daemon/c8d: Implement import
**- What I did**
Implemented `docker import` for containerd integration

**- How I did it**
Write the blob to the content store, calculate the uncompressed data digest and create a manifest and image config for the selected platform (default is host platform if none provided).
I also extracted the common logic (parsing image name into reference and getting a data stream) to the API routes.

If the imported layer archive is uncompressed, it gets compressed with gzip before writing to the content store.
Archives compressed with gzip and zstd are imported as-is.
Xz and bzip2 are recompressed into gzip.

**- How to verify it**

1. Export some image from non-c8d Docker
```bash
$ docker run --name asdf ubuntu sh -c 'apt-get -y update && apt-get install -y fish'
$ docker export asdf >asdf.tar
```

2. Compress it with supported compressions (gz, zstd, xz, bzip2)
3. Import and verify with ctr (docker run also works, but is not yet upstreamed)
<details>
<summary>raw</summary>

```bash
$ docker import asdf.tar asdf:raw
sha256:5cebbe53a996ec2ad6c4101143a8c13f8d6999e5144a75ed21570fb96c93bd6b

# Manifest
$ ctr content get "sha256:5cebbe53a996ec2ad6c4101143a8c13f8d6999e5144a75ed21570fb96c93bd6b"
{
  "schemaVersion": 2,
  "mediaType": "application/vnd.oci.image.manifest.v1+json",
  "config": {
    "mediaType": "application/vnd.oci.image.config.v1+json",
    "digest": "sha256:1150939d6b988df2a1febb4b0fcbdc5245eb4c8335061b2ee239f878f0530a8e",
    "size": 291
  },
  "layers": [
    {
      "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
      "digest": "sha256:0953a62d60750e4932d6dbd7f79b9cf41c371f768ab4a7c7d384c1cdc75ffc99",
      "size": 9934617
    }
  ]
}


# Config
$ ctr content get "sha256:1150939d6b988df2a1febb4b0fcbdc5245eb4c8335061b2ee239f878f0530a8e"
{
  "created": "2022-12-16T14:54:19.593382044Z",
  "architecture": "arm64",
  "os": "linux",
  "config": {},
  "rootfs": {
    "type": "layers",
    "diff_ids": [
      "sha256:4aa1b69e24b6c5dfde91e20440f2feb0c2e85ec0596adef1bd1107005268a826"
    ]
  },
  "history": [
    {
      "created": "2022-12-16T14:54:19.593382044Z",
      "comment": "Imported from -"
    }
  ]
}


# Layer
$ ctr content get "sha256:0953a62d60750e4932d6dbd7f79b9cf41c371f768ab4a7c7d384c1cdc75ffc99" | gzip -d | tar --list | grep 'bin/fish'
usr/bin/fish
usr/bin/fish_indent
usr/bin/fish_key_reader

```
</details>


<details>
<summary>gz</summary>

```bash
$ docker import asdf.tar.gz asdf:gz
sha256:c2b5ae5a12eac4c397445289c1864c5d658b0e958ff1cd83276ccb48d32edadb

# Manifest
$ ctr content get "sha256:c2b5ae5a12eac4c397445289c1864c5d658b0e958ff1cd83276ccb48d32edadb"
{
  "schemaVersion": 2,
  "mediaType": "application/vnd.oci.image.manifest.v1+json",
  "config": {
    "mediaType": "application/vnd.oci.image.config.v1+json",
    "digest": "sha256:892ea7918be6bf49d44b7ef410d4c4eb5212afad5aa101ef563540f51442521c",
    "size": 291
  },
  "layers": [
    {
      "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
      "digest": "sha256:ad31c4743384e7d6be2518fd5d411d4175566057a7e38f6018f577d3132a4d71",
      "size": 9695278
    }
  ]
}


# Config
$ ctr content get "sha256:892ea7918be6bf49d44b7ef410d4c4eb5212afad5aa101ef563540f51442521c"
{
  "created": "2022-12-16T14:54:20.091913877Z",
  "architecture": "arm64",
  "os": "linux",
  "config": {},
  "rootfs": {
    "type": "layers",
    "diff_ids": [
      "sha256:4aa1b69e24b6c5dfde91e20440f2feb0c2e85ec0596adef1bd1107005268a826"
    ]
  },
  "history": [
    {
      "created": "2022-12-16T14:54:20.091913877Z",
      "comment": "Imported from -"
    }
  ]
}


# Layer
$ ctr content get "sha256:ad31c4743384e7d6be2518fd5d411d4175566057a7e38f6018f577d3132a4d71" | gzip -d | tar --list | grep 'bin/fish'
usr/bin/fish
usr/bin/fish_indent
usr/bin/fish_key_reader

```
</details>


