c8d/pull: Handle pull all tags (2nd approach)
- alternative to: https://github.com/moby/moby/pull/46516
- closes https://github.com/moby/moby/pull/46516

**distribution: Add Tags**

Add a function to return tags for the given repository reference. This
is needed to implement the `pull -a` (pull all tags) for containerd
which doesn't directly use distribution, but we need to somehow make an
API call to the registry to obtain the available tags.

**c8d/pull: Handle pull all tags**

Use the distribution code to query the remote repository for tags and
pull them sequentially just like the non-c8d pull.


**- How to verify it**
```diff
$ docker pull -a docker/doodle
-Error response from daemon: failed to resolve reference "docker.io/docker/doodle": object required
+birthday: Pulling from docker/doodle
+ecaf003f35ec: Download complete
+Digest: sha256:4a203cbea73f32a4885f8d9ee05233246445aa95dd4362887058ec663a7b8088
+Status: Downloaded newer image for docker/doodle:birthday
+birthday2019: Pulling from docker/doodle
+Digest: sha256:4a203cbea73f32a4885f8d9ee05233246445aa95dd4362887058ec663a7b8088
+Status: Downloaded newer image for docker/doodle:birthday2019
(...)
+summer: Pulling from docker/doodle
+6fe738df11e6: Download complete
+Digest: sha256:f7fdfd4a030d16a619f32f183307c9f60ddd9206dc92b1c3b1e1c2f1450567fc
+Status: Downloaded newer image for docker/doodle:summer
+summer2019: Pulling from docker/doodle
+Digest: sha256:f7fdfd4a030d16a619f32f183307c9f60ddd9206dc92b1c3b1e1c2f1450567fc
+Status: Downloaded newer image for docker/doodle:summer2019
+docker.io/docker/doodle


```

**- Description for the changelog**
```release-note
- containerd image store: Implement pulling all tags (`docker pull -a`)
```

**- A picture of a cute animal (not mandatory but encouraged)**


distribution/pull.go
package distribution // import "github.com/docker/docker/distribution"

import (
	"context"
	"fmt"

	"github.com/containerd/log"
	"github.com/distribution/reference"
	"github.com/docker/docker/api"
	"github.com/docker/docker/api/types/events"
	refstore "github.com/docker/docker/reference"
	"github.com/opencontainers/go-digest"
	"github.com/pkg/errors"
)

// Pull initiates a pull operation. image is the repository name to pull, and
// tag may be either empty, or indicate a specific tag to pull.
func Pull(ctx context.Context, ref reference.Named, config *ImagePullConfig, local ContentStore) error {
	// Resolve the Repository name from fqn to RepositoryInfo
	repoInfo, err := config.RegistryService.ResolveRepository(ref)
	if err != nil {
		return err
	}

	// makes sure name is not `scratch`
	if err := validateRepoName(repoInfo.Name); err != nil {
		return err
	}

	endpoints, err := config.RegistryService.LookupPullEndpoints(reference.Domain(repoInfo.Name))
	if err != nil {
		return err
	}

	var (
		lastErr error

		// confirmedTLSRegistries is a map indicating which registries
		// are known to be using TLS. There should never be a plaintext
		// retry for any of these.
		confirmedTLSRegistries = make(map[string]struct{})
	)
	for _, endpoint := range endpoints {
		if endpoint.URL.Scheme != "https" {
			if _, confirmedTLS := confirmedTLSRegistries[endpoint.URL.Host]; confirmedTLS {
				log.G(ctx).Debugf("Skipping non-TLS endpoint %s for host/port that appears to use TLS", endpoint.URL)
				continue
			}
		}

		log.G(ctx).Debugf("Trying to pull %s from %s", reference.FamiliarName(repoInfo.Name), endpoint.URL)

		if err := newPuller(endpoint, repoInfo, config, local).pull(ctx, ref); err != nil {
			// Was this pull cancelled? If so, don't try to fall
			// back.
			fallback := false
			select {
			case <-ctx.Done():
			default:
				if fallbackErr, ok := err.(fallbackError); ok {
					fallback = true
					if fallbackErr.transportOK && endpoint.URL.Scheme == "https" {
						confirmedTLSRegistries[endpoint.URL.Host] = struct{}{}
					}
					err = fallbackErr.err
				}
			}
			if fallback {
				lastErr = err
				log.G(ctx).Infof("Attempting next endpoint for pull after error: %v", err)
				continue
			}
			log.G(ctx).Errorf("Not continuing with pull after error: %v", err)
			return translatePullError(err, ref)
		}

		config.ImageEventLogger(reference.FamiliarString(ref), reference.FamiliarName(repoInfo.Name), events.ActionPull)
		return nil
	}

	if lastErr == nil {
		lastErr = fmt.Errorf("no endpoints found for %s", reference.FamiliarString(ref))
	}

	return translatePullError(lastErr, ref)
}

// validateRepoName validates the name of a repository.
func validateRepoName(name reference.Named) error {
	if reference.FamiliarName(name) == api.NoBaseImageSpecifier {
		return errors.WithStack(reservedNameError(api.NoBaseImageSpecifier))
	}
	return nil
}

func addDigestReference(store refstore.Store, ref reference.Named, dgst digest.Digest, id digest.Digest) error {
	dgstRef, err := reference.WithDigest(reference.TrimNamed(ref), dgst)
	if err != nil {
		return err
	}

	if oldTagID, err := store.Get(dgstRef); err == nil {
		if oldTagID != id {
			// Updating digests not supported by reference store
			log.G(context.TODO()).Errorf("Image ID for digest %s changed from %s to %s, cannot update", dgst.String(), oldTagID, id)
		}
		return nil
	} else if err != refstore.ErrDoesNotExist {
		return err
	}

	return store.AddDigest(dgstRef, id, true)
}

