Encapsulation of applier logic: Move Txn related code out of applier.go.

server/embed/etcd.go

@@ -275,7 +275,7 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) {
 
 	e.cfg.logger.Info(
 		"now serving peer/client/metrics",
-		zap.String("local-member-id", e.Server.ID().String()),
+		zap.String("local-member-id", e.Server.MemberId().String()),
 		zap.Strings("initial-advertise-peer-urls", e.cfg.getAPURLs()),
 		zap.Strings("listen-peer-urls", e.cfg.getLPURLs()),
 		zap.Strings("advertise-client-urls", e.cfg.getACURLs()),

server/etcdmain/etcd.go

@@ -25,9 +25,8 @@ import (
 	"go.etcd.io/etcd/client/pkg/v3/types"
 	"go.etcd.io/etcd/pkg/v3/osutil"
 	"go.etcd.io/etcd/server/v3/embed"
-	"go.etcd.io/etcd/server/v3/etcdserver"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/v2discovery"
-
+	"go.etcd.io/etcd/server/v3/etcdserver/errors"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 )
@@ -127,7 +126,7 @@ func startEtcdOrProxyV2(args []string) {
 	}
 
 	if err != nil {
-		if derr, ok := err.(*etcdserver.DiscoveryError); ok {
+		if derr, ok := err.(*errors.DiscoveryError); ok {
 			switch derr.Err {
 			case v2discovery.ErrDuplicateID:
 				lg.Warn(

server/etcdserver/adapters.go

@@ -70,7 +70,7 @@ func (s *serverVersionAdapter) GetDowngradeInfo() *serverversion.DowngradeInfo {
 }
 
 func (s *serverVersionAdapter) GetMembersVersions() map[string]*version.Versions {
-	return getMembersVersions(s.lg, s.cluster, s.id, s.peerRt, s.Cfg.ReqTimeout())
+	return getMembersVersions(s.lg, s.cluster, s.MemberId(), s.peerRt, s.Cfg.ReqTimeout())
 }
 
 func (s *serverVersionAdapter) GetStorageVersion() *semver.Version {

server/etcdserver/api/etcdhttp/peer.go

@@ -26,6 +26,7 @@ import (
 	"go.etcd.io/etcd/server/v3/etcdserver/api"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/membership"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
+	"go.etcd.io/etcd/server/v3/etcdserver/errors"
 	"go.etcd.io/etcd/server/v3/lease/leasehttp"
 
 	"go.uber.org/zap"
@@ -142,7 +143,7 @@ func (h *peerMemberPromoteHandler) ServeHTTP(w http.ResponseWriter, r *http.Requ
 			http.Error(w, err.Error(), http.StatusNotFound)
 		case membership.ErrMemberNotLearner:
 			http.Error(w, err.Error(), http.StatusPreconditionFailed)
-		case etcdserver.ErrLearnerNotReady:
+		case errors.ErrLearnerNotReady:
 			http.Error(w, err.Error(), http.StatusPreconditionFailed)
 		default:
 			writeError(h.lg, w, r, err)

server/etcdserver/api/etcdhttp/utils.go

@@ -17,9 +17,9 @@ package etcdhttp
 import (
 	"net/http"
 
-	"go.etcd.io/etcd/server/v3/etcdserver"
 	httptypes "go.etcd.io/etcd/server/v3/etcdserver/api/etcdhttp/types"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/v2error"
+	"go.etcd.io/etcd/server/v3/etcdserver/errors"
 	"go.uber.org/zap"
 )
 
@@ -57,8 +57,8 @@ func writeError(lg *zap.Logger, w http.ResponseWriter, r *http.Request, err erro
 
 	default:
 		switch err {
-		case etcdserver.ErrTimeoutDueToLeaderFail, etcdserver.ErrTimeoutDueToConnectionLost, etcdserver.ErrNotEnoughStartedMembers,
-			etcdserver.ErrUnhealthy:
+		case errors.ErrTimeoutDueToLeaderFail, errors.ErrTimeoutDueToConnectionLost, errors.ErrNotEnoughStartedMembers,
+			errors.ErrUnhealthy:
 			if lg != nil {
 				lg.Warn(
 					"v2 response error",

server/etcdserver/api/v3rpc/header.go

@@ -17,19 +17,20 @@ package v3rpc
 import (
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/server/v3/etcdserver"
+	"go.etcd.io/etcd/server/v3/etcdserver/apply"
 )
 
 type header struct {
 	clusterID int64
 	memberID  int64
-	sg        etcdserver.RaftStatusGetter
+	sg        apply.RaftStatusGetter
 	rev       func() int64
 }
 
 func newHeader(s *etcdserver.EtcdServer) header {
 	return header{
 		clusterID: int64(s.Cluster().ID()),
-		memberID:  int64(s.ID()),
+		memberID:  int64(s.MemberId()),
 		sg:        s,
 		rev:       func() int64 { return s.KV().Rev() },
 	}

server/etcdserver/api/v3rpc/interceptor.go

@@ -49,7 +49,7 @@ func newUnaryInterceptor(s *etcdserver.EtcdServer) grpc.UnaryServerInterceptor {
 			return nil, rpctypes.ErrGRPCNotCapable
 		}
 
-		if s.IsMemberExist(s.ID()) && s.IsLearner() && !isRPCSupportedForLearner(req) {
+		if s.IsMemberExist(s.MemberId()) && s.IsLearner() && !isRPCSupportedForLearner(req) {
 			return nil, rpctypes.ErrGRPCNotSupportedForLearner
 		}
 
@@ -218,7 +218,7 @@ func newStreamInterceptor(s *etcdserver.EtcdServer) grpc.StreamServerInterceptor
 			return rpctypes.ErrGRPCNotCapable
 		}
 
-		if s.IsMemberExist(s.ID()) && s.IsLearner() && info.FullMethod != snapshotMethod { // learner does not support stream RPC except Snapshot
+		if s.IsMemberExist(s.MemberId()) && s.IsLearner() && info.FullMethod != snapshotMethod { // learner does not support stream RPC except Snapshot
 			return rpctypes.ErrGRPCNotSupportedForLearner
 		}
 

server/etcdserver/api/v3rpc/maintenance.go

@@ -27,6 +27,8 @@ import (
 	"go.etcd.io/etcd/raft/v3"
 	"go.etcd.io/etcd/server/v3/auth"
 	"go.etcd.io/etcd/server/v3/etcdserver"
+	"go.etcd.io/etcd/server/v3/etcdserver/apply"
+	"go.etcd.io/etcd/server/v3/etcdserver/errors"
 	serverversion "go.etcd.io/etcd/server/v3/etcdserver/version"
 	"go.etcd.io/etcd/server/v3/storage/backend"
 	"go.etcd.io/etcd/server/v3/storage/mvcc"
@@ -69,7 +71,7 @@ type ClusterStatusGetter interface {
 
 type maintenanceServer struct {
 	lg  *zap.Logger
-	rg  etcdserver.RaftStatusGetter
+	rg  apply.RaftStatusGetter
 	kg  KVGetter
 	bg  BackendGetter
 	a   Alarmer
@@ -241,7 +243,7 @@ func (ms *maintenanceServer) Status(ctx context.Context, ar *pb.StatusRequest) (
 		resp.StorageVersion = storageVersion.String()
 	}
 	if resp.Leader == raft.None {
-		resp.Errors = append(resp.Errors, etcdserver.ErrNoLeader.Error())
+		resp.Errors = append(resp.Errors, errors.ErrNoLeader.Error())
 	}
 	for _, a := range ms.a.Alarms() {
 		resp.Errors = append(resp.Errors, a.String())
@@ -250,7 +252,7 @@ func (ms *maintenanceServer) Status(ctx context.Context, ar *pb.StatusRequest) (
 }
 
 func (ms *maintenanceServer) MoveLeader(ctx context.Context, tr *pb.MoveLeaderRequest) (*pb.MoveLeaderResponse, error) {
-	if ms.rg.ID() != ms.rg.Leader() {
+	if ms.rg.MemberId() != ms.rg.Leader() {
 		return nil, rpctypes.ErrGRPCNotLeader
 	}
 

server/etcdserver/api/v3rpc/member.go

@@ -106,7 +106,7 @@ func (cs *ClusterServer) MemberPromote(ctx context.Context, r *pb.MemberPromoteR
 }
 
 func (cs *ClusterServer) header() *pb.ResponseHeader {
-	return &pb.ResponseHeader{ClusterId: uint64(cs.cluster.ID()), MemberId: uint64(cs.server.ID()), RaftTerm: cs.server.Term()}
+	return &pb.ResponseHeader{ClusterId: uint64(cs.cluster.ID()), MemberId: uint64(cs.server.MemberId()), RaftTerm: cs.server.Term()}
 }
 
 func membersToProtoMembers(membs []*membership.Member) []*pb.Member {

server/etcdserver/api/v3rpc/quota.go

@@ -53,7 +53,7 @@ func (qa *quotaAlarmer) check(ctx context.Context, r interface{}) error {
 func NewQuotaKVServer(s *etcdserver.EtcdServer) pb.KVServer {
 	return &quotaKVServer{
 		NewKVServer(s),
-		quotaAlarmer{storage.NewBackendQuota(s.Cfg, s.Backend(), "kv"), s, s.ID()},
+		quotaAlarmer{newBackendQuota(s, "kv"), s, s.MemberId()},
 	}
 }
 
@@ -86,6 +86,10 @@ func (s *quotaLeaseServer) LeaseGrant(ctx context.Context, cr *pb.LeaseGrantRequ
 func NewQuotaLeaseServer(s *etcdserver.EtcdServer) pb.LeaseServer {
 	return &quotaLeaseServer{
 		NewLeaseServer(s),
-		quotaAlarmer{storage.NewBackendQuota(s.Cfg, s.Backend(), "lease"), s, s.ID()},
+		quotaAlarmer{newBackendQuota(s, "lease"), s, s.MemberId()},
 	}
 }
+
+func newBackendQuota(s *etcdserver.EtcdServer, name string) storage.Quota {
+	return storage.NewBackendQuota(s.Logger(), s.Cfg.QuotaBackendBytes, s.Backend(), name)
+}

server/etcdserver/api/v3rpc/util.go

@@ -21,8 +21,8 @@ import (
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 	"go.etcd.io/etcd/server/v3/auth"
-	"go.etcd.io/etcd/server/v3/etcdserver"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/membership"
+	"go.etcd.io/etcd/server/v3/etcdserver/errors"
 	"go.etcd.io/etcd/server/v3/etcdserver/version"
 	"go.etcd.io/etcd/server/v3/lease"
 	"go.etcd.io/etcd/server/v3/storage/mvcc"
@@ -32,39 +32,39 @@ import (
 )
 
 var toGRPCErrorMap = map[error]error{
-	membership.ErrIDRemoved:               rpctypes.ErrGRPCMemberNotFound,
-	membership.ErrIDNotFound:              rpctypes.ErrGRPCMemberNotFound,
-	membership.ErrIDExists:                rpctypes.ErrGRPCMemberExist,
-	membership.ErrPeerURLexists:           rpctypes.ErrGRPCPeerURLExist,
-	membership.ErrMemberNotLearner:        rpctypes.ErrGRPCMemberNotLearner,
-	membership.ErrTooManyLearners:         rpctypes.ErrGRPCTooManyLearners,
-	etcdserver.ErrNotEnoughStartedMembers: rpctypes.ErrMemberNotEnoughStarted,
-	etcdserver.ErrLearnerNotReady:         rpctypes.ErrGRPCLearnerNotReady,
-
-	mvcc.ErrCompacted:             rpctypes.ErrGRPCCompacted,
-	mvcc.ErrFutureRev:             rpctypes.ErrGRPCFutureRev,
-	etcdserver.ErrRequestTooLarge: rpctypes.ErrGRPCRequestTooLarge,
-	etcdserver.ErrNoSpace:         rpctypes.ErrGRPCNoSpace,
-	etcdserver.ErrTooManyRequests: rpctypes.ErrTooManyRequests,
-
-	etcdserver.ErrNoLeader:                   rpctypes.ErrGRPCNoLeader,
-	etcdserver.ErrNotLeader:                  rpctypes.ErrGRPCNotLeader,
-	etcdserver.ErrLeaderChanged:              rpctypes.ErrGRPCLeaderChanged,
-	etcdserver.ErrStopped:                    rpctypes.ErrGRPCStopped,
-	etcdserver.ErrTimeout:                    rpctypes.ErrGRPCTimeout,
-	etcdserver.ErrTimeoutDueToLeaderFail:     rpctypes.ErrGRPCTimeoutDueToLeaderFail,
-	etcdserver.ErrTimeoutDueToConnectionLost: rpctypes.ErrGRPCTimeoutDueToConnectionLost,
-	etcdserver.ErrTimeoutWaitAppliedIndex:    rpctypes.ErrGRPCTimeoutWaitAppliedIndex,
-	etcdserver.ErrUnhealthy:                  rpctypes.ErrGRPCUnhealthy,
-	etcdserver.ErrKeyNotFound:                rpctypes.ErrGRPCKeyNotFound,
-	etcdserver.ErrCorrupt:                    rpctypes.ErrGRPCCorrupt,
-	etcdserver.ErrBadLeaderTransferee:        rpctypes.ErrGRPCBadLeaderTransferee,
-
-	etcdserver.ErrClusterVersionUnavailable:   rpctypes.ErrGRPCClusterVersionUnavailable,
-	etcdserver.ErrWrongDowngradeVersionFormat: rpctypes.ErrGRPCWrongDowngradeVersionFormat,
-	version.ErrInvalidDowngradeTargetVersion:  rpctypes.ErrGRPCInvalidDowngradeTargetVersion,
-	version.ErrDowngradeInProcess:             rpctypes.ErrGRPCDowngradeInProcess,
-	version.ErrNoInflightDowngrade:            rpctypes.ErrGRPCNoInflightDowngrade,
+	membership.ErrIDRemoved:           rpctypes.ErrGRPCMemberNotFound,
+	membership.ErrIDNotFound:          rpctypes.ErrGRPCMemberNotFound,
+	membership.ErrIDExists:            rpctypes.ErrGRPCMemberExist,
+	membership.ErrPeerURLexists:       rpctypes.ErrGRPCPeerURLExist,
+	membership.ErrMemberNotLearner:    rpctypes.ErrGRPCMemberNotLearner,
+	membership.ErrTooManyLearners:     rpctypes.ErrGRPCTooManyLearners,
+	errors.ErrNotEnoughStartedMembers: rpctypes.ErrMemberNotEnoughStarted,
+	errors.ErrLearnerNotReady:         rpctypes.ErrGRPCLearnerNotReady,
+
+	mvcc.ErrCompacted:         rpctypes.ErrGRPCCompacted,
+	mvcc.ErrFutureRev:         rpctypes.ErrGRPCFutureRev,
+	errors.ErrRequestTooLarge: rpctypes.ErrGRPCRequestTooLarge,
+	errors.ErrNoSpace:         rpctypes.ErrGRPCNoSpace,
+	errors.ErrTooManyRequests: rpctypes.ErrTooManyRequests,
+
+	errors.ErrNoLeader:                   rpctypes.ErrGRPCNoLeader,
+	errors.ErrNotLeader:                  rpctypes.ErrGRPCNotLeader,
+	errors.ErrLeaderChanged:              rpctypes.ErrGRPCLeaderChanged,
+	errors.ErrStopped:                    rpctypes.ErrGRPCStopped,
+	errors.ErrTimeout:                    rpctypes.ErrGRPCTimeout,
+	errors.ErrTimeoutDueToLeaderFail:     rpctypes.ErrGRPCTimeoutDueToLeaderFail,
+	errors.ErrTimeoutDueToConnectionLost: rpctypes.ErrGRPCTimeoutDueToConnectionLost,
+	errors.ErrTimeoutWaitAppliedIndex:    rpctypes.ErrGRPCTimeoutWaitAppliedIndex,
+	errors.ErrUnhealthy:                  rpctypes.ErrGRPCUnhealthy,
+	errors.ErrKeyNotFound:                rpctypes.ErrGRPCKeyNotFound,
+	errors.ErrCorrupt:                    rpctypes.ErrGRPCCorrupt,
+	errors.ErrBadLeaderTransferee:        rpctypes.ErrGRPCBadLeaderTransferee,
+
+	errors.ErrClusterVersionUnavailable:      rpctypes.ErrGRPCClusterVersionUnavailable,
+	errors.ErrWrongDowngradeVersionFormat:    rpctypes.ErrGRPCWrongDowngradeVersionFormat,
+	version.ErrInvalidDowngradeTargetVersion: rpctypes.ErrGRPCInvalidDowngradeTargetVersion,
+	version.ErrDowngradeInProcess:            rpctypes.ErrGRPCDowngradeInProcess,
+	version.ErrNoInflightDowngrade:           rpctypes.ErrGRPCNoInflightDowngrade,
 
 	lease.ErrLeaseNotFound:    rpctypes.ErrGRPCLeaseNotFound,
 	lease.ErrLeaseExists:      rpctypes.ErrGRPCLeaseExist,

server/etcdserver/api/v3rpc/watch.go

@@ -26,6 +26,7 @@ import (
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 	"go.etcd.io/etcd/server/v3/auth"
 	"go.etcd.io/etcd/server/v3/etcdserver"
+	"go.etcd.io/etcd/server/v3/etcdserver/apply"
 	"go.etcd.io/etcd/server/v3/storage/mvcc"
 
 	"go.uber.org/zap"
@@ -41,7 +42,7 @@ type watchServer struct {
 
 	maxRequestBytes int
 
-	sg        etcdserver.RaftStatusGetter
+	sg        apply.RaftStatusGetter
 	watchable mvcc.WatchableKV
 	ag        AuthGetter
 }
@@ -52,7 +53,7 @@ func NewWatchServer(s *etcdserver.EtcdServer) pb.WatchServer {
 		lg: s.Cfg.Logger,
 
 		clusterID: int64(s.Cluster().ID()),
-		memberID:  int64(s.ID()),
+		memberID:  int64(s.MemberId()),
 
 		maxRequestBytes: int(s.Cfg.MaxRequestBytes + grpcOverheadBytes),
 
@@ -124,7 +125,7 @@ type serverWatchStream struct {
 
 	maxRequestBytes int
 
-	sg        etcdserver.RaftStatusGetter
+	sg        apply.RaftStatusGetter
 	watchable mvcc.WatchableKV
 	ag        AuthGetter