stability enhancement during overload conditions

Signed-off-by: shenmu.wy <[email protected]>

Author: silentred

server/etcdserver/util.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"time"
 
+	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/client/pkg/v3/types"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/membership"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
@@ -42,6 +43,26 @@ func isConnectedFullySince(transport rafthttp.Transporter, since time.Time, self
 	return numConnectedSince(transport, since, self, members) == len(members)
 }
 
+// isTooManyRequest checks whether the gap between applied index and committed
+// index is too large. r *pb.InternalRaftRequest should read-only.
+func isTooManyRequest(ai, ci uint64, r *pb.InternalRaftRequest) bool {
+	// Critical request kinds include LeaseRevoke
+	isCritical := r != nil && r.LeaseRevoke != nil
+
+	// If the gap is larger than maxGapBetweenApplyAndCommitIndex, reject all normal request
+	if !isCritical && ci > ai+maxGapBetweenApplyAndCommitIndex {
+		return true
+	}
+
+	// Make an extra 10% gap buffer for critical requests, so thay can be applied
+	// during overload conditions.
+	if isCritical && ci-ai > maxGapBetweenApplyAndCommitIndex+500 {
+		return true
+	}
+
+	return false
+}
+
 // numConnectedSince counts how many members are connected to the local member
 // since the given time.
 func numConnectedSince(transport rafthttp.Transporter, since time.Time, self types.ID, members []*membership.Member) int {

server/etcdserver/util_test.go

@@ -19,8 +19,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap/zaptest"
 
+	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/client/pkg/v3/types"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/membership"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
@@ -110,3 +112,54 @@ type testStringerFunc func() string
 func (s testStringerFunc) String() string {
 	return s()
 }
+
+func TestIsTooManyRequest(t *testing.T) {
+	tests := []struct {
+		ci     uint64
+		ai     uint64
+		expect bool
+		req    *pb.InternalRaftRequest
+		desc   string
+	}{
+		{
+			ci:     1 + maxGapBetweenApplyAndCommitIndex,
+			ai:     1,
+			expect: false,
+			req:    nil,
+			desc:   "Test nil InternalRaftRequest",
+		},
+		{
+			ci:     1 + maxGapBetweenApplyAndCommitIndex,
+			ai:     1,
+			expect: false,
+			req:    &pb.InternalRaftRequest{},
+			desc:   "Test non-critical request and gap is not larger than maxGapBetweenApplyAndCommitIndex",
+		},
+		{
+			ci:     1 + maxGapBetweenApplyAndCommitIndex + 1,
+			ai:     1,
+			expect: true,
+			req:    &pb.InternalRaftRequest{},
+			desc:   "Test non-critical request and gap is larger than maxGapBetweenApplyAndCommitIndex",
+		},
+		{
+			ci:     1 + maxGapBetweenApplyAndCommitIndex + 1,
+			ai:     1,
+			expect: false,
+			req:    &pb.InternalRaftRequest{LeaseRevoke: &pb.LeaseRevokeRequest{}},
+			desc:   "Test critical request and gap is larger than maxGapBetweenApplyAndCommitIndex",
+		},
+		{
+			ci:     1 + maxGapBetweenApplyAndCommitIndex + 1 + maxGapBetweenApplyAndCommitIndex/10,
+			ai:     1,
+			expect: true,
+			req:    &pb.InternalRaftRequest{LeaseRevoke: &pb.LeaseRevokeRequest{}},
+			desc:   "Test critical request and gap is larger than 110% maxGapBetweenApplyAndCommitIndex",
+		},
+	}
+
+	for _, test := range tests {
+		t.Log(test.desc)
+		assert.Equal(t, test.expect, isTooManyRequest(test.ai, test.ci, test.req))
+	}
+}

server/etcdserver/v3_server.go

@@ -792,7 +792,8 @@ func (s *EtcdServer) doSerialize(ctx context.Context, chk func(*auth.AuthInfo) e
 func (s *EtcdServer) processInternalRaftRequestOnce(ctx context.Context, r pb.InternalRaftRequest) (*apply2.Result, error) {
 	ai := s.getAppliedIndex()
 	ci := s.getCommittedIndex()
-	if ci > ai+maxGapBetweenApplyAndCommitIndex {
+
+	if isTooManyRequest(ai, ci, &r) {
 		return nil, errors.ErrTooManyRequests
 	}