Set the maximum open connections limit in PG and MySQL target configs (#10558)

As the bulk/recursive delete will require multiple connections to open at an instance,
The default open connections limit will be reached which results in the following error

```FATAL:  sorry, too many clients already```

By setting the open connections to a reasonable value - `2`, We ensure that the max open connections
will not be exhausted and lie under bounds.

The queries are simple inserts/updates/deletes which is operational and sufficient with the
the maximum open connection limit is 2.

Fixes #10553

Allow user configuration for MaxOpenConnections

cmd/config/notify/help.go

@@ -340,6 +340,12 @@ var (
 			Optional:    true,
 			Type:        "sentence",
 		},
+		config.HelpKV{
+			Key:         target.PostgresMaxOpenConnections,
+			Description: "To set the maximum number of open connections to the database. The value is set to `2` by default.",
+			Optional:    true,
+			Type:        "number",
+		},
 	}
 
 	HelpMySQL = config.HelpKVS{
@@ -377,6 +383,12 @@ var (
 			Optional:    true,
 			Type:        "sentence",
 		},
+		config.HelpKV{
+			Key:         target.MySQLMaxOpenConnections,
+			Description: "To set the maximum number of open connections to the database. The value is set to `2` by default.",
+			Optional:    true,
+			Type:        "number",
+		},
 	}
 
 	HelpNATS = config.HelpKVS{

cmd/config/notify/legacy.go

@@ -357,6 +357,10 @@ func SetNotifyPostgres(s config.Config, psqName string, cfg target.PostgreSQLArg
 			Key:   target.PostgresQueueLimit,
 			Value: strconv.Itoa(int(cfg.QueueLimit)),
 		},
+		config.KV{
+			Key:   target.PostgresMaxOpenConnections,
+			Value: strconv.Itoa(cfg.MaxOpenConnections),
+		},
 	}
 
 	return nil
@@ -554,6 +558,10 @@ func SetNotifyMySQL(s config.Config, sqlName string, cfg target.MySQLArgs) error
 			Key:   target.MySQLQueueLimit,
 			Value: strconv.Itoa(int(cfg.QueueLimit)),
 		},
+		config.KV{
+			Key:   target.MySQLMaxOpenConnections,
+			Value: strconv.Itoa(cfg.MaxOpenConnections),
+		},
 	}
 
 	return nil

cmd/config/notify/parse.go

@@ -807,6 +807,10 @@ var (
 			Key:   target.MySQLQueueLimit,
 			Value: "0",
 		},
+		config.KV{
+			Key:   target.MySQLMaxOpenConnections,
+			Value: "2",
+		},
 	}
 )
 
