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,25 @@ func isConnectedFullySince(transport rafthttp.Transporter, since time.Time, self
 	return numConnectedSince(transport, since, self, members) == len(members)
 }
 
+// exceedsRequestLimit checks if the committed index is too far ahead of
+// the applied index. Priority requests are allowed up to 10% extra gap.
+func exceedsRequestLimit(ai, ci uint64, r *pb.InternalRaftRequest) bool {
+	if ci <= ai+maxGapBetweenApplyAndCommitIndex {
+		return false
+	}
+	// allow up to 10% extra gap for priority requests
+	if ci <= ai+maxGapBetweenApplyAndCommitIndex*110/100 {
+		if isPriorityRequest(r) {
+			return false
+		}
+	}
+	return true
+}
+
+func isPriorityRequest(r *pb.InternalRaftRequest) bool {
+	return r != nil && r.LeaseRevoke != nil
+}
+
 // 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,55 @@ type testStringerFunc func() string
 func (s testStringerFunc) String() string {
 	return s()
 }
+
+func TestExceedsRequestLimit(t *testing.T) {
+	tests := []struct {
+		name           string
+		ci             uint64
+		ai             uint64
+		expectedResult bool
+		req            *pb.InternalRaftRequest
+	}{
+		{
+			ci:             1 + maxGapBetweenApplyAndCommitIndex,
+			ai:             1,
+			expectedResult: false,
+			req:            nil,
+			name:           "Test nil InternalRaftRequest",
+		},
+		{
+			ci:             1 + maxGapBetweenApplyAndCommitIndex,
+			ai:             1,
+			expectedResult: false,
+			req:            &pb.InternalRaftRequest{},
+			name:           "Test non-critical request and gap is not larger than maxGapBetweenApplyAndCommitIndex",
+		},
+		{
+			ci:             1 + maxGapBetweenApplyAndCommitIndex + 1,
+			ai:             1,
+			expectedResult: true,
+			req:            &pb.InternalRaftRequest{},
+			name:           "Test non-critical request and gap is larger than maxGapBetweenApplyAndCommitIndex",
+		},
+		{
+			ci:             1 + maxGapBetweenApplyAndCommitIndex + 1,
+			ai:             1,
+			expectedResult: false,
+			req:            &pb.InternalRaftRequest{LeaseRevoke: &pb.LeaseRevokeRequest{}},
+			name:           "Test critical request and gap is larger than maxGapBetweenApplyAndCommitIndex",
+		},
+		{
+			ci:             1 + maxGapBetweenApplyAndCommitIndex + 1 + maxGapBetweenApplyAndCommitIndex/10,
+			ai:             1,
+			expectedResult: true,
+			req:            &pb.InternalRaftRequest{LeaseRevoke: &pb.LeaseRevokeRequest{}},
+			name:           "Test critical request and gap is larger than 110% maxGapBetweenApplyAndCommitIndex",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expectedResult, exceedsRequestLimit(tc.ai, tc.ci, tc.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 exceedsRequestLimit(ai, ci, &r) {
 		return nil, errors.ErrTooManyRequests
 	}