<commit_before>/*
Copyright 2017 Google Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

/*
Commands for controlling an external mysql process.

Some commands are issued as exec'd tools, some are handled by connecting via
the mysql protocol.
*/

package mysqlctl

import (
	"bufio"
	"errors"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"bytes"

	"golang.org/x/net/context"
	"vitess.io/vitess/go/mysql"
	"vitess.io/vitess/go/stats"
	"vitess.io/vitess/go/vt/dbconfigs"
	"vitess.io/vitess/go/vt/dbconnpool"
	vtenv "vitess.io/vitess/go/vt/env"
	"vitess.io/vitess/go/vt/hook"
	"vitess.io/vitess/go/vt/log"
	"vitess.io/vitess/go/vt/mysqlctl/mysqlctlclient"
)

var (
	// DisableActiveReparents is a flag to disable active
	// reparents for safety reasons. It is used in three places:
	// 1. in this file to skip registering the commands.
	// 2. in vtctld so it can be exported to the UI (different
	// package, that's why it's exported). That way we can disable
	// menu items there, using features.
	// 3. prevents the vtworker from updating replication topology
	// after restarting replication after a split clone/diff.
	DisableActiveReparents = flag.Bool("disable_active_reparents", false, "if set, do not allow active reparents. Use this to protect a cluster using external reparents.")

	dbaPoolSize    = flag.Int("dba_pool_size", 20, "Size of the connection pool for dba connections")
	dbaIdleTimeout = flag.Duration("dba_idle_timeout", time.Minute, "Idle timeout for dba connections")
	appPoolSize    = flag.Int("app_pool_size", 40, "Size of the connection pool for app connections")
	appIdleTimeout = flag.Duration("app_idle_timeout", time.Minute, "Idle timeout for app connections")

	socketFile        = flag.String("mysqlctl_socket", "", "socket file to use for remote mysqlctl actions (empty for local actions)")
	mycnfTemplateFile = flag.String("mysqlctl_mycnf_template", "", "template file to use for generating the my.cnf file during server init")

	// masterConnectRetry is used in 'SET MASTER' commands
	masterConnectRetry = flag.Duration("master_connect_retry", 10*time.Second, "how long to wait in between slave -> connection attempts. Only precise to the second.")

	dbaMysqlStats      = stats.NewTimings("MysqlDba", "MySQL DBA stats", "operation")
	allprivsMysqlStats = stats.NewTimings("MysqlAllPrivs", "MySQl Stats for all privs", "operation")
	appMysqlStats      = stats.NewTimings("MysqlApp", "MySQL app stats", "operation")
)

// Mysqld is the object that represents a mysqld daemon running on this server.
type Mysqld struct {
	config    *Mycnf
	dbcfgs    *dbconfigs.DBConfigs
	dbaPool   *dbconnpool.ConnectionPool
	appPool   *dbconnpool.ConnectionPool
	tabletDir string

	// mutex protects the fields below.
	mutex         sync.Mutex
	onTermFuncs   []func()
	cancelWaitCmd chan struct{}
}

// NewMysqld creates a Mysqld object based on the provided configuration
// and connection parameters.
func NewMysqld(config *Mycnf, dbcfgs *dbconfigs.DBConfigs, dbconfigsFlags dbconfigs.DBConfigFlag) *Mysqld {
	result := &Mysqld{
		config:    config,
		dbcfgs:    dbcfgs,
		tabletDir: path.Dir(config.DataDir),
	}

	// Create and open the connection pool for dba access.
	if dbconfigs.DbaConfig&dbconfigsFlags != 0 {
		result.dbaPool = dbconnpool.NewConnectionPool("DbaConnPool", *dbaPoolSize, *dbaIdleTimeout)
		result.dbaPool.Open(&dbcfgs.Dba, dbaMysqlStats)
	}

	// Create and open the connection pool for app access.
	if dbconfigs.AppConfig&dbconfigsFlags != 0 {
		result.appPool = dbconnpool.NewConnectionPool("AppConnPool", *appPoolSize, *appIdleTimeout)
		result.appPool.Open(&dbcfgs.App, appMysqlStats)
	}

	return result
}

// Cnf returns the mysql config for the daemon
func (mysqld *Mysqld) Cnf() *Mycnf {
	return mysqld.config
}

