diff --git a/cmd/atenet/internal/router/cmd.go b/cmd/atenet/internal/router/cmd.go index bcb410f88..353037500 100644 --- a/cmd/atenet/internal/router/cmd.go +++ b/cmd/atenet/internal/router/cmd.go @@ -83,7 +83,7 @@ func NewRouterCmd() *cobra.Command { // must propagate to the Service endpoints before the drain starts. cmd.Flags().DurationVar(&cfg.DrainDelay, "drain-delay", 13*time.Second, "How long to keep serving after SIGTERM before starting the drain, covering readiness-probe detection and Service endpoint propagation") cmd.Flags().DurationVar(&cfg.DrainTimeout, "drain-timeout", 0, "Deadline for the ext_proc drain on shutdown; streams still open past it (parked requests included) are forcefully cancelled. 0 (the default) derives --parked-request-budget + the actor route timeout + margin so parked requests always finish normally. Explicit values must be >= --parked-request-budget") - cmd.Flags().StringVar(&cfg.EnvoyAdminAddr, "envoy-admin-address", "127.0.0.1:9901", "Envoy admin interface the shutdown sequence drives to drain the sidecar (healthcheck/fail, drain_listeners, stats polling). Ignored with --atenet-router=agentgateway") + cmd.Flags().StringVar(&cfg.EnvoyAdminAddr, "envoy-admin-address", "localhost:9901", "Envoy admin interface the shutdown sequence drives to drain the sidecar (healthcheck/fail, drain_listeners, stats polling). Ignored with --atenet-router=agentgateway") cmd.Flags().StringVar(&cfg.DrainCompleteFile, "drain-complete-file", defaultDrainCompleteFile, "Marker file created (on a pod-shared emptyDir) once the shutdown drain completes; the dataplane container's preStop hook polls for it so the proxy exits as soon as — and no sooner than — the drain is done. Removed at startup to defuse stale markers. Empty disables the handshake") return cmd diff --git a/cmd/atenet/internal/router/dataplane.go b/cmd/atenet/internal/router/dataplane.go index bd9f2abfc..190e73cd8 100644 --- a/cmd/atenet/internal/router/dataplane.go +++ b/cmd/atenet/internal/router/dataplane.go @@ -39,7 +39,9 @@ type dataplaneHealthCheck struct { func (r atenetRouter) healthCheck() dataplaneHealthCheck { switch r { case atenetRouterEnvoy: - return dataplaneHealthCheck{url: "http://127.0.0.1:9901/ready", expectedBody: "LIVE"} + // localhost, not 127.0.0.1: the admin socket binds `::`, so the dial + // has to be able to fall through to the IPv6 loopback. + return dataplaneHealthCheck{url: "http://localhost:9901/ready", expectedBody: "LIVE"} case atenetRouterAgentgateway: return dataplaneHealthCheck{url: "http://127.0.0.1:15021/healthz/ready", expectedBody: "ready"} default: diff --git a/cmd/atenet/internal/router/drain_test.go b/cmd/atenet/internal/router/drain_test.go index 160e1a887..b4b7083af 100644 --- a/cmd/atenet/internal/router/drain_test.go +++ b/cmd/atenet/internal/router/drain_test.go @@ -16,6 +16,7 @@ package router import ( "context" + "net" "net/http" "net/http/httptest" "os" @@ -280,6 +281,39 @@ func TestEnvoyDrainerDrainsToZero(t *testing.T) { } } +// TestEnvoyDrainerReachesIPv6OnlyAdmin guards the loopback coupling behind +// --envoy-admin-address: the gateway admin sockets bind "::", and a drainer +// pinned to the IPv4 loopback would find nothing listening, read that as an +// exited Envoy, and report a drain it never performed -- silently, since that +// path returns nil. Hence the assertion on the POSTs rather than on the error. +func TestEnvoyDrainerReachesIPv6OnlyAdmin(t *testing.T) { + ln, err := net.Listen("tcp6", "[::1]:0") + if err != nil { + t.Skipf("no IPv6 loopback on this host: %v", err) + } + admin := &fakeEnvoyAdmin{activeSeries: []int{0}} + srv := httptest.NewUnstartedServer(admin.handler()) + srv.Listener.Close() + srv.Listener = ln + srv.Start() + defer srv.Close() + + d := newEnvoyDrainer(net.JoinHostPort("localhost", itoa(ln.Addr().(*net.TCPAddr).Port))) + d.pollInterval = 5 * time.Millisecond + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := d.Drain(ctx); err != nil { + t.Fatalf("Drain: %v", err) + } + + admin.mu.Lock() + defer admin.mu.Unlock() + if len(admin.posts) != 2 { + t.Errorf("admin POSTs = %v, want the drain to have reached the IPv6 loopback", admin.posts) + } +} + // TestEnvoyDrainerAdminGone asserts an unreachable admin interface (Envoy // already exited) is treated as a completed drain, quickly. func TestEnvoyDrainerAdminGone(t *testing.T) { diff --git a/cmd/atenet/internal/router/health_test.go b/cmd/atenet/internal/router/health_test.go index 3c2d52157..a44176c67 100644 --- a/cmd/atenet/internal/router/health_test.go +++ b/cmd/atenet/internal/router/health_test.go @@ -80,7 +80,7 @@ func TestCheckDataplane(t *testing.T) { { name: "envoy", router: atenetRouterEnvoy, - wantURL: "http://127.0.0.1:9901/ready", + wantURL: "http://localhost:9901/ready", response: "LIVE", wantMessage: "LIVE", }, diff --git a/cmd/atenet/internal/router/xds.go b/cmd/atenet/internal/router/xds.go index e2d76d8bf..07b5b1cab 100644 --- a/cmd/atenet/internal/router/xds.go +++ b/cmd/atenet/internal/router/xds.go @@ -1108,6 +1108,27 @@ func (x *XdsServer) buildTracing() *hcmv3.HttpConnectionManager_Tracing { } } +// dualStackAdditionalAddresses returns the IPv6 half of a dual-stack ingress +// listener, to pair with a primary 0.0.0.0 socket on the same port. Ipv4Compat +// stays false: clearing IPV6_V6ONLY would collide with that primary. +func dualStackAdditionalAddresses(port int) []*listenerv3.AdditionalAddress { + return []*listenerv3.AdditionalAddress{ + { + Address: &corev3.Address{ + Address: &corev3.Address_SocketAddress{ + SocketAddress: &corev3.SocketAddress{ + Address: "::", + Ipv4Compat: false, + PortSpecifier: &corev3.SocketAddress_PortValue{ + PortValue: uint32(port), + }, + }, + }, + }, + }, + } +} + func (x *XdsServer) buildListener() *listenerv3.Listener { hcm := x.buildHcm("ingress_http", true) @@ -1123,6 +1144,7 @@ func (x *XdsServer) buildListener() *listenerv3.Listener { }, }, }, + AdditionalAddresses: dualStackAdditionalAddresses(x.ingressPort), FilterChains: []*listenerv3.FilterChain{ { Filters: []*listenerv3.Filter{ @@ -1182,6 +1204,7 @@ func (x *XdsServer) buildHttpsListener() *listenerv3.Listener { }, }, }, + AdditionalAddresses: dualStackAdditionalAddresses(x.httpsPort), FilterChains: []*listenerv3.FilterChain{ { Filters: []*listenerv3.Filter{ @@ -1213,6 +1236,7 @@ func (x *XdsServer) buildConnectTerminateListener() *listenerv3.Listener { }, }, }, + AdditionalAddresses: dualStackAdditionalAddresses(x.connectPlainTextPort), FilterChains: []*listenerv3.FilterChain{ { Filters: []*listenerv3.Filter{ @@ -1246,6 +1270,7 @@ func (x *XdsServer) buildConnectTerminateTLSListener() *listenerv3.Listener { }, }, }, + AdditionalAddresses: dualStackAdditionalAddresses(x.connectTLSPort), FilterChains: []*listenerv3.FilterChain{ { Filters: []*listenerv3.Filter{ diff --git a/cmd/atenet/internal/router/xds_test.go b/cmd/atenet/internal/router/xds_test.go index 6fa5c428b..ea39fb7c4 100644 --- a/cmd/atenet/internal/router/xds_test.go +++ b/cmd/atenet/internal/router/xds_test.go @@ -48,6 +48,36 @@ import ( "github.com/agent-substrate/substrate/internal/atunnel" ) +// assertDualStackIngress checks an ingress listener keeps its 0.0.0.0 primary +// and gains exactly one "::" socket on the same port. +func assertDualStackIngress(t *testing.T, l *listenerv3.Listener, wantPort uint32) { + t.Helper() + + sa := l.GetAddress().GetSocketAddress() + if sa.GetAddress() != "0.0.0.0" { + t.Errorf("Expected address '0.0.0.0', got %s", sa.GetAddress()) + } + if sa.GetPortValue() != wantPort { + t.Errorf("Expected port %d, got %d", wantPort, sa.GetPortValue()) + } + + addrs := l.GetAdditionalAddresses() + if len(addrs) != 1 { + t.Fatalf("Expected 1 additional address on %s, got %d", l.GetName(), len(addrs)) + } + + asa := addrs[0].GetAddress().GetSocketAddress() + if asa.GetAddress() != "::" { + t.Errorf("Expected additional address '::', got %s", asa.GetAddress()) + } + if asa.GetIpv4Compat() { + t.Error("Expected additional address Ipv4Compat to be false") + } + if asa.GetPortValue() != wantPort { + t.Errorf("Expected additional port %d, got %d", wantPort, asa.GetPortValue()) + } +} + func TestXdsServer_UpdateSnapshot(t *testing.T) { server := NewXdsServer(18000) server.SetConfig(8081, 50052, "10.0.0.1") @@ -150,14 +180,7 @@ func TestXdsServer_UpdateSnapshot(t *testing.T) { if raw, exists := listenersMap[IngressHTTPListener]; !exists { t.Errorf("Listener name '%s' is missing from snapshot listeners", IngressHTTPListener) } else { - l := raw.(*listenerv3.Listener) - sa := l.GetAddress().GetSocketAddress() - if sa.GetPortValue() != 8081 { - t.Errorf("Expected port 8081, got %d", sa.GetPortValue()) - } - if sa.GetAddress() != "0.0.0.0" { - t.Errorf("Expected address '0.0.0.0', got %s", sa.GetAddress()) - } + assertDualStackIngress(t, raw.(*listenerv3.Listener), 8081) } } @@ -192,10 +215,7 @@ func TestXdsServer_UpdateSnapshot_WithHttps(t *testing.T) { t.Errorf("Listener name '%s' is missing from snapshot listeners", IngressHTTPSListener) } else { l := raw.(*listenerv3.Listener) - sa := l.GetAddress().GetSocketAddress() - if sa.GetPortValue() != 8443 { - t.Errorf("Expected port 8443, got %d", sa.GetPortValue()) - } + assertDualStackIngress(t, l, 8443) // Verify the TLS config references the serving cert via SDS rather // than embedding it: inline filename DataSources are read only once @@ -354,16 +374,14 @@ func TestXdsServer_UpdateSnapshot_WithConnect(t *testing.T) { } if raw, exists := listenersMap["connect_terminate"]; !exists { t.Error("connect_terminate listener missing") - } else if sa := raw.(*listenerv3.Listener).GetAddress().GetSocketAddress(); sa.GetPortValue() != 8081 { - t.Errorf("Expected connect_terminate port 8081, got %d", sa.GetPortValue()) + } else { + assertDualStackIngress(t, raw.(*listenerv3.Listener), 8081) } if raw, exists := listenersMap["connect_terminate_tls"]; !exists { t.Error("connect_terminate_tls listener missing") } else { l := raw.(*listenerv3.Listener) - if sa := l.GetAddress().GetSocketAddress(); sa.GetPortValue() != 8444 { - t.Errorf("Expected connect_terminate_tls port 8444, got %d", sa.GetPortValue()) - } + assertDualStackIngress(t, l, 8444) ts := l.GetFilterChains()[0].GetTransportSocket() if ts.GetName() != "envoy.transport_sockets.tls" { t.Errorf("Expected connect_terminate_tls to be TLS-wrapped, got transport socket %q", ts.GetName()) diff --git a/manifests/ate-install/atenet-egress.yaml b/manifests/ate-install/atenet-egress.yaml index 591ef31fc..78d4f1e4c 100644 --- a/manifests/ate-install/atenet-egress.yaml +++ b/manifests/ate-install/atenet-egress.yaml @@ -37,12 +37,16 @@ data: envoy.yaml: | admin: address: - socket_address: { address: 0.0.0.0, port_value: 15000 } + # ipv4_compat is load-bearing: the probes below are the kubelet + # dialling the pod IP, which is IPv4 on an IPv4 cluster. + socket_address: { address: "::", ipv4_compat: true, port_value: 15000 } static_resources: listeners: - name: egress address: - socket_address: { address: 0.0.0.0, port_value: 443 } + # ipv4_compat rather than a second socket: IPv4 peers arrive as + # ::ffff: and nothing here reads the peer -- identity is the cert. + socket_address: { address: "::", ipv4_compat: true, port_value: 443 } filter_chains: # Named so ext_proc can read it back as xds.filter_chain_name. Must # match EgressFilterChainName in @@ -304,7 +308,7 @@ spec: # defaults to (that is the ingress gateway's port). Without this the # drain sequence dials a closed port, reads the connection refusal as # "Envoy already exited", and reports a drain it never performed. - - --envoy-admin-address=127.0.0.1:15000 + - --envoy-admin-address=localhost:15000 env: - name: POD_NAME valueFrom: @@ -379,6 +383,7 @@ metadata: namespace: ate-system spec: type: ClusterIP + ipFamilyPolicy: PreferDualStack selector: app: atenet-egress ports: diff --git a/manifests/ate-install/atenet-router.yaml b/manifests/ate-install/atenet-router.yaml index e05e06efb..018dbd0a4 100644 --- a/manifests/ate-install/atenet-router.yaml +++ b/manifests/ate-install/atenet-router.yaml @@ -86,7 +86,10 @@ data: admin: address: socket_address: - address: 0.0.0.0 + # ipv4_compat keeps `kubectl port-forward` to the admin port working; + # in-pod callers dial localhost and need no help. + address: "::" + ipv4_compat: true port_value: 9901 node: @@ -354,6 +357,7 @@ metadata: namespace: ate-system spec: type: ClusterIP + ipFamilyPolicy: PreferDualStack selector: app: atenet-router ports: