c8d: Implement push
- upstreams: https://github.com/rumpl/moby/pull/2
- upstreams: https://github.com/rumpl/moby/pull/74
- upstreams: https://github.com/rumpl/moby/pull/80
- upstreams: https://github.com/rumpl/moby/pull/15


Implements `docker push` under containerd image store.  When
pushing manifest lists that reference a content which is not present in
the local content store, it will attempt to perform the cross-repo mount
the content if possible.

Considering this scenario:

```bash
$ docker pull docker.io/library/busybox
```
This will download manifest list and only host platform-specific
manifest and blobs.

Note, tagging to a different repository (but still the same registry) and pushing:
```bash
$ docker tag docker.io/library/busybox docker.io/private-repo/mybusybox
$ docker push docker.io/private-repo/mybusybox
```

will result in error, because the neither we nor the target repository
doesn't have the manifests that the busybox manifest list references
(because manifests can't be cross-repo mounted).

If for some reason the manifests and configs for all other platforms
would be present in the content store, but only layer blobs were
missing, then the push would work, because the blobs can be cross-repo
mounted (only if we push to the same registry).


The `container.io/distribution.source` label is taken only from the target blob we are pushing. It's a change from fork's behavior.
In fork for each missing content we did search the entire content store for a manifest that includes it and took it's distribution source label:
https://github.com/moby/moby/blob/2670c94d8298523ca90db1b097d0c4ebfc3ccc56/daemon/containerd/lazy_push.go#L367-L489

This was mostly a workaround for buildkit image exporter not fetching all platform-specific content (only the host's platform). Eg. doing `FROM 'ubuntu' | docker buildx build --platform <host>,<other> -` would only fetch ubuntu rootfs for `<host>` platform, but not for `<other>`. Because the built image didn't have any `distribution.source` that would point to the original `docker.io/library/ubuntu` - it was impossible to know where to obtain the blob from, based on the built image alone. This was only possible by traversing all manifests in content store, checking if they have a `distribution.source` label and then traversing them to check if they happen to define this blob.
Since this is no longer needed as buildkit already does `store=true` by default, I decided to change this.

**- What I did**
Implemented push

**- How I did it**
Use containerd remote to push the content
Some missing content can be cross-repo mounted. In this case fake their existence by wrapping the `content.Store` to return fake `content.Info` that includes the `containerd.io/distribution.source` label with the source repository.
That makes the push not end with NotFound error and allows it to perform cross-repo mount.

<details>

<summary>**How to verify it [click to expand]**</summary>

TODO: New examples
</details>


**- A picture of a cute animal (not mandatory but encouraged)**
![mobbunny](https://user-images.githubusercontent.com/5046555/217814826-56c699b1-ff48-4858-8e4c-7a11fc74f783.png)


daemon/images/image_push.go
package images // import "github.com/docker/docker/daemon/images"

import (
	"context"
	"io"
	"time"

	"github.com/docker/distribution/manifest/schema2"
	"github.com/docker/distribution/reference"
	"github.com/docker/docker/api/types/registry"
	"github.com/docker/docker/distribution"
	progressutils "github.com/docker/docker/distribution/utils"
	"github.com/docker/docker/pkg/progress"
)

// PushImage initiates a push operation on the repository named localName.
func (i *ImageService) PushImage(ctx context.Context, image, tag string, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error {
	start := time.Now()
	ref, err := reference.ParseNormalizedNamed(image)
	if err != nil {
		return err
	}
	if tag != "" {
		// Push by digest is not supported, so only tags are supported.
		ref, err = reference.WithTag(ref, tag)
		if err != nil {
			return err
		}
	}

	// Include a buffer so that slow client connections don't affect
	// transfer performance.
	progressChan := make(chan progress.Progress, 100)

	writesDone := make(chan struct{})

	ctx, cancelFunc := context.WithCancel(ctx)

	go func() {
		progressutils.WriteDistributionProgress(cancelFunc, outStream, progressChan)
		close(writesDone)
	}()

	imagePushConfig := &distribution.ImagePushConfig{
		Config: distribution.Config{
			MetaHeaders:      metaHeaders,
			AuthConfig:       authConfig,
			ProgressOutput:   progress.ChanOutput(progressChan),
			RegistryService:  i.registryService,
			ImageEventLogger: i.LogImageEvent,
			MetadataStore:    i.distributionMetadataStore,
			ImageStore:       distribution.NewImageConfigStoreFromStore(i.imageStore),
			ReferenceStore:   i.referenceStore,
		},
		ConfigMediaType: schema2.MediaTypeImageConfig,
		LayerStores:     distribution.NewLayerProvidersFromStore(i.layerStore),
		UploadManager:   i.uploadManager,
	}

	err = distribution.Push(ctx, ref, imagePushConfig)
	close(progressChan)
	<-writesDone
	imageActions.WithValues("push").UpdateSince(start)
	return err
}

package images // import "github.com/docker/docker/daemon/images"

import (
	"context"
	"io"
	"time"

	"github.com/docker/distribution/manifest/schema2"
	"github.com/docker/distribution/reference"
	"github.com/docker/docker/api/types/registry"
	"github.com/docker/docker/distribution"
	progressutils "github.com/docker/docker/distribution/utils"
	"github.com/docker/docker/pkg/progress"
)

// PushImage initiates a push operation on the repository named localName.
func (i *ImageService) PushImage(ctx context.Context, ref reference.Named, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error {
	start := time.Now()
	// Include a buffer so that slow client connections don't affect
	// transfer performance.
	progressChan := make(chan progress.Progress, 100)

	writesDone := make(chan struct{})

	ctx, cancelFunc := context.WithCancel(ctx)

	go func() {
		progressutils.WriteDistributionProgress(cancelFunc, outStream, progressChan)
		close(writesDone)
	}()

	imagePushConfig := &distribution.ImagePushConfig{
		Config: distribution.Config{
			MetaHeaders:      metaHeaders,
			AuthConfig:       authConfig,
			ProgressOutput:   progress.ChanOutput(progressChan),
			RegistryService:  i.registryService,
			ImageEventLogger: i.LogImageEvent,
			MetadataStore:    i.distributionMetadataStore,
			ImageStore:       distribution.NewImageConfigStoreFromStore(i.imageStore),
			ReferenceStore:   i.referenceStore,
		},
		ConfigMediaType: schema2.MediaTypeImageConfig,
		LayerStores:     distribution.NewLayerProvidersFromStore(i.layerStore),
		UploadManager:   i.uploadManager,
	}

	err := distribution.Push(ctx, ref, imagePushConfig)
	close(progressChan)
	<-writesDone
	imageActions.WithValues("push").UpdateSince(start)
	return err
}

Ooooooh, I hadn't fully realized until just now that containerd integration means we _could_ support push-by-digest now! :eyes: