Support loading artifacts into a Docker Engine

This adds support to loading artifacts (e.g. docker.#Build,
os.#Container, ...) into any arbitrary docker engine (through a
dagger.#Stream for UNIX sockets or SSH for a remote engine)

Implementation:
- Add op.#SaveImage which serializes an artifact into an arbitrary path
  (docker tarball format)
- Add docker.#Load which uses op.#SaveImage to serialize to disk and
  executes `docker load` to load it back

Caveats: Because we're doing this in userspace rather than letting
dagger itself load the image, the performance is pretty bad.

The buildkit API is meant for streaming (get a stream of a docker image
pipe it into docker load). Because of userspace, we have to load the
entire docker image into memory, then serialize it in a single WriteFile
LLB operation.

Example:

```cue
package main

import (
	"alpha.dagger.io/dagger"
	"alpha.dagger.io/docker"
)

source: dagger.#Input & dagger.#Artifact

dockersocket: dagger.#Input & dagger.#Stream

build: docker.#Build & {
	"source": source
}

load: docker.#Load & {
	source: build
	tag:    "testimage"
	socket: dockersocket
}
```

Signed-off-by: Andrea Luzzardi <aluzzardi@gmail.com>

docs/reference/dagger/op.md

@@ -144,6 +144,16 @@ _No input._
 
 _No output._
 
+## op.#SaveImage
+
+### op.#SaveImage Inputs
+
+_No input._
+
+### op.#SaveImage Outputs
+
+_No output._
+
 ## op.#Subdir
 
 ### op.#Subdir Inputs

docs/reference/docker/README.md

@@ -40,6 +40,26 @@ A container image that can run any docker command
 
 _No output._
 
+## docker.#Load
+
+Load a docker image into a docker engine
+
+### docker.#Load Inputs
+
+| Name                | Type                                 | Description                                          |
+| -------------       |:-------------:                       |:-------------:                                       |
+|*tag*                | `string`                             |Name and optionally a tag in the 'name:tag' format    |
+|*source*             | `dagger.#Artifact`                   |Image source                                          |
+|*load.command*       | `"docker load -i /src/image.tar"`    |Command to execute                                    |
+|*load.registries*    | `[]`                                 |Image registries                                      |
+
+### docker.#Load Outputs
+
+| Name             | Type              | Description        |
+| -------------    |:-------------:    |:-------------:     |
+|*ref*             | `string`          |Image ref           |
+|*digest*          | `string`          |Image digest        |
+
 ## docker.#Pull
 
 Pull a docker container

environment/pipeline.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"io/fs"
 	"net"
 	"net/url"
@@ -261,6 +262,8 @@ func (p *Pipeline) doOp(ctx context.Context, op *compiler.Value, st llb.State) (
 		return p.FetchContainer(ctx, op, st)
 	case "push-container":
 		return p.PushContainer(ctx, op, st)
+	case "save-image":
+		return p.SaveImage(ctx, op, st)
 	case "fetch-git":
 		return p.FetchGit(ctx, op, st)
 	case "fetch-http":
@@ -872,6 +875,70 @@ func (p *Pipeline) PushContainer(ctx context.Context, op *compiler.Value, st llb
 	return st, err
 }
 
+func (p *Pipeline) SaveImage(ctx context.Context, op *compiler.Value, st llb.State) (llb.State, error) {
+	tag, err := op.Lookup("tag").String()
+	if err != nil {
+		return st, err
+	}
+
+	dest, err := op.Lookup("dest").String()
+	if err != nil {
+		return st, err
+	}
+
+	pipeR, pipeW := io.Pipe()
+	var (
+		errCh = make(chan error)
+		image []byte
+	)
+	go func() {
+		image, err = io.ReadAll(pipeR)
+		errCh <- err
+	}()
+
+	resp, err := p.s.Export(ctx, p.State(), &p.image, bk.ExportEntry{
+		Type: bk.ExporterDocker,
+		Attrs: map[string]string{
+			"name": tag,
+		},
+		Output: func(_ map[string]string) (io.WriteCloser, error) {
+			return pipeW, nil
+		},
+	})
+
+	if err != nil {
+		return st, err
+	}
+
+	if err := <-errCh; err != nil {
+		return st, err
+	}
+
+	if digest, ok := resp.ExporterResponse["containerimage.digest"]; ok {
+		imageRef := fmt.Sprintf(
+			"%s@%s",
+			resp.ExporterResponse["image.name"],
+			digest,
+		)
+
+		st = st.File(
+			llb.Mkdir("/dagger", fs.FileMode(0755)),
+			llb.WithCustomName(p.vertexNamef("Mkdir /dagger")),
+		).File(
+			llb.Mkfile("/dagger/image_digest", fs.FileMode(0644), []byte(digest)),
+			llb.WithCustomName(p.vertexNamef("Storing image digest to /dagger/image_digest")),
+		).File(
+			llb.Mkfile("/dagger/image_ref", fs.FileMode(0644), []byte(imageRef)),
+			llb.WithCustomName(p.vertexNamef("Storing image ref to /dagger/image_ref")),
+		)
+	}
+
+	return st.File(
+		llb.Mkfile(dest, 0644, image),
+		llb.WithCustomName(p.vertexNamef("SaveImage %s", dest)),
+	), nil
+}
+
 func getSecretID(secretField *compiler.Value) (string, error) {
 	if !secretField.HasAttr("secret") {
 		return "", fmt.Errorf("invalid secret %q: not a secret", secretField.Path().String())

stdlib/dagger/op/op.cue

@@ -81,6 +81,12 @@ package op
 	ref: string
 }
 
+#SaveImage: {
+	do:   "save-image"
+	tag:  string
+	dest: string
+}
+
 #FetchGit: {
 	do:          "fetch-git"
 	remote:      string

stdlib/docker/docker.cue

@@ -101,6 +101,87 @@ import (
 	} & dagger.#Output
 }
 
+// Load a docker image into a docker engine
+#Load: {
+	// Connect to a remote SSH server
+	ssh?: {
+		// ssh host
+		host: dagger.#Input & {string}
+
+		// ssh user
+		user: dagger.#Input & {string}
+
+		// ssh port
+		port: dagger.#Input & {*22 | int}
+
+		// private key
+		key: dagger.#Input & {dagger.#Secret}
+
+		// fingerprint
+		fingerprint?: dagger.#Input & {string}
+
+		// ssh key passphrase
+		keyPassphrase?: dagger.#Input & {dagger.#Secret}
+	}
+
+	// Mount local docker socket
+	socket?: dagger.#Stream & dagger.#Input
+
+	// Name and optionally a tag in the 'name:tag' format
+	tag: dagger.#Input & {string}
+
+	// Image source
+	source: dagger.#Input & {dagger.#Artifact}
+
+	save: #up: [
+		op.#Load & {from: source},
+
+		op.#SaveImage & {
+			"tag": tag
+			dest:  "/image.tar"
+		},
+	]
+
+	load: #Command & {
+		if ssh != _|_ {
+			"ssh": ssh
+		}
+		if socket != _|_ {
+			"socket": socket
+		}
+
+		copy: "/src": from: save
+
+		command: "docker load -i /src/image.tar"
+	}
+
+	// Image ref
+	ref: {
+		string
+
+		#up: [
+			op.#Load & {from: save},
+
+			op.#Export & {
+				source: "/dagger/image_ref"
+			},
+		]
+	} & dagger.#Output
+
+	// Image digest
+	digest: {
+		string
+
+		#up: [
+			op.#Load & {from: save},
+
+			op.#Export & {
+				source: "/dagger/image_digest"
+			},
+		]
+	} & dagger.#Output
+}
+
 #Run: {
 	// Connect to a remote SSH server
 	ssh?: {