etcdutl: Implement migrate --force command

Co-authored-by: nic-chen <33000667+nic-chen@users.noreply.github.com>
Co-authored-by: Heisenberg <yuzhiquanlong@gmail.com>

.github/workflows/e2e.yaml

@@ -21,7 +21,7 @@ jobs:
         echo "${TARGET}"
         case "${TARGET}" in
           linux-amd64-e2e)
-            PASSES='build release e2e' MANUAL_VER=v3.4.7 CPU='4' EXPECT_DEBUG='true' COVER='false' RACE='true' ./test.sh 2>&1 | tee test.log
+            PASSES='build release e2e' MANUAL_VER=v3.5.0 CPU='4' EXPECT_DEBUG='true' COVER='false' RACE='true' ./test.sh 2>&1 | tee test.log
             ! egrep "(--- FAIL:|FAIL:|DATA RACE|panic: test timed out|appears to have leaked)" -B50 -A10 test.log
             ;;
           linux-386-e2e)

CHANGELOG-3.6.md

@@ -15,6 +15,7 @@ See [code changes](https://github.com/etcd-io/etcd/compare/v3.5.0...v3.6.0).
 ### etcdutl v3
 
 - Add command to generate [shell completion](https://github.com/etcd-io/etcd/pull/13142).
+- Add `migrate` command for downgrading/upgrading etcd data dir files.
 
 ### Package `server`
 

etcdutl/ctl.go

@@ -45,6 +45,7 @@ func init() {
 		etcdutl.NewSnapshotCommand(),
 		etcdutl.NewVersionCommand(),
 		etcdutl.NewCompletionCommand(),
+		etcdutl.NewMigrateCommand(),
 	)
 }
 

etcdutl/etcdutl/migrate_command.go

@@ -0,0 +1,141 @@
+// Copyright 2021 The etcd Authors
+//
+// 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 etcdutl
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/coreos/go-semver/semver"
+	"github.com/spf13/cobra"
+	"go.uber.org/zap"
+
+	"go.etcd.io/etcd/pkg/v3/cobrautl"
+	"go.etcd.io/etcd/server/v3/storage/backend"
+	"go.etcd.io/etcd/server/v3/storage/datadir"
+	"go.etcd.io/etcd/server/v3/storage/schema"
+)
+
+// NewMigrateCommand prints out the version of etcd.
+func NewMigrateCommand() *cobra.Command {
+	o := newMigrateOptions()
+	cmd := &cobra.Command{
+		Use:   "migrate",
+		Short: "Migrates schema of etcd data dir files to make them compatible with different etcd version",
+		Run: func(cmd *cobra.Command, args []string) {
+			cfg, err := o.Config()
+			if err != nil {
+				cobrautl.ExitWithError(cobrautl.ExitBadArgs, err)
+			}
+			err = migrateCommandFunc(cfg)
+			if err != nil {
+				cobrautl.ExitWithError(cobrautl.ExitError, err)
+			}
+		},
+	}
+	o.AddFlags(cmd)
+	return cmd
+}
+
+type migrateOptions struct {
+	dataDir       string
+	targetVersion string
+	force         bool
+}
+
+func newMigrateOptions() *migrateOptions {
+	return &migrateOptions{}
+}
+
+func (o *migrateOptions) AddFlags(cmd *cobra.Command) {
+	cmd.Flags().StringVar(&o.dataDir, "data-dir", o.dataDir, "Path to the etcd data dir")
+	cmd.MarkFlagRequired("data-dir")
+	cmd.MarkFlagDirname("data-dir")
+
+	cmd.Flags().StringVar(&o.targetVersion, "target-version", o.targetVersion, `Target etcd version to migrate contents of data dir. Minmal value 3.5. Format "X.Y" for example 3.6.`)
+	cmd.MarkFlagRequired("target-version")
+
+	cmd.Flags().BoolVar(&o.force, "force", o.force, "Ignore migration failure and forcefully override storage version. Not recommended.")
+}
+
+func (o *migrateOptions) Config() (*migrateConfig, error) {
+	c := &migrateConfig{
+		force: o.force,
+	}
+	var err error
+	dotCount := strings.Count(o.targetVersion, ".")
+	if dotCount != 1 {
+		return nil, fmt.Errorf(`wrong target version format, expected "X.Y", got %q`, o.targetVersion)
+	}
+	c.targetVersion, err = semver.NewVersion(o.targetVersion + ".0")
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse target version: %w", err)
+	}
+	if c.targetVersion.LessThan(schema.V3_5) {
+		return nil, fmt.Errorf(`target version %q not supported. Minimal "3.5".`, storageVersionToString(c.targetVersion))
+	}
+
+	dbPath := datadir.ToBackendFileName(o.dataDir)
+	c.be = backend.NewDefaultBackend(dbPath)
+
+	return c, nil
+}
+
+type migrateConfig struct {
+	be            backend.Backend
+	targetVersion *semver.Version
+	force         bool
+}
+
+func migrateCommandFunc(c *migrateConfig) error {
+	defer c.be.Close()
+	lg := GetLogger()
+	tx := c.be.BatchTx()
+	tx.Lock()
+	current, err := schema.DetectSchemaVersion(lg, tx)
+	if err != nil {
+		tx.Unlock()
+		lg.Error("failed to detect storage version. Please make sure you are using data dir from etcd v3.5 and older")
+		return err
+	}
+	if *current == *c.targetVersion {
+		tx.Unlock()
+		lg.Info("storage version up-to-date", zap.String("storage-version", storageVersionToString(current)))
+		return nil
+	}
+	if c.force {
+		unsafeMigrateForce(lg, tx, c.targetVersion)
+		tx.Unlock()
+		c.be.ForceCommit()
+		return nil
+	}
+	tx.Unlock()
+	return fmt.Errorf("storage version migration is not yet supported")
+}
+
+func unsafeMigrateForce(lg *zap.Logger, tx backend.BatchTx, target *semver.Version) {
+	// Storage version is only supported since v3.6
+	if target.LessThan(schema.V3_6) {
+		schema.UnsafeClearStorageVersion(tx)
+		lg.Warn("forcefully cleared storage version")
+	} else {
+		schema.UnsafeSetStorageVersion(tx, target)
+		lg.Warn("forcefully set storage version", zap.String("storage-version", storageVersionToString(target)))
+	}
+}
+
+func storageVersionToString(ver *semver.Version) string {
+	return fmt.Sprintf("%d.%d", ver.Major, ver.Minor)
+}

