[WIP] update gorm and related libraries

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/getsentry/sentry-go v0.20.0
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-faker/faker/v4 v4.1.0
-	github.com/go-gormigrate/gormigrate/v2 v2.0.0
+	github.com/go-gormigrate/gormigrate/v2 v2.0.2
 	github.com/go-playground/validator/v10 v10.12.0
 	github.com/go-resty/resty/v2 v2.7.0
 	github.com/goava/di v1.11.1
@@ -29,7 +29,6 @@ require (
 	github.com/lib/pq v1.10.7
 	github.com/libdns/route53 v1.3.2
 	github.com/looplab/fsm v1.0.1
-	github.com/mattn/go-sqlite3 v1.14.3 // indirect
 	github.com/mendsley/gojwk v0.0.0-20141217222730-4d5ec6e58103
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/onsi/gomega v1.27.6
@@ -55,8 +54,8 @@ require (
 	gopkg.in/resty.v1 v1.12.0
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	gorm.io/driver/postgres v1.0.8
-	gorm.io/gorm v1.21.7
+	gorm.io/driver/postgres v1.5.0
+	gorm.io/gorm v1.25.0
 	k8s.io/api v0.25.6
 	k8s.io/apimachinery v0.25.6
 	k8s.io/client-go v0.25.6
@@ -107,16 +106,12 @@ require (
 	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/itchyny/timefmt-go v0.1.5 // indirect
-	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
-	github.com/jackc/pgconn v1.12.0 // indirect
-	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
-	github.com/jackc/pgproto3/v2 v2.3.0 // indirect
-	github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
-	github.com/jackc/pgtype v1.11.0 // indirect
-	github.com/jackc/pgx/v4 v4.16.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/pgx/v4 v4.17.2 // indirect
+	github.com/jackc/pgx/v5 v5.3.1 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
-	github.com/jinzhu/now v1.1.2 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
@@ -139,10 +134,10 @@ require (
 	github.com/sirupsen/logrus v1.9.0 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
-	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.24.0
-	golang.org/x/crypto v0.7.0 // indirect
+	golang.org/x/crypto v0.8.0 // indirect
 	golang.org/x/mod v0.9.0 // indirect
 	golang.org/x/net v0.9.0 // indirect
 	golang.org/x/sys v0.7.0 // indirect

internal/connector/internal/services/connector_cluster.go

@@ -433,7 +433,7 @@ func (k *connectorClusterService) ListConnectorDeployments(ctx context.Context,
 	var resourceList dbapi.ConnectorDeploymentList
 	dbConn := k.connectionFactory.New()
 	// specify preload for annotations only, to avoid skipping deleted connectors
-	dbConn = dbConn.Preload("Annotations").Joins("Status").Joins("ConnectorShardMetadata").Joins("Connector")
+	dbConn = dbConn.Unscoped().Preload("Annotations").Joins("Status").Joins("ConnectorShardMetadata").Joins("Connector")
 
 	pagingMeta := &api.PagingMeta{
 		Page: listArgs.Page,
@@ -505,7 +505,16 @@ func (k *connectorClusterService) UpdateConnectorDeploymentStatus(ctx context.Co
 		return services.HandleGoneError("Connector deployment", "id", deploymentStatus.ID)
 	}
 
-	if err := dbConn.Model(&deploymentStatus).Where("id = ? and version <= ?", deploymentStatus.ID, deploymentStatus.Version).Save(&deploymentStatus).Error; err != nil {
+	// See https://github.com/go-gorm/gorm/issues/6139
+	// We can no longer rely on the Save throwing a constraint violation exception when upserting a record that already exists.
+	// By first checking that the record existences we ensure the Save only updates an existing record vs trying to create a duplicate.
+	existing := dbapi.ConnectorDeploymentStatus{}
+	dbConn.Unscoped().Where("id = ?", deploymentStatus.ID).First(&existing)
+	// If it was not found existing.Version will be zero.
+	if existing.Version > deploymentStatus.Version {
+		return errors.Conflict("failed to update deployment status, probably a stale deployment status version was used: %d", deploymentStatus.Version)
+	}
+	if err := dbConn.Unscoped().Model(&deploymentStatus).Where("id = ? and version <= ?", deploymentStatus.ID, deploymentStatus.Version).Save(&deploymentStatus).Error; err != nil {
 		return errors.Conflict("failed to update deployment status: %s, probably a stale deployment status version was used: %d", err.Error(), deploymentStatus.Version)
 	}
 

internal/kafka/internal/services/clusters_test.go

@@ -662,7 +662,7 @@ func Test_clusterService_Update(t *testing.T) {
 			wantErr: false,
 			want:    nil,
 			setupFn: func() {
-				mocket.Catcher.Reset().NewMock().WithQuery(`UPDATE "clusters" SET "id"=$1,"updated_at"=$2 WHERE "id" = $3`)
+				mocket.Catcher.Reset().NewMock().WithQuery(`UPDATE "clusters" SET "id"=$1,"updated_at"=$2 WHERE "clusters"."deleted_at" IS NULL AND "id" = $3`)
 				mocket.Catcher.NewMock().WithExecException().WithQueryException()
 			},
 		},

internal/kafka/internal/services/kafka_test.go

@@ -1543,7 +1543,7 @@ func Test_kafkaService_RegisterKafkaJob(t *testing.T) {
 				totalCountResponse := []map[string]interface{}{{"count": 0}}
 
 				mocket.Catcher.Reset()
-				mocket.Catcher.NewMock().WithQuery(`SELECT count(1) FROM "kafka_requests" WHERE instance_type = $1 AND owner = $2 AND (organisation_id = $3) AND "kafka_requests"."deleted_at" IS NULL`).
+				mocket.Catcher.NewMock().WithQuery(`SELECT count(*) FROM "kafka_requests" WHERE instance_type = $1 AND owner = $2 AND organisation_id = $3 AND "kafka_requests"."deleted_at" IS NULL`).
 					WithArgs(types.DEVELOPER.String(), testUser, "org-id").
 					WithReply(totalCountResponse)
 				mocket.Catcher.NewMock().
@@ -1619,7 +1619,7 @@ func Test_kafkaService_RegisterKafkaJob(t *testing.T) {
 
 				totalCountResponse := []map[string]interface{}{{"count": 2}}
 
-				mocket.Catcher.NewMock().WithQuery(`SELECT count(1) FROM "kafka_requests" WHERE instance_type = $1 AND owner = $2 AND (organisation_id = $3) AND "kafka_requests"."deleted_at" IS NULL`).
+				mocket.Catcher.NewMock().WithQuery(`SELECT count(*) FROM "kafka_requests" WHERE instance_type = $1 AND owner = $2 AND organisation_id = $3 AND "kafka_requests"."deleted_at" IS NULL`).
 					WithArgs(types.DEVELOPER.String(), testUser, "org-id").
 					WithReply(totalCountResponse)
 				mocket.Catcher.NewMock().WithQueryException().WithExecException()
@@ -2156,7 +2156,7 @@ func Test_kafkaService_List(t *testing.T) {
 
 				// total count query
 				totalCountResponse := []map[string]interface{}{{"count": len(kafkaList)}}
-				mocket.Catcher.NewMock().WithQuery(`SELECT count(1) FROM "kafka_requests"`).WithReply(totalCountResponse)
+				mocket.Catcher.NewMock().WithQuery(`SELECT count(*) FROM "kafka_requests"`).WithReply(totalCountResponse)
 
 				// actual query to return list of kafka requests based on filters
 				query := fmt.Sprintf(`SELECT * FROM "%s"`, kafkaRequestTableName)
@@ -2220,7 +2220,7 @@ func Test_kafkaService_List(t *testing.T) {
 
 				// total count query
 				totalCountResponse := []map[string]interface{}{{"count": len(kafkaList)}}
-				mocket.Catcher.NewMock().WithQuery(`SELECT count(1) FROM "kafka_requests"`).WithReply(totalCountResponse)
+				mocket.Catcher.NewMock().WithQuery(`SELECT count(*) FROM "kafka_requests"`).WithReply(totalCountResponse)
 
 				// actual query to return list of kafka requests based on filters
 				query := fmt.Sprintf(`SELECT * FROM "%s"`, kafkaRequestTableName)
@@ -2270,7 +2270,7 @@ func Test_kafkaService_List(t *testing.T) {
 
 				// total count query
 				totalCountResponse := []map[string]interface{}{{"count": "5"}}
-				mocket.Catcher.NewMock().WithQuery(`SELECT count(1) FROM "kafka_requests"`).WithReply(totalCountResponse)
+				mocket.Catcher.NewMock().WithQuery(`SELECT count(*) FROM "kafka_requests"`).WithReply(totalCountResponse)
 
 				// actual query to return list of kafka requests based on filters
 				query := fmt.Sprintf(`SELECT * FROM "%s"`, kafkaRequestTableName)
@@ -2306,7 +2306,7 @@ func Test_kafkaService_List(t *testing.T) {
 
 				// total count query
 				totalCountResponse := []map[string]interface{}{{"count": len(kafkaList)}}
-				mocket.Catcher.NewMock().WithQuery(`SELECT count(1) FROM "kafka_requests"`).WithReply(totalCountResponse)
+				mocket.Catcher.NewMock().WithQuery(`SELECT count(*) FROM "kafka_requests"`).WithReply(totalCountResponse)
 
 				// actual query to return list of kafka requests based on filters
 				query := fmt.Sprintf(`SELECT * FROM "%s"`, kafkaRequestTableName)

internal/kafka/internal/services/quota/quota_management_list_service_test.go

@@ -394,7 +394,7 @@ func Test_QuotaManagementListReserveQuota(t *testing.T) {
 			setupFn: func() {
 				mocket.Catcher.Reset()
 				mocket.Catcher.NewMock().
-					WithQuery(`SELECT * FROM "kafka_requests" WHERE instance_type = $1 AND (actual_kafka_billing_model = $2 or desired_kafka_billing_model = $3) AND (organisation_id = $4) AND "kafka_requests"."deleted_at" IS NULL`).
+					WithQuery(`SELECT * FROM "kafka_requests" WHERE instance_type = $1 AND (actual_kafka_billing_model = $2 or desired_kafka_billing_model = $3) AND organisation_id = $4 AND "kafka_requests"."deleted_at" IS NULL`).
 					WithArgs(types.STANDARD.String(), "standard", "standard", "org-id").
 					WithReply(converters.ConvertKafkaRequest(buildKafkaRequest(nil)))
 				mocket.Catcher.NewMock().WithExecException().WithQueryException()
@@ -497,7 +497,7 @@ func Test_QuotaManagementListReserveQuota(t *testing.T) {
 			setupFn: func() {
 				mocket.Catcher.Reset()
 				mocket.Catcher.NewMock().
-					WithQuery(`SELECT * FROM "kafka_requests" WHERE instance_type = $1 AND (actual_kafka_billing_model = $2 or desired_kafka_billing_model = $3) AND (organisation_id = $4) AND "kafka_requests"."deleted_at" IS NULL`).
+					WithQuery(`SELECT * FROM "kafka_requests" WHERE instance_type = $1 AND (actual_kafka_billing_model = $2 or desired_kafka_billing_model = $3) AND organisation_id = $4 AND "kafka_requests"."deleted_at" IS NULL`).
 					WithArgs(types.STANDARD.String(), "standard", "standard", "org-id").
 					WithReply(converters.ConvertKafkaRequest(buildKafkaRequest(nil)))
 				mocket.Catcher.NewMock().WithExecException().WithQueryException()

pkg/db/config.go

@@ -26,6 +26,29 @@ type DatabaseConfig struct {
 	NameFile           string `json:"name_file"`
 	UsernameFile       string `json:"username_file"`
 	PasswordFile       string `json:"password_file"`
+
+	EnablePreparedStatements bool
+}
+
+func (d *DatabaseConfig) DeepCopy() *DatabaseConfig {
+	return &DatabaseConfig{
+		DatabaseCaCertFile:       d.DatabaseCaCertFile,
+		Debug:                    d.Debug,
+		Dialect:                  d.Dialect,
+		EnablePreparedStatements: d.EnablePreparedStatements,
+		Host:                     d.Host,
+		HostFile:                 d.HostFile,
+		MaxOpenConnections:       d.MaxOpenConnections,
+		Name:                     d.Name,
+		NameFile:                 d.NameFile,
+		Password:                 d.Password,
+		PasswordFile:             d.PasswordFile,
+		Port:                     d.Port,
+		PortFile:                 d.PortFile,
+		SSLMode:                  d.SSLMode,
+		Username:                 d.Username,
+		UsernameFile:             d.UsernameFile,
+	}
 }
 
 func NewDatabaseConfig() *DatabaseConfig {
@@ -35,12 +58,13 @@ func NewDatabaseConfig() *DatabaseConfig {
 		Debug:              false,
 		MaxOpenConnections: 50,
 
-		HostFile:           "secrets/db.host",
-		PortFile:           "secrets/db.port",
-		UsernameFile:       "secrets/db.user",
-		PasswordFile:       "secrets/db.password",
-		NameFile:           "secrets/db.name",
-		DatabaseCaCertFile: "secrets/db.ca_cert",
+		HostFile:                 "secrets/db.host",
+		PortFile:                 "secrets/db.port",
+		UsernameFile:             "secrets/db.user",
+		PasswordFile:             "secrets/db.password",
+		NameFile:                 "secrets/db.name",
+		DatabaseCaCertFile:       "secrets/db.ca_cert",
+		EnablePreparedStatements: true,
 	}
 }
 

pkg/db/connection.go

@@ -16,13 +16,6 @@ type ConnectionFactory struct {
 	DB     *gorm.DB
 }
 
-var gormConfig *gorm.Config = &gorm.Config{
-	PrepareStmt:       true,
-	AllowGlobalUpdate: false, // change it to true to allow updates without the WHERE clause
-	QueryFields:       true,
-	Logger:            customLoggerWithMetricsCollector{},
-}
-
 // NewConnectionFactory will initialize a singleton ConnectionFactory as needed and return the same instance.
 // Go includes database connection pooling in the platform. Gorm uses the same and provides a method to
 // clone a connection via New(), which is safe for use by concurrent Goroutines.
@@ -31,12 +24,18 @@ func NewConnectionFactory(config *DatabaseConfig) (*ConnectionFactory, func()) {
 	var err error
 	// refer to https://gorm.io/docs/gorm_config.html
 
-	if config.Dialect == "postgres" {
-		db, err = gorm.Open(postgres.Open(config.ConnectionString()), gormConfig)
-	} else {
+	if config.Dialect != "postgres" {
 		// TODO what other dialects do we support?
 		panic(fmt.Sprintf("unsupported DB dialect: %s", config.Dialect))
 	}
+
+	gormConfig := &gorm.Config{
+		PrepareStmt:       config.EnablePreparedStatements,
+		AllowGlobalUpdate: false, // change it to true to allow updates without the WHERE clause
+		QueryFields:       true,
+		Logger:            customLoggerWithMetricsCollector{},
+	}
+	db, err = gorm.Open(postgres.Open(config.ConnectionString()), gormConfig)
 	if err != nil {
 		panic(fmt.Sprintf(
 			"failed to connect to %s database %s with connection string: %s\nError: %s",
@@ -46,6 +45,7 @@ func NewConnectionFactory(config *DatabaseConfig) (*ConnectionFactory, func()) {
 			err.Error(),
 		))
 	}
+
 	sqlDB, sqlDBErr := db.DB()
 	if sqlDBErr != nil {
 		panic(fmt.Errorf("unexpected connection error: %s", sqlDBErr))

pkg/workers/leader_election_mgr_test.go

@@ -128,7 +128,7 @@ func TestLeaderElectionManager_acquireLeaderLease(t *testing.T) {
 					WithArgs("cluster").
 					WithReply([]map[string]interface{}{mockEntry})
 				mocket.Catcher.NewMock().
-					WithQuery(`UPDATE "leader_leases" SET "expires"=$1,"leader"=$2,"updated_at"=$3 WHERE "id" = $4`)
+					WithQuery(`UPDATE "leader_leases" SET "expires"=$1,"leader"=$2,"updated_at"=$3 WHERE "leader_leases"."deleted_at" IS NULL AND "id" = $4`)
 				mocket.Catcher.NewMock().WithQueryException().WithExecException()
 			},
 		},