diff --git a/pkg/compose/logs.go b/pkg/compose/logs.go index 5bacaf76be..2836dea152 100644 --- a/pkg/compose/logs.go +++ b/pkg/compose/logs.go @@ -19,6 +19,8 @@ package compose import ( "context" "io" + "sync" + "time" "github.com/containerd/errdefs" "github.com/moby/moby/api/pkg/stdcopy" @@ -110,19 +112,29 @@ func (s *composeService) logContainer(ctx context.Context, consumer api.LogConsu // while following, ignoring those whose logging driver doesn't support // reading logs func (s *composeService) followStartedContainersLogs(ctx context.Context, eg *errgroup.Group, consumer api.LogConsumer, options api.LogOptions) api.ContainerEventListener { + runEnds := newRunEndTracker() return func(event api.ContainerEvent) { + runEnds.Observe(event) if event.Type != api.ContainerEventStarted { return } + // Captured synchronously: the monitor delivers events in order, so + // the recorded end cannot yet include THIS run's own exit — reading + // it inside the goroutine below could (fast run), and the window + // would drop the whole run. + since := runEnds.Since(event.ID) eg.Go(func() error { res, err := s.apiClient().ContainerInspect(ctx, event.ID, client.ContainerInspectOptions{}) if err != nil { return err } + if since == "" { + since = logsSinceLastRun(res.Container) + } err = s.doLogContainer(ctx, consumer, event.Source, res.Container, api.LogOptions{ Follow: options.Follow, - Since: res.Container.State.StartedAt, + Since: since, Until: options.Until, Tail: options.Tail, Timestamps: options.Timestamps, @@ -136,6 +148,67 @@ func (s *composeService) followStartedContainersLogs(ctx context.Context, eg *er } } +// runEndTracker remembers, per container, when the session last saw it exit — +// the re-attach anchor that stays correct even when the NEW run is already +// over: the event stream is ordered, so at start-event time the recorded +// value is necessarily the PREVIOUS run's end. The inspected FinishedAt +// (logsSinceLastRun) cannot give that guarantee — by the time we inspect, a +// fast run's own FinishedAt has overwritten it and the window would exclude +// everything the run printed. +type runEndTracker struct { + mu sync.Mutex + ends map[string]int64 // container ID → TimeNano of the last observed exit +} + +func newRunEndTracker() *runEndTracker { + return &runEndTracker{ends: map[string]int64{}} +} + +// Observe records exit events (other event types are ignored). +func (t *runEndTracker) Observe(e api.ContainerEvent) { + if e.Type != api.ContainerEventExited || e.Time == 0 { + return + } + t.mu.Lock() + t.ends[e.ID] = e.Time + t.mu.Unlock() +} + +// Since returns the log-window anchor for a container being re-attached: the +// recorded end of its previous run in RFC3339Nano — the same format the +// FinishedAt fallback feeds the logs API — or "" when the session never saw +// it exit (first start). +func (t *runEndTracker) Since(containerID string) string { + t.mu.Lock() + nano, ok := t.ends[containerID] + t.mu.Unlock() + if !ok { + return "" + } + return time.Unix(0, nano).UTC().Format(time.RFC3339Nano) +} + +// logsSinceLastRun returns the FALLBACK log window anchor for a container +// (re)started while we follow the project, used when the session has not +// observed a previous exit (runEndTracker): the previous run's FinishedAt. +// The new run's StartedAt looks like the natural anchor but loses output — +// the daemon starts copying stdout before it records StartedAt, so a fast +// process can get its first lines timestamped just before it, and +// `since=StartedAt` then drops them forever. Nothing can be logged between +// the previous run's end and the new run's start, so FinishedAt captures the +// entire new run without replaying the previous one — UNLESS the new run +// already finished by inspection time (its own FinishedAt shadows the +// previous run's), which is exactly what the tracker protects against. A +// container with no previous run has a zero FinishedAt, which means "no +// lower bound" — equally exact for a fresh container. +func logsSinceLastRun(ctr container.InspectResponse) string { + finished := ctr.State.FinishedAt + if t, err := time.Parse(time.RFC3339Nano, finished); err != nil || t.Unix() <= 0 { + return "" + } + return finished +} + func (s *composeService) doLogContainer(ctx context.Context, consumer api.LogConsumer, name string, ctr container.InspectResponse, options api.LogOptions) error { r, err := s.apiClient().ContainerLogs(ctx, ctr.ID, client.ContainerLogsOptions{ ShowStdout: true, diff --git a/pkg/compose/logs_test.go b/pkg/compose/logs_test.go index b0499f1560..27a9b859ae 100644 --- a/pkg/compose/logs_test.go +++ b/pkg/compose/logs_test.go @@ -245,3 +245,34 @@ func (l *testLogConsumer) LogsForContainer(containerName string) []string { defer l.mu.Unlock() return l.logs[containerName] } + +// TestRunEndTrackerAnchorsOnPreviousRun pins the re-attach anchor against the +// fast-run race seen in CI: with events delivered in order, the anchor +// captured at start-event time is the PREVIOUS run's end — even when the new +// run exits (and is observed) before the log stream is actually opened. +func TestRunEndTrackerAnchorsOnPreviousRun(t *testing.T) { + tr := newRunEndTracker() + + // First start: no previous exit observed → no anchor (caller falls back + // to the inspected FinishedAt). + assert.Equal(t, tr.Since("c1"), "") + + // Run N exits at t=1_000_000_001ns, run N+1 starts: the anchor captured + // at start-event time is run N's end, nanosecond-precise. + tr.Observe(compose.ContainerEvent{Type: compose.ContainerEventExited, ID: "c1", Time: 1_000_000_001}) + anchor := tr.Since("c1") + assert.Equal(t, anchor, "1970-01-01T00:00:01.000000001Z") + + // Run N+1 is fast: its own exit is observed before the log stream opens. + // The anchor captured above must NOT move — reading it after this point + // would exclude everything run N+1 printed. + tr.Observe(compose.ContainerEvent{Type: compose.ContainerEventExited, ID: "c1", Time: 2_000_000_002}) + assert.Equal(t, anchor, "1970-01-01T00:00:01.000000001Z") + // The NEXT start anchors on run N+1's end. + assert.Equal(t, tr.Since("c1"), "1970-01-01T00:00:02.000000002Z") + + // Non-exit events and other containers do not pollute the anchor. + tr.Observe(compose.ContainerEvent{Type: compose.ContainerEventStarted, ID: "c1", Time: 9_000_000_000}) + tr.Observe(compose.ContainerEvent{Type: compose.ContainerEventExited, ID: "c2", Time: 3_000_000_003}) + assert.Equal(t, tr.Since("c1"), "1970-01-01T00:00:02.000000002Z") +} diff --git a/pkg/compose/up.go b/pkg/compose/up.go index 0beb363735..2c5e87869c 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -338,26 +338,35 @@ func (u *upSession) captureExitCodeFrom() api.ContainerEventListener { // followStartedContainers streams logs of containers (re)started after `up`, // so they are followed like the initially attached ones. func (u *upSession) followStartedContainers(attached []string) api.ContainerEventListener { + runEnds := newRunEndTracker() return func(event api.ContainerEvent) { + runEnds.Observe(event) if !shouldFollowStartEvent(event, attached, u.options.Start.AttachTo) { return } + // Captured synchronously — see followStartedContainersLogs: read any + // later, a fast run's own exit could already be recorded and the log + // window would drop the whole run. + since := runEnds.Since(event.ID) u.eg.Go(func() error { - u.appendErr(u.streamContainerLogs(event)) + u.appendErr(u.streamContainerLogs(event, since)) return nil }) } } -func (u *upSession) streamContainerLogs(event api.ContainerEvent) error { +func (u *upSession) streamContainerLogs(event api.ContainerEvent, since string) error { res, err := u.apiClient().ContainerInspect(u.globalCtx, event.ID, client.ContainerInspectOptions{}) if err != nil { return err } + if since == "" { + since = logsSinceLastRun(res.Container) + } err = u.doLogContainer(u.globalCtx, u.options.Start.Attach, event.Source, res.Container, api.LogOptions{ Follow: true, - Since: res.Container.State.StartedAt, + Since: since, }) if errdefs.IsNotImplemented(err) { // container may be configured with logging_driver: none diff --git a/pkg/e2e/assert.go b/pkg/e2e/assert.go index a5b973dd15..c71f7d07b3 100644 --- a/pkg/e2e/assert.go +++ b/pkg/e2e/assert.go @@ -18,26 +18,62 @@ package e2e import ( "encoding/json" + "fmt" "strings" "testing" + "time" - "gotest.tools/v3/assert" - is "gotest.tools/v3/assert/cmp" + "gotest.tools/v3/poll" ) -// RequireServiceState ensures that the container is in the expected state -// (running or exited). +// RequireServiceState ensures that the container reaches the expected state +// (running or exited). The daemon reports state transitions asynchronously +// from everything else a test can observe (a container whose logs already +// flowed may still be listed under its previous state for a moment), so the +// check polls `compose ps` until the state converges instead of asserting on +// a single snapshot. func RequireServiceState(t testing.TB, cli *CLI, service string, state string) { t.Helper() - psRes := cli.RunDockerComposeCmd(t, "ps", "--all", "--format=json", service) - var serviceState map[string]any - assert.NilError(t, json.Unmarshal([]byte(psRes.Stdout()), &serviceState), - "Invalid `compose ps` JSON: command output: %s", - psRes.Combined()) - - assert.Assert(t, is.Equal(service, serviceState["Service"]), "Found ps output for unexpected service") - assert.Assert(t, is.Equal(strings.ToLower(state), strings.ToLower(serviceState["State"].(string))), - "Service %q (%s) not in expected state", - service, serviceState["Name"], - ) + poll.WaitOn(t, func(poll.LogT) poll.Result { + // NoCheck: a non-zero `compose ps` is a transient state here (the + // project may not be registered yet) — and the asserting variant + // would t.FailNow() from the poll goroutine, which terminates it via + // runtime.Goexit without reporting: the poll would hang until its + // opaque timeout instead of surfacing the actual failure below. + psRes := cli.RunDockerComposeCmdNoCheck(t, "ps", "--all", "--format=json", service) + if psRes.ExitCode != 0 { + return poll.Continue("`compose ps %s` exited %d: %s", service, psRes.ExitCode, psRes.Combined()) + } + out := strings.TrimSpace(psRes.Stdout()) + if out == "" { + // The container is not registered yet (creation in progress): + // transient, keep polling. + return poll.Continue("service %q has no `compose ps` entry yet", service) + } + // --format=json emits one JSON object per line, and a service can + // briefly list two containers mid-transition (the old one being + // removed, its replacement being created). Succeed as soon as one + // entry of the target service reaches the expected state; everything + // short of malformed JSON is a transient condition to retry, not a + // hard failure — hard-failing on those is the exact race this helper + // exists to absorb. + var seen []string + for line := range strings.SplitSeq(out, "\n") { + var entry map[string]any + if err := json.Unmarshal([]byte(line), &entry); err != nil { + return poll.Error(fmt.Errorf("invalid `compose ps` JSON: %w: command output: %s", err, psRes.Combined())) + } + if name, _ := entry["Service"].(string); name != service { + // ps was invoked filtered on the service name; a foreign or + // incomplete entry is transient noise. + continue + } + current, _ := entry["State"].(string) + if strings.EqualFold(state, current) { + return poll.Success() + } + seen = append(seen, current) + } + return poll.Continue("service %q is %v, expected %q", service, seen, state) + }, poll.WithTimeout(15*time.Second), poll.WithDelay(200*time.Millisecond)) } diff --git a/pkg/e2e/compose_test.go b/pkg/e2e/compose_test.go index 59d6cdd481..69e92b0427 100644 --- a/pkg/e2e/compose_test.go +++ b/pkg/e2e/compose_test.go @@ -157,7 +157,19 @@ func TestAttachRestart(t *testing.T) { debug) }, 4*time.Minute, 2*time.Second) - assert.Equal(t, strings.Count(res.Stdout(), "failing-1 | world"), 3, res.Combined()) + // The exit notice comes from the events monitor while the log line comes + // from the logs stream compose re-attaches after each restart: two + // asynchronous channels, so the third "world" may land shortly after the + // third exit notice — wait for it rather than asserting a snapshot. + // On failure, dump the daemon's own view of the container log: it + // discriminates a line compose failed to relay (present below, absent + // above) from a line the daemon itself never captured. + c.WaitForCondition(t, func() (bool, string) { + daemonView := icmd.RunCmd(c.NewDockerCmd(t, "logs", "attach-restart-failing-1")).Combined() + return strings.Count(res.Stdout(), + "failing-1 | world") == 3, fmt.Sprintf("'failing-1 | world' not found 3 times in : \n%s\ndaemon log view:\n%s\n", + res.Combined(), daemonView) + }, time.Minute, time.Second) } func TestInitContainer(t *testing.T) { diff --git a/pkg/e2e/compose_up_test.go b/pkg/e2e/compose_up_test.go index 66e8837a02..6ce8768468 100644 --- a/pkg/e2e/compose_up_test.go +++ b/pkg/e2e/compose_up_test.go @@ -52,7 +52,7 @@ func TestUpExitCodeFrom(t *testing.T) { func TestUpExitCodeFromContainerKilled(t *testing.T) { NewScenario(t, "up --exit-code-from must report 143 for a service stopped by the abort"). Step("the watched long-lived service is stopped when another exits", - ComposeCmd("up", "--menu=false", "--exit-code-from=test").MayFail().Within(60*time.Second), + ComposeCmd("up", "--menu=false", "--exit-code-from=test").MayFail().Within(120*time.Second), ExitCode(143)) }