Per-service API client with identifying headers

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()
 }

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
 	}
 

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",

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{