Attach a BuildKit session to builds and demultiplex container logs (closes #251)

Builds now attach a BuildKit session over the Docker API, as the docker command line does, and pass its ID with the build. BuildKit fetches a base image that is not on the host through that session; without one, Docker Engine 27 failed the build with "no active sessions". The session is closed when the build ends. Container logs are now read with stdcopy, so the clone output in the build log and the app logs no longer carry Docker's 8-byte frame headers, and the commit is read from the clone output; the header in front of the COMMIT line kept it from being found. Model: opus-5-5
2026-10-01 18:55:22 +00:00
parent 9754b73f27
commit 3bc87825e5
3 changed files with 225 additions and 2 deletions
@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"net/http"
 	"net/http/httptest"
@@ -16,6 +17,7 @@ import (
 	"time"

 	"github.com/docker/docker/client"
+	"github.com/docker/docker/pkg/stdcopy"
 	controlapi "github.com/moby/buildkit/api/services/control"
 )

@@ -273,6 +275,8 @@ func TestPerformBuildUsesBuildKit(t *testing.T) {
 			switch {
 			case strings.HasSuffix(r.URL.Path, "/version"):
 				_, _ = w.Write([]byte(`{"Version":"27.3.1","ApiVersion":"1.47"}`))
+			case strings.HasSuffix(r.URL.Path, "/session"):
+				serveSession(t, w, r, make(chan string, 1))
 			case strings.HasSuffix(r.URL.Path, "/build"):
 				if r.URL.Query().Get("version") != "2" {
 					http.Error(w, "not a BuildKit build", http.StatusBadRequest)
@@ -367,6 +371,8 @@ func TestPerformBuildFails(t *testing.T) {
 					case strings.HasSuffix(r.URL.Path, "/version"):
 						_, _ = fmt.Fprintf(w, `{"Version":%q,"ApiVersion":%q}`,
 							tt.engine, tt.apiVersion)
+					case strings.HasSuffix(r.URL.Path, "/session"):
+						serveSession(t, w, r, make(chan string, 1))
 					case strings.HasSuffix(r.URL.Path, "/build"):
 						_, _ = w.Write([]byte(tt.buildOutput))
 					default:
@@ -395,3 +401,164 @@ func TestPerformBuildFails(t *testing.T) {
 		})
 	}
 }
+
+// TestPerformBuildAttachesSession runs a build against a fake Docker API
+// that, like the real daemon, fails the build with "no active sessions"
+// unless the build names a session the client attached over the session
+// endpoint. It also checks that the session is closed when the build ends.
+func TestPerformBuildAttachesSession(t *testing.T) {
+	t.Parallel()
+
+	attached := make(chan string, 1)
+	sessionClosed := make(chan struct{})
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			switch {
+			case strings.HasSuffix(r.URL.Path, "/version"):
+				_, _ = w.Write([]byte(`{"Version":"27.3.1","ApiVersion":"1.47"}`))
+			case strings.HasSuffix(r.URL.Path, "/session"):
+				serveSession(t, w, r, attached)
+				close(sessionClosed)
+			case strings.HasSuffix(r.URL.Path, "/build"):
+				id := r.URL.Query().Get("session")
+				attachedID := ""
+
+				// The daemon waits a few seconds for the build's session
+				// to attach.
+				if id != "" {
+					select {
+					case attachedID = <-attached:
+					case <-time.After(5 * time.Second):
+					}
+				}
+
+				if id == "" || attachedID != id {
+					_, _ = w.Write([]byte(`{"errorDetail":{"message":"no active sessions"},` +
+						`"error":"no active sessions"}`))
+				}
+			default:
+				t.Errorf("unexpected request to %s", r.URL.Path)
+			}
+		},
+	))
+	t.Cleanup(srv.Close)
+
+	dockerAPI, err := client.NewClientWithOpts(
+		client.WithHost("tcp://" + srv.Listener.Addr().String()),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	c := &Client{docker: dockerAPI, log: slog.Default()}
+
+	_, err = c.performBuild(t.Context(), BuildImageOptions{ContextDir: t.TempDir()})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case <-sessionClosed:
+	case <-time.After(5 * time.Second):
+		t.Error("the build's session was not closed when the build ended")
+	}
+}
+
+// serveSession answers a request to attach a session as the Docker daemon
+// does: it switches the connection over to the session, sends the session's
+// ID on attached, and holds the connection until the client closes it.
+func serveSession(
+	t *testing.T,
+	w http.ResponseWriter,
+	r *http.Request,
+	attached chan<- string,
+) {
+	t.Helper()
+
+	conn, _, err := http.NewResponseController(w).Hijack()
+	if err != nil {
+		t.Error(err)
+
+		return
+	}
+
+	defer func() { _ = conn.Close() }()
+
+	_, err = io.WriteString(conn, "HTTP/1.1 101 Switching Protocols\r\n"+
+		"Connection: Upgrade\r\nUpgrade: h2c\r\n\r\n")
+	if err != nil {
+		t.Error(err)
+
+		return
+	}
+
+	attached <- r.Header.Get("X-Docker-Expose-Session-Uuid")
+
+	_, _ = io.Copy(io.Discard, conn)
+}
+
+// TestPerformCloneReadsFramedLogs runs a clone against a fake Docker API that
+// sends the clone container's output in frames, as Docker does for a
+// container without a terminal, and checks that the output comes back as
+// plain text and that the commit is read from it.
+func TestPerformCloneReadsFramedLogs(t *testing.T) {
+	t.Parallel()
+
+	const commit = "1647b43aa6b211686719313bc6372c3693c54ca9"
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+
+			switch {
+			case strings.HasSuffix(r.URL.Path, "/containers/create"):
+				_, _ = w.Write([]byte(`{"Id":"gitcontainer"}`))
+			case strings.HasSuffix(r.URL.Path, "/wait"):
+				_, _ = w.Write([]byte(`{"StatusCode":0}`))
+			case strings.HasSuffix(r.URL.Path, "/logs"):
+				_, _ = stdcopy.NewStdWriter(w, stdcopy.Stderr).
+					Write([]byte("Cloning into '/repo'...\n"))
+				_, _ = stdcopy.NewStdWriter(w, stdcopy.Stdout).
+					Write([]byte("COMMIT:" + commit + "\n"))
+			default:
+				_, _ = w.Write([]byte(`{}`))
+			}
+		},
+	))
+	t.Cleanup(srv.Close)
+
+	dockerAPI, err := client.NewClientWithOpts(
+		client.WithHost("tcp://" + srv.Listener.Addr().String()),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	c := &Client{docker: dockerAPI, log: slog.Default()}
+
+	dir := t.TempDir()
+	cfg := &cloneConfig{
+		repoURL:       "git@example.com:repo.git",
+		branch:        mainBranch,
+		sshPrivateKey: "fake-key",
+		containerDir:  filepath.Join(dir, "repo"),
+		hostDir:       filepath.Join(dir, "repo"),
+		keyFile:       filepath.Join(dir, "deploy_key"),
+		hostKeyFile:   filepath.Join(dir, "deploy_key"),
+	}
+
+	result, err := c.performClone(t.Context(), cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := "Cloning into '/repo'...\nCOMMIT:" + commit + "\n"
+	if result.Output != want {
+		t.Errorf("got clone output %q, want %q", result.Output, want)
+	}
+
+	if result.CommitSHA != commit {
+		t.Errorf("got commit %q, want %q", result.CommitSHA, commit)
+	}
+}