atenet: bind the Envoy sockets and Services dual-stack

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

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:

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) {

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",
 		},

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{

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())

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:

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: