All MySQL DBs limited to max 3 concurrent/idle connections #15 (#931)

* v1.1.0

* WIP: copying AUTO_INCREMENT value to ghost table
Initial commit: towards setting up a test suite

Signed-off-by: Shlomi Noach <[email protected]>

* greping for 'expect_table_structure' content

* Adding simple test for 'expect_table_structure' scenario

* adding tests for AUTO_INCREMENT value after row deletes. Should initially fail

* clear event beforehand

* parsing AUTO_INCREMENT from alter query, reading AUTO_INCREMENT from original table, applying AUTO_INCREMENT value onto ghost table if applicable and user has not specified AUTO_INCREMENT in alter statement

* support GetUint64

Signed-off-by: Shlomi Noach <[email protected]>

* minor update to test

Signed-off-by: Shlomi Noach <[email protected]>

* adding test for user defined AUTO_INCREMENT statement

* Generated column as part of UNIQUE (or PRIMARY) KEY

Signed-off-by: Shlomi Noach <[email protected]>

* skip analysis of generated column data type in unique key

Signed-off-by: Shlomi Noach <[email protected]>

* All MySQL DBs limited to max 3 concurrent/idle connections

Signed-off-by: Shlomi Noach <[email protected]>

Author: shlomi-noach

go/logic/throttler.go

@@ -188,9 +188,12 @@ func (this *Throttler) collectControlReplicasLag() {
 		dbUri := connectionConfig.GetDBUri("information_schema")
 
 		var heartbeatValue string
-		if db, _, err := mysql.GetDB(this.migrationContext.Uuid, dbUri); err != nil {
+		db, _, err := mysql.GetDB(this.migrationContext.Uuid, dbUri)
+		if err != nil {
 			return lag, err
-		} else if err = db.QueryRow(replicationLagQuery).Scan(&heartbeatValue); err != nil {
+		}
+
+		if err := db.QueryRow(replicationLagQuery).Scan(&heartbeatValue); err != nil {
 			return lag, err
 		}
 

go/mysql/utils.go

@@ -18,8 +18,11 @@ import (
 	"github.com/outbrain/golib/sqlutils"
 )
 
-const MaxTableNameLength = 64
-const MaxReplicationPasswordLength = 32
+const (
+	MaxTableNameLength           = 64
+	MaxReplicationPasswordLength = 32
+	MaxDBPoolConnections         = 3
+)
 
 type ReplicationLagResult struct {
 	Key InstanceKey
@@ -39,23 +42,22 @@ func (this *ReplicationLagResult) HasLag() bool {
 var knownDBs map[string]*gosql.DB = make(map[string]*gosql.DB)
 var knownDBsMutex = &sync.Mutex{}
 
-func GetDB(migrationUuid string, mysql_uri string) (*gosql.DB, bool, error) {
+func GetDB(migrationUuid string, mysql_uri string) (db *gosql.DB, exists bool, err error) {
 	cacheKey := migrationUuid + ":" + mysql_uri
 
 	knownDBsMutex.Lock()
-	defer func() {
-		knownDBsMutex.Unlock()
-	}()
-
-	var exists bool
-	if _, exists = knownDBs[cacheKey]; !exists {
-		if db, err := gosql.Open("mysql", mysql_uri); err == nil {
-			knownDBs[cacheKey] = db
-		} else {
-			return db, exists, err
+	defer knownDBsMutex.Unlock()
+
+	if db, exists = knownDBs[cacheKey]; !exists {
+		db, err = gosql.Open("mysql", mysql_uri)
+		if err != nil {
+			return nil, false, err
 		}
+		db.SetMaxOpenConns(MaxDBPoolConnections)
+		db.SetMaxIdleConns(MaxDBPoolConnections)
+		knownDBs[cacheKey] = db
 	}
-	return knownDBs[cacheKey], exists, nil
+	return db, exists, nil
 }
 
 // GetReplicationLagFromSlaveStatus returns replication lag for a given db; via SHOW SLAVE STATUS