loki.source.docker: filter out duplicate container IDs

Signed-off-by: Paschalis Tsilias <[email protected]>

component/loki/source/docker/docker.go

@@ -215,13 +215,18 @@ func (c *Component) Update(args component.Arguments) error {
 
 	// Convert input targets into targets to give to tailer.
 	targets := make([]*dt.Target, 0, len(newArgs.Targets))
+	seenTargets := make(map[string]struct{}, len(newArgs.Targets))
 
 	for _, target := range newArgs.Targets {
 		containerID, ok := target[dockerLabelContainerID]
 		if !ok {
 			level.Debug(c.opts.Logger).Log("msg", "docker target did not include container ID label:"+dockerLabelContainerID)
 			continue
 		}
+		if _, seen := seenTargets[containerID]; seen {
+			continue
+		}
+		seenTargets[containerID] = struct{}{}
 
 		var labels = make(model.LabelSet)
 		for k, v := range target {

component/loki/source/docker/docker_test.go

@@ -5,9 +5,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/grafana/agent/component"
 	"github.com/grafana/agent/pkg/flow/componenttest"
 	"github.com/grafana/agent/pkg/util"
 	"github.com/grafana/river"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 )
 
@@ -33,3 +35,30 @@ func Test(t *testing.T) {
 
 	require.NoError(t, ctrl.WaitRunning(time.Minute))
 }
+
+func TestDuplicateTargets(t *testing.T) {
+	// Use host that works on all platforms (including Windows).
+	var cfg = `
+		host       = "tcp://127.0.0.1:9375"
+		targets    = [
+			{__meta_docker_container_id = "foo", __meta_docker_port_private = "8080"},
+			{__meta_docker_container_id = "foo", __meta_docker_port_private = "8080"},
+			{__meta_docker_container_id = "bar", __meta_docker_port_private = "8080"},
+		]
+		forward_to = []
+	`
+
+	var args Arguments
+	err := river.Unmarshal([]byte(cfg), &args)
+	require.NoError(t, err)
+
+	cmp, err := New(component.Options{
+		ID:         "loki.source.docker.test",
+		Logger:     util.TestFlowLogger(t),
+		Registerer: prometheus.NewRegistry(),
+		DataPath:   t.TempDir(),
+	}, args)
+	require.NoError(t, err)
+
+	require.Len(t, cmp.manager.tasks, 2)
+}