diff --git a/internal/component/loki/source/docker/tailer.go b/internal/component/loki/source/docker/tailer.go index 06e92ce5ad8..13bc4763694 100644 --- a/internal/component/loki/source/docker/tailer.go +++ b/internal/component/loki/source/docker/tailer.go @@ -177,11 +177,27 @@ func (t *tailer) stop() { t.wg.Wait() t.logger.Debug("stopped Docker target", "container", t.containerID) - // If the component is not stopping, then it means that the target for this component is gone and that - // we should clear the entry from the positions file. - if !t.componentStopping() { - t.positions.Remove(positions.CursorKey(t.containerID), t.labelsStr) - } + t.removePositionIfContainerDeleted() + } +} + +func (t *tailer) removePositionIfContainerDeleted() { + if t.componentStopping() || t.client == nil { + return + } + + _, err := t.client.ContainerInspect(context.Background(), t.containerID, client.ContainerInspectOptions{}) + switch { + case err == nil: + // The target may have disappeared because the container stopped. Keep its + // position so a later restart resumes where the tailer left off. + return + case cerrdefs.IsNotFound(err): + t.positions.Remove(positions.CursorKey(t.containerID), t.labelsStr) + default: + // Preserve the position when the container's state cannot be determined. + // Re-reading logs is more harmful than retaining a stale cursor. + t.logger.Warn("could not determine whether to remove Docker container position", "id", t.containerID, "error", err) } } diff --git a/internal/component/loki/source/docker/tailer_test.go b/internal/component/loki/source/docker/tailer_test.go index 5d63dcfb654..536bea7c6dd 100644 --- a/internal/component/loki/source/docker/tailer_test.go +++ b/internal/component/loki/source/docker/tailer_test.go @@ -9,6 +9,7 @@ import ( "context" "encoding/binary" "encoding/json" + "errors" "io" "net/http" "net/http/httptest" @@ -265,6 +266,66 @@ func TestTailerStopsWhenContainerNotFound(t *testing.T) { } } +func TestTailerPositionCleanup(t *testing.T) { + tests := []struct { + name string + componentStopping bool + inspectErr error + wantRemoved bool + }{ + { + name: "stopped container preserves position", + }, + { + name: "removed container deletes position", + inspectErr: cerrdefs.ErrNotFound, + wantRemoved: true, + }, + { + name: "inspect error preserves position", + inspectErr: errors.New("inspect failed"), + }, + { + name: "component shutdown preserves position", + componentStopping: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pos := &trackingPositions{Positions: positions.NewNop()} + mock := clientMock{ + running: func() bool { return false }, + finishedAt: func() string { return "0001-01-01T00:00:00Z" }, + inspectErr: func() error { return tt.inspectErr }, + } + tailer := &tailer{ + logger: logging.NewSlogNop(), + positions: pos, + containerID: "container-id", + labelsStr: "{}", + client: mock, + componentStopping: func() bool { return tt.componentStopping }, + running: true, + cancel: func() {}, + } + + tailer.stop() + + require.Equal(t, tt.wantRemoved, pos.removed) + }) + } +} + +type trackingPositions struct { + positions.Positions + removed bool +} + +func (p *trackingPositions) Remove(_, _ string) { + p.removed = true +} + var _ io.ReadCloser = (*stringReader)(nil) func newStringReader(s string) *stringReader {