Check isLeader in compliance checks if function given

cmd/cluster-agent/app/app.go

@@ -333,7 +333,7 @@ func start(cmd *cobra.Command, args []string) error {
 		go func() {
 			defer wg.Done()
 
-			if err := runCompliance(mainCtx); err != nil {
+			if err := runCompliance(mainCtx, apiCl, le.IsLeader); err != nil {
 				log.Errorf("Error while running compliance agent: %v", err)
 			}
 		}()

cmd/cluster-agent/app/compliance.go

@@ -29,14 +29,9 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
-func runCompliance(ctx context.Context) error {
-	apiCl, err := apiserver.WaitForAPIClient(ctx)
-	if err != nil {
-		return err
-	}
-
+func runCompliance(ctx context.Context, apiCl *apiserver.APIClient, isLeader func() bool) error {
 	stopper := restart.NewSerialStopper()
-	if err := startCompliance(stopper, apiCl); err != nil {
+	if err := startCompliance(stopper, apiCl, isLeader); err != nil {
 		return err
 	}
 
@@ -47,7 +42,7 @@ func runCompliance(ctx context.Context) error {
 }
 
 // TODO: Factorize code with pkg/compliance
-func startCompliance(stopper restart.Stopper, apiCl *apiserver.APIClient) error {
+func startCompliance(stopper restart.Stopper, apiCl *apiserver.APIClient, isLeader func() bool) error {
 	httpConnectivity := config.HTTPConnectivityFailure
 	if endpoints, err := config.BuildHTTPEndpoints(); err == nil {
 		httpConnectivity = http.CheckConnectivity(endpoints.Main)
@@ -105,6 +100,7 @@ func startCompliance(stopper restart.Stopper, apiCl *apiserver.APIClient) error
 			return rule.Scope.Includes(compliance.KubernetesClusterScope)
 		}),
 		checks.WithKubernetesClient(apiCl.DynamicCl),
+		checks.WithIsLeader(isLeader),
 	)
 	if err != nil {
 		return err

pkg/compliance/checks/builder.go

@@ -122,6 +122,14 @@ func WithKubernetesClient(cli env.KubeClient) BuilderOption {
 	}
 }
 
+// WithIsLeader allows check runner to know if its a leader instance or not (DCA)
+func WithIsLeader(isLeader func() bool) BuilderOption {
+	return func(b *builder) error {
+		b.isLeaderFunc = isLeader
+		return nil
+	}
+}
+
 // SuiteMatcher checks if a compliance suite is included
 type SuiteMatcher func(*compliance.SuiteMeta) bool
 
@@ -220,6 +228,7 @@ type builder struct {
 	dockerClient env.DockerClient
 	auditClient  env.AuditClient
 	kubeClient   env.KubeClient
+	isLeaderFunc func() bool
 
 	status *status
 }
@@ -506,6 +515,13 @@ func (b *builder) RelativeToHostRoot(path string) string {
 	return b.pathMapper.relativeToHostRoot(path)
 }
 
+func (b *builder) IsLeader() bool {
+	if b.isLeaderFunc != nil {
+		return b.isLeaderFunc()
+	}
+	return true
+}
+
 func (b *builder) EvaluateFromCache(ev eval.Evaluatable) (interface{}, error) {
 	instance := &eval.Instance{
 		Functions: eval.FunctionMap{

pkg/compliance/checks/check.go

@@ -77,6 +77,10 @@ func (c *complianceCheck) IsTelemetryEnabled() bool {
 }
 
 func (c *complianceCheck) Run() error {
+	if !c.IsLeader() {
+		return nil
+	}
+
 	report, err := c.checkable.check(c)
 	if err != nil {
 		log.Warnf("%s: check run failed: %v", c.ruleID, err)

pkg/compliance/checks/check_test.go

@@ -105,6 +105,7 @@ func TestCheckRun(t *testing.T) {
 			}
 
 			if test.configErr == nil {
+				env.On("IsLeader").Return(true)
 				env.On("Reporter").Return(reporter)
 				reporter.On("Report", test.expectEvent).Once()
 				checkable.On("check", check).Return(test.checkReport, test.checkErr)
@@ -115,3 +116,38 @@ func TestCheckRun(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckRunNoLeader(t *testing.T) {
+	const (
+		ruleID       = "rule-id"
+		resourceType = "resource-type"
+		resourceID   = "resource-id"
+	)
+
+	assert := assert.New(t)
+
+	env := &mocks.Env{}
+	defer env.AssertExpectations(t)
+
+	reporter := &mocks.Reporter{}
+	defer reporter.AssertExpectations(t)
+
+	checkable := &mockCheckable{}
+	defer checkable.AssertExpectations(t)
+
+	check := &complianceCheck{
+		Env: env,
+
+		ruleID:       ruleID,
+		resourceType: resourceType,
+		resourceID:   resourceID,
+		checkable:    checkable,
+	}
+
+	// Not leader
+	env.On("IsLeader").Return(false)
+	checkable.AssertNotCalled(t, "check")
+
+	err := check.Run()
+	assert.Nil(err)
+}

pkg/compliance/checks/env/env.go

@@ -31,4 +31,5 @@ type Configuration interface {
 	NormalizeToHostRoot(path string) string
 	RelativeToHostRoot(path string) string
 	EvaluateFromCache(e eval.Evaluatable) (interface{}, error)
+	IsLeader() bool
 }