Merge branch 'master' into dep-updates-dec2023

Author: meiji163

.github/CODEOWNERS

@@ -1 +1 @@
-* @dm-2 @rashiq @meiji163 @timvaillancourt
+* @rashiq @meiji163 @timvaillancourt

doc/command-line-flags.md

@@ -261,6 +261,9 @@ But RocksDB currently lacks a few features support compared to InnoDB:
 
 When `--storage-engine=rocksdb`, `gh-ost` will make some changes necessary (e.g. sets isolation level to `READ_COMMITTED`) to support RocksDB.
 
+### charset
+The default charset for the database connection is utf8mb4, utf8, latin1. The ability to specify character set and collation is supported, eg: utf8mb4_general_ci,utf8_general_ci,latin1. 
+
 ### test-on-replica
 
 Issue the migration on a replica; do not modify data on master. Useful for validating, testing and benchmarking. See [`testing-on-replica`](testing-on-replica.md)

doc/interactive-commands.md

@@ -17,6 +17,7 @@ Both interfaces may serve at the same time. Both respond to simple text command,
 - `help`: shows a brief list of available commands
 - `status`: returns a detailed status summary of migration progress and configuration
 - `sup`: returns a brief status summary of migration progress
+- `cpu-profile`: returns a base64-encoded [`runtime/pprof`](https://pkg.go.dev/runtime/pprof) CPU profile using a duration, default: `30s`. Comma-separated options `gzip` and/or `block` (blocked profile) may follow the profile duration
 - `coordinates`: returns recent (though not exactly up to date) binary log coordinates of the inspected server
 - `applier`: returns the hostname of the applier
 - `inspector`: returns the hostname of the inspector

go/base/context.go

@@ -305,6 +305,15 @@ func (this *MigrationContext) SetConnectionConfig(storageEngine string) error {
 	return nil
 }
 
+func (this *MigrationContext) SetConnectionCharset(charset string) {
+	if charset == "" {
+		charset = "utf8mb4,utf8,latin1"
+	}
+
+	this.InspectorConnectionConfig.Charset = charset
+	this.ApplierConnectionConfig.Charset = charset
+}
+
 func getSafeTableName(baseName string, suffix string) string {
 	name := fmt.Sprintf("_%s_%s", baseName, suffix)
 	if len(name) <= mysql.MaxTableNameLength {

go/cmd/gh-ost/main.go

@@ -57,6 +57,7 @@ func main() {
 	flag.StringVar(&migrationContext.CliMasterPassword, "master-password", "", "MySQL password on master, if different from that on replica. Requires --assume-master-host")
 	flag.StringVar(&migrationContext.ConfigFile, "conf", "", "Config file")
 	askPass := flag.Bool("ask-pass", false, "prompt for MySQL password")
+	charset := flag.String("charset", "utf8mb4,utf8,latin1", "The default charset for the database connection is utf8mb4, utf8, latin1.")
 
 	flag.BoolVar(&migrationContext.UseTLS, "ssl", false, "Enable SSL encrypted connections to MySQL hosts")
 	flag.StringVar(&migrationContext.TLSCACertificate, "ssl-ca", "", "CA certificate in PEM format for TLS connections to MySQL hosts. Requires --ssl")
@@ -191,6 +192,8 @@ func main() {
 		migrationContext.Log.Fatale(err)
 	}
 
+	migrationContext.SetConnectionCharset(*charset)
+
 	if migrationContext.AlterStatement == "" {
 		log.Fatal("--alter must be provided and statement must not be empty")
 	}

go/logic/migrator.go

@@ -1042,7 +1042,14 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 	)
 	w := io.MultiWriter(writers...)
 	fmt.Fprintln(w, status)
-	this.migrationContext.Log.Infof(status)
+
+	// This "hack" is required here because the underlying logging library
+	// github.com/outbrain/golib/log provides two functions Info and Infof; but the arguments of
+	// both these functions are eventually redirected to the same function, which internally calls
+	// fmt.Sprintf. So, the argument of every function called on the DefaultLogger object
+	// migrationContext.Log will eventually pass through fmt.Sprintf, and thus the '%' character
+	// needs to be escaped.
+	this.migrationContext.Log.Info(strings.Replace(status, "%", "%%", 1))
 
 	hooksStatusIntervalSec := this.migrationContext.HooksStatusIntervalSec
 	if hooksStatusIntervalSec > 0 && elapsedSeconds%hooksStatusIntervalSec == 0 {

go/logic/server.go

@@ -7,17 +7,30 @@ package logic
 
 import (
 	"bufio"
+	"bytes"
+	"compress/gzip"
+	"encoding/base64"
+	"errors"
 	"fmt"
 	"io"
 	"net"
 	"os"
+	"runtime"
+	"runtime/pprof"
 	"strconv"
 	"strings"
 	"sync/atomic"
+	"time"
 
 	"github.com/github/gh-ost/go/base"
 )
 
+var (
+	ErrCPUProfilingBadOption  = errors.New("unrecognized cpu profiling option")
+	ErrCPUProfilingInProgress = errors.New("cpu profiling already in progress")
+	defaultCPUProfileDuration = time.Second * 30
+)
+
 type printStatusFunc func(PrintStatusRule, io.Writer)
 
 // Server listens for requests on a socket file or via TCP
@@ -27,6 +40,7 @@ type Server struct {
 	tcpListener      net.Listener
 	hooksExecutor    *HooksExecutor
 	printStatus      printStatusFunc
+	isCPUProfiling   int64
 }
 
 func NewServer(migrationContext *base.MigrationContext, hooksExecutor *HooksExecutor, printStatus printStatusFunc) *Server {
@@ -37,6 +51,54 @@ func NewServer(migrationContext *base.MigrationContext, hooksExecutor *HooksExec
 	}
 }
 
+func (this *Server) runCPUProfile(args string) (io.Reader, error) {
+	duration := defaultCPUProfileDuration
+
+	var err error
+	var blockProfile, useGzip bool
+	if args != "" {
+		s := strings.Split(args, ",")
+		// a duration string must be the 1st field, if any
+		if duration, err = time.ParseDuration(s[0]); err != nil {
+			return nil, err
+		}
+		for _, arg := range s[1:] {
+			switch arg {
+			case "block", "blocked", "blocking":
+				blockProfile = true
+			case "gzip":
+				useGzip = true
+			default:
+				return nil, ErrCPUProfilingBadOption
+			}
+		}
+	}
+
+	if atomic.LoadInt64(&this.isCPUProfiling) > 0 {
+		return nil, ErrCPUProfilingInProgress
+	}
+	atomic.StoreInt64(&this.isCPUProfiling, 1)
+	defer atomic.StoreInt64(&this.isCPUProfiling, 0)
+
+	var buf bytes.Buffer
+	var writer io.Writer = &buf
+	if blockProfile {
+		runtime.SetBlockProfileRate(1)
+		defer runtime.SetBlockProfileRate(0)
+	}
+	if useGzip {
+		writer = gzip.NewWriter(writer)
+	}
+	if err = pprof.StartCPUProfile(writer); err != nil {
+		return nil, err
+	}
+
+	time.Sleep(duration)
+	pprof.StopCPUProfile()
+	this.migrationContext.Log.Infof("Captured %d byte runtime/pprof CPU profile (gzip=%v)", buf.Len(), useGzip)
+	return &buf, nil
+}
+
 func (this *Server) BindSocketFile() (err error) {
 	if this.migrationContext.ServeSocketFile == "" {
 		return nil
@@ -144,6 +206,7 @@ func (this *Server) applyServerCommand(command string, writer *bufio.Writer) (pr
 			fmt.Fprint(writer, `available commands:
 status                               # Print a detailed status message
 sup                                  # Print a short status message
+cpu-profile=<options>                # Print a base64-encoded runtime/pprof CPU profile using a duration, default: 30s. Comma-separated options 'gzip' and/or 'block' (blocked profile) may follow the profile duration
 coordinates                          # Print the currently inspected coordinates
 applier                              # Print the hostname of the applier
 inspector                            # Print the hostname of the inspector
@@ -169,6 +232,12 @@ help                                 # This message
 		return ForcePrintStatusOnlyRule, nil
 	case "info", "status":
 		return ForcePrintStatusAndHintRule, nil
+	case "cpu-profile":
+		cpuProfile, err := this.runCPUProfile(arg)
+		if err == nil {
+			fmt.Fprint(base64.NewEncoder(base64.StdEncoding, writer), cpuProfile)
+		}
+		return NoPrintStatusRule, err
 	case "coordinates":
 		{
 			if argIsQuestion || arg == "" {

go/logic/server_test.go

@@ -0,0 +1,68 @@
+package logic
+
+import (
+	"testing"
+	"time"
+
+	"github.com/github/gh-ost/go/base"
+	"github.com/openark/golib/tests"
+)
+
+func TestServerRunCPUProfile(t *testing.T) {
+	t.Parallel()
+
+	t.Run("failed already running", func(t *testing.T) {
+		s := &Server{isCPUProfiling: 1}
+		profile, err := s.runCPUProfile("15ms")
+		tests.S(t).ExpectEquals(err, ErrCPUProfilingInProgress)
+		tests.S(t).ExpectEquals(profile, nil)
+	})
+
+	t.Run("failed bad duration", func(t *testing.T) {
+		s := &Server{isCPUProfiling: 0}
+		profile, err := s.runCPUProfile("should-fail")
+		tests.S(t).ExpectNotNil(err)
+		tests.S(t).ExpectEquals(profile, nil)
+	})
+
+	t.Run("failed bad option", func(t *testing.T) {
+		s := &Server{isCPUProfiling: 0}
+		profile, err := s.runCPUProfile("10ms,badoption")
+		tests.S(t).ExpectEquals(err, ErrCPUProfilingBadOption)
+		tests.S(t).ExpectEquals(profile, nil)
+	})
+
+	t.Run("success", func(t *testing.T) {
+		s := &Server{
+			isCPUProfiling:   0,
+			migrationContext: base.NewMigrationContext(),
+		}
+		defaultCPUProfileDuration = time.Millisecond * 10
+		profile, err := s.runCPUProfile("")
+		tests.S(t).ExpectNil(err)
+		tests.S(t).ExpectNotEquals(profile, nil)
+		tests.S(t).ExpectEquals(s.isCPUProfiling, int64(0))
+	})
+
+	t.Run("success with block", func(t *testing.T) {
+		s := &Server{
+			isCPUProfiling:   0,
+			migrationContext: base.NewMigrationContext(),
+		}
+		profile, err := s.runCPUProfile("10ms,block")
+		tests.S(t).ExpectNil(err)
+		tests.S(t).ExpectNotEquals(profile, nil)
+		tests.S(t).ExpectEquals(s.isCPUProfiling, int64(0))
+	})
+
+	t.Run("success with block and gzip", func(t *testing.T) {
+		s := &Server{
+			isCPUProfiling:   0,
+			migrationContext: base.NewMigrationContext(),
+		}
+		profile, err := s.runCPUProfile("10ms,block,gzip")
+		tests.S(t).ExpectNil(err)
+		tests.S(t).ExpectNotEquals(profile, nil)
+		tests.S(t).ExpectEquals(s.isCPUProfiling, int64(0))
+	})
+}

go/logic/streamer.go

@@ -193,7 +193,7 @@ func (this *EventsStreamer) StreamEvents(canStopStreaming func() bool) error {
 			} else {
 				successiveFailures = 0
 			}
-			if successiveFailures > this.migrationContext.MaxRetries() {
+			if successiveFailures >= this.migrationContext.MaxRetries() {
 				return fmt.Errorf("%d successive failures in streamer reconnect at coordinates %+v", successiveFailures, this.GetReconnectBinlogCoordinates())
 			}
 

go/mysql/connection.go

@@ -30,6 +30,7 @@ type ConnectionConfig struct {
 	tlsConfig            *tls.Config
 	Timeout              float64
 	TransactionIsolation string
+	Charset              string
 }
 
 func NewConnectionConfig() *ConnectionConfig {
@@ -49,6 +50,7 @@ func (this *ConnectionConfig) DuplicateCredentials(key InstanceKey) *ConnectionC
 		tlsConfig:            this.tlsConfig,
 		Timeout:              this.Timeout,
 		TransactionIsolation: this.TransactionIsolation,
+		Charset:              this.Charset,
 	}
 	config.ImpliedKey = &config.Key
 	return config
@@ -122,10 +124,15 @@ func (this *ConnectionConfig) GetDBUri(databaseName string) string {
 	if this.tlsConfig != nil {
 		tlsOption = TLS_CONFIG_KEY
 	}
+
+	if this.Charset == "" {
+		this.Charset = "utf8mb4,utf8,latin1"
+	}
+
 	connectionParams := []string{
 		"autocommit=true",
-		"charset=utf8mb4,utf8,latin1",
 		"interpolateParams=true",
+		fmt.Sprintf("charset=%s", this.Charset),
 		fmt.Sprintf("tls=%s", tlsOption),
 		fmt.Sprintf("transaction_isolation=%q", this.TransactionIsolation),
 		fmt.Sprintf("timeout=%fs", this.Timeout),

go/mysql/connection_test.go

@@ -30,6 +30,7 @@ func TestNewConnectionConfig(t *testing.T) {
 	test.S(t).ExpectEquals(c.User, "")
 	test.S(t).ExpectEquals(c.Password, "")
 	test.S(t).ExpectEquals(c.TransactionIsolation, "")
+	test.S(t).ExpectEquals(c.Charset, "")
 }
 
 func TestDuplicateCredentials(t *testing.T) {
@@ -42,6 +43,7 @@ func TestDuplicateCredentials(t *testing.T) {
 		ServerName:         "feathers",
 	}
 	c.TransactionIsolation = transactionIsolation
+	c.Charset = "utf8mb4"
 
 	dup := c.DuplicateCredentials(InstanceKey{Hostname: "otherhost", Port: 3310})
 	test.S(t).ExpectEquals(dup.Key.Hostname, "otherhost")
@@ -52,6 +54,7 @@ func TestDuplicateCredentials(t *testing.T) {
 	test.S(t).ExpectEquals(dup.Password, "penguin")
 	test.S(t).ExpectEquals(dup.tlsConfig, c.tlsConfig)
 	test.S(t).ExpectEquals(dup.TransactionIsolation, c.TransactionIsolation)
+	test.S(t).ExpectEquals(dup.Charset, c.Charset)
 }
 
 func TestDuplicate(t *testing.T) {
@@ -60,6 +63,7 @@ func TestDuplicate(t *testing.T) {
 	c.User = "gromit"
 	c.Password = "penguin"
 	c.TransactionIsolation = transactionIsolation
+	c.Charset = "utf8mb4"
 
 	dup := c.Duplicate()
 	test.S(t).ExpectEquals(dup.Key.Hostname, "myhost")
@@ -69,6 +73,7 @@ func TestDuplicate(t *testing.T) {
 	test.S(t).ExpectEquals(dup.User, "gromit")
 	test.S(t).ExpectEquals(dup.Password, "penguin")
 	test.S(t).ExpectEquals(dup.TransactionIsolation, transactionIsolation)
+	test.S(t).ExpectEquals(dup.Charset, "utf8mb4")
 }
 
 func TestGetDBUri(t *testing.T) {
@@ -78,9 +83,10 @@ func TestGetDBUri(t *testing.T) {
 	c.Password = "penguin"
 	c.Timeout = 1.2345
 	c.TransactionIsolation = transactionIsolation
+	c.Charset = "utf8mb4,utf8,latin1"
 
 	uri := c.GetDBUri("test")
-	test.S(t).ExpectEquals(uri, `gromit:penguin@tcp(myhost:3306)/test?autocommit=true&charset=utf8mb4,utf8,latin1&interpolateParams=true&tls=false&transaction_isolation="REPEATABLE-READ"&timeout=1.234500s&readTimeout=1.234500s&writeTimeout=1.234500s`)
+	test.S(t).ExpectEquals(uri, `gromit:penguin@tcp(myhost:3306)/test?autocommit=true&interpolateParams=true&charset=utf8mb4,utf8,latin1&tls=false&transaction_isolation="REPEATABLE-READ"&timeout=1.234500s&readTimeout=1.234500s&writeTimeout=1.234500s`)
 }
 
 func TestGetDBUriWithTLSSetup(t *testing.T) {
@@ -91,7 +97,8 @@ func TestGetDBUriWithTLSSetup(t *testing.T) {
 	c.Timeout = 1.2345
 	c.tlsConfig = &tls.Config{}
 	c.TransactionIsolation = transactionIsolation
+	c.Charset = "utf8mb4_general_ci,utf8_general_ci,latin1"
 
 	uri := c.GetDBUri("test")
-	test.S(t).ExpectEquals(uri, `gromit:penguin@tcp(myhost:3306)/test?autocommit=true&charset=utf8mb4,utf8,latin1&interpolateParams=true&tls=ghost&transaction_isolation="REPEATABLE-READ"&timeout=1.234500s&readTimeout=1.234500s&writeTimeout=1.234500s`)
+	test.S(t).ExpectEquals(uri, `gromit:penguin@tcp(myhost:3306)/test?autocommit=true&interpolateParams=true&charset=utf8mb4_general_ci,utf8_general_ci,latin1&tls=ghost&transaction_isolation="REPEATABLE-READ"&timeout=1.234500s&readTimeout=1.234500s&writeTimeout=1.234500s`)
 }