migration: add scaffolding for the migrations manager

pkg/migration/BUILD.bazel

@@ -0,0 +1,19 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+ name = "migration",
+ srcs = [
+ "manager.go",
+ "migrations.go",
+ ],
+ importpath = "github.com/cockroachdb/cockroach/pkg/migration",
+ visibility = ["//visibility:public"],
+ deps = [
+ "//pkg/kv",
+ "//pkg/kv/kvserver/liveness/livenesspb",
+ "//pkg/roachpb",
+ "//pkg/rpc/nodedialer",
+ "//pkg/sql",
+ "//vendor/github.com/cockroachdb/logtags",
+ ],
+)

pkg/migration/manager.go

@@ -0,0 +1,136 @@
+// Copyright 2020 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+// Package migration captures the facilities needed to define and execute
+// migrations for a crdb cluster. These migrations can be arbitrarily long
+// running, are free to send out arbitrary requests cluster wide, change
+// internal DB state, and much more. They're typically reserved for crdb
+// internal operations and state. Each migration is idempotent in nature, is
+// associated with a specific cluster version, and executed when the cluster
+// version is made activate on every node in the cluster.
+//
+// Examples of migrations that apply would be migrations to move all raft state
+// from one storage engine to another, or purging all usage of the replicated
+// truncated state in KV. A "sister" package of interest is pkg/sqlmigrations.
+package migration
+
+import (
+ "context"
+
+ "github.com/cockroachdb/cockroach/pkg/kv"
+ "github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness/livenesspb"
+ "github.com/cockroachdb/cockroach/pkg/roachpb"
+ "github.com/cockroachdb/cockroach/pkg/rpc/nodedialer"
+ "github.com/cockroachdb/cockroach/pkg/sql"
+ "github.com/cockroachdb/logtags"
+)
+
+// Migration defines a program to be executed once every node in the cluster is
+// (a) running a specific binary version, and (b) has completed all prior
+// migrations.
+//
+// Each migration is associated with a specific internal cluster version and is
+// idempotent in nature. When setting the cluster version (via `SET CLUSTER
+// SETTING version`), a manager process determines the set of migrations needed
+// to bridge the gap between the current active cluster version, and the target
+// one.
+//
+// To introduce a migration, introduce a version key in pkg/clusterversion, and
+// introduce a corresponding internal cluster version for it. See [1] for
+// details. Following that, define a Migration in this package and add it to the
+// Registry. Be sure to key it in with the new cluster version we just added.
+// During cluster upgrades, once the operator is able to set a cluster version
+// setting that's past the version that was introduced (typically the major
+// release version the migration was introduced in), the manager will execute
+// the defined migration before letting the upgrade finalize.
+//
+// [1]: pkg/clusterversion/cockroach_versions.go
+type Migration func(context.Context, *Helper) error
+
+// Manager is the instance responsible for executing migrations across the
+// cluster.
+type Manager struct {
+ dialer *nodedialer.Dialer
+ nl nodeLiveness
+ executor *sql.InternalExecutor
+ db *kv.DB
+}
+
+// Helper captures all the primitives required to fully specify a migration.
+type Helper struct {
+ *Manager
+}
+
+// nodeLiveness is the subset of the interface satisfied by CRDB's node liveness
+// component that the migration manager relies upon.
+type nodeLiveness interface {
+ GetLivenessesFromKV(context.Context) ([]livenesspb.Liveness, error)
+ IsLive(roachpb.NodeID) (bool, error)
+}
+
+// NewManager constructs a new Manager.
+//
+// TODO(irfansharif): We'll need to eventually plumb in on a lease manager here.
+func NewManager(
+ dialer *nodedialer.Dialer, nl nodeLiveness, executor *sql.InternalExecutor, db *kv.DB,
+) *Manager {
+ return &Manager{
+ dialer: dialer,
+ executor: executor,
+ db: db,
+ nl: nl,
+ }
+}
+
+// MigrateTo runs the set of migrations required to upgrade the cluster version
+// to the provided target version.
+//
+// TODO(irfansharif): Do something real here.
+func (m *Manager) MigrateTo(ctx context.Context, targetV roachpb.Version) error {
+ // TODO(irfansharif): Should we inject every ctx here with specific labels
+ // for each migration, so they log distinctly?
+ _ = logtags.AddTag(ctx, "migration-mgr", nil)
+
+ // TODO(irfansharif): We'll need to acquire a lease here and refresh it
+ // throughout during the migration to ensure mutual exclusion.
+
+ // TODO(irfansharif): We'll need to create a system table to store
+ // in-progress state of long running migrations, for introspection.
+
+ // TODO(irfansharif): We'll want to either write to a KV key to record the
+ // version up until which we've already migrated to, or consult the system
+ // table mentioned above. Perhaps it makes sense to consult any given
+ // `StoreClusterVersionKey`, since the manager here will want to push out
+ // cluster version bumps for vX before attempting to migrate into vX+1.
+
+ // TODO(irfansharif): After determining the last completed migration, if
+ // any, we'll be want to assemble the list of remaining migrations to step
+ // through to get to targetV.
+ _ = targetV
+ var vs []roachpb.Version
+
+ for _, version := range vs {
+ _ = &Helper{Manager: m}
+ // TODO(irfansharif): We'll want to out the version gate to every node
+ // in the cluster. Each node will want to persist the version, bump the
+ // local version gates, and then return. The migration associated with
+ // the specific version can then assume that every node in the cluster
+ // has the corresponding version activated.
+
+ // TODO(irfansharif): We'll want to retrieve the right migration off of
+ // our registry of migrations, and execute it.
+ // TODO(irfansharif): We'll want to be able to override which migration
+ // is retrieved here within tests. We could make the registry be a part
+ // of the manager, and all tests to provide their own.
+ _ = Registry[version]
+ }
+
+ return nil
+}

