Applier does not depend on EtcdServer any longer.

All the depencies are explicily passed to the UberApplier factory method.
This commit is contained in:
Piotr Tabor 2022-04-03 23:01:08 +02:00
parent 651de5a057
commit b073129d03
6 changed files with 186 additions and 99 deletions

View File

@ -53,7 +53,7 @@ func (qa *quotaAlarmer) check(ctx context.Context, r interface{}) error {
func NewQuotaKVServer(s *etcdserver.EtcdServer) pb.KVServer { func NewQuotaKVServer(s *etcdserver.EtcdServer) pb.KVServer {
return &quotaKVServer{ return &quotaKVServer{
NewKVServer(s), NewKVServer(s),
quotaAlarmer{storage.NewBackendQuota(s.Cfg, s.Backend(), "kv"), s, s.MemberId()}, 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 { func NewQuotaLeaseServer(s *etcdserver.EtcdServer) pb.LeaseServer {
return &quotaLeaseServer{ return &quotaLeaseServer{
NewLeaseServer(s), NewLeaseServer(s),
quotaAlarmer{storage.NewBackendQuota(s.Cfg, s.Backend(), "lease"), s, s.MemberId()}, 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)
}

View File

@ -25,9 +25,12 @@ import (
"go.etcd.io/etcd/server/v3/auth" "go.etcd.io/etcd/server/v3/auth"
"go.etcd.io/etcd/server/v3/etcdserver/api" "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/membership"
"go.etcd.io/etcd/server/v3/etcdserver/api/v3alarm"
"go.etcd.io/etcd/server/v3/etcdserver/cindex"
"go.etcd.io/etcd/server/v3/etcdserver/version" "go.etcd.io/etcd/server/v3/etcdserver/version"
"go.etcd.io/etcd/server/v3/lease" "go.etcd.io/etcd/server/v3/lease"
serverstorage "go.etcd.io/etcd/server/v3/storage" serverstorage "go.etcd.io/etcd/server/v3/storage"
"go.etcd.io/etcd/server/v3/storage/backend"
"go.etcd.io/etcd/server/v3/storage/mvcc" "go.etcd.io/etcd/server/v3/storage/mvcc"
"github.com/gogo/protobuf/proto" "github.com/gogo/protobuf/proto"
@ -89,13 +92,52 @@ type applierV3 interface {
RoleList(ua *pb.AuthRoleListRequest) (*pb.AuthRoleListResponse, error) RoleList(ua *pb.AuthRoleListRequest) (*pb.AuthRoleListResponse, error)
// processing internal V3 raft request // processing internal V3 raft request
ClusterVersionSet(r *membershippb.ClusterVersionSetRequest, shouldApplyV3 membership.ShouldApplyV3) ClusterVersionSet(r *membershippb.ClusterVersionSetRequest, shouldApplyV3 membership.ShouldApplyV3)
ClusterMemberAttrSet(r *membershippb.ClusterMemberAttrSetRequest, shouldApplyV3 membership.ShouldApplyV3) ClusterMemberAttrSet(r *membershippb.ClusterMemberAttrSetRequest, shouldApplyV3 membership.ShouldApplyV3)
DowngradeInfoSet(r *membershippb.DowngradeInfoSetRequest, shouldApplyV3 membership.ShouldApplyV3) DowngradeInfoSet(r *membershippb.DowngradeInfoSetRequest, shouldApplyV3 membership.ShouldApplyV3)
} }
type SnapshotServer interface {
ForceSnapshot()
}
type applierV3backend struct { type applierV3backend struct {
s *EtcdServer lg *zap.Logger
kv mvcc.KV
alarmStore *v3alarm.AlarmStore
authStore auth.AuthStore
lessor lease.Lessor
cluster *membership.RaftCluster
raftStatus RaftStatusGetter
snapshotServer SnapshotServer
consistentIndex cindex.ConsistentIndexer
txnModeWriteWithSharedBuffer bool
}
func newApplierV3Backend(
lg *zap.Logger,
kv mvcc.KV,
alarmStore *v3alarm.AlarmStore,
authStore auth.AuthStore,
lessor lease.Lessor,
cluster *membership.RaftCluster,
raftStatus RaftStatusGetter,
snapshotServer SnapshotServer,
consistentIndex cindex.ConsistentIndexer,
txnModeWriteWithSharedBuffer bool) applierV3 {
return &applierV3backend{
lg: lg,
kv: kv,
alarmStore: alarmStore,
authStore: authStore,
lessor: lessor,
cluster: cluster,
raftStatus: raftStatus,
snapshotServer: snapshotServer,
consistentIndex: consistentIndex,
txnModeWriteWithSharedBuffer: txnModeWriteWithSharedBuffer}
} }
func (a *applierV3backend) WrapApply(ctx context.Context, r *pb.InternalRaftRequest, shouldApplyV3 membership.ShouldApplyV3, applyFunc ApplyFunc) *applyResult { func (a *applierV3backend) WrapApply(ctx context.Context, r *pb.InternalRaftRequest, shouldApplyV3 membership.ShouldApplyV3, applyFunc ApplyFunc) *applyResult {
@ -103,63 +145,63 @@ func (a *applierV3backend) WrapApply(ctx context.Context, r *pb.InternalRaftRequ
} }
func (a *applierV3backend) Put(ctx context.Context, txn mvcc.TxnWrite, p *pb.PutRequest) (resp *pb.PutResponse, trace *traceutil.Trace, err error) { func (a *applierV3backend) Put(ctx context.Context, txn mvcc.TxnWrite, p *pb.PutRequest) (resp *pb.PutResponse, trace *traceutil.Trace, err error) {
return Put(ctx, a.s.Logger(), a.s.lessor, a.s.KV(), txn, p) return Put(ctx, a.lg, a.lessor, a.kv, txn, p)
} }
func (a *applierV3backend) DeleteRange(txn mvcc.TxnWrite, dr *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) { func (a *applierV3backend) DeleteRange(txn mvcc.TxnWrite, dr *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) {
return DeleteRange(a.s.KV(), txn, dr) return DeleteRange(a.kv, txn, dr)
} }
func (a *applierV3backend) Range(ctx context.Context, txn mvcc.TxnRead, r *pb.RangeRequest) (*pb.RangeResponse, error) { func (a *applierV3backend) Range(ctx context.Context, txn mvcc.TxnRead, r *pb.RangeRequest) (*pb.RangeResponse, error) {
return Range(ctx, a.s.Logger(), a.s.KV(), txn, r) return Range(ctx, a.lg, a.kv, txn, r)
} }
func (a *applierV3backend) Txn(ctx context.Context, rt *pb.TxnRequest) (*pb.TxnResponse, *traceutil.Trace, error) { func (a *applierV3backend) Txn(ctx context.Context, rt *pb.TxnRequest) (*pb.TxnResponse, *traceutil.Trace, error) {
return Txn(ctx, a.s.Logger(), rt, a.s.Cfg.ExperimentalTxnModeWriteWithSharedBuffer, a.s.KV(), a.s.lessor) return Txn(ctx, a.lg, rt, a.txnModeWriteWithSharedBuffer, a.kv, a.lessor)
} }
func (a *applierV3backend) Compaction(compaction *pb.CompactionRequest) (*pb.CompactionResponse, <-chan struct{}, *traceutil.Trace, error) { func (a *applierV3backend) Compaction(compaction *pb.CompactionRequest) (*pb.CompactionResponse, <-chan struct{}, *traceutil.Trace, error) {
resp := &pb.CompactionResponse{} resp := &pb.CompactionResponse{}
resp.Header = &pb.ResponseHeader{} resp.Header = &pb.ResponseHeader{}
trace := traceutil.New("compact", trace := traceutil.New("compact",
a.s.Logger(), a.lg,
traceutil.Field{Key: "revision", Value: compaction.Revision}, traceutil.Field{Key: "revision", Value: compaction.Revision},
) )
ch, err := a.s.KV().Compact(trace, compaction.Revision) ch, err := a.kv.Compact(trace, compaction.Revision)
if err != nil { if err != nil {
return nil, ch, nil, err return nil, ch, nil, err
} }
// get the current revision. which key to get is not important. // get the current revision. which key to get is not important.
rr, _ := a.s.KV().Range(context.TODO(), []byte("compaction"), nil, mvcc.RangeOptions{}) rr, _ := a.kv.Range(context.TODO(), []byte("compaction"), nil, mvcc.RangeOptions{})
resp.Header.Revision = rr.Rev resp.Header.Revision = rr.Rev
return resp, ch, trace, err return resp, ch, trace, err
} }
func (a *applierV3backend) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) { func (a *applierV3backend) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
l, err := a.s.lessor.Grant(lease.LeaseID(lc.ID), lc.TTL) l, err := a.lessor.Grant(lease.LeaseID(lc.ID), lc.TTL)
resp := &pb.LeaseGrantResponse{} resp := &pb.LeaseGrantResponse{}
if err == nil { if err == nil {
resp.ID = int64(l.ID) resp.ID = int64(l.ID)
resp.TTL = l.TTL() resp.TTL = l.TTL()
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) LeaseRevoke(lc *pb.LeaseRevokeRequest) (*pb.LeaseRevokeResponse, error) { func (a *applierV3backend) LeaseRevoke(lc *pb.LeaseRevokeRequest) (*pb.LeaseRevokeResponse, error) {
err := a.s.lessor.Revoke(lease.LeaseID(lc.ID)) err := a.lessor.Revoke(lease.LeaseID(lc.ID))
return &pb.LeaseRevokeResponse{Header: newHeader(a.s)}, err return &pb.LeaseRevokeResponse{Header: a.newHeader()}, err
} }
func (a *applierV3backend) LeaseCheckpoint(lc *pb.LeaseCheckpointRequest) (*pb.LeaseCheckpointResponse, error) { func (a *applierV3backend) LeaseCheckpoint(lc *pb.LeaseCheckpointRequest) (*pb.LeaseCheckpointResponse, error) {
for _, c := range lc.Checkpoints { for _, c := range lc.Checkpoints {
err := a.s.lessor.Checkpoint(lease.LeaseID(c.ID), c.Remaining_TTL) err := a.lessor.Checkpoint(lease.LeaseID(c.ID), c.Remaining_TTL)
if err != nil { if err != nil {
return &pb.LeaseCheckpointResponse{Header: newHeader(a.s)}, err return &pb.LeaseCheckpointResponse{Header: a.newHeader()}, err
} }
} }
return &pb.LeaseCheckpointResponse{Header: newHeader(a.s)}, nil return &pb.LeaseCheckpointResponse{Header: a.newHeader()}, nil
} }
func (a *applierV3backend) Alarm(ar *pb.AlarmRequest) (*pb.AlarmResponse, error) { func (a *applierV3backend) Alarm(ar *pb.AlarmRequest) (*pb.AlarmResponse, error) {
@ -167,18 +209,18 @@ func (a *applierV3backend) Alarm(ar *pb.AlarmRequest) (*pb.AlarmResponse, error)
switch ar.Action { switch ar.Action {
case pb.AlarmRequest_GET: case pb.AlarmRequest_GET:
resp.Alarms = a.s.alarmStore.Get(ar.Alarm) resp.Alarms = a.alarmStore.Get(ar.Alarm)
case pb.AlarmRequest_ACTIVATE: case pb.AlarmRequest_ACTIVATE:
if ar.Alarm == pb.AlarmType_NONE { if ar.Alarm == pb.AlarmType_NONE {
break break
} }
m := a.s.alarmStore.Activate(types.ID(ar.MemberID), ar.Alarm) m := a.alarmStore.Activate(types.ID(ar.MemberID), ar.Alarm)
if m == nil { if m == nil {
break break
} }
resp.Alarms = append(resp.Alarms, m) resp.Alarms = append(resp.Alarms, m)
case pb.AlarmRequest_DEACTIVATE: case pb.AlarmRequest_DEACTIVATE:
m := a.s.alarmStore.Deactivate(types.ID(ar.MemberID), ar.Alarm) m := a.alarmStore.Deactivate(types.ID(ar.MemberID), ar.Alarm)
if m == nil { if m == nil {
break break
} }
@ -214,156 +256,156 @@ func (a *applierV3Capped) LeaseGrant(_ *pb.LeaseGrantRequest) (*pb.LeaseGrantRes
} }
func (a *applierV3backend) AuthEnable() (*pb.AuthEnableResponse, error) { func (a *applierV3backend) AuthEnable() (*pb.AuthEnableResponse, error) {
err := a.s.AuthStore().AuthEnable() err := a.authStore.AuthEnable()
if err != nil { if err != nil {
return nil, err return nil, err
} }
return &pb.AuthEnableResponse{Header: newHeader(a.s)}, nil return &pb.AuthEnableResponse{Header: a.newHeader()}, nil
} }
func (a *applierV3backend) AuthDisable() (*pb.AuthDisableResponse, error) { func (a *applierV3backend) AuthDisable() (*pb.AuthDisableResponse, error) {
a.s.AuthStore().AuthDisable() a.authStore.AuthDisable()
return &pb.AuthDisableResponse{Header: newHeader(a.s)}, nil return &pb.AuthDisableResponse{Header: a.newHeader()}, nil
} }
func (a *applierV3backend) AuthStatus() (*pb.AuthStatusResponse, error) { func (a *applierV3backend) AuthStatus() (*pb.AuthStatusResponse, error) {
enabled := a.s.AuthStore().IsAuthEnabled() enabled := a.authStore.IsAuthEnabled()
authRevision := a.s.AuthStore().Revision() authRevision := a.authStore.Revision()
return &pb.AuthStatusResponse{Header: newHeader(a.s), Enabled: enabled, AuthRevision: authRevision}, nil return &pb.AuthStatusResponse{Header: a.newHeader(), Enabled: enabled, AuthRevision: authRevision}, nil
} }
func (a *applierV3backend) Authenticate(r *pb.InternalAuthenticateRequest) (*pb.AuthenticateResponse, error) { func (a *applierV3backend) Authenticate(r *pb.InternalAuthenticateRequest) (*pb.AuthenticateResponse, error) {
ctx := context.WithValue(context.WithValue(a.s.ctx, auth.AuthenticateParamIndex{}, a.s.consistIndex.ConsistentIndex()), auth.AuthenticateParamSimpleTokenPrefix{}, r.SimpleToken) ctx := context.WithValue(context.WithValue(context.Background(), auth.AuthenticateParamIndex{}, a.consistentIndex.ConsistentIndex()), auth.AuthenticateParamSimpleTokenPrefix{}, r.SimpleToken)
resp, err := a.s.AuthStore().Authenticate(ctx, r.Name, r.Password) resp, err := a.authStore.Authenticate(ctx, r.Name, r.Password)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) UserAdd(r *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, error) { func (a *applierV3backend) UserAdd(r *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, error) {
resp, err := a.s.AuthStore().UserAdd(r) resp, err := a.authStore.UserAdd(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) UserDelete(r *pb.AuthUserDeleteRequest) (*pb.AuthUserDeleteResponse, error) { func (a *applierV3backend) UserDelete(r *pb.AuthUserDeleteRequest) (*pb.AuthUserDeleteResponse, error) {
resp, err := a.s.AuthStore().UserDelete(r) resp, err := a.authStore.UserDelete(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) UserChangePassword(r *pb.AuthUserChangePasswordRequest) (*pb.AuthUserChangePasswordResponse, error) { func (a *applierV3backend) UserChangePassword(r *pb.AuthUserChangePasswordRequest) (*pb.AuthUserChangePasswordResponse, error) {
resp, err := a.s.AuthStore().UserChangePassword(r) resp, err := a.authStore.UserChangePassword(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) UserGrantRole(r *pb.AuthUserGrantRoleRequest) (*pb.AuthUserGrantRoleResponse, error) { func (a *applierV3backend) UserGrantRole(r *pb.AuthUserGrantRoleRequest) (*pb.AuthUserGrantRoleResponse, error) {
resp, err := a.s.AuthStore().UserGrantRole(r) resp, err := a.authStore.UserGrantRole(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) UserGet(r *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error) { func (a *applierV3backend) UserGet(r *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error) {
resp, err := a.s.AuthStore().UserGet(r) resp, err := a.authStore.UserGet(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) UserRevokeRole(r *pb.AuthUserRevokeRoleRequest) (*pb.AuthUserRevokeRoleResponse, error) { func (a *applierV3backend) UserRevokeRole(r *pb.AuthUserRevokeRoleRequest) (*pb.AuthUserRevokeRoleResponse, error) {
resp, err := a.s.AuthStore().UserRevokeRole(r) resp, err := a.authStore.UserRevokeRole(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) RoleAdd(r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, error) { func (a *applierV3backend) RoleAdd(r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, error) {
resp, err := a.s.AuthStore().RoleAdd(r) resp, err := a.authStore.RoleAdd(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) RoleGrantPermission(r *pb.AuthRoleGrantPermissionRequest) (*pb.AuthRoleGrantPermissionResponse, error) { func (a *applierV3backend) RoleGrantPermission(r *pb.AuthRoleGrantPermissionRequest) (*pb.AuthRoleGrantPermissionResponse, error) {
resp, err := a.s.AuthStore().RoleGrantPermission(r) resp, err := a.authStore.RoleGrantPermission(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) RoleGet(r *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error) { func (a *applierV3backend) RoleGet(r *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error) {
resp, err := a.s.AuthStore().RoleGet(r) resp, err := a.authStore.RoleGet(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) RoleRevokePermission(r *pb.AuthRoleRevokePermissionRequest) (*pb.AuthRoleRevokePermissionResponse, error) { func (a *applierV3backend) RoleRevokePermission(r *pb.AuthRoleRevokePermissionRequest) (*pb.AuthRoleRevokePermissionResponse, error) {
resp, err := a.s.AuthStore().RoleRevokePermission(r) resp, err := a.authStore.RoleRevokePermission(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) RoleDelete(r *pb.AuthRoleDeleteRequest) (*pb.AuthRoleDeleteResponse, error) { func (a *applierV3backend) RoleDelete(r *pb.AuthRoleDeleteRequest) (*pb.AuthRoleDeleteResponse, error) {
resp, err := a.s.AuthStore().RoleDelete(r) resp, err := a.authStore.RoleDelete(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) UserList(r *pb.AuthUserListRequest) (*pb.AuthUserListResponse, error) { func (a *applierV3backend) UserList(r *pb.AuthUserListRequest) (*pb.AuthUserListResponse, error) {
resp, err := a.s.AuthStore().UserList(r) resp, err := a.authStore.UserList(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) RoleList(r *pb.AuthRoleListRequest) (*pb.AuthRoleListResponse, error) { func (a *applierV3backend) RoleList(r *pb.AuthRoleListRequest) (*pb.AuthRoleListResponse, error) {
resp, err := a.s.AuthStore().RoleList(r) resp, err := a.authStore.RoleList(r)
if resp != nil { if resp != nil {
resp.Header = newHeader(a.s) resp.Header = a.newHeader()
} }
return resp, err return resp, err
} }
func (a *applierV3backend) ClusterVersionSet(r *membershippb.ClusterVersionSetRequest, shouldApplyV3 membership.ShouldApplyV3) { func (a *applierV3backend) ClusterVersionSet(r *membershippb.ClusterVersionSetRequest, shouldApplyV3 membership.ShouldApplyV3) {
prevVersion := a.s.Cluster().Version() prevVersion := a.cluster.Version()
newVersion := semver.Must(semver.NewVersion(r.Ver)) newVersion := semver.Must(semver.NewVersion(r.Ver))
a.s.cluster.SetVersion(newVersion, api.UpdateCapability, shouldApplyV3) a.cluster.SetVersion(newVersion, api.UpdateCapability, shouldApplyV3)
// Force snapshot after cluster version downgrade. // Force snapshot after cluster version downgrade.
if prevVersion != nil && newVersion.LessThan(*prevVersion) { if prevVersion != nil && newVersion.LessThan(*prevVersion) {
lg := a.s.Logger() lg := a.lg
if lg != nil { if lg != nil {
lg.Info("Cluster version downgrade detected, forcing snapshot", lg.Info("Cluster version downgrade detected, forcing snapshot",
zap.String("prev-cluster-version", prevVersion.String()), zap.String("prev-cluster-version", prevVersion.String()),
zap.String("new-cluster-version", newVersion.String()), zap.String("new-cluster-version", newVersion.String()),
) )
} }
a.s.forceSnapshot = true a.snapshotServer.ForceSnapshot()
} }
} }
func (a *applierV3backend) ClusterMemberAttrSet(r *membershippb.ClusterMemberAttrSetRequest, shouldApplyV3 membership.ShouldApplyV3) { func (a *applierV3backend) ClusterMemberAttrSet(r *membershippb.ClusterMemberAttrSetRequest, shouldApplyV3 membership.ShouldApplyV3) {
a.s.cluster.UpdateAttributes( a.cluster.UpdateAttributes(
types.ID(r.Member_ID), types.ID(r.Member_ID),
membership.Attributes{ membership.Attributes{
Name: r.MemberAttributes.Name, Name: r.MemberAttributes.Name,
@ -378,7 +420,7 @@ func (a *applierV3backend) DowngradeInfoSet(r *membershippb.DowngradeInfoSetRequ
if r.Enabled { if r.Enabled {
d = version.DowngradeInfo{Enabled: true, TargetVersion: r.Ver} d = version.DowngradeInfo{Enabled: true, TargetVersion: r.Ver}
} }
a.s.cluster.SetDowngradeInfo(&d, shouldApplyV3) a.cluster.SetDowngradeInfo(&d, shouldApplyV3)
} }
type quotaApplierV3 struct { type quotaApplierV3 struct {
@ -386,8 +428,8 @@ type quotaApplierV3 struct {
q serverstorage.Quota q serverstorage.Quota
} }
func newQuotaApplierV3(s *EtcdServer, app applierV3) applierV3 { func newQuotaApplierV3(lg *zap.Logger, quotaBackendBytesCfg int64, be backend.Backend, app applierV3) applierV3 {
return &quotaApplierV3{app, serverstorage.NewBackendQuota(s.Cfg, s.Backend(), "v3-applier")} return &quotaApplierV3{app, serverstorage.NewBackendQuota(lg, quotaBackendBytesCfg, be, "v3-applier")}
} }
func (a *quotaApplierV3) Put(ctx context.Context, txn mvcc.TxnWrite, p *pb.PutRequest) (*pb.PutResponse, *traceutil.Trace, error) { func (a *quotaApplierV3) Put(ctx context.Context, txn mvcc.TxnWrite, p *pb.PutRequest) (*pb.PutResponse, *traceutil.Trace, error) {
@ -439,11 +481,11 @@ func removeNeedlessRangeReqs(txn *pb.TxnRequest) {
txn.Failure = f(txn.Failure) txn.Failure = f(txn.Failure)
} }
func newHeader(s *EtcdServer) *pb.ResponseHeader { func (a *applierV3backend) newHeader() *pb.ResponseHeader {
return &pb.ResponseHeader{ return &pb.ResponseHeader{
ClusterId: uint64(s.Cluster().ID()), ClusterId: uint64(a.cluster.ID()),
MemberId: uint64(s.MemberId()), MemberId: uint64(a.raftStatus.MemberId()),
Revision: s.KV().Rev(), Revision: a.kv.Rev(),
RaftTerm: s.Term(), RaftTerm: a.raftStatus.Term(),
} }
} }

View File

@ -391,7 +391,7 @@ func NewServer(cfg config.ServerConfig) (srv *EtcdServer, err error) {
if err = srv.restoreAlarms(); err != nil { if err = srv.restoreAlarms(); err != nil {
return nil, err return nil, err
} }
srv.uberApply = newUberApplier(srv) srv.uberApply = srv.NewUberApplier()
if srv.Cfg.EnableLeaseCheckpoint { if srv.Cfg.EnableLeaseCheckpoint {
// setting checkpointer enables lease checkpoint feature. // setting checkpointer enables lease checkpoint feature.
@ -1072,7 +1072,12 @@ func (s *EtcdServer) applySnapshot(ep *etcdProgress, apply *apply) {
// As backends and implementations like alarmsStore changed, we need // As backends and implementations like alarmsStore changed, we need
// to re-bootstrap Appliers. // to re-bootstrap Appliers.
s.uberApply = newUberApplier(s) s.uberApply = s.NewUberApplier()
}
func (s *EtcdServer) NewUberApplier() *uberApplier {
return newUberApplier(s.lg, s.be, s.KV(), s.alarmStore, s.authStore, s.lessor, s.cluster, s, s, s.consistIndex,
s.Cfg.WarningApplyDuration, s.Cfg.ExperimentalTxnModeWriteWithSharedBuffer, s.Cfg.QuotaBackendBytes)
} }
func verifySnapshotIndex(snapshot raftpb.Snapshot, cindex uint64) { func verifySnapshotIndex(snapshot raftpb.Snapshot, cindex uint64) {
@ -1109,6 +1114,10 @@ func (s *EtcdServer) applyEntries(ep *etcdProgress, apply *apply) {
} }
} }
func (s *EtcdServer) ForceSnapshot() {
s.forceSnapshot = true
}
func (s *EtcdServer) triggerSnapshot(ep *etcdProgress) { func (s *EtcdServer) triggerSnapshot(ep *etcdProgress) {
if !s.shouldSnapshot(ep) { if !s.shouldSnapshot(ep) {
return return

View File

@ -20,8 +20,12 @@ import (
"time" "time"
pb "go.etcd.io/etcd/api/v3/etcdserverpb" pb "go.etcd.io/etcd/api/v3/etcdserverpb"
"go.etcd.io/etcd/server/v3/auth"
"go.etcd.io/etcd/server/v3/etcdserver/api/membership" "go.etcd.io/etcd/server/v3/etcdserver/api/membership"
"go.etcd.io/etcd/server/v3/etcdserver/api/v3alarm" "go.etcd.io/etcd/server/v3/etcdserver/api/v3alarm"
"go.etcd.io/etcd/server/v3/etcdserver/cindex"
"go.etcd.io/etcd/server/v3/lease"
"go.etcd.io/etcd/server/v3/storage/backend"
"go.etcd.io/etcd/server/v3/storage/mvcc" "go.etcd.io/etcd/server/v3/storage/mvcc"
"go.uber.org/zap" "go.uber.org/zap"
) )
@ -39,13 +43,26 @@ type uberApplier struct {
applyV3base applierV3 applyV3base applierV3
} }
func newUberApplier(s *EtcdServer) *uberApplier { func newUberApplier(
applyV3base_ := newApplierV3(s) lg *zap.Logger,
be backend.Backend,
kv mvcc.KV,
alarmStore *v3alarm.AlarmStore,
authStore auth.AuthStore,
lessor lease.Lessor,
cluster *membership.RaftCluster,
raftStatus RaftStatusGetter,
snapshotServer SnapshotServer,
consistentIndex cindex.ConsistentIndexer,
warningApplyDuration time.Duration,
txnModeWriteWithSharedBuffer bool,
quotaBackendBytesCfg int64) *uberApplier {
applyV3base_ := newApplierV3(lg, be, kv, alarmStore, authStore, lessor, cluster, raftStatus, snapshotServer, consistentIndex, txnModeWriteWithSharedBuffer, quotaBackendBytesCfg)
ua := &uberApplier{ ua := &uberApplier{
lg: s.lg, lg: lg,
alarmStore: s.alarmStore, alarmStore: alarmStore,
warningApplyDuration: s.Cfg.WarningApplyDuration, warningApplyDuration: warningApplyDuration,
applyV3: applyV3base_, applyV3: applyV3base_,
applyV3base: applyV3base_, applyV3base: applyV3base_,
} }
@ -53,15 +70,24 @@ func newUberApplier(s *EtcdServer) *uberApplier {
return ua return ua
} }
func newApplierV3Backend(s *EtcdServer) applierV3 { func newApplierV3(
return &applierV3backend{s: s} lg *zap.Logger,
} be backend.Backend,
kv mvcc.KV,
func newApplierV3(s *EtcdServer) applierV3 { alarmStore *v3alarm.AlarmStore,
authStore auth.AuthStore,
lessor lease.Lessor,
cluster *membership.RaftCluster,
raftStatus RaftStatusGetter,
snapshotServer SnapshotServer,
consistentIndex cindex.ConsistentIndexer,
txnModeWriteWithSharedBuffer bool,
quotaBackendBytesCfg int64) applierV3 {
applierBackend := newApplierV3Backend(lg, kv, alarmStore, authStore, lessor, cluster, raftStatus, snapshotServer, consistentIndex, txnModeWriteWithSharedBuffer)
return newAuthApplierV3( return newAuthApplierV3(
s.AuthStore(), authStore,
newQuotaApplierV3(s, newApplierV3Backend(s)), newQuotaApplierV3(lg, quotaBackendBytesCfg, be, applierBackend),
s.lessor, lessor,
) )
} }

View File

@ -392,6 +392,15 @@ func (s *EtcdServer) LeaseTimeToLive(ctx context.Context, r *pb.LeaseTimeToLiveR
return nil, ErrCanceled return nil, ErrCanceled
} }
func (s *EtcdServer) newHeader() *pb.ResponseHeader {
return &pb.ResponseHeader{
ClusterId: uint64(s.cluster.ID()),
MemberId: uint64(s.MemberId()),
Revision: s.KV().Rev(),
RaftTerm: s.Term(),
}
}
// LeaseLeases is really ListLeases !??? // LeaseLeases is really ListLeases !???
func (s *EtcdServer) LeaseLeases(_ context.Context, _ *pb.LeaseLeasesRequest) (*pb.LeaseLeasesResponse, error) { func (s *EtcdServer) LeaseLeases(_ context.Context, _ *pb.LeaseLeasesRequest) (*pb.LeaseLeasesResponse, error) {
ls := s.lessor.Leases() ls := s.lessor.Leases()
@ -399,7 +408,7 @@ func (s *EtcdServer) LeaseLeases(_ context.Context, _ *pb.LeaseLeasesRequest) (*
for i := range ls { for i := range ls {
lss[i] = &pb.LeaseStatus{ID: int64(ls[i].ID)} lss[i] = &pb.LeaseStatus{ID: int64(ls[i].ID)}
} }
return &pb.LeaseLeasesResponse{Header: newHeader(s), Leases: lss}, nil return &pb.LeaseLeasesResponse{Header: s.newHeader(), Leases: lss}, nil
} }
func (s *EtcdServer) waitLeader(ctx context.Context) (*membership.Member, error) { func (s *EtcdServer) waitLeader(ctx context.Context) (*membership.Member, error) {

View File

@ -18,7 +18,6 @@ import (
"sync" "sync"
pb "go.etcd.io/etcd/api/v3/etcdserverpb" pb "go.etcd.io/etcd/api/v3/etcdserverpb"
"go.etcd.io/etcd/server/v3/config"
"go.etcd.io/etcd/server/v3/storage/backend" "go.etcd.io/etcd/server/v3/storage/backend"
humanize "github.com/dustin/go-humanize" humanize "github.com/dustin/go-humanize"
@ -73,23 +72,21 @@ var (
) )
// NewBackendQuota creates a quota layer with the given storage limit. // NewBackendQuota creates a quota layer with the given storage limit.
func NewBackendQuota(cfg config.ServerConfig, be backend.Backend, name string) Quota { func NewBackendQuota(lg *zap.Logger, quotaBackendBytesCfg int64, be backend.Backend, name string) Quota {
lg := cfg.Logger quotaBackendBytes.Set(float64(quotaBackendBytesCfg))
quotaBackendBytes.Set(float64(cfg.QuotaBackendBytes)) if quotaBackendBytesCfg < 0 {
if cfg.QuotaBackendBytes < 0 {
// disable quotas if negative // disable quotas if negative
quotaLogOnce.Do(func() { quotaLogOnce.Do(func() {
lg.Info( lg.Info(
"disabled backend quota", "disabled backend quota",
zap.String("quota-name", name), zap.String("quota-name", name),
zap.Int64("quota-size-bytes", cfg.QuotaBackendBytes), zap.Int64("quota-size-bytes", quotaBackendBytesCfg),
) )
}) })
return &passthroughQuota{} return &passthroughQuota{}
} }
if cfg.QuotaBackendBytes == 0 { if quotaBackendBytesCfg == 0 {
// use default size if no quota size given // use default size if no quota size given
quotaLogOnce.Do(func() { quotaLogOnce.Do(func() {
if lg != nil { if lg != nil {
@ -106,12 +103,12 @@ func NewBackendQuota(cfg config.ServerConfig, be backend.Backend, name string) Q
} }
quotaLogOnce.Do(func() { quotaLogOnce.Do(func() {
if cfg.QuotaBackendBytes > MaxQuotaBytes { if quotaBackendBytesCfg > MaxQuotaBytes {
lg.Warn( lg.Warn(
"quota exceeds the maximum value", "quota exceeds the maximum value",
zap.String("quota-name", name), zap.String("quota-name", name),
zap.Int64("quota-size-bytes", cfg.QuotaBackendBytes), zap.Int64("quota-size-bytes", quotaBackendBytesCfg),
zap.String("quota-size", humanize.Bytes(uint64(cfg.QuotaBackendBytes))), zap.String("quota-size", humanize.Bytes(uint64(quotaBackendBytesCfg))),
zap.Int64("quota-maximum-size-bytes", MaxQuotaBytes), zap.Int64("quota-maximum-size-bytes", MaxQuotaBytes),
zap.String("quota-maximum-size", maxQuotaSize), zap.String("quota-maximum-size", maxQuotaSize),
) )
@ -119,11 +116,11 @@ func NewBackendQuota(cfg config.ServerConfig, be backend.Backend, name string) Q
lg.Info( lg.Info(
"enabled backend quota", "enabled backend quota",
zap.String("quota-name", name), zap.String("quota-name", name),
zap.Int64("quota-size-bytes", cfg.QuotaBackendBytes), zap.Int64("quota-size-bytes", quotaBackendBytesCfg),
zap.String("quota-size", humanize.Bytes(uint64(cfg.QuotaBackendBytes))), zap.String("quota-size", humanize.Bytes(uint64(quotaBackendBytesCfg))),
) )
}) })
return &BackendQuota{be, cfg.QuotaBackendBytes} return &BackendQuota{be, quotaBackendBytesCfg}
} }
func (b *BackendQuota) Available(v interface{}) bool { func (b *BackendQuota) Available(v interface{}) bool {