etcdutl/go.mod

@@ -21,6 +21,7 @@ replace (
 )
 
 require (
+	github.com/coreos/go-semver v0.3.0
 	github.com/dustin/go-humanize v1.0.0
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/spf13/cobra v1.1.3

server/storage/schema/schema.go

@@ -32,7 +32,7 @@ var (
 func UpdateStorageSchema(lg *zap.Logger, tx backend.BatchTx) error {
 	tx.Lock()
 	defer tx.Unlock()
-	v, err := detectStorageVersion(lg, tx)
+	v, err := DetectSchemaVersion(lg, tx)
 	if err != nil {
 		return fmt.Errorf("cannot determine storage version: %w", err)
 	}
@@ -48,7 +48,7 @@ func UpdateStorageSchema(lg *zap.Logger, tx backend.BatchTx) error {
 	return nil
 }
 
-func detectStorageVersion(lg *zap.Logger, tx backend.ReadTx) (*semver.Version, error) {
+func DetectSchemaVersion(lg *zap.Logger, tx backend.ReadTx) (*semver.Version, error) {
 	v := UnsafeReadStorageVersion(tx)
 	if v != nil {
 		return v, nil

server/storage/schema/version.go

@@ -59,3 +59,8 @@ func UnsafeSetStorageVersion(tx backend.BatchTx, v *semver.Version) {
 	sv := semver.Version{Major: v.Major, Minor: v.Minor}
 	tx.UnsafePut(Meta, MetaStorageVersionName, []byte(sv.String()))
 }
+
+// UnsafeClearStorageVersion removes etcd storage version in backend.
+func UnsafeClearStorageVersion(tx backend.BatchTx) {
+	tx.UnsafeDelete(Meta, MetaStorageVersionName)
+}

tests/e2e/utl_migrate_test.go

@@ -0,0 +1,163 @@
+// Copyright 2021 The etcd Authors
+//
+// 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 e2e
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/coreos/go-semver/semver"
+	"github.com/stretchr/testify/assert"
+	"go.etcd.io/etcd/client/pkg/v3/fileutil"
+	"go.etcd.io/etcd/server/v3/storage/backend"
+	"go.etcd.io/etcd/server/v3/storage/schema"
+)
+
+func TestEtctlutlMigrate(t *testing.T) {
+	lastReleaseBinary := binDir + "/etcd-last-release"
+
+	tcs := []struct {
+		name          string
+		targetVersion string
+		binary        string
+		force         bool
+
+		expectLogsSubString  string
+		expectStorageVersion *semver.Version
+	}{
+		{
+			name:                 "Invalid target version string",
+			targetVersion:        "abc",
+			expectLogsSubString:  `Error: wrong target version format, expected "X.Y", got "abc"`,
+			expectStorageVersion: &schema.V3_6,
+		},
+		{
+			name:                 "Invalid target version",
+			targetVersion:        "3.a",
+			expectLogsSubString:  `Error: failed to parse target version: strconv.ParseInt: parsing "a": invalid syntax`,
+			expectStorageVersion: &schema.V3_6,
+		},
+		{
+			name:                 "Target with only major version is invalid",
+			targetVersion:        "3",
+			expectLogsSubString:  `Error: wrong target version format, expected "X.Y", got "3"`,
+			expectStorageVersion: &schema.V3_6,
+		},
+		{
+			name:                 "Target with patch version is invalid",
+			targetVersion:        "3.6.0",
+			expectLogsSubString:  `Error: wrong target version format, expected "X.Y", got "3.6.0"`,
+			expectStorageVersion: &schema.V3_6,
+		},
+		{
+			name:                "Migrate v3.5 to v3.5 is no-op",
+			binary:              lastReleaseBinary,
+			targetVersion:       "3.5",
+			expectLogsSubString: "storage version up-to-date\t" + `{"storage-version": "3.5"}`,
+		},
+		{
+			name:                "Upgrade v3.5 to v3.6 should fail until it's implemented",
+			binary:              lastReleaseBinary,
+			targetVersion:       "3.6",
+			expectLogsSubString: "Error: storage version migration is not yet supported",
+		},
+		{
+			name:                 "Migrate v3.6 to v3.6 is no-op",
+			targetVersion:        "3.6",
+			expectLogsSubString:  "storage version up-to-date\t" + `{"storage-version": "3.6"}`,
+			expectStorageVersion: &schema.V3_6,
+		},
+		{
+			name:                 "Downgrade v3.6 to v3.5 should fail until it's implemented",
+			targetVersion:        "3.5",
+			expectLogsSubString:  "Error: storage version migration is not yet supported",
+			expectStorageVersion: &schema.V3_6,
+		},
+		{
+			name:                "Downgrade v3.6 to v3.5 with force should work",
+			targetVersion:       "3.5",
+			force:               true,
+			expectLogsSubString: "forcfully cleared storage version",
+		},
+		{
+			name:                 "Upgrade v3.6 to v3.7 with force should work",
+			targetVersion:        "3.7",
+			force:                true,
+			expectLogsSubString:  "forcfully set storage version\t" + `{"storage-version": "3.7"}`,
+			expectStorageVersion: &semver.Version{Major: 3, Minor: 7},
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			BeforeTest(t)
+			if tc.binary != "" && !fileutil.Exist(tc.binary) {
+				t.Skipf("%q does not exist", lastReleaseBinary)
+			}
+			dataDirPath := t.TempDir()
+
+			epc, err := newEtcdProcessCluster(t, &etcdProcessClusterConfig{
+				execPath:     tc.binary,
+				dataDirPath:  dataDirPath,
+				clusterSize:  1,
+				initialToken: "new",
+				keepDataDir:  true,
+				// Set low snapshotCount to ensure wal snapshot is done
+				snapshotCount: 1,
+			})
+			if err != nil {
+				t.Fatalf("could not start etcd process cluster (%v)", err)
+			}
+			defer func() {
+				if errC := epc.Close(); errC != nil {
+					t.Fatalf("error closing etcd processes (%v)", errC)
+				}
+			}()
+
+			dialTimeout := 10 * time.Second
+			prefixArgs := []string{ctlBinPath, "--endpoints", strings.Join(epc.EndpointsV3(), ","), "--dial-timeout", dialTimeout.String()}
+
+			t.Log("Write keys to ensure wal snapshot is created and all v3.5 fields are set...")
+			for i := 0; i < 10; i++ {
+				if err = spawnWithExpect(append(prefixArgs, "put", fmt.Sprintf("%d", i), "value"), "OK"); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			t.Log("Stopping the server...")
+			if err = epc.procs[0].Stop(); err != nil {
+				t.Fatal(err)
+			}
+
+			t.Log("etcdutl migrate...")
+			args := []string{utlBinPath, "migrate", "--data-dir", dataDirPath, "--target-version", tc.targetVersion}
+			if tc.force {
+				args = append(args, "--force")
+			}
+			err = spawnWithExpect(args, tc.expectLogsSubString)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			t.Log("etcdutl migrate...")
+			be := backend.NewDefaultBackend(dataDirPath + "/member/snap/db")
+			defer be.Close()
+
+			ver := schema.ReadStorageVersion(be.ReadTx())
+			assert.Equal(t, tc.expectStorageVersion, ver)
+		})
+	}
+}

tests/go.mod

@@ -15,6 +15,7 @@ replace (
 )
 
 require (
+	github.com/coreos/go-semver v0.3.0
 	github.com/dustin/go-humanize v1.0.0
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.5.2