// Copyright 2019-2020 Dolthub, 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. package sqlserver import ( "context" "crypto/tls" "errors" "fmt" "net" "net/http" "os" "path/filepath" "strconv" "strings" "time" "github.com/dolthub/go-mysql-server/eventscheduler" "github.com/dolthub/go-mysql-server/server" "github.com/dolthub/go-mysql-server/server/golden" "github.com/dolthub/go-mysql-server/sql" "github.com/dolthub/go-mysql-server/sql/mysql_db" "github.com/dolthub/go-mysql-server/sql/plan" "github.com/dolthub/go-mysql-server/sql/types" "github.com/dolthub/vitess/go/mysql" "github.com/google/uuid" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/sirupsen/logrus" goerrors "gopkg.in/src-d/go-errors.v1" "github.com/dolthub/dolt/go/cmd/dolt/cli" "github.com/dolthub/dolt/go/cmd/dolt/commands" "github.com/dolthub/dolt/go/cmd/dolt/commands/engine" remotesapi "github.com/dolthub/dolt/go/gen/proto/dolt/services/remotesapi/v1alpha1" "github.com/dolthub/dolt/go/libraries/doltcore/dconfig" "github.com/dolthub/dolt/go/libraries/doltcore/doltdb" "github.com/dolthub/dolt/go/libraries/doltcore/env" "github.com/dolthub/dolt/go/libraries/doltcore/remotesrv" "github.com/dolthub/dolt/go/libraries/doltcore/servercfg" "github.com/dolthub/dolt/go/libraries/doltcore/sqle" "github.com/dolthub/dolt/go/libraries/doltcore/sqle/binlogreplication" "github.com/dolthub/dolt/go/libraries/doltcore/sqle/cluster" _ "github.com/dolthub/dolt/go/libraries/doltcore/sqle/dfunctions" "github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess" "github.com/dolthub/dolt/go/libraries/doltcore/sqlserver" "github.com/dolthub/dolt/go/libraries/events" "github.com/dolthub/dolt/go/libraries/utils/config" "github.com/dolthub/dolt/go/libraries/utils/filesys" httputils "github.com/dolthub/dolt/go/libraries/utils/http" "github.com/dolthub/dolt/go/libraries/utils/svcs" "github.com/dolthub/dolt/go/store/chunks" eventsapi "github.com/dolthub/eventsapi_schema/dolt/services/eventsapi/v1alpha1" ) const ( LocalConnectionUser = "__dolt_local_user__" ApiSqleContextKey = "__sqle_context__" ) // sqlServerHeartbeatIntervalEnvVar is the duration between heartbeats sent to the remote server, used for testing const sqlServerHeartbeatIntervalEnvVar = "DOLT_SQL_SERVER_HEARTBEAT_INTERVAL" // ExternalDisableUsers is called by implementing applications to disable users. This is not used by Dolt itself, // but will break compatibility with implementing applications that do not yet support users. var ExternalDisableUsers bool = false var ErrCouldNotLockDatabase = goerrors.NewKind("database \"%s\" is locked by another dolt process; either clone the database to run a second server, or stop the dolt process which currently holds an exclusive write lock on the database") type Config struct { ServerConfig servercfg.ServerConfig DoltEnv *env.DoltEnv SkipRootUserInit bool Version string Controller *svcs.Controller ProtocolListenerFactory server.ProtocolListenerFunc MCP *MCPConfig // ProviderFactory controls how the DatabaseProvider is instantiated ProviderFactory sqle.ProviderFactory } // Serve starts a MySQL-compatible server. Returns any errors that were encountered. func Serve( ctx context.Context, cfg *Config, ) (startError error, closeError error) { // Code is easier to work through if we assume that serverController is never nil if cfg.Controller == nil { cfg.Controller = svcs.NewController() } ConfigureServices(cfg) go cfg.Controller.Start(ctx) err := cfg.Controller.WaitForStart() if err != nil { return err, nil } return nil, cfg.Controller.WaitForStop() } func ConfigureServices( cfg *Config, ) { controller := cfg.Controller ValidateConfigStep := &svcs.AnonService{ InitF: func(context.Context) error { return servercfg.ValidateConfig(cfg.ServerConfig) }, } controller.Register(ValidateConfigStep) lgr := logrus.StandardLogger() lgr.SetOutput(cli.CliErr) InitLogging := &svcs.AnonService{ InitF: func(context.Context) error { level, err := logrus.ParseLevel(cfg.ServerConfig.LogLevel().String()) if err != nil { return err } logrus.SetLevel(level) switch strings.ToLower(string(cfg.ServerConfig.LogFormat())) { case string(servercfg.LogFormat_JSON): logrus.SetFormatter(&logrus.JSONFormatter{}) case string(servercfg.LogFormat_Text): logrus.SetFormatter(&logrus.TextFormatter{}) default: return fmt.Errorf("unknown log format: %s", cfg.ServerConfig.LogFormat()) } sql.SystemVariables.AddSystemVariables([]sql.SystemVariable{ &sql.MysqlSystemVariable{ Name: dsess.DoltLogLevel, Scope: sql.GetMysqlScope(sql.SystemVariableScope_Global), Dynamic: true, SetVarHintApplies: false, Type: types.NewSystemEnumType(dsess.DoltLogLevel, logrus.PanicLevel.String(), logrus.FatalLevel.String(), logrus.ErrorLevel.String(), logrus.WarnLevel.String(), logrus.InfoLevel.String(), logrus.DebugLevel.String(), logrus.TraceLevel.String(), ), Default: logrus.GetLevel().String(), NotifyChanged: func(ctx *sql.Context, _ sql.SystemVariableScope, v sql.SystemVarValue) error { level, err := logrus.ParseLevel(v.Val.(string)) if err != nil { return fmt.Errorf("could not parse requested log level %s as a log level. dolt_log_level variable value and logging behavior will diverge.", v.Val.(string)) } logrus.SetLevel(level) return nil }, }, }) return nil }, } controller.Register(InitLogging) controller.Register(newHeartbeatService(cfg.Version, cfg.DoltEnv)) fs := cfg.DoltEnv.FS InitFailsafes := &svcs.AnonService{ InitF: func(ctx context.Context) (err error) { cfg.DoltEnv.Config.SetFailsafes(env.DefaultFailsafeConfig) return nil }, } controller.Register(InitFailsafes) var mrEnv *env.MultiRepoEnv InitMultiEnv := &svcs.AnonService{ InitF: func(ctx context.Context) (err error) { mrEnv, err = env.MultiEnvForDirectory(ctx, fs, cfg.DoltEnv) return err }, } controller.Register(InitMultiEnv) AssertNoDatabasesInAccessModeReadOnly := &svcs.AnonService{ InitF: func(ctx context.Context) (err error) { return mrEnv.Iter(func(name string, dEnv *env.DoltEnv) (stop bool, err error) { readOnly, err := dEnv.IsAccessModeReadOnly(ctx) if err != nil { return true, err } if readOnly { return true, ErrCouldNotLockDatabase.New(name) } return false, nil }) }, } controller.Register(AssertNoDatabasesInAccessModeReadOnly) var localCreds *LocalCreds InitServerLocalCreds := &svcs.AnonService{ InitF: func(context.Context) (err error) { localCreds, err = persistServerLocalCreds(cfg.ServerConfig.Port(), cfg.DoltEnv) return err }, StopF: func(_ svcs.RunState) error { RemoveLocalCreds(cfg.DoltEnv.FS) return nil }, } controller.Register(InitServerLocalCreds) // Persist any system variables that have a non-deterministic default value (i.e. @@server_uuid) // We only do this on sql-server startup initially since we want to keep the persisted server_uuid // in the configuration files for a sql-server, and not global for the whole host. // // This load can set global variables that will be read by services which are started later. Those // services may be running background threads of their own. This load should be done early to // prevent races and to prevent different components seeing different configured variables. PersistNondeterministicSystemVarDefaults := &svcs.AnonService{ InitF: func(ctx context.Context) error { err := dsess.PersistSystemVarDefaults(cfg.DoltEnv) if err != nil { logrus.Errorf("unable to persist system variable defaults: %v", err) } // Always return nil, because we don't want an invalid config value to prevent // the server from starting up. return nil }, } controller.Register(PersistNondeterministicSystemVarDefaults) var clusterController *cluster.Controller InitClusterController := &svcs.AnonService{ InitF: func(context.Context) (err error) { clusterController, err = cluster.NewController(lgr, cfg.ServerConfig.ClusterConfig(), mrEnv.Config()) return err }, } controller.Register(InitClusterController) var serverConf server.Config LoadServerConfig := &svcs.AnonService{ InitF: func(context.Context) (err error) { serverConf, err = getConfigFromServerConfig(cfg.ServerConfig, cfg.ProtocolListenerFactory) return err }, } controller.Register(LoadServerConfig) // Create SQL Engine with users var config *engine.SqlEngineConfig InitSqlEngineConfig := &svcs.AnonService{ InitF: func(context.Context) error { config = &engine.SqlEngineConfig{ IsReadOnly: cfg.ServerConfig.ReadOnly(), PrivFilePath: cfg.ServerConfig.PrivilegeFilePath(), BranchCtrlFilePath: cfg.ServerConfig.BranchControlFilePath(), DoltCfgDirPath: cfg.ServerConfig.CfgDir(), ServerUser: cfg.ServerConfig.User(), ServerPass: cfg.ServerConfig.Password(), ServerHost: cfg.ServerConfig.Host(), Autocommit: cfg.ServerConfig.AutoCommit(), DoltTransactionCommit: cfg.ServerConfig.DoltTransactionCommit(), BranchActivityTracking: cfg.ServerConfig.BranchActivityTracking(), JwksConfig: cfg.ServerConfig.JwksConfig(), SystemVariables: cfg.ServerConfig.SystemVars(), ClusterController: clusterController, BinlogReplicaController: binlogreplication.DoltBinlogReplicaController, SkipRootUserInitialization: cfg.SkipRootUserInit, EngineOverrides: cfg.ServerConfig.Overrides(), ProviderFactory: cfg.ProviderFactory, } return nil }, } controller.Register(InitSqlEngineConfig) var esStatus eventscheduler.SchedulerStatus InitEventSchedulerStatus := &svcs.AnonService{ InitF: func(context.Context) (err error) { esStatus, err = getEventSchedulerStatus(cfg.ServerConfig.EventSchedulerStatus()) if err != nil { return err } config.EventSchedulerStatus = esStatus return nil }, } controller.Register(InitEventSchedulerStatus) gcSch := os.Getenv(dconfig.EnvDoltGCScheduler) InitAutoGCController := &svcs.AnonService{ InitF: func(context.Context) error { // Having AutoGCBehavior == nil means that they are using default behavior, which is enabled. if cfg.ServerConfig.AutoGCBehavior() == nil { config.AutoGCController = sqle.NewAutoGCController(chunks.SimpleArchive, chunks.IncrementalGCTablesDisabled, sqle.NewGCScheduler(gcSch), lgr) } else if cfg.ServerConfig.AutoGCBehavior().Enable() { cmp := chunks.GCArchiveLevel(cfg.ServerConfig.AutoGCBehavior().ArchiveLevel()) if cmp < chunks.NoArchive || cmp > chunks.MaxArchiveLevel { return fmt.Errorf("invalid value for %s: %d", cli.ArchiveLevelParam, cmp) } incrementalArchiveSize := cfg.ServerConfig.AutoGCBehavior().IncrementalFileSize() config.AutoGCController = sqle.NewAutoGCController(cmp, incrementalArchiveSize, sqle.NewGCScheduler(gcSch), lgr) } return nil }, } controller.Register(InitAutoGCController) // mySQLServer is going to be populated down below once further services // are initialized. However, we want to block Controller shutdown on all // connections being fully drained from the Server. Stopping the // SQL server itself only stops the connection listener, and inflight // work is shutdown by other services, which can stop things like // replication threads, listeners for other services, etc. // // On the shutdown path, we block for connection draining // right after we have stopped the sql engine itself, which // was responsible for canceling the contexts associated with // all inflight queries. var mySQLServer *server.Server DrainClientConnectionsOnShutdown := &svcs.AnonService{ StopF: func(_ svcs.RunState) error { if mySQLServer != nil { mySQLServer.SessionManager().WaitForClosedConnections() } return nil }, } controller.Register(DrainClientConnectionsOnShutdown) var sqlEngine *engine.SqlEngine InitSqlEngine := &svcs.AnonService{ InitF: func(ctx context.Context) (err error) { if _, err := mrEnv.Config().GetString(env.SqlServerGlobalsPrefix + "." + dsess.DoltStatsPaused); err != nil { // unless otherwise specified, run stats writer alongside server sqlCtx := sql.NewEmptyContext() sql.SystemVariables.SetGlobal(sqlCtx, dsess.DoltStatsPaused, 0) } sqlEngine, err = engine.NewSqlEngine( ctx, mrEnv, config, ) if err != nil { return err } // Set the read_only system variable based on the engine configuration. If cluster replication is enabled, // the read_only value is managed by the cluster role callback instead. if config.ClusterController != nil { return nil } sqlCtx := sql.NewEmptyContext() readOnlyValue := int8(0) if config.IsReadOnly { readOnlyValue = int8(1) } return sql.SystemVariables.SetGlobal(sqlCtx, "read_only", readOnlyValue) }, StopF: func(_ svcs.RunState) error { sqlEngine.Close() return nil }, } controller.Register(InitSqlEngine) // Closing the connections on shutdown attempts to prevent // them from creating new work after we cancel their running // queries up above. GMS should ideally avoid creating new // process list queries or operations for these connections // after they are killed, but it is not currently set up that // way. CloseClientConnectionsOnShutdown := &svcs.AnonService{ StopF: func(_ svcs.RunState) error { if mySQLServer != nil { sm := mySQLServer.SessionManager() return sm.Iter(func(s sql.Session) (bool, error) { sm.KillConnection(s.ID()) return false, nil }) } return nil }, } controller.Register(CloseClientConnectionsOnShutdown) InitStatsController := &svcs.AnonService{ InitF: func(ctx context.Context) error { return sqlEngine.InitStats(ctx) }, } controller.Register(InitStatsController) InitBinlogging := &svcs.AnonService{ InitF: func(ctx context.Context) error { sqlCtx := sql.NewContext(ctx) primaryController := sqlEngine.GetUnderlyingEngine().Analyzer.Catalog.BinlogPrimaryController doltBinlogPrimaryController, ok := primaryController.(*binlogreplication.DoltBinlogPrimaryController) if !ok { return fmt.Errorf("unexpected type of binlog controller: %T", primaryController) } _, logBinValue, ok := sql.SystemVariables.GetGlobal("log_bin") if !ok { return fmt.Errorf("unable to load @@log_bin system variable") } logBin, ok := logBinValue.(int8) if !ok { return fmt.Errorf("unexpected type for @@log_bin system variable: %T", logBinValue) } _, logBinBranchValue, ok := sql.SystemVariables.GetGlobal("log_bin_branch") if !ok { return fmt.Errorf("unable to load @@log_bin_branch system variable") } logBinBranch, ok := logBinBranchValue.(string) if !ok { return fmt.Errorf("unexpected type for @@log_bin_branch system variable: %T", logBinBranchValue) } if logBinBranch != "" { // If an invalid branch has been configured, let the server start up so that it's // easier for customers to correct the value, but log a warning and don't enable // binlog replication. if strings.Contains(logBinBranch, "/") { logrus.Warnf("branch names containing '/' are not supported "+ "for binlog replication. Not enabling binlog replication; fix "+ "@@log_bin_branch value and restart Dolt (current value: %s)", logBinBranch) return nil } binlogreplication.BinlogBranch = logBinBranch } if logBin == 1 { logrus.Infof("Enabling binary logging for branch %s", logBinBranch) binlogProducer, err := binlogreplication.NewBinlogProducer(sqlCtx, cfg.DoltEnv.FS) if err != nil { return err } logManager, err := binlogreplication.NewLogManager(sqlCtx, fs) if err != nil { return err } binlogProducer.LogManager(logManager) doltdb.RegisterDatabaseUpdateListener(binlogProducer) doltBinlogPrimaryController.BinlogProducer(binlogProducer) // Register binlog hooks for database creation/deletion provider := sqlEngine.GetUnderlyingEngine().Analyzer.Catalog.DbProvider if doltProvider, ok := provider.(*sqle.DoltDatabaseProvider); ok { doltProvider.AddInitDatabaseHook(binlogreplication.NewBinlogInitDatabaseHook(nil, doltdb.DatabaseUpdateListeners)) doltProvider.AddDropDatabaseHook(binlogreplication.NewBinlogDropDatabaseHook(nil, doltdb.DatabaseUpdateListeners)) } } return nil }, } controller.Register(InitBinlogging) // MySQL creates a root superuser when the mysql install is first initialized. Depending on the options // specified, the root superuser is created without a password, or with a random password. This varies // slightly in some OS-specific installers. Dolt initializes the root superuser the first time a // sql-server is started and initializes its privileges database. We do this on sql-server initialization, // instead of dolt db initialization, because we only want to create the privileges database when it's // used for a server, and because we want the same root initialization logic when a sql-server is started // for a clone. More details: https://dev.mysql.com/doc/mysql-security-excerpt/8.0/en/default-privileges.html InitImplicitRootSuperUser := &svcs.AnonService{ InitF: func(ctx context.Context) error { // If privileges.db has already been initialized, indicating that this is NOT the // first time sql-server has been launched, then don't initialize the root superuser. if permissionDbExists, err := doesPrivilegesDbExist(cfg.DoltEnv, cfg.ServerConfig.PrivilegeFilePath()); err != nil { return err } else if permissionDbExists { logrus.Debug("privileges.db already exists, not creating root superuser") return nil } // We always persist the privileges.db file, to signal that the privileges system has been initialized mysqlDb := sqlEngine.GetUnderlyingEngine().Analyzer.Catalog.MySQLDb ed := mysqlDb.Editor() defer ed.Close() // Create the root@localhost superuser, unless --skip-root-user-initialization was specified if !config.SkipRootUserInitialization { // Allow the user to override the default root host (localhost) and password (""). // This is particularly useful in a Docker container, where you need to connect // to the sql-server from outside the container and can't rely on localhost. rootHost := "localhost" doltRootHost := os.Getenv(dconfig.EnvDoltRootHost) if doltRootHost != "" { logrus.Infof("Overriding root user host with value from DOLT_ROOT_HOST: %s", doltRootHost) rootHost = doltRootHost } rootPassword := servercfg.DefaultPass doltRootPassword := os.Getenv(dconfig.EnvDoltRootPassword) if doltRootPassword != "" { logrus.Info("Overriding root user password with value from DOLT_ROOT_PASSWORD") rootPassword = doltRootPassword } logrus.Infof("Creating root@%s superuser", rootHost) mysqlDb.AddSuperUser(ed, servercfg.DefaultUser, rootHost, rootPassword) } // TODO: The in-memory filesystem doesn't work with the GMS API // for persisting the privileges database. The filesys API // is in the Dolt layer, so when the file path is passed to // GMS, it expects it to be a path on disk, and errors out. if _, isInMemFs := cfg.DoltEnv.FS.(*filesys.InMemFS); isInMemFs { return nil } else { sqlCtx, err := sqlEngine.NewDefaultContext(context.Background()) if err != nil { return err } return mysqlDb.Persist(sqlCtx, ed) } }, } controller.Register(InitImplicitRootSuperUser) var metListener *metricsListener InitMetricsListener := &svcs.AnonService{ InitF: func(context.Context) (err error) { labels := cfg.ServerConfig.MetricsLabels() path, err := cfg.DoltEnv.FS.Abs(".") if err != nil { logrus.Debug(fmt.Sprintf("Error getting absolute path for metrics listener: %v", err)) path = "" } metricsExposed := cfg.ServerConfig.MetricsHost() != "" && cfg.ServerConfig.MetricsPort() > 0 metListener, err = newMetricsListener(labels, cfg.Version, path, clusterController, metricsExposed) return err }, StopF: func(_ svcs.RunState) error { metListener.Close() return nil }, } controller.Register(InitMetricsListener) InitLockSuperUser := &svcs.AnonService{ InitF: func(ctx context.Context) error { mysqlDb := sqlEngine.GetUnderlyingEngine().Analyzer.Catalog.MySQLDb host := "localhost" // this is reserved for `dolt sql` connections when sql-server is running reservedUser := LocalConnectionUser rd := mysqlDb.Reader() conflict, _ := rd.GetUser(mysql_db.UserPrimaryKey{Host: host, User: reservedUser}) rd.Close() var conflicts []mysql_db.UserPrimaryKey if conflict != nil && !conflict.IsEphemeral { logrus.Warnf("Dropping persisted '%s@%s' because this account name is reserved for Dolt", conflict.User, conflict.Host) conflicts = append(conflicts, mysql_db.UserPrimaryKey{Host: conflict.Host, User: conflict.User}) } ed := mysqlDb.Editor() defer ed.Close() if len(conflicts) > 0 { c := conflicts[0] ed.RemoveUser(c) ed.RemoveRoleEdgesFromKey(mysql_db.RoleEdgesFromKey{ FromHost: c.Host, FromUser: c.User, }) ed.RemoveRoleEdgesToKey(mysql_db.RoleEdgesToKey{ ToHost: c.Host, ToUser: c.User, }) sqlCtx, err := sqlEngine.NewDefaultContext(ctx) if err != nil { return fmt.Errorf("failed to create SQL context: %v", err) } if err := mysqlDb.Persist(sqlCtx, ed); err != nil { return fmt.Errorf("failed to persist changes to privileges database: %v", err) } } mysqlDb.AddEphemeralSuperUser(ed, LocalConnectionUser, "localhost", localCreds.Secret) return nil }, } controller.Register(InitLockSuperUser) DisableMySQLDbIfRequired := &svcs.AnonService{ InitF: func(context.Context) error { if ExternalDisableUsers { mysqlDb := sqlEngine.GetUnderlyingEngine().Analyzer.Catalog.MySQLDb mysqlDb.SetEnabled(false) } return nil }, } controller.Register(DisableMySQLDbIfRequired) type SQLMetricsService struct { enabled bool lis net.Listener srv *http.Server } var metSrv SQLMetricsService RunMetricsServer := &svcs.AnonService{ InitF: func(context.Context) (err error) { if cfg.ServerConfig.MetricsHost() != "" && cfg.ServerConfig.MetricsPort() > 0 { metSrv.enabled = true addr := fmt.Sprintf("%s:%d", cfg.ServerConfig.MetricsHost(), cfg.ServerConfig.MetricsPort()) metSrv.lis, err = net.Listen("tcp", addr) if err != nil { return err } tlsConfig, err := servercfg.LoadMetricsTLSConfig(cfg.ServerConfig) if err != nil { return err } mux := http.NewServeMux() metricsHandler := promhttp.Handler() jwksConfig := cfg.ServerConfig.MetricsJwksConfig() enableMetricsAuth := jwksConfig != nil requireLocalhostAuth := cfg.ServerConfig.MetricsJWTRequiredForLocalhost() logrus.Infof("Starting metrics server. auth_enabled = %t, addr = %s, require_localhost_auth = %t", enableMetricsAuth, addr, requireLocalhostAuth) if enableMetricsAuth { mux.Handle("/metrics", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if !requireLocalhostAuth { isLocal, err := httputils.IsLocalRequest(r) logrus.Info("Metrics JWT not required for localhost isLocal:", isLocal, "err:", err) if err != nil { logrus.Warnf("error checking if request is local for /metrics (assuming remote) request: %v.", err) } else if isLocal { metricsHandler.ServeHTTP(w, r) return } } auth := r.Header.Get("Authorization") if auth == "" || !strings.HasPrefix(auth, "Bearer ") { w.Header().Set("WWW-Authenticate", `Bearer realm="metrics"`) http.Error(w, "unauthorized", http.StatusUnauthorized) return } valid, _, err := validateJWT(jwksConfig, strings.TrimPrefix(auth, "Bearer "), time.Now()) if err != nil { logrus.Warnf("JWT validation error for /metrics: %v", err) http.Error(w, "auth failed", http.StatusUnauthorized) return } else if !valid { logrus.Warnf("JWT validation error for /metrics: JWT token is invalid") http.Error(w, "invalid token", http.StatusUnauthorized) return } metricsHandler.ServeHTTP(w, r) })) } else { mux.Handle("/metrics", metricsHandler) } metSrv.srv = &http.Server{ Addr: addr, Handler: mux, TLSConfig: tlsConfig, } } return nil }, RunF: func(context.Context) { if metSrv.enabled { if metSrv.srv.TLSConfig != nil { _ = metSrv.srv.ServeTLS(metSrv.lis, "", "") } else { _ = metSrv.srv.Serve(metSrv.lis) } } }, StopF: func(rs svcs.RunState) error { if !metSrv.enabled { return nil } if rs == svcs.RunInvoked { metSrv.srv.Close() } else { metSrv.lis.Close() } return nil }, } controller.Register(RunMetricsServer) type RemoteSrvService struct { enabled bool lis remotesrv.Listeners srv *remotesrv.Server } var remoteSrv RemoteSrvService RunRemoteSrv := &svcs.AnonService{ InitF: func(ctx context.Context) error { if cfg.ServerConfig.RemotesapiPort() == nil { return nil } remoteSrv.enabled = true port := *cfg.ServerConfig.RemotesapiPort() apiReadOnly := false if cfg.ServerConfig.RemotesapiReadOnly() != nil { apiReadOnly = *cfg.ServerConfig.RemotesapiReadOnly() } listenaddr := fmt.Sprintf(":%d", port) sqlContextInterceptor := sqle.SqlContextServerInterceptor{ Factory: sqlEngine.NewDefaultContext, } args := remotesrv.ServerArgs{ Logger: logrus.NewEntry(lgr), ReadOnly: apiReadOnly || cfg.ServerConfig.ReadOnly(), HttpListenAddr: listenaddr, GrpcListenAddr: listenaddr, ConcurrencyControl: remotesapi.PushConcurrencyControl_PUSH_CONCURRENCY_CONTROL_ASSERT_WORKING_SET, Options: sqlContextInterceptor.Options(), HttpInterceptor: sqlContextInterceptor.HTTP(nil), } var err error args.FS = sqlEngine.FileSystem() args.DBCache, err = sqle.RemoteSrvDBCache(sqle.GetInterceptorSqlContext, sqle.DoNotCreateUnknownDatabases, sqle.IsNotReplicaWrite) if err != nil { lgr.Errorf("error creating SQL engine context for remotesapi server: %v", err) return err } authenticator := newAccessController(sqle.GetInterceptorSqlContext, sqlEngine.GetUnderlyingEngine().Analyzer.Catalog.MySQLDb) args = sqle.WithUserPasswordAuth(args, authenticator) args.TLSConfig = serverConf.TLSConfig remoteSrv.srv, err = remotesrv.NewServer(args) if err != nil { lgr.Errorf("error creating remotesapi server on port %d: %v", port, err) return err } remoteSrv.lis, err = remoteSrv.srv.Listeners() if err != nil { lgr.Errorf("error starting remotesapi server listeners on port %d: %v", port, err) return err } return nil }, RunF: func(ctx context.Context) { if remoteSrv.enabled { remoteSrv.srv.Serve(remoteSrv.lis) } }, StopF: func(rs svcs.RunState) error { if !remoteSrv.enabled { return nil } if rs == svcs.RunInvoked { remoteSrv.srv.GracefulStop() } else { remoteSrv.lis.Close() } return nil }, } controller.Register(RunRemoteSrv) var clusterRemoteSrv RemoteSrvService RunClusterRemoteSrv := &svcs.AnonService{ InitF: func(context.Context) error { if clusterController == nil { return nil } clusterRemoteSrv.enabled = true args, err := clusterController.RemoteSrvServerArgs(sqlEngine.NewDefaultContext, remotesrv.ServerArgs{ Logger: logrus.NewEntry(lgr), }) if err != nil { lgr.Errorf("error creating SQL engine context for remotesapi server: %v", err) return err } args.FS = sqlEngine.FileSystem() clusterRemoteSrvTLSConfig, err := LoadClusterTLSConfig(cfg.ServerConfig.ClusterConfig()) if err != nil { lgr.Errorf("error starting remotesapi server for cluster config, could not load tls config: %v", err) return err } args.TLSConfig = clusterRemoteSrvTLSConfig clusterRemoteSrv.srv, err = remotesrv.NewServer(args) if err != nil { lgr.Errorf("error creating remotesapi server on port %d: %v", *cfg.ServerConfig.RemotesapiPort(), err) return err } clusterController.RegisterGrpcServices(sqle.GetInterceptorSqlContext, clusterRemoteSrv.srv.GrpcServer()) clusterRemoteSrv.lis, err = clusterRemoteSrv.srv.Listeners() if err != nil { lgr.Errorf("error starting remotesapi server listeners for cluster config on %s: %v", clusterController.RemoteSrvListenAddr(), err) return err } return nil }, RunF: func(context.Context) { if clusterRemoteSrv.enabled { clusterRemoteSrv.srv.Serve(clusterRemoteSrv.lis) } }, StopF: func(rs svcs.RunState) error { if !clusterRemoteSrv.enabled { return nil } if rs == svcs.RunInvoked { clusterRemoteSrv.srv.GracefulStop() } else { clusterRemoteSrv.lis.Close() } return nil }, } controller.Register(RunClusterRemoteSrv) // We still have some startup to do from this point, and we do not run // the SQL server until we are fully booted. We also want to stop the // SQL server as the first thing we stop. However, if startup fails // during initialization, we want to shutdown the SQL server cleanly. // So we track whether the server has been shutdown by either service // which is responsible for it and we only do it here if it hasn't // already been Closed. var sqlServerClosed bool InitSQLServer := &svcs.AnonService{ InitF: func(context.Context) (err error) { v, ok := cfg.ServerConfig.(servercfg.ValidatingServerConfig) if ok && v.GoldenMysqlConnectionString() != "" { mySQLServer, err = server.NewServerWithHandler( serverConf, sqlEngine.GetUnderlyingEngine(), sqlEngine.ContextFactory, newSessionBuilder(sqlEngine, cfg.ServerConfig), metListener, func(h mysql.Handler) (mysql.Handler, error) { return golden.NewValidatingHandler(h, v.GoldenMysqlConnectionString(), logrus.StandardLogger()) }, ) } else { mySQLServer, err = server.NewServer( serverConf, sqlEngine.GetUnderlyingEngine(), sqlEngine.ContextFactory, newSessionBuilder(sqlEngine, cfg.ServerConfig), metListener, ) } if errors.Is(err, server.UnixSocketInUseError) { lgr.Warn("unix socket set up failed: file already in use: ", serverConf.Socket) err = nil } return err }, StopF: func(_ svcs.RunState) (err error) { if !sqlServerClosed { sqlServerClosed = true return mySQLServer.Close() } return nil }, } controller.Register(InitSQLServer) // Automatically restart binlog replication if replication was enabled when the server was last shut down AutoStartBinlogReplica := &svcs.AnonService{ InitF: func(ctx context.Context) error { // If we're unable to restart replication, log an error, but don't prevent the server from starting up sqlCtx, err := sqlEngine.NewDefaultContext(ctx) if err != nil { logrus.Errorf("unable to restart replication, could not create session: %s", err.Error()) return nil } defer sql.SessionEnd(sqlCtx.Session) if err := binlogreplication.DoltBinlogReplicaController.AutoStart(sqlCtx); err != nil { logrus.Errorf("unable to restart replication: %s", err.Error()) } return nil }, } controller.Register(AutoStartBinlogReplica) RunClusterController := &svcs.AnonService{ InitF: func(context.Context) error { if clusterController == nil { return nil } clusterController.ManageQueryConnections( mySQLServer.SessionManager().Iter, sqlEngine.GetUnderlyingEngine().ProcessList.Kill, mySQLServer.SessionManager().KillConnection, ) return nil }, RunF: func(context.Context) { if clusterController == nil { return } clusterController.Run() }, StopF: func(_ svcs.RunState) error { if clusterController == nil { return nil } clusterController.GracefulStop() return nil }, } controller.Register(RunClusterController) runDone := make(chan struct{}) RunSQLServer := &svcs.AnonService{ RunF: func(context.Context) { defer close(runDone) sqlserver.SetRunningServer(mySQLServer) defer sqlserver.UnsetRunningServer() mySQLServer.Start() }, StopF: func(rs svcs.RunState) error { sqlServerClosed = true closeErr := mySQLServer.Close() if rs == svcs.RunInvoked { <-runDone } return closeErr }, } controller.Register(RunSQLServer) // Optionally start an MCP HTTP server registerMCPService(controller, cfg, lgr) } // heartbeatService is a service that sends a heartbeat event to the metrics server once a day type heartbeatService struct { version string eventEmitter events.Emitter interval time.Duration closer func() error } func newHeartbeatService(version string, dEnv *env.DoltEnv) *heartbeatService { metricsDisabled := dEnv.Config.GetStringOrDefault(config.MetricsDisabled, "false") disabled, err := strconv.ParseBool(metricsDisabled) if err != nil || disabled { return &heartbeatService{} // will be defunct on Run() } emitterType, ok := os.LookupEnv(events.EmitterTypeEnvVar) if !ok { emitterType = events.EmitterTypeGrpc } interval, ok := os.LookupEnv(sqlServerHeartbeatIntervalEnvVar) if !ok { interval = "24h" } duration, err := time.ParseDuration(interval) if err != nil { return &heartbeatService{} // will be defunct on Run() } emitter, closer, err := commands.NewEmitter(emitterType, dEnv) if err != nil { return &heartbeatService{} // will be defunct on Run() } events.SetGlobalCollector(events.NewCollector(version, emitter)) return &heartbeatService{ version: version, eventEmitter: emitter, interval: duration, closer: closer, } } func (h *heartbeatService) Init(ctx context.Context) error { return nil } func (h *heartbeatService) Stop(_ svcs.RunState) error { if h.closer != nil { return h.closer() } return nil } func (h *heartbeatService) Run(ctx context.Context) { // Faulty config settings or disabled metrics can cause us to not have a valid event emitter if h.eventEmitter == nil { return } ticker := time.NewTicker(h.interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: t := events.NowTimestamp() logrus.Debugf("sending heartbeat event to %s:%s", events.DefaultMetricsHost, events.DefaultMetricsPort) err := h.eventEmitter.LogEvents(ctx, h.version, []*eventsapi.ClientEvent{ { Id: uuid.New().String(), StartTime: t, EndTime: t, Type: eventsapi.ClientEventType_SQL_SERVER_HEARTBEAT, }, }) if err != nil { logrus.Debugf("failed to send heartbeat event: %v", err) } } } } var _ svcs.Service = &heartbeatService{} func persistServerLocalCreds(port int, dEnv *env.DoltEnv) (*LocalCreds, error) { creds := NewLocalCreds(port) err := WriteLocalCreds(dEnv.FS, creds) if err != nil { return nil, err } return creds, err } // remotesapiAuth facilitates the implementation remotesrv.AccessControl for the remotesapi server. type remotesapiAuth struct { // ctxFactory is a function that returns a new sql.Context. This will create a new context every time it is called, // so it should be called once per API request. ctxFactory func(context.Context) (*sql.Context, error) rawDb *mysql_db.MySQLDb } func newAccessController(ctxFactory func(context.Context) (*sql.Context, error), rawDb *mysql_db.MySQLDb) remotesrv.AccessControl { return &remotesapiAuth{ctxFactory, rawDb} } // ApiAuthenticate checks the provided credentials against the database and return a SQL context if the credentials are // valid. If the credentials are invalid, then a nil context is returned. Failures to authenticate are logged. func (r *remotesapiAuth) ApiAuthenticate(ctx context.Context) (context.Context, error) { creds, err := remotesrv.ExtractBasicAuthCreds(ctx) if err != nil { return nil, err } err = commands.ValidatePasswordWithAuthResponse(r.rawDb, creds.Username, creds.Password) if err != nil { return nil, fmt.Errorf("API Authentication Failure: %v", err) } address := creds.Address if strings.Index(address, ":") > 0 { address, _, err = net.SplitHostPort(creds.Address) if err != nil { return nil, fmt.Errorf("Invalid Host string for authentication: %s", creds.Address) } } sqlCtx, err := r.ctxFactory(ctx) if err != nil { return nil, fmt.Errorf("API Runtime error: %v", err) } sqlCtx.Session.SetClient(sql.Client{User: creds.Username, Address: address, Capabilities: 0}) updatedCtx := context.WithValue(ctx, ApiSqleContextKey, sqlCtx) return updatedCtx, nil } func (r *remotesapiAuth) ApiAuthorize(ctx context.Context, superUserRequired bool) (bool, error) { sqlCtx, ok := ctx.Value(ApiSqleContextKey).(*sql.Context) if !ok { return false, fmt.Errorf("Runtime error: could not get SQL context from context") } privOp := sql.NewDynamicPrivilegedOperation(plan.DynamicPrivilege_CloneAdmin) if superUserRequired { database := sqlCtx.GetCurrentDatabase() subject := sql.PrivilegeCheckSubject{Database: database} privOp = sql.NewPrivilegedOperation(subject, sql.PrivilegeType_Super) } authorized := r.rawDb.UserHasPrivileges(sqlCtx, privOp) if !authorized { if superUserRequired { return false, fmt.Errorf("API Authorization Failure: %s has not been granted SuperUser access", sqlCtx.Session.Client().User) } return false, fmt.Errorf("API Authorization Failure: %s has not been granted CLONE_ADMIN access", sqlCtx.Session.Client().User) } return true, nil } // doesPrivilegesDbExist looks for an existing privileges database as the specified |privilegeFilePath|. If // |privilegeFilePath| is an absolute path, it is used directly. If it is a relative path, then it is resolved // relative to the root of the specified |dEnv|. func doesPrivilegesDbExist(dEnv *env.DoltEnv, privilegeFilePath string) (exists bool, err error) { if !filepath.IsAbs(privilegeFilePath) { privilegeFilePath, err = dEnv.FS.Abs(privilegeFilePath) if err != nil { return false, err } } _, err = os.Stat(privilegeFilePath) if err != nil { if os.IsNotExist(err) { return false, nil } else { return false, err } } return true, nil } func LoadClusterTLSConfig(cfg servercfg.ClusterConfig) (*tls.Config, error) { rcfg := cfg.RemotesAPIConfig() if rcfg.TLSKey() == "" && rcfg.TLSCert() == "" { return nil, nil } c, err := tls.LoadX509KeyPair(rcfg.TLSCert(), rcfg.TLSKey()) if err != nil { return nil, err } return &tls.Config{ Certificates: []tls.Certificate{ c, }, }, nil } func newSessionBuilder(se *engine.SqlEngine, config servercfg.ServerConfig) server.SessionBuilder { userToSessionVars := make(map[string]map[string]interface{}) userVars := config.UserVars() for _, curr := range userVars { userToSessionVars[curr.Name] = curr.Vars } return func(ctx context.Context, conn *mysql.Conn, addr string) (sql.Session, error) { baseSession, err := sql.BaseSessionFromConnection(ctx, conn, addr) if err != nil { return nil, err } dSess, err := se.NewDoltSession(ctx, baseSession) if err != nil { return nil, err } varsForUser := userToSessionVars[conn.User] if len(varsForUser) > 0 { sqlCtx, err := se.NewContext(ctx, dSess) if err != nil { return nil, err } for key, val := range varsForUser { if err = dSess.InitSessionVariable(sqlCtx, key, val); err != nil { return nil, err } } } return dSess, nil } } // getConfigFromServerConfig processes ServerConfig and returns server.Config for sql-server. func getConfigFromServerConfig(serverConfig servercfg.ServerConfig, plf server.ProtocolListenerFunc) (server.Config, error) { serverConf, err := handleProtocolAndAddress(serverConfig) if err != nil { return server.Config{}, err } serverConf.DisableClientMultiStatements = serverConfig.DisableClientMultiStatements() readTimeout := time.Duration(serverConfig.ReadTimeout()) * time.Millisecond writeTimeout := time.Duration(serverConfig.WriteTimeout()) * time.Millisecond tlsConfig, err := servercfg.LoadTLSConfig(serverConfig) if err != nil { return server.Config{}, err } serverConf, err = serverConf.NewConfig() if err != nil { return server.Config{}, err } // The server config is created before system variables are initialized, so we have to use // the config to inform its responses. if verstionString, hasVersionString := serverConfig.SystemVars()["version"]; hasVersionString { serverConf.Version = fmt.Sprint(verstionString) } serverConf.ConnReadTimeout = readTimeout serverConf.ConnWriteTimeout = writeTimeout serverConf.MaxConnections = serverConfig.MaxConnections() serverConf.MaxWaitConnections = serverConfig.MaxWaitConnections() serverConf.MaxWaitConnectionsTimeout = serverConfig.MaxWaitConnectionsTimeout() serverConf.TLSConfig = tlsConfig serverConf.RequireSecureTransport = serverConfig.RequireSecureTransport() serverConf.MaxLoggedQueryLen = serverConfig.MaxLoggedQueryLen() serverConf.EncodeLoggedQuery = serverConfig.ShouldEncodeLoggedQuery() serverConf.ProtocolListenerFactory = plf // If client certs are required, then TLS connections are implicitly required if serverConfig.RequireClientCert() { serverConf.RequireSecureTransport = true } return serverConf, nil } // handleProtocolAndAddress returns new server.Config object with only Protocol and Address defined. func handleProtocolAndAddress(serverConfig servercfg.ServerConfig) (server.Config, error) { serverConf := server.Config{Protocol: "tcp"} portAsString := strconv.Itoa(serverConfig.Port()) hostPort := net.JoinHostPort(serverConfig.Host(), portAsString) if server.PortInUse(hostPort) { portInUseError := fmt.Errorf("Port %s already in use.", portAsString) return server.Config{}, portInUseError } serverConf.Address = hostPort sock, useSock, err := servercfg.CheckForUnixSocket(serverConfig) if err != nil { return server.Config{}, err } if useSock { serverConf.Socket = sock } return serverConf, nil } func getEventSchedulerStatus(status string) (eventscheduler.SchedulerStatus, error) { switch strings.ToLower(status) { case "on", "1": return eventscheduler.SchedulerOn, nil case "off", "0": return eventscheduler.SchedulerOff, nil case "disabled": return eventscheduler.SchedulerDisabled, nil default: return eventscheduler.SchedulerDisabled, fmt.Errorf("Error while setting value '%s' to 'event_scheduler'.", status) } }