K8SPS-392: configure bootstrap readTimeout

cmd/bootstrap/async_replication.go

@@ -78,7 +78,18 @@ func bootstrapAsyncReplication(ctx context.Context) error {
 		return errors.Wrapf(err, "get %s password", apiv1alpha1.UserOperator)
 	}
 
-	db, err := database.NewDatabase(ctx, apiv1alpha1.UserOperator, operatorPass, podIp, mysql.DefaultAdminPort)
+	params := database.DBParams{
+		User: apiv1alpha1.UserOperator,
+		Pass: operatorPass,
+		Host: podIp,
+	}
+	readTimeout, err := getReadTimeout()
+	if err != nil {
+		return errors.Wrap(err, "get read timeout")
+	}
+	params.ReadTimeoutSeconds = readTimeout
+
+	db, err := database.NewDatabase(ctx, params)
 	if err != nil {
 		return errors.Wrap(err, "connect to database")
 	}
@@ -199,7 +210,18 @@ func getTopology(ctx context.Context, fqdn string, peers sets.Set[string]) (stri
 	}
 
 	for _, peer := range sets.List(peers) {
-		db, err := database.NewDatabase(ctx, apiv1alpha1.UserOperator, operatorPass, peer, mysql.DefaultAdminPort)
+		params := database.DBParams{
+			User: apiv1alpha1.UserOperator,
+			Pass: operatorPass,
+			Host: peer,
+		}
+		readTimeout, err := getReadTimeout()
+		if err != nil {
+			return "", nil, errors.Wrap(err, "get read timeout")
+		}
+		params.ReadTimeoutSeconds = readTimeout
+
+		db, err := database.NewDatabase(ctx, params)
 		if err != nil {
 			return "", nil, errors.Wrapf(err, "connect to %s", peer)
 		}
@@ -254,7 +276,18 @@ func selectDonor(ctx context.Context, fqdn, primary string, replicas []string) (
 	}
 
 	for _, replica := range replicas {
-		db, err := database.NewDatabase(ctx, apiv1alpha1.UserOperator, operatorPass, replica, mysql.DefaultAdminPort)
+		params := database.DBParams{
+			User: apiv1alpha1.UserOperator,
+			Pass: operatorPass,
+			Host: replica,
+		}
+		readTimeout, err := getReadTimeout()
+		if err != nil {
+			return "", errors.Wrap(err, "get read timeout")
+		}
+		params.ReadTimeoutSeconds = readTimeout
+
+		db, err := database.NewDatabase(ctx, params)
 		if err != nil {
 			continue
 		}

cmd/bootstrap/utils.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -15,6 +16,7 @@ import (
 	apiv1alpha1 "github.com/percona/percona-server-mysql-operator/api/v1alpha1"
 	"github.com/percona/percona-server-mysql-operator/pkg/k8s"
 	"github.com/percona/percona-server-mysql-operator/pkg/mysql"
+	"github.com/percona/percona-server-mysql-operator/pkg/naming"
 )
 
 func getFQDN(svcName string) (string, error) {
@@ -31,6 +33,22 @@ func getFQDN(svcName string) (string, error) {
 	return fmt.Sprintf("%s.%s.%s", hostname, svcName, namespace), nil
 }
 
+func getReadTimeout() (uint32, error) {
+	s, ok := os.LookupEnv(naming.EnvBootstrapReadTimeout)
+	if !ok {
+		return 0, nil
+	}
+	readTimeout, err := strconv.Atoi(s)
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to parse BOOTSTRAP_READ_TIMEOUT")
+	}
+	if readTimeout < 0 {
+		return 0, errors.New("BOOTSTRAP_READ_TIMEOUT should be a positive value")
+	}
+
+	return uint32(readTimeout), nil
+}
+
 func getSecret(username apiv1alpha1.SystemUser) (string, error) {
 	path := filepath.Join(mysql.CredsMountPath, string(username))
 	sBytes, err := os.ReadFile(path)

cmd/db/db.go

@@ -10,6 +10,7 @@ import (
 
 	apiv1alpha1 "github.com/percona/percona-server-mysql-operator/api/v1alpha1"
 	"github.com/percona/percona-server-mysql-operator/pkg/db"
+	defs "github.com/percona/percona-server-mysql-operator/pkg/mysql"
 )
 
 const defaultChannelName = ""
@@ -22,23 +23,48 @@ type DB struct {
 	db *sql.DB
 }
 
-func NewDatabase(ctx context.Context, user apiv1alpha1.SystemUser, pass, host string, port int32) (*DB, error) {
+type DBParams struct {
+	User apiv1alpha1.SystemUser
+	Pass string
+	Host string
+	Port int32
+
+	ReadTimeoutSeconds uint32
+}
+
+func (p *DBParams) setDefaults() {
+	if p.Port == 0 {
+		p.Port = defs.DefaultAdminPort
+	}
+
+	if p.ReadTimeoutSeconds == 0 {
+		p.ReadTimeoutSeconds = 10
+	}
+}
+
+func (p *DBParams) DSN() string {
+	p.setDefaults()
+
 	config := mysql.NewConfig()
 
-	config.User = string(user)
-	config.Passwd = pass
+	config.User = string(p.User)
+	config.Passwd = p.Pass
 	config.Net = "tcp"
-	config.Addr = fmt.Sprintf("%s:%d", host, port)
+	config.Addr = fmt.Sprintf("%s:%d", p.Host, p.Port)
 	config.DBName = "performance_schema"
 	config.Params = map[string]string{
 		"interpolateParams": "true",
 		"timeout":           "10s",
-		"readTimeout":       "10s",
+		"readTimeout":       fmt.Sprintf("%ds", p.ReadTimeoutSeconds),
 		"writeTimeout":      "10s",
 		"tls":               "preferred",
 	}
 
-	db, err := sql.Open("mysql", config.FormatDSN())
+	return config.FormatDSN()
+}
+
+func NewDatabase(ctx context.Context, params DBParams) (*DB, error) {
+	db, err := sql.Open("mysql", params.DSN())
 	if err != nil {
 		return nil, errors.Wrap(err, "connect to MySQL")
 	}

cmd/healthcheck/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/pkg/errors"
 
 	apiv1alpha1 "github.com/percona/percona-server-mysql-operator/api/v1alpha1"
+	"github.com/percona/percona-server-mysql-operator/cmd/db"
 	database "github.com/percona/percona-server-mysql-operator/cmd/db"
 	mysqldb "github.com/percona/percona-server-mysql-operator/pkg/db"
 	"github.com/percona/percona-server-mysql-operator/pkg/k8s"
@@ -82,7 +83,12 @@ func checkReadinessAsync(ctx context.Context) error {
 		return errors.Wrapf(err, "get %s password", apiv1alpha1.UserMonitor)
 	}
 
-	db, err := database.NewDatabase(ctx, apiv1alpha1.UserMonitor, monitorPass, podIP, mysql.DefaultAdminPort)
+	params := db.DBParams{
+		User: apiv1alpha1.UserMonitor,
+		Pass: monitorPass,
+		Host: podIP,
+	}
+	db, err := database.NewDatabase(ctx, params)
 	if err != nil {
 		return errors.Wrap(err, "connect to db")
 	}
@@ -117,7 +123,12 @@ func checkReadinessGR(ctx context.Context) error {
 		return errors.Wrapf(err, "get %s password", apiv1alpha1.UserMonitor)
 	}
 
-	db, err := database.NewDatabase(ctx, apiv1alpha1.UserMonitor, monitorPass, podIP, mysql.DefaultAdminPort)
+	params := db.DBParams{
+		User: apiv1alpha1.UserMonitor,
+		Pass: monitorPass,
+		Host: podIP,
+	}
+	db, err := database.NewDatabase(ctx, params)
 	if err != nil {
 		return errors.Wrap(err, "connect to db")
 	}
@@ -151,7 +162,12 @@ func checkLivenessAsync(ctx context.Context) error {
 		return errors.Wrapf(err, "get %s password", apiv1alpha1.UserMonitor)
 	}
 
-	db, err := database.NewDatabase(ctx, apiv1alpha1.UserMonitor, monitorPass, podIP, mysql.DefaultAdminPort)
+	params := db.DBParams{
+		User: apiv1alpha1.UserMonitor,
+		Pass: monitorPass,
+		Host: podIP,
+	}
+	db, err := database.NewDatabase(ctx, params)
 	if err != nil {
 		return errors.Wrap(err, "connect to db")
 	}
@@ -171,7 +187,12 @@ func checkLivenessGR(ctx context.Context) error {
 		return errors.Wrapf(err, "get %s password", apiv1alpha1.UserMonitor)
 	}
 
-	db, err := database.NewDatabase(ctx, apiv1alpha1.UserMonitor, monitorPass, podIP, mysql.DefaultAdminPort)
+	params := db.DBParams{
+		User: apiv1alpha1.UserMonitor,
+		Pass: monitorPass,
+		Host: podIP,
+	}
+	db, err := database.NewDatabase(ctx, params)
 	if err != nil {
 		return errors.Wrap(err, "connect to db")
 	}
@@ -202,7 +223,12 @@ func checkReplication(ctx context.Context) error {
 		return errors.Wrapf(err, "get %s password", apiv1alpha1.UserMonitor)
 	}
 
-	db, err := database.NewDatabase(ctx, apiv1alpha1.UserMonitor, monitorPass, podIP, mysql.DefaultAdminPort)
+	params := db.DBParams{
+		User: apiv1alpha1.UserMonitor,
+		Pass: monitorPass,
+		Host: podIP,
+	}
+	db, err := database.NewDatabase(ctx, params)
 	if err != nil {
 		return errors.Wrap(err, "connect to db")
 	}

deploy/cr.yaml

@@ -45,6 +45,9 @@ spec:
 
     size: 3
 
+#    env:
+#    - name: BOOTSTRAP_READ_TIMEOUT
+#      value: "600"
     resources:
       requests:
         memory: 1G

pkg/naming/env.go

@@ -0,0 +1,5 @@
+package naming
+
+const (
+	EnvBootstrapReadTimeout = "BOOTSTRAP_READ_TIMEOUT"
+)