@@ -855,6 +859,17 @@ func GetNotifyMySQL(mysqlKVS map[string]config.KVS) (map[string]target.MySQLArgs
 		if k != config.Default {
 			queueDirEnv = queueDirEnv + config.Default + k
 		}
+
+		maxOpenConnectionsEnv := target.EnvMySQLMaxOpenConnections
+		if k != config.Default {
+			maxOpenConnectionsEnv = maxOpenConnectionsEnv + config.Default + k
+		}
+
+		maxOpenConnections, cErr := strconv.Atoi(env.Get(maxOpenConnectionsEnv, kv.Get(target.MySQLMaxOpenConnections)))
+		if cErr != nil {
+			return nil, cErr
+		}
+
 		mysqlArgs := target.MySQLArgs{
 			Enable:             enabled,
 			Format:             env.Get(formatEnv, kv.Get(target.MySQLFormat)),
@@ -862,6 +877,7 @@ func GetNotifyMySQL(mysqlKVS map[string]config.KVS) (map[string]target.MySQLArgs
 			Table:              env.Get(tableEnv, kv.Get(target.MySQLTable)),
 			QueueDir:           env.Get(queueDirEnv, kv.Get(target.MySQLQueueDir)),
 			QueueLimit:         queueLimit,
+			MaxOpenConnections: maxOpenConnections,
 		}
 		if err = mysqlArgs.Validate(); err != nil {
 			return nil, err
@@ -1235,6 +1251,10 @@ var (
 			Key:   target.PostgresQueueLimit,
 			Value: "0",
 		},
+		config.KV{
+			Key:   target.PostgresMaxOpenConnections,
+			Value: "2",
+		},
 	}
 )
 
@@ -1285,6 +1305,16 @@ func GetNotifyPostgres(postgresKVS map[string]config.KVS) (map[string]target.Pos
 			queueDirEnv = queueDirEnv + config.Default + k
 		}
 
+		maxOpenConnectionsEnv := target.EnvPostgresMaxOpenConnections
+		if k != config.Default {
+			maxOpenConnectionsEnv = maxOpenConnectionsEnv + config.Default + k
+		}
+
+		maxOpenConnections, cErr := strconv.Atoi(env.Get(maxOpenConnectionsEnv, kv.Get(target.PostgresMaxOpenConnections)))
+		if cErr != nil {
+			return nil, cErr
+		}
+
 		psqlArgs := target.PostgreSQLArgs{
 			Enable:             enabled,
 			Format:             env.Get(formatEnv, kv.Get(target.PostgresFormat)),
@@ -1292,6 +1322,7 @@ func GetNotifyPostgres(postgresKVS map[string]config.KVS) (map[string]target.Pos
 			Table:              env.Get(tableEnv, kv.Get(target.PostgresTable)),
 			QueueDir:           env.Get(queueDirEnv, kv.Get(target.PostgresQueueDir)),
 			QueueLimit:         uint64(queueLimit),
+			MaxOpenConnections: maxOpenConnections,
 		}
 		if err = psqlArgs.Validate(); err != nil {
 			return nil, err

docs/bucket/notifications/README.md

@@ -861,6 +861,7 @@ table*              (string)             DB table name to store/update events, t
 format*              (namespace*|access)  'namespace' reflects current bucket/object list and 'access' reflects a journal of object operations, defaults to 'namespace'
 queue_dir            (path)               staging dir for undelivered messages e.g. '/home/events'
 queue_limit          (number)             maximum limit for undelivered messages, defaults to '100000'
+max_open_connections (number)             maximum number of open connections to the database, defaults to '2'
 comment              (sentence)           optionally add a comment to this setting
 ```
 
@@ -877,8 +878,12 @@ MINIO_NOTIFY_POSTGRES_FORMAT*             (namespace*|access)  'namespace' refle
 MINIO_NOTIFY_POSTGRES_QUEUE_DIR            (path)               staging dir for undelivered messages e.g. '/home/events'
 MINIO_NOTIFY_POSTGRES_QUEUE_LIMIT          (number)             maximum limit for undelivered messages, defaults to '100000'
 MINIO_NOTIFY_POSTGRES_COMMENT              (sentence)           optionally add a comment to this setting
+MINIO_NOTIFY_POSTGRES_MAX_OPEN_CONNECTIONS (number)             maximum number of open connections to the database, defaults to '2'
 ```
 
+> NOTE: If the `max_open_connections` key or the environment variable `MINIO_NOTIFY_POSTGRES_MAX_OPEN_CONNECTIONS` is set to `0`, There will be no limit set on the number of
+> open connections to the database. This setting is generally NOT recommended as the behaviour may be inconsistent during recursive deletes in `namespace` format.
+
 MinIO supports persistent event store. The persistent store will backup events when the PostgreSQL connection goes offline and replays it when the broker comes back online. The event store can be configured by setting the directory path in `queue_dir` field and the maximum limit of events in the queue_dir in `queue_limit` field. For eg, the `queue_dir` can be `/home/events` and `queue_limit` can be `1000`. By default, the `queue_limit` is set to 100000.
 
 Note that for illustration here, we have disabled SSL. In the interest of security, for production this is not recommended.
@@ -990,6 +995,7 @@ table*       (string)             DB table name to store/update events, table is
 format*              (namespace*|access)  'namespace' reflects current bucket/object list and 'access' reflects a journal of object operations, defaults to 'namespace'
 queue_dir            (path)               staging dir for undelivered messages e.g. '/home/events'
 queue_limit          (number)             maximum limit for undelivered messages, defaults to '100000'
+max_open_connections (number)             maximum number of open connections to the database, defaults to '2'
 comment              (sentence)           optionally add a comment to this setting
 ```
 
@@ -1005,9 +1011,13 @@ MINIO_NOTIFY_MYSQL_TABLE*       (string)             DB table name to store/upda
 MINIO_NOTIFY_MYSQL_FORMAT*              (namespace*|access)  'namespace' reflects current bucket/object list and 'access' reflects a journal of object operations, defaults to 'namespace'
 MINIO_NOTIFY_MYSQL_QUEUE_DIR            (path)               staging dir for undelivered messages e.g. '/home/events'
 MINIO_NOTIFY_MYSQL_QUEUE_LIMIT          (number)             maximum limit for undelivered messages, defaults to '100000'
+MINIO_NOTIFY_MYSQL_MAX_OPEN_CONNECTIONS (number)             maximum number of open connections to the database, defaults to '2'
 MINIO_NOTIFY_MYSQL_COMMENT              (sentence)           optionally add a comment to this setting
 ```
 
+> NOTE: If the `max_open_connections` key or the environment variable `MINIO_NOTIFY_MYSQL_MAX_OPEN_CONNECTIONS` is set to `0`, There will be no limit set on the number of
+> open connections to the database. This setting is generally NOT recommended as the behaviour may be inconsistent during recursive deletes in `namespace` format.
+
 `dsn_string` is required and is of form `"<user>:<password>@tcp(<host>:<port>)/<database>"`
 
 MinIO supports persistent event store. The persistent store will backup events if MySQL connection goes offline and then replays the stored events when the broken connection comes back up. The event store can be configured by setting a directory path in `queue_dir` field, and the maximum number of events, which can be stored in a `queue_dir`, in `queue_limit` field. For example, `queue_dir` can be set to `/home/events` and `queue_limit` can be set to `1000`. By default, the `queue_limit` is set to `100000`.

go.mod

@@ -40,7 +40,7 @@ require (
 	github.com/klauspost/pgzip v1.2.1
 	github.com/klauspost/readahead v1.3.1
 	github.com/klauspost/reedsolomon v1.9.9
-	github.com/lib/pq v1.7.0
+	github.com/lib/pq v1.8.0
 	github.com/mattn/go-colorable v0.1.4
 	github.com/mattn/go-ieproxy v0.0.1 // indirect
 	github.com/mattn/go-isatty v0.0.8

go.sum

@@ -280,6 +280,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lib/pq v1.7.0 h1:h93mCPfUSkaul3Ka/VG8uZdmW1uMHDGxzu0NWHuJmHY=
 github.com/lib/pq v1.7.0/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/lib/pq v1.8.0 h1:9xohqzkUwzR4Ga4ivdTcawVS89YSDVxXMa3xJX3cGzg=
+github.com/lib/pq v1.8.0/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/mailru/easyjson v0.7.1 h1:mdxE1MF9o53iCb2Ghj1VfWvh7ZOwHpnVG/xwXrV90U8=
 github.com/mailru/easyjson v0.7.1/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=

pkg/event/target/mysql.go

@@ -93,6 +93,7 @@ const (
 	MySQLDatabase           = "database"
 	MySQLQueueLimit         = "queue_limit"
 	MySQLQueueDir           = "queue_dir"
+	MySQLMaxOpenConnections = "max_open_connections"
 
 	EnvMySQLEnable             = "MINIO_NOTIFY_MYSQL_ENABLE"
 	EnvMySQLFormat             = "MINIO_NOTIFY_MYSQL_FORMAT"
@@ -105,6 +106,7 @@ const (
 	EnvMySQLDatabase           = "MINIO_NOTIFY_MYSQL_DATABASE"
 	EnvMySQLQueueLimit         = "MINIO_NOTIFY_MYSQL_QUEUE_LIMIT"
 	EnvMySQLQueueDir           = "MINIO_NOTIFY_MYSQL_QUEUE_DIR"
+	EnvMySQLMaxOpenConnections = "MINIO_NOTIFY_MYSQL_MAX_OPEN_CONNECTIONS"
 )
 
 // MySQLArgs - MySQL target arguments.
@@ -120,6 +122,7 @@ type MySQLArgs struct {
 	Database           string   `json:"database"`
 	QueueDir           string   `json:"queueDir"`
 	QueueLimit         uint64   `json:"queueLimit"`
+	MaxOpenConnections int      `json:"maxOpenConnections"`
 }
 
 // Validate MySQLArgs fields
@@ -162,6 +165,10 @@ func (m MySQLArgs) Validate() error {
 		}
 	}
 
+	if m.MaxOpenConnections < 0 {
+		return errors.New("maxOpenConnections cannot be less than zero")
+	}
+
 	return nil
 }
 
@@ -196,6 +203,10 @@ func (target *MySQLTarget) IsActive() (bool, error) {
 			return false, sErr
 		}
 		target.db = db
+		if target.args.MaxOpenConnections > 0 {
+			// Set the maximum connections limit
+			target.db.SetMaxOpenConns(target.args.MaxOpenConnections)
+		}
 	}
 	if err := target.db.Ping(); err != nil {
 		if IsConnErr(err) {
@@ -384,6 +395,11 @@ func NewMySQLTarget(id string, args MySQLArgs, doneCh <-chan struct{}, loggerOnc
 	}
 	target.db = db
 
+	if args.MaxOpenConnections > 0 {
+		// Set the maximum connections limit
+		target.db.SetMaxOpenConns(args.MaxOpenConnections)
+	}
+
 	var store Store
 
 	if args.QueueDir != "" {

pkg/event/target/postgresql.go

@@ -94,6 +94,7 @@ const (
 	PostgresDatabase           = "database"
 	PostgresQueueDir           = "queue_dir"
 	PostgresQueueLimit         = "queue_limit"
+	PostgresMaxOpenConnections = "max_open_connections"
 
 	EnvPostgresEnable             = "MINIO_NOTIFY_POSTGRES_ENABLE"
 	EnvPostgresFormat             = "MINIO_NOTIFY_POSTGRES_FORMAT"
@@ -106,6 +107,7 @@ const (
 	EnvPostgresDatabase           = "MINIO_NOTIFY_POSTGRES_DATABASE"
 	EnvPostgresQueueDir           = "MINIO_NOTIFY_POSTGRES_QUEUE_DIR"
 	EnvPostgresQueueLimit         = "MINIO_NOTIFY_POSTGRES_QUEUE_LIMIT"
+	EnvPostgresMaxOpenConnections = "MINIO_NOTIFY_POSTGRES_MAX_OPEN_CONNECTIONS"
 )
 
 // PostgreSQLArgs - PostgreSQL target arguments.
@@ -121,6 +123,7 @@ type PostgreSQLArgs struct {
 	Database           string    `json:"database"` // default: same as user
 	QueueDir           string    `json:"queueDir"`
 	QueueLimit         uint64    `json:"queueLimit"`
+	MaxOpenConnections int       `json:"maxOpenConnections"`
 }
 
 // Validate PostgreSQLArgs fields
@@ -160,6 +163,10 @@ func (p PostgreSQLArgs) Validate() error {
 		}
 	}
 
+	if p.MaxOpenConnections < 0 {
+		return errors.New("maxOpenConnections cannot be less than zero")
+	}
+
 	return nil
 }
 
@@ -195,6 +202,10 @@ func (target *PostgreSQLTarget) IsActive() (bool, error) {
 			return false, err
 		}
 		target.db = db
+		if target.args.MaxOpenConnections > 0 {
+			// Set the maximum connections limit
+			target.db.SetMaxOpenConns(target.args.MaxOpenConnections)
+		}
 	}
 	if err := target.db.Ping(); err != nil {
 		if IsConnErr(err) {
@@ -391,6 +402,11 @@ func NewPostgreSQLTarget(id string, args PostgreSQLArgs, doneCh <-chan struct{},
 	}
 	target.db = db
 
+	if args.MaxOpenConnections > 0 {
+		// Set the maximum connections limit
+		target.db.SetMaxOpenConns(args.MaxOpenConnections)
+	}
+
 	var store Store
 
 	if args.QueueDir != "" {

ruleguard.rules.go

@@ -438,7 +438,6 @@ func httpheaderadd(m fluent.Matcher) {
 		Suggest(`$H.Set($KEY, $VALUE)`)
 }
 
-
 func hmacnew(m fluent.Matcher) {
 	m.Match("hmac.New(func() hash.Hash { return $x }, $_)",
 		`$f := func() hash.Hash { return $x }