logs/sds: delete existing scanners when an update with 0 configuratio…

…n is received through RC (#25375)

* logs/sds: delete existing scanners when an update with 0 configuration is received through RC.

* logs/sds: report the errors from all reconfigured pipelines.

* logs/sds: add an unit test for this specific scenario.

* inv -e tidy-all

* Typo.

comp/logs/agent/agent.go

@@ -11,6 +11,7 @@ import (
  "fmt"
  "time"
 
+ "github.com/hashicorp/go-multierror"
  "go.uber.org/atomic"
  "go.uber.org/fx"
 
@@ -308,7 +309,7 @@ func (a *agent) onUpdateSDSRules(updates map[string]state.RawConfig, applyStateC
  var err error
  for _, config := range updates {
  if rerr := a.pipelineProvider.ReconfigureSDSStandardRules(config.Config); rerr != nil {
- err = rerr
+ err = multierror.Append(err, rerr)
  }
  }
 
@@ -333,9 +334,17 @@ func (a *agent) onUpdateSDSRules(updates map[string]state.RawConfig, applyStateC
 func (a *agent) onUpdateSDSAgentConfig(updates map[string]state.RawConfig, applyStateCallback func(string, state.ApplyStatus)) { //nolint:revive
  var err error
 
- for _, config := range updates {
- if rerr := a.pipelineProvider.ReconfigureSDSAgentConfig(config.Config); rerr != nil {
- err = rerr
+ // We received a hit that new updates arrived, but if the list of updates
+ // is empty, it means we don't have any updates applying to this agent anymore
+ // Send a reconfiguration with an empty payload, indicating that
+ // the scanners have to be dropped.
+ if len(updates) == 0 {
+ err = a.pipelineProvider.ReconfigureSDSAgentConfig([]byte("{}"))
+ } else {
+ for _, config := range updates {
+ if rerr := a.pipelineProvider.ReconfigureSDSAgentConfig(config.Config); rerr != nil {
+ err = multierror.Append(err, rerr)
+ }
  }
  }
 

comp/otelcol/logsagentpipeline/go.mod

@@ -109,6 +109,8 @@ require (
  github.com/go-ole/go-ole v1.2.6 // indirect
  github.com/gogo/protobuf v1.3.2 // indirect
  github.com/golang/protobuf v1.5.3 // indirect
+ github.com/hashicorp/errwrap v1.0.0 // indirect
+ github.com/hashicorp/go-multierror v1.1.1 // indirect
  github.com/hashicorp/hcl v1.0.0 // indirect
  github.com/hectane/go-acl v0.0.0-20190604041725-da78bae5fc95 // indirect
  github.com/inconshreveable/mousetrap v1.1.0 // indirect

comp/otelcol/logsagentpipeline/logsagentpipelineimpl/go.mod

@@ -120,6 +120,8 @@ require (
  github.com/go-ole/go-ole v1.2.6 // indirect
  github.com/gogo/protobuf v1.3.2 // indirect
  github.com/golang/protobuf v1.5.3 // indirect
+ github.com/hashicorp/errwrap v1.0.0 // indirect
+ github.com/hashicorp/go-multierror v1.1.1 // indirect
  github.com/hashicorp/hcl v1.0.0 // indirect
  github.com/hectane/go-acl v0.0.0-20190604041725-da78bae5fc95 // indirect
  github.com/inconshreveable/mousetrap v1.1.0 // indirect

pkg/logs/pipeline/go.mod

@@ -64,6 +64,7 @@ require (
  github.com/DataDog/datadog-agent/pkg/status/health v0.54.0-rc.2
  github.com/DataDog/datadog-agent/pkg/util/log v0.54.0-rc.2
  github.com/DataDog/datadog-agent/pkg/util/startstop v0.54.0-rc.2
+ github.com/hashicorp/go-multierror v1.1.1
  github.com/stretchr/testify v1.9.0
  go.uber.org/atomic v1.11.0
 )
@@ -109,6 +110,7 @@ require (
  github.com/go-ole/go-ole v1.2.6 // indirect
  github.com/gogo/protobuf v1.3.2 // indirect
  github.com/golang/protobuf v1.5.3 // indirect
+ github.com/hashicorp/errwrap v1.0.0 // indirect
  github.com/hashicorp/hcl v1.0.0 // indirect
  github.com/hectane/go-acl v0.0.0-20190604041725-da78bae5fc95 // indirect
  github.com/inconshreveable/mousetrap v1.1.0 // indirect

pkg/logs/pipeline/provider.go

@@ -8,6 +8,7 @@ package pipeline
 import (
  "context"
 
+ "github.com/hashicorp/go-multierror"
  "go.uber.org/atomic"
 
  "github.com/DataDog/datadog-agent/comp/core/hostname/hostnameinterface"
@@ -133,7 +134,7 @@ func (p *provider) reconfigureSDS(config []byte, orderType sds.ReconfigureOrderT
  for _, response := range responses {
  err := <-response
  if err != nil {
- rerr = err
+ rerr = multierror.Append(rerr, err)
  }
  close(response)
  }

pkg/logs/sds/scanner_test.go

@@ -203,6 +203,77 @@ func TestCreateScanner(t *testing.T) {
  require.Len(s.configuredRules, 0, "The group is disabled, no rules should be configured.")
 }
 
+// TestEmptyConfiguration validates that the scanner is destroyed when receiving
+// an empty configuration.
+func TestEmptyConfiguration(t *testing.T) {
+ require := require.New(t)
+
+ standardRules := []byte(`
+ {"priority":1,"is_enabled":true,"rules":[
+ {
+ "id":"zero-0",
+ "description":"zero desc",
+ "name":"zero",
+ "definitions": [{"version":1, "pattern":"zero"}]
+ },{
+ "id":"one-1",
+ "description":"one desc",
+ "name":"one",
+ "definitions": [{"version":1, "pattern":"one"}]
+ },{
+ "id":"two-2",
+ "description":"two desc",
+ "name":"two",
+ "definitions": [{"version":1, "pattern":"two"}]
+ }
+ ]}
+ `)
+ agentConfig := []byte(`
+ {"is_enabled":true,"rules":[
+ {
+ "id": "random000",
+ "name":"zero",
+ "definition":{"standard_rule_id":"zero-0"},
+ "match_action":{"type":"Redact","placeholder":"[redacted]"},
+ "is_enabled":true
+ }
+ ]}
+ `)
+
+ s := CreateScanner(0)
+
+ require.NotNil(s, "the scanner should not be nil after a creation")
+
+ err := s.Reconfigure(ReconfigureOrder{
+ Type: StandardRules,
+ Config: standardRules,
+ })
+
+ require.NoError(err, "configuring the standard rules should not fail")
+
+ // configure with one rule
+
+ err = s.Reconfigure(ReconfigureOrder{
+ Type: AgentConfig,
+ Config: agentConfig,
+ })
+
+ require.NoError(err, "this one should not fail since one rule is enabled")
+ require.Len(s.configuredRules, 1, "only one rules should be part of this scanner")
+ require.NotNil(s.Scanner)
+
+ // empty reconfiguration
+
+ err = s.Reconfigure(ReconfigureOrder{
+ Type: AgentConfig,
+ Config: []byte("{}"),
+ })
+
+ require.NoError(err)
+ require.Len(s.configuredRules, 0)
+ require.Nil(s.Scanner)
+}
+
 func TestIsReady(t *testing.T) {
  require := require.New(t)