etcd: extend rate limiting to consider the number of inflight requests

CODEOWNERS

@@ -469,6 +469,7 @@ jenkinsfiles @cilium/ci-structure
 /pkg/promise @cilium/sig-foundations
 /pkg/proxy/ @cilium/proxy
 /pkg/proxy/accesslog @cilium/api
+/pkg/rate/metrics @cilium/metrics
 /pkg/recorder @cilium/sig-datapath
 /pkg/redirectpolicy @cilium/sig-lb
 /pkg/safeio @cilium/sig-agent

clustermesh-apiserver/metrics/metrics.go

@@ -73,6 +73,11 @@ func (mm *metricsManager) Start(hive.HookContext) error {
 		metrics.KVStoreQuorumErrors,
 		metrics.KVStoreSyncQueueSize,
 		metrics.KVStoreInitialSyncCompleted,
+		metrics.APILimiterProcessingDuration,
+		metrics.APILimiterWaitDuration,
+		metrics.APILimiterRequestsInFlight,
+		metrics.APILimiterRateLimit,
+		metrics.APILimiterAdjustmentFactor,
 	)
 
 	mux := http.NewServeMux()

daemon/cmd/api_limits.go

@@ -6,7 +6,6 @@ package cmd
 import (
 	"time"
 
-	"github.com/cilium/cilium/pkg/metrics"
 	"github.com/cilium/cilium/pkg/rate"
 )
 
@@ -94,25 +93,3 @@ var apiRateLimitDefaults = map[string]rate.APILimiterParameters{
 		MinParallelRequests:         2,
 	},
 }
-
-type apiRateLimitingMetrics struct{}
-
-func (a *apiRateLimitingMetrics) ProcessedRequest(name string, v rate.MetricsValues) {
-	metrics.APILimiterProcessingDuration.WithLabelValues(name, "mean").Set(v.MeanProcessingDuration)
-	metrics.APILimiterProcessingDuration.WithLabelValues(name, "estimated").Set(v.EstimatedProcessingDuration)
-	metrics.APILimiterWaitDuration.WithLabelValues(name, "mean").Set(v.MeanWaitDuration)
-	metrics.APILimiterWaitDuration.WithLabelValues(name, "max").Set(v.MaxWaitDuration.Seconds())
-	metrics.APILimiterWaitDuration.WithLabelValues(name, "min").Set(v.MinWaitDuration.Seconds())
-	metrics.APILimiterRequestsInFlight.WithLabelValues(name, "in-flight").Set(float64(v.CurrentRequestsInFlight))
-	metrics.APILimiterRequestsInFlight.WithLabelValues(name, "limit").Set(float64(v.ParallelRequests))
-	metrics.APILimiterRateLimit.WithLabelValues(name, "limit").Set(float64(v.Limit))
-	metrics.APILimiterRateLimit.WithLabelValues(name, "burst").Set(float64(v.Burst))
-	metrics.APILimiterAdjustmentFactor.WithLabelValues(name).Set(v.AdjustmentFactor)
-
-	if v.Outcome == "" {
-		metrics.APILimiterWaitHistoryDuration.WithLabelValues(name).Observe(v.WaitDuration.Seconds())
-		v.Outcome = metrics.Error2Outcome(v.Error)
-	}
-
-	metrics.APILimiterProcessedRequests.WithLabelValues(name, v.Outcome).Inc()
-}

daemon/cmd/daemon.go

@@ -85,6 +85,7 @@ import (
 	policyAPI "github.com/cilium/cilium/pkg/policy/api"
 	"github.com/cilium/cilium/pkg/proxy"
 	"github.com/cilium/cilium/pkg/rate"
+	ratemetrics "github.com/cilium/cilium/pkg/rate/metrics"
 	"github.com/cilium/cilium/pkg/recorder"
 	"github.com/cilium/cilium/pkg/redirectpolicy"
 	"github.com/cilium/cilium/pkg/service"
@@ -418,7 +419,7 @@ func newDaemon(ctx context.Context, cleaner *daemonCleanup, params *daemonParams
 		log.WithField("mtu", configuredMTU).Info("Overwriting MTU based on CNI configuration")
 	}
 
-	apiLimiterSet, err := rate.NewAPILimiterSet(option.Config.APIRateLimit, apiRateLimitDefaults, &apiRateLimitingMetrics{})
+	apiLimiterSet, err := rate.NewAPILimiterSet(option.Config.APIRateLimit, apiRateLimitDefaults, ratemetrics.APILimiterObserver())
 	if err != nil {
 		log.WithError(err).Error("unable to configure API rate limiting")
 		return nil, nil, fmt.Errorf("unable to configure API rate limiting: %w", err)

pkg/clustermesh/remote_cluster.go

@@ -300,7 +300,7 @@ func (rc *remoteCluster) makeEtcdOpts() map[string]string {
 
 	for key, value := range option.Config.KVStoreOpt {
 		switch key {
-		case kvstore.EtcdRateLimitOption, kvstore.EtcdListLimitOption,
+		case kvstore.EtcdRateLimitOption, kvstore.EtcdMaxInflightOption, kvstore.EtcdListLimitOption,
 			kvstore.EtcdOptionKeepAliveHeartbeat, kvstore.EtcdOptionKeepAliveTimeout:
 			opts[key] = value
 		}