introduce service hooks

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

cmd/formatter/logs.go

@@ -61,21 +61,32 @@ func (l *logConsumer) Register(name string) {
 }
 
 func (l *logConsumer) register(name string) *presenter {
-	cf := monochrome
-	if l.color {
-		if name == api.WatchLogger {
-			cf = makeColorFunc("92")
-		} else {
-			cf = nextColor()
+	var p *presenter
+	root, _, found := strings.Cut(name, " ")
+	if found {
+		parent := l.getPresenter(root)
+		p = &presenter{
+			colors: parent.colors,
+			name:   name,
+			prefix: parent.prefix,
+		}
+	} else {
+		cf := monochrome
+		if l.color {
+			if name == api.WatchLogger {
+				cf = makeColorFunc("92")
+			} else {
+				cf = nextColor()
+			}
+		}
+		p = &presenter{
+			colors: cf,
+			name:   name,
 		}
-	}
-	p := &presenter{
-		colors: cf,
-		name:   name,
 	}
 	l.presenters.Store(name, p)
+	l.computeWidth()
 	if l.prefix {
-		l.computeWidth()
 		l.presenters.Range(func(key, value interface{}) bool {
 			p := value.(*presenter)
 			p.setPrefix(l.width)

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/Microsoft/go-winio v0.6.2
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d
 	github.com/buger/goterm v1.0.4
-	github.com/compose-spec/compose-go/v2 v2.2.1-0.20241003145835-48d3a5bbf4ea
+	github.com/compose-spec/compose-go/v2 v2.2.1-0.20241007090213-a59035ad2bf4
 	github.com/containerd/containerd v1.7.22
 	github.com/containerd/platforms v0.2.1
 	github.com/davecgh/go-spew v1.1.1

go.sum

@@ -85,8 +85,8 @@ github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa h1:jQCWAUqqlij9Pgj2i/P
 github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa/go.mod h1:x/1Gn8zydmfq8dk6e9PdstVsDgu9RuyIIJqAaF//0IM=
 github.com/codahale/rfc6979 v0.0.0-20141003034818-6a90f24967eb h1:EDmT6Q9Zs+SbUoc7Ik9EfrFqcylYqgPZ9ANSbTAntnE=
 github.com/codahale/rfc6979 v0.0.0-20141003034818-6a90f24967eb/go.mod h1:ZjrT6AXHbDs86ZSdt/osfBi5qfexBrKUdONk989Wnk4=
-github.com/compose-spec/compose-go/v2 v2.2.1-0.20241003145835-48d3a5bbf4ea h1:BU/Sx/dAU6f64sDad58igm4OwwI1Z1uvV5E0ZKv4CZ8=
-github.com/compose-spec/compose-go/v2 v2.2.1-0.20241003145835-48d3a5bbf4ea/go.mod h1:lFN0DrMxIncJGYAXTfWuajfwj5haBJqrBkarHcnjJKc=
+github.com/compose-spec/compose-go/v2 v2.2.1-0.20241007090213-a59035ad2bf4 h1:2FWtPQWe/tkeGuwxk5x03luRw5pzPhPCRfzfeVw56vo=
+github.com/compose-spec/compose-go/v2 v2.2.1-0.20241007090213-a59035ad2bf4/go.mod h1:lFN0DrMxIncJGYAXTfWuajfwj5haBJqrBkarHcnjJKc=
 github.com/containerd/cgroups v1.1.0 h1:v8rEWFl6EoqHB+swVNjVoCJE8o3jX7e8nqBGPLaDFBM=
 github.com/containerd/cgroups v1.1.0/go.mod h1:6ppBcbh/NOOUU+dMKrykgaBnK9lCIBxHqJDGwsa1mIw=
 github.com/containerd/console v1.0.4 h1:F2g4+oChYvBTsASRTz8NP6iIAi97J3TtSAsLbIFn4ro=

pkg/api/api.go

@@ -637,6 +637,8 @@ const (
 	ContainerEventExit
 	// UserCancel user cancelled compose up, we are stopping containers
 	UserCancel
+	// HookEventLog is a ContainerEvent of type log on stdout by service hook
+	HookEventLog
 )
 
 // Separator is used for naming components

pkg/compose/convergence.go

@@ -149,7 +149,7 @@ func (c *convergence) ensureService(ctx context.Context, project *types.Project,
 			container := container
 			traceOpts := append(tracing.ServiceOptions(service), tracing.ContainerOptions(container)...)
 			eg.Go(tracing.SpanWrapFuncForErrGroup(ctx, "service/scale/down", traceOpts, func(ctx context.Context) error {
-				return c.service.stopAndRemoveContainer(ctx, container, timeout, false)
+				return c.service.stopAndRemoveContainer(ctx, container, &service, timeout, false)
 			}))
 			continue
 		}
@@ -224,7 +224,7 @@ func (c *convergence) stopDependentContainers(ctx context.Context, project *type
 	dependents := project.GetDependentsForService(service)
 	for _, name := range dependents {
 		dependents := c.getObservedState(name)
-		err := c.service.stopContainers(ctx, w, dependents, nil)
+		err := c.service.stopContainers(ctx, w, &service, dependents, nil)
 		if err != nil {
 			return err
 		}
@@ -769,7 +769,10 @@ func (s *composeService) isServiceCompleted(ctx context.Context, containers Cont
 	return false, 0, nil
 }
 
-func (s *composeService) startService(ctx context.Context, project *types.Project, service types.ServiceConfig, containers Containers, timeout time.Duration) error {
+func (s *composeService) startService(ctx context.Context,
+	project *types.Project, service types.ServiceConfig,
+	containers Containers, listener api.ContainerEventListener,
+	timeout time.Duration) error {
 	if service.Deploy != nil && service.Deploy.Replicas != nil && *service.Deploy.Replicas == 0 {
 		return nil
 	}
@@ -793,10 +796,18 @@ func (s *composeService) startService(ctx context.Context, project *types.Projec
 		}
 		eventName := getContainerProgressName(container)
 		w.Event(progress.StartingEvent(eventName))
-		err := s.apiClient().ContainerStart(ctx, container.ID, containerType.StartOptions{})
+		err = s.apiClient().ContainerStart(ctx, container.ID, containerType.StartOptions{})
 		if err != nil {
 			return err
 		}
+
+		for _, hook := range service.PostStart {
+			err = s.runHook(ctx, container, service, hook, listener)
+			if err != nil {
+				return err
+			}
+		}
+
 		w.Event(progress.StartedEvent(eventName))
 	}
 	return nil

pkg/compose/create.go

@@ -104,7 +104,7 @@ func (s *composeService) create(ctx context.Context, project *types.Project, opt
 	orphans := observedState.filter(isOrphaned(project))
 	if len(orphans) > 0 && !options.IgnoreOrphans {
 		if options.RemoveOrphans {
-			err := s.removeContainers(ctx, orphans, nil, false)
+			err := s.removeContainers(ctx, orphans, nil, nil, false)
 			if err != nil {
 				return err
 			}

pkg/compose/down.go

@@ -85,7 +85,8 @@ func (s *composeService) down(ctx context.Context, projectName string, options a
 
 	err = InReverseDependencyOrder(ctx, project, func(c context.Context, service string) error {
 		serviceContainers := containers.filter(isService(service))
-		err := s.removeContainers(ctx, serviceContainers, options.Timeout, options.Volumes)
+		serv := project.Services[service]
+		err := s.removeContainers(ctx, serviceContainers, &serv, options.Timeout, options.Volumes)
 		return err
 	}, WithRootNodesAndDown(options.Services))
 	if err != nil {
@@ -94,7 +95,7 @@ func (s *composeService) down(ctx context.Context, projectName string, options a
 
 	orphans := containers.filter(isOrphaned(project))
 	if options.RemoveOrphans && len(orphans) > 0 {
-		err := s.removeContainers(ctx, orphans, options.Timeout, false)
+		err := s.removeContainers(ctx, orphans, nil, options.Timeout, false)
 		if err != nil {
 			return err
 		}
@@ -296,9 +297,19 @@ func (s *composeService) removeVolume(ctx context.Context, id string, w progress
 	return err
 }
 
-func (s *composeService) stopContainer(ctx context.Context, w progress.Writer, container moby.Container, timeout *time.Duration) error {
+func (s *composeService) stopContainer(ctx context.Context, w progress.Writer, service *types.ServiceConfig, container moby.Container, timeout *time.Duration) error {
 	eventName := getContainerProgressName(container)
 	w.Event(progress.StoppingEvent(eventName))
+
+	if service != nil {
+		for _, hook := range service.PreStop {
+			err := s.runHook(ctx, container, *service, hook, nil)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
 	timeoutInSecond := utils.DurationSecondToInt(timeout)
 	err := s.apiClient().ContainerStop(ctx, container.ID, containerType.StopOptions{Timeout: timeoutInSecond})
 	if err != nil {
@@ -309,32 +320,32 @@ func (s *composeService) stopContainer(ctx context.Context, w progress.Writer, c
 	return nil
 }
 
-func (s *composeService) stopContainers(ctx context.Context, w progress.Writer, containers []moby.Container, timeout *time.Duration) error {
+func (s *composeService) stopContainers(ctx context.Context, w progress.Writer, serv *types.ServiceConfig, containers []moby.Container, timeout *time.Duration) error {
 	eg, ctx := errgroup.WithContext(ctx)
 	for _, container := range containers {
 		container := container
 		eg.Go(func() error {
-			return s.stopContainer(ctx, w, container, timeout)
+			return s.stopContainer(ctx, w, serv, container, timeout)
 		})
 	}
 	return eg.Wait()
 }
 
-func (s *composeService) removeContainers(ctx context.Context, containers []moby.Container, timeout *time.Duration, volumes bool) error {
+func (s *composeService) removeContainers(ctx context.Context, containers []moby.Container, service *types.ServiceConfig, timeout *time.Duration, volumes bool) error {
 	eg, _ := errgroup.WithContext(ctx)
 	for _, container := range containers {
 		container := container
 		eg.Go(func() error {
-			return s.stopAndRemoveContainer(ctx, container, timeout, volumes)
+			return s.stopAndRemoveContainer(ctx, container, service, timeout, volumes)
 		})
 	}
 	return eg.Wait()
 }
 
-func (s *composeService) stopAndRemoveContainer(ctx context.Context, container moby.Container, timeout *time.Duration, volumes bool) error {
+func (s *composeService) stopAndRemoveContainer(ctx context.Context, container moby.Container, service *types.ServiceConfig, timeout *time.Duration, volumes bool) error {
 	w := progress.ContextWriter(ctx)
 	eventName := getContainerProgressName(container)
-	err := s.stopContainer(ctx, w, container, timeout)
+	err := s.stopContainer(ctx, w, service, container, timeout)
 	if errdefs.IsNotFound(err) {
 		w.Event(progress.RemovedEvent(eventName))
 		return nil

pkg/compose/hook.go

@@ -0,0 +1,122 @@
+/*
+   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"
+	"io"
+	"time"
+
+	"github.com/compose-spec/compose-go/v2/types"
+	"github.com/docker/compose/v2/pkg/api"
+	"github.com/docker/compose/v2/pkg/utils"
+	moby "github.com/docker/docker/api/types"
+	containerType "github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/pkg/stdcopy"
+)
+
+func (s composeService) runHook(ctx context.Context, container moby.Container, service types.ServiceConfig, hook types.ServiceHook, listener api.ContainerEventListener) error {
+	wOut := utils.GetWriter(func(line string) {
+		listener(api.ContainerEvent{
+			Type:      api.HookEventLog,
+			Container: getContainerNameWithoutProject(container) + " ->",
+			ID:        container.ID,
+			Service:   service.Name,
+			Line:      line,
+		})
+	})
+	defer wOut.Close() //nolint:errcheck
+
+	detached := listener == nil
+	exec, err := s.apiClient().ContainerExecCreate(ctx, container.ID, containerType.ExecOptions{
+		User:         hook.User,
+		Privileged:   hook.Privileged,
+		Env:          ToMobyEnv(hook.Environment),
+		WorkingDir:   hook.WorkingDir,
+		Cmd:          hook.Command,
+		Detach:       detached,
+		AttachStdout: !detached,
+		AttachStderr: !detached,
+	})
+	if err != nil {
+		return err
+	}
+
+	if detached {
+		return s.runWaitExec(ctx, exec, service, listener)
+	}
+
+	height, width := s.stdout().GetTtySize()
+	consoleSize := &[2]uint{height, width}
+	attach, err := s.apiClient().ContainerExecAttach(ctx, exec.ID, containerType.ExecAttachOptions{
+		Tty:         service.Tty,
+		ConsoleSize: consoleSize,
+	})
+	if err != nil {
+		return err
+	}
+	defer attach.Close()
+
+	if service.Tty {
+		_, err = io.Copy(wOut, attach.Reader)
+	} else {
+		_, err = stdcopy.StdCopy(wOut, wOut, attach.Reader)
+	}
+	if err != nil {
+		return err
+	}
+
+	inspected, err := s.apiClient().ContainerExecInspect(ctx, exec.ID)
+	if err != nil {
+		return err
+	}
+	if inspected.ExitCode != 0 {
+		return fmt.Errorf("%s hook exited with status %d", service.Name, inspected.ExitCode)
+	}
+	return nil
+}
+
+func (s composeService) runWaitExec(ctx context.Context, exec moby.IDResponse, service types.ServiceConfig, listener api.ContainerEventListener) error {
+	err := s.apiClient().ContainerExecStart(ctx, exec.ID, containerType.ExecStartOptions{
+		Detach: listener == nil,
+		Tty:    service.Tty,
+	})
+	if err != nil {
+		return nil
+	}
+
+	// We miss a ContainerExecWait API
+	tick := time.NewTicker(100 * time.Millisecond)
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-tick.C:
+			inspect, err := s.apiClient().ContainerExecInspect(ctx, exec.ID)
+			if err != nil {
+				return nil
+			}
+			if !inspect.Running {
+				if inspect.ExitCode != 0 {
+					return fmt.Errorf("%s hook exited with status %d", service.Name, inspect.ExitCode)
+				}
+				return nil
+			}
+		}
+	}
+}

pkg/compose/printer.go

@@ -148,7 +148,7 @@ func (p *printer) Run(cascade api.Cascade, exitCodeFrom string, stopFn func() er
 					// Last container terminated, done
 					return exitCode, nil
 				}
-			case api.ContainerEventLog:
+			case api.ContainerEventLog, api.HookEventLog:
 				if !aborting {
 					p.consumer.Log(container, event.Line)
 				}

pkg/compose/start.go

@@ -129,7 +129,7 @@ func (s *composeService) start(ctx context.Context, projectName string, options
 			return err
 		}
 
-		return s.startService(ctx, project, service, containers, options.WaitTimeout)
+		return s.startService(ctx, project, service, containers, listener, options.WaitTimeout)
 	})
 	if err != nil {
 		return err

pkg/compose/stop.go

@@ -54,6 +54,7 @@ func (s *composeService) stop(ctx context.Context, projectName string, options a
 		if !utils.StringContains(options.Services, service) {
 			return nil
 		}
-		return s.stopContainers(ctx, w, containers.filter(isService(service)).filter(isNotOneOff), options.Timeout)
+		serv := project.Services[service]
+		return s.stopContainers(ctx, w, &serv, containers.filter(isService(service)).filter(isNotOneOff), options.Timeout)
 	})
 }