<details>
<summary>zstd</summary>

```bash
$ docker import asdf.tar.zst asdf:zstd
sha256:a49f3a56f566e6a9741b5b0a568768c007cb4a962c506cbb2867d283384795d0

# Manifest
$ ctr content get "sha256:a49f3a56f566e6a9741b5b0a568768c007cb4a962c506cbb2867d283384795d0"
{
  "schemaVersion": 2,
  "mediaType": "application/vnd.oci.image.manifest.v1+json",
  "config": {
    "mediaType": "application/vnd.oci.image.config.v1+json",
    "digest": "sha256:f5357c0fb73a3681c1ae09749df39e2b1d3e379ca2c92498a3aafe020414e413",
    "size": 291
  },
  "layers": [
    {
      "mediaType": "application/vnd.oci.image.layer.v1.tar+zstd",
      "digest": "sha256:7f7585dbea4e0ea7aa7790d4b81f0c324fe801e77e28a372e46e4c96adda3fc0",
      "size": 9513145
    }
  ]
}


# Config
$ ctr content get "sha256:f5357c0fb73a3681c1ae09749df39e2b1d3e379ca2c92498a3aafe020414e413"
{
  "created": "2022-12-16T14:54:20.491820044Z",
  "architecture": "arm64",
  "os": "linux",
  "config": {},
  "rootfs": {
    "type": "layers",
    "diff_ids": [
      "sha256:4aa1b69e24b6c5dfde91e20440f2feb0c2e85ec0596adef1bd1107005268a826"
    ]
  },
  "history": [
    {
      "created": "2022-12-16T14:54:20.491820044Z",
      "comment": "Imported from -"
    }
  ]
}


# Layer
$ ctr content get "sha256:7f7585dbea4e0ea7aa7790d4b81f0c324fe801e77e28a372e46e4c96adda3fc0" | zstd -d | tar --list | grep 'bin/fish'
usr/bin/fish
usr/bin/fish_indent
usr/bin/fish_key_reader

```
</details>


<details>
<summary>xz</summary>

```bash
$ docker import asdf.tar.xz asdf:xz
sha256:0c867e1189a713d170c8afec8f75b9dc67f204d9e49c95b8626a99a848cd681a

# Manifest
$ ctr content get "sha256:0c867e1189a713d170c8afec8f75b9dc67f204d9e49c95b8626a99a848cd681a"
{
  "schemaVersion": 2,
  "mediaType": "application/vnd.oci.image.manifest.v1+json",
  "config": {
    "mediaType": "application/vnd.oci.image.config.v1+json",
    "digest": "sha256:de62b7c145c11db01ddebfe4373a224c462c7885fa9b5006dde141c18ba157f6",
    "size": 291
  },
  "layers": [
    {
      "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
      "digest": "sha256:0953a62d60750e4932d6dbd7f79b9cf41c371f768ab4a7c7d384c1cdc75ffc99",
      "size": 9934617
    }
  ]
}


# Config
$ ctr content get "sha256:de62b7c145c11db01ddebfe4373a224c462c7885fa9b5006dde141c18ba157f6"
{
  "created": "2022-12-16T14:54:21.443434628Z",
  "architecture": "arm64",
  "os": "linux",
  "config": {},
  "rootfs": {
    "type": "layers",
    "diff_ids": [
      "sha256:4aa1b69e24b6c5dfde91e20440f2feb0c2e85ec0596adef1bd1107005268a826"
    ]
  },
  "history": [
    {
      "created": "2022-12-16T14:54:21.443434628Z",
      "comment": "Imported from -"
    }
  ]
}


# Layer
$ ctr content get "sha256:0953a62d60750e4932d6dbd7f79b9cf41c371f768ab4a7c7d384c1cdc75ffc99" | gzip -d | tar --list | grep 'bin/fish'
usr/bin/fish
usr/bin/fish_indent
usr/bin/fish_key_reader

```
</details>


<details>
<summary>bz2</summary>

