docker: read from now if container restarts

pkg/acquisition/modules/docker/config.go

@@ -125,6 +125,10 @@ func (d *Source) UnmarshalConfig(yamlConfig []byte) error {
 		d.compiledServiceID = append(d.compiledServiceID, compiled)
 	}
 
+	if d.Config.Since != "" && d.Config.Mode == configuration.TAIL_MODE && d.logger != nil {
+		d.logger.Warn("since is ignored in tail mode: containers and services are read from the moment they are discovered, to avoid re-reading logs when they restart")
+	}
+
 	if d.Config.Since == "" {
 		d.Config.Since = time.Now().UTC().Format(time.RFC3339)
 	}

pkg/acquisition/modules/docker/docker_test.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/crowdsecurity/go-cs-lib/cstest"
 
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/docker/tracker"
 	"github.com/crowdsecurity/crowdsec/pkg/metrics"
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
@@ -685,3 +686,63 @@ func TestParseLabelsNestedCollisionDoesNotPanic(t *testing.T) {
 		assert.Equal(t, first, parseLabels(labels))
 	}
 }
+
+// A restarted container (or service) keeps the logs of its previous runs: tailing it again
+// from the configured "since" would replay everything we already sent to the parsers.
+func TestCheckTailsFromNow(t *testing.T) {
+	ctx := t.Context()
+
+	tests := []struct {
+		name   string
+		config string
+		check  func(*Source, context.Context, chan *ContainerConfig, chan *ContainerConfig) error
+	}{
+		{
+			name: "container",
+			config: fmt.Sprintf(`
+source: docker
+container_name:
+ - %s
+since: 2020-01-01T00:00:00Z`, testContainerName),
+			check: func(d *Source, ctx context.Context, monitChan chan *ContainerConfig, deleteChan chan *ContainerConfig) error {
+				return d.checkContainers(ctx, monitChan, deleteChan)
+			},
+		},
+		{
+			name: "service",
+			config: fmt.Sprintf(`
+source: docker
+service_name:
+ - %s
+since: 2020-01-01T00:00:00Z`, testServiceName),
+			check: func(d *Source, ctx context.Context, monitChan chan *ContainerConfig, deleteChan chan *ContainerConfig) error {
+				return d.checkServices(ctx, monitChan, deleteChan)
+			},
+		},
+	}
+
+	for _, ts := range tests {
+		t.Run(ts.name, func(t *testing.T) {
+			dockerSource := &Source{logger: log.WithField("type", ModuleName)}
+
+			require.NoError(t, dockerSource.UnmarshalConfig([]byte(ts.config)))
+
+			dockerSource.runningContainerState = tracker.NewTracker[*ContainerConfig]()
+			dockerSource.runningServiceState = tracker.NewTracker[*ContainerConfig]()
+			dockerSource.Client = &mockDockerCli{}
+
+			before := time.Now().UTC()
+
+			monitChan := make(chan *ContainerConfig, 1)
+			deleteChan := make(chan *ContainerConfig, 1)
+
+			require.NoError(t, ts.check(dockerSource, ctx, monitChan, deleteChan))
+
+			cfg := <-monitChan
+
+			since, err := time.Parse(time.RFC3339Nano, cfg.logOptions.Since)
+			require.NoError(t, err)
+			assert.False(t, since.Before(before), "expected to tail from now, got %s", cfg.logOptions.Since)
+		})
+	}
+}

pkg/acquisition/modules/docker/run.go

@@ -10,13 +10,13 @@ import (
 	"time"
 
 	backoff "github.com/cenkalti/backoff/v5"
+	"github.com/containerd/errdefs"
 	dockerContainer "github.com/moby/moby/api/types/container"
 	dockerTypesEvents "github.com/moby/moby/api/types/events"
 	dockerTypesSwarm "github.com/moby/moby/api/types/swarm"
 	"github.com/moby/moby/client"
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
-	"github.com/containerd/errdefs"
 	"gopkg.in/tomb.v2"
 
 	"github.com/crowdsecurity/dlog"
@@ -28,15 +28,15 @@ import (
 type BackOffFactory func() backoff.BackOff
 
 func newDockerBackOffFactory() BackOffFactory {
-    return func() backoff.BackOff {
-        exp := backoff.NewExponentialBackOff()
-        exp.InitialInterval = 2 * time.Second
-        exp.Multiplier = 2.5
-        exp.MaxInterval = 2 * time.Minute
-        exp.RandomizationFactor = 0.5
-
-        return exp
-    }
+	return func() backoff.BackOff {
+		exp := backoff.NewExponentialBackOff()
+		exp.InitialInterval = 2 * time.Second
+		exp.Multiplier = 2.5
+		exp.MaxInterval = 2 * time.Minute
+		exp.RandomizationFactor = 0.5
+
+		return exp
+	}
 }
 
 // OneShotAcquisition reads a set of file and returns when done
@@ -278,6 +278,9 @@ func (d *Source) checkServices(ctx context.Context, monitChan chan *ContainerCon
 		}
 
 		if serviceConfig := d.EvalService(ctx, service); serviceConfig != nil {
+			// same as containers: the logs of the previous tasks are still there
+			serviceConfig.logOptions.Since = time.Now().UTC().Format(time.RFC3339Nano)
+
 			monitChan <- serviceConfig
 		}
 	}
@@ -329,6 +332,10 @@ func (d *Source) checkContainers(ctx context.Context, monitChan chan *ContainerC
 		}
 
 		if containerConfig := d.EvalContainer(ctx, container); containerConfig != nil {
+			// A restarted container keeps the logs of its previous runs, which we already
+			// read: always tail from now on, whatever the configured "since".
+			containerConfig.logOptions.Since = time.Now().UTC().Format(time.RFC3339Nano)
+
 			monitChan <- containerConfig
 		}
 	}
@@ -345,8 +352,8 @@ func (d *Source) checkContainers(ctx context.Context, monitChan chan *ContainerC
 }
 
 type subscription struct {
-    events <-chan dockerTypesEvents.Message
-    errs   <-chan error
+	events <-chan dockerTypesEvents.Message
+	errs   <-chan error
 }
 
 func (d *Source) trySubscribeEvents(ctx context.Context) (*subscription, error) {

pkg/acquisition/schemas/docker.yaml

@@ -70,6 +70,8 @@ properties:
     format: date-time
     description: >
       RFC3339 lower-bound timestamp; defaults to the current UTC time if omitted.
+      Only used in cat mode: in tail mode, containers and services are read from
+      the moment they are discovered.
   until:
     type: string
     format: date-time