feat(compose): partially implement image resolver (pinner) but disable for now

pkg/client/deploy/deploy.go

@@ -11,6 +11,7 @@ import (
 type Client interface {
 	api.ContainerClient
 	api.DNSClient
+	api.ImageClient
 	api.MachineClient
 	api.ServiceClient
 }
@@ -67,10 +68,15 @@ func (d *Deployment) Plan(ctx context.Context) (Plan, error) {
 	if err != nil && !errors.Is(err, api.ErrNotFound) {
 		return Plan{}, fmt.Errorf("get cluster domain: %w", err)
 	}
-
+	// TODO: enable image resolver when it's ready to pin digests and look up existing images.
+	//imageResolver := &ImageDigestResolver{
+	//	Ctx:    ctx,
+	//	Client: d.cli,
+	//}
 	specResolver := &ServiceSpecResolver{
 		// If the domain is not found (not reserved), an empty domain is used for the resolver.
 		ClusterDomain: clusterDomain,
+		//ImageResolver: imageResolver,
 	}
 
 	resolvedSpec, err := specResolver.Resolve(d.Spec)

pkg/client/deploy/resolver.go

@@ -1,21 +1,23 @@
 package deploy
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"github.com/distribution/reference"
+	"github.com/docker/docker/api/types"
+	"github.com/opencontainers/go-digest"
 	"github.com/psviderski/uncloud/internal/secret"
 	"github.com/psviderski/uncloud/pkg/api"
+	"google.golang.org/grpc/codes"
 	"strings"
+	"time"
 )
 
-type ImageDigestResolver interface {
-	Resolve(image string) (string, error)
-}
-
 // ServiceSpecResolver transforms user-provided service specs into deployment-ready form.
 type ServiceSpecResolver struct {
 	ClusterDomain string
-	ImageResolver ImageDigestResolver
+	ImageResolver *ImageDigestResolver
 }
 
 // Resolve transforms a service spec into its fully resolved form ready for deployment.
@@ -29,8 +31,8 @@ func (r *ServiceSpecResolver) Resolve(spec api.ServiceSpec) (api.ServiceSpec, er
 	steps := []func(*api.ServiceSpec) error{
 		r.applyDefaults,
 		r.resolveServiceName,
-		r.resolveImageDigest,
 		r.expandIngressPorts,
+		r.resolveImageDigest,
 	}
 
 	for _, step := range steps {
@@ -85,21 +87,6 @@ func (r *ServiceSpecResolver) resolveServiceName(spec *api.ServiceSpec) error {
 	return nil
 }
 
-func (r *ServiceSpecResolver) resolveImageDigest(spec *api.ServiceSpec) error {
-	if r.ImageResolver == nil {
-		// Skip digest resolution when no resolver is provided.
-		return nil
-	}
-
-	imageDigest, err := r.ImageResolver.Resolve(spec.Container.Image)
-	if err != nil {
-		return fmt.Errorf("resolve image digest: %w", err)
-	}
-	spec.Container.Image = imageDigest
-
-	return nil
-}
-
 // expandIngressPorts processes HTTP(S) ingress ports in a service spec by:
 // 1. Setting a default hostname (service-name.cluster-domain) for ports without a hostname.
 // 2. Duplicating a port with a cluster domain hostname for ports with external domains.
@@ -138,3 +125,186 @@ func (r *ServiceSpecResolver) expandIngressPorts(spec *api.ServiceSpec) error {
 
 	return nil
 }
+
+func (r *ServiceSpecResolver) resolveImageDigest(spec *api.ServiceSpec) error {
+	if r.ImageResolver == nil {
+		// Skip digest resolution when no resolver is provided.
+		return nil
+	}
+
+	image, err := r.ImageResolver.Resolve(spec.Container.Image, spec.Container.PullPolicy)
+	if err != nil {
+		return fmt.Errorf("resolve image digest: %w", err)
+	}
+	spec.Container.Image = image
+
+	return nil
+}
+
+type ImageResolverClient interface {
+	api.ImageClient
+	api.MachineClient
+}
+
+type ImageDigestResolver struct {
+	Ctx    context.Context
+	Client ImageResolverClient
+}
+
+func (r *ImageDigestResolver) Resolve(image, policy string) (string, error) {
+	if r.Ctx == nil {
+		r.Ctx = context.Background()
+	}
+
+	ref, err := reference.ParseNormalizedNamed(image)
+	if err != nil {
+		return "", fmt.Errorf("parse image: %w", err)
+	}
+	if _, ok := ref.(reference.Canonical); ok {
+		// The image is already pinned to a digest.
+		return image, nil
+	}
+
+	switch policy {
+	case api.PullPolicyAlways:
+		return r.resolveAlways(image)
+	case api.PullPolicyMissing:
+		return r.resolveMissing(image)
+	case api.PullPolicyNever:
+		return "", fmt.Errorf("pull policy '%s' is not supported yet", policy)
+	}
+	return image, nil
+}
+
+// resolveAlways resolves the image to the image with the digest by querying the registry from all machines.
+func (r *ImageDigestResolver) resolveAlways(image string) (string, error) {
+	// TODO: broadcast to a subset of machines in large clusters to avoid being rate-limited by the registry.
+	ctx, err := api.ProxyMachinesContext(r.Ctx, r.Client, nil)
+	if err != nil {
+		return "", fmt.Errorf("create request context to broadcast to all machines: %w", err)
+	}
+
+	remoteImages, err := r.Client.InspectRemoteImage(ctx, image)
+	if err != nil {
+		return "", fmt.Errorf("inspect image '%s' in registry from all machines: %w", image, err)
+	}
+	if len(remoteImages) == 0 {
+		return "", fmt.Errorf("inspect image '%s' in registry from all machines: unexpected empty response", image)
+	}
+
+	for _, ri := range remoteImages {
+		if ri.Metadata != nil && ri.Metadata.Error != "" {
+			// Save the last error to return it if all machines fail to inspect the image.
+			err = fmt.Errorf("inspect image '%s' in registry on machine '%s': %s",
+				image, ri.Metadata.Machine, ri.Metadata.Error)
+			continue
+		}
+
+		return reference.FamiliarString(ri.Image.Reference), nil
+	}
+
+	return "", err
+}
+
+func (r *ImageDigestResolver) resolveMissing(image string) (string, error) {
+	ctx, err := api.ProxyMachinesContext(r.Ctx, r.Client, nil)
+	if err != nil {
+		return "", fmt.Errorf("create request context to broadcast to all machines: %w", err)
+	}
+
+	machineImages, err := r.Client.InspectImage(ctx, image)
+	if err != nil {
+		if errors.Is(err, api.ErrNotFound) {
+			// If the image is missing on all machines, the 'missing' policy is equivalent to 'always'.
+			return r.resolveAlways(image)
+		}
+		return "", fmt.Errorf("inspect image '%s' on all machines: %w", image, err)
+	}
+
+	var availableImages []types.ImageInspect
+	for _, mi := range machineImages {
+		// Metadata can be nil if the request was proxied to only one machine.
+		if mi.Metadata != nil && mi.Metadata.Error != "" {
+			if codes.Code(mi.Metadata.Status.Code) != codes.NotFound {
+				fmt.Printf("WARNING: failed to inspect image '%s' on machine '%s': %s\n",
+					image, mi.Metadata.Machine, mi.Metadata.Error)
+			}
+			continue
+		}
+
+		availableImages = append(availableImages, mi.Image)
+	}
+
+	if len(availableImages) == 0 {
+		// If the image is missing on all machines, the 'missing' policy is equivalent to 'always'.
+		return r.resolveAlways(image)
+	}
+
+	// Find the latest image with a RepoDigest.
+	var latestDigest digest.Digest
+	var latestCreated time.Time
+	for _, img := range availableImages {
+		if len(img.RepoDigests) == 0 {
+			continue
+		}
+
+		// TODO: handle multiple RepoDigests. This could happen for example if the same image was pulled twice using
+		// 	both its index (multi-arch) digest and manifest (platform-specific) digest:
+		//  {
+		//    "Id": "sha256:6fee7566e4273ee6078f08e167e36434b35f72152232a5e6f1446288817dabe5",
+		//    "RepoTags": [
+		//        "traefik/whoami:latest"
+		//    ],
+		//    "RepoDigests": [
+		//        "traefik/whoami@sha256:200689790a0a0ea48ca45992e0450bc26ccab5307375b41c84dfc4f2475937ab",
+		//        "traefik/whoami@sha256:4f90b33ddca9c4d4f06527070d6e503b16d71016edea036842be2a84e60c91cb"
+		//    ],
+		//    ...
+		//  }
+		//  Should the registry be queried to find out which digest to use?
+		repoDigest := img.RepoDigests[0]
+		created, err := time.Parse(time.RFC3339Nano, img.Created)
+		if err != nil {
+			continue
+		}
+
+		if created.After(latestCreated) {
+			ref, err := reference.ParseNormalizedNamed(repoDigest)
+			if err != nil {
+				continue
+			}
+			if c, ok := ref.(reference.Canonical); ok {
+				latestDigest = c.Digest()
+				latestCreated = created
+			}
+		}
+	}
+
+	if latestDigest != "" {
+		return imageWithDigest(image, latestDigest)
+	}
+
+	// Don't pin the digest if no RepoDigests were found. This means the available images were not pulled from
+	// a registry but built locally or loaded from an archive. In this case, the available images (could be multiple
+	// for different platforms) should be copied to other machines to be able to run service containers on them.
+	return image, nil
+}
+
+// imageWithDigest adds a digest to an image string if it doesn't already contain one.
+func imageWithDigest(image string, dig digest.Digest) (string, error) {
+	ref, err := reference.ParseNormalizedNamed(image)
+	if err != nil {
+		return "", fmt.Errorf("parse image: %w", err)
+	}
+
+	if _, ok := ref.(reference.Canonical); !ok {
+		// Preserves the original tag if present.
+		img, err := reference.WithDigest(ref, dig)
+		if err != nil {
+			return "", fmt.Errorf("add digest to image: %w", err)
+		}
+		return reference.FamiliarString(img), nil
+	}
+
+	return image, nil
+}