diff --git a/pkg/compose/compose.go b/pkg/compose/compose.go index 33ed81af9b..ed3887ed52 100644 --- a/pkg/compose/compose.go +++ b/pkg/compose/compose.go @@ -37,6 +37,7 @@ import ( "github.com/moby/moby/client" "github.com/sirupsen/logrus" + "github.com/docker/compose/v5/internal" "github.com/docker/compose/v5/pkg/api" "github.com/docker/compose/v5/pkg/dryrun" ) @@ -217,6 +218,7 @@ type composeService struct { dryRun bool runtimeAPIVersion runtimeVersionCache + serviceClients serviceClientCache } // Close releases any connections/resources held by the underlying clients. @@ -225,6 +227,7 @@ type composeService struct { // will get cleaned up at about the same time regardless even if not invoked. func (s *composeService) Close() error { var errs []error + errs = append(errs, s.serviceClients.Close()) if s.dockerCli != nil { errs = append(errs, s.apiClient().Close()) } @@ -235,6 +238,104 @@ func (s *composeService) apiClient() client.APIClient { return s.dockerCli.Client() } +// Compose tags the Docker API requests it makes on behalf of a specific service +// with these headers, so the daemon can attribute the request to the project +// and service it originates from. +const ( + composeProjectHeader = "X-Docker-Compose-Project" + composeServiceHeader = "X-Docker-Compose-Service" +) + +type serviceClientKey struct { + project string + service string +} + +// serviceClientCache memoizes one API client per project/service. Each client +// tags its requests with headers identifying the service it acts on, so a +// separate client is required per service (the Docker client applies custom +// headers per instance, not per request). Clients are closed by +// composeService.Close(). +type serviceClientCache struct { + mu sync.Mutex + clients map[serviceClientKey]client.APIClient +} + +// Close closes every cached client and empties the cache. +func (c *serviceClientCache) Close() error { + c.mu.Lock() + defer c.mu.Unlock() + var errs []error + for _, cli := range c.clients { + errs = append(errs, cli.Close()) + } + c.clients = nil + return errors.Join(errs...) +} + +// serviceClient returns an API client whose requests are tagged with headers +// identifying projectName/serviceName. Clients are cached and reused; the +// returned client is owned and closed by the composeService, so callers must +// not close it themselves. Use it for Docker API calls scoped to a single +// service; project-wide calls (networks, volumes, events, project-wide +// container listings, …) should keep using apiClient. +func (s *composeService) serviceClient(projectName, serviceName string) (client.APIClient, error) { + key := serviceClientKey{project: projectName, service: serviceName} + + s.serviceClients.mu.Lock() + defer s.serviceClients.mu.Unlock() + if c, ok := s.serviceClients.clients[key]; ok { + return c, nil + } + + c, err := s.newServiceClient(projectName, serviceName) + if err != nil { + return nil, err + } + if s.serviceClients.clients == nil { + s.serviceClients.clients = make(map[serviceClientKey]client.APIClient) + } + s.serviceClients.clients[key] = c + return c, nil +} + +// newServiceClient builds an API client that mirrors the CLI-provided client +// (same endpoint, TLS and negotiated API version) and tags every request with +// headers identifying the project and service on whose behalf it acts. +func (s *composeService) newServiceClient(projectName, serviceName string) (client.APIClient, error) { + endpoint := s.dockerCli.DockerEndpoint() + opts, err := endpoint.ClientOpts() + if err != nil { + return nil, err + } + + headers := map[string]string{} + for k, v := range s.configFile().HTTPHeaders { + headers[k] = v + } + headers[composeProjectHeader] = projectName + headers[composeServiceHeader] = serviceName + + opts = append(opts, + client.WithAPIVersion(s.apiClient().ClientVersion()), + client.WithUserAgent("compose/"+internal.Version), + client.WithHTTPHeaders(headers), + ) + apiClient, err := client.New(opts...) + if err != nil { + return nil, err + } + + // In dry-run mode the shared client is a DryRunClient that intercepts + // mutating calls (e.g. ImagePull returns a fake stream without contacting + // the daemon). A freshly-built client would bypass that and hit the daemon, + // so wrap the scoped client to preserve dry-run semantics. + if s.dryRun { + return dryrun.NewDryRunClient(apiClient, s.dockerCli) + } + return apiClient, nil +} + func (s *composeService) configFile() *configfile.ConfigFile { return s.dockerCli.ConfigFile() } diff --git a/pkg/compose/convergence.go b/pkg/compose/convergence.go index 67917f20db..38b907d739 100644 --- a/pkg/compose/convergence.go +++ b/pkg/compose/convergence.go @@ -348,7 +348,12 @@ func (s *composeService) createMobyContainer(ctx context.Context, project *types plat = &p } - response, err := s.apiClient().ContainerCreate(ctx, client.ContainerCreateOptions{ + apiClient, err := s.serviceClient(project.Name, service.Name) + if err != nil { + return created, err + } + + response, err := apiClient.ContainerCreate(ctx, client.ContainerCreateOptions{ Name: name, Platform: plat, Config: cfgs.Container, @@ -383,20 +388,20 @@ func (s *composeService) createMobyContainer(ctx context.Context, project *types } epSettings, err := createEndpointSettings(project, service, number, networkKey, cfgs.Links, opts.UseNetworkAliases) if err != nil { - _, _ = s.apiClient().ContainerRemove(ctx, response.ID, client.ContainerRemoveOptions{Force: true}) + _, _ = apiClient.ContainerRemove(ctx, response.ID, client.ContainerRemoveOptions{Force: true}) return created, err } - if _, err := s.apiClient().NetworkConnect(ctx, mobyNetworkName, client.NetworkConnectOptions{ + if _, err := apiClient.NetworkConnect(ctx, mobyNetworkName, client.NetworkConnectOptions{ Container: response.ID, EndpointConfig: epSettings, }); err != nil { - _, _ = s.apiClient().ContainerRemove(ctx, response.ID, client.ContainerRemoveOptions{Force: true}) + _, _ = apiClient.ContainerRemove(ctx, response.ID, client.ContainerRemoveOptions{Force: true}) return created, err } } } - res, err := s.apiClient().ContainerInspect(ctx, response.ID, client.ContainerInspectOptions{}) + res, err := apiClient.ContainerInspect(ctx, response.ID, client.ContainerInspectOptions{}) if err != nil { return created, err } @@ -570,9 +575,14 @@ func (s *composeService) startServiceContainer(ctx context.Context, project *typ return err } + apiClient, err := s.serviceClient(project.Name, service.Name) + if err != nil { + return err + } + eventName := getContainerProgressName(ctr) s.events.On(newEvent(eventName, api.Working, api.StatusStarting)) - if _, err := s.apiClient().ContainerStart(ctx, ctr.ID, client.ContainerStartOptions{}); err != nil { + if _, err := apiClient.ContainerStart(ctx, ctr.ID, client.ContainerStartOptions{}); err != nil { return err } diff --git a/pkg/compose/convergence_test.go b/pkg/compose/convergence_test.go index 5b4514d067..c4f61a3490 100644 --- a/pkg/compose/convergence_test.go +++ b/pkg/compose/convergence_test.go @@ -424,6 +424,15 @@ func TestIsServiceHealthy(t *testing.T) { }) } +// seedServiceClient pre-populates the per-service client cache so that +// service-scoped calls (ContainerCreate, ContainerStart, …) resolve to the +// given mock instead of building a real client from the Docker endpoint. +func seedServiceClient(s *composeService, project, service string, c client.APIClient) { + s.serviceClients.clients = map[serviceClientKey]client.APIClient{ + {project: project, service: service}: c, + } +} + func TestCreateMobyContainer(t *testing.T) { mockCtrl := gomock.NewController(t) defer mockCtrl.Finish() @@ -435,6 +444,7 @@ func TestCreateMobyContainer(t *testing.T) { cli.EXPECT().ConfigFile().Return(&configfile.ConfigFile{}).AnyTimes() apiClient.EXPECT().DaemonHost().Return("").AnyTimes() apiClient.EXPECT().ImageInspect(anyCancellableContext(), gomock.Any()).Return(client.ImageInspectResult{}, nil).AnyTimes() + seedServiceClient(tested.(*composeService), "bork", "test", apiClient) apiClient.EXPECT().Ping(gomock.Any(), client.PingOptions{NegotiateAPIVersion: true}).Return(client.PingResult{ APIVersion: "1.44", @@ -539,6 +549,7 @@ func TestCreateMobyContainerLegacyAPI(t *testing.T) { apiClient.EXPECT().Ping(gomock.Any(), client.PingOptions{NegotiateAPIVersion: true}). Return(client.PingResult{APIVersion: "1.43"}, nil).AnyTimes() apiClient.EXPECT().ClientVersion().Return("1.43").AnyTimes() + seedServiceClient(tested.(*composeService), "bork", "test", apiClient) service := types.ServiceConfig{ Name: "test", @@ -628,6 +639,7 @@ func TestCreateMobyContainerLegacyAPI_NetworkConnectFailure(t *testing.T) { apiClient.EXPECT().Ping(gomock.Any(), client.PingOptions{NegotiateAPIVersion: true}). Return(client.PingResult{APIVersion: "1.43"}, nil).AnyTimes() apiClient.EXPECT().ClientVersion().Return("1.43").AnyTimes() + seedServiceClient(tested.(*composeService), "bork", "test", apiClient) service := types.ServiceConfig{ Name: "test", diff --git a/pkg/compose/pull.go b/pkg/compose/pull.go index ce14a5b5ce..13788f09c2 100644 --- a/pkg/compose/pull.go +++ b/pkg/compose/pull.go @@ -115,7 +115,7 @@ func (s *composeService) pull(ctx context.Context, project *types.Project, opts idx := i eg.Go(func() error { - _, err := s.pullServiceImage(ctx, service, opts.Quiet, project.Environment["DOCKER_DEFAULT_PLATFORM"]) + _, err := s.pullServiceImage(ctx, project.Name, service, opts.Quiet, project.Environment["DOCKER_DEFAULT_PLATFORM"]) if err != nil { pullErrors[idx] = err if service.Build != nil { @@ -173,7 +173,24 @@ func getUnwrappedErrorMessage(err error) string { return err.Error() } -func (s *composeService) pullServiceImage(ctx context.Context, service types.ServiceConfig, quietPull bool, defaultPlatform string) (string, error) { +// resolvePullPlatforms returns the OCI platforms to request for an image pull, +// falling back to defaultPlatform when the service declares none. +func resolvePullPlatforms(servicePlatform, defaultPlatform string) ([]ocispec.Platform, error) { + platform := servicePlatform + if platform == "" { + platform = defaultPlatform + } + if platform == "" { + return nil, nil + } + p, err := platforms.Parse(platform) + if err != nil { + return nil, err + } + return []ocispec.Platform{p}, nil +} + +func (s *composeService) pullServiceImage(ctx context.Context, projectName string, service types.ServiceConfig, quietPull bool, defaultPlatform string) (string, error) { resource := "Image " + service.Image s.events.On(newEvent(resource, api.Working, api.StatusPulling)) ref, err := reference.ParseNormalizedNamed(service.Image) @@ -186,21 +203,17 @@ func (s *composeService) pullServiceImage(ctx context.Context, service types.Ser return "", err } - platform := service.Platform - if platform == "" { - platform = defaultPlatform + ociPlatforms, err := resolvePullPlatforms(service.Platform, defaultPlatform) + if err != nil { + return "", err } - var ociPlatforms []ocispec.Platform - if platform != "" { - p, err := platforms.Parse(platform) - if err != nil { - return "", err - } - ociPlatforms = append(ociPlatforms, p) + apiClient, err := s.serviceClient(projectName, service.Name) + if err != nil { + return "", err } - stream, err := s.apiClient().ImagePull(ctx, service.Image, client.ImagePullOptions{ + stream, err := apiClient.ImagePull(ctx, service.Image, client.ImagePullOptions{ RegistryAuth: encodedAuth, Platforms: ociPlatforms, }) @@ -324,7 +337,7 @@ func (s *composeService) pullRequiredImages(ctx context.Context, project *types. var mutex sync.Mutex for name, service := range needPull { eg.Go(func() error { - id, err := s.pullServiceImage(ctx, service, quietPull, project.Environment["DOCKER_DEFAULT_PLATFORM"]) + id, err := s.pullServiceImage(ctx, project.Name, service, quietPull, project.Environment["DOCKER_DEFAULT_PLATFORM"]) mutex.Lock() defer mutex.Unlock() pulledImages[name] = api.ImageSummary{ diff --git a/pkg/compose/pull_test.go b/pkg/compose/pull_test.go new file mode 100644 index 0000000000..3146065e23 --- /dev/null +++ b/pkg/compose/pull_test.go @@ -0,0 +1,115 @@ +/* + 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 ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/docker/cli/cli/config/configfile" + "github.com/docker/cli/cli/context/docker" + "github.com/moby/moby/client" + "go.uber.org/mock/gomock" + "gotest.tools/v3/assert" +) + +func TestResolvePullPlatforms(t *testing.T) { + tests := []struct { + name string + servicePlatform string + defaultPlatform string + wantLen int + wantErr bool + }{ + {name: "none", wantLen: 0}, + {name: "service platform", servicePlatform: "linux/amd64", wantLen: 1}, + {name: "default platform fallback", defaultPlatform: "linux/arm64", wantLen: 1}, + {name: "service overrides default", servicePlatform: "linux/amd64", defaultPlatform: "linux/arm64", wantLen: 1}, + {name: "invalid", servicePlatform: "not a platform!!", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := resolvePullPlatforms(tt.servicePlatform, tt.defaultPlatform) + if tt.wantErr { + assert.Assert(t, err != nil) + return + } + assert.NilError(t, err) + assert.Equal(t, len(got), tt.wantLen) + }) + } +} + +// TestServiceClientHeaders verifies that requests made through a per-service +// client carry headers identifying the Compose project and service on whose +// behalf the request is made, and that CLI-configured headers are preserved. +func TestServiceClientHeaders(t *testing.T) { + mockCtrl := gomock.NewController(t) + defer mockCtrl.Finish() + + var gotProject, gotService, gotConfigHeader string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "/images/create") { + gotProject = r.Header.Get(composeProjectHeader) + gotService = r.Header.Get(composeServiceHeader) + gotConfigHeader = r.Header.Get("X-Custom") + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"Pulling"}` + "\n")) + })) + defer srv.Close() + + api, cli := prepareMocks(mockCtrl) + // Use a "tcp" scheme so the moby client dials the httptest server over HTTP. + host := "tcp://" + strings.TrimPrefix(srv.URL, "http://") + cli.EXPECT().DockerEndpoint().Return(docker.Endpoint{ + EndpointMeta: docker.EndpointMeta{Host: host}, + }).AnyTimes() + cli.EXPECT().ConfigFile().Return(&configfile.ConfigFile{ + HTTPHeaders: map[string]string{"X-Custom": "config-value"}, + }).AnyTimes() + api.EXPECT().ClientVersion().Return("1.51").AnyTimes() + api.EXPECT().Close().Return(nil).AnyTimes() + + s := composeService{dockerCli: cli} + + apiClient, err := s.serviceClient("myProject", "myService") + assert.NilError(t, err) + + stream, err := apiClient.ImagePull(t.Context(), "alpine:latest", client.ImagePullOptions{}) + assert.NilError(t, err) + _, _ = io.Copy(io.Discard, stream) + _ = stream.Close() + + assert.Equal(t, gotProject, "myProject") + assert.Equal(t, gotService, "myService") + // Headers configured on the CLI (e.g. via config.json) are preserved. + assert.Equal(t, gotConfigHeader, "config-value") + + // The same service resolves to the cached client; a different service does not. + again, err := s.serviceClient("myProject", "myService") + assert.NilError(t, err) + assert.Equal(t, again, apiClient) + other, err := s.serviceClient("myProject", "otherService") + assert.NilError(t, err) + assert.Assert(t, other != apiClient) + + assert.NilError(t, s.Close()) +}