Fix duplicate logs from docker containers (#6201)

* Fix duplicate logs from docker containers

* changelog

CHANGELOG.md

@@ -36,6 +36,8 @@ Main (unreleased)
 - Fix a duplicate metrics registration panic when sending metrics to an static
   mode metric instance's write handler. (@tpaschalis)
 
+- Fix issue causing duplicate logs when a docker target is restarted. (@captncraig)
+
 ### Other changes
 
 - Removed support for Windows 2012 in line with Microsoft end of life. (@mattdurham)

component/loki/source/docker/internal/dockertarget/target.go

@@ -219,6 +219,7 @@ func (t *Target) process(r io.Reader, logStreamLset model.LabelSet) {
 		// labels (e.g. duplicated and relabeled), but this shouldn't be the
 		// case anyway.
 		t.positions.Put(positions.CursorKey(t.containerName), t.labelsStr, ts.Unix())
+		t.since = ts.Unix()
 	}
 }
 

component/loki/source/docker/internal/dockertarget/target_test.go

@@ -31,7 +31,13 @@ func TestDockerTarget(t *testing.T) {
 	h := func(w http.ResponseWriter, r *http.Request) {
 		switch path := r.URL.Path; {
 		case strings.HasSuffix(path, "/logs"):
-			dat, err := os.ReadFile("testdata/flog.log")
+			var filePath string
+			if strings.Contains(r.URL.RawQuery, "since=0") {
+				filePath = "testdata/flog.log"
+			} else {
+				filePath = "testdata/flog_after_restart.log"
+			}
+			dat, err := os.ReadFile(filePath)
 			require.NoError(t, err)
 			_, err = w.Write(dat)
 			require.NoError(t, err)
@@ -97,4 +103,28 @@ func TestDockerTarget(t *testing.T) {
 		actualLines = append(actualLines, entry.Line)
 	}
 	require.ElementsMatch(t, actualLines, expectedLines)
+
+	// restart target to simulate container restart
+	tgt.StartIfNotRunning()
+	entryHandler.Clear()
+	require.Eventually(t, func() bool {
+		return len(entryHandler.Received()) >= 5
+	}, 5*time.Second, 100*time.Millisecond)
+
+	receivedAfterRestart := entryHandler.Received()
+	sort.Slice(receivedAfterRestart, func(i, j int) bool {
+		return receivedAfterRestart[i].Timestamp.Before(receivedAfterRestart[j].Timestamp)
+	})
+	actualLinesAfterRestart := make([]string, 0, 5)
+	for _, entry := range receivedAfterRestart[:5] {
+		actualLinesAfterRestart = append(actualLinesAfterRestart, entry.Line)
+	}
+	expectedLinesAfterRestart := []string{
+		"243.115.12.215 - - [09/Dec/2023:09:16:57 +0000] \"DELETE /morph/exploit/granular HTTP/1.0\" 500 26468",
+		"221.41.123.237 - - [09/Dec/2023:09:16:57 +0000] \"DELETE /user-centric/whiteboard HTTP/2.0\" 205 22487",
+		"89.111.144.144 - - [09/Dec/2023:09:16:57 +0000] \"DELETE /open-source/e-commerce HTTP/1.0\" 401 11092",
+		"62.180.191.187 - - [09/Dec/2023:09:16:57 +0000] \"DELETE /cultivate/integrate/technologies HTTP/2.0\" 302 12979",
+		"156.249.2.192 - - [09/Dec/2023:09:16:57 +0000] \"POST /revolutionize/mesh/metrics HTTP/2.0\" 401 5297",
+	}
+	require.ElementsMatch(t, actualLinesAfterRestart, expectedLinesAfterRestart)
 }