use logs API with Since to collect the very first logs after restart

Signed-off-by: Nicolas De Loof <[email protected]>

Author: ndeloof

pkg/api/api.go

@@ -658,7 +658,7 @@ type ContainerEvent struct {
 	ID      string
 	Service string
 	Line    string
-	// ContainerEventExited only
+	// ExitCode is only set on ContainerEventExited events
 	ExitCode   int
 	Restarting bool
 }

pkg/compose/convergence.go

@@ -640,23 +640,23 @@ func (s *composeService) recreateContainer(ctx context.Context, project *types.P
 		UseNetworkAliases: true,
 		Labels:            mergeLabels(service.Labels, service.CustomLabels).Add(api.ContainerReplaceLabel, replaced.ID),
 	}
-	created, err = s.createMobyContainer(ctx, project, service, tmpName, number, inherited, opts, w)
+	timeoutInSecond := utils.DurationSecondToInt(timeout)
+	err = s.apiClient().ContainerStop(ctx, replaced.ID, containerType.StopOptions{Timeout: timeoutInSecond})
 	if err != nil {
 		return created, err
 	}
 
-	timeoutInSecond := utils.DurationSecondToInt(timeout)
-	err = s.apiClient().ContainerStop(ctx, replaced.ID, containerType.StopOptions{Timeout: timeoutInSecond})
+	err = s.apiClient().ContainerRename(ctx, replaced.ID, tmpName)
 	if err != nil {
 		return created, err
 	}
 
-	err = s.apiClient().ContainerRemove(ctx, replaced.ID, containerType.RemoveOptions{})
+	created, err = s.createMobyContainer(ctx, project, service, name, number, inherited, opts, w)
 	if err != nil {
 		return created, err
 	}
 
-	err = s.apiClient().ContainerRename(ctx, created.ID, name)
+	err = s.apiClient().ContainerRemove(ctx, replaced.ID, containerType.RemoveOptions{})
 	if err != nil {
 		return created, err
 	}