// TabletDir returns the main tablet directory.
// It's a method so it can be accessed through the MysqlDaemon interface.
func (mysqld *Mysqld) TabletDir() string {
	return mysqld.tabletDir
}

// RunMysqlUpgrade will run the mysql_upgrade program on the current
// install.  Will be called only when mysqld is running with no
// network and no grant tables.
func (mysqld *Mysqld) RunMysqlUpgrade() error {
	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.RunMysqlUpgrade() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.RunMysqlUpgrade(context.TODO())
	}

	// Find mysql_upgrade. If not there, we do nothing.
	dir, err := vtenv.VtMysqlRoot()
	if err != nil {
		log.Warningf("VT_MYSQL_ROOT not set, skipping mysql_upgrade step: %v", err)
		return nil
	}
	name, err := binaryPath(dir, "mysql_upgrade")
	if err != nil {
		log.Warningf("mysql_upgrade binary not present, skipping it: %v", err)
		return nil
	}

	// Since we started mysql with --skip-grant-tables, we should
	// be able to run mysql_upgrade without any valid user or
	// password. However, mysql_upgrade executes a 'flush
	// privileges' right in the middle, and then subsequent
	// commands fail if we don't use valid credentials. So let's
	// use dba credentials.
	params, err := dbconfigs.WithCredentials(&mysqld.dbcfgs.Dba)
	if err != nil {
		return err
	}
	cnf, err := mysqld.defaultsExtraFile(&params)
	if err != nil {
		return err
	}
	defer os.Remove(cnf)

	// Run the program, if it fails, we fail.  Note in this
	// moment, mysqld is running with no grant tables on the local
	// socket only, so this doesn't need any user or password.
	args := []string{
		// --defaults-file=* must be the first arg.
		"--defaults-file=" + cnf,
		"--force", // Don't complain if it's already been upgraded.
	}
	cmd := exec.Command(name, args...)
	cmd.Env = []string{os.ExpandEnv("LD_LIBRARY_PATH=$VT_MYSQL_ROOT/lib/mysql")}
	out, err := cmd.CombinedOutput()
	log.Infof("mysql_upgrade output: %s", out)
	return err
}

// Start will start the mysql daemon, either by running the
// 'mysqld_start' hook, or by running mysqld_safe in the background.
// If a mysqlctld address is provided in a flag, Start will run
// remotely.  When waiting for mysqld to start, we will use
// the dba user.
func (mysqld *Mysqld) Start(ctx context.Context, mysqldArgs ...string) error {
	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.Start() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.Start(ctx, mysqldArgs...)
	}

	if err := mysqld.startNoWait(ctx, mysqldArgs...); err != nil {
		return err
	}

	return mysqld.Wait(ctx)
}

