Add opt-in scale-to-zero for idle services

README.md

@@ -50,6 +50,38 @@ Each deployment takes over all the traffic from the previously deployed
 instance. As soon as Kamal Proxy determines that the new instance is healthy,
 it will route all new traffic to that instance.
 
+### Opt-in scale to zero
+
+Services can stop their write containers after an idle period and wake them on
+the next application request:
+
+    kamal-proxy run --docker-socket /var/run/docker.sock
+    kamal-proxy deploy service1 --target web-1:3000 --idle-timeout 15m --idle-wake-timeout 30s
+
+`--idle-timeout` defaults to `0` (disabled). `--idle-wake-timeout` defaults to
+`30s` and bounds how long each request waits for Docker start and a successful
+configured health check. The target hostname (`web-1` above) must be the Docker
+container name. `DOCKER_SOCKET` and `KAMAL_PROXY_DOCKER_SOCKET` are equivalents
+of the run flag.
+
+Requests are held before their bodies are read, so POST bodies are forwarded
+unchanged after a successful wake. Concurrent wake requests are coalesced.
+Open streaming responses and WebSockets count as activity/in-flight work and
+prevent sleeping until they close; a new stream or WebSocket is held during
+wake like any other request. Health-check requests do not wake or reset an idle
+service and receive success while it is stopping, sleeping, or waking.
+
+Mounting the Docker socket gives the proxy host-level container control. Only
+enable this feature where that trust is acceptable; the lifecycle calls are
+isolated behind the `ContainerLifecycle` interface so they can be moved to an
+external service later.
+
+The Docker client negotiates the API version once from the daemon's unversioned
+`/version` endpoint and caches it for start/stop calls. If that endpoint is
+unavailable or returns a non-success status, it falls back to the legacy
+`v1.41` paths for compatibility with restricted socket proxies; a successful
+but malformed version response is rejected instead of guessing.
+
 The `deploy` command also waits for traffic to drain from the old instance before
 returning. This means it's safe to remove the old instance as soon as `deploy`
 returns successfully, without interrupting any in-flight requests.

internal/cmd/deploy.go

@@ -52,6 +52,9 @@ func newDeployCommand() *deployCommand {
 	deployCommand.cmd.Flags().DurationVar(&deployCommand.args.ServiceOptions.WriterAffinityTimeout, "writer-affinity-timeout", server.DefaultWriterAffinityTimeout, "Time after a write before read requests will be routed to readers")
 	deployCommand.cmd.Flags().BoolVar(&deployCommand.args.ServiceOptions.ReadTargetsAcceptWebsockets, "read-target-websockets", false, "Route WebSocket traffic to read targets, when available")
 
+	deployCommand.cmd.Flags().DurationVar(&deployCommand.args.ServiceOptions.IdleTimeout, "idle-timeout", 0, "Stop container after this duration of inactivity (0 to disable)")
+	deployCommand.cmd.Flags().DurationVar(&deployCommand.args.ServiceOptions.IdleWakeTimeout, "idle-wake-timeout", server.DefaultIdleWakeTimeout, "Max time to hold request while waking container")
+
 	deployCommand.cmd.Flags().DurationVar(&deployCommand.args.TargetOptions.ResponseTimeout, "target-timeout", server.DefaultTargetTimeout, "Maximum time to wait for the target server to respond when serving requests")
 
 	deployCommand.cmd.Flags().BoolVar(&deployCommand.args.TargetOptions.BufferRequests, "buffer-requests", false, "Buffer requests before forwarding to target")

internal/cmd/run.go

@@ -29,15 +29,18 @@ func newRunCommand() *runCommand {
 	runCommand.cmd.Flags().IntVar(&globalConfig.HttpsPort, "https-port", getEnvInt("HTTPS_PORT", server.DefaultHttpsPort), "Port to serve HTTPS traffic on")
 	runCommand.cmd.Flags().IntVar(&globalConfig.MetricsPort, "metrics-port", getEnvInt("METRICS_PORT", 0), "Publish metrics on the specified port (default zero to disable)")
 	runCommand.cmd.Flags().BoolVar(&globalConfig.HTTP3Enabled, "http3", false, "Enable HTTP/3")
+	runCommand.cmd.Flags().StringVar(&globalConfig.DockerSocketPath, "docker-socket", getEnvString("DOCKER_SOCKET", server.DefaultDockerSocketPath), "Path to Docker socket")
 
 	return runCommand
 }
 
 func (c *runCommand) run(cmd *cobra.Command, args []string) error {
 	c.setLogger()
 
-	router := server.NewRouter(globalConfig.StatePath())
-	router.RestoreLastSavedState()
+	router := server.NewRouter(globalConfig.StatePath(), globalConfig.DockerSocketPath)
+	if err := router.RestoreLastSavedState(); err != nil {
+		return err
+	}
 
 	s := server.NewServer(&globalConfig, router)
 	err := s.Start()

internal/cmd/run_test.go

@@ -0,0 +1,20 @@
+package cmd
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRunCommandReturnsStateRestoreError(t *testing.T) {
+	previous := globalConfig
+	t.Cleanup(func() { globalConfig = previous })
+	globalConfig.AlternateConfigDir = t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(globalConfig.AlternateConfigDir, "kamal-proxy.state"), []byte("invalid"), 0o600))
+
+	err := newRunCommand().run(nil, nil)
+
+	require.ErrorContains(t, err, "invalid character 'i'")
+}