pkg/compose/logs.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"io"
-	"time"
 
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/errdefs"
@@ -75,9 +74,10 @@ func (s *composeService) Logs(
 
 	if options.Follow {
 		printer := newLogPrinter(consumer)
-		eg.Go(printer.Run)
 
-		monitor := newMonitor(s.apiClient(), options.Project)
+		monitor := newMonitor(s.apiClient(), projectName)
+		monitor.withServices(options.Services)
+		monitor.withListener(printer.HandleEvent)
 		monitor.withListener(func(event api.ContainerEvent) {
 			if event.Type == api.ContainerEventStarted {
 				eg.Go(func() error {
@@ -88,7 +88,7 @@ func (s *composeService) Logs(
 
 					err = s.doLogContainer(ctx, consumer, event.Source, ctr, api.LogOptions{
 						Follow:     options.Follow,
-						Since:      time.Unix(0, event.Time).Format(time.RFC3339Nano),
+						Since:      ctr.State.StartedAt,
 						Until:      options.Until,
 						Tail:       options.Tail,
 						Timestamps: options.Timestamps,
@@ -103,7 +103,6 @@ func (s *composeService) Logs(
 			}
 		})
 		eg.Go(func() error {
-			defer printer.Stop()
 			return monitor.Start(ctx)
 		})
 	}

pkg/compose/monitor.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"strconv"
 
-	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/containerd/errdefs"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/events"
@@ -34,23 +33,23 @@ import (
 
 type monitor struct {
 	api     client.APIClient
-	project *types.Project
+	project string
 	// services tells us which service to consider and those we can ignore, maybe ran by a concurrent compose command
 	services  map[string]bool
 	listeners []api.ContainerEventListener
 }
 
-func newMonitor(api client.APIClient, project *types.Project) *monitor {
-	services := map[string]bool{}
-	if project != nil {
-		for name := range project.Services {
-			services[name] = true
-		}
-	}
+func newMonitor(api client.APIClient, project string) *monitor {
 	return &monitor{
 		api:      api,
 		project:  project,
-		services: services,
+		services: map[string]bool{},
+	}
+}
+
+func (c *monitor) withServices(services []string) {
+	for _, name := range services {
+		c.services[name] = true
 	}
 }
 
@@ -62,7 +61,7 @@ func (c *monitor) Start(ctx context.Context) error {
 	initialState, err := c.api.ContainerList(ctx, container.ListOptions{
 		All: true,
 		Filters: filters.NewArgs(
-			projectFilter(c.project.Name),
+			projectFilter(c.project),
 			oneOffFilter(false),
 			hasConfigHashLabel(),
 		),
@@ -84,16 +83,19 @@ func (c *monitor) Start(ctx context.Context) error {
 	evtCh, errCh := c.api.Events(ctx, events.ListOptions{
 		Filters: filters.NewArgs(
 			filters.Arg("type", "container"),
-			projectFilter(c.project.Name)),
+			projectFilter(c.project)),
 	})
 	for {
+		if len(containers) == 0 {
+			return nil
+		}
 		select {
 		case <-ctx.Done():
 			return nil
 		case err := <-errCh:
 			return err
 		case event := <-evtCh:
-			if !c.services[event.Actor.Attributes[api.ServiceLabel]] {
+			if len(c.services) > 0 && !c.services[event.Actor.Attributes[api.ServiceLabel]] {
 				continue
 			}
 			ctr, err := c.getContainerSummary(event)
@@ -103,24 +105,35 @@ func (c *monitor) Start(ctx context.Context) error {
 
 			switch event.Action {
 			case events.ActionCreate:
-				containers.Add(ctr.ID)
+				if len(c.services) == 0 || c.services[ctr.Labels[api.ServiceLabel]] {
+					containers.Add(ctr.ID)
+				}
+				evtType := api.ContainerEventCreated
+				if _, ok := ctr.Labels[api.ContainerReplaceLabel]; ok {
+					evtType = api.ContainerEventRecreated
+				}
 				for _, listener := range c.listeners {
-					listener(newContainerEvent(event.TimeNano, ctr, api.ContainerEventCreated))
+					listener(newContainerEvent(event.TimeNano, ctr, evtType))
 				}
 				logrus.Debugf("container %s created", ctr.Name)
 			case events.ActionStart:
 				restarted := restarting.Has(ctr.ID)
-				for _, listener := range c.listeners {
-					listener(newContainerEvent(event.TimeNano, ctr, api.ContainerEventStarted, func(e *api.ContainerEvent) {
-						e.Restarting = restarted
-					}))
-				}
 				if restarted {
 					logrus.Debugf("container %s restarted", ctr.Name)
+					for _, listener := range c.listeners {
+						listener(newContainerEvent(event.TimeNano, ctr, api.ContainerEventStarted, func(e *api.ContainerEvent) {
+							e.Restarting = restarted
+						}))
+					}
 				} else {
 					logrus.Debugf("container %s started", ctr.Name)
+					for _, listener := range c.listeners {
+						listener(newContainerEvent(event.TimeNano, ctr, api.ContainerEventStarted))
+					}
+				}
+				if len(c.services) == 0 || c.services[ctr.Labels[api.ServiceLabel]] {
+					containers.Add(ctr.ID)
 				}
-				containers.Add(ctr.ID)
 			case events.ActionRestart:
 				for _, listener := range c.listeners {
 					listener(newContainerEvent(event.TimeNano, ctr, api.ContainerEventRestarted))
@@ -159,9 +172,6 @@ func (c *monitor) Start(ctx context.Context) error {
 				}
 			}
 		}
-		if len(containers) == 0 {
-			return nil
-		}
 	}
 }
 
@@ -192,7 +202,7 @@ func (c *monitor) getContainerSummary(event events.Message) (*api.ContainerSumma
 	ctr := &api.ContainerSummary{
 		ID:      event.Actor.ID,
 		Name:    event.Actor.Attributes["name"],
-		Project: c.project.Name,
+		Project: c.project,
 		Service: event.Actor.Attributes[api.ServiceLabel],
 		Labels:  event.Actor.Attributes, // More than just labels, but that'c the closest the API gives us
 	}

pkg/compose/printer.go

@@ -18,80 +18,36 @@ package compose
 
 import (
 	"fmt"
-	"sync"
 
 	"github.com/docker/compose/v2/pkg/api"
 )
 
 // logPrinter watch application containers and collect their logs
 type logPrinter interface {
 	HandleEvent(event api.ContainerEvent)
-	Run() error
-	Stop()
 }
 
 type printer struct {
-	queue    chan api.ContainerEvent
 	consumer api.LogConsumer
-	stopCh   chan struct{} // stopCh is a signal channel for producers to stop sending events to the queue
-	stop     sync.Once
 }
 
 // newLogPrinter builds a LogPrinter passing containers logs to LogConsumer
 func newLogPrinter(consumer api.LogConsumer) logPrinter {
 	printer := printer{
 		consumer: consumer,
-		queue:    make(chan api.ContainerEvent),
-		stopCh:   make(chan struct{}),
-		stop:     sync.Once{},
 	}
 	return &printer
 }
 
-func (p *printer) Stop() {
-	p.stop.Do(func() {
-		close(p.stopCh)
-		for {
-			select {
-			case <-p.queue:
-				// purge the queue to free producers goroutines
-				// p.queue will be garbage collected
-			default:
-				return
-			}
-		}
-	})
-}
-
 func (p *printer) HandleEvent(event api.ContainerEvent) {
-	select {
-	case <-p.stopCh:
-		return
-	default:
-		p.queue <- event
-	}
-}
-
-func (p *printer) Run() error {
-	defer p.Stop()
-
-	// containers we are tracking. Use true when container is running, false after we receive a stop|die signal
-	for {
-		select {
-		case <-p.stopCh:
-			return nil
-		case event := <-p.queue:
-			switch event.Type {
-			case api.ContainerEventExited, api.ContainerEventStopped, api.ContainerEventRecreated, api.ContainerEventRestarted:
-				p.consumer.Status(event.Source, fmt.Sprintf("exited with code %d", event.ExitCode))
-				if event.Type == api.ContainerEventRecreated {
-					p.consumer.Status(event.Source, "has been recreated")
-				}
-			case api.ContainerEventLog, api.HookEventLog:
-				p.consumer.Log(event.Source, event.Line)
-			case api.ContainerEventErr:
-				p.consumer.Err(event.Source, event.Line)
-			}
-		}
+	switch event.Type {
+	case api.ContainerEventExited:
+		p.consumer.Status(event.Source, fmt.Sprintf("exited with code %d", event.ExitCode))
+	case api.ContainerEventRecreated:
+		p.consumer.Status(event.Source, "has been recreated")
+	case api.ContainerEventLog, api.HookEventLog:
+		p.consumer.Log(event.Source, event.Line)
+	case api.ContainerEventErr:
+		p.consumer.Err(event.Source, event.Line)
 	}
 }