[ha-agent] Run agent integration only on leader agent

comp/collector/collector/collectorimpl/agent_check_metadata_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/DataDog/datadog-agent/comp/aggregator/demultiplexer/demultiplexerimpl"
 	"github.com/DataDog/datadog-agent/comp/core"
 	"github.com/DataDog/datadog-agent/comp/core/config"
+	haagentmock "github.com/DataDog/datadog-agent/comp/haagent/mock"
 	"github.com/DataDog/datadog-agent/pkg/collector/externalhost"
 	"github.com/DataDog/datadog-agent/pkg/serializer"
 	"github.com/DataDog/datadog-agent/pkg/util/fxutil"
@@ -36,6 +37,7 @@ func TestExternalHostTags(t *testing.T) {
 	c := newCollector(fxutil.Test[dependencies](t,
 		core.MockBundle(),
 		demultiplexerimpl.MockModule(),
+		haagentmock.Module(),
 		fx.Provide(func() optional.Option[serializer.MetricSerializer] {
 			return optional.NewNoneOption[serializer.MetricSerializer]()
 		}),

comp/collector/collector/collectorimpl/collector.go

@@ -24,6 +24,7 @@ import (
 	flaretypes "github.com/DataDog/datadog-agent/comp/core/flare/types"
 	log "github.com/DataDog/datadog-agent/comp/core/log/def"
 	"github.com/DataDog/datadog-agent/comp/core/status"
+	haagent "github.com/DataDog/datadog-agent/comp/haagent/def"
 	metadata "github.com/DataDog/datadog-agent/comp/metadata/runner/runnerimpl"
 	"github.com/DataDog/datadog-agent/pkg/aggregator/sender"
 	pkgCollector "github.com/DataDog/datadog-agent/pkg/collector"
@@ -48,17 +49,19 @@ const (
 type dependencies struct {
 	fx.In
 
-	Lc     fx.Lifecycle
-	Config config.Component
-	Log    log.Component
+	Lc      fx.Lifecycle
+	Config  config.Component
+	Log     log.Component
+	HaAgent haagent.Component
 
 	SenderManager    sender.SenderManager
 	MetricSerializer optional.Option[serializer.MetricSerializer]
 }
 
 type collectorImpl struct {
-	log    log.Component
-	config config.Component
+	log     log.Component
+	config  config.Component
+	haAgent haagent.Component
 
 	senderManager    sender.SenderManager
 	metricSerializer optional.Option[serializer.MetricSerializer]
@@ -119,6 +122,7 @@ func newCollector(deps dependencies) *collectorImpl {
 	c := &collectorImpl{
 		log:                deps.Log,
 		config:             deps.Config,
+		haAgent:            deps.HaAgent,
 		senderManager:      deps.SenderManager,
 		metricSerializer:   deps.MetricSerializer,
 		checks:             make(map[checkid.ID]*middleware.CheckWrapper),
@@ -186,7 +190,7 @@ func (c *collectorImpl) start(_ context.Context) error {
 	c.m.Lock()
 	defer c.m.Unlock()
 
-	run := runner.NewRunner(c.senderManager)
+	run := runner.NewRunner(c.senderManager, c.haAgent)
 	sched := scheduler.NewScheduler(run.GetChan())
 
 	// let the runner some visibility into the scheduler

comp/collector/collector/collectorimpl/collector_demux_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/DataDog/datadog-agent/comp/core/config"
 	log "github.com/DataDog/datadog-agent/comp/core/log/def"
 	logmock "github.com/DataDog/datadog-agent/comp/core/log/mock"
+	haagentmock "github.com/DataDog/datadog-agent/comp/haagent/mock"
 	checkid "github.com/DataDog/datadog-agent/pkg/collector/check/id"
 
 	"github.com/DataDog/datadog-agent/comp/aggregator/demultiplexer"
@@ -85,6 +86,7 @@ func (suite *CollectorDemuxTestSuite) SetupTest() {
 	suite.SenderManagerMock = NewSenderManagerMock(suite.demux)
 	suite.c = newCollector(fxutil.Test[dependencies](suite.T(),
 		core.MockBundle(),
+		haagentmock.Module(),
 		fx.Provide(func() sender.SenderManager {
 			return suite.SenderManagerMock
 		}),

comp/collector/collector/collectorimpl/collector_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/DataDog/datadog-agent/comp/collector/collector/collectorimpl/internal/middleware"
 	"github.com/DataDog/datadog-agent/comp/core"
 	"github.com/DataDog/datadog-agent/comp/core/config"
+	haagentmock "github.com/DataDog/datadog-agent/comp/haagent/mock"
 	"github.com/DataDog/datadog-agent/pkg/aggregator"
 	"github.com/DataDog/datadog-agent/pkg/collector/check"
 	checkid "github.com/DataDog/datadog-agent/pkg/collector/check/id"
@@ -97,6 +98,7 @@ func (suite *CollectorTestSuite) SetupTest() {
 	suite.c = newCollector(fxutil.Test[dependencies](suite.T(),
 		core.MockBundle(),
 		demultiplexerimpl.MockModule(),
+		haagentmock.Module(),
 		fx.Provide(func() optional.Option[serializer.MetricSerializer] {
 			return optional.NewNoneOption[serializer.MetricSerializer]()
 		}),

comp/haagent/def/component.go

@@ -22,4 +22,7 @@ type Component interface {
 	// SetLeader takes the leader agent hostname as input, if it matches the current agent hostname,
 	// the isLeader state is set to true, otherwise false.
 	SetLeader(leaderAgentHostname string)
+
+	// IsHaIntegration returns true if the integration type is an HA Integration
+	IsHaIntegration(checkType string) bool
 }

comp/haagent/impl/config.go

@@ -7,16 +7,42 @@ package haagentimpl
 
 import (
 	"github.com/DataDog/datadog-agent/comp/core/config"
+	log "github.com/DataDog/datadog-agent/comp/core/log/def"
 )
 
+type integrationConfig struct {
+	Name string `mapstructure:"name"`
+}
+
 type haAgentConfigs struct {
-	enabled bool
-	group   string
+	enabled            bool
+	group              string
+	isHaIntegrationMap map[string]bool
 }
 
-func newHaAgentConfigs(agentConfig config.Component) *haAgentConfigs {
+func newHaAgentConfigs(agentConfig config.Component, logger log.Component) *haAgentConfigs {
+	var integrationConfigs []integrationConfig
+	// TODO: DO NOT EXPOSE integrations CONFIG TO USERS
+	// TODO: DO NOT EXPOSE integrations CONFIG TO USERS
+	// TODO: DO NOT EXPOSE integrations CONFIG TO USERS
+	// TODO: DO NOT EXPOSE integrations CONFIG TO USERS
+
+	err := agentConfig.UnmarshalKey("ha_agent.integrations", &integrationConfigs)
+	if err != nil {
+		logger.Errorf("Error while reading 'ha_agent.integrationConfigs' settings: %v", err)
+	}
+	logger.Debugf("integration configs: %v", integrationConfigs)
+	integrationMap := make(map[string]bool)
+	for _, intg := range integrationConfigs {
+		if intg.Name == "" {
+			logger.Warn("Setging 'ha_agent.integrationConfigs[].name' shouldn't be empty")
+			continue
+		}
+		integrationMap[intg.Name] = true
+	}
 	return &haAgentConfigs{
-		enabled: agentConfig.GetBool("ha_agent.enabled"),
-		group:   agentConfig.GetString("ha_agent.group"),
+		enabled:            agentConfig.GetBool("ha_agent.enabled"),
+		group:              agentConfig.GetString("ha_agent.group"),
+		isHaIntegrationMap: integrationMap,
 	}
 }

comp/haagent/impl/config_test.go

@@ -0,0 +1,87 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2024-present Datadog, Inc.
+
+package haagentimpl
+
+import (
+	"testing"
+
+	"github.com/DataDog/datadog-agent/comp/core/config"
+	logmock "github.com/DataDog/datadog-agent/comp/core/log/mock"
+	"github.com/DataDog/datadog-agent/pkg/util/fxutil"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/fx"
+)
+
+func Test_newHaAgentConfigs(t *testing.T) {
+	tests := []struct {
+		name           string
+		agentConfigs   map[string]interface{}
+		expectedConfig *haAgentConfigs
+	}{
+		{
+			name: "ok",
+			agentConfigs: map[string]interface{}{
+				"ha_agent.enabled": true,
+				"ha_agent.group":   "mygroup01",
+				"ha_agent.integrations": []integrationConfig{
+					{Name: "snmp"},
+					{Name: "mysql"},
+				},
+			},
+			expectedConfig: &haAgentConfigs{
+				enabled: true,
+				group:   "mygroup01",
+				isHaIntegrationMap: map[string]bool{
+					"snmp":  true,
+					"mysql": true,
+				},
+			},
+		},
+		{
+			name: "invalid integrations",
+			agentConfigs: map[string]interface{}{
+				"ha_agent.enabled":      true,
+				"ha_agent.group":        "mygroup01",
+				"ha_agent.integrations": "abc",
+			},
+			expectedConfig: &haAgentConfigs{
+				enabled:            true,
+				group:              "mygroup01",
+				isHaIntegrationMap: map[string]bool{},
+			},
+		},
+		{
+			name: "empty integrationConfig name is skipped",
+			agentConfigs: map[string]interface{}{
+				"ha_agent.enabled": true,
+				"ha_agent.group":   "mygroup01",
+				"ha_agent.integrations": []integrationConfig{
+					{Name: "snmp"},
+					{Name: "mysql"},
+					{Name: ""},
+				},
+			},
+			expectedConfig: &haAgentConfigs{
+				enabled: true,
+				group:   "mygroup01",
+				isHaIntegrationMap: map[string]bool{
+					"snmp":  true,
+					"mysql": true,
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			logComponent := logmock.New(t)
+			agentConfigComponent := fxutil.Test[config.Component](t, fx.Options(
+				config.MockModule(),
+				fx.Replace(config.MockParams{Overrides: tt.agentConfigs}),
+			))
+			assert.Equal(t, tt.expectedConfig, newHaAgentConfigs(agentConfigComponent, logComponent))
+		})
+	}
+}

comp/haagent/impl/haagent.go

@@ -50,6 +50,10 @@ func (h *haAgentImpl) SetLeader(leaderAgentHostname string) {
 	h.isLeader.Store(agentHostname == leaderAgentHostname)
 }
 
+func (h *haAgentImpl) IsHaIntegration(integrationName string) bool {
+	return h.haAgentConfigs.isHaIntegrationMap[integrationName]
+}
+
 func (h *haAgentImpl) onHaAgentUpdate(updates map[string]state.RawConfig, applyStateCallback func(string, state.ApplyStatus)) {
 	h.log.Debugf("Updates received: count=%d", len(updates))
 

comp/haagent/impl/haagent_comp.go

@@ -28,7 +28,7 @@ type Provides struct {
 
 // NewComponent creates a new haagent component
 func NewComponent(reqs Requires) (Provides, error) {
-	haAgentConfigs := newHaAgentConfigs(reqs.AgentConfig)
+	haAgentConfigs := newHaAgentConfigs(reqs.AgentConfig, reqs.Logger)
 	haAgent := newHaAgentImpl(reqs.Logger, haAgentConfigs)
 	var rcListener rctypes.ListenerProvider
 	if haAgent.Enabled() {

comp/haagent/impl/haagent_test.go

@@ -68,6 +68,20 @@ func Test_IsLeader_SetLeader(t *testing.T) {
 	assert.True(t, haAgent.IsLeader())
 }
 
+func Test_IsHaIntegration(t *testing.T) {
+	agentConfigs := map[string]interface{}{
+		"ha_agent.integrations": []integrationConfig{
+			{Name: "snmp"},
+			{Name: "mysql"},
+		},
+	}
+	haAgent := newTestHaAgentComponent(t, agentConfigs)
+
+	assert.True(t, haAgent.IsHaIntegration("snmp"))
+	assert.True(t, haAgent.IsHaIntegration("mysql"))
+	assert.False(t, haAgent.IsHaIntegration("cpu"))
+}
+
 func Test_RCListener(t *testing.T) {
 	tests := []struct {
 		name             string

comp/haagent/mock/mock.go

@@ -44,6 +44,10 @@ func (m *mockHaAgent) SetEnabled(enabled bool) {
 	m.enabled = enabled
 }
 
+func (m *mockHaAgent) IsHaIntegration(_ string) bool {
+	return false
+}
+
 // Component is the component type.
 type Component interface {
 	haagent.Component

pkg/collector/runner/runner.go

@@ -14,6 +14,7 @@ import (
 
 	"go.uber.org/atomic"
 
+	haagent "github.com/DataDog/datadog-agent/comp/haagent/def"
 	"github.com/DataDog/datadog-agent/pkg/aggregator/sender"
 	"github.com/DataDog/datadog-agent/pkg/collector/check"
 	checkid "github.com/DataDog/datadog-agent/pkg/collector/check/id"
@@ -40,6 +41,7 @@ var (
 // Runner is the object in charge of running all the checks
 type Runner struct {
 	senderManager       sender.SenderManager
+	haAgent             haagent.Component
 	isRunning           *atomic.Bool
 	id                  int                           // Globally unique identifier for the Runner
 	workers             map[int]*worker.Worker        // Workers currrently under this Runner's management
@@ -52,11 +54,12 @@ type Runner struct {
 }
 
 // NewRunner takes the number of desired goroutines processing incoming checks.
-func NewRunner(senderManager sender.SenderManager) *Runner {
+func NewRunner(senderManager sender.SenderManager, haAgent haagent.Component) *Runner {
 	numWorkers := pkgconfigsetup.Datadog().GetInt("check_runners")
 
 	r := &Runner{
 		senderManager:       senderManager,
+		haAgent:             haAgent,
 		id:                  int(runnerIDGenerator.Inc()),
 		isRunning:           atomic.NewBool(true),
 		workers:             make(map[int]*worker.Worker),
@@ -117,6 +120,7 @@ func (r *Runner) AddWorker() {
 func (r *Runner) newWorker() (*worker.Worker, error) {
 	worker, err := worker.NewWorker(
 		r.senderManager,
+		r.haAgent,
 		r.id,
 		int(workerIDGenerator.Inc()),
 		r.pendingChecksChan,