Disable deny writing when the growing segment size exceeds the watermark (#26163)

Signed-off-by: Nbigsheeper <yihao.dai@zilliz.com>

configs/milvus.yaml

@@ -582,10 +582,11 @@ quotaAndLimits:
       queryNodeMemoryLowWaterLevel: 0.85 # (0, 1], memoryLowWaterLevel in QueryNodes
       queryNodeMemoryHighWaterLevel: 0.95 # (0, 1], memoryHighWaterLevel in QueryNodes
     growingSegmentsSizeProtection:
-      # 1. No action will be taken if the ratio of growing segments size is less than the low water level.
-      # 2. The DML rate will be reduced if the ratio of growing segments size is greater than the low water level and less than the high water level.
-      # 3. All DML requests will be rejected if the ratio of growing segments size is greater than the high water level.
+      # No action will be taken if the growing segments size is less than the low watermark.
+      # When the growing segments size exceeds the low watermark, the dml rate will be reduced,
+      # but the rate will not be lower than `minRateRatio * dmlRate`.
       enabled: false
+      minRateRatio: 0.5
       lowWaterLevel: 0.2
       highWaterLevel: 0.4
     diskProtection:

internal/rootcoord/quota_center.go

@@ -658,24 +658,19 @@ func (q *QuotaCenter) getGrowingSegmentsSizeFactor() map[int64]float64 {
 		if cur <= low {
 			continue
 		}
-		if cur >= high {
-			log.RatedWarn(10, "QuotaCenter: QueryNode growing segments size to high water level",
-				zap.String("Node", fmt.Sprintf("%s-%d", typeutil.QueryNodeRole, nodeID)),
-				zap.Int64s("collections", metric.Effect.CollectionIDs),
-				zap.Int64("segmentsSize", metric.GrowingSegmentsSize),
-				zap.Uint64("TotalMem", metric.Hms.Memory),
-				zap.Float64("highWaterLevel", high))
-			updateCollectionFactor(0, metric.Effect.CollectionIDs)
-			continue
-		}
 		factor := (high - cur) / (high - low)
+		if factor < Params.QuotaConfig.GrowingSegmentsSizeMinRateRatio.GetAsFloat() {
+			factor = Params.QuotaConfig.GrowingSegmentsSizeMinRateRatio.GetAsFloat()
+		}
 		updateCollectionFactor(factor, metric.Effect.CollectionIDs)
-		log.RatedWarn(10, "QuotaCenter: QueryNode growing segments size to low water level, limit writing rate",
+		log.RatedWarn(10, "QuotaCenter: QueryNode growing segments size exceeds watermark, limit writing rate",
 			zap.String("Node", fmt.Sprintf("%s-%d", typeutil.QueryNodeRole, nodeID)),
 			zap.Int64s("collections", metric.Effect.CollectionIDs),
 			zap.Int64("segmentsSize", metric.GrowingSegmentsSize),
 			zap.Uint64("TotalMem", metric.Hms.Memory),
-			zap.Float64("lowWaterLevel", low))
+			zap.Float64("highWatermark", high),
+			zap.Float64("lowWatermark", low),
+			zap.Float64("factor", factor))
 	}
 	return collectionFactor
 }

internal/rootcoord/quota_center_test.go

