limiter: Integrate adaptive limit to pack-objects limiterqmnguyen0711/integration-adaptive-calculation

This commit integrates the adaptive limit to pack-objects limiter of the
Gitaly server. It introduces some new configurations for pack-objects
limiting. When Gitaly server starts, it creates an adaptive calculator.
The calculator adjusts the adaptive limit which is already integrated
into the concurrency limit in prior commits.

The integration is hidden behind `gitaly_adaptive_limit_pack_objects`
feature flag. If this flag is not enabled, the limiter uses a static
limit from `max_concurrency` config.

4 files changed, 63 insertions, 5 deletions

diff --git a/internal/cli/gitaly/serve.go b/internal/cli/gitaly/serve.go
index b0a0c3833..814b282d9 100644
--- a/internal/cli/gitaly/serve.go
+++ b/internal/cli/gitaly/serve.go
@@ -19,6 +19,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v16/internal/bootstrap/starter"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/cache"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/cgroups"
+	"gitlab.com/gitlab-org/gitaly/v16/internal/featureflag"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/git"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/git/catfile"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/git/housekeeping"
@@ -42,6 +43,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v16/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/helper/env"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/limiter"
+	"gitlab.com/gitlab-org/gitaly/v16/internal/limiter/watchers"
 	glog "gitlab.com/gitlab-org/gitaly/v16/internal/log"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/streamcache"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/tempdir"
@@ -272,11 +274,43 @@ func run(cfg config.Cfg) error {
 		return fmt.Errorf("disk cache walkers: %w", err)
 	}
 
+	adaptiveLimits := []limiter.AdaptiveLimiter{}
+
 	//  The pack-objects limit below is static at this stage. It's always equal to the initial limit, which uses
 	//  MaxConcurrency config.
-	packObjectLimit := limiter.NewAdaptiveLimit("packObjects", limiter.AdaptiveSetting{
-		Initial: cfg.PackObjectsLimiting.MaxConcurrency,
-	})
+	var packObjectLimit *limiter.AdaptiveLimit
+	if cfg.PackObjectsLimiting.Adaptive && featureflag.AdaptiveLimitPackObjects.IsEnabled(ctx) {
+		packObjectLimit = limiter.NewAdaptiveLimit("packObjects", limiter.AdaptiveSetting{
+			Initial:       cfg.PackObjectsLimiting.InitialLimit,
+			Max:           cfg.PackObjectsLimiting.MaxLimit,
+			Min:           cfg.PackObjectsLimiting.MinLimit,
+			BackoffFactor: limiter.DefaultBackoffFactor,
+		})
+		adaptiveLimits = append(adaptiveLimits, packObjectLimit)
+	} else {
+		packObjectLimit = limiter.NewAdaptiveLimit("packObjects", limiter.AdaptiveSetting{
+			Initial: cfg.PackObjectsLimiting.MaxConcurrency,
+		})
+	}
+
+	// Always initializes the calculator. Adjust pack-object limit adaptively only if the feature flag is on.
+	adaptiveCalculator := limiter.NewAdaptiveCalculator(
+		limiter.DefaultCalibrateFrequency,
+		log.NewEntry(log.StandardLogger()),
+		adaptiveLimits,
+		[]limiter.ResourceWatcher{
+			watchers.NewCgroupCPUWatcher(cgroupMgr),
+			watchers.NewCgroupMemoryWatcher(cgroupMgr),
+		},
+	)
+	prometheus.MustRegister(adaptiveCalculator)
+
+	stop, err := adaptiveCalculator.Start(ctx)
+	if err != nil {
+		log.WithError(err).Warn("error starting adaptive limiter calculator")
+	}
+	defer stop()
+
 	concurrencyLimitHandler := limithandler.New(
 		cfg,
 		limithandler.LimitConcurrencyByRepo,
diff --git a/internal/featureflag/ff_adaptive_limit_pack_objects.go b/internal/featureflag/ff_adaptive_limit_pack_objects.go
new file mode 100644
index 000000000..8552b8b8e
--- /dev/null
+++ b/internal/featureflag/ff_adaptive_limit_pack_objects.go
@@ -0,0 +1,9 @@
+package featureflag
+
+// AdaptiveLimitPackObjects enables adaptive concurrency limit for pack-objects
+var AdaptiveLimitPackObjects = NewFeatureFlag(
+	"adaptive_limit_pack_objects",
+	"v16.3.0",
+	"https://gitlab.com/gitlab-org/gitaly/-/issues/5379",
+	false,
+)
diff --git a/internal/gitaly/config/config.go b/internal/gitaly/config/config.go
index 61d9261f0..33902ef8a 100644
--- a/internal/gitaly/config/config.go
+++ b/internal/gitaly/config/config.go
@@ -412,8 +412,18 @@ type RateLimiting struct {
 // Requests that come in after the maximum number of concurrent pack objects
 // processes have been reached will wait.
 type PackObjectsLimiting struct {
-	// MaxConcurrency is the maximum number of concurrent pack objects processes
-	// for a given key.
+	// Adaptive determines the behavior of the concurrency limit. If set to true, the concurrency limit is dynamic
+	// and starts at InitialLimit, then adjusts within the range [MinLimit, MaxLimit] based on current resource
+	// usage. If set to false, the concurrency limit is static and is set to MaxConcurrency.
+	Adaptive bool `toml:"adaptive,omitempty" json:"adaptive,omitempty"`
+	// InitialLimit is the concurrency limit to start with.
+	InitialLimit int `toml:"initial_limit,omitempty" json:"initial_limit,omitempty"`
+	// MaxLimit is the minimum adaptive concurrency limit.
+	MaxLimit int `toml:"max_limit,omitempty" json:"max_limit,omitempty"`
+	// MinLimit is the mini adaptive concurrency limit.
+	MinLimit int `toml:"min_limit,omitempty" json:"min_limit,omitempty"`
+	// MaxConcurrency is the static maximum number of concurrent pack objects processes for a given key. This config
+	// is used only if Adaptive is false.
 	MaxConcurrency int `toml:"max_concurrency,omitempty" json:"max_concurrency,omitempty"`
 	// MaxQueueWait is the maximum time a request can remain in the concurrency queue
 	// waiting to be picked up by Gitaly.
diff --git a/internal/limiter/adaptive_calculator.go b/internal/limiter/adaptive_calculator.go
index 793fc853c..ecbf1eb63 100644
--- a/internal/limiter/adaptive_calculator.go
+++ b/internal/limiter/adaptive_calculator.go
@@ -17,6 +17,11 @@ const (
 	// MaximumWatcherTimeout is the number of maximum allowed timeout when polling backoff events from watchers.
 	// When this threshold is reached, a timeout polling is treated as a backoff event.
 	MaximumWatcherTimeout = 5
+	// DefaultCalibrateFrequency is the default time period between two calibrations.
+	DefaultCalibrateFrequency = 30 * time.Second
+	// DefaultBackoffFactor is the default recommended backoff factor when the concurrency decreases. By default,
+	// the factor is 0.5, meaning the limit is cut off by half when a backoff event occurs.
+	DefaultBackoffFactor = 0.5
 )
 
 // BackoffEvent is a signal that the current system is under pressure. It's returned by the watchers under the