Merge pull request #19451 from fuweid/add-downgrade-status

*: support DowngradeInfo field in maintenence.Status API

Author: ahrtr

Documentation/dev-guide/apispec/swagger/rpc.swagger.json

@@ -2135,6 +2135,19 @@
         }
       }
     },
+    "etcdserverpbDowngradeInfo": {
+      "type": "object",
+      "properties": {
+        "enabled": {
+          "type": "boolean",
+          "description": "enabled indicates whether the cluster is enabled to downgrade."
+        },
+        "targetVersion": {
+          "type": "string",
+          "description": "targetVersion is the target downgrade version."
+        }
+      }
+    },
     "etcdserverpbDowngradeRequest": {
       "type": "object",
       "properties": {
@@ -2840,6 +2853,10 @@
           "type": "string",
           "format": "int64",
           "title": "dbSizeQuota is the configured etcd storage quota in bytes (the value passed to etcd instance by flag --quota-backend-bytes)"
+        },
+        "downgradeInfo": {
+          "$ref": "#/definitions/etcdserverpbDowngradeInfo",
+          "description": "downgradeInfo indicates if there is downgrade process."
         }
       }
     },

api/etcdserverpb/rpc.pb.go

@@ -1198,6 +1198,15 @@ message StatusResponse {
   string storageVersion = 11 [(versionpb.etcd_version_field)="3.6"];
   // dbSizeQuota is the configured etcd storage quota in bytes (the value passed to etcd instance by flag --quota-backend-bytes)
   int64 dbSizeQuota = 12 [(versionpb.etcd_version_field)="3.6"];
+  // downgradeInfo indicates if there is downgrade process.
+  DowngradeInfo downgradeInfo = 13 [(versionpb.etcd_version_field)="3.6"];
+}
+
+message DowngradeInfo {
+	// enabled indicates whether the cluster is enabled to downgrade.
+  bool enabled = 1;
+	// targetVersion is the target downgrade version.
+  string targetVersion = 2;
 }
 
 message AuthEnableRequest {

api/etcdserverpb/rpc.proto

@@ -17,6 +17,7 @@ package command
 import (
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 
 	"github.com/dustin/go-humanize"
@@ -220,7 +221,7 @@ func makeEndpointHealthTable(healthList []epHealth) (hdr []string, rows [][]stri
 func makeEndpointStatusTable(statusList []epStatus) (hdr []string, rows [][]string) {
 	hdr = []string{
 		"endpoint", "ID", "version", "storage version", "db size", "in use", "percentage not in use", "quota", "is leader", "is learner", "raft term",
-		"raft index", "raft applied index", "errors",
+		"raft index", "raft applied index", "errors", "downgrade target version", "downgrade enabled",
 	}
 	for _, status := range statusList {
 		rows = append(rows, []string{
@@ -238,6 +239,8 @@ func makeEndpointStatusTable(statusList []epStatus) (hdr []string, rows [][]stri
 			fmt.Sprint(status.Resp.RaftIndex),
 			fmt.Sprint(status.Resp.RaftAppliedIndex),
 			fmt.Sprint(strings.Join(status.Resp.Errors, ", ")),
+			status.Resp.DowngradeInfo.GetTargetVersion(),
+			strconv.FormatBool(status.Resp.DowngradeInfo.GetEnabled()),
 		})
 	}
 	return hdr, rows

etcdctl/ctlv3/command/printer.go

@@ -203,6 +203,8 @@ func (p *fieldsPrinter) EndpointStatus(eps []epStatus) {
 		fmt.Println(`"RaftAppliedIndex" :`, ep.Resp.RaftAppliedIndex)
 		fmt.Println(`"Errors" :`, ep.Resp.Errors)
 		fmt.Printf("\"Endpoint\" : %q\n", ep.Ep)
+		fmt.Printf("\"DowngradeTargetVersion\" : %q\n", ep.Resp.DowngradeInfo.GetTargetVersion())
+		fmt.Println(`"DowngradeEnabled" :`, ep.Resp.DowngradeInfo.GetEnabled())
 		fmt.Println()
 	}
 }

etcdctl/ctlv3/command/printer_fields.go

@@ -150,6 +150,9 @@ etcdserverpb.DeleteRangeResponse: "3.0"
 etcdserverpb.DeleteRangeResponse.deleted: ""
 etcdserverpb.DeleteRangeResponse.header: ""
 etcdserverpb.DeleteRangeResponse.prev_kvs: "3.1"
+etcdserverpb.DowngradeInfo: ""
+etcdserverpb.DowngradeInfo.enabled: ""
+etcdserverpb.DowngradeInfo.targetVersion: ""
 etcdserverpb.DowngradeRequest: "3.5"
 etcdserverpb.DowngradeRequest.CANCEL: ""
 etcdserverpb.DowngradeRequest.DowngradeAction: "3.5"
@@ -382,6 +385,7 @@ etcdserverpb.StatusResponse: "3.0"
 etcdserverpb.StatusResponse.dbSize: ""
 etcdserverpb.StatusResponse.dbSizeInUse: "3.4"
 etcdserverpb.StatusResponse.dbSizeQuota: "3.6"
+etcdserverpb.StatusResponse.downgradeInfo: "3.6"
 etcdserverpb.StatusResponse.errors: "3.4"
 etcdserverpb.StatusResponse.header: ""
 etcdserverpb.StatusResponse.isLearner: "3.4"

scripts/etcd_version_annotations.txt

@@ -262,10 +262,17 @@ func (ms *maintenanceServer) Status(ctx context.Context, ar *pb.StatusRequest) (
 		DbSizeInUse:      ms.bg.Backend().SizeInUse(),
 		IsLearner:        ms.cs.IsLearner(),
 		DbSizeQuota:      ms.cg.Config().QuotaBackendBytes,
+		DowngradeInfo:    &pb.DowngradeInfo{Enabled: false},
 	}
 	if storageVersion := ms.vs.GetStorageVersion(); storageVersion != nil {
 		resp.StorageVersion = storageVersion.String()
 	}
+	if downgradeInfo := ms.vs.GetDowngradeInfo(); downgradeInfo != nil {
+		resp.DowngradeInfo = &pb.DowngradeInfo{
+			Enabled:       downgradeInfo.Enabled,
+			TargetVersion: downgradeInfo.TargetVersion,
+		}
+	}
 	if resp.Leader == raft.None {
 		resp.Errors = append(resp.Errors, errors.ErrNoLeader.Error())
 	}

server/etcdserver/api/v3rpc/maintenance.go

@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/version"
 	"go.etcd.io/etcd/client/pkg/v3/fileutil"
 	"go.etcd.io/etcd/client/pkg/v3/types"
@@ -51,6 +52,10 @@ func TestDowngradeUpgradeClusterOf1(t *testing.T) {
 	testDowngradeUpgrade(t, 1, 1, false, noCancellation)
 }
 
+func TestDowngradeUpgrade2InClusterOf3(t *testing.T) {
+	testDowngradeUpgrade(t, 2, 3, false, noCancellation)
+}
+
 func TestDowngradeUpgradeClusterOf3(t *testing.T) {
 	testDowngradeUpgrade(t, 3, 3, false, noCancellation)
 }
@@ -128,6 +133,9 @@ func testDowngradeUpgrade(t *testing.T, numberOfMembersToDowngrade int, clusterS
 		time.Sleep(etcdserver.HealthInterval)
 	}
 
+	t.Log("Downgrade should be disabled")
+	e2e.ValidateDowngradeInfo(t, epc, &pb.DowngradeInfo{Enabled: false})
+
 	t.Log("Adding member to test membership, but a learner avoid breaking quorum")
 	resp, err := cc.MemberAddAsLearner(context.Background(), "fake1", []string{"http://127.0.0.1:1001"})
 	require.NoError(t, err)
@@ -150,6 +158,10 @@ func testDowngradeUpgrade(t *testing.T, numberOfMembersToDowngrade int, clusterS
 		return // No need to perform downgrading, end the test here
 	}
 	e2e.DowngradeEnable(t, epc, lastVersion)
+
+	t.Log("Downgrade should be enabled")
+	e2e.ValidateDowngradeInfo(t, epc, &pb.DowngradeInfo{Enabled: true, TargetVersion: lastClusterVersion.String()})
+
 	if triggerCancellation == cancelRightAfterEnable {
 		t.Logf("Cancelling downgrade right after enabling (no node is downgraded yet)")
 		e2e.DowngradeCancel(t, epc)
@@ -165,7 +177,7 @@ func testDowngradeUpgrade(t *testing.T, numberOfMembersToDowngrade int, clusterS
 	err = e2e.DowngradeUpgradeMembersByID(t, nil, epc, membersToChange, currentVersion, lastClusterVersion)
 	require.NoError(t, err)
 	if len(membersToChange) == len(epc.Procs) {
-		e2e.AssertProcessLogs(t, leader(t, epc), "the cluster has been downgraded")
+		e2e.AssertProcessLogs(t, epc.Procs[epc.WaitLeader(t)], "the cluster has been downgraded")
 	}
 
 	t.Log("Downgrade complete")
@@ -202,6 +214,14 @@ func testDowngradeUpgrade(t *testing.T, numberOfMembersToDowngrade int, clusterS
 	require.NoError(t, err)
 	t.Log("Upgrade complete")
 
+	if triggerCancellation == noCancellation && numberOfMembersToDowngrade < clusterSize {
+		t.Log("Downgrade should be still enabled")
+		e2e.ValidateDowngradeInfo(t, epc, &pb.DowngradeInfo{Enabled: true, TargetVersion: lastClusterVersion.String()})
+	} else {
+		t.Log("Downgrade should be disabled")
+		e2e.ValidateDowngradeInfo(t, epc, &pb.DowngradeInfo{Enabled: false})
+	}
+
 	afterMembers, afterKV = getMembersAndKeys(t, cc)
 	assert.Equal(t, beforeKV.Kvs, afterKV.Kvs)
 	assert.Equal(t, beforeMembers.Members, afterMembers.Members)
@@ -224,27 +244,6 @@ func newCluster(t *testing.T, clusterSize int, snapshotCount uint64) *e2e.EtcdPr
 	return epc
 }
 
-func leader(t *testing.T, epc *e2e.EtcdProcessCluster) e2e.EtcdProcess {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-	defer cancel()
-	for i := 0; i < len(epc.Procs); i++ {
-		endpoints := epc.Procs[i].EndpointsGRPC()
-		cli, err := clientv3.New(clientv3.Config{
-			Endpoints:   endpoints,
-			DialTimeout: 3 * time.Second,
-		})
-		require.NoError(t, err)
-		defer cli.Close()
-		resp, err := cli.Status(ctx, endpoints[0])
-		require.NoError(t, err)
-		if resp.Header.GetMemberId() == resp.Leader {
-			return epc.Procs[i]
-		}
-	}
-	t.Fatal("Leader not found")
-	return nil
-}
-
 func generateSnapshot(t *testing.T, snapshotCount uint64, cc *e2e.EtcdctlV3) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

tests/e2e/cluster_downgrade_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 
+	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/version"
 	"go.etcd.io/etcd/tests/v3/framework/testutils"
 )
@@ -46,7 +47,6 @@ func DowngradeEnable(t *testing.T, epc *EtcdProcessCluster, ver *semver.Version)
 			Server:  OffsetMinor(ver, 1).String(),
 			Storage: ver.String(),
 		})
-		AssertProcessLogs(t, epc.Procs[i], "The server is ready to downgrade")
 	}
 
 	t.Log("Cluster is ready for downgrade")
@@ -82,6 +82,51 @@ func DowngradeCancel(t *testing.T, epc *EtcdProcessCluster) {
 	t.Log("Cluster downgrade cancellation is completed")
 }
 
+func ValidateDowngradeInfo(t *testing.T, clus *EtcdProcessCluster, expected *pb.DowngradeInfo) {
+	cfg := clus.Cfg
+
+	for i := 0; i < len(clus.Procs); i++ {
+		member := clus.Procs[i]
+		mc := member.Etcdctl()
+		mName := member.Config().Name
+
+		testutils.ExecuteWithTimeout(t, 1*time.Minute, func() {
+			for {
+				statuses, err := mc.Status(context.Background())
+				if err != nil {
+					cfg.Logger.Warn("failed to get member status and retrying",
+						zap.Error(err),
+						zap.String("member", mName))
+
+					time.Sleep(time.Second)
+					continue
+				}
+
+				require.Lenf(t, statuses, 1, "member %s", mName)
+				got := (*pb.StatusResponse)(statuses[0]).GetDowngradeInfo()
+
+				if got.GetEnabled() == expected.GetEnabled() && got.GetTargetVersion() == expected.GetTargetVersion() {
+					cfg.Logger.Info("DowngradeInfo match", zap.String("member", mName))
+					break
+				}
+
+				cfg.Logger.Warn("DowngradeInfo didn't match retrying",
+					zap.String("member", mName),
+					zap.Dict("expected",
+						zap.Bool("Enabled", expected.GetEnabled()),
+						zap.String("TargetVersion", expected.GetTargetVersion()),
+					),
+					zap.Dict("got",
+						zap.Bool("Enabled", got.GetEnabled()),
+						zap.String("TargetVersion", got.GetTargetVersion()),
+					),
+				)
+				time.Sleep(time.Second)
+			}
+		})
+	}
+}
+
 func DowngradeUpgradeMembers(t *testing.T, lg *zap.Logger, clus *EtcdProcessCluster, numberOfMembersToChange int, currentVersion, targetVersion *semver.Version) error {
 	membersToChange := rand.Perm(len(clus.Procs))[:numberOfMembersToChange]
 	t.Logf("Elect members for operations on members: %v", membersToChange)
@@ -100,7 +145,6 @@ func DowngradeUpgradeMembersByID(t *testing.T, lg *zap.Logger, clus *EtcdProcess
 		opString = "downgrading"
 		newExecPath = BinPath.EtcdLastRelease
 	}
-
 	for _, memberID := range membersToChange {
 		member := clus.Procs[memberID]
 		if member.Config().ExecPath == newExecPath {
@@ -117,11 +161,16 @@ func DowngradeUpgradeMembersByID(t *testing.T, lg *zap.Logger, clus *EtcdProcess
 			return err
 		}
 	}
+
+	clusterVersion := targetVersion.String()
+	if !isDowngrade && len(membersToChange) != len(clus.Procs) {
+		clusterVersion = currentVersion.String()
+	}
 	lg.Info("Validating versions")
 	for _, memberID := range membersToChange {
 		member := clus.Procs[memberID]
 		ValidateVersion(t, clus.Cfg, member, version.Versions{
-			Cluster: targetVersion.String(),
+			Cluster: clusterVersion,
 			Server:  targetVersion.String(),
 		})
 	}

tests/framework/e2e/downgrade.go

@@ -232,6 +232,13 @@ func (f memberDowngradeUpgrade) Inject(ctx context.Context, t *testing.T, lg *za
 	if err != nil {
 		return nil, err
 	}
+
+	// NOTE: By default, the leader can cancel the downgrade once all members
+	// have reached the target version. However, determining the final stable
+	// cluster version after an upgrade can be challenging. To ensure stability,
+	// we should wait for leader to cancel downgrade process.
+	e2e.AssertProcessLogs(t, clus.Procs[clus.WaitLeader(t)], "the cluster has been downgraded")
+
 	// partial upgrade the cluster
 	numberOfMembersToUpgrade := rand.Int()%len(clus.Procs) + 1
 	err = e2e.DowngradeUpgradeMembers(t, lg, clus, numberOfMembersToUpgrade, lastVersion, currentVersion)