From fd75494c0e58f8973189cfc725d5812ce51aca47 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Sun, 19 Apr 2026 21:30:46 +0200 Subject: [PATCH] Reconciliation: Executor Signed-off-by: Nicolas De Loof --- pkg/compose/executor.go | 432 +++++++++++++++++++++++++++++++++++ pkg/compose/executor_test.go | 127 ++++++++++ 2 files changed, 559 insertions(+) create mode 100644 pkg/compose/executor.go create mode 100644 pkg/compose/executor_test.go diff --git a/pkg/compose/executor.go b/pkg/compose/executor.go new file mode 100644 index 000000000..da43a969a --- /dev/null +++ b/pkg/compose/executor.go @@ -0,0 +1,432 @@ +/* + 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" + "fmt" + "sync" + + "github.com/compose-spec/compose-go/v2/types" + "github.com/moby/moby/client" + "golang.org/x/sync/errgroup" + + "github.com/docker/compose/v5/pkg/api" + "github.com/docker/compose/v5/pkg/utils" +) + +// planExecutor executes a reconciliation Plan by walking the DAG and performing +// each atomic operation via the Docker API. It carries no decision logic — all +// decisions were made by the reconciler when building the plan. +type planExecutor struct { + compose *composeService + project *types.Project + pctx *reconciliationContext +} + +// reconciliationContext holds results produced by completed nodes so that downstream +// nodes can reference them (e.g. a RenameContainer node needs the container ID +// created by a prior CreateContainer node). +type reconciliationContext struct { + mu sync.Mutex + results map[int]operationResult +} + +type operationResult struct { + ContainerID string + ContainerName string +} + +func (pc *reconciliationContext) set(nodeID int, r operationResult) { + pc.mu.Lock() + defer pc.mu.Unlock() + pc.results[nodeID] = r +} + +func (pc *reconciliationContext) get(nodeID int) operationResult { + pc.mu.Lock() + defer pc.mu.Unlock() + return pc.results[nodeID] +} + +// executePlan walks the plan DAG, executing nodes in parallel where possible +// while respecting dependency edges. It emits progress events and handles +// group-based event aggregation for composite operations like recreate. +func (s *composeService) executePlan(ctx context.Context, project *types.Project, plan *Plan) error { + if plan.IsEmpty() { + return nil + } + + exec := &planExecutor{ + compose: s, + project: project, + pctx: &reconciliationContext{results: map[int]operationResult{}}, + } + + // Build a done-channel per node so dependents can wait + done := make(map[int]chan struct{}, len(plan.Nodes)) + for _, node := range plan.Nodes { + done[node.ID] = make(chan struct{}) + } + + // Track group event state: first node emits Working, last emits Done + groups := exec.buildGroupTracker(plan) + + eg, ctx := errgroup.WithContext(ctx) + for _, node := range plan.Nodes { + eg.Go(func() error { + // Wait for all dependencies + for _, dep := range node.DependsOn { + select { + case <-done[dep.ID]: + case <-ctx.Done(): + return ctx.Err() + } + } + + // Emit group start event if this is the first node of a group + groups.onNodeStart(node, s.events) + + err := exec.executeNode(ctx, node) + + if err == nil { + // Emit group done event if this is the last node of a group + groups.onNodeDone(node, s.events) + } else if ctx.Err() == nil { + groups.onNodeError(node, s.events, err) + } + + close(done[node.ID]) + return err + }) + } + + return eg.Wait() +} + +// executeNode dispatches a single plan node to the appropriate API call. +func (exec *planExecutor) executeNode(ctx context.Context, node *PlanNode) error { + op := node.Operation + switch op.Type { + case OpCreateNetwork: + return exec.execCreateNetwork(ctx, op) + case OpRemoveNetwork: + return exec.execRemoveNetwork(ctx, op) + case OpDisconnectNetwork: + return exec.execDisconnectNetwork(ctx, op) + case OpConnectNetwork: + return exec.execConnectNetwork(ctx, op) + case OpCreateVolume: + return exec.execCreateVolume(ctx, op) + case OpRemoveVolume: + return exec.execRemoveVolume(ctx, op) + case OpCreateContainer: + return exec.execCreateContainer(ctx, node) + case OpStartContainer: + return exec.execStartContainer(ctx, op) + case OpStopContainer: + return exec.execStopContainer(ctx, op) + case OpRemoveContainer: + return exec.execRemoveContainer(ctx, op) + case OpRenameContainer: + return exec.execRenameContainer(ctx, node) + default: + return fmt.Errorf("unknown operation type: %s", op.Type) + } +} + +// --- Network operations --- + +func (exec *planExecutor) execCreateNetwork(ctx context.Context, op Operation) error { + _, err := exec.compose.ensureNetwork(ctx, exec.project, op.Name, op.Network) + return err +} + +func (exec *planExecutor) execRemoveNetwork(ctx context.Context, op Operation) error { + _, err := exec.compose.apiClient().NetworkRemove(ctx, op.Name, client.NetworkRemoveOptions{}) + return err +} + +func (exec *planExecutor) execDisconnectNetwork(ctx context.Context, op Operation) error { + _, err := exec.compose.apiClient().NetworkDisconnect(ctx, op.Name, client.NetworkDisconnectOptions{ + Container: op.Container.ID, + Force: true, + }) + return err +} + +func (exec *planExecutor) execConnectNetwork(ctx context.Context, op Operation) error { + _, err := exec.compose.apiClient().NetworkConnect(ctx, op.Name, client.NetworkConnectOptions{ + Container: op.Container.ID, + }) + return err +} + +// --- Volume operations --- + +func (exec *planExecutor) execCreateVolume(ctx context.Context, op Operation) error { + return exec.compose.createVolume(ctx, *op.Volume) +} + +func (exec *planExecutor) execRemoveVolume(ctx context.Context, op Operation) error { + _, err := exec.compose.apiClient().VolumeRemove(ctx, op.Name, client.VolumeRemoveOptions{Force: true}) + return err +} + +// --- Container operations --- + +func (exec *planExecutor) execCreateContainer(ctx context.Context, node *PlanNode) error { + op := node.Operation + labels := mergeLabels(op.Service.Labels, op.Service.CustomLabels) + if op.Container != nil { + // This is a recreate: add the replace label + replacedName := op.Service.ContainerName + if replacedName == "" { + replacedName = fmt.Sprintf("%s%s%d", op.Service.Name, api.Separator, op.Number) + } + labels = labels.Add(api.ContainerReplaceLabel, replacedName) + } + + opts := createOptions{ + AutoRemove: false, + AttachStdin: false, + UseNetworkAliases: true, + Labels: labels, + } + ctr, err := exec.compose.createMobyContainer(ctx, exec.project, *op.Service, op.Name, op.Number, op.Inherited, opts) + if err != nil { + return err + } + + exec.pctx.set(node.ID, operationResult{ + ContainerID: ctr.ID, + ContainerName: op.Name, + }) + return nil +} + +func (exec *planExecutor) execStartContainer(ctx context.Context, op Operation) error { + startMx.Lock() + defer startMx.Unlock() + _, err := exec.compose.apiClient().ContainerStart(ctx, op.Container.ID, client.ContainerStartOptions{}) + return err +} + +func (exec *planExecutor) execStopContainer(ctx context.Context, op Operation) error { + _, err := exec.compose.apiClient().ContainerStop(ctx, op.Container.ID, client.ContainerStopOptions{ + Timeout: utils.DurationSecondToInt(op.Timeout), + }) + return err +} + +func (exec *planExecutor) execRemoveContainer(ctx context.Context, op Operation) error { + _, err := exec.compose.apiClient().ContainerRemove(ctx, op.Container.ID, client.ContainerRemoveOptions{Force: true}) + return err +} + +func (exec *planExecutor) execRenameContainer(ctx context.Context, node *PlanNode) error { + // Find the CreateContainer node in our dependencies to get the created container ID + var createdID string + for _, dep := range node.DependsOn { + r := exec.pctx.get(dep.ID) + if r.ContainerID != "" { + createdID = r.ContainerID + break + } + // Also check transitive: the create node may not be a direct dep + // Walk up the group chain + } + // Fallback: walk all results in the group + if createdID == "" { + for _, n := range node.DependsOn { + createdID = exec.findCreatedIDInChain(n) + if createdID != "" { + break + } + } + } + if createdID == "" { + return fmt.Errorf("rename: could not find created container ID for node #%d", node.ID) + } + + _, err := exec.compose.apiClient().ContainerRename(ctx, createdID, client.ContainerRenameOptions{ + NewName: node.Operation.Name, + }) + return err +} + +// findCreatedIDInChain walks the dependency chain to find a CreateContainer result. +func (exec *planExecutor) findCreatedIDInChain(node *PlanNode) string { + r := exec.pctx.get(node.ID) + if r.ContainerID != "" { + return r.ContainerID + } + for _, dep := range node.DependsOn { + if id := exec.findCreatedIDInChain(dep); id != "" { + return id + } + } + return "" +} + +// --- Group event tracking --- + +// groupTracker manages event emission for grouped nodes (e.g. recreate). +// The first node starting emits Working, the last finishing emits Done. +type groupTracker struct { + mu sync.Mutex + groups map[string]*groupState +} + +type groupState struct { + eventName string // e.g. "Container myproject-web-1" + total int // total nodes in this group + started int // nodes that have started + done int // nodes that have completed +} + +func (exec *planExecutor) buildGroupTracker(plan *Plan) *groupTracker { + gt := &groupTracker{groups: map[string]*groupState{}} + for _, node := range plan.Nodes { + if node.Group == "" { + continue + } + if _, ok := gt.groups[node.Group]; !ok { + // Derive the event name from the container being recreated + eventName := node.Group // fallback + if node.Operation.Container != nil { + eventName = getContainerProgressName(*node.Operation.Container) + } + gt.groups[node.Group] = &groupState{eventName: eventName} + } + gt.groups[node.Group].total++ + } + return gt +} + +func (gt *groupTracker) onNodeStart(node *PlanNode, events api.EventProcessor) { + if node.Group == "" { + // Ungrouped: emit individual event + emitStartEvent(node, events) + return + } + gt.mu.Lock() + defer gt.mu.Unlock() + gs := gt.groups[node.Group] + gs.started++ + if gs.started == 1 { + events.On(newEvent(gs.eventName, api.Working, "Recreate")) + } +} + +func (gt *groupTracker) onNodeDone(node *PlanNode, events api.EventProcessor) { + if node.Group == "" { + emitDoneEvent(node, events) + return + } + gt.mu.Lock() + defer gt.mu.Unlock() + gs := gt.groups[node.Group] + gs.done++ + if gs.done == gs.total { + events.On(newEvent(gs.eventName, api.Done, "Recreated")) + } +} + +func (gt *groupTracker) onNodeError(node *PlanNode, events api.EventProcessor, err error) { + if node.Group == "" { + emitErrorEvent(node, events, err) + return + } + gt.mu.Lock() + defer gt.mu.Unlock() + gs := gt.groups[node.Group] + events.On(api.Resource{ + ID: gs.eventName, + Status: api.Error, + Text: err.Error(), + }) +} + +// emitStartEvent emits the appropriate Working event for an ungrouped node. +func emitStartEvent(node *PlanNode, events api.EventProcessor) { + op := node.Operation + switch op.Type { + case OpCreateContainer: + events.On(creatingEvent("Container " + op.Name)) + case OpStartContainer: + name := getContainerProgressName(*op.Container) + events.On(newEvent(name, api.Working, api.StatusStarting)) + case OpStopContainer: + events.On(stoppingEvent(getContainerProgressName(*op.Container))) + case OpRemoveContainer: + events.On(removingEvent(getContainerProgressName(*op.Container))) + case OpCreateNetwork: + events.On(creatingEvent("Network " + op.Name)) + case OpRemoveNetwork: + events.On(removingEvent("Network " + op.Name)) + case OpCreateVolume: + events.On(creatingEvent("Volume " + op.Name)) + case OpRemoveVolume: + events.On(removingEvent("Volume " + op.Name)) + } +} + +// emitDoneEvent emits the appropriate Done event for an ungrouped node. +func emitDoneEvent(node *PlanNode, events api.EventProcessor) { + op := node.Operation + switch op.Type { + case OpCreateContainer: + events.On(createdEvent("Container " + op.Name)) + case OpStartContainer: + name := getContainerProgressName(*op.Container) + events.On(newEvent(name, api.Done, api.StatusStarted)) + case OpStopContainer: + events.On(stoppedEvent(getContainerProgressName(*op.Container))) + case OpRemoveContainer: + events.On(removedEvent(getContainerProgressName(*op.Container))) + case OpCreateNetwork: + events.On(createdEvent("Network " + op.Name)) + case OpRemoveNetwork: + events.On(removedEvent("Network " + op.Name)) + case OpCreateVolume: + events.On(createdEvent("Volume " + op.Name)) + case OpRemoveVolume: + events.On(removedEvent("Volume " + op.Name)) + } +} + +// emitErrorEvent emits an error event for an ungrouped node. +func emitErrorEvent(node *PlanNode, events api.EventProcessor, err error) { + op := node.Operation + var id string + switch { + case op.Container != nil: + id = getContainerProgressName(*op.Container) + default: + id = op.ResourceID + } + events.On(api.Resource{ + ID: id, + Status: api.Error, + Text: err.Error(), + }) +} + +// getContainerProgressName returns the display name for a container in progress output. +// Re-exported from convergence.go for use by the executor. +var _ = getContainerProgressName // ensure it stays accessible diff --git a/pkg/compose/executor_test.go b/pkg/compose/executor_test.go new file mode 100644 index 000000000..21a3c783a --- /dev/null +++ b/pkg/compose/executor_test.go @@ -0,0 +1,127 @@ +/* + 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" + "testing" + + "github.com/compose-spec/compose-go/v2/types" + "github.com/moby/moby/api/types/container" + "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" +) + +// noopEventProcessor discards all events. +type noopEventProcessor struct{} + +func (noopEventProcessor) Start(_ context.Context, _ string) {} +func (noopEventProcessor) On(_ ...api.Resource) {} +func (noopEventProcessor) Done(_ string, _ bool) {} + +func newTestService(t *testing.T) (*composeService, *mocks.MockAPIClient) { + t.Helper() + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + return svc.(*composeService), apiClient +} + +func TestExecutePlanEmpty(t *testing.T) { + svc, _ := newTestService(t) + err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, &Plan{}) + assert.NilError(t, err) +} + +func TestExecutePlanCreateNetwork(t *testing.T) { + svc, apiClient := newTestService(t) + + nw := types.NetworkConfig{Name: "test_default"} + project := &types.Project{ + Name: "test", + Networks: types.Networks{"default": nw}, + } + + // ensureNetwork: inspect → not found, list → empty, create + apiClient.EXPECT().NetworkInspect(gomock.Any(), "test_default", gomock.Any()). + Return(client.NetworkInspectResult{}, notFoundError{}) + apiClient.EXPECT().NetworkList(gomock.Any(), gomock.Any()). + Return(client.NetworkListResult{}, nil) + apiClient.EXPECT().NetworkCreate(gomock.Any(), "test_default", gomock.Any()). + Return(client.NetworkCreateResult{ID: "net1"}, nil) + + plan := &Plan{} + plan.addNode(Operation{ + Type: OpCreateNetwork, + ResourceID: "network:default", + Cause: "not found", + Name: nw.Name, + Network: &nw, + }, "") + + err := svc.executePlan(t.Context(), project, plan) + assert.NilError(t, err) +} + +func TestExecutePlanStopRemoveContainer(t *testing.T) { + svc, apiClient := newTestService(t) + + ctr := container.Summary{ + ID: "c1", + Names: []string{"/test-web-1"}, + Labels: map[string]string{ + api.ServiceLabel: "web", + api.ContainerNumberLabel: "1", + }, + } + + apiClient.EXPECT().ContainerStop(gomock.Any(), "c1", gomock.Any()). + Return(client.ContainerStopResult{}, nil) + apiClient.EXPECT().ContainerRemove(gomock.Any(), "c1", gomock.Any()). + Return(client.ContainerRemoveResult{}, nil) + + plan := &Plan{} + stopNode := plan.addNode(Operation{ + Type: OpStopContainer, + ResourceID: "service:web:1", + Cause: "scale down", + Container: &ctr, + }, "") + plan.addNode(Operation{ + Type: OpRemoveContainer, + ResourceID: "service:web:1", + Cause: "scale down", + Container: &ctr, + }, "", stopNode) + + err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, plan) + assert.NilError(t, err) +} + +// notFoundError implements the errdefs.ErrNotFound interface for test mocks. +type notFoundError struct{} + +func (notFoundError) Error() string { return "not found" } +func (notFoundError) NotFound() {}