docker-compose/pkg/compose/compose.go

/*
   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"
	"fmt"
	"io"
	"strconv"
	"strings"
	"sync"

	"github.com/compose-spec/compose-go/v2/types"
	"github.com/docker/buildx/store/storeutil"
	"github.com/docker/cli/cli/command"
	"github.com/docker/cli/cli/config/configfile"
	"github.com/docker/cli/cli/flags"
	"github.com/docker/cli/cli/streams"
	"github.com/jonboulle/clockwork"
	"github.com/moby/moby/api/types/container"
	"github.com/moby/moby/api/types/swarm"
	"github.com/moby/moby/client"
	"github.com/sirupsen/logrus"

	"github.com/docker/compose/v5/pkg/api"
	"github.com/docker/compose/v5/pkg/dryrun"
)

type Option func(service *composeService) error

// NewComposeService creates a Compose service using Docker CLI.
// This is the standard constructor that requires command.Cli for full functionality.
//
// Example usage:
//
//	dockerCli, _ := command.NewDockerCli()
//	service := NewComposeService(dockerCli)
//
// For advanced configuration with custom overrides, use ServiceOption functions:
//
//	service := NewComposeService(dockerCli,
//	    WithPrompt(prompt.NewPrompt(cli.In(), cli.Out()).Confirm),
//	    WithOutputStream(customOut),
//	    WithErrorStream(customErr),
//	    WithInputStream(customIn))
//
// Or set all streams at once:
//
//	service := NewComposeService(dockerCli,
//	    WithStreams(customOut, customErr, customIn))
func NewComposeService(dockerCli command.Cli, options ...Option) (api.Compose, error) {
	s := &composeService{
		dockerCli:      dockerCli,
		clock:          clockwork.NewRealClock(),
		maxConcurrency: -1,
		dryRun:         false,
	}
	for _, option := range options {
		if err := option(s); err != nil {
			return nil, err
		}
	}
	if s.prompt == nil {
		s.prompt = func(message string, defaultValue bool) (bool, error) {
			fmt.Println(message)
			logrus.Warning("Compose is running without a 'prompt' component to interact with user")
			return defaultValue, nil
		}
	}
	if s.events == nil {
		s.events = &ignore{}
	}

	// If custom streams were provided, wrap the Docker CLI to use them
	if s.outStream != nil || s.errStream != nil || s.inStream != nil {
		s.dockerCli = s.wrapDockerCliWithStreams(dockerCli)
	}

	return s, nil
}

// WithStreams sets custom I/O streams for output and interaction
func WithStreams(out, err io.Writer, in io.Reader) Option {
	return func(s *composeService) error {
		s.outStream = out
		s.errStream = err
		s.inStream = in
		return nil
	}
}

// WithOutputStream sets a custom output stream
func WithOutputStream(out io.Writer) Option {
	return func(s *composeService) error {
		s.outStream = out
		return nil
	}
}

// WithErrorStream sets a custom error stream
func WithErrorStream(err io.Writer) Option {
	return func(s *composeService) error {
		s.errStream = err
		return nil
	}
}

// WithInputStream sets a custom input stream
func WithInputStream(in io.Reader) Option {
	return func(s *composeService) error {
		s.inStream = in
		return nil
	}
}

// WithContextInfo sets custom Docker context information
func WithContextInfo(info api.ContextInfo) Option {
	return func(s *composeService) error {
		s.contextInfo = info
		return nil
	}
}

// WithProxyConfig sets custom HTTP proxy configuration for builds
func WithProxyConfig(config map[string]string) Option {
	return func(s *composeService) error {
		s.proxyConfig = config
		return nil
	}
}

// WithPrompt configure a UI component for Compose service to interact with user and confirm actions
func WithPrompt(prompt Prompt) Option {
	return func(s *composeService) error {
		s.prompt = prompt
		return nil
	}
}

// WithMaxConcurrency defines upper limit for concurrent operations against engine API
func WithMaxConcurrency(maxConcurrency int) Option {
	return func(s *composeService) error {
		s.maxConcurrency = maxConcurrency
		return nil
	}
}

// WithDryRun configure Compose to run without actually applying changes
func WithDryRun(s *composeService) error {
	s.dryRun = true
	cli, err := command.NewDockerCli()
	if err != nil {
		return err
	}

	options := flags.NewClientOptions()
	options.Context = s.dockerCli.CurrentContext()
	err = cli.Initialize(options, command.WithInitializeClient(func(cli *command.DockerCli) (client.APIClient, error) {
		return dryrun.NewDryRunClient(s.apiClient(), s.dockerCli)
	}))
	if err != nil {
		return err
	}
	s.dockerCli = cli
	return nil
}

type Prompt func(message string, defaultValue bool) (bool, error)

// AlwaysOkPrompt returns a Prompt implementation that always returns true without user interaction.
func AlwaysOkPrompt() Prompt {
	return func(message string, defaultValue bool) (bool, error) {
		return true, nil
	}
}

// WithEventProcessor configure component to get notified on Compose operation and progress events.
// Typically used to configure a progress UI
func WithEventProcessor(bus api.EventProcessor) Option {
	return func(s *composeService) error {
		s.events = bus
		return nil
	}
}

type composeService struct {
	dockerCli command.Cli
	// prompt is used to interact with user and confirm actions
	prompt Prompt
	// eventBus collects tasks execution events
	events api.EventProcessor

	// Optional overrides for specific components (for SDK users)
	outStream   io.Writer
	errStream   io.Writer
	inStream    io.Reader
	contextInfo api.ContextInfo
	proxyConfig map[string]string

	clock          clockwork.Clock
	maxConcurrency int
	dryRun         bool

	runtimeAPIVersion runtimeVersionCache
}

// Close releases any connections/resources held by the underlying clients.
//
// In practice, this service has the same lifetime as the process, so everything
// will get cleaned up at about the same time regardless even if not invoked.
func (s *composeService) Close() error {
	var errs []error
	if s.dockerCli != nil {
		errs = append(errs, s.apiClient().Close())
	}
	return errors.Join(errs...)
}

func (s *composeService) apiClient() client.APIClient {
	return s.dockerCli.Client()
}

func (s *composeService) configFile() *configfile.ConfigFile {
	return s.dockerCli.ConfigFile()
}

// getContextInfo returns the context info - either custom override or dockerCli adapter
func (s *composeService) getContextInfo() api.ContextInfo {
	if s.contextInfo != nil {
		return s.contextInfo
	}
	return &dockerCliContextInfo{cli: s.dockerCli}
}

// getProxyConfig returns the proxy config - either custom override or environment-based
func (s *composeService) getProxyConfig() map[string]string {
	if s.proxyConfig != nil {
		return s.proxyConfig
	}
	return storeutil.GetProxyConfig(s.dockerCli)
}

func (s *composeService) stdout() *streams.Out {
	return s.dockerCli.Out()
}

func (s *composeService) stdin() *streams.In {
	return s.dockerCli.In()
}

func (s *composeService) stderr() *streams.Out {
	return s.dockerCli.Err()
}

// readCloserAdapter adapts io.Reader to io.ReadCloser
type readCloserAdapter struct {
	r io.Reader
}

func (r *readCloserAdapter) Read(p []byte) (int, error) {
	return r.r.Read(p)
}

func (r *readCloserAdapter) Close() error {
	return nil
}

// wrapDockerCliWithStreams wraps the Docker CLI to intercept and override stream methods
func (s *composeService) wrapDockerCliWithStreams(baseCli command.Cli) command.Cli {
	wrapper := &streamOverrideWrapper{
		Cli: baseCli,
	}

	// Wrap custom streams in Docker CLI's stream types
	if s.outStream != nil {
		wrapper.outStream = streams.NewOut(s.outStream)
	}
	if s.errStream != nil {
		wrapper.errStream = streams.NewOut(s.errStream)
	}
	if s.inStream != nil {
		wrapper.inStream = streams.NewIn(&readCloserAdapter{r: s.inStream})
	}

	return wrapper
}

// streamOverrideWrapper wraps command.Cli to override streams with custom implementations
type streamOverrideWrapper struct {
	command.Cli
	outStream *streams.Out
	errStream *streams.Out
	inStream  *streams.In
}

func (w *streamOverrideWrapper) Out() *streams.Out {
	if w.outStream != nil {
		return w.outStream
	}
	return w.Cli.Out()
}

func (w *streamOverrideWrapper) Err() *streams.Out {
	if w.errStream != nil {
		return w.errStream
	}
	return w.Cli.Err()
}

func (w *streamOverrideWrapper) In() *streams.In {
	if w.inStream != nil {
		return w.inStream
	}
	return w.Cli.In()
}

func getCanonicalContainerName(c container.Summary) string {
	if len(c.Names) == 0 {
		// corner case, sometime happens on removal. return short ID as a safeguard value
		return c.ID[:12]
	}
	// Names return container canonical name /foo  + link aliases /linked_by/foo
	for _, name := range c.Names {
		if strings.LastIndex(name, "/") == 0 {
			return name[1:]
		}
	}

	return strings.TrimPrefix(c.Names[0], "/")
}

func getContainerNameWithoutProject(c container.Summary) string {
	project := c.Labels[api.ProjectLabel]
	defaultName := getDefaultContainerName(project, c.Labels[api.ServiceLabel], c.Labels[api.ContainerNumberLabel])
	name := getCanonicalContainerName(c)
	if name != defaultName {
		// service declares a custom container_name
		return name
	}
	return name[len(project)+1:]
}

// projectFromName builds a types.Project based on actual resources with compose labels set
func (s *composeService) projectFromName(containers Containers, projectName string, services ...string) (*types.Project, error) {
	project := &types.Project{
		Name:     projectName,
		Services: types.Services{},
	}
	if len(containers) == 0 {
		return project, fmt.Errorf("no container found for project %q: %w", projectName, api.ErrNotFound)
	}
	set := types.Services{}
	for _, c := range containers {
		serviceLabel, ok := c.Labels[api.ServiceLabel]
		if !ok {
			serviceLabel = getCanonicalContainerName(c)
		}
		service, ok := set[serviceLabel]
		if !ok {
			service = types.ServiceConfig{
				Name:   serviceLabel,
				Image:  c.Image,
				Labels: c.Labels,
			}
		}
		service.Scale = increment(service.Scale)
		set[serviceLabel] = service
	}
	for name, service := range set {
		dependencies := service.Labels[api.DependenciesLabel]
		if dependencies != "" {
			service.DependsOn = types.DependsOnConfig{}
			for dc := range strings.SplitSeq(dependencies, ",") {
				dcArr := strings.Split(dc, ":")
				condition := ServiceConditionRunningOrHealthy
				// Let's restart the dependency by default if we don't have the info stored in the label
				restart := true
				required := true
				dependency := dcArr[0]

				// backward compatibility
				if len(dcArr) > 1 {
					condition = dcArr[1]
					if len(dcArr) > 2 {
						restart, _ = strconv.ParseBool(dcArr[2])
					}
				}
				service.DependsOn[dependency] = types.ServiceDependency{Condition: condition, Restart: restart, Required: required}
			}
			set[name] = service
		}
	}
	project.Services = set

SERVICES:
	for _, qs := range services {
		for _, es := range project.Services {
			if es.Name == qs {
				continue SERVICES
			}
		}
		return project, fmt.Errorf("no such service: %q: %w", qs, api.ErrNotFound)
	}
	project, err := project.WithSelectedServices(services)
	if err != nil {
		return project, err
	}

	return project, nil
}

func increment(scale *int) *int {
	i := 1
	if scale != nil {
		i = *scale + 1
	}
	return &i
}

func (s *composeService) actualVolumes(ctx context.Context, projectName string) (types.Volumes, error) {
	opts := client.VolumeListOptions{
		Filters: projectFilter(projectName),
	}
	volumes, err := s.apiClient().VolumeList(ctx, opts)
	if err != nil {
		return nil, err
	}

	actual := types.Volumes{}
	for _, vol := range volumes.Items {
		actual[vol.Labels[api.VolumeLabel]] = types.VolumeConfig{
			Name:   vol.Name,
			Driver: vol.Driver,
			Labels: vol.Labels,
		}
	}
	return actual, nil
}

func (s *composeService) actualNetworks(ctx context.Context, projectName string) (types.Networks, error) {
	networks, err := s.apiClient().NetworkList(ctx, client.NetworkListOptions{
		Filters: projectFilter(projectName),
	})
	if err != nil {
		return nil, err
	}

	actual := types.Networks{}
	for _, net := range networks.Items {
		actual[net.Labels[api.NetworkLabel]] = types.NetworkConfig{
			Name:   net.Name,
			Driver: net.Driver,
			Labels: net.Labels,
		}
	}
	return actual, nil
}

var swarmEnabled = struct {
	once sync.Once
	val  bool
	err  error
}{}

func (s *composeService) isSwarmEnabled(ctx context.Context) (bool, error) {
	swarmEnabled.once.Do(func() {
		res, err := s.apiClient().Info(ctx, client.InfoOptions{})
		if err != nil {
			swarmEnabled.err = err
		}
		switch res.Info.Swarm.LocalNodeState {
		case swarm.LocalNodeStateInactive, swarm.LocalNodeStateLocked:
			swarmEnabled.val = false
		default:
			swarmEnabled.val = true
		}
	})
	return swarmEnabled.val, swarmEnabled.err
}

// runtimeVersionCache caches a version string after a successful lookup.
// Errors (including context cancellation) are not cached so that
// subsequent calls can retry with a fresh context.
type runtimeVersionCache struct {
	mu  sync.Mutex
	val string
}

// RuntimeAPIVersion returns the negotiated API version that will be used for
// requests to the Docker daemon. It triggers version negotiation via Ping so
// that version-gated request shaping matches the version subsequent API calls
// will actually use.
//
// After negotiation, Compose should never rely on features or request attributes
// not defined by this API version, even if the daemon's raw version is higher.
func (s *composeService) RuntimeAPIVersion(ctx context.Context) (string, error) {
	s.runtimeAPIVersion.mu.Lock()
	defer s.runtimeAPIVersion.mu.Unlock()
	if s.runtimeAPIVersion.val != "" {
		return s.runtimeAPIVersion.val, nil
	}

	cli := s.apiClient()
	_, err := cli.Ping(ctx, client.PingOptions{NegotiateAPIVersion: true})
	if err != nil {
		return "", err
	}

	version := cli.ClientVersion()
	if version == "" {
		return "", fmt.Errorf("docker client returned empty version after successful API negotiation")
	}

	s.runtimeAPIVersion.val = version
	return s.runtimeAPIVersion.val, nil
}