[clusteragent/clusterchecks/dispatcher] Pick random node for new checks when advanced dispatching is enabled (DataDog#18914)

Author: davidor

pkg/clusteragent/clusterchecks/dispatcher_main.go

@@ -147,7 +147,7 @@ func (d *dispatcher) reschedule(configs []integration.Config) {
 
 // add stores and delegates a given configuration
 func (d *dispatcher) add(config integration.Config) {
-	target := d.getLeastBusyNode()
+	target := d.getNodeToScheduleCheck()
 	if target == "" {
 		// If no node is found, store it in the danglingConfigs map for retrying later.
 		log.Warnf("No available node to dispatch %s:%s on, will retry later", config.Name, config.Digest())

pkg/clusteragent/clusterchecks/dispatcher_nodes.go

@@ -9,6 +9,7 @@ package clusterchecks
 
 import (
 	"fmt"
+	"math/rand"
 	"time"
 
 	"github.com/DataDog/datadog-agent/pkg/autodiscovery/integration"
@@ -75,38 +76,57 @@ func (d *dispatcher) processNodeStatus(nodeName, clientIP string, status types.N
 	return false
 }
 
-// getLeastBusyNode returns the name of the node that is assigned
-// the lowest number of checks. In case of equality, one is chosen
-// randomly, based on map iterations being randomized.
-func (d *dispatcher) getLeastBusyNode() string {
-	var leastBusyNode string
-	minCheckCount := int(-1)
-	minBusyness := int(-1)
+// getNodeToScheduleCheck returns the node where a new check should be scheduled
+
+// Advanced dispatching relies on the check stats fetched from the cluster check
+// runners API to distribute the checks. The stats are only updated when the
+// checks are rebalanced, they are not updated every time a check is scheduled.
+// That's why it's not a good idea to pick the least busy node. Rebalance
+// happens every few minutes, so all the checks added during that time would get
+// scheduled to the same node. It's a better solution to pick a random node and
+// rely on rebalancing to distribute when needed.
+//
+// On the other hand, when advanced dispatching is not used, we can pick the
+// node with fewer checks. It's because the number of checks is kept up to date.
+func (d *dispatcher) getNodeToScheduleCheck() string {
+	if d.advancedDispatching {
+		return d.getRandomNode()
+	}
+
+	return d.getNodeWithLessChecks()
+}
+
+func (d *dispatcher) getRandomNode() string {
+	d.store.RLock()
+	defer d.store.RUnlock()
+
+	var nodes []string
+	for name := range d.store.nodes {
+		nodes = append(nodes, name)
+	}
+
+	if len(nodes) == 0 {
+		return ""
+	}
+
+	return nodes[rand.Intn(len(nodes))]
+}
 
+func (d *dispatcher) getNodeWithLessChecks() string {
 	d.store.RLock()
 	defer d.store.RUnlock()
 
+	var selectedNode string
+	minNumChecks := 0
+
 	for name, store := range d.store.nodes {
-		if name == "" {
-			continue
-		}
-		if d.advancedDispatching && store.busyness > defaultBusynessValue {
-			// dispatching based on clc runners stats
-			// only when advancedDispatching is true and
-			// started collecting busyness values
-			if minBusyness == -1 || store.busyness < minBusyness {
-				leastBusyNode = name
-				minBusyness = store.busyness
-			}
-		} else {
-			// count-based round robin dispatching
-			if minCheckCount == -1 || len(store.digestToConfig) < minCheckCount {
-				leastBusyNode = name
-				minCheckCount = len(store.digestToConfig)
-			}
+		if selectedNode == "" || len(store.digestToConfig) < minNumChecks {
+			selectedNode = name
+			minNumChecks = len(store.digestToConfig)
 		}
 	}
-	return leastBusyNode
+
+	return selectedNode
 }
 
 // expireNodes iterates over nodes and removes the ones that have not

pkg/clusteragent/clusterchecks/dispatcher_test.go

@@ -231,26 +231,26 @@ func TestProcessNodeStatus(t *testing.T) {
 	requireNotLocked(t, dispatcher.store)
 }
 
-func TestGetLeastBusyNode(t *testing.T) {
+func TestGetNodeWithLessChecks(t *testing.T) {
 	dispatcher := newDispatcher()
 
 	// No node registered -> empty string
-	assert.Equal(t, "", dispatcher.getLeastBusyNode())
+	assert.Equal(t, "", dispatcher.getNodeWithLessChecks())
 
 	// 1 config on node1, 2 on node2
 	dispatcher.addConfig(generateIntegration("A"), "node1")
 	dispatcher.addConfig(generateIntegration("B"), "node2")
 	dispatcher.addConfig(generateIntegration("C"), "node2")
-	assert.Equal(t, "node1", dispatcher.getLeastBusyNode())
+	assert.Equal(t, "node1", dispatcher.getNodeWithLessChecks())
 
 	// 3 configs on node1, 2 on node2
 	dispatcher.addConfig(generateIntegration("D"), "node1")
 	dispatcher.addConfig(generateIntegration("E"), "node1")
-	assert.Equal(t, "node2", dispatcher.getLeastBusyNode())
+	assert.Equal(t, "node2", dispatcher.getNodeWithLessChecks())
 
 	// Add an empty node3
 	dispatcher.processNodeStatus("node3", "10.0.0.3", types.NodeStatus{})
-	assert.Equal(t, "node3", dispatcher.getLeastBusyNode())
+	assert.Equal(t, "node3", dispatcher.getNodeWithLessChecks())
 
 	requireNotLocked(t, dispatcher.store)
 }

releasenotes-dca/notes/schedule-to-random-when-using-advance-dispatching-03915e197be5ad36.yaml

@@ -0,0 +1,13 @@
+# Each section from every release note are combined when the
+# CHANGELOG.rst is rendered. So the text needs to be worded so that
+# it does not depend on any information only available in another
+# section. This may mean repeating some details, but each section
+# must be readable independently of the other.
+#
+# Each section note must be formatted as reStructuredText.
+---
+fixes:
+  - |
+    Fixed a bug in the advanced dispatching of cluster checks. All the checks
+    scheduled since the last rebalance were being scheduled in the same node.
+    Now they should be distributed among the available nodes.