(refactoring) introduce monitor to manage containers events and application termination

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

Author: ndeloof

cmd/formatter/logs.go

@@ -56,10 +56,6 @@ func NewLogConsumer(ctx context.Context, stdout, stderr io.Writer, color, prefix
 	}
 }
 
-func (l *logConsumer) Register(name string) {
-	l.register(name)
-}
-
 func (l *logConsumer) register(name string) *presenter {
 	var p *presenter
 	root, _, found := strings.Cut(name, " ")

pkg/api/api.go

@@ -647,24 +647,25 @@ type LogConsumer interface {
 	Log(containerName, message string)
 	Err(containerName, message string)
 	Status(container, msg string)
-	Register(container string)
 }
 
 // ContainerEventListener is a callback to process ContainerEvent from services
 type ContainerEventListener func(event ContainerEvent)
 
 // ContainerEvent notify an event has been collected on source container implementing Service
 type ContainerEvent struct {
-	Type int
-	// Container is the name of the container _without the project prefix_.
+	Type      int
+	Time      int64
+	Container *ContainerSummary
+	// Source is the name of the container _without the project prefix_.
 	//
 	// This is only suitable for display purposes within Compose, as it's
 	// not guaranteed to be unique across services.
-	Container string
-	ID        string
-	Service   string
-	Line      string
-	// ContainerEventExit only
+	Source  string
+	ID      string
+	Service string
+	Line    string
+	// ContainerEventExited only
 	ExitCode   int
 	Restarting bool
 }
@@ -674,17 +675,19 @@ const (
 	ContainerEventLog = iota
 	// ContainerEventErr is a ContainerEvent of type log on stderr. Line is set
 	ContainerEventErr
-	// ContainerEventAttach is a ContainerEvent of type attach. First event sent about a container
-	ContainerEventAttach
+	// ContainerEventStarted let consumer know a container has been started
+	ContainerEventStarted
+	// ContainerEventRestarted let consumer know a container has been restarted
+	ContainerEventRestarted
 	// ContainerEventStopped is a ContainerEvent of type stopped.
 	ContainerEventStopped
+	// ContainerEventCreated let consumer know a new container has been created
+	ContainerEventCreated
 	// ContainerEventRecreated let consumer know container stopped but his being replaced
 	ContainerEventRecreated
-	// ContainerEventExit is a ContainerEvent of type exit. ExitCode is set
-	ContainerEventExit
+	// ContainerEventExited is a ContainerEvent of type exit. ExitCode is set
+	ContainerEventExited
 	// UserCancel user cancelled compose up, we are stopping containers
-	UserCancel
-	// HookEventLog is a ContainerEvent of type log on stdout by service hook
 	HookEventLog
 )
 

pkg/compose/attach.go

@@ -61,41 +61,37 @@ func (s *composeService) attach(ctx context.Context, project *types.Project, lis
 }
 
 func (s *composeService) attachContainer(ctx context.Context, container containerType.Summary, listener api.ContainerEventListener) error {
-	serviceName := container.Labels[api.ServiceLabel]
-	containerName := getContainerNameWithoutProject(container)
-
-	listener(api.ContainerEvent{
-		Type:      api.ContainerEventAttach,
-		Container: containerName,
-		ID:        container.ID,
-		Service:   serviceName,
-	})
+	service := container.Labels[api.ServiceLabel]
+	name := getContainerNameWithoutProject(container)
+	return s.doAttachContainer(ctx, service, container.ID, name, listener)
+}
+
+func (s *composeService) doAttachContainer(ctx context.Context, service, id, name string, listener api.ContainerEventListener) error {
+	inspect, err := s.apiClient().ContainerInspect(ctx, id)
+	if err != nil {
+		return err
+	}
 
 	wOut := utils.GetWriter(func(line string) {
 		listener(api.ContainerEvent{
-			Type:      api.ContainerEventLog,
-			Container: containerName,
-			ID:        container.ID,
-			Service:   serviceName,
-			Line:      line,
+			Type:    api.ContainerEventLog,
+			Source:  name,
+			ID:      id,
+			Service: service,
+			Line:    line,
 		})
 	})
 	wErr := utils.GetWriter(func(line string) {
 		listener(api.ContainerEvent{
-			Type:      api.ContainerEventErr,
-			Container: containerName,
-			ID:        container.ID,
-			Service:   serviceName,
-			Line:      line,
+			Type:    api.ContainerEventErr,
+			Source:  name,
+			ID:      id,
+			Service: service,
+			Line:    line,
 		})
 	})
 
-	inspect, err := s.apiClient().ContainerInspect(ctx, container.ID)
-	if err != nil {
-		return err
-	}
-
-	_, _, err = s.attachContainerStreams(ctx, container.ID, inspect.Config.Tty, nil, wOut, wErr)
+	_, _, err = s.attachContainerStreams(ctx, id, inspect.Config.Tty, nil, wOut, wErr)
 	return err
 }
 

pkg/compose/containers.go

@@ -128,12 +128,6 @@ func isService(services ...string) containerPredicate {
 	}
 }
 
