diff --git a/cmd/admin-handlers_test.go b/cmd/admin-handlers_test.go index 81e95e632..d460d09b3 100644 --- a/cmd/admin-handlers_test.go +++ b/cmd/admin-handlers_test.go @@ -192,7 +192,9 @@ var ( "format": "namespace", "address": "", "password": "", - "key": "" + "key": "", + "queueDir": "", + "queueLimit": 0 } }, "webhook": { diff --git a/cmd/config-current.go b/cmd/config-current.go index b9591dd3e..2b53d2c15 100644 --- a/cmd/config-current.go +++ b/cmd/config-current.go @@ -391,7 +391,7 @@ func (s *serverConfig) TestNotificationTargets() error { if !v.Enable { continue } - t, err := target.NewRedisTarget(k, v) + t, err := target.NewRedisTarget(k, v, GlobalServiceDoneCh) if err != nil { return fmt.Errorf("redis(%s): %s", k, err.Error()) } @@ -752,7 +752,7 @@ func getNotificationTargets(config *serverConfig) *event.TargetList { for id, args := range config.Notify.Redis { if args.Enable { - newTarget, err := target.NewRedisTarget(id, args) + newTarget, err := target.NewRedisTarget(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 ec4e9c3bb..3c18d669e 100644 --- a/cmd/config-current_test.go +++ b/cmd/config-current_test.go @@ -227,7 +227,7 @@ func TestValidateConfig(t *testing.T) { {`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "elasticsearch": { "1": { "enable": true, "format": "namespace", "url": "example.com", "index": "myindex", "queueDir": "", "queueLimit": 0 } }}}`, true}, // Test 25 - Test Format for Redis - {`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "redis": { "1": { "enable": true, "format": "invalid", "address": "example.com:80", "password": "xxx", "key": "key1" } }}}`, false}, + {`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "redis": { "1": { "enable": true, "format": "invalid", "address": "example.com:80", "password": "xxx", "key": "key1", "queueDir": "", "queueLimit": 0 } }}}`, false}, // Test 26 - Test valid Format for Redis {`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "redis": { "1": { "enable": true, "format": "namespace", "address": "example.com:80", "password": "xxx", "key": "key1" } }}}`, true}, diff --git a/dockerscripts/check-user b/dockerscripts/check-user new file mode 100755 index 000000000..f7d126f66 Binary files /dev/null and b/dockerscripts/check-user differ diff --git a/docs/bucket/notifications/README.md b/docs/bucket/notifications/README.md index 805df0db4..b166681ed 100644 --- a/docs/bucket/notifications/README.md +++ b/docs/bucket/notifications/README.md @@ -460,11 +460,15 @@ An example of Redis configuration is as follows: "format": "namespace", "address": "127.0.0.1:6379", "password": "yoursecret", - "key": "bucketevents" + "key": "bucketevents", + "queueDir": "", + "queueLimit": 0 } } ``` +MinIO supports persistent event store. The persistent store will backup events when the Redis 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. ```sh diff --git a/docs/config/config.sample.json b/docs/config/config.sample.json index be22b956c..d43e8df2a 100644 --- a/docs/config/config.sample.json +++ b/docs/config/config.sample.json @@ -157,7 +157,9 @@ "format": "", "address": "", "password": "", - "key": "" + "key": "", + "queueDir": "", + "queueLimit": 0 } }, "webhook": { diff --git a/pkg/event/target/nsq.go b/pkg/event/target/nsq.go index b2ae34f0c..1bb92cd28 100644 --- a/pkg/event/target/nsq.go +++ b/pkg/event/target/nsq.go @@ -20,11 +20,9 @@ import ( "crypto/tls" "encoding/json" "errors" - "net" "net/url" "os" "path/filepath" - "syscall" "github.com/nsqio/go-nsq" @@ -90,7 +88,7 @@ func (target *NSQTarget) Save(eventData event.Event) error { } if err := target.producer.Ping(); err != nil { // To treat "connection refused" errors as errNotConnected. - if isConnRefusedErr(err) { + if IsConnRefusedErr(err) { return errNotConnected } return err @@ -98,20 +96,6 @@ func (target *NSQTarget) Save(eventData event.Event) error { return target.send(eventData) } -// 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) @@ -133,7 +117,7 @@ func (target *NSQTarget) Send(eventKey string) error { if err := target.producer.Ping(); err != nil { // To treat "connection refused" errors as errNotConnected. - if isConnRefusedErr(err) { + if IsConnRefusedErr(err) { return errNotConnected } return err @@ -198,7 +182,7 @@ func NewNSQTarget(id string, args NSQArgs, doneCh <-chan struct{}) (*NSQTarget, if err := target.producer.Ping(); err != nil { // To treat "connection refused" errors as errNotConnected. - if target.store == nil || !isConnRefusedErr(err) { + if target.store == nil || !IsConnRefusedErr(err) { return nil, err } } diff --git a/pkg/event/target/redis.go b/pkg/event/target/redis.go index 2db62ba97..259b81ded 100644 --- a/pkg/event/target/redis.go +++ b/pkg/event/target/redis.go @@ -17,24 +17,31 @@ package target import ( + "context" "encoding/json" + "errors" "fmt" "net/url" + "os" + "path/filepath" "strings" "time" "github.com/gomodule/redigo/redis" + "github.com/minio/minio/cmd/logger" "github.com/minio/minio/pkg/event" xnet "github.com/minio/minio/pkg/net" ) // RedisArgs - Redis target arguments. type RedisArgs struct { - Enable bool `json:"enable"` - Format string `json:"format"` - Addr xnet.Host `json:"address"` - Password string `json:"password"` - Key string `json:"key"` + Enable bool `json:"enable"` + Format string `json:"format"` + Addr xnet.Host `json:"address"` + Password string `json:"password"` + Key string `json:"key"` + QueueDir string `json:"queueDir"` + QueueLimit uint64 `json:"queueLimit"` } // Validate RedisArgs fields @@ -54,14 +61,45 @@ func (r RedisArgs) Validate() error { return fmt.Errorf("empty key") } + if r.QueueDir != "" { + if !filepath.IsAbs(r.QueueDir) { + return errors.New("queueDir path should be absolute") + } + } + if r.QueueLimit > 10000 { + return errors.New("queueLimit should not exceed 10000") + } + + return nil +} + +func (r RedisArgs) validateFormat(c redis.Conn) error { + typeAvailable, err := redis.String(c.Do("TYPE", r.Key)) + if err != nil { + return err + } + + if typeAvailable != "none" { + expectedType := "hash" + if r.Format == event.AccessFormat { + expectedType = "list" + } + + if typeAvailable != expectedType { + return fmt.Errorf("expected type %v does not match with available type %v", expectedType, typeAvailable) + } + } + return nil } // RedisTarget - Redis target. type RedisTarget struct { - id event.TargetID - args RedisArgs - pool *redis.Pool + id event.TargetID + args RedisArgs + pool *redis.Pool + store Store + firstPing bool } // ID - returns target ID. @@ -69,16 +107,32 @@ func (target *RedisTarget) ID() event.TargetID { return target.id } -// Save - Sends event directly without persisting. +// Save - saves the events to the store if questore is configured, which will be replayed when the redis connection is active. func (target *RedisTarget) Save(eventData event.Event) error { + if target.store != nil { + return target.store.Put(eventData) + } + conn := target.pool.Get() + defer func() { + cErr := conn.Close() + logger.LogOnceIf(context.Background(), cErr, target.ID()) + }() + _, pingErr := conn.Do("PING") + if pingErr != nil { + if IsConnRefusedErr(pingErr) { + return errNotConnected + } + return pingErr + } return target.send(eventData) } +// send - sends an event to the redis. func (target *RedisTarget) send(eventData event.Event) error { conn := target.pool.Get() defer func() { - // FIXME: log returned error. ignore time being. - _ = conn.Close() + cErr := conn.Close() + logger.LogOnceIf(context.Background(), cErr, target.ID()) }() if target.args.Format == event.NamespaceFormat { @@ -98,7 +152,9 @@ func (target *RedisTarget) send(eventData event.Event) error { _, err = conn.Do("HSET", target.args.Key, key, data) } - return err + if err != nil { + return err + } } if target.args.Format == event.AccessFormat { @@ -106,16 +162,58 @@ func (target *RedisTarget) send(eventData event.Event) error { if err != nil { return err } - _, err = conn.Do("RPUSH", target.args.Key, data) - return err + if _, err := conn.Do("RPUSH", target.args.Key, data); err != nil { + return err + } } return nil } -// Send - interface compatible method does no-op. +// Send - reads an event from store and sends it to redis. func (target *RedisTarget) Send(eventKey string) error { - return nil + conn := target.pool.Get() + defer func() { + cErr := conn.Close() + logger.LogOnceIf(context.Background(), cErr, target.ID()) + }() + _, pingErr := conn.Do("PING") + if pingErr != nil { + if IsConnRefusedErr(pingErr) { + return errNotConnected + } + return pingErr + } + + if !target.firstPing { + if err := target.args.validateFormat(conn); err != nil { + if IsConnRefusedErr(err) { + return errNotConnected + } + return err + } + target.firstPing = true + } + + 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 would've been already been sent successfully. + if os.IsNotExist(eErr) { + return nil + } + return eErr + } + + if err := target.send(eventData); err != nil { + if IsConnRefusedErr(err) { + return errNotConnected + } + return err + } + + // Delete the event from store. + return target.store.Del(eventKey) } // Close - does nothing and available for interface compatibility. @@ -124,7 +222,7 @@ func (target *RedisTarget) Close() error { } // NewRedisTarget - creates new Redis target. -func NewRedisTarget(id string, args RedisArgs) (*RedisTarget, error) { +func NewRedisTarget(id string, args RedisArgs, doneCh <-chan struct{}) (*RedisTarget, error) { pool := &redis.Pool{ MaxIdle: 3, IdleTimeout: 2 * 60 * time.Second, @@ -139,8 +237,9 @@ func NewRedisTarget(id string, args RedisArgs) (*RedisTarget, error) { } if _, err = conn.Do("AUTH", args.Password); err != nil { - // FIXME: log returned error. ignore time being. - _ = conn.Close() + cErr := conn.Close() + targetID := event.TargetID{ID: id, Name: "redis"} + logger.LogOnceIf(context.Background(), cErr, targetID.String()) return nil, err } @@ -152,35 +251,47 @@ func NewRedisTarget(id string, args RedisArgs) (*RedisTarget, error) { }, } - conn := pool.Get() + var store Store + + if args.QueueDir != "" { + queueDir := filepath.Join(args.QueueDir, storePrefix+"-redis-"+id) + store = NewQueueStore(queueDir, args.QueueLimit) + if oErr := store.Open(); oErr != nil { + return nil, oErr + } + } + + target := &RedisTarget{ + id: event.TargetID{ID: id, Name: "redis"}, + args: args, + pool: pool, + store: store, + } + + conn := target.pool.Get() defer func() { - // FIXME: log returned error. ignore time being. - _ = conn.Close() + cErr := conn.Close() + logger.LogOnceIf(context.Background(), cErr, target.ID()) }() - if _, err := conn.Do("PING"); err != nil { - return nil, err - } - - typeAvailable, err := redis.String(conn.Do("TYPE", args.Key)) - if err != nil { - return nil, err - } - - if typeAvailable != "none" { - expectedType := "hash" - if args.Format == event.AccessFormat { - expectedType = "list" + _, pingErr := conn.Do("PING") + if pingErr != nil { + if target.store == nil || !IsConnRefusedErr(pingErr) { + return nil, pingErr } - - if typeAvailable != expectedType { - return nil, fmt.Errorf("expected type %v does not match with available type %v", expectedType, typeAvailable) + } else { + if err := target.args.validateFormat(conn); err != nil { + return nil, err } + target.firstPing = true } - return &RedisTarget{ - id: event.TargetID{ID: id, Name: "redis"}, - args: args, - pool: pool, - }, nil + 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/store.go b/pkg/event/target/store.go index 1b94ca7e1..ae30683e1 100644 --- a/pkg/event/target/store.go +++ b/pkg/event/target/store.go @@ -79,6 +79,20 @@ func replayEvents(store Store, doneCh <-chan struct{}) <-chan string { return eventKeyCh } +// 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 +} + // isConnResetErr - Checks for connection reset errors. func isConnResetErr(err error) bool { if opErr, ok := err.(*net.OpError); ok { diff --git a/pkg/event/target/webhook.go b/pkg/event/target/webhook.go index 2f2a4a0b3..c95bdab1d 100644 --- a/pkg/event/target/webhook.go +++ b/pkg/event/target/webhook.go @@ -30,7 +30,6 @@ import ( "net/url" "os" "path/filepath" - "syscall" "time" "github.com/minio/minio/pkg/event" @@ -134,20 +133,6 @@ func (target *WebhookTarget) send(eventData event.Event) error { return nil } -// IsConnRefusedErr - To check for "connection refused" errors. -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 - reads an event from store and sends it to webhook. func (target *WebhookTarget) Send(eventKey string) error {