refactor: one handler per container event in monitor.Start, add tests

monitor.Start (cognitive complexity 77) inlined the handling of all four container actions in the event loop. Each action moves to an onContainerX handler, with the recurring bits named: watched() (is this service's container ours to track), notify() (broadcast to listeners), initialContainers() (seed the tracking set). The loop drops to complexity 19 and reads as: seed, subscribe, dispatch until no containers remain. The tracking sets stay explicit parameters so data flow remains visible. The monitor had no unit test despite driving up/logs termination; the refactor is locked by a new suite covering the full lifecycle (created/recreated/started/restarted/exited, default-name trimming, exit codes), the restart detection through both engine states (Restarting, and Running per moby/moby#45538), the already-removed container on die, service filtering, the no-container fast path, events stream errors, context cancellation, and exit-code parse errors. No behavior change: same events, same ordering, same termination conditions. Co-Authored-By: Claude Fable 5 <[email protected]> Signed-off-by: Nicolas De Loof <[email protected]>
2026-10-12 02:13:08 +00:00 · 2026-08-18 13:37:48 +02:00
parent e964454893
commit ef54fcaa41
2 changed files with 326 additions and 76 deletions
@@ -52,30 +52,13 @@ func (c *monitor) withServices(services []string) {
 }

 // Start runs monitor to detect application events and return after termination
-//
-// FIXME(ndeloof) complete migration to gocognit
-//
-//nolint:gocognit
 func (c *monitor) Start(ctx context.Context) error {
-	// collect initial application container
-	initialState, err := c.apiClient.ContainerList(ctx, client.ContainerListOptions{
-		All: true,
-		Filters: projectFilter(c.project).Add("label",
-			oneOffFilter(false),
-			api.ConfigHashLabel,
-		),
-	})
+	// containers is the set of container IDs the application is based on
+	containers, err := c.initialContainers(ctx)
 	if err != nil {
 		return err
 	}
-
-	// containers is the set if container IDs the application is based on
-	containers := utils.Set[string]{}
-	for _, ctr := range initialState.Items {
-		if len(c.services) == 0 || c.services[ctr.Labels[api.ServiceLabel]] {
-			containers.Add(ctr.ID)
-		}
-	}
+	// restarting tracks containers which exited but are configured to restart on exit
 	restarting := utils.Set[string]{}

 	res := c.apiClient.Events(ctx, client.EventsListOptions{
@@ -91,7 +74,7 @@ func (c *monitor) Start(ctx context.Context) error {
 		case err := <-res.Err:
 			return err
 		case event := <-res.Messages:
-			if len(c.services) > 0 && !c.services[event.Actor.Attributes[api.ServiceLabel]] {
+			if !c.watched(event.Actor.Attributes[api.ServiceLabel]) {
 				continue
 			}
 			ctr, err := c.getContainerSummary(event)
@@ -101,71 +84,114 @@ func (c *monitor) Start(ctx context.Context) error {

 			switch event.Action {
 			case events.ActionCreate:
-				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, evtType))
-				}
-				logrus.Debugf("container %s created", ctr.Name)
+				c.onContainerCreate(event, ctr, containers)
 			case events.ActionStart:
-				restarted := restarting.Has(ctr.ID)
-				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)
-				}
+				c.onContainerStart(event, ctr, containers, restarting)
 			case events.ActionRestart:
-				for _, listener := range c.listeners {
-					listener(newContainerEvent(event.TimeNano, ctr, api.ContainerEventRestarted))
-				}
-				logrus.Debugf("container %s restarted", ctr.Name)
+				c.onContainerRestart(event, ctr)
 			case events.ActionDie:
-				logrus.Debugf("container %s exited with code %d", ctr.Name, ctr.ExitCode)
-				inspect, err := c.apiClient.ContainerInspect(ctx, event.Actor.ID, client.ContainerInspectOptions{})
-				if errdefs.IsNotFound(err) {
-					// Source is already removed
-				} else if err != nil {
+				err := c.onContainerDie(ctx, event, ctr, containers, restarting)
+				if err != nil {
 					return err
 				}
-
-				if inspect.Container.State != nil && (inspect.Container.State.Restarting || inspect.Container.State.Running) {
-					// State.Restarting is set by engine when container is configured to restart on exit
-					// on ContainerRestart it doesn't (see https://github.com/moby/moby/issues/45538)
-					// container state still is reported as "running"
-					logrus.Debugf("container %s is restarting", ctr.Name)
-					restarting.Add(ctr.ID)
-					for _, listener := range c.listeners {
-						listener(newContainerEvent(event.TimeNano, ctr, api.ContainerEventExited, func(e *api.ContainerEvent) {
-							e.Restarting = true
-						}))
-					}
-				} else {
-					for _, listener := range c.listeners {
-						listener(newContainerEvent(event.TimeNano, ctr, api.ContainerEventExited))
-					}
-					containers.Remove(ctr.ID)
-				}
 			}
 		}
 	}
 }

