Ensure dangling cluster checks can be re-scheduled

pkg/clusteragent/clusterchecks/dispatcher_nodes.go

@@ -105,6 +105,8 @@ func (d *dispatcher) expireNodes() {
 			}
 			for digest, config := range node.digestToConfig {
 				delete(d.store.digestToNode, digest)
+				// Dangling configs are meant to be rescheduled, ensure they re-enter the pool of schedulable Cluster Checks.
+				config.ClusterCheck = true
 				d.store.danglingConfigs[digest] = config
 				danglingConfigs.Inc()
 			}

pkg/clusteragent/clusterchecks/dispatcher_test.go

@@ -228,6 +228,56 @@ func TestExpireNodes(t *testing.T) {
 	requireNotLocked(t, dispatcher.store)
 }
 
+func TestRescheduleDanglingFromExpiredNodes(t *testing.T) {
+	// This test case can represent a rollout of the cluster check workers
+	dispatcher := newDispatcher()
+
+	// Register a node with a correct status & schedule a Check
+	dispatcher.processNodeStatus("nodeA", types.NodeStatus{})
+	dispatcher.Schedule([]integration.Config{
+		generateIntegration("A")})
+
+	// Ensure it's dispatch correctly
+	allConfigs, err := dispatcher.getAllConfigs()
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(allConfigs))
+	assert.Equal(t, []string{"A"}, extractCheckNames(allConfigs))
+
+	// Ensure it's running correctly
+	configsA, _, err := dispatcher.getNodeConfigs("nodeA")
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(configsA))
+
+	// Expire NodeA
+	dispatcher.store.nodes["nodeA"].heartbeat = timestampNow() - 35
+
+	// Assert config becomes dangling when we trigger a expireNodes.
+	assert.Equal(t, 0, len(dispatcher.store.danglingConfigs))
+	dispatcher.expireNodes()
+	assert.Equal(t, 0, len(dispatcher.store.nodes))
+	assert.Equal(t, 1, len(dispatcher.store.danglingConfigs))
+
+	requireNotLocked(t, dispatcher.store)
+
+	// Register new node as healthy
+	dispatcher.processNodeStatus("nodeB", types.NodeStatus{})
+
+	// Ensure we have 1 dangling to schedule, as new available node is registered
+	assert.True(t, dispatcher.shouldDispatchDanling())
+	configs := dispatcher.retrieveAndClearDangling()
+	// Assert the check is scheduled
+	dispatcher.Schedule(configs)
+	danglingConfig, err := dispatcher.getAllConfigs()
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(danglingConfig))
+	assert.Equal(t, []string{"A"}, extractCheckNames(danglingConfig))
+
+	// Make sure make sure the dangling check is rescheduled on the new node
+	configsB, _, err := dispatcher.getNodeConfigs("nodeB")
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(configsB))
+}
+
 func TestDispatchFourConfigsTwoNodes(t *testing.T) {
 	dispatcher := newDispatcher()