diff --git a/cmd/admin-handlers_test.go b/cmd/admin-handlers_test.go index 761c29cbb..81e95e632 100644 --- a/cmd/admin-handlers_test.go +++ b/cmd/admin-handlers_test.go @@ -168,7 +168,9 @@ var ( "tls": { "enable": false, "skipVerify": false - } + }, + "queueDir": "", + "queueLimit": 0 } }, "postgresql": { diff --git a/cmd/config-current.go b/cmd/config-current.go index 2fdb3f8b1..b9591dd3e 100644 --- a/cmd/config-current.go +++ b/cmd/config-current.go @@ -369,7 +369,7 @@ func (s *serverConfig) TestNotificationTargets() error { if !v.Enable { continue } - t, err := target.NewNSQTarget(k, v) + t, err := target.NewNSQTarget(k, v, GlobalServiceDoneCh) if err != nil { return fmt.Errorf("nsq(%s): %s", k, err.Error()) } @@ -724,7 +724,7 @@ func getNotificationTargets(config *serverConfig) *event.TargetList { for id, args := range config.Notify.NSQ { if args.Enable { - newTarget, err := target.NewNSQTarget(id, args) + newTarget, err := target.NewNSQTarget(id, args, GlobalServiceDoneCh) if err != nil { logger.LogIf(context.Background(), err) continue diff --git a/cmd/config-current_test.go b/cmd/config-current_test.go index ad2679a7c..ec4e9c3bb 100644 --- a/cmd/config-current_test.go +++ b/cmd/config-current_test.go @@ -236,7 +236,7 @@ func TestValidateConfig(t *testing.T) { {`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "mqtt": { "1": { "enable": true, "broker": "", "topic": "", "qos": 0, "username": "", "password": "", "queueDir": "", "queueLimit": 0}}}}`, false}, // Test 28 - Test NSQ - {`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "nsq": { "1": { "enable": true, "nsqdAddress": "", "topic": ""} }}}`, false}, + {`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "nsq": { "1": { "enable": true, "nsqdAddress": "", "topic": "", "queueDir": "", "queueLimit": 0} }}}`, false}, } for i, testCase := range testCases { diff --git a/docs/bucket/notifications/README.md b/docs/bucket/notifications/README.md index a6a82255c..805df0db4 100644 --- a/docs/bucket/notifications/README.md +++ b/docs/bucket/notifications/README.md @@ -1110,12 +1110,17 @@ An example configuration for NSQ is shown below: "tls": { "enable": false, "skipVerify": true - } + }, + "queueDir": "", + "queueLimit": 0 } } + +MinIO supports persistent event store. The persistent store will backup events when the NSQ broker goes offline and replays it when the broker comes back online. The event store can be configured by setting the directory path in `queueDir` field and the maximum limit of events in the queueDir in `queueLimit` field. For eg, the `queueDir` can be `/home/events` and `queueLimit` can be `1000`. By default, the `queueLimit` is set to 10000. + ``` -To update the configuration, use `mc admin config get` command to get the current configuration file for the minio deployment in json format, and save it locally. +To update the configuration, use `mc admin config get` command to get the current configuration file for the MinIO deployment in json format, and save it locally. ```sh $ mc admin config get myminio/ > /tmp/myconfig diff --git a/docs/config/config.sample.json b/docs/config/config.sample.json index 363584400..be22b956c 100644 --- a/docs/config/config.sample.json +++ b/docs/config/config.sample.json @@ -133,7 +133,9 @@ "tls": { "enable": false, "skipVerify": true - } + }, + "queueDir": "", + "queueLimit": 0 } }, "postgresql": { diff --git a/pkg/event/target/nsq.go b/pkg/event/target/nsq.go index 1d5e34d08..b2ae34f0c 100644 --- a/pkg/event/target/nsq.go +++ b/pkg/event/target/nsq.go @@ -20,7 +20,11 @@ import ( "crypto/tls" "encoding/json" "errors" + "net" "net/url" + "os" + "path/filepath" + "syscall" "github.com/nsqio/go-nsq" @@ -37,6 +41,8 @@ type NSQArgs struct { Enable bool `json:"enable"` SkipVerify bool `json:"skipVerify"` } `json:"tls"` + QueueDir string `json:"queueDir"` + QueueLimit uint64 `json:"queueLimit"` } // Validate NSQArgs fields @@ -52,6 +58,14 @@ func (n NSQArgs) Validate() error { if n.Topic == "" { return errors.New("empty topic") } + if n.QueueDir != "" { + if !filepath.IsAbs(n.QueueDir) { + return errors.New("queueDir path should be absolute") + } + } + if n.QueueLimit > 10000 { + return errors.New("queueLimit should not exceed 10000") + } return nil } @@ -61,6 +75,7 @@ type NSQTarget struct { id event.TargetID args NSQArgs producer *nsq.Producer + store Store } // ID - returns target ID. @@ -68,12 +83,37 @@ func (target *NSQTarget) ID() event.TargetID { return target.id } -// Save - Sends event directly without persisting. +// Save - saves the events to the store which will be replayed when the nsq connection is active. func (target *NSQTarget) Save(eventData event.Event) error { + if target.store != nil { + return target.store.Put(eventData) + } + if err := target.producer.Ping(); err != nil { + // To treat "connection refused" errors as errNotConnected. + if isConnRefusedErr(err) { + return errNotConnected + } + return err + } return target.send(eventData) } -func (target *NSQTarget) send(eventData event.Event) (err error) { +// isConnRefusedErr - To check fot "connection refused" error. +func isConnRefusedErr(err error) bool { + if opErr, ok := err.(*net.OpError); ok { + if sysErr, ok := opErr.Err.(*os.SyscallError); ok { + if errno, ok := sysErr.Err.(syscall.Errno); ok { + if errno == syscall.ECONNREFUSED { + return true + } + } + } + } + return false +} + +// send - sends an event to the NSQ. +func (target *NSQTarget) send(eventData event.Event) error { objectName, err := url.QueryUnescape(eventData.S3.Object.Key) if err != nil { return err @@ -85,14 +125,36 @@ func (target *NSQTarget) send(eventData event.Event) (err error) { return err } - err = target.producer.Publish(target.args.Topic, data) - - return err + return target.producer.Publish(target.args.Topic, data) } -// Send - interface compatible method does no-op. +// Send - reads an event from store and sends it to NSQ. func (target *NSQTarget) Send(eventKey string) error { - return nil + + if err := target.producer.Ping(); err != nil { + // To treat "connection refused" errors as errNotConnected. + if isConnRefusedErr(err) { + return errNotConnected + } + return err + } + + eventData, eErr := target.store.Get(eventKey) + if eErr != nil { + // The last event key in a successful batch will be sent in the channel atmost once by the replayEvents() + // Such events will not exist and wouldve been already been sent successfully. + if os.IsNotExist(eErr) { + return nil + } + return eErr + } + + if err := target.send(eventData); err != nil { + return err + } + + // Delete the event from store. + return target.store.Del(eventKey) } // Close - closes underneath connections to NSQD server. @@ -103,7 +165,7 @@ func (target *NSQTarget) Close() (err error) { } // NewNSQTarget - creates new NSQ target. -func NewNSQTarget(id string, args NSQArgs) (*NSQTarget, error) { +func NewNSQTarget(id string, args NSQArgs, doneCh <-chan struct{}) (*NSQTarget, error) { config := nsq.NewConfig() if args.TLS.Enable { config.TlsV1 = true @@ -111,15 +173,42 @@ func NewNSQTarget(id string, args NSQArgs) (*NSQTarget, error) { InsecureSkipVerify: args.TLS.SkipVerify, } } - producer, err := nsq.NewProducer(args.NSQDAddress.String(), config) + var store Store + + if args.QueueDir != "" { + queueDir := filepath.Join(args.QueueDir, storePrefix+"-nsq-"+id) + store = NewQueueStore(queueDir, args.QueueLimit) + if oErr := store.Open(); oErr != nil { + return nil, oErr + } + } + + producer, err := nsq.NewProducer(args.NSQDAddress.String(), config) if err != nil { return nil, err } - return &NSQTarget{ + target := &NSQTarget{ id: event.TargetID{ID: id, Name: "nsq"}, args: args, producer: producer, - }, nil + store: store, + } + + if err := target.producer.Ping(); err != nil { + // To treat "connection refused" errors as errNotConnected. + if target.store == nil || !isConnRefusedErr(err) { + return nil, err + } + } + + if target.store != nil { + // Replays the events from the store. + eventKeyCh := replayEvents(target.store, doneCh) + // Start replaying events from the store. + go sendEvents(target, eventKeyCh, doneCh) + } + + return target, nil } diff --git a/pkg/event/target/nsq_test.go b/pkg/event/target/nsq_test.go index 67619f864..12f67142b 100644 --- a/pkg/event/target/nsq_test.go +++ b/pkg/event/target/nsq_test.go @@ -17,71 +17,10 @@ package target import ( - "reflect" - "testing" - - "github.com/minio/minio/pkg/event" - "github.com/minio/minio/pkg/net" xnet "github.com/minio/minio/pkg/net" - "github.com/nsqio/go-nsq" + "testing" ) -func TestNewNSQTarget(t *testing.T) { - type args struct { - id string - args NSQArgs - } - tests := []struct { - name string - args args - want *NSQTarget - wantErr bool - }{ - { - name: "test1", - args: args{ - id: "id", - args: NSQArgs{ - Enable: true, - Topic: "", - TLS: struct { - Enable bool `json:"enable"` - SkipVerify bool `json:"skipVerify"` - }{true, true}, - }, - }, - want: &NSQTarget{ - id: event.TargetID{ID: "id", Name: "nsq"}, - args: NSQArgs{ - Enable: true, - NSQDAddress: net.Host{}, - Topic: "", - TLS: struct { - Enable bool `json:"enable"` - SkipVerify bool `json:"skipVerify"` - }{true, true}, - }, - producer: &nsq.Producer{}, - }, - wantErr: false, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got, err := NewNSQTarget(tt.args.id, tt.args.args) - // dirty hack, otherwhise cannot compare the pointers: - tt.want.producer = got.producer - if (err != nil) != tt.wantErr { - t.Errorf("NewNSQTarget() error = %v, wantErr %v", err, tt.wantErr) - return - } - if !reflect.DeepEqual(got, tt.want) { - t.Errorf("NewNSQTarget() = %v, want %v", got, tt.want) - } - }) - } -} - func TestNSQArgs_Validate(t *testing.T) { type fields struct { Enable bool