// startNoWait is the internal version of Start, and it doesn't wait.
func (mysqld *Mysqld) startNoWait(ctx context.Context, mysqldArgs ...string) error {
	var name string
	ts := fmt.Sprintf("Mysqld.Start(%v)", time.Now().Unix())

	// try the mysqld start hook, if any
	switch hr := hook.NewHook("mysqld_start", mysqldArgs).Execute(); hr.ExitStatus {
	case hook.HOOK_SUCCESS:
		// hook exists and worked, we can keep going
		name = "mysqld_start hook"
	case hook.HOOK_DOES_NOT_EXIST:
		// hook doesn't exist, run mysqld_safe ourselves
		log.Infof("%v: No mysqld_start hook, running mysqld_safe directly", ts)
		dir, err := vtenv.VtMysqlRoot()
		if err != nil {
			return err
		}
		name, err = binaryPath(dir, "mysqld_safe")
		if err != nil {
			log.Warningf("%v: trying to launch mysqld instead", err)
			name, err = binaryPath(dir, "mysqld")
			// If this also fails, return an error.
			if err != nil {
				return err
			}
		}
		arg := []string{
			"--defaults-file=" + mysqld.config.path}
		arg = append(arg, mysqldArgs...)<commit_msg>Warning about mysqld_safe not being found is futile<commit_after>/*
Copyright 2017 Google Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

/*
Commands for controlling an external mysql process.

Some commands are issued as exec'd tools, some are handled by connecting via
the mysql protocol.
*/

package mysqlctl

import (
	"bufio"
	"errors"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"bytes"

	"golang.org/x/net/context"
	"vitess.io/vitess/go/mysql"
	"vitess.io/vitess/go/stats"
	"vitess.io/vitess/go/vt/dbconfigs"
	"vitess.io/vitess/go/vt/dbconnpool"
	vtenv "vitess.io/vitess/go/vt/env"
	"vitess.io/vitess/go/vt/hook"
	"vitess.io/vitess/go/vt/log"
	"vitess.io/vitess/go/vt/mysqlctl/mysqlctlclient"
)

var (
	// DisableActiveReparents is a flag to disable active
	// reparents for safety reasons. It is used in three places:
	// 1. in this file to skip registering the commands.
	// 2. in vtctld so it can be exported to the UI (different
	// package, that's why it's exported). That way we can disable
	// menu items there, using features.
	// 3. prevents the vtworker from updating replication topology
	// after restarting replication after a split clone/diff.
	DisableActiveReparents = flag.Bool("disable_active_reparents", false, "if set, do not allow active reparents. Use this to protect a cluster using external reparents.")

	dbaPoolSize    = flag.Int("dba_pool_size", 20, "Size of the connection pool for dba connections")
	dbaIdleTimeout = flag.Duration("dba_idle_timeout", time.Minute, "Idle timeout for dba connections")
	appPoolSize    = flag.Int("app_pool_size", 40, "Size of the connection pool for app connections")
	appIdleTimeout = flag.Duration("app_idle_timeout", time.Minute, "Idle timeout for app connections")

	socketFile        = flag.String("mysqlctl_socket", "", "socket file to use for remote mysqlctl actions (empty for local actions)")
	mycnfTemplateFile = flag.String("mysqlctl_mycnf_template", "", "template file to use for generating the my.cnf file during server init")

	// masterConnectRetry is used in 'SET MASTER' commands
	masterConnectRetry = flag.Duration("master_connect_retry", 10*time.Second, "how long to wait in between slave -> connection attempts. Only precise to the second.")

	dbaMysqlStats      = stats.NewTimings("MysqlDba", "MySQL DBA stats", "operation")
	allprivsMysqlStats = stats.NewTimings("MysqlAllPrivs", "MySQl Stats for all privs", "operation")
	appMysqlStats      = stats.NewTimings("MysqlApp", "MySQL app stats", "operation")
)

// Mysqld is the object that represents a mysqld daemon running on this server.
type Mysqld struct {
	config    *Mycnf
	dbcfgs    *dbconfigs.DBConfigs
	dbaPool   *dbconnpool.ConnectionPool
	appPool   *dbconnpool.ConnectionPool
	tabletDir string

	// mutex protects the fields below.
	mutex         sync.Mutex
	onTermFuncs   []func()
	cancelWaitCmd chan struct{}
}

// NewMysqld creates a Mysqld object based on the provided configuration
// and connection parameters.
func NewMysqld(config *Mycnf, dbcfgs *dbconfigs.DBConfigs, dbconfigsFlags dbconfigs.DBConfigFlag) *Mysqld {
	result := &Mysqld{
		config:    config,
		dbcfgs:    dbcfgs,
		tabletDir: path.Dir(config.DataDir),
	}

	// Create and open the connection pool for dba access.
	if dbconfigs.DbaConfig&dbconfigsFlags != 0 {
		result.dbaPool = dbconnpool.NewConnectionPool("DbaConnPool", *dbaPoolSize, *dbaIdleTimeout)
		result.dbaPool.Open(&dbcfgs.Dba, dbaMysqlStats)
	}

	// Create and open the connection pool for app access.
	if dbconfigs.AppConfig&dbconfigsFlags != 0 {
		result.appPool = dbconnpool.NewConnectionPool("AppConnPool", *appPoolSize, *appIdleTimeout)
		result.appPool.Open(&dbcfgs.App, appMysqlStats)
	}

	return result
}

// Cnf returns the mysql config for the daemon
func (mysqld *Mysqld) Cnf() *Mycnf {
	return mysqld.config
}

// TabletDir returns the main tablet directory.
// It's a method so it can be accessed through the MysqlDaemon interface.
func (mysqld *Mysqld) TabletDir() string {
	return mysqld.tabletDir
}

// RunMysqlUpgrade will run the mysql_upgrade program on the current
// install.  Will be called only when mysqld is running with no
// network and no grant tables.
func (mysqld *Mysqld) RunMysqlUpgrade() error {
	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.RunMysqlUpgrade() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.RunMysqlUpgrade(context.TODO())
	}

	// Find mysql_upgrade. If not there, we do nothing.
	dir, err := vtenv.VtMysqlRoot()
	if err != nil {
		log.Warningf("VT_MYSQL_ROOT not set, skipping mysql_upgrade step: %v", err)
		return nil
	}
	name, err := binaryPath(dir, "mysql_upgrade")
	if err != nil {
		log.Warningf("mysql_upgrade binary not present, skipping it: %v", err)
		return nil
	}

	// Since we started mysql with --skip-grant-tables, we should
	// be able to run mysql_upgrade without any valid user or
	// password. However, mysql_upgrade executes a 'flush
	// privileges' right in the middle, and then subsequent
	// commands fail if we don't use valid credentials. So let's
	// use dba credentials.
	params, err := dbconfigs.WithCredentials(&mysqld.dbcfgs.Dba)
	if err != nil {
		return err
	}
	cnf, err := mysqld.defaultsExtraFile(&params)
	if err != nil {
		return err
	}
	defer os.Remove(cnf)

	// Run the program, if it fails, we fail.  Note in this
	// moment, mysqld is running with no grant tables on the local
	// socket only, so this doesn't need any user or password.
	args := []string{
		// --defaults-file=* must be the first arg.
		"--defaults-file=" + cnf,
		"--force", // Don't complain if it's already been upgraded.
	}
	cmd := exec.Command(name, args...)
	cmd.Env = []string{os.ExpandEnv("LD_LIBRARY_PATH=$VT_MYSQL_ROOT/lib/mysql")}
	out, err := cmd.CombinedOutput()
	log.Infof("mysql_upgrade output: %s", out)
	return err
}

// Start will start the mysql daemon, either by running the
// 'mysqld_start' hook, or by running mysqld_safe in the background.
// If a mysqlctld address is provided in a flag, Start will run
// remotely.  When waiting for mysqld to start, we will use
// the dba user.
func (mysqld *Mysqld) Start(ctx context.Context, mysqldArgs ...string) error {
	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.Start() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.Start(ctx, mysqldArgs...)
	}

	if err := mysqld.startNoWait(ctx, mysqldArgs...); err != nil {
		return err
	}

	return mysqld.Wait(ctx)
}

