diff --git a/cmd/compose/compose.go b/cmd/compose/compose.go index 9f537caf83..58e519ad9c 100644 --- a/cmd/compose/compose.go +++ b/cmd/compose/compose.go @@ -434,9 +434,6 @@ func (o *BackendOptions) Add(option compose.Option) { } // RootCommand returns the compose command with its child commands -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func RootCommand(dockerCli command.Cli, backendOptions *BackendOptions) *cobra.Command { opts := ProjectOptions{} var ( @@ -467,48 +464,24 @@ func RootCommand(dockerCli command.Cli, backendOptions *BackendOptions) *cobra.C } }, PersistentPreRunE: func(cmd *cobra.Command, args []string) error { - parent := cmd.Root() - if parent != nil { - parentPrerun := parent.PersistentPreRunE - if parentPrerun != nil { - err := parentPrerun(cmd, args) - if err != nil { - return err - } - } + err := runParentPreRun(cmd, args) + if err != nil { + return err } if verbose { logrus.SetLevel(logrus.TraceLevel) } - err := setEnvWithDotEnv(opts, dockerCli) + err = setEnvWithDotEnv(opts, dockerCli) if err != nil { return err } - if noAnsi { - if ansi != "auto" { - return errors.New(`cannot specify DEPRECATED "--no-ansi" and "--ansi". Please use only "--ansi"`) - } - ansi = "never" - fmt.Fprint(os.Stderr, "option '--no-ansi' is DEPRECATED ! Please use '--ansi' instead.\n") - } - if v, ok := os.LookupEnv("COMPOSE_ANSI"); ok && !cmd.Flags().Changed("ansi") { - ansi = v - } - formatter.SetANSIMode(dockerCli, ansi) - - if noColor, ok := os.LookupEnv("NO_COLOR"); ok && noColor != "" { - display.NoColor() - formatter.SetANSIMode(dockerCli, formatter.Never) - } - - switch ansi { - case "never": - display.Mode = display.ModePlain - case "always": - display.Mode = display.ModeTTY + ansi, err = resolveAnsiMode(cmd, ansi, noAnsi) + if err != nil { + return err } + applyDisplayMode(dockerCli, ansi) detached, _ := cmd.Flags().GetBool("detach") ep, err := selectEventProcessor(dockerCli, opts.Progress, ansi, detached) @@ -517,41 +490,14 @@ func RootCommand(dockerCli command.Cli, backendOptions *BackendOptions) *cobra.C } backendOptions.Add(compose.WithEventProcessor(ep)) - // (4) options validation / normalization - if opts.WorkDir != "" { - if opts.ProjectDir != "" { - return errors.New(`cannot specify DEPRECATED "--workdir" and "--project-directory". Please use only "--project-directory" instead`) - } - opts.ProjectDir = opts.WorkDir - fmt.Fprint(os.Stderr, aec.Apply("option '--workdir' is DEPRECATED at root level! Please use '--project-directory' instead.\n", aec.RedF)) - } - for i, file := range opts.EnvFiles { - file = composepaths.ExpandUser(file) - if !filepath.IsAbs(file) { - file, err := filepath.Abs(file) - if err != nil { - return err - } - opts.EnvFiles[i] = file - } else { - opts.EnvFiles[i] = file - } - } - - composeCmd := cmd - for composeCmd.Name() != PluginName { - if !composeCmd.HasParent() { - return fmt.Errorf("error parsing command line, expected %q", PluginName) - } - composeCmd = composeCmd.Parent() + err = normalizeProjectOptions(&opts) + if err != nil { + return err } - if v, ok := os.LookupEnv(ComposeParallelLimit); ok && !composeCmd.Flags().Changed("parallel") { - i, err := strconv.Atoi(v) - if err != nil { - return fmt.Errorf("%s must be an integer (found: %q)", ComposeParallelLimit, v) - } - parallel = i + parallel, err = resolveMaxConcurrency(cmd, parallel) + if err != nil { + return err } if parallel > 0 { logrus.Debugf("Limiting max concurrency to %d jobs", parallel) @@ -644,6 +590,98 @@ func RootCommand(dockerCli command.Cli, backendOptions *BackendOptions) *cobra.C return c } +// runParentPreRun invokes the docker CLI root command's PersistentPreRunE, +// which cobra doesn't chain automatically. +func runParentPreRun(cmd *cobra.Command, args []string) error { + parent := cmd.Root() + if parent == nil { + return nil + } + if prerun := parent.PersistentPreRunE; prerun != nil { + return prerun(cmd, args) + } + return nil +} + +// resolveAnsiMode reconciles --ansi with the deprecated --no-ansi flag and +// the COMPOSE_ANSI environment variable (flag wins over environment). +func resolveAnsiMode(cmd *cobra.Command, ansi string, noAnsi bool) (string, error) { + if noAnsi { + if ansi != "auto" { + return "", errors.New(`cannot specify DEPRECATED "--no-ansi" and "--ansi". Please use only "--ansi"`) + } + ansi = "never" + fmt.Fprint(os.Stderr, "option '--no-ansi' is DEPRECATED ! Please use '--ansi' instead.\n") + } + if v, ok := os.LookupEnv("COMPOSE_ANSI"); ok && !cmd.Flags().Changed("ansi") { + ansi = v + } + return ansi, nil +} + +// applyDisplayMode configures ANSI output and the progress display mode, +// honoring the NO_COLOR convention (https://no-color.org). +func applyDisplayMode(dockerCli command.Cli, ansi string) { + formatter.SetANSIMode(dockerCli, ansi) + + if noColor, ok := os.LookupEnv("NO_COLOR"); ok && noColor != "" { + display.NoColor() + formatter.SetANSIMode(dockerCli, formatter.Never) + } + + switch ansi { + case "never": + display.Mode = display.ModePlain + case "always": + display.Mode = display.ModeTTY + } +} + +// normalizeProjectOptions handles the deprecated --workdir flag and makes +// env-file paths absolute. +func normalizeProjectOptions(opts *ProjectOptions) error { + if opts.WorkDir != "" { + if opts.ProjectDir != "" { + return errors.New(`cannot specify DEPRECATED "--workdir" and "--project-directory". Please use only "--project-directory" instead`) + } + opts.ProjectDir = opts.WorkDir + fmt.Fprint(os.Stderr, aec.Apply("option '--workdir' is DEPRECATED at root level! Please use '--project-directory' instead.\n", aec.RedF)) + } + for i, file := range opts.EnvFiles { + file = composepaths.ExpandUser(file) + if !filepath.IsAbs(file) { + abs, err := filepath.Abs(file) + if err != nil { + return err + } + file = abs + } + opts.EnvFiles[i] = file + } + return nil +} + +// resolveMaxConcurrency returns the parallelism limit: COMPOSE_PARALLEL_LIMIT +// applies unless --parallel was set explicitly on the compose command. +func resolveMaxConcurrency(cmd *cobra.Command, parallel int) (int, error) { + composeCmd := cmd + for composeCmd.Name() != PluginName { + if !composeCmd.HasParent() { + return 0, fmt.Errorf("error parsing command line, expected %q", PluginName) + } + composeCmd = composeCmd.Parent() + } + + if v, ok := os.LookupEnv(ComposeParallelLimit); ok && !composeCmd.Flags().Changed("parallel") { + i, err := strconv.Atoi(v) + if err != nil { + return 0, fmt.Errorf("%s must be an integer (found: %q)", ComposeParallelLimit, v) + } + parallel = i + } + return parallel, nil +} + func stdinfo(dockerCli command.Cli) io.Writer { if stdioToStdout { return dockerCli.Out() diff --git a/pkg/bridge/convert.go b/pkg/bridge/convert.go index f4589afb1f..d544131423 100644 --- a/pkg/bridge/convert.go +++ b/pkg/bridge/convert.go @@ -162,47 +162,13 @@ func convert(ctx context.Context, dockerCli command.Cli, model map[string]any, o } // LoadAdditionalResources loads additional resources from the project, such as image references, secrets, configs and exposed ports -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func LoadAdditionalResources(ctx context.Context, dockerCLI command.Cli, project *types.Project) (*types.Project, error) { for name, service := range project.Services { - imageName := api.GetImageNameOrDefault(service, project.Name) - - var inspect image.InspectResponse - if service.Build != nil && service.Image == "" { - result, err := dockerCLI.Client().ImageInspect(ctx, imageName) - if err != nil { - if !errdefs.IsNotFound(err) { - return nil, err - } - logrus.Warnf("image %s for service %s not found locally; Dockerfile-exposed ports will not be included — run `docker compose build` first to include them", imageName, name) - } - inspect = result.InspectResponse - } else { - var err error - inspect, err = inspectWithPull(ctx, dockerCLI, imageName) - if err != nil { - return nil, err - } - } - service.Image = imageName - exposed := utils.Set[string]{} - exposed.AddAll(service.Expose...) - if inspect.Config != nil { - for port := range inspect.Config.ExposedPorts { - p, err := network.ParsePort(port) - if err != nil { - return nil, err - } - exposed.Add(strconv.Itoa(int(p.Num()))) - } - } - for _, port := range service.Ports { - exposed.Add(strconv.Itoa(int(port.Target))) + updated, err := loadServiceImageResources(ctx, dockerCLI, project.Name, name, service) + if err != nil { + return nil, err } - service.Expose = exposed.Elements() - project.Services[name] = service + project.Services[name] = updated } for name, secret := range project.Secrets { @@ -224,6 +190,51 @@ func LoadAdditionalResources(ctx context.Context, dockerCLI command.Cli, project return project, nil } +// loadServiceImageResources resolves the service image and merges the ports +// it exposes into the service's Expose list +func loadServiceImageResources(ctx context.Context, dockerCLI command.Cli, projectName, name string, service types.ServiceConfig) (types.ServiceConfig, error) { + imageName := api.GetImageNameOrDefault(service, projectName) + + inspect, err := inspectServiceImage(ctx, dockerCLI, name, imageName, service) + if err != nil { + return service, err + } + + service.Image = imageName + exposed := utils.Set[string]{} + exposed.AddAll(service.Expose...) + if inspect.Config != nil { + for port := range inspect.Config.ExposedPorts { + p, err := network.ParsePort(port) + if err != nil { + return service, err + } + exposed.Add(strconv.Itoa(int(p.Num()))) + } + } + for _, port := range service.Ports { + exposed.Add(strconv.Itoa(int(port.Target))) + } + service.Expose = exposed.Elements() + return service, nil +} + +// inspectServiceImage inspects the service image, pulling it when needed; a +// buildable image missing locally only degrades to a warning +func inspectServiceImage(ctx context.Context, dockerCLI command.Cli, name, imageName string, service types.ServiceConfig) (image.InspectResponse, error) { + if service.Build != nil && service.Image == "" { + result, err := dockerCLI.Client().ImageInspect(ctx, imageName) + if err != nil { + if !errdefs.IsNotFound(err) { + return image.InspectResponse{}, err + } + logrus.Warnf("image %s for service %s not found locally; Dockerfile-exposed ports will not be included — run `docker compose build` first to include them", imageName, name) + } + return result.InspectResponse, nil + } + return inspectWithPull(ctx, dockerCLI, imageName) +} + func loadFileObject(conf types.FileObjectConfig) (types.FileObjectConfig, error) { if !conf.External { switch { diff --git a/pkg/compose/build_classic.go b/pkg/compose/build_classic.go index cc8f12fea9..799e3138ee 100644 --- a/pkg/compose/build_classic.go +++ b/pkg/compose/build_classic.go @@ -121,31 +121,22 @@ func (s *composeService) doBuildClassic(ctx context.Context, project *types.Proj return imageIDs, err } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit -func (s *composeService) doBuildImage(ctx context.Context, project *types.Project, service types.ServiceConfig, options api.BuildOptions) (string, error) { - var ( - buildCtx io.ReadCloser - dockerfileCtx io.ReadCloser - contextDir string - relDockerfile string - ) +// classicBuildContext is the build context resolved for the classic builder: +// either an archive stream (buildCtx) or a local directory still to be tarred +// (contextDir), with an optional out-of-context Dockerfile stream. cleanup +// must run once the build is done. +type classicBuildContext struct { + buildCtx io.ReadCloser + dockerfileCtx io.ReadCloser + contextDir string + relDockerfile string + cleanup func() +} - if len(service.Build.Platforms) > 1 { - return "", fmt.Errorf("the classic builder doesn't support multi-arch build, set DOCKER_BUILDKIT=1 to use BuildKit") - } - if service.Build.Privileged { - return "", fmt.Errorf("the classic builder doesn't support privileged mode, set DOCKER_BUILDKIT=1 to use BuildKit") - } - if len(service.Build.AdditionalContexts) > 0 { - return "", fmt.Errorf("the classic builder doesn't support additional contexts, set DOCKER_BUILDKIT=1 to use BuildKit") - } - if len(service.Build.SSH) > 0 { - return "", fmt.Errorf("the classic builder doesn't support SSH keys, set DOCKER_BUILDKIT=1 to use BuildKit") - } - if len(service.Build.Secrets) > 0 { - return "", fmt.Errorf("the classic builder doesn't support secrets, set DOCKER_BUILDKIT=1 to use BuildKit") +func (s *composeService) doBuildImage(ctx context.Context, project *types.Project, service types.ServiceConfig, options api.BuildOptions) (string, error) { + err := checkClassicBuilderSupported(service) + if err != nil { + return "", err } if service.Build.Labels == nil { @@ -158,73 +149,25 @@ func (s *composeService) doBuildImage(ctx context.Context, project *types.Projec progBuff := s.stdout() buildBuff := s.stdout() - contextType, err := build.DetectContextType(specifiedContext) + bctx, err := prepareClassicBuildContext(specifiedContext, dockerfileName, progBuff) if err != nil { return "", err } + defer bctx.cleanup() - switch contextType { - case build.ContextTypeStdin: - return "", fmt.Errorf("building from STDIN is not supported") - case build.ContextTypeLocal: - contextDir, relDockerfile, err = build.GetContextFromLocalDir(specifiedContext, dockerfileName) - if err != nil { - return "", fmt.Errorf("unable to prepare context: %w", err) - } - if strings.HasPrefix(relDockerfile, ".."+string(filepath.Separator)) { - // Dockerfile is outside build-context; read the Dockerfile and pass it as dockerfileCtx - dockerfileCtx, err = os.Open(dockerfileName) - if err != nil { - return "", fmt.Errorf("unable to open Dockerfile: %w", err) - } - defer dockerfileCtx.Close() //nolint:errcheck - } - case build.ContextTypeGit: - var tempDir string - tempDir, relDockerfile, err = build.GetContextFromGitURL(specifiedContext, dockerfileName) - if err != nil { - return "", fmt.Errorf("unable to prepare context: %w", err) - } - defer func() { - _ = os.RemoveAll(tempDir) - }() - contextDir = tempDir - case build.ContextTypeRemote: - buildCtx, relDockerfile, err = build.GetContextFromURL(progBuff, specifiedContext, dockerfileName) - if err != nil { - return "", fmt.Errorf("unable to prepare context: %w", err) - } - default: - return "", fmt.Errorf("unable to prepare context: path %q not found", specifiedContext) - } - + buildCtx := bctx.buildCtx + relDockerfile := bctx.relDockerfile // read from a directory into tar archive if buildCtx == nil { - excludes, err := build.ReadDockerignore(contextDir) - if err != nil { - return "", err - } - - if err := build.ValidateContextDirectory(contextDir, excludes); err != nil { - return "", fmt.Errorf("checking context: %w", err) - } - - // And canonicalize dockerfile name to a platform-independent one - relDockerfile = filepath.ToSlash(relDockerfile) - - excludes = build.TrimBuildFilesFromExcludes(excludes, relDockerfile, false) - buildCtx, err = archive.TarWithOptions(contextDir, &archive.TarOptions{ - ExcludePatterns: excludes, - ChownOpts: &archive.ChownOpts{UID: 0, GID: 0}, - }) + buildCtx, relDockerfile, err = archiveBuildContext(bctx.contextDir, relDockerfile) if err != nil { return "", err } } // replace Dockerfile if it was added from stdin or a file outside the build-context, and there is archive context - if dockerfileCtx != nil && buildCtx != nil { - buildCtx, relDockerfile, err = build.AddDockerfileToBuildContext(dockerfileCtx, buildCtx) + if bctx.dockerfileCtx != nil && buildCtx != nil { + buildCtx, relDockerfile, err = build.AddDockerfileToBuildContext(bctx.dockerfileCtx, buildCtx) if err != nil { return "", err } @@ -239,24 +182,10 @@ func (s *composeService) doBuildImage(ctx context.Context, project *types.Projec progressOutput := streamformatter.NewProgressOutput(progBuff) body := progress.NewProgressReader(buildCtx, progressOutput, 0, "", "Sending build context to Docker daemon") - configFile := s.configFile() - creds, err := configFile.GetAllCredentials() + authConfigs, err := s.classicAuthConfigs() if err != nil { return "", err } - authConfigs := make(map[string]registry.AuthConfig, len(creds)) - for k, authConfig := range creds { - authConfigs[k] = registry.AuthConfig{ - Username: authConfig.Username, - Password: authConfig.Password, - ServerAddress: authConfig.ServerAddress, - - // TODO(thaJeztah): Are these expected to be included? See https://github.com/docker/cli/pull/6516#discussion_r2387586472 - Auth: authConfig.Auth, - IdentityToken: authConfig.IdentityToken, - RegistryToken: authConfig.RegistryToken, - } - } buildOpts := imageBuildOptions(s.getProxyConfig(), project, service, options) imageName := api.GetImageNameOrDefault(service, project.Name) buildOpts.Tags = append(buildOpts.Tags, imageName) @@ -299,6 +228,119 @@ func (s *composeService) doBuildImage(ctx context.Context, project *types.Projec return imageID, nil } +// checkClassicBuilderSupported rejects build options the classic builder +// doesn't implement +func checkClassicBuilderSupported(service types.ServiceConfig) error { + if len(service.Build.Platforms) > 1 { + return fmt.Errorf("the classic builder doesn't support multi-arch build, set DOCKER_BUILDKIT=1 to use BuildKit") + } + if service.Build.Privileged { + return fmt.Errorf("the classic builder doesn't support privileged mode, set DOCKER_BUILDKIT=1 to use BuildKit") + } + if len(service.Build.AdditionalContexts) > 0 { + return fmt.Errorf("the classic builder doesn't support additional contexts, set DOCKER_BUILDKIT=1 to use BuildKit") + } + if len(service.Build.SSH) > 0 { + return fmt.Errorf("the classic builder doesn't support SSH keys, set DOCKER_BUILDKIT=1 to use BuildKit") + } + if len(service.Build.Secrets) > 0 { + return fmt.Errorf("the classic builder doesn't support secrets, set DOCKER_BUILDKIT=1 to use BuildKit") + } + return nil +} + +// prepareClassicBuildContext resolves the build context per its type (local +// directory, git URL, remote URL) +func prepareClassicBuildContext(specifiedContext, dockerfileName string, progBuff io.Writer) (*classicBuildContext, error) { + contextType, err := build.DetectContextType(specifiedContext) + if err != nil { + return nil, err + } + + bctx := &classicBuildContext{cleanup: func() {}} + switch contextType { + case build.ContextTypeStdin: + return nil, fmt.Errorf("building from STDIN is not supported") + case build.ContextTypeLocal: + bctx.contextDir, bctx.relDockerfile, err = build.GetContextFromLocalDir(specifiedContext, dockerfileName) + if err != nil { + return nil, fmt.Errorf("unable to prepare context: %w", err) + } + if strings.HasPrefix(bctx.relDockerfile, ".."+string(filepath.Separator)) { + // Dockerfile is outside build-context; read the Dockerfile and pass it as dockerfileCtx + dockerfileCtx, err := os.Open(dockerfileName) + if err != nil { + return nil, fmt.Errorf("unable to open Dockerfile: %w", err) + } + bctx.dockerfileCtx = dockerfileCtx + bctx.cleanup = func() { _ = dockerfileCtx.Close() } + } + case build.ContextTypeGit: + var tempDir string + tempDir, bctx.relDockerfile, err = build.GetContextFromGitURL(specifiedContext, dockerfileName) + if err != nil { + return nil, fmt.Errorf("unable to prepare context: %w", err) + } + bctx.contextDir = tempDir + bctx.cleanup = func() { _ = os.RemoveAll(tempDir) } + case build.ContextTypeRemote: + bctx.buildCtx, bctx.relDockerfile, err = build.GetContextFromURL(progBuff, specifiedContext, dockerfileName) + if err != nil { + return nil, fmt.Errorf("unable to prepare context: %w", err) + } + default: + return nil, fmt.Errorf("unable to prepare context: path %q not found", specifiedContext) + } + return bctx, nil +} + +// archiveBuildContext tars a local context directory, honoring .dockerignore +// and canonicalizing the dockerfile name to a platform-independent one +func archiveBuildContext(contextDir, relDockerfile string) (io.ReadCloser, string, error) { + excludes, err := build.ReadDockerignore(contextDir) + if err != nil { + return nil, "", err + } + + if err := build.ValidateContextDirectory(contextDir, excludes); err != nil { + return nil, "", fmt.Errorf("checking context: %w", err) + } + + relDockerfile = filepath.ToSlash(relDockerfile) + + excludes = build.TrimBuildFilesFromExcludes(excludes, relDockerfile, false) + buildCtx, err := archive.TarWithOptions(contextDir, &archive.TarOptions{ + ExcludePatterns: excludes, + ChownOpts: &archive.ChownOpts{UID: 0, GID: 0}, + }) + if err != nil { + return nil, "", err + } + return buildCtx, relDockerfile, nil +} + +// classicAuthConfigs converts the CLI credentials to the engine's auth config +func (s *composeService) classicAuthConfigs() (map[string]registry.AuthConfig, error) { + creds, err := s.configFile().GetAllCredentials() + if err != nil { + return nil, err + } + authConfigs := make(map[string]registry.AuthConfig, len(creds)) + for k, authConfig := range creds { + authConfigs[k] = registry.AuthConfig{ + Username: authConfig.Username, + Password: authConfig.Password, + ServerAddress: authConfig.ServerAddress, + + // TODO(thaJeztah): Are these expected to be included? See https://github.com/docker/cli/pull/6516#discussion_r2387586472 + Auth: authConfig.Auth, + IdentityToken: authConfig.IdentityToken, + RegistryToken: authConfig.RegistryToken, + } + } + return authConfigs, nil +} + func imageBuildOptions(proxyConfigs map[string]string, project *types.Project, service types.ServiceConfig, options api.BuildOptions) client.ImageBuildOptions { config := service.Build return client.ImageBuildOptions{ diff --git a/pkg/compose/convergence.go b/pkg/compose/convergence.go index 59a39b60e7..95726963c3 100644 --- a/pkg/compose/convergence.go +++ b/pkg/compose/convergence.go @@ -153,9 +153,6 @@ func containerReasonEvents(containers Containers, eventFunc func(string, string) // ServiceConditionRunningOrHealthy is a service condition on status running or healthy const ServiceConditionRunningOrHealthy = "running_or_healthy" -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) waitDependencies(ctx context.Context, project *types.Project, dependant string, dependencies types.DependsOnConfig, containers Containers, timeout time.Duration) error { if timeout > 0 { withTimeout, cancelFunc := context.WithTimeout(ctx, timeout) @@ -181,79 +178,7 @@ func (s *composeService) waitDependencies(ctx context.Context, project *types.Pr } eg.Go(func() error { - ticker := time.NewTicker(500 * time.Millisecond) - defer ticker.Stop() - for { - select { - case <-ticker.C: - case <-ctx.Done(): - return nil - } - switch config.Condition { - case ServiceConditionRunningOrHealthy: - isHealthy, err := s.isServiceHealthy(ctx, waitingFor, true) - if err != nil { - if !config.Required { - s.events.On(containerReasonEvents(waitingFor, skippedEvent, - fmt.Sprintf("optional dependency %q is not running or is unhealthy", dep))...) - logrus.Warnf("optional dependency %q is not running or is unhealthy: %s", dep, err.Error()) - return nil - } - return err - } - if isHealthy { - s.events.On(containerEvents(waitingFor, healthy)...) - return nil - } - case types.ServiceConditionHealthy: - isHealthy, err := s.isServiceHealthy(ctx, waitingFor, false) - if err != nil { - if !config.Required { - s.events.On(containerReasonEvents(waitingFor, skippedEvent, - fmt.Sprintf("optional dependency %q failed to start", dep))...) - logrus.Warnf("optional dependency %q failed to start: %s", dep, err.Error()) - return nil - } - s.events.On(containerEvents(waitingFor, func(s string) api.Resource { - return errorEventf(s, "dependency %s failed to start", dep) - })...) - return fmt.Errorf("dependency failed to start: %w", err) - } - if isHealthy { - s.events.On(containerEvents(waitingFor, healthy)...) - return nil - } - case types.ServiceConditionCompletedSuccessfully: - isExited, code, err := s.isServiceCompleted(ctx, waitingFor) - if err != nil { - return err - } - if isExited { - if code == 0 { - s.events.On(containerEvents(waitingFor, exited)...) - return nil - } - - messageSuffix := fmt.Sprintf("%q didn't complete successfully: exit %d", dep, code) - if !config.Required { - // optional -> mark as skipped & don't propagate error - s.events.On(containerReasonEvents(waitingFor, skippedEvent, - fmt.Sprintf("optional dependency %s", messageSuffix))...) - logrus.Warnf("optional dependency %s", messageSuffix) - return nil - } - - msg := fmt.Sprintf("service %s", messageSuffix) - s.events.On(containerEvents(waitingFor, func(s string) api.Resource { - return errorEventf(s, "service %s", messageSuffix) - })...) - return errors.New(msg) - } - default: - logrus.Warnf("unsupported depends_on condition: %s", config.Condition) - return nil - } - } + return s.waitDependency(ctx, dep, config, waitingFor) }) } err := eg.Wait() @@ -263,6 +188,105 @@ func (s *composeService) waitDependencies(ctx context.Context, project *types.Pr return err } +// waitDependency polls the dependency's containers until its depends_on +// condition is satisfied (done), definitively failed (err), or ctx is +// cancelled. Each check reports (done, err): (false, nil) means keep polling. +func (s *composeService) waitDependency(ctx context.Context, dep string, config types.ServiceDependency, waitingFor Containers) error { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-ticker.C: + case <-ctx.Done(): + return nil + } + var ( + done bool + err error + ) + switch config.Condition { + case ServiceConditionRunningOrHealthy: + done, err = s.checkDependencyRunningOrHealthy(ctx, dep, config, waitingFor) + case types.ServiceConditionHealthy: + done, err = s.checkDependencyHealthy(ctx, dep, config, waitingFor) + case types.ServiceConditionCompletedSuccessfully: + done, err = s.checkDependencyCompleted(ctx, dep, config, waitingFor) + default: + logrus.Warnf("unsupported depends_on condition: %s", config.Condition) + return nil + } + if done || err != nil { + return err + } + } +} + +func (s *composeService) checkDependencyRunningOrHealthy(ctx context.Context, dep string, config types.ServiceDependency, waitingFor Containers) (bool, error) { + isHealthy, err := s.isServiceHealthy(ctx, waitingFor, true) + if err != nil { + if !config.Required { + s.events.On(containerReasonEvents(waitingFor, skippedEvent, + fmt.Sprintf("optional dependency %q is not running or is unhealthy", dep))...) + logrus.Warnf("optional dependency %q is not running or is unhealthy: %s", dep, err.Error()) + return true, nil + } + return false, err + } + if isHealthy { + s.events.On(containerEvents(waitingFor, healthy)...) + } + return isHealthy, nil +} + +func (s *composeService) checkDependencyHealthy(ctx context.Context, dep string, config types.ServiceDependency, waitingFor Containers) (bool, error) { + isHealthy, err := s.isServiceHealthy(ctx, waitingFor, false) + if err != nil { + if !config.Required { + s.events.On(containerReasonEvents(waitingFor, skippedEvent, + fmt.Sprintf("optional dependency %q failed to start", dep))...) + logrus.Warnf("optional dependency %q failed to start: %s", dep, err.Error()) + return true, nil + } + s.events.On(containerEvents(waitingFor, func(s string) api.Resource { + return errorEventf(s, "dependency %s failed to start", dep) + })...) + return false, fmt.Errorf("dependency failed to start: %w", err) + } + if isHealthy { + s.events.On(containerEvents(waitingFor, healthy)...) + } + return isHealthy, nil +} + +func (s *composeService) checkDependencyCompleted(ctx context.Context, dep string, config types.ServiceDependency, waitingFor Containers) (bool, error) { + isExited, code, err := s.isServiceCompleted(ctx, waitingFor) + if err != nil { + return false, err + } + if !isExited { + return false, nil + } + if code == 0 { + s.events.On(containerEvents(waitingFor, exited)...) + return true, nil + } + + messageSuffix := fmt.Sprintf("%q didn't complete successfully: exit %d", dep, code) + if !config.Required { + // optional -> mark as skipped & don't propagate error + s.events.On(containerReasonEvents(waitingFor, skippedEvent, + fmt.Sprintf("optional dependency %s", messageSuffix))...) + logrus.Warnf("optional dependency %s", messageSuffix) + return true, nil + } + + msg := fmt.Sprintf("service %s", messageSuffix) + s.events.On(containerEvents(waitingFor, func(s string) api.Resource { + return errorEventf(s, "service %s", messageSuffix) + })...) + return false, errors.New(msg) +} + func shouldWaitForDependency(serviceName string, dependencyConfig types.ServiceDependency, project *types.Project) (bool, error) { if dependencyConfig.Condition == types.ServiceConditionStarted { // already managed by InDependencyOrder diff --git a/pkg/compose/create.go b/pkg/compose/create.go index e3a86b6637..229bed9c88 100644 --- a/pkg/compose/create.go +++ b/pkg/compose/create.go @@ -473,15 +473,10 @@ func getAliases(project *types.Project, service types.ServiceConfig, serviceInde return aliases } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func createEndpointSettings(p *types.Project, service types.ServiceConfig, serviceIndex int, networkKey string, links []string, useNetworkAliases bool) (*network.EndpointSettings, error) { - const ifname = "com.docker.network.endpoint.ifname" - config := service.Networks[networkKey] - var ipam *network.EndpointIPAMConfig var ( + ipam *network.EndpointIPAMConfig ipv4Address netip.Addr ipv6Address netip.Addr macAddress string @@ -490,46 +485,12 @@ func createEndpointSettings(p *types.Project, service types.ServiceConfig, servi ) if config != nil { var err error - if config.Ipv4Address != "" { - ipv4Address, err = netip.ParseAddr(config.Ipv4Address) - if err != nil { - return nil, fmt.Errorf("invalid IPv4 address: %w", err) - } - } - if config.Ipv6Address != "" { - ipv6Address, err = netip.ParseAddr(config.Ipv6Address) - if err != nil { - return nil, fmt.Errorf("invalid IPv6 address: %w", err) - } - } - var linkLocalIPs []netip.Addr - for _, link := range config.LinkLocalIPs { - if link == "" { - continue - } - llIP, err := netip.ParseAddr(link) - if err != nil { - return nil, fmt.Errorf("invalid link-local IP: %w", err) - } - linkLocalIPs = append(linkLocalIPs, llIP) - } - - ipam = &network.EndpointIPAMConfig{ - IPv4Address: ipv4Address.Unmap(), - IPv6Address: ipv6Address, - LinkLocalIPs: linkLocalIPs, + ipam, ipv4Address, ipv6Address, err = parseEndpointIPAM(config) + if err != nil { + return nil, err } macAddress = config.MacAddress - driverOpts = config.DriverOpts - if config.InterfaceName != "" { - if driverOpts == nil { - driverOpts = map[string]string{} - } - if name, ok := driverOpts[ifname]; ok && name != config.InterfaceName { - logrus.Warnf("ignoring services.%s.networks.%s.interface_name as %s driver_opts is already declared", service.Name, networkKey, ifname) - } - driverOpts[ifname] = config.InterfaceName - } + driverOpts = endpointDriverOpts(service, networkKey, config) gwPriority = config.GatewayPriority } var ma network.HardwareAddr @@ -553,6 +514,63 @@ func createEndpointSettings(p *types.Project, service types.ServiceConfig, servi }, nil } +// parseEndpointIPAM parses the static addresses configured for an endpoint +func parseEndpointIPAM(config *types.ServiceNetworkConfig) (*network.EndpointIPAMConfig, netip.Addr, netip.Addr, error) { + var ( + ipv4Address netip.Addr + ipv6Address netip.Addr + err error + ) + if config.Ipv4Address != "" { + ipv4Address, err = netip.ParseAddr(config.Ipv4Address) + if err != nil { + return nil, ipv4Address, ipv6Address, fmt.Errorf("invalid IPv4 address: %w", err) + } + } + if config.Ipv6Address != "" { + ipv6Address, err = netip.ParseAddr(config.Ipv6Address) + if err != nil { + return nil, ipv4Address, ipv6Address, fmt.Errorf("invalid IPv6 address: %w", err) + } + } + var linkLocalIPs []netip.Addr + for _, link := range config.LinkLocalIPs { + if link == "" { + continue + } + llIP, err := netip.ParseAddr(link) + if err != nil { + return nil, ipv4Address, ipv6Address, fmt.Errorf("invalid link-local IP: %w", err) + } + linkLocalIPs = append(linkLocalIPs, llIP) + } + + ipam := &network.EndpointIPAMConfig{ + IPv4Address: ipv4Address.Unmap(), + IPv6Address: ipv6Address, + LinkLocalIPs: linkLocalIPs, + } + return ipam, ipv4Address, ipv6Address, nil +} + +// endpointDriverOpts merges interface_name into the endpoint driver_opts +func endpointDriverOpts(service types.ServiceConfig, networkKey string, config *types.ServiceNetworkConfig) types.Options { + const ifname = "com.docker.network.endpoint.ifname" + + driverOpts := config.DriverOpts + if config.InterfaceName == "" { + return driverOpts + } + if driverOpts == nil { + driverOpts = map[string]string{} + } + if name, ok := driverOpts[ifname]; ok && name != config.InterfaceName { + logrus.Warnf("ignoring services.%s.networks.%s.interface_name as %s driver_opts is already declared", service.Name, networkKey, ifname) + } + driverOpts[ifname] = config.InterfaceName + return driverOpts +} + // copy/pasted from https://github.com/docker/cli/blob/9de1b162f/cli/command/container/opts.go#L673-L697 + RelativePath // TODO find so way to share this code with docker/cli func parseSecurityOpts(p *types.Project, securityOpts []string) ([]string, bool, error) { @@ -933,9 +951,6 @@ func getDependentServiceFromMode(mode string) string { return "" } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) buildContainerVolumes( ctx context.Context, p types.Project, @@ -956,46 +971,70 @@ func (s *composeService) buildContainerVolumes( // `Mount` is preferred but does not offer option to created host path if missing // so `Bind` API is used here with raw volume string // see https://github.com/moby/moby/issues/43483 - v := findVolumeByTarget(service.Volumes, m.Target) - if v != nil { - if v.Type != types.VolumeTypeBind { - v.Source = m.Source - } - if !bindRequiresMountAPI(v.Bind) { - source := m.Source - if vol := findVolumeByName(p.Volumes, m.Source); vol != nil { - source = m.Source - } - binds = append(binds, toBindString(source, v)) - continue - } + if bind, ok := bindStringForMount(service, m); ok { + binds = append(binds, bind) + continue } case mount.TypeVolume: - v := findVolumeByTarget(service.Volumes, m.Target) - vol := findVolumeByName(p.Volumes, m.Source) - if v != nil && vol != nil { - // Prefer the bind API if no advanced option is used, to preserve backward compatibility - if !volumeRequiresMountAPI(v.Volume) { - binds = append(binds, toBindString(vol.Name, v)) - continue - } + if bind, ok := volumeBindString(p, service, m); ok { + binds = append(binds, bind) + continue } case mount.TypeImage: - // The daemon validates image mounts against the negotiated API version - // from the request path, not the server's own max version. - version, err := s.RuntimeAPIVersion(ctx) + err := s.checkImageMountSupported(ctx) if err != nil { return nil, nil, err } - if versions.LessThan(version, apiVersion148) { - return nil, nil, fmt.Errorf("volume with type=image require Docker Engine %s or later", dockerEngineV28) - } } mounts = append(mounts, m) } return binds, mounts, nil } +// bindStringForMount returns the legacy Bind-API string for a bind mount +// which doesn't require the Mount API +func bindStringForMount(service types.ServiceConfig, m mount.Mount) (string, bool) { + v := findVolumeByTarget(service.Volumes, m.Target) + if v == nil { + return "", false + } + if v.Type != types.VolumeTypeBind { + v.Source = m.Source + } + if bindRequiresMountAPI(v.Bind) { + return "", false + } + return toBindString(m.Source, v), true +} + +// volumeBindString returns the legacy Bind-API string for a volume mount +// without advanced options, preferred to preserve backward compatibility +func volumeBindString(p types.Project, service types.ServiceConfig, m mount.Mount) (string, bool) { + v := findVolumeByTarget(service.Volumes, m.Target) + vol := findVolumeByName(p.Volumes, m.Source) + if v == nil || vol == nil { + return "", false + } + if volumeRequiresMountAPI(v.Volume) { + return "", false + } + return toBindString(vol.Name, v), true +} + +// checkImageMountSupported verifies the negotiated API version supports image +// mounts. The daemon validates image mounts against the negotiated API +// version from the request path, not the server's own max version. +func (s *composeService) checkImageMountSupported(ctx context.Context) error { + version, err := s.RuntimeAPIVersion(ctx) + if err != nil { + return err + } + if versions.LessThan(version, apiVersion148) { + return fmt.Errorf("volume with type=image require Docker Engine %s or later", dockerEngineV28) + } + return nil +} + func toBindString(name string, v *types.ServiceVolumeConfig) string { access := "rw" if v.ReadOnly { diff --git a/pkg/compose/logs.go b/pkg/compose/logs.go index 08b520ce95..5bacaf76be 100644 --- a/pkg/compose/logs.go +++ b/pkg/compose/logs.go @@ -31,50 +31,21 @@ import ( "github.com/docker/compose/v5/pkg/utils" ) -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) Logs( ctx context.Context, projectName string, consumer api.LogConsumer, options api.LogOptions, ) error { - var containers Containers - var err error - - if options.Index > 0 { - ctr, err := s.getSpecifiedContainer(ctx, projectName, oneOffExclude, true, options.Services[0], options.Index) - if err != nil { - return err - } - containers = append(containers, ctr) - } else { - containers, err = s.getContainers(ctx, projectName, oneOffExclude, true, options.Services...) - if err != nil { - return err - } - } - - if options.Project != nil && len(options.Services) == 0 { - // we run with an explicit compose.yaml, so only consider services defined in this file - options.Services = options.Project.ServiceNames() - containers = containers.filter(isService(options.Services...)) + containers, err := s.selectLogsContainers(ctx, projectName, &options) + if err != nil { + return err } eg, ctx := errgroup.WithContext(ctx) for _, ctr := range containers { eg.Go(func() error { - res, err := s.apiClient().ContainerInspect(ctx, ctr.ID, client.ContainerInspectOptions{}) - if err != nil { - return err - } - err = s.doLogContainer(ctx, consumer, getContainerNameWithoutProject(ctr), res.Container, options) - if errdefs.IsNotImplemented(err) { - logrus.Warnf("Can't retrieve logs for %q: %s", getCanonicalContainerName(ctr), err.Error()) - return nil - } - return err + return s.logContainer(ctx, consumer, ctr, options) }) } @@ -88,29 +59,7 @@ func (s *composeService) Logs( monitor.withServices(options.Project.ServiceNames()) } monitor.withListener(printer.HandleEvent) - monitor.withListener(func(event api.ContainerEvent) { - if event.Type == api.ContainerEventStarted { - eg.Go(func() error { - res, err := s.apiClient().ContainerInspect(ctx, event.ID, client.ContainerInspectOptions{}) - if err != nil { - return err - } - - err = s.doLogContainer(ctx, consumer, event.Source, res.Container, api.LogOptions{ - Follow: options.Follow, - Since: res.Container.State.StartedAt, - Until: options.Until, - Tail: options.Tail, - Timestamps: options.Timestamps, - }) - if errdefs.IsNotImplemented(err) { - // ignore - return nil - } - return err - }) - } - }) + monitor.withListener(s.followStartedContainersLogs(ctx, eg, consumer, options)) eg.Go(func() error { // pass ctx so monitor will immediately stop on SIGINT return monitor.Start(ctx) @@ -120,6 +69,73 @@ func (s *composeService) Logs( return eg.Wait() } +// selectLogsContainers returns the containers to stream logs from, per the +// requested services, container index, and project +func (s *composeService) selectLogsContainers(ctx context.Context, projectName string, options *api.LogOptions) (Containers, error) { + if options.Index > 0 { + ctr, err := s.getSpecifiedContainer(ctx, projectName, oneOffExclude, true, options.Services[0], options.Index) + if err != nil { + return nil, err + } + return Containers{ctr}, nil + } + containers, err := s.getContainers(ctx, projectName, oneOffExclude, true, options.Services...) + if err != nil { + return nil, err + } + if options.Project != nil && len(options.Services) == 0 { + // we run with an explicit compose.yaml, so only consider services defined in this file + options.Services = options.Project.ServiceNames() + containers = containers.filter(isService(options.Services...)) + } + return containers, nil +} + +// logContainer streams a container's logs, warning when its logging driver +// doesn't support reading logs +func (s *composeService) logContainer(ctx context.Context, consumer api.LogConsumer, ctr container.Summary, options api.LogOptions) error { + res, err := s.apiClient().ContainerInspect(ctx, ctr.ID, client.ContainerInspectOptions{}) + if err != nil { + return err + } + err = s.doLogContainer(ctx, consumer, getContainerNameWithoutProject(ctr), res.Container, options) + if errdefs.IsNotImplemented(err) { + logrus.Warnf("Can't retrieve logs for %q: %s", getCanonicalContainerName(ctr), err.Error()) + return nil + } + return err +} + +// followStartedContainersLogs streams the logs of containers (re)started +// while following, ignoring those whose logging driver doesn't support +// reading logs +func (s *composeService) followStartedContainersLogs(ctx context.Context, eg *errgroup.Group, consumer api.LogConsumer, options api.LogOptions) api.ContainerEventListener { + return func(event api.ContainerEvent) { + if event.Type != api.ContainerEventStarted { + return + } + eg.Go(func() error { + res, err := s.apiClient().ContainerInspect(ctx, event.ID, client.ContainerInspectOptions{}) + if err != nil { + return err + } + + err = s.doLogContainer(ctx, consumer, event.Source, res.Container, api.LogOptions{ + Follow: options.Follow, + Since: res.Container.State.StartedAt, + Until: options.Until, + Tail: options.Tail, + Timestamps: options.Timestamps, + }) + if errdefs.IsNotImplemented(err) { + // ignore + return nil + } + return err + }) + } +} + 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, client.ContainerLogsOptions{ ShowStdout: true, diff --git a/pkg/compose/monitor.go b/pkg/compose/monitor.go index 461b8643e0..50092faed3 100644 --- a/pkg/compose/monitor.go +++ b/pkg/compose/monitor.go @@ -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]) diff --git a/pkg/compose/monitor_test.go b/pkg/compose/monitor_test.go new file mode 100644 index 0000000000..54b9be8922 --- /dev/null +++ b/pkg/compose/monitor_test.go @@ -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 "--" name to "-" + 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") +} diff --git a/pkg/compose/ps.go b/pkg/compose/ps.go index af79619f50..f3b70d0a7e 100644 --- a/pkg/compose/ps.go +++ b/pkg/compose/ps.go @@ -28,9 +28,6 @@ import ( "github.com/docker/compose/v5/pkg/api" ) -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) Ps(ctx context.Context, projectName string, options api.PsOptions) ([]api.ContainerSummary, error) { projectName = strings.ToLower(projectName) oneOff := oneOffExclude @@ -49,88 +46,110 @@ func (s *composeService) Ps(ctx context.Context, projectName string, options api eg, ctx := errgroup.WithContext(ctx) for i, ctr := range containers { eg.Go(func() error { - publishers := make([]api.PortPublisher, len(ctr.Ports)) - sort.Slice(ctr.Ports, func(i, j int) bool { - return ctr.Ports[i].PrivatePort < ctr.Ports[j].PrivatePort - }) - for i, p := range ctr.Ports { - var url string - if p.IP.IsValid() { - url = p.IP.String() - } - publishers[i] = api.PortPublisher{ - URL: url, // TODO(thaJeztah); change this to a netip.Addr ?? - TargetPort: int(p.PrivatePort), - PublishedPort: int(p.PublicPort), - Protocol: p.Type, - } - } + var err error + summary[i], err = s.containerSummary(ctx, ctr) + return err + }) + } + return summary, eg.Wait() +} - inspect, err := s.apiClient().ContainerInspect(ctx, ctr.ID, client.ContainerInspectOptions{}) - if err != nil { - return err - } +// containerSummary builds the api summary for a container, inspecting it to +// retrieve its health and exit code +func (s *composeService) containerSummary(ctx context.Context, ctr container.Summary) (api.ContainerSummary, error) { + inspect, err := s.apiClient().ContainerInspect(ctx, ctr.ID, client.ContainerInspectOptions{}) + if err != nil { + return api.ContainerSummary{}, err + } + health, exitCode := containerHealthAndExitCode(inspect) + mounts, localVolumes := containerMounts(ctr) + return api.ContainerSummary{ + ID: ctr.ID, + Name: getCanonicalContainerName(ctr), + Names: ctr.Names, + Image: ctr.Image, + Project: ctr.Labels[api.ProjectLabel], + Service: ctr.Labels[api.ServiceLabel], + Command: ctr.Command, + State: ctr.State, + Status: ctr.Status, + Created: ctr.Created, + Labels: ctr.Labels, + SizeRw: ctr.SizeRw, + SizeRootFs: ctr.SizeRootFs, + Mounts: mounts, + LocalVolumes: localVolumes, + Networks: containerNetworks(ctr), + Health: health, + ExitCode: exitCode, + Publishers: containerPublishers(ctr), + }, nil +} - var ( - health container.HealthStatus - exitCode int - ) - if inspect.Container.State != nil { - switch inspect.Container.State.Status { - case container.StateRunning: - if inspect.Container.State.Health != nil { - health = inspect.Container.State.Health.Status - } - case container.StateExited, container.StateDead: - exitCode = inspect.Container.State.ExitCode - } - } +func containerPublishers(ctr container.Summary) []api.PortPublisher { + sort.Slice(ctr.Ports, func(i, j int) bool { + return ctr.Ports[i].PrivatePort < ctr.Ports[j].PrivatePort + }) + publishers := make([]api.PortPublisher, len(ctr.Ports)) + for i, p := range ctr.Ports { + var url string + if p.IP.IsValid() { + url = p.IP.String() + } + publishers[i] = api.PortPublisher{ + URL: url, // TODO(thaJeztah); change this to a netip.Addr ?? + TargetPort: int(p.PrivatePort), + PublishedPort: int(p.PublicPort), + Protocol: p.Type, + } + } + return publishers +} - var ( - local int - mounts []string - ) - for _, m := range ctr.Mounts { - name := m.Name - if name == "" { - name = m.Source - } - if m.Driver == "local" { - local++ - } - mounts = append(mounts, name) - } +func containerHealthAndExitCode(inspect client.ContainerInspectResult) (container.HealthStatus, int) { + var ( + health container.HealthStatus + exitCode int + ) + state := inspect.Container.State + if state == nil { + return health, exitCode + } + switch state.Status { + case container.StateRunning: + if state.Health != nil { + health = state.Health.Status + } + case container.StateExited, container.StateDead: + exitCode = state.ExitCode + } + return health, exitCode +} - var networks []string - if ctr.NetworkSettings != nil { - for k := range ctr.NetworkSettings.Networks { - networks = append(networks, k) - } - } +func containerMounts(ctr container.Summary) ([]string, int) { + var ( + local int + mounts []string + ) + for _, m := range ctr.Mounts { + name := m.Name + if name == "" { + name = m.Source + } + if m.Driver == "local" { + local++ + } + mounts = append(mounts, name) + } + return mounts, local +} - summary[i] = api.ContainerSummary{ - ID: ctr.ID, - Name: getCanonicalContainerName(ctr), - Names: ctr.Names, - Image: ctr.Image, - Project: ctr.Labels[api.ProjectLabel], - Service: ctr.Labels[api.ServiceLabel], - Command: ctr.Command, - State: ctr.State, - Status: ctr.Status, - Created: ctr.Created, - Labels: ctr.Labels, - SizeRw: ctr.SizeRw, - SizeRootFs: ctr.SizeRootFs, - Mounts: mounts, - LocalVolumes: local, - Networks: networks, - Health: health, - ExitCode: exitCode, - Publishers: publishers, - } - return nil - }) +func containerNetworks(ctr container.Summary) []string { + var networks []string + if ctr.NetworkSettings != nil { + for k := range ctr.NetworkSettings.Networks { + networks = append(networks, k) + } } - return summary, eg.Wait() + return networks } diff --git a/pkg/compose/publish.go b/pkg/compose/publish.go index 5598a2cf70..e49522ed3c 100644 --- a/pkg/compose/publish.go +++ b/pkg/compose/publish.go @@ -33,6 +33,7 @@ import ( "github.com/DefangLabs/secret-detector/pkg/secrets" "github.com/compose-spec/compose-go/v2/loader" "github.com/compose-spec/compose-go/v2/types" + "github.com/containerd/containerd/v2/core/remotes" "github.com/distribution/reference" "github.com/opencontainers/go-digest" "github.com/opencontainers/image-spec/specs-go" @@ -51,9 +52,6 @@ func (s *composeService) Publish(ctx context.Context, project *types.Project, re }, "publish", s.events) } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) publish(ctx context.Context, project *types.Project, repository string, options api.PublishOptions) error { project, err := project.WithProfiles([]string{"*"}) if err != nil { @@ -89,78 +87,91 @@ func (s *composeService) publish(ctx context.Context, project *types.Project, re } } if !s.dryRun { - named, err := reference.ParseDockerRef(repository) + err = s.pushComposeArtifact(ctx, project, repository, layers, options) if err != nil { return err } + } + s.events.On(api.Resource{ + ID: repository, + Text: "published", + Status: api.Done, + }) + return nil +} - var insecureRegistries []string - if options.InsecureRegistry { - insecureRegistries = append(insecureRegistries, reference.Domain(named)) - } +// pushComposeArtifact pushes the compose artifact manifest to the repository, +// and the application image index when publishing a full application +func (s *composeService) pushComposeArtifact(ctx context.Context, project *types.Project, repository string, layers []v1.Descriptor, options api.PublishOptions) error { + named, err := reference.ParseDockerRef(repository) + if err != nil { + return err + } + + var insecureRegistries []string + if options.InsecureRegistry { + insecureRegistries = append(insecureRegistries, reference.Domain(named)) + } - resolver := oci.NewResolver(s.configFile(), desktop.ProxyTransportFor(ctx, s.apiClient()), insecureRegistries...) + resolver := oci.NewResolver(s.configFile(), desktop.ProxyTransportFor(ctx, s.apiClient()), insecureRegistries...) - descriptor, err := oci.PushManifest(ctx, resolver, named, layers, options.OCIVersion) + descriptor, err := oci.PushManifest(ctx, resolver, named, layers, options.OCIVersion) + if err != nil { + s.events.On(api.Resource{ + ID: repository, + Text: "publishing", + Status: api.Error, + }) + return err + } + + if options.Application { + return pushApplicationIndex(ctx, resolver, named, descriptor, project) + } + return nil +} + +// pushApplicationIndex pushes an image index referencing every service image, +// so the application can be pulled as a single artifact +func pushApplicationIndex(ctx context.Context, resolver remotes.Resolver, named reference.Named, descriptor v1.Descriptor, project *types.Project) error { + manifests := []v1.Descriptor{} + for _, service := range project.Services { + ref, err := reference.ParseDockerRef(service.Image) if err != nil { - s.events.On(api.Resource{ - ID: repository, - Text: "publishing", - Status: api.Error, - }) return err } - if options.Application { - manifests := []v1.Descriptor{} - for _, service := range project.Services { - ref, err := reference.ParseDockerRef(service.Image) - if err != nil { - return err - } - - manifest, err := oci.Copy(ctx, resolver, ref, named) - if err != nil { - return err - } - manifests = append(manifests, manifest) - } - - descriptor.Data = nil - index, err := json.Marshal(v1.Index{ - Versioned: specs.Versioned{SchemaVersion: 2}, - MediaType: v1.MediaTypeImageIndex, - Manifests: manifests, - Subject: &descriptor, - Annotations: map[string]string{ - "com.docker.compose.version": api.ComposeVersion, - }, - }) - if err != nil { - return err - } - imagesDescriptor := v1.Descriptor{ - MediaType: v1.MediaTypeImageIndex, - ArtifactType: oci.ComposeProjectArtifactType, - Digest: digest.FromString(string(index)), - Size: int64(len(index)), - Annotations: map[string]string{ - "com.docker.compose.version": api.ComposeVersion, - }, - Data: index, - } - err = oci.Push(ctx, resolver, reference.TrimNamed(named), imagesDescriptor) - if err != nil { - return err - } + manifest, err := oci.Copy(ctx, resolver, ref, named) + if err != nil { + return err } + manifests = append(manifests, manifest) } - s.events.On(api.Resource{ - ID: repository, - Text: "published", - Status: api.Done, + + descriptor.Data = nil + index, err := json.Marshal(v1.Index{ + Versioned: specs.Versioned{SchemaVersion: 2}, + MediaType: v1.MediaTypeImageIndex, + Manifests: manifests, + Subject: &descriptor, + Annotations: map[string]string{ + "com.docker.compose.version": api.ComposeVersion, + }, }) - return nil + if err != nil { + return err + } + imagesDescriptor := v1.Descriptor{ + MediaType: v1.MediaTypeImageIndex, + ArtifactType: oci.ComposeProjectArtifactType, + Digest: digest.FromString(string(index)), + Size: int64(len(index)), + Annotations: map[string]string{ + "com.docker.compose.version": api.ComposeVersion, + }, + Data: index, + } + return oci.Push(ctx, resolver, reference.TrimNamed(named), imagesDescriptor) } func (s *composeService) createLayers(ctx context.Context, project *types.Project, options api.PublishOptions) ([]v1.Descriptor, error) { @@ -686,12 +697,10 @@ func (s *composeService) checkForBindMount(project *types.Project) map[string][] return allFindings } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) checkForSensitiveData(ctx context.Context, project *types.Project) ([]secrets.DetectedSecret, error) { var allFindings []secrets.DetectedSecret scan := scanner.NewDefaultScanner() + // Check all compose files for _, file := range project.ComposeFiles { in, err := composeFileAsByteReader(ctx, file, project) @@ -705,48 +714,78 @@ func (s *composeService) checkForSensitiveData(ctx context.Context, project *typ } allFindings = append(allFindings, findings...) } + + // Check env files for _, service := range project.Services { - // Check env files - for _, envFile := range service.EnvFiles { - if _, statErr := os.Stat(envFile.Path); statErr != nil { - if !os.IsNotExist(statErr) { - return nil, fmt.Errorf("failed to access env file %s: %w", envFile.Path, statErr) - } - if envFile.Required { - return nil, fmt.Errorf("env file %s not found", envFile.Path) - } - continue - } - findings, err := scan.ScanFile(envFile.Path) - if err != nil { - return nil, fmt.Errorf("failed to scan env file %s: %w", envFile.Path, err) - } - allFindings = append(allFindings, findings...) + findings, err := scanEnvFiles(scan, service) + if err != nil { + return nil, err } + allFindings = append(allFindings, findings...) } // Check configs defined by files + configFiles := make([]string, 0, len(project.Configs)) for _, config := range project.Configs { - if config.File != "" { - findings, err := scan.ScanFile(config.File) - if err != nil { - return nil, fmt.Errorf("failed to scan config file %s: %w", config.File, err) - } - allFindings = append(allFindings, findings...) - } + configFiles = append(configFiles, config.File) } + findings, err := scanFiles(scan, "config", configFiles) + if err != nil { + return nil, err + } + allFindings = append(allFindings, findings...) // Check secrets defined by files + secretFiles := make([]string, 0, len(project.Secrets)) for _, secret := range project.Secrets { - if secret.File != "" { - findings, err := scan.ScanFile(secret.File) - if err != nil { - return nil, fmt.Errorf("failed to scan secret file %s: %w", secret.File, err) + secretFiles = append(secretFiles, secret.File) + } + findings, err = scanFiles(scan, "secret", secretFiles) + if err != nil { + return nil, err + } + allFindings = append(allFindings, findings...) + + return allFindings, nil +} + +// scanEnvFiles scans a service's env files for sensitive data; a missing env +// file is only an error when the service requires it +func scanEnvFiles(scan secrets.Scanner, service types.ServiceConfig) ([]secrets.DetectedSecret, error) { + var allFindings []secrets.DetectedSecret + for _, envFile := range service.EnvFiles { + if _, statErr := os.Stat(envFile.Path); statErr != nil { + if !os.IsNotExist(statErr) { + return nil, fmt.Errorf("failed to access env file %s: %w", envFile.Path, statErr) + } + if envFile.Required { + return nil, fmt.Errorf("env file %s not found", envFile.Path) } - allFindings = append(allFindings, findings...) + continue + } + findings, err := scan.ScanFile(envFile.Path) + if err != nil { + return nil, fmt.Errorf("failed to scan env file %s: %w", envFile.Path, err) } + allFindings = append(allFindings, findings...) } + return allFindings, nil +} +// scanFiles scans file-based resources (configs, secrets) for sensitive data, +// ignoring resources not defined by a file +func scanFiles(scan secrets.Scanner, kind string, paths []string) ([]secrets.DetectedSecret, error) { + var allFindings []secrets.DetectedSecret + for _, path := range paths { + if path == "" { + continue + } + findings, err := scan.ScanFile(path) + if err != nil { + return nil, fmt.Errorf("failed to scan %s file %s: %w", kind, path, err) + } + allFindings = append(allFindings, findings...) + } return allFindings, nil } diff --git a/pkg/compose/pull.go b/pkg/compose/pull.go index 486c2624f1..0852293039 100644 --- a/pkg/compose/pull.go +++ b/pkg/compose/pull.go @@ -50,9 +50,21 @@ func (s *composeService) Pull(ctx context.Context, project *types.Project, optio }, "pull", s.events) } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit +// imagePuller tracks the state of a `compose pull` run: the images already +// scheduled (a same image may back several services), per-service pull +// failures, and services whose image must be built as a fallback. +type imagePuller struct { + *composeService + project *types.Project + opts api.PullOptions + images map[string]api.ImageSummary + eg *errgroup.Group + scheduled map[string]string // image -> first service pulling it + pullErrors []error + mu sync.Mutex + mustBuild []string +} + func (s *composeService) pull(ctx context.Context, project *types.Project, opts api.PullOptions) error { images, _, err := s.getLocalImagesDigests(ctx, project) if err != nil { @@ -62,114 +74,132 @@ func (s *composeService) pull(ctx context.Context, project *types.Project, opts eg, ctx := errgroup.WithContext(ctx) eg.SetLimit(s.maxConcurrency) - var ( - mustBuild []string - pullErrors = make([]error, len(project.Services)) - imagesBeingPulled = map[string]string{} - ) + p := &imagePuller{ + composeService: s, + project: project, + opts: opts, + images: images, + eg: eg, + scheduled: map[string]string{}, + pullErrors: make([]error, len(project.Services)), + } + + err = p.pullServiceImages(ctx) + if err == nil { + err = p.pullHookImages(ctx) + } + if err != nil { + // join already-scheduled pulls before returning: bailing out with + // goroutines still in flight would leak them past pull()'s return + return errors.Join(err, eg.Wait()) + } + + err = eg.Wait() + + if len(p.mustBuild) > 0 { + logrus.Warnf("WARNING: Some service image(s) must be built from source by running:\n docker compose build %s", strings.Join(p.mustBuild, " ")) + } + + if err != nil { + return err + } + if opts.IgnoreFailures { + return nil + } + return errors.Join(p.pullErrors...) +} +// pullServiceImages schedules a pull for each service image which requires +// one, emitting a Skipped event for the others +func (p *imagePuller) pullServiceImages(ctx context.Context) error { i := 0 - for name, service := range project.Services { + for name, service := range p.project.Services { if service.Image == "" { - s.events.On(api.Resource{ - ID: name, - Status: api.Done, - Text: "Skipped", - Details: "No image to be pulled", - }) + p.eventSkippedPull(name, "No image to be pulled") continue } - pullRequired, skipReason, err := shouldPullImage(service, images) + pullRequired, skipReason, err := shouldPullImage(service, p.images) if err != nil { - // join already-scheduled pulls before returning: bailing out with - // goroutines still in flight would leak them past pull()'s return - return errors.Join(err, eg.Wait()) + return err } if !pullRequired { - s.events.On(api.Resource{ - ID: "Image " + service.Image, - Status: api.Done, - Text: "Skipped", - Details: skipReason, - }) + p.eventSkippedPull("Image "+service.Image, skipReason) continue } - - if service.Build != nil && opts.IgnoreBuildable { - s.events.On(api.Resource{ - ID: "Image " + service.Image, - Status: api.Done, - Text: "Skipped", - Details: "Image can be built", - }) + if service.Build != nil && p.opts.IgnoreBuildable { + p.eventSkippedPull("Image "+service.Image, "Image can be built") continue } - - if _, ok := imagesBeingPulled[service.Image]; ok { + if _, ok := p.scheduled[service.Image]; ok { continue } - - imagesBeingPulled[service.Image] = service.Name + p.scheduled[service.Image] = service.Name idx := i - eg.Go(func() error { - err := s.pullServiceImage(ctx, service, opts.Quiet, project.Environment["DOCKER_DEFAULT_PLATFORM"]) - if err != nil { - pullErrors[idx] = err - if service.Build != nil { - mustBuild = append(mustBuild, service.Name) - } - if !opts.IgnoreFailures && service.Build == nil { - if s.dryRun { - s.events.On(errorEventf("Image "+service.Image, - "error pulling image: %s", service.Image)) - } - // fail fast if image can't be pulled nor built - return err - } - } - return nil + p.eg.Go(func() error { + return p.runServicePull(ctx, idx, service) }) i++ } + return nil +} - // pre_start hook images run as ephemeral init containers with their own - // registry image. They have no pull policy of their own, so we inherit the - // parent service's policy for skip decisions — through the same - // shouldPullImage decision as the service image. Unlike the service - // image, a hook image can't be built, so `build` falls back to - // pull-if-missing instead of exempting it from pulling. - for name, service := range project.Services { +// runServicePull pulls a service image, recording the failure and whether the +// image could be built instead, per the fail-fast rules of `compose pull` +func (p *imagePuller) runServicePull(ctx context.Context, idx int, service types.ServiceConfig) error { + err := p.pullServiceImage(ctx, service, p.opts.Quiet, p.project.Environment["DOCKER_DEFAULT_PLATFORM"]) + if err == nil { + return nil + } + p.pullErrors[idx] = err + if service.Build != nil { + p.mu.Lock() + p.mustBuild = append(p.mustBuild, service.Name) + p.mu.Unlock() + } + if !p.opts.IgnoreFailures && service.Build == nil { + if p.dryRun { + p.events.On(errorEventf("Image "+service.Image, + "error pulling image: %s", service.Image)) + } + // fail fast if image can't be pulled nor built + return err + } + return nil +} + +// pullHookImages schedules pulls for pre_start hook images, which run as +// ephemeral init containers with their own registry image. They have no pull +// policy of their own, so we inherit the parent service's policy for skip +// decisions — through the same shouldPullImage decision as the service image. +// Unlike the service image, a hook image can't be built, so `build` falls +// back to pull-if-missing instead of exempting it from pulling. +func (p *imagePuller) pullHookImages(ctx context.Context) error { + for name, service := range p.project.Services { hookPolicy := service.PullPolicy if hookPolicy == types.PullPolicyBuild { hookPolicy = types.PullPolicyMissing } - for _, img := range api.GetDependentImages(service, project.Name) { - pullRequired, skipReason, err := shouldPullImage(types.ServiceConfig{Name: name, Image: img, PullPolicy: hookPolicy}, images) + for _, img := range api.GetDependentImages(service, p.project.Name) { + pullRequired, skipReason, err := shouldPullImage(types.ServiceConfig{Name: name, Image: img, PullPolicy: hookPolicy}, p.images) if err != nil { - // same as the service loop: never leave scheduled pulls unjoined - return errors.Join(err, eg.Wait()) + return err } if !pullRequired { if skipReason != "" { - s.events.On(api.Resource{ - ID: "Image " + img, - Status: api.Done, - Text: "Skipped", - Details: skipReason, - }) + p.eventSkippedPull("Image "+img, skipReason) } continue } - if _, ok := imagesBeingPulled[img]; ok { + if _, ok := p.scheduled[img]; ok { continue } - imagesBeingPulled[img] = name + p.scheduled[img] = name hookService := types.ServiceConfig{Name: name, Image: img} - eg.Go(func() error { - err := s.pullServiceImage(ctx, hookService, opts.Quiet, project.Environment["DOCKER_DEFAULT_PLATFORM"]) - if err != nil && !opts.IgnoreFailures { + p.eg.Go(func() error { + err := p.pullServiceImage(ctx, hookService, p.opts.Quiet, p.project.Environment["DOCKER_DEFAULT_PLATFORM"]) + if err != nil && !p.opts.IgnoreFailures { // fail fast: a hook image can't be built as a fallback return err } @@ -177,20 +207,16 @@ func (s *composeService) pull(ctx context.Context, project *types.Project, opts }) } } + return nil +} - err = eg.Wait() - - if len(mustBuild) > 0 { - logrus.Warnf("WARNING: Some service image(s) must be built from source by running:\n docker compose build %s", strings.Join(mustBuild, " ")) - } - - if err != nil { - return err - } - if opts.IgnoreFailures { - return nil - } - return errors.Join(pullErrors...) +func (p *imagePuller) eventSkippedPull(id, details string) { + p.events.On(api.Resource{ + ID: id, + Status: api.Done, + Text: "Skipped", + Details: details, + }) } // shouldPullImage decides whether `compose pull` refreshes a service's image, diff --git a/pkg/compose/restart.go b/pkg/compose/restart.go index 587e57095b..461a0257c5 100644 --- a/pkg/compose/restart.go +++ b/pkg/compose/restart.go @@ -21,6 +21,7 @@ import ( "strings" "github.com/compose-spec/compose-go/v2/types" + "github.com/moby/moby/api/types/container" "github.com/moby/moby/client" "golang.org/x/sync/errgroup" @@ -34,27 +35,50 @@ func (s *composeService) Restart(ctx context.Context, projectName string, option }, "restart", s.events) } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) restart(ctx context.Context, projectName string, options api.RestartOptions) error { containers, err := s.getContainers(ctx, projectName, oneOffExclude, true) if err != nil { return err } + project, err := s.prepareRestartProject(ctx, containers, projectName, options) + if err != nil { + return err + } + + return InDependencyOrder(ctx, project, func(c context.Context, service string) error { + config := project.Services[service] + err := s.waitDependencies(ctx, project, service, config.DependsOn, containers, 0) + if err != nil { + return err + } + + eg, ctx := errgroup.WithContext(ctx) + for _, ctr := range containers.filter(isService(service)) { + eg.Go(func() error { + return s.restartContainer(ctx, project.Services[service], ctr, options) + }) + } + return eg.Wait() + }) +} + +// prepareRestartProject resolves the project restart applies to, restricted +// to the requested services and the depends_on relations with restart: true +func (s *composeService) prepareRestartProject(ctx context.Context, containers Containers, projectName string, options api.RestartOptions) (*types.Project, error) { project := options.Project + var err error if project == nil { project, err = s.getProjectWithResources(ctx, containers, projectName) if err != nil { - return err + return nil, err } } if options.NoDeps { project, err = project.WithSelectedServices(options.Services, types.IgnoreDependencies) if err != nil { - return err + return nil, err } } @@ -68,51 +92,41 @@ func (s *composeService) restart(ctx context.Context, projectName string, option return s, nil }) if err != nil { - return err + return nil, err } if len(options.Services) != 0 { project, err = project.WithSelectedServices(options.Services, types.IncludeDependents) if err != nil { - return err + return nil, err } } + return project, nil +} - return InDependencyOrder(ctx, project, func(c context.Context, service string) error { - config := project.Services[service] - err = s.waitDependencies(ctx, project, service, config.DependsOn, containers, 0) +// restartContainer restarts a container, running its pre_stop and post_start +// hooks around the restart +func (s *composeService) restartContainer(ctx context.Context, def types.ServiceConfig, ctr container.Summary, options api.RestartOptions) error { + for _, hook := range def.PreStop { + err := s.runHook(ctx, ctr, def, hook, nil) if err != nil { return err } - - eg, ctx := errgroup.WithContext(ctx) - for _, ctr := range containers.filter(isService(service)) { - eg.Go(func() error { - def := project.Services[service] - for _, hook := range def.PreStop { - err = s.runHook(ctx, ctr, def, hook, nil) - if err != nil { - return err - } - } - eventName := getContainerProgressName(ctr) - s.events.On(newEvent(eventName, api.Working, api.StatusRestarting)) - _, err = s.apiClient().ContainerRestart(ctx, ctr.ID, client.ContainerRestartOptions{ - Timeout: utils.DurationSecondToInt(options.Timeout), - }) - if err != nil { - return err - } - s.events.On(newEvent(eventName, api.Done, api.StatusStarted)) - for _, hook := range def.PostStart { - err = s.runHook(ctx, ctr, def, hook, nil) - if err != nil { - return err - } - } - return nil - }) - } - return eg.Wait() + } + eventName := getContainerProgressName(ctr) + s.events.On(newEvent(eventName, api.Working, api.StatusRestarting)) + _, err := s.apiClient().ContainerRestart(ctx, ctr.ID, client.ContainerRestartOptions{ + Timeout: utils.DurationSecondToInt(options.Timeout), }) + if err != nil { + return err + } + s.events.On(newEvent(eventName, api.Done, api.StatusStarted)) + for _, hook := range def.PostStart { + err := s.runHook(ctx, ctr, def, hook, nil) + if err != nil { + return err + } + } + return nil } diff --git a/pkg/compose/up.go b/pkg/compose/up.go index 074ea80975..0beb363735 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -41,9 +41,6 @@ import ( "github.com/docker/compose/v5/pkg/api" ) -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) Up(ctx context.Context, project *types.Project, options api.UpOptions) error { err := Run(ctx, tracing.SpanWrapFunc("project/up", tracing.ProjectOptions(ctx, project), func(ctx context.Context) error { err := s.create(ctx, project, options.Create) @@ -66,7 +63,31 @@ func (s *composeService) Up(ctx context.Context, project *types.Project, options _, _ = fmt.Fprintln(s.stdout(), "end of 'compose up' output, interactive run is not supported in dry-run mode") return err } + return s.runInteractiveUp(ctx, project, options) +} + +// upSession carries the state shared between the goroutines driving an +// interactive `compose up` once services are created: the errgroup running +// them, the collected errors, and the application exit status. +type upSession struct { + *composeService + project *types.Project + options api.UpOptions + printer logPrinter + watcher *Watcher + menu *formatter.LogKeyboard + globalCtx context.Context + cancel context.CancelFunc + + signalChan chan os.Signal + isTerminated atomic.Bool + eg errgroup.Group + mu sync.Mutex + errs []error + exitCode int +} +func (s *composeService) runInteractiveUp(ctx context.Context, project *types.Project, options api.UpOptions) error { // if we get a second signal during shutdown, we kill the services // immediately, so the channel needs to have sufficient capacity or // we might miss a signal while setting up the second channel read @@ -74,29 +95,15 @@ func (s *composeService) Up(ctx context.Context, project *types.Project, options signalChan := make(chan os.Signal, 2) signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM) defer signal.Stop(signalChan) - var isTerminated atomic.Bool - - var ( - logConsumer = options.Start.Attach - navigationMenu *formatter.LogKeyboard - kEvents <-chan keyboard.KeyEvent - ) - if options.Start.NavigationMenu { - kEvents, err = keyboard.GetKeys(100) - if err != nil { - logrus.Warnf("could not start menu, an error occurred while starting: %v", err) - options.Start.NavigationMenu = false - } else { - defer keyboard.Close() //nolint:errcheck - isDockerDesktopActive, err := s.isDesktopIntegrationActive(ctx) - if err != nil { - return err - } - isLogsViewEnabled := s.isDesktopFeatureActive(ctx, desktop.FeatureLogsTab) - tracing.KeyboardMetrics(ctx, options.Start.NavigationMenu, isDockerDesktopActive, isLogsViewEnabled) - navigationMenu = formatter.NewKeyboardManager(isDockerDesktopActive, isLogsViewEnabled, signalChan) - logConsumer = navigationMenu.Decorate(logConsumer) - } + + logConsumer := options.Start.Attach + navigationMenu, kEvents, err := s.setupNavigationMenu(ctx, &options, signalChan) + if err != nil { + return err + } + if navigationMenu != nil { + defer keyboard.Close() //nolint:errcheck + logConsumer = navigationMenu.Decorate(logConsumer) } watcher, err := NewWatcher(project, options, s.watch, logConsumer) @@ -108,8 +115,6 @@ func (s *composeService) Up(ctx context.Context, project *types.Project, options navigationMenu.EnableWatch(options.Start.Watch, watcher) } - printer := newLogPrinter(logConsumer) - // global context to handle canceling goroutines globalCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -118,79 +123,27 @@ func (s *composeService) Up(ctx context.Context, project *types.Project, options navigationMenu.EnableDetach(cancel) } - var ( - eg errgroup.Group - mu sync.Mutex - errs []error - ) - - appendErr := func(err error) { - if err != nil { - mu.Lock() - errs = append(errs, err) - mu.Unlock() - } + u := &upSession{ + composeService: s, + project: project, + options: options, + printer: newLogPrinter(logConsumer), + watcher: watcher, + menu: navigationMenu, + globalCtx: globalCtx, + cancel: cancel, + signalChan: signalChan, } - eg.Go(func() error { - first := true - gracefulTeardown := func() { - first = false - s.events.On(newEvent(api.ResourceCompose, api.Working, api.StatusStopping, "Gracefully Stopping... press Ctrl+C again to force")) - eg.Go(func() error { - err = s.stop(context.WithoutCancel(globalCtx), project.Name, api.StopOptions{ - Services: options.Create.Services, - Project: project, - }, printer.HandleEvent) - appendErr(err) - return nil - }) - isTerminated.Store(true) - } - - for { - select { - case <-globalCtx.Done(): - if watcher != nil { - return watcher.Stop() - } - return nil - case <-ctx.Done(): - if first { - gracefulTeardown() - } - case <-signalChan: - if first { - _ = keyboard.Close() - gracefulTeardown() - break - } - eg.Go(func() error { - err := s.kill(context.WithoutCancel(globalCtx), project.Name, api.KillOptions{ - Services: options.Create.Services, - Project: project, - All: true, - }) - // Ignore errors indicating that some of the containers were already stopped or removed. - if errdefs.IsNotFound(err) || errdefs.IsConflict(err) || errors.Is(err, api.ErrNoResources) { - return nil - } - - appendErr(err) - return nil - }) - return nil - case event := <-kEvents: - navigationMenu.HandleKeyEvents(globalCtx, event, project, options) - } - } + u.eg.Go(func() error { + return u.runEventLoop(ctx, kEvents) }) if options.Start.Watch && watcher != nil { if err := watcher.Start(globalCtx); err != nil { // cancel the global context to terminate background goroutines cancel() - _ = eg.Wait() + _ = u.eg.Wait() return err } } @@ -202,105 +155,214 @@ func (s *composeService) Up(ctx context.Context, project *types.Project, options // Start.AttachTo have been already curated with only the services to monitor monitor.withServices(options.Start.AttachTo) } - monitor.withListener(printer.HandleEvent) + monitor.withListener(u.printer.HandleEvent) - var exitCode int if options.Start.OnExit != api.CascadeIgnore { - once := true - // detect first container to exit to trigger application shutdown - monitor.withListener(func(event api.ContainerEvent) { - if once && event.Type == api.ContainerEventExited { - if options.Start.OnExit == api.CascadeFail && event.ExitCode == 0 { - return - } - once = false - exitCode = event.ExitCode - s.events.On(newEvent(api.ResourceCompose, api.Working, api.StatusStopping, "Aborting on container exit...")) - eg.Go(func() error { - err = s.stop(context.WithoutCancel(globalCtx), project.Name, api.StopOptions{ - Services: options.Create.Services, - Project: project, - }, printer.HandleEvent) - appendErr(err) - return nil - }) - } - }) + monitor.withListener(u.stopOnFirstExit()) } - if options.Start.ExitCodeFrom != "" { - once := true - // capture exit code from first container to exit with selected service - monitor.withListener(func(event api.ContainerEvent) { - if once && event.Type == api.ContainerEventExited && event.Service == options.Start.ExitCodeFrom { - exitCode = event.ExitCode - once = false - } - }) + monitor.withListener(u.captureExitCodeFrom()) } - containers, err := s.attach(globalCtx, project, printer.HandleEvent, options.Start.AttachTo) + containers, err := s.attach(globalCtx, project, u.printer.HandleEvent, options.Start.AttachTo) if err != nil { cancel() - _ = eg.Wait() + _ = u.eg.Wait() return err } attached := make([]string, len(containers)) for i, ctr := range containers { attached[i] = ctr.ID } + monitor.withListener(u.followStartedContainers(attached)) - monitor.withListener(func(event api.ContainerEvent) { - if !shouldFollowStartEvent(event, attached, options.Start.AttachTo) { - return - } - eg.Go(func() error { - res, err := s.apiClient().ContainerInspect(globalCtx, event.ID, client.ContainerInspectOptions{}) - if err != nil { - appendErr(err) - return nil - } - - err = s.doLogContainer(globalCtx, options.Start.Attach, event.Source, res.Container, api.LogOptions{ - Follow: true, - Since: res.Container.State.StartedAt, - }) - if errdefs.IsNotImplemented(err) { - // container may be configured with logging_driver: none - // as container already started, we might miss the very first logs. But still better than none - err := s.doAttachContainer(globalCtx, event.Service, event.ID, event.Source, printer.HandleEvent) - appendErr(err) - return nil - } - appendErr(err) - return nil - }) - }) - - eg.Go(func() error { + u.eg.Go(func() error { err := monitor.Start(globalCtx) // cancel the global context to terminate signal-handler goroutines cancel() - appendErr(err) + u.appendErr(err) return nil }) // We use the parent context without cancellation as we manage sigterm to stop the stack - err = s.start(context.WithoutCancel(ctx), project.Name, options.Start, printer.HandleEvent) - if err != nil && !isTerminated.Load() { // Ignore error if the process is terminated + err = s.start(context.WithoutCancel(ctx), project.Name, options.Start, u.printer.HandleEvent) + if err != nil && !u.isTerminated.Load() { // Ignore error if the process is terminated cancel() - _ = eg.Wait() + _ = u.eg.Wait() return err } - _ = eg.Wait() - err = errors.Join(errs...) - if exitCode != 0 { + _ = u.eg.Wait() + err = errors.Join(u.errs...) + if u.exitCode != 0 { errMsg := "" if err != nil { errMsg = err.Error() } - return cli.StatusError{StatusCode: exitCode, Status: errMsg} + return cli.StatusError{StatusCode: u.exitCode, Status: errMsg} + } + return err +} + +// setupNavigationMenu initializes the interactive keyboard menu when enabled. +// It returns a nil menu when the menu is disabled, or when the keyboard can't +// be grabbed — then disabling the option. +func (s *composeService) setupNavigationMenu(ctx context.Context, options *api.UpOptions, signalChan chan os.Signal) (*formatter.LogKeyboard, <-chan keyboard.KeyEvent, error) { + if !options.Start.NavigationMenu { + return nil, nil, nil + } + kEvents, err := keyboard.GetKeys(100) + if err != nil { + logrus.Warnf("could not start menu, an error occurred while starting: %v", err) + options.Start.NavigationMenu = false + return nil, nil, nil + } + isDockerDesktopActive, err := s.isDesktopIntegrationActive(ctx) + if err != nil { + _ = keyboard.Close() + return nil, nil, err + } + isLogsViewEnabled := s.isDesktopFeatureActive(ctx, desktop.FeatureLogsTab) + tracing.KeyboardMetrics(ctx, options.Start.NavigationMenu, isDockerDesktopActive, isLogsViewEnabled) + return formatter.NewKeyboardManager(isDockerDesktopActive, isLogsViewEnabled, signalChan), kEvents, nil +} + +func (u *upSession) appendErr(err error) { + if err != nil { + u.mu.Lock() + u.errs = append(u.errs, err) + u.mu.Unlock() + } +} + +// runEventLoop reacts to cancellation, SIGINT/SIGTERM and keyboard input until +// the application terminates: a first interruption triggers a graceful stop, +// a second one kills the services. +func (u *upSession) runEventLoop(ctx context.Context, kEvents <-chan keyboard.KeyEvent) error { + first := true + gracefulTeardown := func() { + first = false + u.events.On(newEvent(api.ResourceCompose, api.Working, api.StatusStopping, "Gracefully Stopping... press Ctrl+C again to force")) + u.stopApplication() + u.isTerminated.Store(true) + } + + for { + select { + case <-u.globalCtx.Done(): + if u.watcher != nil { + return u.watcher.Stop() + } + return nil + case <-ctx.Done(): + if first { + gracefulTeardown() + } + case <-u.signalChan: + if first { + _ = keyboard.Close() + gracefulTeardown() + break + } + u.killApplication() + return nil + case event := <-kEvents: + u.menu.HandleKeyEvents(u.globalCtx, event, u.project, u.options) + } + } +} + +// stopApplication requests a graceful stop of the application services in +// background; the error is collected for the final report. +func (u *upSession) stopApplication() { + u.eg.Go(func() error { + err := u.stop(context.WithoutCancel(u.globalCtx), u.project.Name, api.StopOptions{ + Services: u.options.Create.Services, + Project: u.project, + }, u.printer.HandleEvent) + u.appendErr(err) + return nil + }) +} + +// killApplication kills the application services in background; the error is +// collected for the final report. +func (u *upSession) killApplication() { + u.eg.Go(func() error { + err := u.kill(context.WithoutCancel(u.globalCtx), u.project.Name, api.KillOptions{ + Services: u.options.Create.Services, + Project: u.project, + All: true, + }) + // Ignore errors indicating that some of the containers were already stopped or removed. + if errdefs.IsNotFound(err) || errdefs.IsConflict(err) || errors.Is(err, api.ErrNoResources) { + return nil + } + + u.appendErr(err) + return nil + }) +} + +// stopOnFirstExit detects the first container to exit — per the on-exit +// cascade policy — to trigger application shutdown and record the +// application exit code. +func (u *upSession) stopOnFirstExit() api.ContainerEventListener { + once := true + return func(event api.ContainerEvent) { + if !once || event.Type != api.ContainerEventExited { + return + } + if u.options.Start.OnExit == api.CascadeFail && event.ExitCode == 0 { + return + } + once = false + u.exitCode = event.ExitCode + u.events.On(newEvent(api.ResourceCompose, api.Working, api.StatusStopping, "Aborting on container exit...")) + u.stopApplication() + } +} + +// captureExitCodeFrom captures the exit code of the first container to exit +// for the service selected by --exit-code-from +func (u *upSession) captureExitCodeFrom() api.ContainerEventListener { + once := true + return func(event api.ContainerEvent) { + if once && event.Type == api.ContainerEventExited && event.Service == u.options.Start.ExitCodeFrom { + u.exitCode = event.ExitCode + once = false + } + } +} + +// followStartedContainers streams logs of containers (re)started after `up`, +// so they are followed like the initially attached ones. +func (u *upSession) followStartedContainers(attached []string) api.ContainerEventListener { + return func(event api.ContainerEvent) { + if !shouldFollowStartEvent(event, attached, u.options.Start.AttachTo) { + return + } + u.eg.Go(func() error { + u.appendErr(u.streamContainerLogs(event)) + return nil + }) + } +} + +func (u *upSession) streamContainerLogs(event api.ContainerEvent) error { + res, err := u.apiClient().ContainerInspect(u.globalCtx, event.ID, client.ContainerInspectOptions{}) + if err != nil { + return err + } + + err = u.doLogContainer(u.globalCtx, u.options.Start.Attach, event.Source, res.Container, api.LogOptions{ + Follow: true, + Since: res.Container.State.StartedAt, + }) + if errdefs.IsNotImplemented(err) { + // container may be configured with logging_driver: none + // as container already started, we might miss the very first logs. But still better than none + return u.doAttachContainer(u.globalCtx, event.Service, event.ID, event.Source, u.printer.HandleEvent) } return err } diff --git a/pkg/compose/viz_test.go b/pkg/compose/viz_test.go index d72d793a90..2ecbb5ab06 100644 --- a/pkg/compose/viz_test.go +++ b/pkg/compose/viz_test.go @@ -29,9 +29,6 @@ import ( "github.com/docker/compose/v5/pkg/mocks" ) -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func TestViz(t *testing.T) { project := types.Project{ Name: "viz-test", @@ -134,50 +131,14 @@ func TestViz(t *testing.T) { assert.Check(t, is.Contains(graphStr, "\n ")) assert.Check(t, !is.Contains(graphStr, "\n ")().Success(), graphStr) - // check digraph name - assert.Check(t, is.Contains(graphStr, "digraph \""+project.Name+"\"")) - - // check nodes - for _, service := range project.Services { - assert.Check(t, is.Contains(graphStr, "\""+service.Name+"\" [style=\"filled\"")) - } + assertVizGraphNodes(t, graphStr, project) // check node attributes assert.Check(t, !is.Contains(graphStr, "Networks")().Success()) assert.Check(t, !is.Contains(graphStr, "Image")().Success()) assert.Check(t, !is.Contains(graphStr, "Ports")().Success()) - // check edges that SHOULD exist in the generated graph - allowedEdges := make(map[string][]string) - for name, service := range project.Services { - allowed := make([]string, 0, len(service.DependsOn)) - for depName := range service.DependsOn { - allowed = append(allowed, depName) - } - allowedEdges[name] = allowed - } - for serviceName, dependencies := range allowedEdges { - for _, dependencyName := range dependencies { - assert.Check(t, is.Contains(graphStr, "\""+serviceName+"\" -> \""+dependencyName+"\"")) - } - } - - // check edges that SHOULD NOT exist in the generated graph - forbiddenEdges := make(map[string][]string) - for name, service := range project.Services { - forbiddenEdges[name] = make([]string, 0, len(project.ServiceNames())-len(service.DependsOn)) - for _, serviceName := range project.ServiceNames() { - _, edgeExists := service.DependsOn[serviceName] - if !edgeExists { - forbiddenEdges[name] = append(forbiddenEdges[name], serviceName) - } - } - } - for serviceName, forbiddenDeps := range forbiddenEdges { - for _, forbiddenDep := range forbiddenDeps { - assert.Check(t, !is.Contains(graphStr, "\""+serviceName+"\" -> \""+forbiddenDep+"\"")().Success()) - } - } + assertVizDependencyEdges(t, graphStr, project) }) t.Run("viz (with ports, networks and image)", func(t *testing.T) { @@ -193,13 +154,7 @@ func TestViz(t *testing.T) { assert.Check(t, is.Contains(graphStr, "\n\t")) assert.Check(t, !is.Contains(graphStr, "\n\t\t")().Success(), graphStr) - // check digraph name - assert.Check(t, is.Contains(graphStr, "digraph \""+project.Name+"\"")) - - // check nodes - for _, service := range project.Services { - assert.Check(t, is.Contains(graphStr, "\""+service.Name+"\" [style=\"filled\"")) - } + assertVizGraphNodes(t, graphStr, project) // check node attributes assert.Check(t, is.Contains(graphStr, "Networks")) @@ -218,3 +173,29 @@ func TestViz(t *testing.T) { } }) } + +// assertVizGraphNodes checks the digraph is named after the project and has a +// node per service +func assertVizGraphNodes(t *testing.T, graphStr string, project types.Project) { + t.Helper() + assert.Check(t, is.Contains(graphStr, "digraph \""+project.Name+"\"")) + for _, service := range project.Services { + assert.Check(t, is.Contains(graphStr, "\""+service.Name+"\" [style=\"filled\"")) + } +} + +// assertVizDependencyEdges checks the graph has an edge per depends_on +// relation, and none between independent services +func assertVizDependencyEdges(t *testing.T, graphStr string, project types.Project) { + t.Helper() + for name, service := range project.Services { + for _, other := range project.ServiceNames() { + edge := "\"" + name + "\" -> \"" + other + "\"" + if _, expected := service.DependsOn[other]; expected { + assert.Check(t, is.Contains(graphStr, edge)) + } else { + assert.Check(t, !is.Contains(graphStr, edge)().Success()) + } + } + } +} diff --git a/pkg/compose/watch.go b/pkg/compose/watch.go index 94d518c195..e75538364b 100644 --- a/pkg/compose/watch.go +++ b/pkg/compose/watch.go @@ -186,9 +186,6 @@ func (r watchRule) Matches(event watch.FileEvent) *sync.PathMapping { } } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) watch(ctx context.Context, project *types.Project, options api.WatchOptions) (func() error, error) { var err error if project, err = project.WithSelectedServices(options.Services); err != nil { @@ -218,47 +215,16 @@ func (s *composeService) watch(ctx context.Context, project *types.Project, opti continue } - for _, trigger := range config.Watch { - if trigger.Action == types.WatchActionRebuild { - if service.Build == nil { - return nil, fmt.Errorf("can't watch service %q with action %s without a build context", service.Name, types.WatchActionRebuild) - } - if options.Build == nil { - return nil, fmt.Errorf("--no-build is incompatible with watch action %s in service %s", types.WatchActionRebuild, service.Name) - } - // set the service to always be built - watch triggers `Up()` when it receives a rebuild event - service.PullPolicy = types.PullPolicyBuild - project.Services[serviceName] = service - } + service, err = prepareRebuildTriggers(project, serviceName, service, config, options) + if err != nil { + return nil, err } - for _, trigger := range config.Watch { - if isSync(trigger) && checkIfPathAlreadyBindMounted(trigger.Path, service.Volumes) { - logrus.Warnf("path '%s' also declared by a bind mount volume, this path won't be monitored!\n", trigger.Path) - continue - } else { - shouldInitialSync := trigger.InitialSync - - // Check legacy extension attribute for backward compatibility - if !shouldInitialSync { - var legacyInitialSync bool - success, err := trigger.Extensions.Get("x-initialSync", &legacyInitialSync) - if err == nil && success && legacyInitialSync { - shouldInitialSync = true - logrus.Warnf("x-initialSync is DEPRECATED, please use the official `initial_sync` attribute\n") - } - } - - if shouldInitialSync && isSync(trigger) { - // Need to check that initial files meant to be synced from the watch action are in the container - err := s.initialSync(ctx, project, service, trigger, syncer) - if err != nil { - return nil, err - } - } - } - paths = append(paths, trigger.Path) + triggerPaths, err := s.watchTriggerPaths(ctx, project, service, config, syncer) + if err != nil { + return nil, err } + paths = append(paths, triggerPaths...) serviceWatchRules, err := getWatchRules(config, service) if err != nil { @@ -295,6 +261,63 @@ func (s *composeService) watch(ctx context.Context, project *types.Project, opti }, nil } +// prepareRebuildTriggers validates rebuild watch actions and marks the +// service to always be built — watch triggers `Up()` when it receives a +// rebuild event. It returns the possibly-updated service config. +func prepareRebuildTriggers(project *types.Project, serviceName string, service types.ServiceConfig, config *types.DevelopConfig, options api.WatchOptions) (types.ServiceConfig, error) { + for _, trigger := range config.Watch { + if trigger.Action != types.WatchActionRebuild { + continue + } + if service.Build == nil { + return service, fmt.Errorf("can't watch service %q with action %s without a build context", service.Name, types.WatchActionRebuild) + } + if options.Build == nil { + return service, fmt.Errorf("--no-build is incompatible with watch action %s in service %s", types.WatchActionRebuild, service.Name) + } + service.PullPolicy = types.PullPolicyBuild + project.Services[serviceName] = service + } + return service, nil +} + +// watchTriggerPaths collects the paths to monitor for a service, skipping +// (with a warning) those already covered by a bind mount volume, and runs the +// initial sync for sync triggers requesting one. +func (s *composeService) watchTriggerPaths(ctx context.Context, project *types.Project, service types.ServiceConfig, config *types.DevelopConfig, syncer sync.Syncer) ([]string, error) { + var paths []string + for _, trigger := range config.Watch { + if isSync(trigger) && checkIfPathAlreadyBindMounted(trigger.Path, service.Volumes) { + logrus.Warnf("path '%s' also declared by a bind mount volume, this path won't be monitored!\n", trigger.Path) + continue + } + if initialSyncRequested(trigger) && isSync(trigger) { + // Need to check that initial files meant to be synced from the watch action are in the container + err := s.initialSync(ctx, project, service, trigger, syncer) + if err != nil { + return nil, err + } + } + paths = append(paths, trigger.Path) + } + return paths, nil +} + +// initialSyncRequested tells whether a sync trigger requests an initial sync, +// honoring the DEPRECATED x-initialSync extension attribute +func initialSyncRequested(trigger types.Trigger) bool { + if trigger.InitialSync { + return true + } + var legacyInitialSync bool + success, err := trigger.Extensions.Get("x-initialSync", &legacyInitialSync) + if err == nil && success && legacyInitialSync { + logrus.Warnf("x-initialSync is DEPRECATED, please use the official `initial_sync` attribute\n") + return true + } + return false +} + func getWatchRules(config *types.DevelopConfig, service types.ServiceConfig) ([]watchRule, error) { var rules []watchRule @@ -777,10 +800,6 @@ func (s *composeService) initialSync(ctx context.Context, project *types.Project } // Syncs files from develop.watch.path if they have been modified after the image has been created -// -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (s *composeService) initialSyncFiles(ctx context.Context, project *types.Project, service types.ServiceConfig, trigger types.Trigger, ignore watch.PathMatcher) ([]*sync.PathMapping, error) { fi, err := os.Stat(trigger.Path) if err != nil { @@ -790,57 +809,66 @@ func (s *composeService) initialSyncFiles(ctx context.Context, project *types.Pr if err != nil { return nil, err } - var pathsToCopy []*sync.PathMapping switch mode := fi.Mode(); { case mode.IsDir(): // process directory - err = filepath.WalkDir(trigger.Path, func(path string, d fs.DirEntry, err error) error { - if err != nil { - // handle possible path err, just in case... - return err - } - if trigger.Path == path { - // walk starts at the root directory - return nil - } - if shouldIgnore(filepath.Base(path), ignore) || checkIfPathAlreadyBindMounted(path, service.Volumes) { - // By definition sync ignores bind mounted paths - if d.IsDir() { - // skip folder - return fs.SkipDir - } - return nil // skip file - } - info, err := d.Info() - if err != nil { - return err - } - if !d.IsDir() { - if info.ModTime().Before(timeImageCreated) { - // skip file if it was modified before image creation - return nil - } - rel, err := filepath.Rel(trigger.Path, path) - if err != nil { - return err - } - // only copy files (and not full directories) - pathsToCopy = append(pathsToCopy, &sync.PathMapping{ - HostPath: path, - ContainerPath: filepath.Join(trigger.Target, rel), - }) - } - return nil - }) + return initialSyncDirectory(trigger, service, ignore, timeImageCreated) case mode.IsRegular(): // process file if fi.ModTime().After(timeImageCreated) && !shouldIgnore(filepath.Base(trigger.Path), ignore) && !checkIfPathAlreadyBindMounted(trigger.Path, service.Volumes) { - pathsToCopy = append(pathsToCopy, &sync.PathMapping{ + return []*sync.PathMapping{{ HostPath: trigger.Path, ContainerPath: trigger.Target, - }) + }}, nil } } + return nil, nil +} + +// initialSyncDirectory collects the files of a watched directory which were +// modified after the image was created, skipping ignored and bind-mounted +// paths +func initialSyncDirectory(trigger types.Trigger, service types.ServiceConfig, ignore watch.PathMatcher, timeImageCreated time.Time) ([]*sync.PathMapping, error) { + var pathsToCopy []*sync.PathMapping + err := filepath.WalkDir(trigger.Path, func(path string, d fs.DirEntry, err error) error { + if err != nil { + // handle possible path err, just in case... + return err + } + if trigger.Path == path { + // walk starts at the root directory + return nil + } + if shouldIgnore(filepath.Base(path), ignore) || checkIfPathAlreadyBindMounted(path, service.Volumes) { + // By definition sync ignores bind mounted paths + if d.IsDir() { + // skip folder + return fs.SkipDir + } + return nil // skip file + } + info, err := d.Info() + if err != nil { + return err + } + if d.IsDir() { + return nil + } + if info.ModTime().Before(timeImageCreated) { + // skip file if it was modified before image creation + return nil + } + rel, err := filepath.Rel(trigger.Path, path) + if err != nil { + return err + } + // only copy files (and not full directories) + pathsToCopy = append(pathsToCopy, &sync.PathMapping{ + HostPath: path, + ContainerPath: filepath.Join(trigger.Target, rel), + }) + return nil + }) return pathsToCopy, err } diff --git a/pkg/remote/oci.go b/pkg/remote/oci.go index cbc43edddc..1c14d31505 100644 --- a/pkg/remote/oci.go +++ b/pkg/remote/oci.go @@ -113,9 +113,6 @@ func (g *ociRemoteLoader) Accept(path string) bool { return strings.HasPrefix(path, OciPrefix) } -// FIXME(ndeloof) complete migration to gocognit -// -//nolint:gocognit func (g *ociRemoteLoader) Load(ctx context.Context, path string) (string, error) { enabled, err := ociRemoteLoaderEnabled() if err != nil { @@ -131,69 +128,90 @@ func (g *ociRemoteLoader) Load(ctx context.Context, path string) (string, error) local, ok := g.known[path] if !ok { - ref, err := reference.ParseDockerRef(path[len(OciPrefix):]) + local, err = g.pullComposeArtifact(ctx, path) if err != nil { return "", err } + g.known[path] = local + } + return filepath.Join(local, "compose.yaml"), nil +} - resolver := oci.NewResolver(g.dockerCli.ConfigFile(), g.httpTransport(ctx), g.insecureRegistries...) +// pullComposeArtifact resolves an oci:// path and pulls the compose artifact +// files into the local cache, unless already cached +func (g *ociRemoteLoader) pullComposeArtifact(ctx context.Context, path string) (string, error) { + ref, err := reference.ParseDockerRef(path[len(OciPrefix):]) + if err != nil { + return "", err + } - descriptor, content, err := oci.Get(ctx, resolver, ref) - if err != nil { - return "", fmt.Errorf("failed to pull OCI resource %q: %w", ref, err) - } + resolver := oci.NewResolver(g.dockerCli.ConfigFile(), g.httpTransport(ctx), g.insecureRegistries...) - cache, err := cacheDir() + descriptor, content, err := oci.Get(ctx, resolver, ref) + if err != nil { + return "", fmt.Errorf("failed to pull OCI resource %q: %w", ref, err) + } + + cache, err := cacheDir() + if err != nil { + return "", fmt.Errorf("initializing remote resource cache: %w", err) + } + + local := filepath.Join(cache, descriptor.Digest.Hex()) + if _, err = os.Stat(local); !os.IsNotExist(err) { + return local, nil + } + + // a Compose application bundle is published as an image index + if images.IsIndexType(descriptor.MediaType) { + content, err = g.resolveComposeManifest(ctx, resolver, ref, content) if err != nil { - return "", fmt.Errorf("initializing remote resource cache: %w", err) + return "", err } + } - local = filepath.Join(cache, descriptor.Digest.Hex()) - if _, err = os.Stat(local); os.IsNotExist(err) { - - // a Compose application bundle is published as an image index - if images.IsIndexType(descriptor.MediaType) { - var index spec.Index - err = json.Unmarshal(content, &index) - if err != nil { - return "", err - } - found := false - for _, manifest := range index.Manifests { - if manifest.ArtifactType != oci.ComposeProjectArtifactType { - continue - } - found = true - digested, err := reference.WithDigest(ref, manifest.Digest) - if err != nil { - return "", err - } - descriptor, content, err = oci.Get(ctx, resolver, digested) - if err != nil { - return "", fmt.Errorf("failed to pull OCI resource %q: %w", ref, err) - } - } - if !found { - return "", fmt.Errorf("OCI index %s doesn't refer to compose artifacts", ref) - } - } + var manifest spec.Manifest + err = json.Unmarshal(content, &manifest) + if err != nil { + return "", err + } - var manifest spec.Manifest - err = json.Unmarshal(content, &manifest) - if err != nil { - return "", err - } + err = g.pullComposeFiles(ctx, local, manifest, ref, resolver) + if err != nil { + // we need to clean up the directory to be sure we won't leave empty files behind + _ = os.RemoveAll(local) + return "", err + } + return local, nil +} - err = g.pullComposeFiles(ctx, local, manifest, ref, resolver) - if err != nil { - // we need to clean up the directory to be sure we won't leave empty files behind - _ = os.RemoveAll(local) - return "", err - } +// resolveComposeManifest returns the content of the compose artifact manifest +// referenced by an image index +func (g *ociRemoteLoader) resolveComposeManifest(ctx context.Context, resolver remotes.Resolver, ref reference.Named, content []byte) ([]byte, error) { + var index spec.Index + err := json.Unmarshal(content, &index) + if err != nil { + return nil, err + } + found := false + for _, manifest := range index.Manifests { + if manifest.ArtifactType != oci.ComposeProjectArtifactType { + continue + } + found = true + digested, err := reference.WithDigest(ref, manifest.Digest) + if err != nil { + return nil, err + } + _, content, err = oci.Get(ctx, resolver, digested) + if err != nil { + return nil, fmt.Errorf("failed to pull OCI resource %q: %w", ref, err) } - g.known[path] = local } - return filepath.Join(local, "compose.yaml"), nil + if !found { + return nil, fmt.Errorf("OCI index %s doesn't refer to compose artifacts", ref) + } + return content, nil } func (g *ociRemoteLoader) Dir(path string) string {