```bash
$ docker import asdf.tar.bz2 asdf:bz2
sha256:d52042ccf7a9b929544e8511d76f02d58f0162dd9acb1d837ad1717dd2e27dde

# Manifest
$ ctr content get "sha256:d52042ccf7a9b929544e8511d76f02d58f0162dd9acb1d837ad1717dd2e27dde"
{
  "schemaVersion": 2,
  "mediaType": "application/vnd.oci.image.manifest.v1+json",
  "config": {
    "mediaType": "application/vnd.oci.image.config.v1+json",
    "digest": "sha256:47bfa38ef6af5ca2e4c8b0460d367fcce5430443f7ce22400bc3ad233e415014",
    "size": 291
  },
  "layers": [
    {
      "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
      "digest": "sha256:0953a62d60750e4932d6dbd7f79b9cf41c371f768ab4a7c7d384c1cdc75ffc99",
      "size": 9934617
    }
  ]
}


# Config
$ ctr content get "sha256:47bfa38ef6af5ca2e4c8b0460d367fcce5430443f7ce22400bc3ad233e415014"
{
  "created": "2022-12-16T14:54:23.271932879Z",
  "architecture": "arm64",
  "os": "linux",
  "config": {},
  "rootfs": {
    "type": "layers",
    "diff_ids": [
      "sha256:4aa1b69e24b6c5dfde91e20440f2feb0c2e85ec0596adef1bd1107005268a826"
    ]
  },
  "history": [
    {
      "created": "2022-12-16T14:54:23.271932879Z",
      "comment": "Imported from -"
    }
  ]
}


# Layer
$ ctr content get "sha256:0953a62d60750e4932d6dbd7f79b9cf41c371f768ab4a7c7d384c1cdc75ffc99" | gzip -d | tar --list | grep 'bin/fish'
usr/bin/fish
usr/bin/fish_indent
usr/bin/fish_key_reader

```
</details>


**- Description for the changelog**
Support `docker import` when containerd snapshotter is enabled


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


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

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/containerd/containerd/platforms"
	"github.com/docker/distribution/reference"
	"github.com/docker/docker/api/types/container"
	"github.com/docker/docker/builder/dockerfile"
	"github.com/docker/docker/builder/remotecontext"
	"github.com/docker/docker/dockerversion"
	"github.com/docker/docker/errdefs"
	"github.com/docker/docker/image"
	"github.com/docker/docker/layer"
	"github.com/docker/docker/pkg/archive"
	"github.com/docker/docker/pkg/progress"
	"github.com/docker/docker/pkg/streamformatter"
	"github.com/docker/docker/pkg/system"
	specs "github.com/opencontainers/image-spec/specs-go/v1"
	"github.com/pkg/errors"
)

// ImportImage imports an image, getting the archived layer data either from
// inConfig (if src is "-"), or from a URI specified in src. Progress output is
// written to outStream. Repository and tag names can optionally be given in
// the repo and tag arguments, respectively.
func (i *ImageService) ImportImage(ctx context.Context, src string, repository string, platform *specs.Platform, tag string, msg string, inConfig io.ReadCloser, outStream io.Writer, changes []string) error {
	var (
		rc     io.ReadCloser
		resp   *http.Response
		newRef reference.Named
	)

	if repository != "" {
		var err error
		newRef, err = reference.ParseNormalizedNamed(repository)
		if err != nil {
			return errdefs.InvalidParameter(err)
		}
		if _, isCanonical := newRef.(reference.Canonical); isCanonical {
			return errdefs.InvalidParameter(errors.New("cannot import digest reference"))
		}

		if tag != "" {
			newRef, err = reference.WithTag(newRef, tag)
			if err != nil {
				return errdefs.InvalidParameter(err)
			}
		}
	}

	// Normalize platform - default to the operating system and architecture if not supplied.
	if platform == nil {
		p := platforms.DefaultSpec()
		platform = &p
	}
	if !system.IsOSSupported(platform.OS) {
		return errdefs.InvalidParameter(system.ErrNotSupportedOperatingSystem)
	}
	config, err := dockerfile.BuildFromConfig(ctx, &container.Config{}, changes, platform.OS)
	if err != nil {
		return err
	}
	if src == "-" {
		rc = inConfig
	} else {
		inConfig.Close()
		if len(strings.Split(src, "://")) == 1 {
			src = "http://" + src
		}
		u, err := url.Parse(src)
		if err != nil {
			return errdefs.InvalidParameter(err)
		}

		resp, err = remotecontext.GetWithStatusError(u.String())
		if err != nil {
			return err
		}
		outStream.Write(streamformatter.FormatStatus("", "Downloading from %s", u))
		progressOutput := streamformatter.NewJSONProgressOutput(outStream, true)
		rc = progress.NewProgressReader(resp.Body, progressOutput, resp.ContentLength, "", "Importing")
	}

	defer rc.Close()
	if len(msg) == 0 {
		msg = "Imported from " + src
	}

	inflatedLayerData, err := archive.DecompressStream(rc)
	if err != nil {
		return err
	}
	l, err := i.layerStore.Register(inflatedLayerData, "")
	if err != nil {
		return err
	}
	defer layer.ReleaseAndLog(i.layerStore, l)

	created := time.Now().UTC()
	imgConfig, err := json.Marshal(&image.Image{
		V1Image: image.V1Image{
			DockerVersion: dockerversion.Version,
			Config:        config,
			Architecture:  platform.Architecture,
			Variant:       platform.Variant,
			OS:            platform.OS,
			Created:       created,
			Comment:       msg,
		},
		RootFS: &image.RootFS{
			Type:    "layers",
			DiffIDs: []layer.DiffID{l.DiffID()},
		},
		History: []image.History{{
			Created: created,
			Comment: msg,
		}},
	})
	if err != nil {
		return err
	}

	id, err := i.imageStore.Create(imgConfig)
	if err != nil {
		return err
	}

	// FIXME: connect with commit code and call refstore directly
	if newRef != nil {
		if err := i.TagImageWithReference(id, newRef); err != nil {
			return err
		}
	}

	i.LogImageEvent(id.String(), id.String(), "import")
	outStream.Write(streamformatter.FormatStatus("", id.String()))
	return nil
}

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