-func isRunning() containerPredicate {
-	return func(c container.Summary) bool {
-		return c.State == "running"
-	}
-}
-
 // isOrphaned is a predicate to select containers without a matching service definition in compose project
 func isOrphaned(project *types.Project) containerPredicate {
 	services := append(project.ServiceNames(), project.DisabledServiceNames()...)

pkg/compose/hook.go

@@ -32,11 +32,11 @@ import (
 func (s composeService) runHook(ctx context.Context, ctr container.Summary, service types.ServiceConfig, hook types.ServiceHook, listener api.ContainerEventListener) error {
 	wOut := utils.GetWriter(func(line string) {
 		listener(api.ContainerEvent{
-			Type:      api.HookEventLog,
-			Container: getContainerNameWithoutProject(ctr) + " ->",
-			ID:        ctr.ID,
-			Service:   service.Name,
-			Line:      line,
+			Type:    api.HookEventLog,
+			Source:  getContainerNameWithoutProject(ctr) + " ->",
+			ID:      ctr.ID,
+			Service: service.Name,
+			Line:    line,
 		})
 	})
 	defer wOut.Close() //nolint:errcheck

pkg/compose/logs.go

@@ -62,7 +62,7 @@ func (s *composeService) Logs(
 	eg, ctx := errgroup.WithContext(ctx)
 	for _, ctr := range containers {
 		eg.Go(func() error {
-			err := s.logContainers(ctx, consumer, ctr, options)
+			err := s.logContainer(ctx, consumer, ctr, options)
 			if errdefs.IsNotImplemented(err) {
 				logrus.Warnf("Can't retrieve logs for %q: %s", getCanonicalContainerName(ctr), err.Error())
 				return nil
@@ -72,34 +72,21 @@ func (s *composeService) Logs(
 	}
 
 	if options.Follow {
-		containers = containers.filter(isRunning())
 		printer := newLogPrinter(consumer)
-		eg.Go(func() error {
-			_, err := printer.Run(api.CascadeIgnore, "", nil)
-			return err
-		})
-
-		for _, c := range containers {
-			printer.HandleEvent(api.ContainerEvent{
-				Type:      api.ContainerEventAttach,
-				Container: getContainerNameWithoutProject(c),
-				ID:        c.ID,
-				Service:   c.Labels[api.ServiceLabel],
-			})
-		}
+		eg.Go(printer.Run)
 
-		eg.Go(func() error {
-			err := s.watchContainers(ctx, projectName, options.Services, nil, printer.HandleEvent, containers, func(c container.Summary, t time.Time) error {
-				printer.HandleEvent(api.ContainerEvent{
-					Type:      api.ContainerEventAttach,
-					Container: getContainerNameWithoutProject(c),
-					ID:        c.ID,
-					Service:   c.Labels[api.ServiceLabel],
-				})
+		monitor := newMonitor(s.apiClient(), options.Project)
+		monitor.withListener(func(event api.ContainerEvent) {
+			if event.Type == api.ContainerEventStarted {
 				eg.Go(func() error {
-					err := s.logContainers(ctx, consumer, c, api.LogOptions{
+					ctr, err := s.apiClient().ContainerInspect(ctx, event.ID)
+					if err != nil {
+						return err
+					}
+
+					err = s.doLogContainer(ctx, consumer, event.Source, ctr, api.LogOptions{
 						Follow:     options.Follow,
-						Since:      t.Format(time.RFC3339Nano),
+						Since:      time.Unix(0, event.Time).Format(time.RFC3339Nano),
 						Until:      options.Until,
 						Tail:       options.Tail,
 						Timestamps: options.Timestamps,
@@ -110,31 +97,28 @@ func (s *composeService) Logs(
 					}
 					return err
 				})
-				return nil
-			}, func(c container.Summary, t time.Time) error {
-				printer.HandleEvent(api.ContainerEvent{
-					Type:      api.ContainerEventAttach,
-					Container: "", // actual name will be set by start event
-					ID:        c.ID,
-					Service:   c.Labels[api.ServiceLabel],
-				})
-				return nil
-			})
-			printer.Stop()
-			return err
+			}
+		})
+		eg.Go(func() error {
+			defer printer.Stop()
+			return monitor.Start(ctx)
 		})
 	}
 
 	return eg.Wait()
 }
 
-func (s *composeService) logContainers(ctx context.Context, consumer api.LogConsumer, c container.Summary, options api.LogOptions) error {
-	cnt, err := s.apiClient().ContainerInspect(ctx, c.ID)
+func (s *composeService) logContainer(ctx context.Context, consumer api.LogConsumer, c container.Summary, options api.LogOptions) error {
+	ctr, err := s.apiClient().ContainerInspect(ctx, c.ID)
 	if err != nil {
 		return err
 	}
+	name := getContainerNameWithoutProject(c)
+	return s.doLogContainer(ctx, consumer, name, ctr, options)
+}
 
-	r, err := s.apiClient().ContainerLogs(ctx, cnt.ID, container.LogsOptions{
+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, container.LogsOptions{
 		ShowStdout: true,
 		ShowStderr: true,
 		Follow:     options.Follow,
@@ -148,11 +132,10 @@ func (s *composeService) logContainers(ctx context.Context, consumer api.LogCons
 	}
 	defer r.Close() //nolint:errcheck
 
-	name := getContainerNameWithoutProject(c)
 	w := utils.GetWriter(func(line string) {
 		consumer.Log(name, line)
 	})
-	if cnt.Config.Tty {
+	if ctr.Config.Tty {
 		_, err = io.Copy(w, r)
 	} else {
 		_, err = stdcopy.StdCopy(w, w, r)

pkg/compose/logs_test.go

@@ -189,8 +189,6 @@ func (l *testLogConsumer) Err(containerName, message string) {
 
 func (l *testLogConsumer) Status(containerName, msg string) {}
 
-func (l *testLogConsumer) Register(containerName string) {}
-
 func (l *testLogConsumer) LogsForContainer(containerName string) []string {
 	l.mu.Lock()
 	defer l.mu.Unlock()