leader election

DataDog · Dec 28, 2024 · adf5884 · adf5884
1 parent 27a1812
commit adf5884
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 7 deletions.
diff --git a/cmd/cluster-agent/api/v2/series/series.go b/cmd/cluster-agent/api/v2/series/series.go
@@ -22,14 +22,30 @@ import (
 )
 
 const (
-	encodingGzip    = "gzip"
-	encodingDeflate = "deflate"
+	encodingGzip           = "gzip"
+	encodingDeflate        = "deflate"
+	loadMetricsHandlerName = "load-metrics-handler"
 )
 
 // InstallNodeMetricsEndpoints register handler for node metrics collection
-func InstallNodeMetricsEndpoints(ctx context.Context, r *mux.Router, _ config.Component) {
-	handler := newSeriesHandler(ctx)
-	r.HandleFunc("/series", api.WithTelemetryWrapper("load-metrics-handler", handler.handle)).Methods("POST")
+func InstallNodeMetricsEndpoints(ctx context.Context, r *mux.Router, cfg config.Component) {
+	leaderHander := newSeriesHandler(ctx)
+	handler := api.WithLeaderProxyHandler(
+		loadMetricsHandlerName,
+		func(w http.ResponseWriter, r *http.Request) bool { // preHandler
+			if !cfg.GetBool("autoscaling.workload.enabled") {
+				http.Error(w, "Autoscaling workload failover store is disabled on the cluster agent", http.StatusServiceUnavailable)
+				return false
+			}
+			if r.Body == nil {
+				http.Error(w, "Request body is empty", http.StatusBadRequest)
+				return false
+			}
+			return true
+		},
+		leaderHander.handle,
+	)
+	r.HandleFunc("/series", api.WithTelemetryWrapper(loadMetricsHandlerName, handler)).Methods("POST")
 }
 
 // Handler handles the series request and store the metrics to loadstore

diff --git a/cmd/cluster-agent/subcommands/start/command.go b/cmd/cluster-agent/subcommands/start/command.go
@@ -267,8 +267,10 @@ func start(log log.Component,
 		}
 	}()
 
-	// Setup the leader forwarder for language detection and cluster checks
-	if config.GetBool("cluster_checks.enabled") || (config.GetBool("language_detection.enabled") && config.GetBool("language_detection.reporting.enabled")) {
+	// Setup the leader forwarder for autoscaling failover store, language detection and cluster checks
+	if config.GetBool("cluster_checks.enabled") ||
+		(config.GetBool("language_detection.enabled") && config.GetBool("language_detection.reporting.enabled")) ||
+		config.GetBool("autoscaling.workload.enabled") {
 		apidca.NewGlobalLeaderForwarder(
 			config.GetInt("cluster_agent.cmd_port"),
 			config.GetInt("cluster_agent.max_connections"),