import (
	"context"
	"encoding/json"
	"io"
	"time"

	"github.com/containerd/containerd/platforms"
	"github.com/docker/distribution/reference"
	"github.com/docker/docker/api/types/container"
	"github.com/docker/docker/builder/dockerfile"
	"github.com/docker/docker/dockerversion"
	"github.com/docker/docker/errdefs"
	"github.com/docker/docker/image"
	"github.com/docker/docker/layer"
	"github.com/docker/docker/pkg/archive"
	"github.com/docker/docker/pkg/system"
	specs "github.com/opencontainers/image-spec/specs-go/v1"
)

// ImportImage imports an image, getting the archived layer data from layerReader.
// Uncompressed layer archive is passed to the layerStore and handled by the
// underlying graph driver.
// Image is tagged with the given reference.
// If the platform is nil, the default host platform is used.
// Message is used as the image's history comment.
// Image configuration is derived from the dockerfile instructions in changes.
func (i *ImageService) ImportImage(ctx context.Context, newRef reference.Named, platform *specs.Platform, msg string, layerReader io.Reader, changes []string) (image.ID, error) {
	if platform == nil {
		def := platforms.DefaultSpec()
		platform = &def
	}
	if !system.IsOSSupported(platform.OS) {
		return "", errdefs.InvalidParameter(system.ErrNotSupportedOperatingSystem)
	}

	config, err := dockerfile.BuildFromConfig(ctx, &container.Config{}, changes, platform.OS)
	if err != nil {
		return "", errdefs.InvalidParameter(err)
	}

	inflatedLayerData, err := archive.DecompressStream(layerReader)
	if err != nil {
		return "", err
	}
	l, err := i.layerStore.Register(inflatedLayerData, "")
	if err != nil {
		return "", err
	}
	defer layer.ReleaseAndLog(i.layerStore, l)

	created := time.Now().UTC()
	imgConfig, err := json.Marshal(&image.Image{
		V1Image: image.V1Image{
			DockerVersion: dockerversion.Version,
			Config:        config,
			Architecture:  platform.Architecture,
			Variant:       platform.Variant,
			OS:            platform.OS,
			Created:       created,
			Comment:       msg,
		},
		RootFS: &image.RootFS{
			Type:    "layers",
			DiffIDs: []layer.DiffID{l.DiffID()},
		},
		History: []image.History{{
			Created: created,
			Comment: msg,
		}},
	})
	if err != nil {
		return "", err
	}

	id, err := i.imageStore.Create(imgConfig)
	if err != nil {
		return "", err
	}

	if newRef != nil {
		if err := i.TagImageWithReference(id, newRef); err != nil {
			return "", err
		}
	}

	i.LogImageEvent(id.String(), id.String(), "import")
	return id, nil
}

Thanks, fixed!