containerd integration: image pull
**- What I did**

Upstreaming:

* https://github.com/rumpl/moby/pull/12
* https://github.com/rumpl/moby/pull/16
* ~~https://github.com/rumpl/moby/pull/42~~
* https://github.com/rumpl/moby/pull/64
* https://github.com/rumpl/moby/pull/80
* https://github.com/rumpl/moby/pull/84

Implements `docker pull` with containerd.

~~Also includes stargz and nydus snapshotter handling, these two are lazy ones and need additional handlers during pull~~ (pre-emptively removing this from the changeset to avoid too much comments, will add this one later)

Limiting concurrent downloads is not yet implemented, see [here](https://github.com/rumpl/moby/pull/95) for the discussion, the way containerd handles limits is different than what moby does, I think we can implement limiting in a followup

_Note: I will squash the commits once it's reviewed, it's easier to review with different commits_

daemon/containerd/resolver.go
package containerd

import (
	"github.com/containerd/containerd/remotes"
	"github.com/containerd/containerd/remotes/docker"
	registrytypes "github.com/docker/docker/api/types/registry"
	"github.com/docker/docker/registry"
	"github.com/sirupsen/logrus"
)

func newResolverFromAuthConfig(authConfig *registrytypes.AuthConfig) remotes.Resolver {
	opts := []docker.RegistryOpt{}
	if authConfig != nil {
		cfgHost := registry.ConvertToHostname(authConfig.ServerAddress)
		if cfgHost == registry.IndexHostname {
			cfgHost = registry.DefaultRegistryHost
		}
		authorizer := docker.NewDockerAuthorizer(docker.WithAuthCreds(func(host string) (string, string, error) {
			if cfgHost != host {
				logrus.WithField("host", host).WithField("cfgHost", cfgHost).Warn("Host doesn't match")
				return "", "", nil
			}
			if authConfig.IdentityToken != "" {
				return "", authConfig.IdentityToken, nil
			}
			return authConfig.Username, authConfig.Password, nil
		}))

		opts = append(opts, docker.WithAuthorizer(authorizer))
	}

	return docker.NewResolver(docker.ResolverOptions{
		Hosts: docker.ConfigureDefaultRegistries(opts...),
	})
}

package containerd

import (
	"net/http"
	"strings"

	"github.com/containerd/containerd/remotes"
	"github.com/containerd/containerd/remotes/docker"
	registrytypes "github.com/docker/docker/api/types/registry"
	"github.com/docker/docker/registry"
	"github.com/sirupsen/logrus"
)

func (i *ImageService) newResolverFromAuthConfig(authConfig *registrytypes.AuthConfig) (remotes.Resolver, docker.StatusTracker) {
	tracker := docker.NewInMemoryTracker()
	hostsFn := i.registryHosts.RegistryHosts()

	hosts := hostsWrapper(hostsFn, authConfig, i.registryService)

	return docker.NewResolver(docker.ResolverOptions{
		Hosts:   hosts,
		Tracker: tracker,
	}), tracker
}

func hostsWrapper(hostsFn docker.RegistryHosts, authConfig *registrytypes.AuthConfig, regService registry.Service) docker.RegistryHosts {
	return func(n string) ([]docker.RegistryHost, error) {
		hosts, err := hostsFn(n)
		if err != nil {
			return nil, err
		}

		for i := range hosts {
			if hosts[i].Authorizer == nil {
				var opts []docker.AuthorizerOpt
				if authConfig != nil {
					opts = append(opts, authorizationCredsFromAuthConfig(*authConfig))
				}
				hosts[i].Authorizer = docker.NewDockerAuthorizer(opts...)

				isInsecure := regService.IsInsecureRegistry(hosts[i].Host)
				if hosts[i].Client.Transport != nil && isInsecure {
					hosts[i].Client.Transport = httpFallback{super: hosts[i].Client.Transport}
				}
			}
		}
		return hosts, nil
	}
}

func authorizationCredsFromAuthConfig(authConfig registrytypes.AuthConfig) docker.AuthorizerOpt {
	cfgHost := registry.ConvertToHostname(authConfig.ServerAddress)
	if cfgHost == registry.IndexHostname {
		cfgHost = registry.DefaultRegistryHost
	}

	return docker.WithAuthCreds(func(host string) (string, string, error) {
		if cfgHost != host {
			logrus.WithField("host", host).WithField("cfgHost", cfgHost).Warn("Host doesn't match")
			return "", "", nil
		}
		if authConfig.IdentityToken != "" {
			return "", authConfig.IdentityToken, nil
		}
		return authConfig.Username, authConfig.Password, nil
	})
}

type httpFallback struct {
	super http.RoundTripper
}

func (f httpFallback) RoundTrip(r *http.Request) (*http.Response, error) {
	resp, err := f.super.RoundTrip(r)
	if err != nil {
		if strings.Contains(err.Error(), "http: server gave HTTP response to HTTPS client") {
			plain := r.Clone(r.Context())
			plain.URL.Scheme = "http"
			return http.DefaultTransport.RoundTrip(plain)
		}
	}

	return resp, err
}

https://github.com/golang/go/wiki/CodeReviewComments#initialisms
https://pkg.go.dev/net/http#Request.Clone

```suggestion
			plain := r.Clone(r.Context())
			plain.URL.Scheme = "http"
			return http.DefaultTransport.RoundTrip(plain)
```