diff --git a/docker-bake.hcl b/docker-bake.hcl index a6da4bb9aa..975b35fb98 100644 --- a/docker-bake.hcl +++ b/docker-bake.hcl @@ -166,3 +166,12 @@ target "image-module-cross" { "windows/arm64", ] } + +target "relay-image" { + context = "./relay" + tags = ["docker/compose-relay:v1"] + platforms = [ + "linux/amd64", + "linux/arm64", + ] +} diff --git a/docs/examples/provider.go b/docs/examples/provider.go index 83ab901993..bf28bd5e0a 100644 --- a/docs/examples/provider.go +++ b/docs/examples/provider.go @@ -17,9 +17,13 @@ package main import ( + "bufio" "encoding/json" "fmt" + "net" "os" + "os/exec" + "strings" "time" "github.com/spf13/cobra" @@ -32,6 +36,7 @@ func main() { Use: "demo", } cmd.AddCommand(composeCommand()) + cmd.AddCommand(serveDemoCommand()) err := cmd.Execute() if err != nil { fmt.Fprintln(os.Stderr, err) @@ -85,6 +90,44 @@ func composeCommand() *cobra.Command { return c } +// serveDemoCommand is the detached helper process behind the +// publish-endpoint demonstration: a TCP server on the given address +// answering every connection with a fixed HTTP response, exiting on its own +// after three minutes. It owns the port from bind to exit: the bound address +// is reported on stdout once listening, so the parent never has to probe or +// pre-reserve the port (no TOCTOU window, and works on Windows where handing +// a socket over ExtraFiles is not supported). +func serveDemoCommand() *cobra.Command { + return &cobra.Command{ + Use: "serve-demo ADDR", + Hidden: true, + Args: cobra.ExactArgs(1), + RunE: func(_ *cobra.Command, args []string) error { + listener, err := net.Listen("tcp", args[0]) + if err != nil { + return err + } + fmt.Println(listener.Addr().String()) + go func() { + time.Sleep(3 * time.Minute) + os.Exit(0) + }() + for { + conn, err := listener.Accept() + if err != nil { + return err + } + go func() { + defer func() { _ = conn.Close() }() + buf := make([]byte, 1024) + _, _ = conn.Read(buf) + _, _ = conn.Write([]byte("HTTP/1.1 200 OK\r\nContent-Length: 19\r\nConnection: close\r\n\r\nhello from provider")) + }() + } + }, + } +} + const lineSeparator = "\n" func up(options options, args []string) { @@ -115,6 +158,42 @@ func up(options options, args []string) { setenv, _ := json.Marshal(map[string]string{"type": "setenv", "message": "CONFIG_TYPE=" + config.Provider.Type}) fmt.Println(string(setenv)) + // When asked to, stand up a real endpoint on the host and publish it, so + // compose deploys a relay and consumers reach it as http://:80. + if os.Getenv("PROVIDER_DEMO_ENDPOINT") != "" { + // The subprocess binds the port itself and reports the resulting + // address on its stdout; only then is the endpoint published. This + // avoids the two races of a pre-reserved port: another process + // grabbing it between release and re-bind, and publish-endpoint + // pointing at a server that is not listening yet. + // All interfaces, not loopback: on a plain Linux engine host-gateway + // is the bridge IP, which cannot reach a host loopback bind. + server := exec.Command(os.Args[0], "serve-demo", "0.0.0.0:0") + stdout, err := server.StdoutPipe() + if err != nil { + fmt.Printf(`{ "type": "error", "message": "demo endpoint: %v" }%s`, err, lineSeparator) + return + } + if err := server.Start(); err != nil { + fmt.Printf(`{ "type": "error", "message": "demo endpoint: %v" }%s`, err, lineSeparator) + return + } + // A crashed subprocess closes the pipe (EOF below); a hung one would + // block the read forever, so kill it after a deadline — the read then + // fails with EOF and lands on the same error path. + watchdog := time.AfterFunc(30*time.Second, func() { _ = server.Process.Kill() }) + addr, err := bufio.NewReader(stdout).ReadString('\n') + watchdog.Stop() + if err != nil { + fmt.Printf(`{ "type": "error", "message": "demo endpoint did not come up: %v" }%s`, err, lineSeparator) + return + } + // the endpoint is announced as seen from THIS process's host — + // the relay translates loopback into the container-visible name + _, port, _ := net.SplitHostPort(strings.TrimSpace(addr)) + fmt.Printf(`{ "type": "publish-endpoint", "message": "80=localhost:%s" }%s`, port, lineSeparator) + } + for i := 0; i < options.size; i += 10 { time.Sleep(1 * time.Second) fmt.Printf(`{ "type": "info", "message": "Processing ... %d%%" }%s`, i*100/options.size, lineSeparator) diff --git a/docs/extension.md b/docs/extension.md index 4a06f8cb01..e3b0abd291 100644 --- a/docs/extension.md +++ b/docs/extension.md @@ -60,6 +60,25 @@ JSON messages MUST include a `type` and a `message` attribute. - `rawsetenv`: Same as `setenv`, but the variable is injected as-is without the service name prefix. Useful when applications require exact variable names that cannot be altered. - `debug`: Those messages could help debugging the provider, but are not rendered to the user by default. They are rendered when Compose is started with `--verbose` flag. - `get-service-config`: Asks Compose for the resolved configuration of the service the provider manages. See next section. +- `publish-endpoint`: Declares where a network endpoint of the provider's resource is actually reachable. The + message is `"=:"` — the port consumers know on the left, the real location on the + right, as seen FROM THE PROVIDER'S HOST (typically a port published on the host): + ```json + { "type": "publish-endpoint", "message": "80=localhost:49152" } + ``` + The provider does not need to know how containers reach its host: the relay translates a loopback (or + unspecified) upstream host into `host.docker.internal` — resolved through the `host-gateway` extra_host + Compose injects — while routable addresses pass through untouched. + When a provider publishes at least one endpoint, Compose deploys a **relay container** in place of the service: + a minimal TCP forwarder (`docker/compose-relay` — set `COMPOSE_RELAY_IMAGE` to pull the image from an internal + registry instead of Docker Hub) joining the networks of the services that depend on the + provider service, aliased with the service name. Consumers then reach the resource at the compose-native + address — `http://:` — with no injected variables involved. The relay is a regular + project container (standard compose labels, canonical `--1` name), so `ps`, `logs`, `stop` + and `down` treat it as the service; it additionally carries the `com.docker.compose.relay` label identifying + its role, and process-level commands (`exec`, `cp`) refuse it. The relay is recreated when the published + endpoints change, and removed by `down` like any project container. TCP only; the message may be repeated, + one per port. ## Requesting the service configuration diff --git a/pkg/api/labels.go b/pkg/api/labels.go index 9122fb5a83..19b4e1bd3c 100644 --- a/pkg/api/labels.go +++ b/pkg/api/labels.go @@ -43,6 +43,12 @@ const ( EnvironmentFileLabel = "com.docker.compose.project.environment_file" // OneoffLabel stores value 'True' for one-off containers created by `compose run` OneoffLabel = "com.docker.compose.oneoff" + // RelayLabel marks the network relay container compose deploys in place + // of a provider-managed service (see the publish-endpoint provider + // message). Its value is a hash of the relay's routes, used to decide + // whether an existing relay can be kept on the next up. Commands that + // act on a service's process (exec, ...) refuse relay containers. + RelayLabel = "com.docker.compose.relay" // SlugLabel stores unique slug used for one-off container identity SlugLabel = "com.docker.compose.slug" // ImageDigestLabel stores digest of the container image used to run service diff --git a/pkg/compose/cp.go b/pkg/compose/cp.go index c76e31e49e..523b6a0222 100644 --- a/pkg/compose/cp.go +++ b/pkg/compose/cp.go @@ -121,6 +121,9 @@ func (s *composeService) listContainersTargetedForCopy(ctx context.Context, proj if err != nil { return nil, err } + if err := checkRelayTarget(ctr, serviceName, "cp"); err != nil { + return nil, err + } return append(containers, ctr), nil default: withOneOff := oneOffExclude @@ -131,6 +134,11 @@ func (s *composeService) listContainersTargetedForCopy(ctx context.Context, proj if err != nil { return nil, err } + for _, ctr := range containers { + if err := checkRelayTarget(ctr, serviceName, "cp"); err != nil { + return nil, err + } + } if len(containers) < 1 { return nil, fmt.Errorf("no container found for service %q", serviceName) diff --git a/pkg/compose/exec.go b/pkg/compose/exec.go index 30a5d08002..3d7536bf72 100644 --- a/pkg/compose/exec.go +++ b/pkg/compose/exec.go @@ -33,6 +33,9 @@ func (s *composeService) Exec(ctx context.Context, projectName string, options a if err != nil { return 0, err } + if err := checkRelayTarget(target, options.Service, "exec"); err != nil { + return 0, err + } exec := container.NewExecOptions() exec.Interactive = options.Interactive diff --git a/pkg/compose/monitor.go b/pkg/compose/monitor.go index 58bb9341cf..85d13062d1 100644 --- a/pkg/compose/monitor.go +++ b/pkg/compose/monitor.go @@ -184,13 +184,21 @@ func (c *monitor) initialContainers(ctx context.Context) (utils.Set[string], err } containers := utils.Set[string]{} for _, ctr := range initialState.Items { - if c.watched(ctr.Labels[api.ServiceLabel]) { + if c.watched(ctr.Labels[api.ServiceLabel]) && !isRelay(ctr.Labels) { containers.Add(ctr.ID) } } return containers, nil } +// isRelay reports whether labels identify a provider-relay container. Relays +// are long-lived infrastructure standing in for a provider's resource: they +// never terminate on their own, so counting them among the application's +// containers would keep an attached `up` waiting forever. +func isRelay(labels map[string]string) bool { + return labels[api.RelayLabel] != "" +} + // watched tells whether a service's containers are watched by this monitor. // An empty service set means "the whole application". func (c *monitor) watched(service string) bool { @@ -205,7 +213,7 @@ func (c *monitor) notify(event api.ContainerEvent) { } func (c *monitor) onContainerCreate(event events.Message, ctr *api.ContainerSummary, containers utils.Set[string]) { - if c.watched(ctr.Labels[api.ServiceLabel]) { + if c.watched(ctr.Labels[api.ServiceLabel]) && !isRelay(ctr.Labels) { containers.Add(ctr.ID) } evtType := api.ContainerEventCreated @@ -226,7 +234,7 @@ func (c *monitor) onContainerStart(event events.Message, ctr *api.ContainerSumma logrus.Debugf("container %s started", ctr.Name) c.notify(newContainerEvent(event.TimeNano, ctr, api.ContainerEventStarted)) } - if c.watched(ctr.Labels[api.ServiceLabel]) { + if c.watched(ctr.Labels[api.ServiceLabel]) && !isRelay(ctr.Labels) { containers.Add(ctr.ID) } } diff --git a/pkg/compose/plugins.go b/pkg/compose/plugins.go index 84a40775e0..9af0082d13 100644 --- a/pkg/compose/plugins.go +++ b/pkg/compose/plugins.go @@ -23,9 +23,11 @@ import ( "errors" "fmt" "io" + "net" "os" "os/exec" "path/filepath" + "strconv" "strings" "sync" @@ -50,6 +52,7 @@ const ( SetEnvType = "setenv" RawSetEnvType = "rawsetenv" DebugType = "debug" + PublishEndpointType = "publish-endpoint" providerMetadataDirectory = "compose/providers" // GetServiceConfigType is a message the provider sends to receive, on @@ -61,6 +64,11 @@ const ( type pluginVariables struct { prefixed types.Mapping raw types.Mapping + // endpoints are "port=host:port" publish-endpoint messages: container + // port the consumers know, mapped to where the provider's resource + // actually listens. When present, compose deploys a relay container + // under the service's name on the consumers' networks. + endpoints map[int]string } var mux sync.Mutex @@ -107,9 +115,30 @@ func (s *composeService) runPlugin(ctx context.Context, project *types.Project, project.Services[name] = s } } + if command == "up" && len(variables.endpoints) > 0 { + if err := s.ensureServiceRelay(ctx, project, service, variables.endpoints); err != nil { + return err + } + } return nil } +// parseEndpointMessage decodes a publish-endpoint payload: "80=host:port". +func parseEndpointMessage(message string) (int, string, error) { + portPart, upstream, found := strings.Cut(message, "=") + if !found { + return 0, "", fmt.Errorf("publish-endpoint %q: want port=host:port", message) + } + port, err := strconv.Atoi(portPart) + if err != nil || port < 1 || port > 65535 { + return 0, "", fmt.Errorf("publish-endpoint %q: invalid port", message) + } + if _, _, err := net.SplitHostPort(upstream); err != nil { + return 0, "", fmt.Errorf("publish-endpoint %q: invalid endpoint: %w", message, err) + } + return port, upstream, nil +} + func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service types.ServiceConfig) (pluginVariables, error) { var action string switch command { @@ -180,8 +209,9 @@ func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service ty defer func() { _ = stdout.Close() }() variables := pluginVariables{ - prefixed: types.Mapping{}, - raw: types.Mapping{}, + prefixed: types.Mapping{}, + raw: types.Mapping{}, + endpoints: map[int]string{}, } for { @@ -224,6 +254,12 @@ func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service ty defer stdinMu.Unlock() _, _ = stdin.Write(payload) }() + case PublishEndpointType: + port, upstream, err := parseEndpointMessage(msg.Message) + if err != nil { + return pluginVariables{}, fmt.Errorf("invalid response from plugin: %w", err) + } + variables.endpoints[port] = upstream case DebugType: logrus.Debugf("%s: %s", service.Name, msg.Message) default: diff --git a/pkg/compose/relay.go b/pkg/compose/relay.go new file mode 100644 index 0000000000..6457636333 --- /dev/null +++ b/pkg/compose/relay.go @@ -0,0 +1,315 @@ +/* + Copyright 2020 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "os" + "sort" + "strings" + "time" + + "github.com/compose-spec/compose-go/v2/types" + "github.com/containerd/errdefs" + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/api/types/network" + "github.com/moby/moby/client" + "github.com/sirupsen/logrus" + + "github.com/docker/compose/v5/pkg/api" +) + +// defaultRelayImage is the published network-relay image deployed in place of +// a provider service that published endpoints. Overridable for development +// and air-gapped setups with COMPOSE_RELAY_IMAGE. +const defaultRelayImage = "docker/compose-relay:v1" + +func relayImage() string { + if img := os.Getenv("COMPOSE_RELAY_IMAGE"); img != "" { + return img + } + return defaultRelayImage +} + +// relayRoutesSpec renders endpoints as the relay's RELAY_ROUTES value, +// canonically ordered so it doubles as the identity the relay label hashes. +func relayRoutesSpec(endpoints map[int]string) string { + ports := make([]int, 0, len(endpoints)) + for port := range endpoints { + ports = append(ports, port) + } + sort.Ints(ports) + routes := make([]string, 0, len(ports)) + for _, port := range ports { + routes = append(routes, fmt.Sprintf("%d=%s", port, endpoints[port])) + } + return strings.Join(routes, ",") +} + +func relayIdentity(routes string) string { + digest := sha256.Sum256([]byte(relayImage() + "|" + routes)) + return hex.EncodeToString(digest[:])[:12] +} + +// relayNetworks returns the compose network keys the relay must join: the +// union of the networks of every service depending on the provider service — +// the consumers the relay exists for — falling back to the project default. +func relayNetworks(project *types.Project, service types.ServiceConfig) []string { + set := map[string]bool{} + for _, s := range project.Services { + if _, ok := s.DependsOn[service.Name]; !ok { + continue + } + for key := range s.Networks { + // a resolved project declares every service network, but guard + // anyway: an unknown key would yield an empty network name later + if _, ok := project.Networks[key]; ok { + set[key] = true + } + } + } + if len(set) == 0 { + if _, ok := project.Networks["default"]; ok { + set["default"] = true + } + } + keys := make([]string, 0, len(set)) + for key := range set { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} + +// ensureServiceRelay converges the relay container standing in for a provider +// service that published endpoints: consumers reach the provider's resource +// at the compose-native address (http://:) through it. The +// relay is a regular project container (standard compose labels, canonical +// name, service alias on the consumers' networks) so label-driven commands +// treat it as the service, plus the RelayLabel identifying its role — the +// reconciler leaves provider services' containers alone, and process-level +// commands (exec) refuse it. +func (s *composeService) ensureServiceRelay(ctx context.Context, project *types.Project, service types.ServiceConfig, endpoints map[int]string) error { + routes := relayRoutesSpec(endpoints) + identity := relayIdentity(routes) + name := getContainerName(project.Name, service, 1) + + existing, err := s.findRelayContainer(ctx, project.Name, service.Name) + if err != nil { + return err + } + if existing != nil { + if existing.Labels[api.RelayLabel] == identity { + switch existing.State { + case container.StateRunning, container.StateRestarting: + // Up to date; restarting means Docker is already recovering + // it — recreating would tear down in-flight connections. + return nil + case container.StateCreated, container.StateExited: + // Routes unchanged: start the existing container rather than + // recreating it. + if _, err := s.apiClient().ContainerStart(ctx, existing.ID, client.ContainerStartOptions{}); err != nil { + return fmt.Errorf("start relay for service %s: %w", service.Name, err) + } + return nil + case container.StatePaused: + if _, err := s.apiClient().ContainerUnpause(ctx, existing.ID, client.ContainerUnpauseOptions{}); err != nil { + return fmt.Errorf("unpause relay for service %s: %w", service.Name, err) + } + return nil + } + } + if existing.State == container.StateRemoving { + // the daemon is already removing it: a concurrent ContainerRemove + // fails with "removal already in progress", so wait for the name + // to free up instead + if err := s.waitRelayRemoved(ctx, project.Name, service.Name); err != nil { + return err + } + } else if _, err := s.apiClient().ContainerRemove(ctx, existing.ID, client.ContainerRemoveOptions{Force: true}); err != nil { + return fmt.Errorf("remove stale relay for service %s: %w", service.Name, err) + } + } + + networkKeys := relayNetworks(project, service) + if len(networkKeys) == 0 { + logrus.Warnf("service %q published endpoints but no service depends on it and the project has no default network; skipping relay", service.Name) + return nil + } + + s.events.On(creatingEvent("Relay " + name)) + id, err := s.createRelayContainer(ctx, project, service, name, routes, identity, networkKeys) + if err != nil { + return err + } + if _, err := s.apiClient().ContainerStart(ctx, id, client.ContainerStartOptions{}); err != nil { + return fmt.Errorf("start relay for service %s: %w", service.Name, err) + } + s.events.On(createdEvent("Relay " + name)) + return nil +} + +// waitRelayRemoved polls until the service's relay container is gone, giving +// an in-progress daemon-side removal time to release the container's name. +func (s *composeService) waitRelayRemoved(ctx context.Context, projectName, serviceName string) error { + deadline := time.Now().Add(30 * time.Second) + for { + existing, err := s.findRelayContainer(ctx, projectName, serviceName) + if err != nil { + return err + } + if existing == nil { + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf("relay for service %s is stuck being removed", serviceName) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(250 * time.Millisecond): + } + } +} + +// findRelayContainer returns the service's relay container, if any. +func (s *composeService) findRelayContainer(ctx context.Context, projectName, serviceName string) (*container.Summary, error) { + f := projectFilter(projectName) + f.Add("label", serviceFilter(serviceName)) + f.Add("label", api.RelayLabel) + result, err := s.apiClient().ContainerList(ctx, client.ContainerListOptions{ + All: true, + Filters: f, + }) + if err != nil { + return nil, err + } + if len(result.Items) == 0 { + return nil, nil + } + return &result.Items[0], nil +} + +func (s *composeService) createRelayContainer(ctx context.Context, project *types.Project, service types.ServiceConfig, name, routes, identity string, networkKeys []string) (string, error) { + labels := types.Labels{ + api.ProjectLabel: project.Name, + api.ServiceLabel: service.Name, + api.VersionLabel: api.ComposeVersion, + api.ConfigFilesLabel: strings.Join(project.ComposeFiles, ","), + api.WorkingDirLabel: project.WorkingDir, + api.ContainerNumberLabel: "1", + api.OneoffLabel: "False", + // Label-driven commands filter on ConfigHashLabel presence: without + // it the relay would be invisible to ps/stop/exec run without the + // compose file. The relay identity doubles as its config hash. + api.ConfigHashLabel: identity, + api.RelayLabel: identity, + } + + config := &container.Config{ + Image: relayImage(), + Env: []string{"RELAY_ROUTES=" + routes}, + Labels: labels, + } + hostConfig := &container.HostConfig{ + // host.docker.internal resolves natively on Docker Desktop; the + // host-gateway mapping makes the same upstream host name work on a + // plain Linux engine. + ExtraHosts: []string{"host.docker.internal:host-gateway"}, + RestartPolicy: container.RestartPolicy{ + Name: container.RestartPolicyUnlessStopped, + }, + } + + // First network at creation, remaining ones connected afterwards — the + // engine accepts a single endpoint in the create payload. + endpointSettings := func(_ string) *network.EndpointSettings { + return &network.EndpointSettings{Aliases: []string{service.Name}} + } + first := project.Networks[networkKeys[0]].Name + networking := &network.NetworkingConfig{ + EndpointsConfig: map[string]*network.EndpointSettings{ + first: endpointSettings(first), + }, + } + + created, err := s.apiClient().ContainerCreate(ctx, client.ContainerCreateOptions{ + Name: name, + Config: config, + HostConfig: hostConfig, + NetworkingConfig: networking, + }) + if errdefs.IsNotFound(err) { + if err := s.pullRelayImage(ctx); err != nil { + return "", err + } + created, err = s.apiClient().ContainerCreate(ctx, client.ContainerCreateOptions{ + Name: name, + Config: config, + HostConfig: hostConfig, + NetworkingConfig: networking, + }) + } + if err != nil { + return "", fmt.Errorf("create relay for service %s: %w", service.Name, err) + } + + for _, key := range networkKeys[1:] { + netName := project.Networks[key].Name + if _, err := s.apiClient().NetworkConnect(ctx, netName, client.NetworkConnectOptions{ + Container: created.ID, + EndpointConfig: endpointSettings(netName), + }); err != nil { + // remove the half-connected container: left in place (with its + // restart policy) it would serve only a subset of the consumers' + // networks, and its identity would shield it from recreation + if _, rmErr := s.apiClient().ContainerRemove(ctx, created.ID, client.ContainerRemoveOptions{Force: true}); rmErr != nil { + logrus.Warnf("removing half-connected relay %s: %v", name, rmErr) + } + return "", fmt.Errorf("connect relay for service %s to network %s: %w", service.Name, netName, err) + } + } + return created.ID, nil +} + +func (s *composeService) pullRelayImage(ctx context.Context) error { + image := relayImage() + s.events.On(newEvent(image, api.Working, "Pulling")) + response, err := s.apiClient().ImagePull(ctx, image, client.ImagePullOptions{}) + if err != nil { + return fmt.Errorf("pull %s: %w", image, err) + } + defer func() { _ = response.Close() }() + if err := response.Wait(ctx); err != nil { + return fmt.Errorf("pull %s: %w", image, err) + } + s.events.On(newEvent(image, api.Done, "Pulled")) + return nil +} + +// checkRelayTarget refuses process-level operations on a relay container: it +// stands in for the provider's resource on the network, but there is no +// service process in it to act on. +func checkRelayTarget(target container.Summary, serviceName, operation string) error { + if target.Labels[api.RelayLabel] == "" { + return nil + } + return fmt.Errorf("service %q is managed by a provider: its container is a network relay and does not support %s", serviceName, operation) +} diff --git a/pkg/compose/relay_test.go b/pkg/compose/relay_test.go new file mode 100644 index 0000000000..55767b22de --- /dev/null +++ b/pkg/compose/relay_test.go @@ -0,0 +1,113 @@ +/* + Copyright 2020 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "testing" + + "github.com/compose-spec/compose-go/v2/types" + "github.com/moby/moby/api/types/container" + "gotest.tools/v3/assert" + + "github.com/docker/compose/v5/pkg/api" +) + +// COMPOSE_RELAY_IMAGE lets internal-registry users substitute their own copy +// of the relay image for the Docker Hub default — and the identity hash +// covers the image, so switching it recreates existing relays. +func TestRelayImageOverride(t *testing.T) { + t.Setenv("COMPOSE_RELAY_IMAGE", "") + assert.Equal(t, relayImage(), defaultRelayImage) + defaultIdentity := relayIdentity("80=host.docker.internal:1") + + t.Setenv("COMPOSE_RELAY_IMAGE", "registry.corp.example/infra/compose-relay:v1") + assert.Equal(t, relayImage(), "registry.corp.example/infra/compose-relay:v1") + assert.Assert(t, relayIdentity("80=host.docker.internal:1") != defaultIdentity) +} + +func TestParseEndpointMessage(t *testing.T) { + port, upstream, err := parseEndpointMessage("80=host.docker.internal:49152") + assert.NilError(t, err) + assert.Equal(t, port, 80) + assert.Equal(t, upstream, "host.docker.internal:49152") + + for _, invalid := range []string{"80", "abc=host:1", "0=host:1", "80=nohostport"} { + _, _, err := parseEndpointMessage(invalid) + assert.Assert(t, err != nil, "expected error for %q", invalid) + } +} + +// relayRoutesSpec is canonically ordered: it is both the relay's runtime +// configuration and the identity hashed into its label, so map iteration +// order must never leak into it. +func TestRelayRoutesSpec(t *testing.T) { + routes := relayRoutesSpec(map[int]string{ + 9000: "host.docker.internal:30001", + 80: "host.docker.internal:49152", + }) + assert.Equal(t, routes, "80=host.docker.internal:49152,9000=host.docker.internal:30001") + + id1 := relayIdentity(routes) + id2 := relayIdentity(routes) + assert.Equal(t, id1, id2) + assert.Assert(t, id1 != relayIdentity("80=host.docker.internal:49153")) +} + +// The relay joins the networks of the services depending on the provider — +// its consumers — and falls back to the project default network. +func TestRelayNetworks(t *testing.T) { + db := types.ServiceConfig{Name: "db", Provider: &types.ServiceProviderConfig{Type: "test"}} + project := &types.Project{ + Name: "p", + Services: types.Services{ + "db": db, + "app": { + Name: "app", + DependsOn: types.DependsOnConfig{"db": {}}, + Networks: map[string]*types.ServiceNetworkConfig{"backend": nil}, + }, + "web": { + Name: "web", + DependsOn: types.DependsOnConfig{"db": {}}, + Networks: map[string]*types.ServiceNetworkConfig{"frontend": nil, "backend": nil}, + }, + "other": { + Name: "other", + Networks: map[string]*types.ServiceNetworkConfig{"private": nil}, + }, + }, + Networks: types.Networks{"default": {}, "backend": {}, "frontend": {}, "private": {}}, + } + + assert.DeepEqual(t, relayNetworks(project, db), []string{"backend", "frontend"}) + + // no consumer: fall back to the project default network + lonely := types.ServiceConfig{Name: "lonely", Provider: &types.ServiceProviderConfig{Type: "test"}} + assert.DeepEqual(t, relayNetworks(project, lonely), []string{"default"}) +} + +// Process-level commands refuse relay containers: there is no service +// process in them to act on. +func TestCheckRelayTarget(t *testing.T) { + relay := container.Summary{Labels: map[string]string{api.RelayLabel: "abc123"}} + err := checkRelayTarget(relay, "db", "exec") + assert.ErrorContains(t, err, "network relay") + assert.ErrorContains(t, err, "exec") + + regular := container.Summary{Labels: map[string]string{api.ServiceLabel: "db"}} + assert.NilError(t, checkRelayTarget(regular, "db", "exec")) +} diff --git a/pkg/e2e/providers_test.go b/pkg/e2e/providers_test.go index c2b5245f3e..18147ba51c 100644 --- a/pkg/e2e/providers_test.go +++ b/pkg/e2e/providers_test.go @@ -22,7 +22,10 @@ import ( "fmt" "os" "path/filepath" + "strings" "testing" + + "gotest.tools/v3/assert" ) // providerScenario creates a scenario whose commands can resolve the @@ -103,3 +106,22 @@ func TestProviderRawSetEnvOverridesInheritedEnvMapForm(t *testing.T) { OutputContains("test-1 | CLOUD_REGION=us-east-1"), OutputContains("overrides environment variable")) } + +func TestProviderPublishEndpoint(t *testing.T) { + // The example provider stands up a real endpoint on the host and + // publishes it; compose deploys a relay under the service's name, so the + // consumer reaches the provider's resource at the compose-native + // address http://db. The relay stands in for the service but refuses + // process-level commands. + relayImage := "compose-relay-e2e" + s := providerScenario(t, "a published endpoint must be reachable at the service's compose-native address") + s.CLI().RunCmd(t, "docker", "build", "-t", relayImage, "../../relay") + s.Env("PROVIDER_DEMO_ENDPOINT=1", "COMPOSE_RELAY_IMAGE="+relayImage) + s.Step("the consumer fetches through the relay at http://db", + ComposeCmd("up"), + OutputContains("test-1 | hello from provider")) + + res := s.CLI().RunDockerComposeCmdNoCheck(t, "--project-name", "e2e-provider-publish-endpoint", "exec", "db", "true") + assert.Assert(t, res.ExitCode != 0, "exec on a relay container must fail") + assert.Assert(t, strings.Contains(res.Combined(), "network relay"), res.Combined()) +} diff --git a/pkg/e2e/testdata/TestProviderPublishEndpoint/compose.yaml b/pkg/e2e/testdata/TestProviderPublishEndpoint/compose.yaml new file mode 100644 index 0000000000..885287740d --- /dev/null +++ b/pkg/e2e/testdata/TestProviderPublishEndpoint/compose.yaml @@ -0,0 +1,13 @@ +services: + test: + image: alpine + command: wget -qO- -T 10 http://db + depends_on: + - db + db: + provider: + type: example-provider + options: + name: db + type: test1 + size: 1 diff --git a/relay/Dockerfile b/relay/Dockerfile new file mode 100644 index 0000000000..a878279d02 --- /dev/null +++ b/relay/Dockerfile @@ -0,0 +1,28 @@ +# syntax=docker/dockerfile:1 + + +# Copyright 2020 Docker Compose CLI authors + +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at + +# http://www.apache.org/licenses/LICENSE-2.0 + +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +ARG GO_VERSION=1.26.8 + +FROM golang:${GO_VERSION}-alpine AS build +WORKDIR /src +COPY go.mod main.go ./ +RUN CGO_ENABLED=0 go build -trimpath -ldflags="-w -s" -o /compose-relay . + +FROM scratch +COPY --from=build /compose-relay /compose-relay +USER 65532:65532 +ENTRYPOINT ["/compose-relay"] diff --git a/relay/README.md b/relay/README.md new file mode 100644 index 0000000000..7fcc3ce373 --- /dev/null +++ b/relay/README.md @@ -0,0 +1,74 @@ +# compose-relay + +`compose-relay` is the network gateway Docker Compose deploys **in place of a +provider-managed service**, so the other services of the project reach the +provider's resource at the compose-native address — `http://:` — +even though that resource lives outside the compose network. + +## Why it exists + +A service can delegate its implementation to an external +[provider](../docs/extension.md): + +```yaml +services: + web: + build: . + depends_on: + - database + + database: + provider: + type: awesomecloud +``` + +The provider's resource (a cloud database, a sandbox, a host process, ...) is +not a container on the project network: `web` cannot resolve `database`, and +the resource's ports are typically published somewhere on the host, at +addresses and port numbers the application does not know. Until now consumers +had to read injected environment variables to locate it. + +When the provider declares where each endpoint actually listens, with one +`publish-endpoint` message per port: + +```json +{ "type": "publish-endpoint", "message": "5432=localhost:49152" } +``` + +Compose deploys this relay in place of the service. `web` then connects to +`database:5432` exactly as if the service were a regular container; the relay +forwards the connection to the real endpoint. + +## How it runs + +Compose creates the relay container from the published +[`docker/compose-relay`](https://hub.docker.com/r/docker/compose-relay) image +(override with `COMPOSE_RELAY_IMAGE`, e.g. for air-gapped setups or local +development) with: + +- the service's canonical container name (`--1`) and a + network alias set to the service name, on the networks of every service + that depends on the provider service; +- the standard compose labels, so label-driven commands (`ps`, `logs`, + `stop`, `down`) treat it as the service — plus the + `com.docker.compose.relay` label identifying its role. Its value is a hash + of the routes, letting `up` keep an up-to-date relay and recreate a stale + one. Commands that act on a service's process (`exec`, `cp`) refuse relay + containers; +- the routes as environment: + + ``` + RELAY_ROUTES=5432=localhost:49152[,=:...] + ``` + +The binary listens on every declared container port and forwards each +connection to its endpoint. TCP only, with half-close propagation so +protocols relying on EOF work through the relay. It is intentionally minimal: +a static Go binary on a `scratch` image, no configuration reload — Compose +recreates the relay when the published endpoints change. + +## Building + +```console +$ docker buildx bake relay-image +``` diff --git a/relay/go.mod b/relay/go.mod new file mode 100644 index 0000000000..7b4c943749 --- /dev/null +++ b/relay/go.mod @@ -0,0 +1,3 @@ +module github.com/docker/compose-relay + +go 1.23 diff --git a/relay/main.go b/relay/main.go new file mode 100644 index 0000000000..510d001986 --- /dev/null +++ b/relay/main.go @@ -0,0 +1,179 @@ +/* + Copyright 2020 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +// compose-relay is the network gateway Docker Compose deploys in place of a +// provider-managed service. It joins the project networks under the service's +// name and forwards each declared container port to the endpoint the provider +// published (typically a port on the host gateway), so consumers keep using +// the compose-native address (http://:) for a resource that +// lives outside the compose network. +// +// Routes come from the RELAY_ROUTES environment variable: +// +// RELAY_ROUTES=80=host.docker.internal:49152,9000=host.docker.internal:30001 +// +// One listener per route, TCP only. +package main + +import ( + "context" + "errors" + "fmt" + "io" + "log" + "net" + "os" + "os/signal" + "strconv" + "strings" + "sync" + "syscall" + "time" +) + +func main() { + routes, err := parseRoutes(os.Getenv("RELAY_ROUTES")) + if err != nil { + log.Fatalf("RELAY_ROUTES: %v", err) + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) + defer stop() + + var wg sync.WaitGroup + for port, upstream := range routes { + listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) + if err != nil { + log.Fatalf("listen :%d: %v", port, err) + } + log.Printf("relaying :%d -> %s", port, upstream) + wg.Add(1) + go func() { + defer wg.Done() + serve(ctx, listener, upstream, &wg) + }() + go func() { + <-ctx.Done() + _ = listener.Close() + }() + } + wg.Wait() +} + +// parseRoutes decodes "port=host:port[,port=host:port...]". +// +// A provider announces endpoints as seen from ITS host ("localhost:49152"): +// where a resource actually lives is not its concern to translate. The relay +// is the component that knows it runs inside a container, so a loopback (or +// unspecified) upstream host is rewritten here to host.docker.internal — the +// container-visible name for the host, resolved through the host-gateway +// extra_host compose injects. Routable addresses pass through untouched. +func parseRoutes(spec string) (map[int]string, error) { + if strings.TrimSpace(spec) == "" { + return nil, errors.New("no routes configured") + } + routes := map[int]string{} + for _, entry := range strings.Split(spec, ",") { + portPart, upstream, found := strings.Cut(entry, "=") + if !found { + return nil, fmt.Errorf("invalid route %q (want port=host:port)", entry) + } + port, err := strconv.Atoi(portPart) + if err != nil || port < 1 || port > 65535 { + return nil, fmt.Errorf("invalid port in route %q", entry) + } + host, hostPort, err := net.SplitHostPort(upstream) + if err != nil { + return nil, fmt.Errorf("invalid upstream in route %q: %v", entry, err) + } + if hostIsContainerLocal(host) { + upstream = net.JoinHostPort("host.docker.internal", hostPort) + } + routes[port] = upstream + } + return routes, nil +} + +// hostIsContainerLocal reports whether a host announced by the provider +// designates the provider's own host machine (loopback or unspecified) — +// unreachable under that name from inside a container. +func hostIsContainerLocal(host string) bool { + if host == "localhost" { + return true + } + ip := net.ParseIP(host) + return ip != nil && (ip.IsLoopback() || ip.IsUnspecified()) +} + +// serve accepts connections until the listener closes. In-flight forwards +// join the WaitGroup, so on SIGTERM the process stops accepting (listeners +// close) but drains established connections until they finish — or until the +// engine's stop timeout escalates to SIGKILL. +func serve(ctx context.Context, listener net.Listener, upstream string, wg *sync.WaitGroup) { + backoff := 5 * time.Millisecond + for { + conn, err := listener.Accept() + if err != nil { + if ctx.Err() != nil { + return + } + // Back off before retrying: some accept errors (e.g. EMFILE) + // persist until a descriptor is freed, and a tight loop would + // spin the CPU and flood the log. + log.Printf("accept on %s: %v", listener.Addr(), err) + time.Sleep(backoff) + if backoff < time.Second { + backoff *= 2 + } + continue + } + backoff = 5 * time.Millisecond + wg.Add(1) + go func() { + defer wg.Done() + forward(conn, upstream) + }() + } +} + +// forward deliberately takes no context: a connection accepted at the +// shutdown boundary (context cancelled, listener not yet closed) must still +// be served — that is the drain contract — and a cancelled context would make +// DialContext fail instantly. The dialer timeout bounds the dial instead. +func forward(downstream net.Conn, upstream string) { + defer downstream.Close() + dialer := net.Dialer{Timeout: 10 * time.Second} + up, err := dialer.Dial("tcp", upstream) + if err != nil { + log.Printf("dial %s: %v", upstream, err) + return + } + defer up.Close() + + done := make(chan struct{}, 2) + go func() { _, _ = io.Copy(up, downstream); closeWrite(up); done <- struct{}{} }() + go func() { _, _ = io.Copy(downstream, up); closeWrite(downstream); done <- struct{}{} }() + <-done + <-done +} + +// closeWrite propagates half-closes so protocols relying on EOF work through +// the relay. +func closeWrite(conn net.Conn) { + if tcp, ok := conn.(*net.TCPConn); ok { + _ = tcp.CloseWrite() + } +} diff --git a/relay/main_test.go b/relay/main_test.go new file mode 100644 index 0000000000..bf3adcc943 --- /dev/null +++ b/relay/main_test.go @@ -0,0 +1,50 @@ +/* + Copyright 2026 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package main + +import ( + "reflect" + "testing" +) + +// A provider announces endpoints as seen from its host; the relay, which +// knows it runs inside a container, rewrites host-local upstreams to +// host.docker.internal and leaves routable addresses untouched. +func TestParseRoutesTranslatesHostLocalUpstreams(t *testing.T) { + routes, err := parseRoutes("80=localhost:49152,81=127.0.0.1:5734,82=[::1]:5735,83=0.0.0.0:5736,443=192.168.1.10:8443") + if err != nil { + t.Fatal(err) + } + want := map[int]string{ + 80: "host.docker.internal:49152", + 81: "host.docker.internal:5734", + 82: "host.docker.internal:5735", + 83: "host.docker.internal:5736", + 443: "192.168.1.10:8443", + } + if !reflect.DeepEqual(routes, want) { + t.Fatalf("got %v, want %v", routes, want) + } +} + +func TestParseRoutesRejectsMalformedEntries(t *testing.T) { + for _, spec := range []string{"", "80", "80=nohostport", "0=localhost:1", "x=localhost:1"} { + if _, err := parseRoutes(spec); err == nil { + t.Errorf("parseRoutes(%q): expected error", spec) + } + } +}