From 7106e6c2468d79393a3a88a4fa92966e0cc0859c Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Wed, 29 Jul 2026 14:01:24 +0200 Subject: [PATCH] feat(server): Docker socket client behind the ContainerLifecycle seam The backend for scale-to-zero (#19). Three Engine API endpoints over a unix socket with no external dependency -- the official SDK would add a large dependency tree to a proxy that has none. Still nothing calls it: the idle controller is only wired up when the request gate lands. Reaching this socket is root-equivalent on the host, so the client is only constructed when the operator passes --docker-socket. Two defects were found by attacking the first draft against a live daemon, and both are fixed here rather than left for review: /stop now sends `t`. Docker's SIGTERM-to-SIGKILL wait otherwise comes from the container's own StopTimeout, which the proxy cannot see -- kamal passes `options:` straight through to docker run, and compose has stop_grace_period. Measured against a real daemon: a container created with --stop-timeout 60 made StopContainer return a deadline error at 30s while the daemon carried on and killed the container at 60s. The controller reads that error as a failed stop, rolls the service back to active and puts its targets back -- for a container Docker is about to kill. The service then serves 502s from a live-looking target until the next full idle period. Deriving `t` from the caller's remaining budget puts the daemon's deadline inside ours instead of outside it. Redirects are no longer followed. Docker's router cleans the decoded path and answers 301 to the canonical form, and Go rewrites a redirected POST as a GET. Measured: `ContainerExists("web-1/")` returned nil while `StartContainer` on the same reference returned 404, because GET /json exists and GET /start does not -- so the deploy preflight accepts precisely the reference the wake path can never start. Against a redirecting socket proxy a stop could report success having stopped nothing. Nothing in the Engine API legitimately redirects. Version negotiation caches only a success, so a daemon that was briefly unreachable does not pin the client to the fallback for the life of the process. /version is read with a 1 MB limit where errors get 4 KB: a plugin-heavy host's Components array is far larger than an error body, and truncating that JSON would silently drop a healthy daemon to the fallback version. 304 is success on both start and stop, which is what makes a coalesced wake against an already-running container work. Refs #19 --- internal/server/docker_client.go | 245 ++++++++++++++++ internal/server/docker_client_test.go | 390 ++++++++++++++++++++++++++ 2 files changed, 635 insertions(+) create mode 100644 internal/server/docker_client.go create mode 100644 internal/server/docker_client_test.go diff --git a/internal/server/docker_client.go b/internal/server/docker_client.go new file mode 100644 index 0000000..365a29d --- /dev/null +++ b/internal/server/docker_client.go @@ -0,0 +1,245 @@ +package server + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strconv" + "strings" + "sync" + "time" +) + +const ( + // fallbackDockerAPIVersion is used when /version cannot be read. Hardened + // socket proxies routinely allow the container endpoints while denying + // /version, so failing to negotiate must not stop the client working. 1.41 + // ships with Docker 20.10 and is old enough that any daemon we could + // plausibly be talking to accepts it. + fallbackDockerAPIVersion = "1.41" + + // maxDockerErrorBodyBytes caps how much of a failure response reaches the + // error string, which travels to the operator's terminal over net/rpc. + maxDockerErrorBodyBytes = 4 << 10 + + // maxDockerVersionBodyBytes is deliberately far larger: /version on a + // plugin-heavy host carries a Components array listing every plugin, and + // truncating that JSON turns a healthy daemon into a parse failure that + // silently drops the client to the fallback version. + maxDockerVersionBodyBytes = 1 << 20 + + dockerRequestTimeout = 30 * time.Second + + // defaultDockerStopTimeout is what a stop with no caller deadline asks the + // daemon for. It matches Docker's own default so behaviour is unchanged for + // containers that never configured one. + defaultDockerStopTimeout = 10 * time.Second + + // dockerStopMargin keeps the daemon's SIGKILL strictly inside our budget, so + // the response arrives before our context expires rather than racing it. + dockerStopMargin = 5 * time.Second +) + +// DockerClient talks to the Docker daemon over its unix socket. It implements +// ContainerLifecycle with no external dependency: the Engine API calls it needs +// are three endpoints, and pulling in the official SDK for them would add a +// large dependency tree to a proxy that otherwise has none. +// +// Reaching this socket is root-equivalent on the host. It is only constructed +// when the operator passes --docker-socket. +type DockerClient struct { + socketPath string + http *http.Client + + lock sync.Mutex + apiVersion string +} + +func NewDockerClient(socketPath string) *DockerClient { + return &DockerClient{ + socketPath: socketPath, + http: &http.Client{ + Timeout: dockerRequestTimeout, + + // Docker's router cleans the decoded path and answers 301 to the + // canonical form, and Go rewrites a redirected POST as a GET. Following + // that would turn a start into a read: ContainerExists would pass on a + // reference StartContainer can never start, so the deploy preflight + // would accept exactly what it exists to reject. Against a redirecting + // socket proxy a stop could report success having stopped nothing. + // Nothing in the Engine API legitimately redirects. + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + + Transport: &http.Transport{ + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + var dialer net.Dialer + return dialer.DialContext(ctx, "unix", socketPath) + }, + }, + }, + } +} + +func (c *DockerClient) StartContainer(ctx context.Context, ref string) error { + return c.send(ctx, http.MethodPost, c.containerPath(ctx, ref, "start"), ref) +} + +// StopContainer stops the container, telling the daemon how long to wait between +// SIGTERM and SIGKILL rather than letting the container's own StopTimeout decide. +// +// Without that, the two deadlines are independent and the daemon's is the one we +// cannot see: kamal passes `options:` straight through to `docker run`, so a +// container created with --stop-timeout 60 keeps the daemon working long after +// our context expires. We would report a failure for a stop the daemon then +// completes, and the controller would roll the service back to active and put +// its targets back for a container Docker is about to kill. +func (c *DockerClient) StopContainer(ctx context.Context, ref string) error { + path := c.containerPath(ctx, ref, "stop") + "?t=" + strconv.Itoa(stopTimeoutSeconds(ctx)) + return c.send(ctx, http.MethodPost, path, ref) +} + +// stopTimeoutSeconds leaves the daemon a kill deadline that lands inside our own +// budget, with a margin for the round trip. A caller with no deadline still gets +// a bound: an unbounded stop is what this exists to prevent. +func stopTimeoutSeconds(ctx context.Context) int { + deadline, ok := ctx.Deadline() + if !ok { + return int(defaultDockerStopTimeout.Seconds()) + } + + remaining := time.Until(deadline) - dockerStopMargin + if remaining < time.Second { + // Already out of budget. One second still beats zero, which Docker reads + // as "SIGKILL immediately" and denies the app any chance to shut down. + return 1 + } + + return int(remaining.Seconds()) +} + +// ContainerExists inspects the container so a deploy can reject a reference that +// would otherwise only fail hours later at the first idle timeout. +func (c *DockerClient) ContainerExists(ctx context.Context, ref string) error { + return c.send(ctx, http.MethodGet, c.containerPath(ctx, ref, "json"), ref) +} + +func (c *DockerClient) send(ctx context.Context, method, path, ref string) error { + resp, err := c.do(ctx, method, path) + if err != nil { + return err + } + defer resp.Body.Close() + + return c.classify(resp, ref) +} + +// containerPath builds /v/containers//. The reference is +// path-escaped because a container name is operator-supplied and a bare name +// with a slash would otherwise change which endpoint is addressed. +func (c *DockerClient) containerPath(ctx context.Context, ref, action string) string { + return fmt.Sprintf("/v%s/containers/%s/%s", c.negotiateAPIVersion(ctx), url.PathEscape(ref), action) +} + +func (c *DockerClient) do(ctx context.Context, method, path string) (*http.Response, error) { + // The host is ignored -- the transport always dials the socket -- but the URL + // still needs one to be well formed. + req, err := http.NewRequestWithContext(ctx, method, "http://docker"+path, http.NoBody) + if err != nil { + return nil, err + } + + resp, err := c.http.Do(req) + if err != nil { + return nil, fmt.Errorf("docker socket %s: %w", c.socketPath, err) + } + + return resp, nil +} + +// classify turns a response into the sentinel errors the preflight and the wake +// path branch on. +func (c *DockerClient) classify(resp *http.Response, ref string) error { + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + return nil + + // Already in the requested state. Starting a running container or stopping a + // stopped one is exactly what a coalesced wake or a redundant sleep does, and + // neither is a failure. + case resp.StatusCode == http.StatusNotModified: + return nil + + case resp.StatusCode == http.StatusNotFound: + return fmt.Errorf("%w: %s", ErrContainerNotFound, ref) + + case resp.StatusCode == http.StatusForbidden: + return fmt.Errorf("%w: %s", ErrContainerInspectForbidden, ref) + + default: + return fmt.Errorf("docker returned %s for container %s: %s", + resp.Status, ref, readLimited(resp.Body, maxDockerErrorBodyBytes)) + } +} + +// negotiateAPIVersion asks the daemon which API version it speaks, caching only +// a successful answer. Caching a failure would pin the client to the fallback +// for the life of the process even after the daemon came back. +func (c *DockerClient) negotiateAPIVersion(ctx context.Context) string { + c.lock.Lock() + cached := c.apiVersion + c.lock.Unlock() + + if cached != "" { + return cached + } + + version := c.readAPIVersion(ctx) + if version == "" { + return fallbackDockerAPIVersion + } + + c.lock.Lock() + c.apiVersion = version + c.lock.Unlock() + + return version +} + +// readAPIVersion returns "" for every failure mode -- unreachable socket, denied +// endpoint, non-JSON body, JSON without the field -- so the caller has a single +// fallback path rather than four. +func (c *DockerClient) readAPIVersion(ctx context.Context) string { + resp, err := c.do(ctx, http.MethodGet, "/version") + if err != nil { + return "" + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return "" + } + + var payload struct { + APIVersion string `json:"ApiVersion"` + } + if err := json.NewDecoder(io.LimitReader(resp.Body, maxDockerVersionBodyBytes)).Decode(&payload); err != nil { + return "" + } + + return payload.APIVersion +} + +func readLimited(r io.Reader, limit int64) string { + body, err := io.ReadAll(io.LimitReader(r, limit)) + if err != nil { + return "" + } + + return strings.TrimSpace(string(body)) +} diff --git a/internal/server/docker_client_test.go b/internal/server/docker_client_test.go new file mode 100644 index 0000000..de3ea42 --- /dev/null +++ b/internal/server/docker_client_test.go @@ -0,0 +1,390 @@ +package server + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// testDockerDaemon serves a handler on a real unix socket, so the client is +// exercised over the same transport it uses in production rather than a stub. +func testDockerDaemon(t *testing.T, handler http.HandlerFunc) (*DockerClient, *recordedRequests) { + t.Helper() + + // macOS caps sun_path at 104 bytes and t.TempDir() paths are long, so the + // socket goes somewhere short rather than in the test's own temp dir. + dir, err := os.MkdirTemp("", "kp") + require.NoError(t, err) + t.Cleanup(func() { os.RemoveAll(dir) }) + + socketPath := filepath.Join(dir, "d.sock") + + listener, err := net.Listen("unix", socketPath) + require.NoError(t, err) + + recorded := &recordedRequests{} + + server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // RequestURI, not URL.Path: Go hands the handler a decoded path, so + // asserting on that cannot tell an escaped reference from an unescaped one. + recorded.add(r.Method + " " + r.RequestURI) + handler(w, r) + })} + + go server.Serve(listener) + t.Cleanup(func() { _ = server.Close() }) + + return NewDockerClient(socketPath), recorded +} + +type recordedRequests struct { + mu sync.Mutex + paths []string +} + +func (r *recordedRequests) add(path string) { + r.mu.Lock() + defer r.mu.Unlock() + r.paths = append(r.paths, path) +} + +func (r *recordedRequests) all() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string{}, r.paths...) +} + +func TestDockerClient_NegotiatesAndUsesVersionedPaths(t *testing.T) { + client, recorded := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + w.WriteHeader(http.StatusNoContent) + }) + + require.NoError(t, client.StartContainer(context.Background(), "web-1")) + + assert.Equal(t, []string{"GET /version", "POST /v1.44/containers/web-1/start"}, recorded.all()) + + // The version is negotiated once and reused. + require.NoError(t, client.StopContainer(context.Background(), "web-1")) + assert.Equal(t, []string{ + "GET /version", + "POST /v1.44/containers/web-1/start", + "POST /v1.44/containers/web-1/stop?t=10", + }, recorded.all()) +} + +func TestDockerClient_PathEscapesTheContainerReference(t *testing.T) { + client, recorded := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + w.WriteHeader(http.StatusNoContent) + }) + + require.NoError(t, client.StartContainer(context.Background(), "we/ird name")) + + assert.Contains(t, recorded.all(), "POST /v1.44/containers/we%2Fird%20name/start", + "an unescaped slash would change which endpoint is addressed") +} + +// Starting an already-running container answers 304. That is what lets a wake +// coalesced behind another wake succeed, and what lets a proxy whose state file +// said "sleeping" for a container that never stopped heal itself. +func TestDockerClient_TreatsNotModifiedAsSuccess(t *testing.T) { + client, _ := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + w.WriteHeader(http.StatusNotModified) + }) + + assert.NoError(t, client.StartContainer(context.Background(), "web-1")) + assert.NoError(t, client.StopContainer(context.Background(), "web-1")) +} + +func TestDockerClient_FallsBackWhenVersionIsUnavailable(t *testing.T) { + tests := []struct { + name string + handler http.HandlerFunc + }{ + { + // A hardened socket proxy that allows start/stop but not /version. + name: "version endpoint forbidden", + handler: func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + w.WriteHeader(http.StatusForbidden) + return + } + w.WriteHeader(http.StatusNoContent) + }, + }, + { + name: "version payload is not JSON", + handler: func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, "nope") + return + } + w.WriteHeader(http.StatusNoContent) + }, + }, + { + name: "version payload omits ApiVersion", + handler: func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"Os":"linux"}`) + return + } + w.WriteHeader(http.StatusNoContent) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, recorded := testDockerDaemon(t, tt.handler) + + require.NoError(t, client.StartContainer(context.Background(), "web-1")) + + assert.Contains(t, recorded.all(), "POST /v"+fallbackDockerAPIVersion+"/containers/web-1/start", + "an unusable /version must not stop the client working") + }) + } +} + +// Caching a negotiation failure would pin the client to the fallback version for +// the life of the process, even after the daemon came back. +func TestDockerClient_DoesNotCacheANegotiationFailure(t *testing.T) { + var versionCalls atomic.Int64 + + client, recorded := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + if versionCalls.Add(1) == 1 { + w.WriteHeader(http.StatusInternalServerError) + return + } + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + w.WriteHeader(http.StatusNoContent) + }) + + require.NoError(t, client.StartContainer(context.Background(), "web-1")) + require.NoError(t, client.StartContainer(context.Background(), "web-1")) + + assert.Equal(t, int64(2), versionCalls.Load(), "a failed negotiation must be retried") + assert.Contains(t, recorded.all(), "POST /v"+fallbackDockerAPIVersion+"/containers/web-1/start") + assert.Contains(t, recorded.all(), "POST /v1.44/containers/web-1/start") +} + +func TestDockerClient_ClassifiesMissingAndForbidden(t *testing.T) { + tests := []struct { + status int + expected error + }{ + {status: http.StatusNotFound, expected: ErrContainerNotFound}, + {status: http.StatusForbidden, expected: ErrContainerInspectForbidden}, + } + + for _, tt := range tests { + t.Run(fmt.Sprint(tt.status), func(t *testing.T) { + client, _ := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + w.WriteHeader(tt.status) + }) + + require.ErrorIs(t, client.ContainerExists(context.Background(), "web-1"), tt.expected) + require.ErrorIs(t, client.StartContainer(context.Background(), "web-1"), tt.expected) + }) + } +} + +func TestDockerClient_ContainerExistsAcceptsAKnownContainer(t *testing.T) { + client, recorded := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + fmt.Fprint(w, `{"Id":"deadbeef"}`) + }) + + require.NoError(t, client.ContainerExists(context.Background(), "web-1")) + assert.Contains(t, recorded.all(), "GET /v1.44/containers/web-1/json") +} + +func TestDockerClient_TruncatesLongErrorBodies(t *testing.T) { + client, _ := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + w.WriteHeader(http.StatusInternalServerError) + fmt.Fprint(w, strings.Repeat("x", 64*1024)) + }) + + err := client.StartContainer(context.Background(), "web-1") + require.Error(t, err) + assert.Less(t, len(err.Error()), maxDockerErrorBodyBytes+512, + "a daemon returning a huge body must not put all of it in the error") +} + +// A plugin-heavy host's /version payload is far larger than an error body, and +// capping both at the same size truncates the JSON into a parse failure -- which +// would silently drop the client to the fallback version on a healthy daemon. +func TestDockerClient_ReadsALargeVersionPayload(t *testing.T) { + client, recorded := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"Padding":%q,"ApiVersion":"1.44"}`, strings.Repeat("p", 32*1024)) + return + } + w.WriteHeader(http.StatusNoContent) + }) + + require.NoError(t, client.StartContainer(context.Background(), "web-1")) + assert.Contains(t, recorded.all(), "POST /v1.44/containers/web-1/start", + "a large but valid /version payload must still negotiate") +} + +func TestDockerClient_SurfacesAnUnreachableSocket(t *testing.T) { + client := NewDockerClient(filepath.Join(t.TempDir(), "absent.sock")) + + err := client.StartContainer(context.Background(), "web-1") + require.Error(t, err) + assert.NotErrorIs(t, err, ErrContainerNotFound, + "a dead socket is not the same as a missing container") +} + +func TestDockerClient_RespectsContextCancellation(t *testing.T) { + client, _ := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + <-r.Context().Done() + }) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := client.StartContainer(ctx, "web-1") + require.Error(t, err) + assert.True(t, errors.Is(err, context.Canceled), "got %v", err) +} + +// Docker's own SIGTERM-to-SIGKILL wait comes from the container's StopTimeout, +// which the proxy cannot see -- kamal passes `options:` straight through to +// docker run, and compose has stop_grace_period. Left unbounded, a container +// configured to wait longer than our budget makes the client give up on a stop +// the daemon then completes: the controller rolls back to active and puts the +// targets back for a container Docker is about to kill. +func TestDockerClient_BoundsTheDaemonStopInsideTheCallersDeadline(t *testing.T) { + client, recorded := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + w.WriteHeader(http.StatusNoContent) + }) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + require.NoError(t, client.StopContainer(ctx, "web-1")) + + var stopPath string + for _, p := range recorded.all() { + if strings.Contains(p, "/stop") { + stopPath = p + } + } + require.NotEmpty(t, stopPath) + + seconds := stopTimeoutFromPath(t, stopPath) + assert.Positive(t, seconds, "the daemon needs a kill deadline") + assert.Less(t, seconds, 30, "it has to land inside our own budget, not on it") +} + +func TestDockerClient_StopWithoutADeadlineStillBoundsTheDaemon(t *testing.T) { + client, recorded := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + w.WriteHeader(http.StatusNoContent) + }) + + require.NoError(t, client.StopContainer(context.Background(), "web-1")) + + var stopPath string + for _, p := range recorded.all() { + if strings.Contains(p, "/stop") { + stopPath = p + } + } + assert.Positive(t, stopTimeoutFromPath(t, stopPath)) +} + +func stopTimeoutFromPath(t *testing.T, path string) int { + t.Helper() + + _, query, found := strings.Cut(path, "?") + require.True(t, found, "stop must carry a timeout: %s", path) + + values, err := url.ParseQuery(query) + require.NoError(t, err) + + seconds, err := strconv.Atoi(values.Get("t")) + require.NoError(t, err, "t must be an integer: %s", path) + return seconds +} + +// Docker's router cleans the decoded path and answers 301 to the canonical form, +// and Go rewrites a redirected POST as a GET. Following that turns a start into +// a read: ContainerExists passes on a reference StartContainer can never start, +// so the deploy preflight accepts exactly what it exists to reject. Against a +// redirecting socket proxy a stop could even report success having stopped +// nothing. Nothing in the Engine API legitimately redirects. +func TestDockerClient_DoesNotFollowRedirects(t *testing.T) { + client, recorded := testDockerDaemon(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.44"}`) + return + } + if strings.HasSuffix(r.URL.Path, "/start") || strings.HasSuffix(r.URL.Path, "/stop") { + http.Redirect(w, r, "/ok", http.StatusMovedPermanently) + return + } + w.WriteHeader(http.StatusNoContent) + }) + + err := client.StartContainer(context.Background(), "web-1") + require.Error(t, err, "a redirect must not be silently re-issued as a GET") + + for _, p := range recorded.all() { + assert.NotContains(t, p, "/ok", "the client must not have followed the redirect") + } +}