enhance: Remove docker acquis internal timer use docker events

pkg/acquisition/modules/docker/docker.go

@@ -11,16 +11,17 @@
 	"strings"
 	"time"
 
+	"github.com/crowdsecurity/dlog"
 	dockerTypes "github.com/docker/docker/api/types"
 	dockerContainer "github.com/docker/docker/api/types/container"
+	dockerTypesEvents "github.com/docker/docker/api/types/events"
+	dockerFilter "github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/client"
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 	"gopkg.in/tomb.v2"
 	"gopkg.in/yaml.v2"
 
-	"github.com/crowdsecurity/dlog"
-
 	"github.com/crowdsecurity/crowdsec/pkg/acquisition/configuration"
 	"github.com/crowdsecurity/crowdsec/pkg/types"
 )
@@ -53,7 +54,6 @@
 	runningContainerState map[string]*ContainerConfig
 	compiledContainerName []*regexp.Regexp
 	compiledContainerID   []*regexp.Regexp
-	CheckIntervalDuration time.Duration
 	logger                *log.Entry
 	Client                client.CommonAPIClient
 	t                     *tomb.Tomb
@@ -75,9 +75,8 @@
 
 func (d *DockerSource) UnmarshalConfig(yamlConfig []byte) error {
 	d.Config = DockerConfiguration{
-		FollowStdout:  true, // default
-		FollowStdErr:  true, // default
-		CheckInterval: "1s", // default
+		FollowStdout: true, // default
+		FollowStdErr: true, // default
 	}
 
 	err := yaml.UnmarshalStrict(yamlConfig, &d.Config)
@@ -97,9 +96,8 @@
 		return errors.New("use_container_labels and container_name, container_id, container_id_regexp, container_name_regexp are mutually exclusive")
 	}
 
