Merge pull request #32388 from nishanttotla/pin-by-digest-on-client-alternative

Moving docker service digest pinning to client side

api/server/router/swarm/backend.go

@@ -19,8 +19,8 @@ type Backend interface {
 
 	GetServices(basictypes.ServiceListOptions) ([]types.Service, error)
 	GetService(idOrName string, insertDefaults bool) (types.Service, error)
-	CreateService(types.ServiceSpec, string) (*basictypes.ServiceCreateResponse, error)
-	UpdateService(string, uint64, types.ServiceSpec, basictypes.ServiceUpdateOptions) (*basictypes.ServiceUpdateResponse, error)
+	CreateService(types.ServiceSpec, string, bool) (*basictypes.ServiceCreateResponse, error)
+	UpdateService(string, uint64, types.ServiceSpec, basictypes.ServiceUpdateOptions, bool) (*basictypes.ServiceUpdateResponse, error)
 	RemoveService(string) error
 
 	ServiceLogs(context.Context, *backend.LogSelector, *basictypes.ContainerLogsOptions) (<-chan *backend.LogMessage, error)

api/server/router/swarm/cluster_routes.go

@@ -13,6 +13,7 @@ import (
 	"github.com/docker/docker/api/types/backend"
 	"github.com/docker/docker/api/types/filters"
 	types "github.com/docker/docker/api/types/swarm"
+	"github.com/docker/docker/api/types/versions"
 	"golang.org/x/net/context"
 )
 
@@ -178,8 +179,13 @@ func (sr *swarmRouter) createService(ctx context.Context, w http.ResponseWriter,
 
 	// Get returns "" if the header does not exist
 	encodedAuth := r.Header.Get("X-Registry-Auth")
+	cliVersion := r.Header.Get("version")
+	queryRegistry := false
+	if cliVersion != "" && versions.LessThan(cliVersion, "1.30") {
+		queryRegistry = true
+	}
 
-	resp, err := sr.backend.CreateService(service, encodedAuth)
+	resp, err := sr.backend.CreateService(service, encodedAuth, queryRegistry)
 	if err != nil {
 		logrus.Errorf("Error creating service %s: %v", service.Name, err)
 		return err
@@ -207,8 +213,13 @@ func (sr *swarmRouter) updateService(ctx context.Context, w http.ResponseWriter,
 	flags.EncodedRegistryAuth = r.Header.Get("X-Registry-Auth")
 	flags.RegistryAuthFrom = r.URL.Query().Get("registryAuthFrom")
 	flags.Rollback = r.URL.Query().Get("rollback")
+	cliVersion := r.Header.Get("version")
+	queryRegistry := false
+	if cliVersion != "" && versions.LessThan(cliVersion, "1.30") {
+		queryRegistry = true
+	}
 
-	resp, err := sr.backend.UpdateService(vars["id"], version, service, flags)
+	resp, err := sr.backend.UpdateService(vars["id"], version, service, flags, queryRegistry)
 	if err != nil {
 		logrus.Errorf("Error updating service %s: %v", vars["id"], err)
 		return err

api/types/client.go

@@ -276,6 +276,12 @@ type ServiceCreateOptions struct {
 	//
 	// This field follows the format of the X-Registry-Auth header.
 	EncodedRegistryAuth string
+
+	// QueryRegistry indicates whether the service update requires
+	// contacting a registry. A registry may be contacted to retrieve
+	// the image digest and manifest, which in turn can be used to update
+	// platform or other information about the service.
+	QueryRegistry bool
 }
 
 // ServiceCreateResponse contains the information returned to a client
@@ -315,6 +321,12 @@ type ServiceUpdateOptions struct {
 	// The valid values are "previous" and "none". An empty value is the
 	// same as "none".
 	Rollback string
+
+	// QueryRegistry indicates whether the service update requires
+	// contacting a registry. A registry may be contacted to retrieve
+	// the image digest and manifest, which in turn can be used to update
+	// platform or other information about the service.
+	QueryRegistry bool
 }
 
 // ServiceListOptions holds parameters to list services with.

client/distribution_inspect.go

@@ -0,0 +1,31 @@
+package client
+
+import (
+	"encoding/json"
+	"net/url"
+
+	registrytypes "github.com/docker/docker/api/types/registry"
+	"golang.org/x/net/context"
+)
+
+// DistributionInspect returns the image digest with full Manifest
+func (cli *Client) DistributionInspect(ctx context.Context, image, encodedRegistryAuth string) (registrytypes.DistributionInspect, error) {
+	var headers map[string][]string
+
+	if encodedRegistryAuth != "" {
+		headers = map[string][]string{
+			"X-Registry-Auth": {encodedRegistryAuth},
+		}
+	}
+
+	// Contact the registry to retrieve digest and platform information
+	var distributionInspect registrytypes.DistributionInspect
+	resp, err := cli.get(ctx, "/distribution/"+image+"/json", url.Values{}, headers)
+	if err != nil {
+		return distributionInspect, err
+	}
+
+	err = json.NewDecoder(resp.body).Decode(&distributionInspect)
+	ensureReaderClosed(resp)
+	return distributionInspect, err
+}

client/interface.go

@@ -20,6 +20,7 @@ import (
 type CommonAPIClient interface {
 	ConfigAPIClient
 	ContainerAPIClient
+	DistributionAPIClient
 	ImageAPIClient
 	NodeAPIClient
 	NetworkAPIClient
@@ -69,6 +70,11 @@ type ContainerAPIClient interface {
 	ContainersPrune(ctx context.Context, pruneFilters filters.Args) (types.ContainersPruneReport, error)
 }
 
+// DistributionAPIClient defines API client methods for the registry
+type DistributionAPIClient interface {
+	DistributionInspect(ctx context.Context, image, encodedRegistryAuth string) (registry.DistributionInspect, error)
+}
+
 // ImageAPIClient defines API client methods for the images
 type ImageAPIClient interface {
 	ImageBuild(ctx context.Context, context io.Reader, options types.ImageBuildOptions) (types.ImageBuildResponse, error)

client/service_create.go

@@ -2,22 +2,39 @@ package client
 
 import (
 	"encoding/json"
+	"fmt"
 
+	"github.com/docker/distribution/reference"
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/swarm"
+	"github.com/opencontainers/go-digest"
 	"golang.org/x/net/context"
 )
 
 // ServiceCreate creates a new Service.
 func (cli *Client) ServiceCreate(ctx context.Context, service swarm.ServiceSpec, options types.ServiceCreateOptions) (types.ServiceCreateResponse, error) {
-	var headers map[string][]string
+	var distErr error
+
+	headers := map[string][]string{
+		"version": {cli.version},
+	}
 
 	if options.EncodedRegistryAuth != "" {
-		headers = map[string][]string{
-			"X-Registry-Auth": {options.EncodedRegistryAuth},
-		}
+		headers["X-Registry-Auth"] = []string{options.EncodedRegistryAuth}
 	}
 
+	// Contact the registry to retrieve digest and platform information
+	if options.QueryRegistry {
+		distributionInspect, err := cli.DistributionInspect(ctx, service.TaskTemplate.ContainerSpec.Image, options.EncodedRegistryAuth)
+		distErr = err
+		if err == nil {
+			// now pin by digest if the image doesn't already contain a digest
+			img := imageWithDigestString(service.TaskTemplate.ContainerSpec.Image, distributionInspect.Descriptor.Digest)
+			if img != "" {
+				service.TaskTemplate.ContainerSpec.Image = img
+			}
+		}
+	}
 	var response types.ServiceCreateResponse
 	resp, err := cli.post(ctx, "/services/create", nil, service, headers)
 	if err != nil {
@@ -25,6 +42,38 @@ func (cli *Client) ServiceCreate(ctx context.Context, service swarm.ServiceSpec,
 	}
 
 	err = json.NewDecoder(resp.body).Decode(&response)
+
+	if distErr != nil {
+		response.Warnings = append(response.Warnings, digestWarning(service.TaskTemplate.ContainerSpec.Image))
+	}
+
 	ensureReaderClosed(resp)
 	return response, err
 }
+
+// imageWithDigestString takes an image string and a digest, and updates
+// the image string if it didn't originally contain a digest. It assumes
+// that the image string is not an image ID
+func imageWithDigestString(image string, dgst digest.Digest) string {
+	isCanonical := false
+	ref, err := reference.ParseAnyReference(image)
+	if err == nil {
+		_, isCanonical = ref.(reference.Canonical)
+
+		if !isCanonical {
+			namedRef, _ := ref.(reference.Named)
+			img, err := reference.WithDigest(namedRef, dgst)
+			if err == nil {
+				return img.String()
+			}
+		}
+	}
+	return ""
+}
+
+// digestWarning constructs a formatted warning string using the
+// image name that could not be pinned by digest. The formatting
+// is hardcoded, but could me made smarter in the future
+func digestWarning(image string) string {
+	return fmt.Sprintf("image %s could not be accessed on a registry to record\nits digest. Each node will access %s independently,\npossibly leading to different nodes running different\nversions of the image.\n", image, image)
+}

client/service_update.go

@@ -13,14 +13,16 @@ import (
 // ServiceUpdate updates a Service.
 func (cli *Client) ServiceUpdate(ctx context.Context, serviceID string, version swarm.Version, service swarm.ServiceSpec, options types.ServiceUpdateOptions) (types.ServiceUpdateResponse, error) {
 	var (
-		headers map[string][]string
 		query   = url.Values{}
+		distErr error
 	)
 
+	headers := map[string][]string{
+		"version": {cli.version},
+	}
+
 	if options.EncodedRegistryAuth != "" {
-		headers = map[string][]string{
-			"X-Registry-Auth": {options.EncodedRegistryAuth},
-		}
+		headers["X-Registry-Auth"] = []string{options.EncodedRegistryAuth}
 	}
 
 	if options.RegistryAuthFrom != "" {
@@ -33,6 +35,20 @@ func (cli *Client) ServiceUpdate(ctx context.Context, serviceID string, version
 
 	query.Set("version", strconv.FormatUint(version.Index, 10))
 
+	// Contact the registry to retrieve digest and platform information
+	// This happens only when the image has changed
+	if options.QueryRegistry {
+		distributionInspect, err := cli.DistributionInspect(ctx, service.TaskTemplate.ContainerSpec.Image, options.EncodedRegistryAuth)
+		distErr = err
+		if err == nil {
+			// now pin by digest if the image doesn't already contain a digest
+			img := imageWithDigestString(service.TaskTemplate.ContainerSpec.Image, distributionInspect.Descriptor.Digest)
+			if img != "" {
+				service.TaskTemplate.ContainerSpec.Image = img
+			}
+		}
+	}
+
 	var response types.ServiceUpdateResponse
 	resp, err := cli.post(ctx, "/services/"+serviceID+"/update", query, service, headers)
 	if err != nil {
@@ -40,6 +56,11 @@ func (cli *Client) ServiceUpdate(ctx context.Context, serviceID string, version
 	}
 
 	err = json.NewDecoder(resp.body).Decode(&response)
+
+	if distErr != nil {
+		response.Warnings = append(response.Warnings, digestWarning(service.TaskTemplate.ContainerSpec.Image))
+	}
+
 	ensureReaderClosed(resp)
 	return response, err
 }

client/utils.go

@@ -1,9 +1,10 @@
 package client
 
 import (
-	"github.com/docker/docker/api/types/filters"
 	"net/url"
 	"regexp"
+
+	"github.com/docker/docker/api/types/filters"
 )
 
 var headerRegexp = regexp.MustCompile(`\ADocker/.+\s\((.+)\)\z`)

daemon/cluster/services.go

@@ -117,7 +117,7 @@ func (c *Cluster) GetService(input string, insertDefaults bool) (types.Service,
 }
 
 // CreateService creates a new service in a managed swarm cluster.
-func (c *Cluster) CreateService(s types.ServiceSpec, encodedAuth string) (*apitypes.ServiceCreateResponse, error) {
+func (c *Cluster) CreateService(s types.ServiceSpec, encodedAuth string, queryRegistry bool) (*apitypes.ServiceCreateResponse, error) {
 	var resp *apitypes.ServiceCreateResponse
 	err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
 		err := c.populateNetworkID(ctx, state.controlClient, &s)
@@ -151,8 +151,11 @@ func (c *Cluster) CreateService(s types.ServiceSpec, encodedAuth string) (*apity
 				}
 			}
 
-			// pin image by digest
-			if os.Getenv("DOCKER_SERVICE_PREFER_OFFLINE_IMAGE") != "1" {
+			// pin image by digest for API versions < 1.30
+			// TODO(nishanttotla): The check on "DOCKER_SERVICE_PREFER_OFFLINE_IMAGE"
+			// should be removed in the future. Since integration tests only use the
+			// latest API version, so this is no longer required.
+			if os.Getenv("DOCKER_SERVICE_PREFER_OFFLINE_IMAGE") != "1" && queryRegistry {
 				digestImage, err := c.imageWithDigestString(ctx, ctnr.Image, authConfig)
 				if err != nil {
 					logrus.Warnf("unable to pin image %s to digest: %s", ctnr.Image, err.Error())
@@ -193,7 +196,7 @@ func (c *Cluster) CreateService(s types.ServiceSpec, encodedAuth string) (*apity
 }
 
 // UpdateService updates existing service to match new properties.
-func (c *Cluster) UpdateService(serviceIDOrName string, version uint64, spec types.ServiceSpec, flags apitypes.ServiceUpdateOptions) (*apitypes.ServiceUpdateResponse, error) {
+func (c *Cluster) UpdateService(serviceIDOrName string, version uint64, spec types.ServiceSpec, flags apitypes.ServiceUpdateOptions, queryRegistry bool) (*apitypes.ServiceUpdateResponse, error) {
 	var resp *apitypes.ServiceUpdateResponse
 
 	err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
@@ -256,8 +259,11 @@ func (c *Cluster) UpdateService(serviceIDOrName string, version uint64, spec typ
 
 		resp = &apitypes.ServiceUpdateResponse{}
 
-		// pin image by digest
-		if os.Getenv("DOCKER_SERVICE_PREFER_OFFLINE_IMAGE") != "1" {
+		// pin image by digest for API versions < 1.30
+		// TODO(nishanttotla): The check on "DOCKER_SERVICE_PREFER_OFFLINE_IMAGE"
+		// should be removed in the future. Since integration tests only use the
+		// latest API version, so this is no longer required.
+		if os.Getenv("DOCKER_SERVICE_PREFER_OFFLINE_IMAGE") != "1" && queryRegistry {
 			digestImage, err := c.imageWithDigestString(ctx, newCtnr.Image, authConfig)
 			if err != nil {
 				logrus.Warnf("unable to pin image %s to digest: %s", newCtnr.Image, err.Error())