fix(libstack): pull images sequentially and respect COMPOSE_PARALLEL_LIMIT [BE-12930] (#2555)

2026-05-06 15:16:14 +12:00
parent bf56a6c913
commit 7e526c4df7
3 changed files with 259 additions and 5 deletions
@@ -1,6 +1,7 @@
 package compose

 import (
+	"bufio"
 	"context"
 	"errors"
 	"fmt"
@@ -12,20 +13,28 @@ import (
 	"strings"
 	"sync"

+	"github.com/distribution/reference"
 	portainer "github.com/portainer/portainer/api"
 	"github.com/portainer/portainer/api/logs"
 	"github.com/portainer/portainer/pkg/libstack"
+	retry "github.com/portainer/portainer/pkg/retry"

 	"github.com/compose-spec/compose-go/v2/cli"
 	"github.com/compose-spec/compose-go/v2/types"
+	cerrdefs "github.com/containerd/errdefs"
 	"github.com/docker/cli/cli/command"
+	configtypes "github.com/docker/cli/cli/config/types"
 	"github.com/docker/cli/cli/flags"
 	cmdcompose "github.com/docker/compose/v2/cmd/compose"
 	"github.com/docker/compose/v2/pkg/api"
 	"github.com/docker/compose/v2/pkg/compose"
 	"github.com/docker/compose/v2/pkg/utils"
+	"github.com/docker/docker/api/types/image"
+	registrytypes "github.com/docker/docker/api/types/registry"
+	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/docker/docker/registry"
 	"github.com/rs/zerolog/log"
+	"github.com/segmentio/encoding/json"
 	"github.com/sirupsen/logrus"
 )

@@ -230,10 +239,139 @@ func (c *ComposeDeployer) Remove(ctx context.Context, projectName string, filePa
 	return nil
 }

+// Separator is used for naming components
+const separator = "-"
+
+// getImageNameOrDefault computes the default image name for a service
+func getImageNameOrDefault(service types.ServiceConfig, projectName string) string {
+	imageName := service.Image
+	if imageName == "" {
+		imageName = projectName + separator + service.Name
+	}
+	return imageName
+}
+
+// encodeRegistryAuth finds the registry credentials for the given image and returns
+// the base64-encoded auth string expected by the Docker service API.
+// Returns an empty string (no error) when no matching credentials are found.
+func encodeRegistryAuth(image string, registries []configtypes.AuthConfig) (string, error) {
+	named, err := reference.ParseNormalizedNamed(image)
+	if err != nil {
+		return "", fmt.Errorf("failed to parse image reference %q: %w", image, err)
+	}
+
+	domain := reference.Domain(named)
+	if domain == "docker.io" {
+		domain = registry.IndexServer
+	}
+
+	for _, r := range registries {
+		if r.ServerAddress == domain {
+			encoded, err := registrytypes.EncodeAuthConfig(registrytypes.AuthConfig{
+				Username:      r.Username,
+				Password:      r.Password,
+				ServerAddress: r.ServerAddress,
+				Auth:          r.Auth,
+				IdentityToken: r.IdentityToken,
+				RegistryToken: r.RegistryToken,
+			})
+			if err != nil {
+				return "", fmt.Errorf("failed to encode auth for registry %s: %w", domain, err)
+			}
+			return encoded, nil
+		}
+	}
+
+	return "", nil
+}
+
 // Pull pulls images
 func (c *ComposeDeployer) Pull(ctx context.Context, filePaths []string, options libstack.Options) error {
-	if err := c.withComposeService(ctx, filePaths, options, func(composeService api.Compose, project *types.Project) error {
-		return composeService.Pull(ctx, project, api.PullOptions{})
+	if err := withCli(ctx, options, func(ctx context.Context, cli *command.DockerCli) error {
+		project, err := createProject(ctx, filePaths, options)
+		if err != nil {
+			return fmt.Errorf("failed to create compose project: %w", err)
+		}
+
+		for _, s := range project.Services {
+			imageName := getImageNameOrDefault(s, project.Name)
+			encodedAuth, err := encodeRegistryAuth(imageName, options.Registries)
+			if err != nil {
+				return fmt.Errorf("failed to encode registry auth: %w", err)
+			}
+
+			_, err = retry.RetryWithWarnings("Pull image: "+imageName, retry.Default, func() (string, error) {
+				_, err := cli.Client().ImageInspect(ctx, imageName)
+				if cerrdefs.IsNotFound(err) {
+					reader, err := cli.Client().ImagePull(ctx, imageName, image.PullOptions{
+						Platform:     s.Platform,
+						RegistryAuth: encodedAuth,
+					})
+					if err != nil {
+						return "", fmt.Errorf("failed to pull image: %w", err)
+					}
+
+					defer logs.CloseAndLogErr(reader)
+
+					scanner := bufio.NewScanner(reader)
+					for scanner.Scan() {
+						message := scanner.Text()
+						log.Debug().
+							Str("ProjectName", options.ProjectName).
+							Str("Host", options.Host).
+							Str("Image", imageName).
+							Msg(message)
+
+						var m jsonmessage.JSONMessage
+						err := json.Unmarshal([]byte(message), &m)
+						if err != nil {
+							log.Error().
+								Err(err).
+								Str("ProjectName", options.ProjectName).
+								Str("Host", options.Host).
+								Str("Image", imageName).
+								Msg("ComposeDeployer.Pull: failed to json Unmarshal image pull message.")
+							return "", fmt.Errorf("failed to json Unmarshal image pull message: %w", err)
+						}
+
+						if m.Error != nil {
+							log.Error().
+								Err(m.Error).
+								Str("ProjectName", options.ProjectName).
+								Str("Host", options.Host).
+								Str("Image", imageName).
+								Msg("ComposeDeployer.Pull: error pulling image")
+							return "", fmt.Errorf("error pulling image: %w", m.Error)
+						}
+					}
+					if err := scanner.Err(); err != nil {
+						log.Error().
+							Err(err).
+							Str("ProjectName", options.ProjectName).
+							Str("Host", options.Host).
+							Str("Image", imageName).
+							Msg("ComposeDeployer.Pull: error reading from pull reader")
+						return "", fmt.Errorf("error reading from pull reader: %w", err)
+					}
+
+					return "", nil
+				} else if err != nil {
+					return "", fmt.Errorf("failed to inspect image: %w", err)
+				} else {
+					return "", nil
+				}
+			})
+			if err != nil {
+				log.Error().
+					Err(err).
+					Str("ProjectName", options.ProjectName).
+					Str("Host", options.Host).
+					Str("Image", imageName).
+					Msg("ComposeDeployer.Pull: failed to pull image")
+				return fmt.Errorf("failed to pull image: %w", err)
+			}
+		}
+		return nil
 	}); err != nil {
 		return fmt.Errorf("compose pull operation failed: %w", err)
 	}
@@ -355,10 +493,15 @@ func createProject(ctx context.Context, configFilepaths []string, options libsta
 	}

 	var osPortainerEnvVars []string
+	var composeEnvVars []string
 	for _, ev := range os.Environ() {
 		if strings.HasPrefix(ev, portainerEnvVarsPrefix) {
 			osPortainerEnvVars = append(osPortainerEnvVars, ev)
 		}
+
+		if strings.HasPrefix(ev, "COMPOSE_") {
+			composeEnvVars = append(composeEnvVars, ev)
+		}
 	}

 	projectOptions, err := cli.NewProjectOptions(configFilepaths,
@@ -367,6 +510,7 @@ func createProject(ctx context.Context, configFilepaths []string, options libsta
 		cli.WithoutEnvironmentResolution,
 		cli.WithResolvedPaths(!slices.Contains(options.ConfigOptions, "--no-path-resolution")),
 		cli.WithEnv(osPortainerEnvVars),
+		cli.WithEnv(composeEnvVars),
 		cli.WithEnv(options.Env),
 		cli.WithEnvFiles(envFiles...),
 		func(o *cli.ProjectOptions) error {
@@ -139,7 +139,6 @@ configs:
 	require.NoError(t, err)

 	require.False(t, containerExists(containerName))
-
 }

 func TestRun(t *testing.T) {
@@ -1144,6 +1143,23 @@ func Test_createProject(t *testing.T) {
 			},
 			expectedProject: expectedSimpleComposeProject("", map[string]string{"PORTAINER_WEB_FOLDER": "html-1"}),
 		},
+		{
+			name: "OS COMPOSE_ vars are passed to project",
+			filesToCreate: map[string]string{
+				"docker-compose.yml": testSimpleComposeConfig,
+			},
+			configFilepaths: []string{dir + "/docker-compose.yml"},
+			options: libstack.Options{
+				ProjectName: projectName,
+			},
+			osEnv: map[string]string{
+				"COMPOSE_PARALLEL_LIMIT": "4",
+				"other_var":              "something",
+			},
+			expectedProject: expectedSimpleComposeProject("", map[string]string{
+				"COMPOSE_PARALLEL_LIMIT": "4",
+			}),
+		},
 		{
 			name: "Env Vars in compose file, compose env file, env, os, and env_file",
 			filesToCreate: map[string]string{
@@ -1450,6 +1466,100 @@ func Test_CredentialsStore_Behavior(t *testing.T) {
 	})
 }

+func TestGetImageNameOrDefault(t *testing.T) {
+	testCases := []struct {
+		name         string
+		service      types.ServiceConfig
+		projectName  string
+		expectedName string
+	}{
+		{
+			name:         "service with explicit image",
+			service:      types.ServiceConfig{Name: "web", Image: "nginx:latest"},
+			projectName:  "myproject",
+			expectedName: "nginx:latest",
+		},
+		{
+			name:         "service without image uses default",
+			service:      types.ServiceConfig{Name: "web", Image: ""},
+			projectName:  "myproject",
+			expectedName: "myproject-web",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			gotName := getImageNameOrDefault(tc.service, tc.projectName)
+			require.Equal(t, tc.expectedName, gotName)
+		})
+	}
+}
+
+func TestEncodeRegistryAuth(t *testing.T) {
+	testCases := []struct {
+		name               string
+		image              string
+		registries         []configtypes.AuthConfig
+		expectRegistryAuth string
+		expectError        string
+	}{
+		{
+			name:  "matching registry returns encoded auth",
+			image: "myregistry.example.com/myimage:latest",
+			registries: []configtypes.AuthConfig{
+				{ServerAddress: "myregistry.example.com", Username: "user", Password: "pass"},
+			},
+			expectRegistryAuth: "eyJ1c2VybmFtZSI6InVzZXIiLCJwYXNzd29yZCI6InBhc3MiLCJzZXJ2ZXJhZGRyZXNzIjoibXlyZWdpc3RyeS5leGFtcGxlLmNvbSJ9",
+		},
+		{
+			name:  "unknown registry returns empty string",
+			image: "myregistry.example.com/myimage:latest",
+			registries: []configtypes.AuthConfig{
+				{ServerAddress: "other.registry.com", Username: "user", Password: "pass"},
+			},
+			expectRegistryAuth: "",
+		},
+		{
+			name:  "docker.io image matches index server",
+			image: "alpine:latest",
+			registries: []configtypes.AuthConfig{
+				{ServerAddress: "https://index.docker.io/v1/", Username: "user", Password: "pass"},
+			},
+			expectRegistryAuth: "eyJ1c2VybmFtZSI6InVzZXIiLCJwYXNzd29yZCI6InBhc3MiLCJzZXJ2ZXJhZGRyZXNzIjoiaHR0cHM6Ly9pbmRleC5kb2NrZXIuaW8vdjEvIn0=",
+		},
+		{
+			name:               "invalid image reference returns error",
+			image:              "INVALID::IMAGE",
+			registries:         []configtypes.AuthConfig{},
+			expectRegistryAuth: "",
+			expectError:        `failed to parse image reference "INVALID::IMAGE": invalid reference format: repository name (library/INVALID) must be lowercase`,
+		},
+		{
+			name:  "multiple registries only matching one used",
+			image: "registry-b.example.com/image:latest",
+			registries: []configtypes.AuthConfig{
+				{ServerAddress: "registry-a.example.com", Username: "user-a", Password: "pass-a"},
+				{ServerAddress: "registry-b.example.com", Username: "user-b", Password: "pass-b"},
+			},
+			expectRegistryAuth: "eyJ1c2VybmFtZSI6InVzZXItYiIsInBhc3N3b3JkIjoicGFzcy1iIiwic2VydmVyYWRkcmVzcyI6InJlZ2lzdHJ5LWIuZXhhbXBsZS5jb20ifQ==",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			gotRegistryAuth, err := encodeRegistryAuth(tc.image, tc.registries)
+
+			if tc.expectError == "" {
+				require.NoError(t, err)
+			} else {
+				require.ErrorContains(t, err, tc.expectError)
+			}
+
+			require.Equal(t, tc.expectRegistryAuth, gotRegistryAuth)
+		})
+	}
+}
+
 func createMockComposeService(command.Cli, ...compose.Option) api.Compose {
 	return &mockComposeService{}
 }