// startNoWait is the internal version of Start, and it doesn't wait.
func (mysqld *Mysqld) startNoWait(ctx context.Context, mysqldArgs ...string) error {
	var name string
	ts := fmt.Sprintf("Mysqld.Start(%v)", time.Now().Unix())

	// try the mysqld start hook, if any
	switch hr := hook.NewHook("mysqld_start", mysqldArgs).Execute(); hr.ExitStatus {
	case hook.HOOK_SUCCESS:
		// hook exists and worked, we can keep going
		name = "mysqld_start hook"
	case hook.HOOK_DOES_NOT_EXIST:
		// hook doesn't exist, run mysqld_safe ourselves
		log.Infof("%v: No mysqld_start hook, running mysqld_safe directly", ts)
		dir, err := vtenv.VtMysqlRoot()
		if err != nil {
			return err
		}
		name, err = binaryPath(dir, "mysqld_safe")
		if err != nil {
			// The movement to use systemd means that mysqld_safe is not always provided.
			// This should not be considered an issue do do not generate a warning.
			log.Infof("%v: trying to launch mysqld instead", err)
			name, err = binaryPath(dir, "mysqld")
			// If this also fails, return an error.
			if err != nil {
				return err
			}
		}
		arg := []string{
			"--defaults-file=" + mysqld.config.path}
		arg = append(arg, mysqldArgs...)
{'fraction_non_alphanumeric': 0.11040916337013638, 'fraction_numerical': 0.002243608667414536, 'mean_word_length': 3.9569798068481123, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '84589', 'n_tokens_mistral': 5945, 'n_tokens_neox': 5577, 'n_words': 1867}