diff --git a/pkg/container/docker/client.go b/pkg/container/docker/client.go index e05c5155ea..b8ab1f73a0 100644 --- a/pkg/container/docker/client.go +++ b/pkg/container/docker/client.go @@ -115,6 +115,7 @@ type deployOps interface { portBindings map[string][]runtime.PortBinding, isolateNetwork bool, ) error + getContainerNetworkIP(ctx context.Context, containerName, networkName string) (string, error) } // Client implements the Deployer interface for Docker (and compatible runtimes) @@ -323,9 +324,17 @@ func (c *Client) DeployWorkload( return 0, err // extractFirstPort already wraps the error with context. } if effectiveIsolation { - // SetupIngress runs after createMcpContainer so the reverse-proxy upstream - // resolves on first probe (a squid ingress created earlier would cache the - // negative DNS lookup and never recover). + // Point the ingress reverse proxy at the MCP container's IP rather than + // its name. Ordering ingress after createMcpContainer is not enough: under + // concurrent startup the container's DNS record can still lag, and the + // squid cache_peer caches the failed lookup and never recovers within the + // readiness window (see #6063). An IP has no lookup, so a transient + // connection failure is retried instead of latching the peer dead. + upstreamIP, ipErr := c.ops.getContainerNetworkIP(ctx, name, networkName) + if ipErr != nil { + return 0, fmt.Errorf("failed to resolve MCP container IP for ingress: %w", ipErr) + } + pspec.UpstreamHost = upstreamIP hostPort, err = c.proxy.SetupIngress(ctx, pspec, egress) if err != nil { return 0, fmt.Errorf("failed to set up ingress proxy: %w", err) @@ -1184,11 +1193,38 @@ func (c *Client) createNetwork( _, err = c.client.NetworkCreate(ctx, name, networkCreate) if err != nil { + // The existence check above is not atomic with the create: another + // process (e.g. a second `thv run` sharing the "toolhive-external" + // network) can create the same network in between. Docker rejects a + // duplicate name with a conflict error, so treat that as success — the + // network is already in the state we wanted. Without this, workloads + // that start concurrently race here and all but one fail to launch. + if errdefs.IsConflict(err) { + return nil + } return err } return nil } +// getContainerNetworkIP returns a container's IP address on the named network. +// Used to give the ingress proxy a DNS-free upstream address (see proxySpec.UpstreamHost). +func (c *Client) getContainerNetworkIP(ctx context.Context, containerName, networkName string) (string, error) { + inspectResult, err := c.client.ContainerInspect(ctx, containerName, mobyclient.ContainerInspectOptions{}) + if err != nil { + return "", fmt.Errorf("failed to inspect container %s: %w", containerName, err) + } + netSettings, ok := inspectResult.Container.NetworkSettings.Networks[networkName] + if !ok { + return "", fmt.Errorf("network %s not found in container %s network settings", networkName, containerName) + } + // EndpointSettings.IPAddress is a netip.Addr; render it to string form. + if !netSettings.IPAddress.IsValid() { + return "", fmt.Errorf("container %s has no IP address on network %s", containerName, networkName) + } + return netSettings.IPAddress.String(), nil +} + // getDockerBridgeGatewayIP returns the gateway IP of the Docker default bridge // network by inspecting it at runtime. This handles platform differences: // Linux Docker Engine uses 172.17.0.1 by default, while Docker Desktop on macOS @@ -1636,8 +1672,9 @@ func mergeEnvVars(base, extra map[string]string) map[string]string { // setupIngressContainer creates the ingress Squid reverse-proxy container for // the workload and returns the host-side port it is bound on. -func (c *Client) setupIngressContainer(ctx context.Context, containerName string, upstreamPort int, attachStdio bool, - externalEndpointsConfig map[string]*network.EndpointSettings, networkPermissions *permissions.NetworkPermissions) (int, error) { +func (c *Client) setupIngressContainer(ctx context.Context, containerName, upstreamHost string, upstreamPort int, + attachStdio bool, externalEndpointsConfig map[string]*network.EndpointSettings, + networkPermissions *permissions.NetworkPermissions) (int, error) { // A random port avoids every same-image workload racing to bind the same // preferred port when starting concurrently (see #6063). squidPort, err := networking.FindOrUsePort(0) @@ -1660,6 +1697,7 @@ func (c *Client) setupIngressContainer(ctx context.Context, containerName string ctx, c, containerName, + upstreamHost, ingressContainerName, attachStdio, upstreamPort, diff --git a/pkg/container/docker/client_deploy_test.go b/pkg/container/docker/client_deploy_test.go index 019cbefb6e..d61abb3ca8 100644 --- a/pkg/container/docker/client_deploy_test.go +++ b/pkg/container/docker/client_deploy_test.go @@ -32,6 +32,11 @@ type fakeDeployOps struct { dnsID string dnsIP string + // mcpIP is returned by getContainerNetworkIP as the MCP container's IP that + // the ingress proxy targets. errMcpIP injects a lookup failure. + mcpIP string + errMcpIP error + mcpCalled bool mcpName string mcpNetworkName string @@ -75,6 +80,10 @@ func (f *fakeDeployOps) createDnsContainer(_ context.Context, _ string, _ bool, return f.dnsID, f.dnsIP, f.errDNS } +func (f *fakeDeployOps) getContainerNetworkIP(_ context.Context, _, _ string) (string, error) { + return f.mcpIP, f.errMcpIP +} + func (f *fakeDeployOps) createMcpContainer( _ context.Context, name string, @@ -234,6 +243,7 @@ func TestDeployWorkload_SSE_IsolatedNetwork_ReturnsIngressPortAndPassesDNS(t *te fops := &fakeDeployOps{ dnsIP: "172.18.0.20", + mcpIP: "172.18.0.21", } fproxy := &fakeNetworkProxy{ ingressPort: 18081, @@ -269,6 +279,9 @@ func TestDeployWorkload_SSE_IsolatedNetwork_ReturnsIngressPortAndPassesDNS(t *te assert.True(t, fproxy.setupCalled) // The upstream port should be the first exposed port (8080). assert.Equal(t, 8080, fproxy.capturedSpec.UpstreamPort) + // The ingress proxy must target the MCP container's IP (not its name) so it + // has no DNS dependency to latch on under concurrent startup (see #6063). + assert.Equal(t, "172.18.0.21", fproxy.capturedSpec.UpstreamHost) require.True(t, fops.mcpCalled) assert.Equal(t, "172.18.0.20", fops.mcpAdditionalDNS, "additionalDNS passed to MCP container should come from DNS container IP") } diff --git a/pkg/container/docker/client_network_test.go b/pkg/container/docker/client_network_test.go new file mode 100644 index 0000000000..0c990aecbd --- /dev/null +++ b/pkg/container/docker/client_network_test.go @@ -0,0 +1,96 @@ +// SPDX-FileCopyrightText: Copyright 2025 Stacklok, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package docker + +import ( + "context" + "fmt" + "os" + "sync" + "testing" + + mobyclient "github.com/moby/moby/client" + "github.com/stretchr/testify/require" + + "github.com/stacklok/toolhive/pkg/container/docker/sdk" +) + +// TestCreateNetwork_ConcurrentSameName reproduces the shared-network startup +// race: multiple workloads (e.g. several `thv run` sharing "toolhive-external") +// call createNetwork for the same name at once. The existence check and the +// create are not atomic, so without idempotent conflict handling all but one +// caller fails with "network ... already exists" and its workload never +// launches. This test drives that exact contention against a real daemon and +// asserts every caller succeeds with exactly one network created. +// +// Skips when Docker is unavailable, matching the real-daemon tests in this +// package (see envoy_test.go). +func TestCreateNetwork_ConcurrentSameName(t *testing.T) { + t.Parallel() + + ctx := context.Background() + mc, _, _, err := sdk.NewDockerClient(ctx) + if err != nil { + t.Skipf("docker not available; skipping network create race test: %v", err) + } + c := &Client{client: mc} + + // Unique per test process so parallel runs / leftovers never collide. + name := fmt.Sprintf("toolhive-test-netrace-%d", os.Getpid()) + labels := map[string]string{"toolhive": "true"} + t.Cleanup(func() { removeNetworksByName(context.Background(), mc, name) }) + removeNetworksByName(ctx, mc, name) // start clean + + const workers = 8 + var wg sync.WaitGroup + start := make(chan struct{}) + errs := make([]error, workers) + for i := 0; i < workers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start // release all goroutines together to maximize contention + errs[i] = c.createNetwork(ctx, name, labels, false) + }(i) + } + close(start) + wg.Wait() + + for i, e := range errs { + require.NoErrorf(t, e, "worker %d: concurrent createNetwork must treat a lost race as success", i) + } + + require.Equal(t, 1, countNetworksByName(ctx, mc, name), + "exactly one shared network should exist after the concurrent race") +} + +func countNetworksByName(ctx context.Context, mc *mobyclient.Client, name string) int { + networks, err := mc.NetworkList(ctx, mobyclient.NetworkListOptions{ + Filters: mobyclient.Filters{}.Add("name", name), + }) + if err != nil { + return -1 + } + n := 0 + for _, item := range networks.Items { + if item.Name == name { + n++ + } + } + return n +} + +func removeNetworksByName(ctx context.Context, mc *mobyclient.Client, name string) { + networks, err := mc.NetworkList(ctx, mobyclient.NetworkListOptions{ + Filters: mobyclient.Filters{}.Add("name", name), + }) + if err != nil { + return + } + for _, item := range networks.Items { + if item.Name == name { + _, _ = mc.NetworkRemove(ctx, item.ID, mobyclient.NetworkRemoveOptions{}) + } + } +} diff --git a/pkg/container/docker/envoy.go b/pkg/container/docker/envoy.go index 38decd6176..143e48d832 100644 --- a/pkg/container/docker/envoy.go +++ b/pkg/container/docker/envoy.go @@ -776,6 +776,12 @@ func ingressDomains(_ proxySpec) []string { // buildIngressCluster returns the STRICT_DNS upstream cluster for the ingress // listener, pointing at spec.WorkloadName:spec.UpstreamPort. func buildIngressCluster(spec proxySpec) envoyCluster { + // Prefer the resolved upstream IP (no DNS dependency); fall back to the + // workload name. See proxySpec.UpstreamHost. + upstreamHost := spec.UpstreamHost + if upstreamHost == "" { + upstreamHost = spec.WorkloadName + } return envoyCluster{ Name: ingressClusterName, ConnectTimeout: "10s", @@ -790,7 +796,7 @@ func buildIngressCluster(spec proxySpec) envoyCluster { Endpoint: envoyEndpointAddress{ Address: envoyAddress{ SocketAddress: envoySocketAddress{ - Address: spec.WorkloadName, + Address: upstreamHost, PortValue: spec.UpstreamPort, }, }, diff --git a/pkg/container/docker/networkproxy.go b/pkg/container/docker/networkproxy.go index c1ad87b7a1..7b27a74a34 100644 --- a/pkg/container/docker/networkproxy.go +++ b/pkg/container/docker/networkproxy.go @@ -57,6 +57,13 @@ type networkProxy interface { type proxySpec struct { // WorkloadName is the base name of the MCP container (e.g. "myserver"). WorkloadName string + // UpstreamHost is the address the ingress reverse proxy connects to for the + // MCP container. It is the container's resolved IP on the internal network + // (not its name) so the ingress proxy has no DNS dependency: under + // concurrent startup the container's DNS record can lag, and a hostname-based + // Squid cache_peer caches the failed lookup and never recovers within the + // readiness window (see #6063). Empty falls back to WorkloadName. + UpstreamHost string // Permissions holds the network permission profile from the workload's // permission profile, governing what outbound traffic is allowed. Permissions *permissions.NetworkPermissions diff --git a/pkg/container/docker/squid.go b/pkg/container/docker/squid.go index 94452a3155..f784e3d802 100644 --- a/pkg/container/docker/squid.go +++ b/pkg/container/docker/squid.go @@ -42,6 +42,7 @@ func createIngressSquidContainer( ctx context.Context, c *Client, containerName string, + upstreamHost string, squidContainerName string, attachStdio bool, upstreamPort int, @@ -51,7 +52,7 @@ func createIngressSquidContainer( portBindings map[string][]runtime.PortBinding, networkPermissions *permissions.NetworkPermissions, ) (string, error) { - squidConfPath, err := createTempIngressSquidConf(containerName, upstreamPort, squidPort, networkPermissions) + squidConfPath, err := createTempIngressSquidConf(containerName, upstreamHost, upstreamPort, squidPort, networkPermissions) if err != nil { return "", fmt.Errorf("failed to create temporary squid.conf: %w", err) } @@ -385,8 +386,14 @@ func (s *squidProxy) SetupIngress(ctx context.Context, spec proxySpec, _ egressR if spec.TransportType == "stdio" || spec.UpstreamPort == 0 { return 0, nil } + // Prefer the resolved upstream IP so the cache_peer has no DNS dependency; + // fall back to the workload name when unset (see proxySpec.UpstreamHost). + upstreamHost := spec.UpstreamHost + if upstreamHost == "" { + upstreamHost = spec.WorkloadName + } ingressPort, err := s.client.setupIngressContainer( - ctx, spec.WorkloadName, spec.UpstreamPort, spec.AttachStdio, spec.Endpoints, spec.Permissions, + ctx, spec.WorkloadName, upstreamHost, spec.UpstreamPort, spec.AttachStdio, spec.Endpoints, spec.Permissions, ) if err != nil { return 0, err @@ -396,6 +403,7 @@ func (s *squidProxy) SetupIngress(ctx context.Context, spec proxySpec, _ egressR func createTempIngressSquidConf( serverHostname string, + upstreamHost string, upstreamPort int, squidPort int, networkPermissions *permissions.NetworkPermissions, @@ -404,7 +412,7 @@ func createTempIngressSquidConf( writeCommonConfig(&sb, serverHostname, proxyIngress) - writeIngressProxyConfig(&sb, serverHostname, upstreamPort, squidPort, networkPermissions) + writeIngressProxyConfig(&sb, serverHostname, upstreamHost, upstreamPort, squidPort, networkPermissions) sb.WriteString("http_access deny all\n") tmpFile, err := os.CreateTemp("", "squid-*.conf") @@ -433,21 +441,25 @@ func createTempIngressSquidConf( func writeIngressProxyConfig( sb *strings.Builder, serverHostname string, + upstreamHost string, upstreamPort int, squidPort int, networkPermissions *permissions.NetworkPermissions, ) { portNum := strconv.Itoa(upstreamPort) squidPortNum := strconv.Itoa(squidPort) - // standby=2 keeps warm idle connections open to the upstream so the first - // request after a cold start (notably a long-lived GET SSE stream that a - // server-initiated request rides on) is forwarded without paying inline DNS - // + TCP connect latency. Without it, the cold first GET races behind a - // later POST that reuses a warmed path, reordering server->client streams. + // cache_peer targets upstreamHost (the MCP container's IP), not its name, so + // the peer has no DNS lookup to cache-and-latch on a cold-start miss (#6063). + // defaultsite keeps the name for the origin Host header. standby=2 keeps warm + // idle connections open to the upstream so the first request after a cold + // start (notably a long-lived GET SSE stream that a server-initiated request + // rides on) is forwarded without paying inline TCP connect latency. Without + // it, the cold first GET races behind a later POST that reuses a warmed path, + // reordering server->client streams. sb.WriteString( "\n# Reverse proxy setup for port " + portNum + "\n" + "http_port 0.0.0.0:" + squidPortNum + " accel defaultsite=" + serverHostname + "\n" + - "cache_peer " + serverHostname + " parent " + portNum + " 0 no-query originserver name=origin_" + + "cache_peer " + upstreamHost + " parent " + portNum + " 0 no-query originserver name=origin_" + portNum + " connect-timeout=5 connect-fail-limit=5 standby=2\n") // Check if inbound network permissions are configured diff --git a/pkg/container/docker/squid_test.go b/pkg/container/docker/squid_test.go index 6761ec2afe..0e99d630f2 100644 --- a/pkg/container/docker/squid_test.go +++ b/pkg/container/docker/squid_test.go @@ -193,7 +193,7 @@ func TestCreateTempEgressSquidConf_WithACLs(t *testing.T) { func TestCreateTempIngressSquidConf_Basics(t *testing.T) { t.Parallel() - fp, err := createTempIngressSquidConf("svc-example", 8080, 18080, nil) + fp, err := createTempIngressSquidConf("svc-example", "10.89.0.7", 8080, 18080, nil) require.NoError(t, err) t.Cleanup(func() { _ = os.Remove(fp) }) @@ -203,8 +203,10 @@ func TestCreateTempIngressSquidConf_Basics(t *testing.T) { assert.Contains(t, s, "visible_hostname svc-example-ingress") assert.Contains(t, s, "\n# Reverse proxy setup for port 8080\n") + // defaultsite keeps the name; cache_peer targets the resolved upstream IP so + // the peer has no DNS lookup to latch on (see #6063). assert.Contains(t, s, "http_port 0.0.0.0:18080 accel defaultsite=svc-example") - assert.Contains(t, s, "cache_peer svc-example parent 8080 0 no-query originserver name=origin_8080") + assert.Contains(t, s, "cache_peer 10.89.0.7 parent 8080 0 no-query originserver name=origin_8080") // standby=2 pre-warms upstream connections so a cold first GET SSE stream is // not reordered behind a later POST (fixes the sampling conformance flake). assert.Contains(t, s, "connect-timeout=5 connect-fail-limit=5 standby=2") @@ -226,7 +228,7 @@ func TestCreateTempIngressSquidConf_WithOverrideHosts(t *testing.T) { }, } - fp, err := createTempIngressSquidConf("svc-example", 8080, 18080, networkPermissions) + fp, err := createTempIngressSquidConf("svc-example", "10.89.0.7", 8080, 18080, networkPermissions) require.NoError(t, err) t.Cleanup(func() { _ = os.Remove(fp) }) @@ -237,7 +239,7 @@ func TestCreateTempIngressSquidConf_WithOverrideHosts(t *testing.T) { assert.Contains(t, s, "visible_hostname svc-example-ingress") assert.Contains(t, s, "\n# Reverse proxy setup for port 8080\n") assert.Contains(t, s, "http_port 0.0.0.0:18080 accel defaultsite=svc-example") - assert.Contains(t, s, "cache_peer svc-example parent 8080 0 no-query originserver name=origin_8080") + assert.Contains(t, s, "cache_peer 10.89.0.7 parent 8080 0 no-query originserver name=origin_8080") // Test that override mode is used - no default ACLs assert.NotContains(t, s, "acl site_8080 dstdomain svc-example") @@ -269,7 +271,7 @@ func TestCreateTempIngressSquidConf_EmptyInboundHosts(t *testing.T) { }, } - fp, err := createTempIngressSquidConf("svc-example", 8080, 18080, networkPermissions) + fp, err := createTempIngressSquidConf("svc-example", "10.89.0.7", 8080, 18080, networkPermissions) require.NoError(t, err) t.Cleanup(func() { _ = os.Remove(fp) }) @@ -499,7 +501,7 @@ func TestTempFilesWrittenToSystemTempDir(t *testing.T) { require.NoError(t, err) t.Cleanup(func() { _ = os.Remove(fp1) }) - fp2, err := createTempIngressSquidConf("s2", 8081, 18081, nil) + fp2, err := createTempIngressSquidConf("s2", "10.89.0.8", 8081, 18081, nil) require.NoError(t, err) t.Cleanup(func() { _ = os.Remove(fp2) })