-	d.CheckIntervalDuration, err = time.ParseDuration(d.Config.CheckInterval)
-	if err != nil {
-		return fmt.Errorf("parsing 'check_interval' parameters: %s", d.CheckIntervalDuration)
+	if d.Config.CheckInterval != "" {
+		d.logger.Warn("check_interval is deprecated, it will be removed in a future version")
 	}
 
 	if d.Config.Mode == "" {
@@ -495,63 +493,155 @@
 	return nil
 }
 
+func (d *DockerSource) checkContainers(ctx context.Context, monitChan chan *ContainerConfig, deleteChan chan *ContainerConfig) error {
+	// to track for garbage collection
+	runningContainersID := make(map[string]bool)
+
+	runningContainers, err := d.Client.ContainerList(ctx, dockerContainer.ListOptions{})
+	if err != nil {
+		if strings.Contains(strings.ToLower(err.Error()), "cannot connect to the docker daemon at") {
+			for idx, container := range d.runningContainerState {
+				if d.runningContainerState[idx].t.Alive() {
+					d.logger.Infof("killing tail for container %s", container.Name)
+					d.runningContainerState[idx].t.Kill(nil)
+
+					if err := d.runningContainerState[idx].t.Wait(); err != nil {
+						d.logger.Infof("error while waiting for death of %s : %s", container.Name, err)
+					}
+				}
+
+				delete(d.runningContainerState, idx)
+			}
+		} else {
+			log.Errorf("container list err: %s", err)
+		}
+
+		return err
+	}
+
+	for _, container := range runningContainers {
+		runningContainersID[container.ID] = true
+
+		// don't need to re eval an already monitored container
+		if _, ok := d.runningContainerState[container.ID]; ok {
+			continue
+		}
+
+		if containerConfig := d.EvalContainer(ctx, container); containerConfig != nil {
+			monitChan <- containerConfig
+		}
+	}
+
+	for containerStateID, containerConfig := range d.runningContainerState {
+		if _, ok := runningContainersID[containerStateID]; !ok {
+			deleteChan <- containerConfig
+		}
+	}
+
+	d.logger.Tracef("Reading logs from %d containers", len(d.runningContainerState))
+	return nil
+}
+
 func (d *DockerSource) WatchContainer(ctx context.Context, monitChan chan *ContainerConfig, deleteChan chan *ContainerConfig) error {
-	ticker := time.NewTicker(d.CheckIntervalDuration)
-	d.logger.Infof("Container watcher started, interval: %s", d.CheckIntervalDuration.String())
+	err := d.checkContainers(ctx, monitChan, deleteChan)
+	if err != nil {
+		return err
+	}
+
+	f := dockerFilter.NewArgs()
+	f.Add("type", "container")
+	options := dockerTypesEvents.ListOptions{
+		Filters: f,
+	}
+	eventsChan, errChan := d.Client.Events(ctx, options)
+
+	const (
+		initialBackoff       = 2 * time.Second
+		backoffFactor        = 2
+		errorRetryMaxBackoff = 60 * time.Second
+	)
+	errorRetryBackoff := initialBackoff
 
 	for {
 		select {
 		case <-d.t.Dying():
 			d.logger.Infof("stopping container watcher")
 			return nil
-		case <-ticker.C:
-			// to track for garbage collection
-			runningContainersID := make(map[string]bool)
 
-			runningContainers, err := d.Client.ContainerList(ctx, dockerContainer.ListOptions{})
-			if err != nil {
-				if strings.Contains(strings.ToLower(err.Error()), "cannot connect to the docker daemon at") {
-					for idx, container := range d.runningContainerState {
-						if d.runningContainerState[idx].t.Alive() {
-							d.logger.Infof("killing tail for container %s", container.Name)
-							d.runningContainerState[idx].t.Kill(nil)
-
-							if err := d.runningContainerState[idx].t.Wait(); err != nil {
-								d.logger.Infof("error while waiting for death of %s : %s", container.Name, err)
-							}
-						}
-
-						delete(d.runningContainerState, idx)
-					}
-				} else {
-					log.Errorf("container list err: %s", err)
+		case event := <-eventsChan:
+			if event.Action == dockerTypesEvents.ActionStart || event.Action == dockerTypesEvents.ActionDie {
+				if err := d.checkContainers(ctx, monitChan, deleteChan); err != nil {
+					d.logger.Warnf("Failed to check containers: %v", err)
 				}
+			}
 
+		case err := <-errChan:
+			if err == nil {
 				continue
 			}
 
-			for _, container := range runningContainers {
-				runningContainersID[container.ID] = true
+			d.logger.Errorf("Docker events error: %v", err)
 
-				// don't need to re eval an already monitored container
-				if _, ok := d.runningContainerState[container.ID]; ok {
-					continue
-				}
+			// Multiple retry attempts within the error case
+			retries := 0
 
-				if containerConfig := d.EvalContainer(ctx, container); containerConfig != nil {
-					monitChan <- containerConfig
+			d.logger.Info("Attempting to reconnect to Docker events stream")
+			for {
+				// Check for cancellation before sleeping
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case <-d.t.Dying():
+					return nil
+				default:
+					// Continue with reconnection attempt
 				}
-			}
 
-			for containerStateID, containerConfig := range d.runningContainerState {
-				if _, ok := runningContainersID[containerStateID]; !ok {
-					deleteChan <- containerConfig
+				// Sleep with backoff
+				d.logger.Debugf("Retry %d: Waiting %s before reconnecting to Docker events",
+					retries+1, errorRetryBackoff)
+
+				select {
+				case <-time.After(errorRetryBackoff):
+					// Continue after backoff
+				case <-ctx.Done():
+					return ctx.Err()
+				case <-d.t.Dying():
+					return nil
 				}
-			}
 
-			d.logger.Tracef("Reading logs from %d containers", len(d.runningContainerState))
+				// Try to reconnect
+				newEventsChan, newErrChan := d.Client.Events(ctx, options)
 
-			ticker.Reset(d.CheckIntervalDuration)
+				// Check if connection is immediately broken
+				select {
+				case err := <-newErrChan:
+					// Connection failed immediately
+					retries++
+					errorRetryBackoff *= backoffFactor
+					if errorRetryBackoff > errorRetryMaxBackoff {
+						errorRetryBackoff = errorRetryMaxBackoff
+					}
+					d.logger.Errorf("Failed to reconnect to Docker events (attempt %d): %v",
+						retries, err)
+
+				default:
+					// No immediate error, seems to have reconnected successfully
+					errorRetryBackoff = initialBackoff
+					eventsChan = newEventsChan
+					errChan = newErrChan
+					goto reconnected
+				}
+			}
+		reconnected:
+			d.logger.Info("Successfully reconnected to Docker events")
+			// We check containers after a reconnection because the docker daemon might have restarted
+			// and the container tombs may have self deleted
+			if err := d.checkContainers(ctx, monitChan, deleteChan); err != nil {
+				d.logger.Warnf("Failed to check containers: %v", err)
+			}
+			// Continue in the main loop with new channels
+			continue
 		}
 	}
 }

pkg/acquisition/modules/docker/docker_test.go

@@ -13,6 +13,7 @@ import (
 
 	dockerTypes "github.com/docker/docker/api/types"
 	dockerContainer "github.com/docker/docker/api/types/container"
+	dockerTypesEvents "github.com/docker/docker/api/types/events"
 	"github.com/docker/docker/client"
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -50,6 +51,15 @@ source: docker`,
 			config: `
 mode: cat
 source: docker
+container_name:
+ - toto`,
+			expectedErr: "",
+		},
+		{
+			config: `
+mode: cat
+source: docker
+check_interval: 10s
 container_name:
  - toto`,
 			expectedErr: "",
@@ -273,6 +283,14 @@ func (cli *mockDockerCli) ContainerInspect(ctx context.Context, c string) (docke
 	return r, nil
 }
 
+// Since we are mocking the docker client, we return channels that will never be used
+func (cli *mockDockerCli) Events(ctx context.Context, options dockerTypesEvents.ListOptions) (<-chan dockerTypesEvents.Message, <-chan error) {
+	eventsChan := make(chan dockerTypesEvents.Message)
+	errChan := make(chan error)
+
+	return eventsChan, errChan
+}
+
 func TestOneShot(t *testing.T) {
 	ctx := t.Context()