add support for dumping large database test

Signed-off-by: Avi Deitcher <avi@deitcher.net>

Author: deitch

pkg/core/dump.go

@@ -111,6 +111,7 @@ func (e *Executor) Dump(ctx context.Context, opts DumpOptions) (DumpResults, err
 		Routines:            routines,
 		SuppressUseDatabase: suppressUseDatabase,
 		MaxAllowedPacket:    maxAllowedPacket,
+		PostDumpDelay:       opts.PostDumpDelay,
 	}, dw); err != nil {
 		dbDumpSpan.SetStatus(codes.Error, err.Error())
 		dbDumpSpan.End()

pkg/core/dumpoptions.go

@@ -1,6 +1,8 @@
 package core
 
 import (
+	"time"
+
 	"github.com/databacker/mysql-backup/pkg/compression"
 	"github.com/databacker/mysql-backup/pkg/database"
 	"github.com/databacker/mysql-backup/pkg/encrypt"
@@ -25,4 +27,6 @@ type DumpOptions struct {
 	MaxAllowedPacket    int
 	Run                 uuid.UUID
 	FilenamePattern     string
+	// PostDumpDelay inafter each dump is complete, while holding connection open. Do not use outside of tests.
+	PostDumpDelay time.Duration
 }

pkg/database/connection.go

@@ -8,10 +8,11 @@ import (
 )
 
 type Connection struct {
-	User string
-	Pass string
-	Host string
-	Port int
+	User            string
+	Pass            string
+	Host            string
+	Port            int
+	MultiStatements bool
 }
 
 func (c Connection) MySQL() string {
@@ -27,5 +28,6 @@ func (c Connection) MySQL() string {
 	}
 	config.ParseTime = true
 	config.TLSConfig = "preferred"
+	config.MultiStatements = c.MultiStatements
 	return config.FormatDSN()
 }

pkg/database/dump.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"time"
 
 	"github.com/databacker/mysql-backup/pkg/database/mysql"
 )
@@ -14,6 +15,8 @@ type DumpOpts struct {
 	Routines            bool
 	SuppressUseDatabase bool
 	MaxAllowedPacket    int
+	// PostDumpDelay after each dump is complete, while holding connection open. Do not use outside of tests.
+	PostDumpDelay time.Duration
 }
 
 func Dump(ctx context.Context, dbconn Connection, opts DumpOpts, writers []DumpWriter) error {
@@ -42,6 +45,7 @@ func Dump(ctx context.Context, dbconn Connection, opts DumpOpts, writers []DumpW
 				Routines:            opts.Routines,
 				SuppressUseDatabase: opts.SuppressUseDatabase,
 				MaxAllowedPacket:    opts.MaxAllowedPacket,
+				PostDumpDelay:       opts.PostDumpDelay,
 			}
 			if err := dumper.Dump(); err != nil {
 				return fmt.Errorf("failed to dump database %s: %v", schema, err)

pkg/database/mysql/dump.go

@@ -52,6 +52,7 @@ type Data struct {
 	SuppressUseDatabase bool
 	Charset             string
 	Collation           string
+	PostDumpDelay       time.Duration
 
 	tx                 *sql.Tx
 	headerTmpl         *template.Template
@@ -254,6 +255,9 @@ func (data *Data) Dump() error {
 		return data.err
 	}
 
+	if data.PostDumpDelay > 0 {
+		time.Sleep(data.PostDumpDelay)
+	}
 	meta.CompleteTime = time.Now().UTC().Format("2006-01-02 15:04:05")
 	return data.footerTmpl.Execute(data.Out, meta)
 }

test/README.md

@@ -71,3 +71,11 @@ base/ - base of the backup area
 	backups/ - the directory where backups are stored
 		15674832/ - one target's backup
 		88725436/ - another target's backup
+
+## Databases in Parallel
+
+The behaviour of backing up multiple databases in parallel can have an impact on connections.
+
+To test management, there is a database test `TestIntegration/parallel_databases`. It sets up a database server
+with multiples databases, each with a table of a MB or a few. It then injects a delay on each backup of 10
+seconds, and then exits. Every second during the backup, it reports the number of connections.

test/backup_test.go

@@ -31,7 +31,6 @@ import (
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
-	imagetypes "github.com/docker/docker/api/types/image"
 	"github.com/docker/docker/client"
 	"github.com/docker/docker/pkg/stdcopy"
 	"github.com/docker/go-connections/nat"
@@ -546,44 +545,12 @@ func setup(dc *dockerContext, base, backupFile, compactBackupFile string) (mysql
 	s3server := httptest.NewServer(s3.Server())
 	s3url = s3server.URL
 
-	// start the mysql container; configure it for lots of debug logging, in case we need it
-	mysqlConf := `
-[mysqld]
-log_error       =/var/log/mysql/mysql_error.log
-general_log_file=/var/log/mysql/mysql.log
-general_log     =1
-slow_query_log  =1
-slow_query_log_file=/var/log/mysql/mysql_slow.log
-long_query_time =2
-log_queries_not_using_indexes = 1
-`
-	confFile := filepath.Join(base, "log.cnf")
-	if err := os.WriteFile(confFile, []byte(mysqlConf), 0644); err != nil {
-		return mysql, smb, s3url, s3backend, fmt.Errorf("failed to write mysql config file: %v", err)
-	}
-	logDir := filepath.Join(base, "mysql_logs")
-	if err := os.Mkdir(logDir, 0755); err != nil {
-		return mysql, smb, s3url, s3backend, fmt.Errorf("failed to create mysql log directory: %v", err)
-	}
-	// ensure we have mysql image
-	resp, err := dc.cli.ImagePull(context.Background(), mysqlImage, imagetypes.PullOptions{})
-	if err != nil {
-		return mysql, smb, s3url, s3backend, fmt.Errorf("failed to pull mysql image: %v", err)
-	}
-	_, _ = io.Copy(os.Stdout, resp)
-	_ = resp.Close()
-	mysqlCID, mysqlPort, err := dc.startContainer(mysqlImage, "mysql", "3306/tcp", []string{fmt.Sprintf("%s:/etc/mysql/conf.d/log.conf:ro", confFile), fmt.Sprintf("%s:/var/log/mysql", logDir)}, nil, []string{
-		fmt.Sprintf("MYSQL_ROOT_PASSWORD=%s", mysqlRootPass),
-		"MYSQL_DATABASE=tester",
-		fmt.Sprintf("MYSQL_USER=%s", mysqlUser),
-		fmt.Sprintf("MYSQL_PASSWORD=%s", mysqlPass),
-	})
+	mysql, err = startDatabase(dc, base, mysqlImage, "mysql")
 	if err != nil {
-		return
+		return mysql, smb, s3url, s3backend, fmt.Errorf("failed to start mysql container: %v", err)
 	}
-	mysql = containerPort{name: "mysql", id: mysqlCID, port: mysqlPort}
 
-	if err = dc.waitForDBConnectionAndGrantPrivileges(mysqlCID, mysqlRootUser, mysqlRootPass); err != nil {
+	if err = dc.waitForDBConnectionAndGrantPrivileges(mysql.id, mysqlRootUser, mysqlRootPass); err != nil {
 		return
 	}
 
@@ -890,58 +857,119 @@ func populatePrePost(base string, targets []backupTarget) (err error) {
 	return nil
 }
 
-func startDatabase(dc *dockerContext, baseDir, image, name string) (containerPort, error) {
-	resp, err := dc.cli.ImagePull(context.Background(), image, imagetypes.PullOptions{})
-	if err != nil {
-		return containerPort{}, fmt.Errorf("failed to pull mysql image: %v", err)
-	}
-	_, _ = io.Copy(os.Stdout, resp)
-	_ = resp.Close()
-
-	// start the mysql container; configure it for lots of debug logging, in case we need it
-	mysqlConf := `
-[mysqld]
-log_error       =/var/log/mysql/mysql_error.log
-general_log_file=/var/log/mysql/mysql.log
-general_log     =1
-slow_query_log  =1
-slow_query_log_file=/var/log/mysql/mysql_slow.log
-long_query_time =2
-log_queries_not_using_indexes = 1
-`
-	if err := os.Mkdir(baseDir, 0o755); err != nil {
-		return containerPort{}, fmt.Errorf("failed to create mysql base directory: %v", err)
-	}
-	confFile := filepath.Join(baseDir, "log.cnf")
-	if err := os.WriteFile(confFile, []byte(mysqlConf), 0644); err != nil {
-		return containerPort{}, fmt.Errorf("failed to write mysql config file: %v", err)
-	}
-	logDir := filepath.Join(baseDir, "mysql_logs")
-	if err := os.Mkdir(logDir, 0755); err != nil {
-		return containerPort{}, fmt.Errorf("failed to create mysql log directory: %v", err)
-	}
-
-	// start mysql
-	cid, port, err := dc.startContainer(
-		image, name, "3306/tcp", []string{fmt.Sprintf("%s:/etc/mysql/conf.d/log.conf:ro", confFile), fmt.Sprintf("%s:/var/log/mysql", logDir)}, nil, []string{
-			fmt.Sprintf("MYSQL_ROOT_PASSWORD=%s", mysqlRootPass),
-			"MYSQL_DATABASE=tester",
-			fmt.Sprintf("MYSQL_USER=%s", mysqlUser),
-			fmt.Sprintf("MYSQL_PASSWORD=%s", mysqlPass),
-		})
-	if err != nil {
-		return containerPort{}, fmt.Errorf("failed to start mysql container: %v", err)
-	}
-	return containerPort{name: name, id: cid, port: port}, nil
-}
-
 func TestIntegration(t *testing.T) {
 	CheckSkipIntegration(t, "integration")
 	syscall.Umask(0)
 	dc, err := getDockerContext()
 	if err != nil {
 		t.Fatalf("failed to get docker client: %v", err)
 	}
+	t.Run("parallel databases", func(t *testing.T) {
+		base := t.TempDir()
+		mysql, err := startDatabase(dc, base, mysqlImage, "mysql-parallel")
+		defer func() {
+			// log the results before tearing down, if requested
+			if err := logContainers(dc, mysql.id); err != nil {
+				log.Errorf("failed to get logs from service containers: %v", err)
+			}
+
+			// tear everything down
+			if err := teardown(dc, mysql.id); err != nil {
+				log.Errorf("failed to teardown test: %v", err)
+			}
+		}()
+
+		if err != nil {
+			t.Fatalf("failed to start large mysql database: %v", err)
+		}
+		if err = dc.waitForDBConnectionAndGrantPrivileges(mysql.id, mysqlRootUser, mysqlRootPass); err != nil {
+			t.Fatalf("failed to wait for DB connection: %v", err)
+		}
+		dbconn := database.Connection{
+			User:            mysqlRootUser,
+			Pass:            mysqlRootPass,
+			Host:            "localhost",
+			Port:            mysql.port,
+			MultiStatements: true,
+		}
+
+		db, err := sql.Open("mysql", dbconn.MySQL())
+		if err != nil {
+			t.Fatalf("failed to open connection to database: %v", err)
+		}
+
+		var dbCount uint = 10
+		var tableSize uint64 = 1
+		t.Logf("Setting up database server with %d databases, with one table of %d MB each", dbCount, tableSize)
+		if err := setupLargeDatabase(db, dbCount, tableSize*1024*1024); err != nil {
+			t.Fatalf("failed to setup large database: %v", err)
+		}
+		// now just run a backup to a temporary directory
+		t.Logf("Running backup for large database")
+		backupDir := t.TempDir()
+		executor := &core.Executor{}
+		executor.SetLogger(log.New())
+
+		store, err := storage.ParseURL(backupDir, credentials.Creds{})
+		if err != nil {
+			t.Fatalf("invalid target url: %v", err)
+		}
+
+		dumpOptions := core.DumpOptions{
+			Compressor: &compression.GzipCompressor{},
+			DBConn: database.Connection{
+				User: mysqlRootUser,
+				Pass: mysqlRootPass,
+				Host: "localhost",
+				Port: mysql.port,
+			},
+			Targets:       []storage.Storage{store},
+			PostDumpDelay: 5 * time.Second, // for testing only, make them delay 10 seconds
+		}
+		ctx := context.Background()
+		start := time.Now()
+		errChan := make(chan error, 1)
+		t.Logf("Starting dump test for large database at %s", start)
+		go func() {
+			_, err := executor.Dump(ctx, dumpOptions)
+			errChan <- err
+		}()
+		ticker := time.NewTicker(1 * time.Second)
+		defer ticker.Stop()
+
+	loop:
+		for {
+			select {
+			case err := <-errChan:
+				if err != nil {
+					t.Fatalf("failed to run dump test: %v", err)
+				}
+				ticker.Stop()
+				break loop
+			case <-ticker.C:
+				// every interval, report how many connections are running to the database
+				tr, err := getStatus(db, "Threads_running")
+				if err != nil {
+					t.Fatalf("failed to get Threads_running status: %v", err)
+				}
+
+				tc, err := getStatus(db, "Threads_connected")
+				if err != nil {
+					t.Fatalf("failed to get Threads_connected status: %v", err)
+				}
+				uTotal, uActive, err := getProcesslistCounts(db)
+				if err != nil {
+					t.Fatalf("failed to get processlist counts: %v", err)
+				}
+
+				t.Logf("[%s]\tthreads_running=%d\tthreads_connected=%d\topen_user=%d\tactive_user=%d\n",
+					time.Now().Format("15:04:05"),
+					tr, tc, uTotal, uActive)
+
+			}
+		}
+		t.Logf("Dump completed at %s in %s", time.Now(), time.Since(start))
+	})
 	t.Run("dump", func(t *testing.T) {
 		var (
 			err        error