weightedroundrobin: Implements slow start [WiP]

davinci26 · davinci26 · commit ac72d6601e60 · 2025-11-01T14:28:26.000-04:00
Implements slow start functionality as described in [gRFC A100](grpc/proposal#498) I still need to do some work on tests and do an e2e experiment with xDS to ensure that things are working as expected, but the core implementation is in a state that can be reviewed. At this phase I'm mainly looking for feedback on the overall approach and design and in the test code because I am worried that I might be missing some test utility that would make testing with `time` easier. Pending Items before this is ready to be merged: - [ ] Add more unit tests for slow start config cases - [ ] Manually verify with an xDS experiment that slow start is working as expected Release Notes * weightedroundrobin: Implements slow start functionality as described in [gRFC A100](grpc/proposal#498) Signed-off-by: sotiris <sotiris.nanopoulos@reddit.com>
diff --git a/balancer/weightedroundrobin/balancer.go b/balancer/weightedroundrobin/balancer.go
@@ -30,6 +30,7 @@ package weightedroundrobin
 import (
 	"encoding/json"
 	"fmt"
+	"math"
 	rand "math/rand/v2"
 	"sync"
 	"sync/atomic"
@@ -91,6 +92,15 @@ var (
 		OptionalLabels: []string{"grpc.lb.locality"},
 		Default:        false,
 	})
+
+	endpointsInSlowStartMetric = estats.RegisterInt64Count(estats.MetricDescriptor{
+		Name:           "grpc.lb.wrr.endpoints_in_slow_start",
+		Description:    "EXPERIMENTAL. Number of endpoints currently in the slow start period. This is incremented when a new scheduler is created.",
+		Unit:           "{endpoint}",
+		Labels:         []string{"grpc.target"},
+		OptionalLabels: []string{"grpc.lb.locality"},
+		Default:        false,
+	})
 )
 
 func init() {
@@ -132,6 +142,28 @@ func (bb) ParseConfig(js json.RawMessage) (serviceconfig.LoadBalancingConfig, er
 		return nil, fmt.Errorf("wrr: errorUtilizationPenalty must be non-negative")
 	}
 
+	// Validate slow start configuration if provided
+	if lbCfg.SlowStartConfig != nil {
+		ssc := lbCfg.SlowStartConfig
+		if ssc.Aggression != 0 && ssc.Aggression <= 0 {
+			return nil, fmt.Errorf("wrr: slowStartConfig.aggression must be greater than 0.0")
+		}
+		if ssc.MinWeightPercent != 0 && (ssc.MinWeightPercent < 0 || ssc.MinWeightPercent > 100) {
+			return nil, fmt.Errorf("wrr: slowStartConfig.minWeightPercent must be in range [0.0, 100.0]")
+		}
+		if ssc.SlowStartWindow != 0 && ssc.SlowStartWindow <= 0 {
+			return nil, fmt.Errorf("wrr: slowStartConfig.slowStartWindow must be greater than 0")
+		}
+
+		// Set defaults for slow start config
+		if ssc.Aggression == 0 {
+			ssc.Aggression = 1.0
+		}
+		if ssc.MinWeightPercent == 0 {
+			ssc.MinWeightPercent = 10.0
+		}
+	}
+
 	// For easier comparisons later, ensure the OOB reporting period is unset
 	// (0s) when OOB reports are disabled.
 	if !lbCfg.EnableOOBLoadReport {
@@ -361,6 +393,7 @@ func (b *wrrBalancer) updateSubConnState(sc balancer.SubConn, state balancer.Sub
 		ew.mu.Lock()
 		ew.nonEmptySince = time.Time{}
 		ew.lastUpdated = time.Time{}
+		ew.readySince = internal.TimeNow() // Set readySince for slow start period
 		cfg := ew.cfg
 		ew.mu.Unlock()
 		ew.updateORCAListener(cfg)
@@ -380,6 +413,9 @@ func (b *wrrBalancer) updateSubConnState(sc balancer.SubConn, state balancer.Sub
 		if ew.stopORCAListener != nil {
 			ew.stopORCAListener()
 		}
+		ew.mu.Lock()
+		ew.readySince = time.Time{} // Reset readySince when going non-READY
+		ew.mu.Unlock()
 		ew.pickedSC = nil
 	}
 }
@@ -427,7 +463,9 @@ func (p *picker) endpointWeights(recordMetrics bool) []float64 {
 	wp := make([]float64, len(p.weightedPickers))
 	now := internal.TimeNow()
 	for i, wpi := range p.weightedPickers {
-		wp[i] = wpi.weightedEndpoint.weight(now, time.Duration(p.cfg.WeightExpirationPeriod), time.Duration(p.cfg.BlackoutPeriod), recordMetrics)
+		baseWeight := wpi.weightedEndpoint.weight(now, time.Duration(p.cfg.WeightExpirationPeriod), time.Duration(p.cfg.BlackoutPeriod), recordMetrics)
+		slowStartScale := wpi.weightedEndpoint.slowStartScale(now, recordMetrics)
+		wp[i] = baseWeight * slowStartScale
 	}
 	return wp
 }
@@ -519,6 +557,7 @@ type endpointWeight struct {
 	weightVal     float64
 	nonEmptySince time.Time
 	lastUpdated   time.Time
+	readySince    time.Time // Time when endpoint transitioned to ready state (for slow start period)
 	cfg           *lbConfig
 }
 
@@ -634,3 +673,37 @@ func (w *endpointWeight) weight(now time.Time, weightExpirationPeriod, blackoutP
 
 	return w.weightVal
 }
+
+// slowStartScale returns the scaling factor for slow start.
+// Returns 1.0 if slow start is not configured or endpoint is outside slow start window.
+func (w *endpointWeight) slowStartScale(now time.Time, recordMetrics bool) float64 {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	// No config, slow start config, or readySince not set
+	if w.cfg == nil || w.cfg.SlowStartConfig == nil || w.readySince.Equal(time.Time{}) {
+		return 1.0
+	}
+
+	if recordMetrics {
+		defer func() {
+			endpointsInSlowStartMetric.Record(w.metricsRecorder, 1, w.target, w.locality)
+		}()
+	}
+
+	ssc := w.cfg.SlowStartConfig
+	slowStartWindow := time.Duration(ssc.SlowStartWindow)
+	timeSinceReady := now.Sub(w.readySince)
+	// Outside slow start window
+	if timeSinceReady >= slowStartWindow {
+		return 1.0
+	}
+
+	// Calculate scaling factor using the formula from gRFC A100:
+	// scaled_weight = weight * max(min_weight_percent/100, time_factor ^ (1/aggression))
+	// where time_factor = max(time_since_ready_seconds, 1) / slow_start_window_seconds
+	timeFactor := math.Max(timeSinceReady.Seconds(), 1.0) / slowStartWindow.Seconds()
+	minScale := ssc.MinWeightPercent / 100.0
+	scale := math.Pow(timeFactor, 1.0/ssc.Aggression)
+	return math.Max(minScale, scale)
+}
diff --git a/balancer/weightedroundrobin/balancer_test.go b/balancer/weightedroundrobin/balancer_test.go
@@ -67,28 +67,54 @@ func stringp(s string) *string    { return &s }
 
 var (
 	perCallConfig = iwrr.LBConfig{
+		EnableOOBLoadReport:     boolp(false),
+		OOBReportingPeriod:      stringp("0.005s"),
+		BlackoutPeriod:          stringp("0s"),
+		WeightExpirationPeriod:  stringp("60s"),
+		WeightUpdatePeriod:      stringp(".050s"),
+		ErrorUtilizationPenalty: float64p(0.),
+	}
+	perCallConfigWithSlowStart = iwrr.LBConfig{
 		EnableOOBLoadReport:     boolp(false),
 		OOBReportingPeriod:      stringp("0.005s"),
 		BlackoutPeriod:          stringp("0s"),
 		WeightExpirationPeriod:  stringp("60s"),
 		WeightUpdatePeriod:      stringp(".050s"),
 		ErrorUtilizationPenalty: float64p(0),
+		SlowStartConfig: &iwrr.SlowStartConfig{
+			SlowStartWindow:  stringp("30s"),
+			Aggression:       float64p(1.0),
+			MinWeightPercent: float64p(10.0),
+		},
 	}
 	oobConfig = iwrr.LBConfig{
 		EnableOOBLoadReport:     boolp(true),
 		OOBReportingPeriod:      stringp("0.005s"),
 		BlackoutPeriod:          stringp("0s"),
 		WeightExpirationPeriod:  stringp("60s"),
 		WeightUpdatePeriod:      stringp(".050s"),
-		ErrorUtilizationPenalty: float64p(0),
+		ErrorUtilizationPenalty: float64p(0.),
+	}
+	oobConfigWithSlowStart = iwrr.LBConfig{
+		EnableOOBLoadReport:     boolp(true),
+		OOBReportingPeriod:      stringp("0.005s"),
+		BlackoutPeriod:          stringp("0s"),
+		WeightExpirationPeriod:  stringp("60s"),
+		WeightUpdatePeriod:      stringp(".050s"),
+		ErrorUtilizationPenalty: float64p(0.),
+		SlowStartConfig: &iwrr.SlowStartConfig{
+			SlowStartWindow:  stringp("30s"),
+			Aggression:       float64p(1.0),
+			MinWeightPercent: float64p(10.0),
+		},
 	}
 	testMetricsConfig = iwrr.LBConfig{
 		EnableOOBLoadReport:     boolp(false),
 		OOBReportingPeriod:      stringp("0.005s"),
 		BlackoutPeriod:          stringp("0s"),
 		WeightExpirationPeriod:  stringp("60s"),
 		WeightUpdatePeriod:      stringp("30s"),
-		ErrorUtilizationPenalty: float64p(0),
+		ErrorUtilizationPenalty: float64p(0.0),
 	}
 )
 
@@ -176,8 +202,11 @@ func (s) TestBalancer_OneAddress(t *testing.T) {
 		cfg iwrr.LBConfig
 	}{
 		{rt: reportNone, cfg: perCallConfig},
+		{rt: reportCall, cfg: perCallConfigWithSlowStart},
 		{rt: reportCall, cfg: perCallConfig},
+		{rt: reportCall, cfg: perCallConfigWithSlowStart},
 		{rt: reportOOB, cfg: oobConfig},
+		{rt: reportOOB, cfg: oobConfigWithSlowStart},
 	}
 
 	for _, tc := range testCases {
@@ -652,6 +681,66 @@ func (s) TestBalancer_TwoAddresses_BlackoutPeriod(t *testing.T) {
 	}
 }
 
+func (s) TestBalancer_SlowStart(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	srv1 := startServer(t, reportBoth)
+	srv2 := startServer(t, reportBoth)
+	srv3 := startServer(t, reportBoth)
+
+	// srv1, srv2, srv3 all start loaded equally.
+	srv1.oobMetrics.SetQPS(10.0)
+	srv1.oobMetrics.SetApplicationUtilization(1.0)
+
+	srv2.oobMetrics.SetQPS(10.0)
+	srv2.oobMetrics.SetApplicationUtilization(1.0)
+
+	srv3.oobMetrics.SetQPS(10.0)
+	srv3.oobMetrics.SetApplicationUtilization(1.0)
+
+	slowStartPeriod := 2 * time.Second
+	slowStartConfig := &iwrr.SlowStartConfig{
+		SlowStartWindow:  stringp("2s"),
+		Aggression:       float64p(1.0),
+		MinWeightPercent: float64p(10.0),
+	}
+	cfg := oobConfig
+	cfg.BlackoutPeriod = stringp("0.1s")
+	cfg.SlowStartConfig = slowStartConfig
+	sc := svcConfig(t, cfg)
+
+	if err := srv1.StartClient(grpc.WithDefaultServiceConfig(sc)); err != nil {
+		t.Fatalf("Error starting client: %v", err)
+	}
+
+	addrs := []resolver.Address{{Addr: srv1.Address}, {Addr: srv2.Address}}
+	srv1.R.UpdateState(resolver.State{Addresses: addrs})
+
+	// Call each backend once to ensure the weights have been received.
+	ensureReached(ctx, t, srv1.Client, 2)
+	time.Sleep(slowStartPeriod + weightUpdatePeriod)
+	checkWeights(ctx, t, srvWeight{srv1, 10}, srvWeight{srv2, 10})
+
+	// Add backend 3
+	addrs = append(addrs, resolver.Address{Addr: srv3.Address})
+	srv1.R.UpdateState(resolver.State{Addresses: addrs})
+	ensureReached(ctx, t, srv1.Client, 3)
+
+	// validate that srv3 is in slow start
+	time.Sleep(weightUpdatePeriod)
+	checkWeights(ctx, t, srvWeight{srv1, 10}, srvWeight{srv2, 10}, srvWeight{srv3, 5})
+
+	// validate that srv3 exits slow start
+	time.Sleep(slowStartPeriod + weightUpdatePeriod)
+	checkWeights(ctx, t, srvWeight{srv1, 10}, srvWeight{srv2, 10}, srvWeight{srv3, 10})
+	fmt.Printf("Test log ---------------------> End\n")
+	if srv1.CC != nil {
+		srv1.CC.Close()
+		time.Sleep(weightUpdatePeriod) // Wait for cleanup
+	}
+}
+
 // Tests that the weight expiration period causes backends to use 0 as their
 // weight (meaning to use the average weight) once the expiration period
 // elapses.
@@ -789,7 +878,9 @@ func (s) TestBalancer_AddressesChanging(t *testing.T) {
 func ensureReached(ctx context.Context, t *testing.T, c testgrpc.TestServiceClient, n int) {
 	t.Helper()
 	reached := make(map[string]struct{})
+	i := 0
 	for len(reached) != n {
+		i++
 		var peer peer.Peer
 		if _, err := c.EmptyCall(ctx, &testpb.Empty{}, grpc.Peer(&peer)); err != nil {
 			t.Fatalf("Error from EmptyCall: %v", err)
diff --git a/balancer/weightedroundrobin/config.go b/balancer/weightedroundrobin/config.go
@@ -56,4 +56,35 @@ type lbConfig struct {
 	// The multiplier used to adjust endpoint weights with the error rate
 	// calculated as eps/qps. Default is 1.0.
 	ErrorUtilizationPenalty float64 `json:"errorUtilizationPenalty,omitempty"`
+
+	// Configuration for slow start feature
+	SlowStartConfig *SlowStartConfig `json:"slowStartConfig,omitempty"`
+}
+
+// SlowStartConfig contains configuration for the slow start feature.
+type SlowStartConfig struct {
+	// Represents the size of slow start window.
+	//
+	// The newly created endpoint remains in slow start mode starting from its creation time
+	// for the duration of slow start window.
+	SlowStartWindow iserviceconfig.Duration `json:"slowStartWindow,omitempty"`
+
+	// This parameter controls the speed of traffic increase over the slow start window. Defaults to 1.0,
+	// so that endpoint would get linearly increasing amount of traffic.
+	// When increasing the value for this parameter, the speed of traffic ramp-up increases non-linearly.
+	// The value of aggression parameter must be greater than 0.0.
+	// By tuning the parameter, it is possible to achieve polynomial or exponential shape of ramp-up curve.
+	//
+	// During slow start window, effective weight of an endpoint would be scaled with time factor and aggression:
+	// ``new_weight = weight * max(min_weight_percent / 100, time_factor ^ (1 / aggression))``,
+	// where ``time_factor = max(time_since_start_seconds, 1) / slow_start_window_seconds``.
+	//
+	// As time progresses, more and more traffic would be sent to endpoint, which is in slow start window.
+	// Once endpoint exits slow start, time_factor and aggression no longer affect its weight.
+	Aggression float64 `json:"aggression,omitempty"`
+
+	// Configures the minimum percentage of the original weight that will be used for an endpoint
+	// in slow start. This helps to avoid a scenario in which endpoints receive no traffic during the
+	// slow start window. Valid range is [0.0, 100.0]. If the value is not specified, the default is 10%.
+	MinWeightPercent float64 `json:"minWeightPercent,omitempty"`
 }
diff --git a/balancer/weightedroundrobin/internal/internal.go b/balancer/weightedroundrobin/internal/internal.go
@@ -31,12 +31,19 @@ var AllowAnyWeightUpdatePeriod bool
 // LBConfig allows tests to produce a JSON form of the config from the struct
 // instead of using a string.
 type LBConfig struct {
-	EnableOOBLoadReport     *bool    `json:"enableOobLoadReport,omitempty"`
-	OOBReportingPeriod      *string  `json:"oobReportingPeriod,omitempty"`
-	BlackoutPeriod          *string  `json:"blackoutPeriod,omitempty"`
-	WeightExpirationPeriod  *string  `json:"weightExpirationPeriod,omitempty"`
-	WeightUpdatePeriod      *string  `json:"weightUpdatePeriod,omitempty"`
-	ErrorUtilizationPenalty *float64 `json:"errorUtilizationPenalty,omitempty"`
+	EnableOOBLoadReport     *bool            `json:"enableOobLoadReport,omitempty"`
+	OOBReportingPeriod      *string          `json:"oobReportingPeriod,omitempty"`
+	BlackoutPeriod          *string          `json:"blackoutPeriod,omitempty"`
+	WeightExpirationPeriod  *string          `json:"weightExpirationPeriod,omitempty"`
+	WeightUpdatePeriod      *string          `json:"weightUpdatePeriod,omitempty"`
+	ErrorUtilizationPenalty *float64         `json:"errorUtilizationPenalty,omitempty"`
+	SlowStartConfig         *SlowStartConfig `json:"slowStartConfig,omitempty"`
+}
+
+type SlowStartConfig struct {
+	SlowStartWindow  *string  `json:"slowStartWindow,omitempty"`
+	Aggression       *float64 `json:"aggression,omitempty"`
+	MinWeightPercent *float64 `json:"minWeightPercent,omitempty"`
 }
 
 // TimeNow can be overridden by tests to return a different value for the
diff --git a/go.mod b/go.mod
@@ -4,9 +4,9 @@ go 1.24.0
 
 require (
 	github.com/cespare/xxhash/v2 v2.3.0
-	github.com/cncf/xds/go v0.0.0-20251022180443-0feb69152e9f
-	github.com/envoyproxy/go-control-plane v0.13.5-0.20251024222203-75eaa193e329
-	github.com/envoyproxy/go-control-plane/envoy v1.35.0
+	github.com/cncf/xds/go v0.0.0-20251014123835-2ee22ca58382
+	github.com/envoyproxy/go-control-plane v0.13.4
+	github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251015221300-4138018a492b
 	github.com/golang/glog v1.2.5
 	github.com/golang/protobuf v1.5.4
 	github.com/google/go-cmp v0.7.0
diff --git a/go.sum b/go.sum
@@ -6,14 +6,14 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0/go.mod h1:P4WPRUkOhJC13W//jWpyfJNDAIpvRbAUIYLX/4jtlE0=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
-github.com/cncf/xds/go v0.0.0-20251022180443-0feb69152e9f h1:Y8xYupdHxryycyPlc9Y+bSQAYZnetRJ70VMVKm5CKI0=
-github.com/cncf/xds/go v0.0.0-20251022180443-0feb69152e9f/go.mod h1:HlzOvOjVBOfTGSRXRyY0OiCS/3J1akRGQQpRO/7zyF4=
+github.com/cncf/xds/go v0.0.0-20251014123835-2ee22ca58382 h1:5IeUoAZvqwF6LcCnV99NbhrGKN6ihZgahJv5jKjmZ3k=
+github.com/cncf/xds/go v0.0.0-20251014123835-2ee22ca58382/go.mod h1:HlzOvOjVBOfTGSRXRyY0OiCS/3J1akRGQQpRO/7zyF4=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/envoyproxy/go-control-plane v0.13.5-0.20251024222203-75eaa193e329 h1:K+fnvUM0VZ7ZFJf0n4L/BRlnsb9pL/GuDG6FqaH+PwM=
-github.com/envoyproxy/go-control-plane v0.13.5-0.20251024222203-75eaa193e329/go.mod h1:Alz8LEClvR7xKsrq3qzoc4N0guvVNSS8KmSChGYr9hs=
-github.com/envoyproxy/go-control-plane/envoy v1.35.0 h1:ixjkELDE+ru6idPxcHLj8LBVc2bFP7iBytj353BoHUo=
-github.com/envoyproxy/go-control-plane/envoy v1.35.0/go.mod h1:09qwbGVuSWWAyN5t/b3iyVfz5+z8QWGrzkoqm/8SbEs=
+github.com/envoyproxy/go-control-plane v0.13.4 h1:zEqyPVyku6IvWCFwux4x9RxkLOMUL+1vC9xUFv5l2/M=
+github.com/envoyproxy/go-control-plane v0.13.4/go.mod h1:kDfuBlDVsSj2MjrLEtRWtHlsWIFcGyB2RMO44Dc5GZA=
+github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251015221300-4138018a492b h1:rDyuvoLwwYo/TeYQF/6Ag7BG/LYbrnLp3qvXzzF5JRk=
+github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251015221300-4138018a492b/go.mod h1:ty89S1YCCVruQAm9OtKeEkQLTb+Lkz0k8v9W0Oxsv98=
 github.com/envoyproxy/go-control-plane/ratelimit v0.1.0 h1:/G9QYbddjL25KvtKTv3an9lx6VBE2cnb8wp1vEGNYGI=
 github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4=
 github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8=
diff --git a/internal/xds/xdsclient/xdslbregistry/converter/converter.go b/internal/xds/xdsclient/xdslbregistry/converter/converter.go
