diff --git a/pkg/client/deploy/deploy.go b/pkg/client/deploy/deploy.go index 1640f34a..dd897216 100644 --- a/pkg/client/deploy/deploy.go +++ b/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) diff --git a/pkg/client/deploy/resolver.go b/pkg/client/deploy/resolver.go index 91b69880..7dfdbb57 100644 --- a/pkg/client/deploy/resolver.go +++ b/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 +}