@@ -438,6 +438,7 @@ func TestQuotaCenter(t *testing.T) {
 		meta := mockrootcoord.NewIMetaTable(t)
 		meta.EXPECT().GetCollectionByID(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, merr.ErrCollectionNotFound).Maybe()
 		quotaCenter := NewQuotaCenter(pcm, qc, &dataCoordMockForQuota{}, core.tsoAllocator, meta)
+		defaultRatio := Params.QuotaConfig.GrowingSegmentsSizeMinRateRatio.GetAsFloat()
 		tests := []struct {
 			low            float64
 			high           float64
@@ -449,15 +450,15 @@ func TestQuotaCenter(t *testing.T) {
 			{0.8, 0.9, 80, 100, 1},
 			{0.8, 0.9, 82, 100, 0.8},
 			{0.8, 0.9, 85, 100, 0.5},
-			{0.8, 0.9, 88, 100, 0.2},
-			{0.8, 0.9, 90, 100, 0},
+			{0.8, 0.9, 88, 100, defaultRatio},
+			{0.8, 0.9, 90, 100, defaultRatio},
 
 			{0.85, 0.95, 25, 100, 1},
 			{0.85, 0.95, 85, 100, 1},
 			{0.85, 0.95, 87, 100, 0.8},
 			{0.85, 0.95, 90, 100, 0.5},
-			{0.85, 0.95, 93, 100, 0.2},
-			{0.85, 0.95, 95, 100, 0},
+			{0.85, 0.95, 93, 100, defaultRatio},
+			{0.85, 0.95, 95, 100, defaultRatio},
 		}
 
 		quotaCenter.writableCollections = append(quotaCenter.writableCollections, 1, 2, 3)

pkg/util/paramtable/quota_param.go

@@ -107,6 +107,7 @@ type quotaConfig struct {
 	QueryNodeMemoryLowWaterLevel         ParamItem `refreshable:"true"`
 	QueryNodeMemoryHighWaterLevel        ParamItem `refreshable:"true"`
 	GrowingSegmentsSizeProtectionEnabled ParamItem `refreshable:"true"`
+	GrowingSegmentsSizeMinRateRatio      ParamItem `refreshable:"true"`
 	GrowingSegmentsSizeLowWaterLevel     ParamItem `refreshable:"true"`
 	GrowingSegmentsSizeHighWaterLevel    ParamItem `refreshable:"true"`
 	DiskProtectionEnabled                ParamItem `refreshable:"true"`
@@ -1014,13 +1015,29 @@ When memory usage < memoryLowWaterLevel, no action.`,
 		Key:          "quotaAndLimits.limitWriting.growingSegmentsSizeProtection.enabled",
 		Version:      "2.2.9",
 		DefaultValue: "false",
-		Doc: `1. No action will be taken if the ratio of growing segments size is less than the low water level.
-2. The DML rate will be reduced if the ratio of growing segments size is greater than the low water level and less than the high water level.
-3. All DML requests will be rejected if the ratio of growing segments size is greater than the high water level.`,
+		Doc: `No action will be taken if the growing segments size is less than the low watermark.
+When the growing segments size exceeds the low watermark, the dml rate will be reduced,
+but the rate will not be lower than minRateRatio * dmlRate.`,
 		Export: true,
 	}
 	p.GrowingSegmentsSizeProtectionEnabled.Init(base.mgr)
 
+	defaultGrowingSegSizeMinRateRatio := "0.5"
+	p.GrowingSegmentsSizeMinRateRatio = ParamItem{
+		Key:          "quotaAndLimits.limitWriting.growingSegmentsSizeProtection.minRateRatio",
+		Version:      "2.3.0",
+		DefaultValue: defaultGrowingSegSizeMinRateRatio,
+		Formatter: func(v string) string {
+			level := getAsFloat(v)
+			if level <= 0 || level > 1 {
+				return defaultGrowingSegSizeMinRateRatio
+			}
+			return v
+		},
+		Export: true,
+	}
+	p.GrowingSegmentsSizeMinRateRatio.Init(base.mgr)
+
 	defaultGrowingSegSizeLowWaterLevel := "0.2"
 	p.GrowingSegmentsSizeLowWaterLevel = ParamItem{
 		Key:          "quotaAndLimits.limitWriting.growingSegmentsSizeProtection.lowWaterLevel",

pkg/util/paramtable/quota_param_test.go

@@ -187,6 +187,7 @@ func TestQuotaParam(t *testing.T) {
 		assert.Equal(t, defaultLowWaterLevel, qc.QueryNodeMemoryLowWaterLevel.GetAsFloat())
 		assert.Equal(t, defaultHighWaterLevel, qc.QueryNodeMemoryHighWaterLevel.GetAsFloat())
 		assert.Equal(t, false, qc.GrowingSegmentsSizeProtectionEnabled.GetAsBool())
+		assert.Equal(t, 0.5, qc.GrowingSegmentsSizeMinRateRatio.GetAsFloat())
 		assert.Equal(t, 0.2, qc.GrowingSegmentsSizeLowWaterLevel.GetAsFloat())
 		assert.Equal(t, 0.4, qc.GrowingSegmentsSizeHighWaterLevel.GetAsFloat())
 		assert.Equal(t, true, qc.DiskProtectionEnabled.GetAsBool())