internal/cmd/util.go

@@ -47,6 +47,14 @@ func getEnvInt(key string, defaultValue int) int {
 	return intValue
 }
 
+func getEnvString(key, defaultValue string) string {
+	value, ok := findEnv(key)
+	if !ok {
+		return defaultValue
+	}
+	return value
+}
+
 func getEnvBool(key string, defaultValue bool) bool {
 	value, ok := findEnv(key)
 	if !ok {

internal/server/config.go

@@ -8,8 +8,9 @@ import (
 )
 
 const (
-	DefaultHttpPort  = 80
-	DefaultHttpsPort = 443
+	DefaultHttpPort         = 80
+	DefaultHttpsPort        = 443
+	DefaultDockerSocketPath = "/var/run/docker.sock"
 )
 
 type Config struct {
@@ -20,6 +21,7 @@ type Config struct {
 	HTTP3Enabled bool
 
 	AlternateConfigDir string
+	DockerSocketPath   string
 }
 
 func (c Config) SocketPath() string {

internal/server/docker_client.go

@@ -0,0 +1,141 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+	"sync"
+)
+
+const (
+	legacyDockerAPIVersion = "1.41"
+	maxDockerErrorBody     = 4096
+)
+
+type DockerClient struct {
+	httpClient *http.Client
+
+	versionMu  sync.Mutex
+	versionSet bool
+	apiVersion string
+	versionErr error
+}
+
+func NewDockerClient(socketPath string) *DockerClient {
+	return &DockerClient{
+		httpClient: &http.Client{
+			Transport: &http.Transport{
+				DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+					return (&net.Dialer{}).DialContext(ctx, "unix", socketPath)
+				},
+			},
+		},
+	}
+}
+
+func (c *DockerClient) StopContainer(ctx context.Context, name string) error {
+	return c.containerAction(ctx, name, "stop")
+}
+
+func (c *DockerClient) StartContainer(ctx context.Context, name string) error {
+	return c.containerAction(ctx, name, "start")
+}
+
+func (c *DockerClient) containerAction(ctx context.Context, name, action string) error {
+	version, err := c.negotiatedVersion(ctx)
+	if err != nil {
+		return err
+	}
+	endpoint := fmt.Sprintf("http://localhost/v%s/containers/%s/%s", version, url.PathEscape(name), action)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, nil)
+	if err != nil {
+		return err
+	}
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotModified {
+		return dockerResponseError(action, resp)
+	}
+	return nil
+}
+
+func (c *DockerClient) negotiatedVersion(ctx context.Context) (string, error) {
+	c.versionMu.Lock()
+	defer c.versionMu.Unlock()
+	if c.versionSet {
+		return c.apiVersion, c.versionErr
+	}
+
+	negotiationCtx, cancel := dockerNegotiationContext(ctx)
+	defer cancel()
+	version, err := c.queryVersion(negotiationCtx)
+	if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+		c.apiVersion, c.versionErr, c.versionSet = version, err, true
+	}
+	return version, err
+}
+
+func dockerNegotiationContext(ctx context.Context) (context.Context, context.CancelFunc) {
+	if deadline, ok := ctx.Deadline(); ok {
+		return context.WithDeadline(context.Background(), deadline)
+	}
+	return context.WithCancel(context.Background())
+}
+
+func (c *DockerClient) queryVersion(ctx context.Context) (string, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost/version", nil)
+	if err != nil {
+		return "", err
+	}
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		if ctx.Err() != nil {
+			return "", ctx.Err()
+		}
+		// Some compatible Docker proxies do not expose /version. Preserve the
+		// legacy behavior and let the versioned operation return the useful error.
+		return legacyDockerAPIVersion, nil
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return legacyDockerAPIVersion, nil
+	}
+	var version struct {
+		APIVersion string `json:"ApiVersion"`
+	}
+	if err := json.NewDecoder(io.LimitReader(resp.Body, maxDockerErrorBody+1)).Decode(&version); err != nil {
+		return "", fmt.Errorf("invalid docker /version response: %w", err)
+	}
+	if version.APIVersion == "" {
+		return "", errors.New("docker /version response has no ApiVersion")
+	}
+	return version.APIVersion, nil
+}
+
+func dockerResponseError(action string, resp *http.Response) error {
+	body, err := io.ReadAll(io.LimitReader(resp.Body, maxDockerErrorBody+1))
+	if err != nil {
+		return fmt.Errorf("docker %s returned status %d (reading error body: %w)", action, resp.StatusCode, err)
+	}
+	truncated := len(body) > maxDockerErrorBody
+	if truncated {
+		body = body[:maxDockerErrorBody]
+	}
+	message := strings.TrimSpace(string(body))
+	if message == "" {
+		return fmt.Errorf("docker %s returned status %d", action, resp.StatusCode)
+	}
+	if truncated {
+		message += "…"
+	}
+	return fmt.Errorf("docker %s returned status %d: %s", action, resp.StatusCode, message)
+}