diff --git a/pkg/compose/build.go b/pkg/compose/build.go index b422b5079..e70b16d5e 100644 --- a/pkg/compose/build.go +++ b/pkg/compose/build.go @@ -106,7 +106,7 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti return s.doBuildClassic(ctx, project, serviceToBuild, options) } -func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, buildOpts *api.BuildOptions, quietPull bool) error { +func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, buildOpts *api.BuildOptions, quietPull bool, skipProviders bool) error { for name, service := range project.Services { if service.Provider == nil && service.Image == "" && service.Build == nil { return fmt.Errorf("invalid service %q. Must specify either image or build", name) @@ -127,6 +127,7 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types. return err } + built := map[string]string{} if buildOpts != nil { err = tracing.SpanWrapFunc("project/build", tracing.ProjectOptions(ctx, project), func(ctx context.Context) error { @@ -136,6 +137,7 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types. } for name, digest := range builtImages { + built[name] = digest images[name] = api.ImageSummary{ Repository: name, ID: digest, @@ -150,6 +152,20 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types. } } + // provider-backed services get their turn in the image phase: the + // provider makes the image available to ITS runtime (weak contract on + // the up path: a usable version present suffices) + if !skipProviders { + err = tracing.SpanWrapFunc("project/provider-pull", tracing.ProjectOptions(ctx, project), + func(ctx context.Context) error { + return s.ensureProviderImages(ctx, project, built, providerPullPolicyMissing) + }, + )(ctx) + if err != nil { + return err + } + } + // set digest as com.docker.compose.image label so we can detect outdated // containers — the single writer of that label, so the platform-pinned // resolution below can't be overwritten by another code path diff --git a/pkg/compose/create.go b/pkg/compose/create.go index 9ac5199db..01cc3aacc 100644 --- a/pkg/compose/create.go +++ b/pkg/compose/create.go @@ -85,7 +85,7 @@ func (s *composeService) create(ctx context.Context, project *types.Project, opt return err } - err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull) + err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull, options.SkipProviders) if err != nil { return err } diff --git a/pkg/compose/images_test.go b/pkg/compose/images_test.go index a1b53ad47..53921e97b 100644 --- a/pkg/compose/images_test.go +++ b/pkg/compose/images_test.go @@ -423,7 +423,7 @@ func TestPlatformPinnedDigest(t *testing.T) { Return(client.ImageInspectResult{InspectResponse: multiPlatform}, nil) project := newProject() - assert.NilError(t, tested.ensureImagesExists(t.Context(), project, nil, true)) + assert.NilError(t, tested.ensureImagesExists(t.Context(), project, nil, true, false)) assert.Equal(t, project.Services["app"].CustomLabels[compose.ImageDigestLabel], "sha256:s390x") }) diff --git a/pkg/compose/provider_images.go b/pkg/compose/provider_images.go new file mode 100644 index 000000000..0150474b3 --- /dev/null +++ b/pkg/compose/provider_images.go @@ -0,0 +1,132 @@ +/* + Copyright 2020 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/compose-spec/compose-go/v2/types" + "golang.org/x/sync/errgroup" + + "github.com/docker/compose/v5/pkg/api" +) + +const ( + // providerImageSourceLocal: the desired state is the local daemon's + // image; the provider synchronizes from it (get-image) when its + // bookkeeping says it holds something else. + providerImageSourceLocal = "local" + // providerImageSourceRegistry: the reference is resolvable upstream and + // upstream is the authority; local facts are an optimization, never an + // obligation. + providerImageSourceRegistry = "registry" + + // providerPullPolicyMissing (up path): a usable version present in the + // provider's runtime suffices. + providerPullPolicyMissing = "missing" + // providerPullPolicyAlways (compose pull): the provider must ensure it + // holds the latest version of the authority. + providerPullPolicyAlways = "always" +) + +// ensureProviderImages gives provider-backed services their turn in the image +// phase: for every service whose provider declares the pull command (metadata +// opt-in, like stop), compose invokes +// +// compose pull --image= [--digest=… --created=…] --source=… --policy=… +// +// The provider owns making the image available to ITS runtime: resolving the +// reference upstream, or requesting the local bytes with a get-image message. +// built maps the image names (re)built by the current run, which forces the +// local-source verdict for them. +func (s *composeService) ensureProviderImages(ctx context.Context, project *types.Project, built map[string]string, policy string) error { + // deliberately NOT errgroup.WithContext: providers are independent, and + // fail-fast cancellation would kill the siblings of the first failure + // mid-run, mangling their reports into context-canceled noise. Every + // provider runs to completion and every failure is reported. + var eg errgroup.Group + eg.SetLimit(s.maxConcurrency) + var mu sync.Mutex + var errs []error + for _, service := range project.Services { + if service.Provider == nil { + continue + } + image := api.GetImageNameOrDefault(service, project.Name) + _, justBuilt := built[image] + source := providerImageSource(service, justBuilt) + eg.Go(func() error { + if err := s.runProviderPull(ctx, project, service, image, source, policy); err != nil { + mu.Lock() + errs = append(errs, fmt.Errorf("provider service %q: %w", service.Name, err)) + mu.Unlock() + } + return nil + }) + } + _ = eg.Wait() + return errors.Join(errs...) +} + +// providerImageSource computes the authority VERDICT for one invocation — +// compose owns this arbitration (pull_policy, flags, what this run built) so +// providers never re-implement it: +// - "local": the desired state is the local daemon's image — a declared +// build with no pullable reference, pull_policy build, or an image the +// current run just (re)built; +// - "registry": the provider resolves the reference upstream — including +// the build-in-CI workflow where a build section is only the recipe used +// to publish the image consumers pull. +func providerImageSource(service types.ServiceConfig, justBuilt bool) string { + if service.Build == nil { + return providerImageSourceRegistry + } + if service.Image == "" || justBuilt { + return providerImageSourceLocal + } + if policy, _, err := service.GetPullPolicy(); err == nil && policy == types.PullPolicyBuild { + return providerImageSourceLocal + } + return providerImageSourceRegistry +} + +func (s *composeService) runProviderPull(ctx context.Context, project *types.Project, service types.ServiceConfig, image, source, policy string) error { + args := []string{"--image=" + image, "--source=" + source, "--policy=" + policy} + if digest, created, ok := s.localImageFacts(ctx, image); ok { + args = append(args, "--digest="+digest) + if created != "" { + args = append(args, "--created="+created) + } + } + return s.runPlugin(ctx, project, service, "pull", args...) +} + +// localImageFacts reports the state of the local daemon cache for ref: the +// image ID and its creation time. These describe the cache, they are not +// instructions — the provider persists them as the bookkeeping keys of what +// it ingested (digest as identity test, created as the ordering fallback for +// backends that cannot preserve digests). +func (s *composeService) localImageFacts(ctx context.Context, ref string) (digest, created string, ok bool) { + inspected, err := s.apiClient().ImageInspect(ctx, ref) + if err != nil { + return "", "", false + } + return inspected.ID, inspected.Created, true +} diff --git a/pkg/compose/provider_images_test.go b/pkg/compose/provider_images_test.go new file mode 100644 index 000000000..578f37ba6 --- /dev/null +++ b/pkg/compose/provider_images_test.go @@ -0,0 +1,266 @@ +/* + 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 ( + "bufio" + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http/httputil" + "os" + "os/exec" + "strconv" + "testing" + + "github.com/compose-spec/compose-go/v2/types" + "go.uber.org/mock/gomock" + "gotest.tools/v3/assert" + + "github.com/docker/compose/v5/pkg/mocks" +) + +// providerSourceService builds a ServiceConfig by assignment: Image and +// PullPolicy are fields promoted from the embedded ContainerSpec, which +// struct literals cannot set before go1.27. +func providerSourceService(build *types.BuildConfig, image, pullPolicy string) types.ServiceConfig { + var s types.ServiceConfig + s.Build = build + s.Image = image + s.PullPolicy = pullPolicy + return s +} + +func TestProviderImageSource(t *testing.T) { + build := &types.BuildConfig{Context: "."} + tests := []struct { + name string + service types.ServiceConfig + justBuilt bool + want string + }{ + {"no build: registry authority", providerSourceService(nil, "nginx", ""), false, providerImageSourceRegistry}, + {"build without image: local-only name", providerSourceService(build, "", ""), false, providerImageSourceLocal}, + {"build just run: local wins for this invocation", providerSourceService(build, "repo/app", ""), true, providerImageSourceLocal}, + {"build as CI recipe, default policy: registry authority", providerSourceService(build, "repo/app", ""), false, providerImageSourceRegistry}, + {"pull_policy build: local authority", providerSourceService(build, "repo/app", types.PullPolicyBuild), false, providerImageSourceLocal}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, providerImageSource(tc.service, tc.justBuilt), tc.want) + }) + } +} + +// TestStreamImageTo locks the wire format of the get-image answer: one JSON +// announce line, then the tar bytes as an HTTP/1.1 chunked body terminated by +// the zero chunk and a final CRLF — decodable with a stock chunked reader. +func TestStreamImageTo(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + s := svc.(*composeService) + + payload := bytes.Repeat([]byte("compose-image-bytes\x00\x17\xff"), 100_000) // > one chunk, binary-hostile values included + apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-app"}). + Return(fakeImageSaveResult{io.NopCloser(bytes.NewReader(payload))}, nil) + + var out bytes.Buffer + assert.NilError(t, s.streamImageTo(t.Context(), &out, "proj-app", "")) + + rd := bufio.NewReader(&out) + line, err := rd.ReadString('\n') + assert.NilError(t, err) + var announce imageStreamAnswer + assert.NilError(t, json.Unmarshal([]byte(line), &announce)) + assert.Equal(t, announce.Type, ImageStreamType) + assert.Equal(t, announce.Encoding, "chunked") + assert.Equal(t, announce.Error, "") + + body, err := io.ReadAll(httputil.NewChunkedReader(rd)) + assert.NilError(t, err) + assert.Assert(t, bytes.Equal(body, payload), "chunked round-trip must be byte-identical (%d vs %d bytes)", len(body), len(payload)) + + // nothing may follow the zero chunk: a stock chunked reader stops there + // without consuming further bytes, and any leftover would corrupt the + // next answer of a provider requesting several images + rest, err := io.ReadAll(rd) + assert.NilError(t, err) + assert.Equal(t, string(rest), "", "no byte may follow the terminating zero chunk") +} + +// TestStreamImageTo_SaveError: when the daemon cannot export the image, the +// announce line carries the error and no stream follows — the provider is +// never left waiting. +func TestStreamImageTo_SaveError(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + s := svc.(*composeService) + + apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-app"}). + Return(nil, errors.New("no such image")) + + var out bytes.Buffer + assert.ErrorContains(t, s.streamImageTo(t.Context(), &out, "proj-app", ""), "no such image") + + var announce imageStreamAnswer + assert.NilError(t, json.Unmarshal(out.Bytes(), &announce)) + assert.Equal(t, announce.Type, ImageStreamType) + assert.ErrorContains(t, errors.New(announce.Error), "no such image") + assert.Equal(t, announce.Encoding, "") +} + +type fakeImageSaveResult struct{ io.ReadCloser } + +// TestExecutePlugin_GetImage runs the pull command against a fake provider +// (this test binary re-executed, see TestHelperProviderPull): the get-image +// request must be answered with the announced chunked stream, which the +// provider consumes and acknowledges by hashing. +func TestExecutePlugin_GetImage(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + + payload := bytes.Repeat([]byte{0x17, 0x00, 0xAB}, 500_000) // ETB and NUL laden + sum := sha256.Sum256(payload) + apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-db"}). + Return(fakeImageSaveResult{io.NopCloser(bytes.NewReader(payload))}, nil) + + cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull") + cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1") + + service := types.ServiceConfig{ + Name: "db", + Provider: &types.ServiceProviderConfig{Type: "fake"}, + } + variables, err := svc.(*composeService).executePlugin(t.Context(), &types.Project{Name: "proj"}, cmd, "pull", service) + assert.NilError(t, err) + assert.Equal(t, variables.prefixed["SHA"], hex.EncodeToString(sum[:])) + assert.Equal(t, variables.prefixed["LEN"], strconv.Itoa(len(payload))) +} + +// TestHelperProviderPull is not a test: it is the fake provider process +// spawned by TestExecutePlugin_GetImage. It requests the image, decodes the +// chunked stream with the stock reader, and reports what it received. +func TestHelperProviderPull(t *testing.T) { + if os.Getenv("GO_WANT_HELPER_PROCESS") != "1" { + t.Skip("helper process for TestExecutePlugin_GetImage") + } + emit := func(msg JsonMessage) { + if err := json.NewEncoder(os.Stdout).Encode(msg); err != nil { + os.Exit(1) + } + } + emit(JsonMessage{Type: GetImageType, Message: "proj-db"}) + + stdin := bufio.NewReader(os.Stdin) + line, err := stdin.ReadString('\n') + if err != nil { + emit(JsonMessage{Type: ErrorType, Message: "reading announce: " + err.Error()}) + os.Exit(1) + } + var announce imageStreamAnswer + if err := json.Unmarshal([]byte(line), &announce); err != nil || announce.Error != "" || announce.Encoding != "chunked" { + emit(JsonMessage{Type: ErrorType, Message: fmt.Sprintf("bad announce %q (err %v)", line, err)}) + os.Exit(1) + } + h := sha256.New() + n, err := io.Copy(h, httputil.NewChunkedReader(stdin)) + if err != nil { + emit(JsonMessage{Type: ErrorType, Message: "reading stream: " + err.Error()}) + os.Exit(1) + } + emit(JsonMessage{Type: SetEnvType, Message: "SHA=" + hex.EncodeToString(h.Sum(nil))}) + emit(JsonMessage{Type: SetEnvType, Message: fmt.Sprintf("LEN=%d", n)}) + os.Exit(0) +} + +// TestExecutePlugin_GetImageRefusedOutsidePull: image distribution belongs to +// the pull command — a get-image emitted during up is a protocol error, so an +// image export can never stall another lifecycle command. +func TestExecutePlugin_GetImageRefusedOutsidePull(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + + cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull") + cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1") + + service := types.ServiceConfig{ + Name: "db", + Provider: &types.ServiceProviderConfig{Type: "fake"}, + } + _, err = svc.(*composeService).executePlugin(t.Context(), &types.Project{Name: "proj"}, cmd, "up", service) + assert.ErrorContains(t, err, "get-image is only supported during the pull command") +} + +// failingReader errors after serving its prefix, simulating a daemon export +// dying mid-stream. +type failingReader struct{ data []byte } + +func (f *failingReader) Read(p []byte) (int, error) { + if len(f.data) == 0 { + return 0, errors.New("export died mid-stream") + } + n := copy(p, f.data) + f.data = f.data[n:] + return n, nil +} +func (f *failingReader) Close() error { return nil } + +// TestExecutePlugin_GetImageAbortedMidStream: a transfer failing after the +// success announce must end observably — compose closes the answer channel, +// the provider reads EOF mid-chunk and reports, instead of blocking forever +// on a stream nobody will finish. +func TestExecutePlugin_GetImageAbortedMidStream(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + + apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-db"}). + Return(&failingReader{data: bytes.Repeat([]byte{0xAB}, 4096)}, nil) + + cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull") + cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1") + + service := types.ServiceConfig{ + Name: "db", + Provider: &types.ServiceProviderConfig{Type: "fake"}, + } + _, err = svc.(*composeService).executePlugin(t.Context(), &types.Project{Name: "proj"}, cmd, "pull", service) + assert.ErrorContains(t, err, "reading stream") +} diff --git a/pkg/compose/pull.go b/pkg/compose/pull.go index 45ba070ef..2456cbeb0 100644 --- a/pkg/compose/pull.go +++ b/pkg/compose/pull.go @@ -64,6 +64,7 @@ type imagePuller struct { } func (s *composeService) pull(ctx context.Context, project *types.Project, opts api.PullOptions) error { + rootCtx := ctx images, _, err := s.getLocalImagesDigests(ctx, project) if err != nil { return err @@ -97,8 +98,20 @@ func (s *composeService) pull(ctx context.Context, project *types.Project, opts logrus.Warnf("WARNING: Some service image(s) must be built from source by running:\n docker compose build %s", strings.Join(p.mustBuild, " ")) } + // provider-backed services: pull is the strong contract — the provider + // must ensure its runtime holds the latest version of the authority. + // It runs even when a regular pull failed: skipping the provider phase + // because an unrelated service failed would silently leave provider + // runtimes stale. Its error is a per-service pull error and follows + // their exact regime — reported below, suppressed by IgnoreFailures. + // rootCtx, not ctx: the errgroup context above is canceled once Wait + // returns, and would kill the provider on the spot. + p.pullErrors = append(p.pullErrors, s.ensureProviderImages(rootCtx, project, nil, providerPullPolicyAlways)) + if err != nil { - return err + // fatal errgroup error: report it with whatever per-service context + // accumulated (provider phase included) instead of dropping it + return errors.Join(append(p.pullErrors, err)...) } if opts.IgnoreFailures { return nil diff --git a/pkg/compose/run.go b/pkg/compose/run.go index 1f233fe32..081a90829 100644 --- a/pkg/compose/run.go +++ b/pkg/compose/run.go @@ -160,7 +160,7 @@ func (s *composeService) prepareRun(ctx context.Context, project *types.Project, // Only ensure image exists for the target service, dependencies were already handled by startDependencies buildOpts := prepareBuildOptions(options) - if err := s.ensureImagesExists(ctx, project, buildOpts, options.QuietPull); err != nil { // all dependencies already checked, but might miss service img + if err := s.ensureImagesExists(ctx, project, buildOpts, options.QuietPull, false); err != nil { // all dependencies already checked, but might miss service img return prepareRunResult{}, err }