+// initialContainers collects the application's containers at startup,
+// restricted to the services this monitor watches
+func (c *monitor) initialContainers(ctx context.Context) (utils.Set[string], error) {
+	initialState, err := c.apiClient.ContainerList(ctx, client.ContainerListOptions{
+		All: true,
+		Filters: projectFilter(c.project).Add("label",
+			oneOffFilter(false),
+			api.ConfigHashLabel,
+		),
+	})
+	if err != nil {
+		return nil, err
+	}
+	containers := utils.Set[string]{}
+	for _, ctr := range initialState.Items {
+		if c.watched(ctr.Labels[api.ServiceLabel]) {
+			containers.Add(ctr.ID)
+		}
+	}
+	return containers, nil
+}
+
+// watched tells whether a service's containers are watched by this monitor.
+// An empty service set means "the whole application".
+func (c *monitor) watched(service string) bool {
+	return len(c.services) == 0 || c.services[service]
+}
+
+// notify broadcasts a container event to the registered listeners
+func (c *monitor) notify(event api.ContainerEvent) {
+	for _, listener := range c.listeners {
+		listener(event)
+	}
+}
+
+func (c *monitor) onContainerCreate(event events.Message, ctr *api.ContainerSummary, containers utils.Set[string]) {
+	if c.watched(ctr.Labels[api.ServiceLabel]) {
+		containers.Add(ctr.ID)
+	}
+	evtType := api.ContainerEventCreated
+	if _, ok := ctr.Labels[api.ContainerReplaceLabel]; ok {
+		evtType = api.ContainerEventRecreated
+	}
+	c.notify(newContainerEvent(event.TimeNano, ctr, evtType))
+	logrus.Debugf("container %s created", ctr.Name)
+}
+
+func (c *monitor) onContainerStart(event events.Message, ctr *api.ContainerSummary, containers, restarting utils.Set[string]) {
+	if restarting.Has(ctr.ID) {
+		logrus.Debugf("container %s restarted", ctr.Name)
+		c.notify(newContainerEvent(event.TimeNano, ctr, api.ContainerEventStarted, func(e *api.ContainerEvent) {
+			e.Restarting = true
+		}))
+	} else {
+		logrus.Debugf("container %s started", ctr.Name)
+		c.notify(newContainerEvent(event.TimeNano, ctr, api.ContainerEventStarted))
+	}
+	if c.watched(ctr.Labels[api.ServiceLabel]) {
+		containers.Add(ctr.ID)
+	}
+}
+
+func (c *monitor) onContainerRestart(event events.Message, ctr *api.ContainerSummary) {
+	c.notify(newContainerEvent(event.TimeNano, ctr, api.ContainerEventRestarted))
+	logrus.Debugf("container %s restarted", ctr.Name)
+}
+
+func (c *monitor) onContainerDie(ctx context.Context, event events.Message, ctr *api.ContainerSummary, containers, restarting utils.Set[string]) error {
+	logrus.Debugf("container %s exited with code %d", ctr.Name, ctr.ExitCode)
+	inspect, err := c.apiClient.ContainerInspect(ctx, event.Actor.ID, client.ContainerInspectOptions{})
+	if errdefs.IsNotFound(err) {
+		// Source is already removed
+	} else if err != nil {
+		return err
+	}
+
+	if inspect.Container.State != nil && (inspect.Container.State.Restarting || inspect.Container.State.Running) {
+		// State.Restarting is set by engine when container is configured to restart on exit
+		// on ContainerRestart it doesn't (see https://github.com/moby/moby/issues/45538)
+		// container state still is reported as "running"
+		logrus.Debugf("container %s is restarting", ctr.Name)
+		restarting.Add(ctr.ID)
+		c.notify(newContainerEvent(event.TimeNano, ctr, api.ContainerEventExited, func(e *api.ContainerEvent) {
+			e.Restarting = true
+		}))
+		return nil
+	}
+
+	c.notify(newContainerEvent(event.TimeNano, ctr, api.ContainerEventExited))
+	containers.Remove(ctr.ID)
+	return nil
+}
+
 func newContainerEvent(timeNano int64, ctr *api.ContainerSummary, eventType int, opts ...func(e *api.ContainerEvent)) api.ContainerEvent {
 	name := ctr.Name
 	defaultName := getDefaultContainerName(ctr.Project, ctr.Labels[api.ServiceLabel], ctr.Labels[api.ContainerNumberLabel])
@@ -0,0 +1,224 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package compose
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/containerd/errdefs"
+	"github.com/google/go-cmp/cmp"
+	"github.com/moby/moby/api/types/container"
+	"github.com/moby/moby/api/types/events"
+	"github.com/moby/moby/client"
+	"go.uber.org/mock/gomock"
+	"gotest.tools/v3/assert"
+
+	"github.com/docker/compose/v5/pkg/api"
+	"github.com/docker/compose/v5/pkg/mocks"
+)
+
+// recordedEvent is the projection of api.ContainerEvent the monitor tests
+// assert on
+type recordedEvent struct {
+	eventType  int
+	id         string
+	source     string
+	restarting bool
+	exitCode   int
+}
+
+var cmpRecordedEvents = cmp.AllowUnexported(recordedEvent{})
+
+func recordEvents(m *monitor) *[]recordedEvent {
+	var got []recordedEvent
+	m.withListener(func(e api.ContainerEvent) {
+		got = append(got, recordedEvent{
+			eventType:  e.Type,
+			id:         e.ID,
+			source:     e.Source,
+			restarting: e.Restarting,
+			exitCode:   e.ExitCode,
+		})
+	})
+	return &got
+}
+
+func containerMessage(action events.Action, id, name, service string, extra map[string]string) events.Message {
+	attributes := map[string]string{
+		"name":                   name,
+		api.ServiceLabel:         service,
+		api.ContainerNumberLabel: "1",
+	}
+	for k, v := range extra {
+		attributes[k] = v
+	}
+	return events.Message{
+		Action: action,
+		Actor:  events.Actor{ID: id, Attributes: attributes},
+	}
+}
+
+func inspectResult(running, restarting bool) client.ContainerInspectResult {
+	return client.ContainerInspectResult{
+		Container: container.InspectResponse{
+			State: &container.State{Running: running, Restarting: restarting},
+		},
+	}
+}
+
+func expectEventStream(apiClient *mocks.MockAPIClient, initial []container.Summary, capacity int) (chan events.Message, chan error) {
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{Items: initial}, nil)
+	messages := make(chan events.Message, capacity)
+	errs := make(chan error, 1)
+	apiClient.EXPECT().Events(gomock.Any(), gomock.Any()).
+		Return(client.EventsResult{Messages: messages, Err: errs})
+	return messages, errs
+}
+
+func TestMonitorStartLifecycle(t *testing.T) {
+	apiClient := mocks.NewMockAPIClient(gomock.NewController(t))
+	m := newMonitor(apiClient, "p")
+	got := recordEvents(m)
+
+	messages, _ := expectEventStream(apiClient, []container.Summary{
+		{ID: "c1", Labels: map[string]string{api.ServiceLabel: "db"}},
+	}, 10)
+
+	gomock.InOrder(
+		// c2 exits but is configured to restart on exit
+		apiClient.EXPECT().ContainerInspect(gomock.Any(), "c2", gomock.Any()).Return(inspectResult(false, true), nil),
+		// c2 exits and restarts again, but this time the engine reports state
+		// "running" instead of "restarting" (see moby/moby#45538)
+		apiClient.EXPECT().ContainerInspect(gomock.Any(), "c2", gomock.Any()).Return(inspectResult(true, false), nil),
+		// c3 is already removed when we inspect it
+		apiClient.EXPECT().ContainerInspect(gomock.Any(), "c3", gomock.Any()).Return(client.ContainerInspectResult{}, errdefs.ErrNotFound),
+		apiClient.EXPECT().ContainerInspect(gomock.Any(), "c2", gomock.Any()).Return(inspectResult(false, false), nil),
+		apiClient.EXPECT().ContainerInspect(gomock.Any(), "c1", gomock.Any()).Return(inspectResult(false, false), nil),
+	)
+
+	// the monitor trims the default "<project>-<service>-<number>" name to "<service>-<number>"
+	defaultName := getDefaultContainerName("p", "web", "1")
+	messages <- containerMessage(events.ActionCreate, "c2", defaultName, "web", nil)
+	messages <- containerMessage(events.ActionCreate, "c3", "custom-name", "job", map[string]string{api.ContainerReplaceLabel: "old"})
+	messages <- containerMessage(events.ActionRestart, "c2", defaultName, "web", nil)
+	messages <- containerMessage(events.ActionDie, "c2", defaultName, "web", map[string]string{"exitCode": "1"})
+	messages <- containerMessage(events.ActionStart, "c2", defaultName, "web", nil)
+	messages <- containerMessage(events.ActionDie, "c2", defaultName, "web", map[string]string{"exitCode": "1"})
+	messages <- containerMessage(events.ActionStart, "c2", defaultName, "web", nil)
+	messages <- containerMessage(events.ActionDie, "c3", "custom-name", "job", map[string]string{"exitCode": "0"})
+	messages <- containerMessage(events.ActionDie, "c2", defaultName, "web", map[string]string{"exitCode": "1"})
+	messages <- containerMessage(events.ActionDie, "c1", "c1-name", "db", map[string]string{"exitCode": "0"})
+
+	err := m.Start(t.Context())
+	assert.NilError(t, err)
+
+	assert.DeepEqual(t, *got, []recordedEvent{
+		{eventType: api.ContainerEventCreated, id: "c2", source: "web-1"},
+		{eventType: api.ContainerEventRecreated, id: "c3", source: "custom-name"},
+		{eventType: api.ContainerEventRestarted, id: "c2", source: "web-1"},
+		{eventType: api.ContainerEventExited, id: "c2", source: "web-1", restarting: true, exitCode: 1},
+		{eventType: api.ContainerEventStarted, id: "c2", source: "web-1", restarting: true},
+		{eventType: api.ContainerEventExited, id: "c2", source: "web-1", restarting: true, exitCode: 1},
+		{eventType: api.ContainerEventStarted, id: "c2", source: "web-1", restarting: true},
+		{eventType: api.ContainerEventExited, id: "c3", source: "custom-name"},
+		{eventType: api.ContainerEventExited, id: "c2", source: "web-1", exitCode: 1},
+		{eventType: api.ContainerEventExited, id: "c1", source: "c1-name"},
+	}, cmpRecordedEvents)
+}
+
+func TestMonitorStartServiceFilter(t *testing.T) {
+	apiClient := mocks.NewMockAPIClient(gomock.NewController(t))
+	m := newMonitor(apiClient, "p")
+	m.withServices([]string{"web"})
+	got := recordEvents(m)
+
+	// the db container is not watched, so it doesn't count towards termination
+	messages, _ := expectEventStream(apiClient, []container.Summary{
+		{ID: "c1", Labels: map[string]string{api.ServiceLabel: "web"}},
+		{ID: "c2", Labels: map[string]string{api.ServiceLabel: "db"}},
+	}, 10)
+
+	// no ContainerInspect expectation for c2: its event must be ignored
+	apiClient.EXPECT().ContainerInspect(gomock.Any(), "c1", gomock.Any()).Return(inspectResult(false, false), nil)
+
+	messages <- containerMessage(events.ActionDie, "c2", "c2-name", "db", map[string]string{"exitCode": "1"})
+	messages <- containerMessage(events.ActionDie, "c1", "c1-name", "web", map[string]string{"exitCode": "0"})
+
+	err := m.Start(t.Context())
+	assert.NilError(t, err)
+
+	assert.DeepEqual(t, *got, []recordedEvent{
+		{eventType: api.ContainerEventExited, id: "c1", source: "c1-name"},
+	}, cmpRecordedEvents)
+}
+
+func TestMonitorStartNoContainers(t *testing.T) {
+	apiClient := mocks.NewMockAPIClient(gomock.NewController(t))
+	m := newMonitor(apiClient, "p")
+
+	expectEventStream(apiClient, nil, 1)
+
+	err := m.Start(t.Context())
+	assert.NilError(t, err)
+}
+
+func TestMonitorStartEventsError(t *testing.T) {
+	apiClient := mocks.NewMockAPIClient(gomock.NewController(t))
+	m := newMonitor(apiClient, "p")
+
+	_, errs := expectEventStream(apiClient, []container.Summary{
+		{ID: "c1", Labels: map[string]string{api.ServiceLabel: "db"}},
+	}, 1)
+
+	sentinel := errors.New("events stream failed")
+	errs <- sentinel
+
+	err := m.Start(t.Context())
+	assert.ErrorIs(t, err, sentinel)
+}
+
+func TestMonitorStartContextCancelled(t *testing.T) {
+	apiClient := mocks.NewMockAPIClient(gomock.NewController(t))
+	m := newMonitor(apiClient, "p")
+
+	expectEventStream(apiClient, []container.Summary{
+		{ID: "c1", Labels: map[string]string{api.ServiceLabel: "db"}},
+	}, 1)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	err := m.Start(ctx)
+	assert.NilError(t, err)
+}
+
+func TestMonitorStartBadExitCode(t *testing.T) {
+	apiClient := mocks.NewMockAPIClient(gomock.NewController(t))
+	m := newMonitor(apiClient, "p")
+
+	messages, _ := expectEventStream(apiClient, []container.Summary{
+		{ID: "c1", Labels: map[string]string{api.ServiceLabel: "db"}},
+	}, 1)
+
+	messages <- containerMessage(events.ActionDie, "c1", "c1-name", "db", map[string]string{"exitCode": "not-a-number"})
+
+	err := m.Start(t.Context())
+	assert.ErrorContains(t, err, "not-a-number")
+}