pkg/migration/migrations.go

@@ -0,0 +1,25 @@
+// Copyright 2020 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package migration
+
+import "github.com/cockroachdb/cockroach/pkg/roachpb"
+
+// Registry defines the global mapping between a version, and the associated
+// migration. The migration is only executed after a cluster-wide bump of the
+// version gate.
+var Registry = make(map[roachpb.Version]Migration)
+
+func init() {
+ // TODO(irfansharif): We'll want to register individual migrations with
+ // specific internal cluster versions here.
+ //
+ // Registry[clusterversion.VersionByKey(clusterversion.VersionWhatever)] = WhateverMigration
+}

pkg/server/BUILD.bazel

@@ -68,6 +68,7 @@ go_library(
  "//pkg/kv/kvserver/protectedts/ptprovider",
  "//pkg/kv/kvserver/protectedts/ptreconcile",
  "//pkg/kv/kvserver/reports",
+ "//pkg/migration",
  "//pkg/roachpb",
  "//pkg/rpc",
  "//pkg/rpc/nodedialer",

pkg/server/server_sql.go

@@ -32,6 +32,7 @@ import (
  "github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvtenant"
  "github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
  "github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts"
+ "github.com/cockroachdb/cockroach/pkg/migration"
  "github.com/cockroachdb/cockroach/pkg/roachpb"
  "github.com/cockroachdb/cockroach/pkg/rpc"
  "github.com/cockroachdb/cockroach/pkg/rpc/nodedialer"
@@ -111,7 +112,8 @@ type sqlServer struct {
 type sqlServerOptionalKVArgs struct {
  // nodesStatusServer gives access to the NodesStatus service.
  nodesStatusServer serverpb.OptionalNodesStatusServer
- // Narrowed down version of *NodeLiveness. Used by jobs and DistSQLPlanner
+ // Narrowed down version of *NodeLiveness. Used by jobs, DistSQLPlanner, and
+ // migration manager.
  nodeLiveness optionalnodeliveness.Container
  // Gossip is relied upon by distSQLCfg (execinfra.ServerConfig), the executor
  // config, the DistSQL planner, the table statistics cache, the statements
@@ -429,8 +431,9 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*sqlServer, error) {
  }
 
  var isLive func(roachpb.NodeID) (bool, error)
- if nl, ok := cfg.nodeLiveness.Optional(47900); ok {
- isLive = nl.IsLive
+ nodeLiveness, ok := cfg.nodeLiveness.Optional(47900)
+ if ok {
+ isLive = nodeLiveness.IsLive
  } else {
  // We're on a SQL tenant, so this is the only node DistSQL will ever
  // schedule on - always returning true is fine.
@@ -633,6 +636,21 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*sqlServer, error) {
  )
  execCfg.StmtDiagnosticsRecorder = stmtDiagnosticsRegistry
 
+ if cfg.TenantID == roachpb.SystemTenantID {
+ // We only need to attach a version upgrade hook if we're the system
+ // tenant. Regular tenants are disallowed from changing cluster
+ // versions.
+ migrationMgr := migration.NewManager(
+ cfg.nodeDialer,
+ nodeLiveness,
+ cfg.circularInternalExecutor,
+ cfg.db,
+ )
+ execCfg.VersionUpgradeHook = func(ctx context.Context, targetV roachpb.Version) error {
+ return migrationMgr.MigrateTo(ctx, targetV)
+ }
+ }
+
  temporaryObjectCleaner := sql.NewTemporaryObjectCleaner(
  cfg.Settings,
  cfg.db,

pkg/sql/exec_util.go

@@ -736,6 +736,11 @@ type ExecutorConfig struct {
  HydratedTables *hydratedtables.Cache
 
  GCJobNotifier *gcjobnotifier.Notifier
+
+ // VersionUpgradeHook is called after validating a `SET CLUSTER SETTING
+ // version` but before executing it. It can carry out arbitrary migrations
+ // that allow us to eventually remove legacy code.
+ VersionUpgradeHook func(ctx context.Context, to roachpb.Version) error
 }
 
 // Organization returns the value of cluster.organization.

pkg/sql/set_cluster_setting.go

@@ -19,6 +19,7 @@ import (
 
  "github.com/cockroachdb/cockroach/pkg/clusterversion"
  "github.com/cockroachdb/cockroach/pkg/kv"
+ "github.com/cockroachdb/cockroach/pkg/roachpb"
  "github.com/cockroachdb/cockroach/pkg/security"
  "github.com/cockroachdb/cockroach/pkg/server/telemetry"
  "github.com/cockroachdb/cockroach/pkg/settings"
@@ -47,6 +48,10 @@ type setClusterSettingNode struct {
  setting settings.WritableSetting
  // If value is nil, the setting should be reset.
  value tree.TypedExpr
+ // versionUpgradeHook is called after validating a `SET CLUSTER SETTING
+ // version` but before executing it. It can carry out arbitrary migrations
+ // that allow us to eventually remove legacy code.
+ versionUpgradeHook func(ctx context.Context, to roachpb.Version) error
 }
 
 func checkPrivilegesForSetting(ctx context.Context, p *planner, name string, action string) error {
@@ -156,7 +161,11 @@ func (p *planner) SetClusterSetting(
  }
  }
 
- return &setClusterSettingNode{name: name, st: st, setting: setting, value: value}, nil
+ csNode := setClusterSettingNode{
+ name: name, st: st, setting: setting, value: value,
+ versionUpgradeHook: p.execCfg.VersionUpgradeHook,
+ }
+ return &csNode, nil
 }
 
 func (n *setClusterSettingNode) startExec(params runParams) error {
@@ -209,7 +218,8 @@ func (n *setClusterSettingNode) startExec(params runParams) error {
  }
  reportedValue = tree.AsStringWithFlags(value, tree.FmtBareStrings)
  var prev tree.Datum
- if _, ok := n.setting.(*settings.VersionSetting); ok {
+ _, isSetVersion := n.setting.(*settings.VersionSetting)
+ if isSetVersion {
  datums, err := execCfg.InternalExecutor.QueryRowEx(
  ctx, "retrieve-prev-setting", txn,
  sessiondata.InternalExecutorOverride{User: security.RootUserName()},
@@ -232,6 +242,18 @@ func (n *setClusterSettingNode) startExec(params runParams) error {
  if err != nil {
  return err
  }
+
+ if isSetVersion {
+ // toSettingString already validated the input, and checked to
+ // see that we are allowed to transition. Let's call into our
+ // upgrade hook to run migrations, if any.
+ versionStr := string(*value.(*tree.DString))
+ targetVersion := roachpb.MustParseVersion(versionStr)
+ if err := n.versionUpgradeHook(ctx, targetVersion); err != nil {
+ return err
+ }
+ }
+
  if _, err = execCfg.InternalExecutor.ExecEx(
  ctx, "update-setting", txn,
  sessiondata.InternalExecutorOverride{User: security.RootUserName()},