package distribution // import "github.com/docker/docker/distribution"

import (
	"context"
	"fmt"

	"github.com/containerd/log"
	"github.com/distribution/reference"
	"github.com/docker/docker/api"
	"github.com/docker/docker/api/types/events"
	refstore "github.com/docker/docker/reference"
	"github.com/docker/docker/registry"
	"github.com/opencontainers/go-digest"
	"github.com/pkg/errors"
)

// Pull initiates a pull operation. image is the repository name to pull, and
// tag may be either empty, or indicate a specific tag to pull.
func Pull(ctx context.Context, ref reference.Named, config *ImagePullConfig, local ContentStore) error {
	repoInfo, err := pullEndpoints(ctx, config.RegistryService, ref, func(ctx context.Context, repoInfo registry.RepositoryInfo, endpoint registry.APIEndpoint) error {
		log.G(ctx).Debugf("Trying to pull %s from %s", reference.FamiliarName(repoInfo.Name), endpoint.URL)
		puller := newPuller(endpoint, &repoInfo, config, local)
		return puller.pull(ctx, ref)
	})

	if err == nil {
		config.ImageEventLogger(reference.FamiliarString(ref), reference.FamiliarName(repoInfo.Name), events.ActionPull)
	}

	return err
}

// Tags returns available tags for the given image in the remote repository.
func Tags(ctx context.Context, ref reference.Named, config *Config) ([]string, error) {
	var tags []string
	_, err := pullEndpoints(ctx, config.RegistryService, ref, func(ctx context.Context, repoInfo registry.RepositoryInfo, endpoint registry.APIEndpoint) error {
		repo, err := newRepository(ctx, &repoInfo, endpoint, config.MetaHeaders, config.AuthConfig, "pull")
		if err != nil {
			return err
		}

		tags, err = repo.Tags(ctx).All(ctx)
		return err
	})

	return tags, err
}

// validateRepoName validates the name of a repository.
func validateRepoName(name reference.Named) error {
	if reference.FamiliarName(name) == api.NoBaseImageSpecifier {
		return errors.WithStack(reservedNameError(api.NoBaseImageSpecifier))
	}
	return nil
}

func addDigestReference(store refstore.Store, ref reference.Named, dgst digest.Digest, id digest.Digest) error {
	dgstRef, err := reference.WithDigest(reference.TrimNamed(ref), dgst)
	if err != nil {
		return err
	}

	if oldTagID, err := store.Get(dgstRef); err == nil {
		if oldTagID != id {
			// Updating digests not supported by reference store
			log.G(context.TODO()).Errorf("Image ID for digest %s changed from %s to %s, cannot update", dgst.String(), oldTagID, id)
		}
		return nil
	} else if err != refstore.ErrDoesNotExist {
		return err
	}

	return store.AddDigest(dgstRef, id, true)
}

func pullEndpoints(ctx context.Context, registryService RegistryResolver, ref reference.Named,
	f func(context.Context, registry.RepositoryInfo, registry.APIEndpoint) error,
) (*registry.RepositoryInfo, error) {
	// Resolve the Repository name from fqn to RepositoryInfo
	repoInfo, err := registryService.ResolveRepository(ref)
	if err != nil {
		return nil, err
	}

	// makes sure name is not `scratch`
	if err := validateRepoName(repoInfo.Name); err != nil {
		return repoInfo, err
	}

	endpoints, err := registryService.LookupPullEndpoints(reference.Domain(repoInfo.Name))
	if err != nil {
		return repoInfo, err
	}

	var (
		lastErr error

		// confirmedTLSRegistries is a map indicating which registries
		// are known to be using TLS. There should never be a plaintext
		// retry for any of these.
		confirmedTLSRegistries = make(map[string]struct{})
	)
	for _, endpoint := range endpoints {
		if endpoint.URL.Scheme != "https" {
			if _, confirmedTLS := confirmedTLSRegistries[endpoint.URL.Host]; confirmedTLS {
				log.G(ctx).Debugf("Skipping non-TLS endpoint %s for host/port that appears to use TLS", endpoint.URL)
				continue
			}
		}

		log.G(ctx).Debugf("Trying to pull %s from %s", reference.FamiliarName(repoInfo.Name), endpoint.URL)

		if err := f(ctx, *repoInfo, endpoint); err != nil {
			if _, ok := err.(fallbackError); !ok && continueOnError(err, endpoint.Mirror) {
				err = fallbackError{
					err:         err,
					transportOK: true,
				}
			}

			// Was this pull cancelled? If so, don't try to fall
			// back.
			fallback := false
			select {
			case <-ctx.Done():
			default:
				if fallbackErr, ok := err.(fallbackError); ok {
					fallback = true
					if fallbackErr.transportOK && endpoint.URL.Scheme == "https" {
						confirmedTLSRegistries[endpoint.URL.Host] = struct{}{}
					}
					err = fallbackErr.err
				}
			}
			if fallback {
				lastErr = err
				log.G(ctx).Infof("Attempting next endpoint for pull after error: %v", err)
				continue
			}
			log.G(ctx).Errorf("Not continuing with pull after error: %v", err)
			return repoInfo, translatePullError(err, ref)
		}

		return repoInfo, nil
	}

	if lastErr == nil {
		lastErr = fmt.Errorf("no endpoints found for %s", reference.FamiliarString(ref))
	}

	return repoInfo, translatePullError(lastErr, ref)
}

Was looking if we could move this to an `internal` package, but that complicates things as well. Perhaps an alternative could be to call it out in the GoDoc as "for internal use", but we can do so in a follow-up if we think that's a good idea.