feat: Add Postgres ConnPool Config (#5226)

Add Postgres ConnPool Config for user to tune the ConnPool.

Signed-off-by: bruce <[email protected]>

Author: weichou1229

cmd/core-common-config-bootstrapper/res/configuration.yaml

@@ -46,6 +46,9 @@ all-services:
     Port: 5432
     Timeout: "5s"
     Type: "postgres"
+    MaxConns: 4
+    MaxConnIdleTime: "30m"
+    MaxConnLifetime: "1h"
 
   MessageBus:
     Protocol: "mqtt"

cmd/core-data/res/configuration.yaml

@@ -15,6 +15,11 @@ Service:
   Host: "localhost"
   StartupMsg: "This is the Core Data Microservice"
 
+Database:
+  MaxConns: 4
+  MaxConnIdleTime: "30m"
+  MaxConnLifetime: "1h"
+
 Clients:
   core-metadata:
     Protocol: http

cmd/core-keeper/res/configuration.yaml

@@ -39,6 +39,9 @@ Database:
   Port: 5432
   Timeout: "5s"
   Type: "postgres"
+  MaxConns: 4
+  MaxConnIdleTime: "30m"
+  MaxConnLifetime: "1h"
 
 MessageBus:
   Protocol: "mqtt"

internal/pkg/bootstrap/handlers/database.go

@@ -67,10 +67,13 @@ func (d Database) newDBClient(
 	databaseInfo := d.database.GetDatabaseInfo()
 
 	databaseConfig := db.Configuration{
-		Host:     databaseInfo.Host,
-		Port:     databaseInfo.Port,
-		Password: credentials.Password,
-		Timeout:  databaseInfo.Timeout,
+		Host:            databaseInfo.Host,
+		Port:            databaseInfo.Port,
+		Password:        credentials.Password,
+		Timeout:         databaseInfo.Timeout,
+		MaxConns:        databaseInfo.MaxConns,
+		MaxConnIdleTime: databaseInfo.MaxConnIdleTime,
+		MaxConnLifetime: databaseInfo.MaxConnLifetime,
 	}
 
 	switch databaseInfo.Type {

internal/pkg/db/db.go

@@ -29,12 +29,15 @@ var (
 )
 
 type Configuration struct {
-	DbType       string
-	Host         string
-	Port         int
-	Timeout      string
-	DatabaseName string
-	Username     string
-	Password     string
-	BatchSize    int
+	DbType          string
+	Host            string
+	Port            int
+	Timeout         string
+	DatabaseName    string
+	Username        string
+	Password        string
+	BatchSize       int
+	MaxConns        int
+	MaxConnIdleTime string
+	MaxConnLifetime string
 }

internal/pkg/db/postgres/client.go

@@ -12,6 +12,7 @@ import (
 	"net/url"
 	"os"
 	"sync"
+	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
 
@@ -22,6 +23,9 @@ import (
 )
 
 const defaultDBName = "edgex_db"
+const defaultMaxConns = int32(4)
+const defaultMaxConnIdleTime = time.Minute * 30
+const defaultMaxConnLifetime = time.Hour
 
 var once sync.Once
 var dc *Client
@@ -35,17 +39,15 @@ type Client struct {
 
 // NewClient returns a pointer to the Postgres client
 func NewClient(ctx context.Context, config db.Configuration, lc logger.LoggingClient, schemaName, serviceKey, serviceVersion string, sqlFiles embed.FS) (*Client, errors.EdgeX) {
-	// Get the database name from the environment variable
-	databaseName := os.Getenv("EDGEX_DBNAME")
-	if databaseName == "" {
-		databaseName = defaultDBName
-	}
-
 	var edgeXerr errors.EdgeX
 	once.Do(func() {
-		// use url encode to prevent special characters in the connection string
-		connectionStr := "postgres://" + fmt.Sprintf("%s:%s@%s:%d/%s", url.PathEscape(config.Username), url.PathEscape(config.Password), url.PathEscape(config.Host), config.Port, url.PathEscape(databaseName))
-		dbPool, err := pgxpool.New(ctx, connectionStr)
+		connPoolConfig, err := connPoolConConfig(config, lc)
+		if err != nil {
+			edgeXerr = WrapDBError("fail to parse pg conn pool config", err)
+			return
+		}
+
+		dbPool, err := pgxpool.NewWithConfig(ctx, connPoolConfig)
 		if err != nil {
 			edgeXerr = WrapDBError("fail to create pg connection pool", err)
 		}
@@ -78,6 +80,35 @@ func NewClient(ctx context.Context, config db.Configuration, lc logger.LoggingCl
 	return dc, nil
 }
 
+func connPoolConConfig(config db.Configuration, lc logger.LoggingClient) (*pgxpool.Config, error) {
+	// Get the database name from the environment variable
+	databaseName := os.Getenv("EDGEX_DBNAME")
+	if databaseName == "" {
+		databaseName = defaultDBName
+	}
+
+	// use url encode to prevent special characters in the connection string
+	connectionStr := "postgres://" + fmt.Sprintf("%s:%s@%s:%d/%s", url.PathEscape(config.Username), url.PathEscape(config.Password), url.PathEscape(config.Host), config.Port, url.PathEscape(databaseName))
+	connPoolConfig, err := pgxpool.ParseConfig(connectionStr)
+	if err != nil {
+		return nil, WrapDBError("fail to parse pg connection string", err)
+	}
+	connPoolConfig.MaxConns = int32(config.MaxConns)
+	if connPoolConfig.MaxConns <= 0 {
+		lc.Errorf("The MaxConns too small, use default '%d'", defaultMaxConns)
+		connPoolConfig.MaxConns = defaultMaxConns
+	}
+	if connPoolConfig.MaxConnIdleTime, err = time.ParseDuration(config.MaxConnIdleTime); err != nil {
+		connPoolConfig.MaxConnIdleTime = defaultMaxConnIdleTime
+		lc.Errorf("Fail to parse pg conn pool MaxConnIdleTime config, use default '%v', err: %v", defaultMaxConnIdleTime, err)
+	}
+	if connPoolConfig.MaxConnLifetime, err = time.ParseDuration(config.MaxConnLifetime); err != nil {
+		connPoolConfig.MaxConnLifetime = defaultMaxConnLifetime
+		lc.Errorf("Fail to parse pg conn pool MaxConnLifetime config, use default '%v', err: %v", defaultMaxConnLifetime, err)
+	}
+	return connPoolConfig, nil
+}
+
 // CloseSession closes the connections to postgres
 func (c *Client) CloseSession() {
 	c.ConnPool.Close()