diff --git a/internal/e2e/dns_client.go b/internal/e2e/dns_client.go new file mode 100644 index 000000000..76db25d4f --- /dev/null +++ b/internal/e2e/dns_client.go @@ -0,0 +1,162 @@ +// Copyright 2026 Google LLC +// +// 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 e2e + +import ( + "context" + "errors" + "fmt" + "net" + "strconv" + "strings" + "time" + + "github.com/agent-substrate/substrate/internal/ateclient" + "github.com/agent-substrate/substrate/internal/portforward" + "k8s.io/client-go/kubernetes" +) + +const ( + dnsNamespace = "ate-system" + dnsService = "dns" + // dnsServicePort is the Service port; the Service exposes 53 twice, once + // UDP and once TCP, and the port-forward tunnel is TCP either way. + dnsServicePort = 53 +) + +// DNSRcode is how the server answered, at the granularity net.Resolver exposes. +type DNSRcode int + +const ( + // DNSAnswered is NOERROR with at least one address of the queried family. + DNSAnswered DNSRcode = iota + // DNSEmpty is "this name has no address in this family": NODATA (NOERROR + // with an empty answer section) or NXDOMAIN. net.Resolver reports both as + // DNSError.IsNotFound and the standard library offers no way to tell them + // apart, which is fine here — both are benign to every stub resolver, and + // that benign-ness is the property under test. + DNSEmpty + // DNSFailed is SERVFAIL, REFUSED, a timeout, or a malformed reply: anything + // net.Resolver classifies as the server misbehaving. No query into the actor + // zone should ever produce it — not a non-A qtype, not a name that fails the + // actor regex — and that is the regression these tests exist to catch. + DNSFailed +) + +func (r DNSRcode) String() string { + switch r { + case DNSAnswered: + return "answered" + case DNSEmpty: + return "no-such-host (NODATA or NXDOMAIN)" + case DNSFailed: + return "server failure (SERVFAIL/REFUSED/timeout)" + default: + return "unknown" + } +} + +// DNSClient resolves names against the ate-system/dns CoreDNS Service over a +// port-forward. +// +// Querying that Service directly, rather than going through the cluster's own +// resolver, is deliberate: the delegation that would make actor names resolvable +// cluster-wide is a patch to the kube-system/kube-dns ConfigMap, which only +// exists on GKE (cmd/atenet/internal/dns/dns.go reconcileKubeDNSConfig hits the +// IsNotFound branch on kind and upstream Kubernetes). Pointing at the Service is +// the only way to assert the zone's behavior on every cluster we test on. +type DNSClient struct { + resolver *net.Resolver + stop func() +} + +// NewDNSClient establishes a port-forward to the atenet DNS Service. Call Close +// to tear it down. +func NewDNSClient(ctx context.Context) (*DNSClient, error) { + config, err := ateclient.LoadConfig(KubeConfig, KubeContext) + if err != nil { + return nil, fmt.Errorf("loading kubeconfig: %w", err) + } + clientset, err := kubernetes.NewForConfig(config) + if err != nil { + return nil, fmt.Errorf("creating k8s client: %w", err) + } + + localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, dnsNamespace, dnsService, dnsServicePort) + if err != nil { + return nil, err + } + addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(localPort)) + + return &DNSClient{ + stop: stop, + resolver: &net.Resolver{ + // PreferGo keeps us on Go's own resolver on every platform. cgo's + // would ignore Dial entirely and query the host's nameservers. + PreferGo: true, + // Surface a per-family failure instead of hiding it behind the + // other family's success. + StrictErrors: true, + Dial: func(ctx context.Context, _, _ string) (net.Conn, error) { + // The port-forward is a TCP tunnel, so every query goes over TCP + // whatever network the resolver asked for. net.Resolver selects + // stream framing for any conn that is not a net.PacketConn, so + // returning a TCP conn here is transparent to it. The requested + // server address is ignored: there is exactly one server. + var d net.Dialer + return d.DialContext(ctx, "tcp", addr) + }, + }, + }, nil +} + +// Close tears down the port-forward. +func (c *DNSClient) Close() { + if c.stop != nil { + c.stop() + } +} + +// Lookup resolves name in a single address family — network is "ip4" for an A +// query or "ip6" for a AAAA query — and reports the addresses alongside how the +// server answered. A DNSFailed result is returned with the underlying error for +// the failure message; DNSEmpty is returned with a nil error because it is a +// valid answer, not a fault. +func (c *DNSClient) Lookup(ctx context.Context, network, name string) ([]string, DNSRcode, error) { + // Root the name so the resolver skips the host's search list and ndots + // handling, which would otherwise make the query depend on where the test + // runs. + if !strings.HasSuffix(name, ".") { + name += "." + } + + lookupCtx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + + addrs, err := c.resolver.LookupNetIP(lookupCtx, network, name) + if err == nil { + ips := make([]string, 0, len(addrs)) + for _, a := range addrs { + ips = append(ips, a.Unmap().String()) + } + return ips, DNSAnswered, nil + } + + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) && dnsErr.IsNotFound { + return nil, DNSEmpty, nil + } + return nil, DNSFailed, fmt.Errorf("%s query for %q: %w", network, name, err) +} diff --git a/internal/e2e/ipfamily.go b/internal/e2e/ipfamily.go new file mode 100644 index 000000000..3d72fcc04 --- /dev/null +++ b/internal/e2e/ipfamily.go @@ -0,0 +1,69 @@ +// Copyright 2026 Google LLC +// +// 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 e2e + +import ( + "context" + "fmt" + "net/netip" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// clusterIPsByFamily splits a Service's cluster IPs into its IPv4 and IPv6 +// entries, returning "" for a family the Service does not have. A Service with +// no ipFamilyPolicy is SingleStack, so on a dual-stack cluster it still has +// exactly one ClusterIP and one of the two return values is empty — which is +// what makes this the right thing to gate a dual-stack assertion on. +// +// Spec.ClusterIPs is preferred over the singular Spec.ClusterIP, with a +// fallback for the latter because a Service object built by hand (or by a fake +// client) may only set the scalar. +func clusterIPsByFamily(svc *corev1.Service) (v4, v6 string) { + ips := svc.Spec.ClusterIPs + if len(ips) == 0 && svc.Spec.ClusterIP != "" { + ips = []string{svc.Spec.ClusterIP} + } + for _, ip := range ips { + if ip == "" || ip == corev1.ClusterIPNone { + continue + } + // netip rather than net.IP: net.IP.To4 returns non-nil for a v4-mapped + // v6 address and would misfile it as IPv4. + addr, err := netip.ParseAddr(ip) + if err != nil { + continue + } + switch { + case addr.Is4() && v4 == "": + v4 = ip + case addr.Is6() && !addr.Is4In6() && v6 == "": + v6 = ip + } + } + return v4, v6 +} + +// RouterClusterIPs returns the atenet-router Service's IPv4 and IPv6 +// ClusterIPs. Either may be "". +func RouterClusterIPs(ctx context.Context) (v4, v6 string, err error) { + svc, err := GetClients().K8s.CoreV1().Services(RouterNamespace).Get(ctx, RouterService, metav1.GetOptions{}) + if err != nil { + return "", "", fmt.Errorf("getting Service %s/%s: %w", RouterNamespace, RouterService, err) + } + v4, v6 = clusterIPsByFamily(svc) + return v4, v6, nil +} diff --git a/internal/e2e/router_client.go b/internal/e2e/router_client.go index 4a0c006b2..7992ba036 100644 --- a/internal/e2e/router_client.go +++ b/internal/e2e/router_client.go @@ -36,8 +36,11 @@ import ( ) const ( - routerNamespace = "ate-system" - routerService = "atenet-router" + // RouterNamespace and RouterService locate the atenet router. Exported so + // that suites addressing the same Service or its pods do not have to + // redeclare them. + RouterNamespace = "ate-system" + RouterService = "atenet-router" // routerConnectServicePort is atenet-router's Service port for // CONNECT-tunneled traffic (see manifests/ate-install/atenet-router.yaml). // It is a distinct listener from the plain HTTP one Get/PostJSON use: @@ -79,7 +82,7 @@ func NewRouterClient(ctx context.Context) (*RouterClient, error) { return nil, fmt.Errorf("creating k8s client: %w", err) } - localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, routerNamespace, routerService, 80) + localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, RouterNamespace, RouterService, 80) if err != nil { return nil, err } @@ -183,7 +186,7 @@ func (c *RouterClient) Connect(ctx context.Context, actorRef resources.ActorRef, // in one test don't each pay for a fresh port-forward. func (c *RouterClient) ensureConnectPortForward(ctx context.Context) error { c.connectOnce.Do(func() { - localPort, stop, err := portforward.ServicePortForward(ctx, c.config, c.clientset, routerNamespace, routerService, routerConnectServicePort) + localPort, stop, err := portforward.ServicePortForward(ctx, c.config, c.clientset, RouterNamespace, RouterService, routerConnectServicePort) if err != nil { c.connectErr = fmt.Errorf("port-forwarding to the router's CONNECT listener: %w", err) return diff --git a/internal/e2e/statusz.go b/internal/e2e/statusz.go index a04a07141..1890a036f 100644 --- a/internal/e2e/statusz.go +++ b/internal/e2e/statusz.go @@ -51,7 +51,7 @@ func NewStatuszClient(ctx context.Context) (*StatuszClient, error) { return nil, fmt.Errorf("creating k8s client: %w", err) } - localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, routerNamespace, routerService, routerStatusPort) + localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, RouterNamespace, RouterService, routerStatusPort) if err != nil { return nil, err } diff --git a/internal/e2e/suites/networking/dns_test.go b/internal/e2e/suites/networking/dns_test.go new file mode 100644 index 000000000..b8abf001c --- /dev/null +++ b/internal/e2e/suites/networking/dns_test.go @@ -0,0 +1,163 @@ +// Copyright 2026 Google LLC +// +// 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 networking + +import ( + "context" + "slices" + "testing" + + "github.com/agent-substrate/substrate/internal/e2e" + "github.com/agent-substrate/substrate/internal/resources" +) + +// The actor zone is served by a CoreDNS `template` block, which answers for any +// name matching .. whether or not that actor exists. +// These tests therefore need no actor fixture — they are asserting the zone's +// behavior, not an actor's. +func probeActorDNSName() string { + return resources.ActorDNSName(resources.ActorRef{Atespace: networkingAtespace, Name: "dns-probe"}) +} + +func mustDNSClient(t *testing.T, ctx context.Context) *e2e.DNSClient { + t.Helper() + dns, err := e2e.NewDNSClient(ctx) + if err != nil { + t.Fatalf("NewDNSClient: %v", err) + } + t.Cleanup(dns.Close) + return dns +} + +// TestActorDNSZone asserts that the actor zone answers an A query with the +// router's ClusterIP, and — the part no other test covers — that everything +// else it is asked returns a *benign* rcode rather than SERVFAIL. +// +// The rcode matters more than the missing record. NODATA and NXDOMAIN are what +// every stub resolver expects for "there is no address here"; SERVFAIL is a +// transport fault, and resolvers disagree about it. musl maps it to EAI_AGAIN +// and abandons the whole getaddrinfo — and because musl issues the A and AAAA +// queries in parallel, one SERVFAIL sinks the other with it, so an Alpine-based +// client cannot resolve an actor name at all, not even its A record. glibc +// retries and pays the resolver timeout instead. And unlike NODATA and +// NXDOMAIN, SERVFAIL carries no SOA to cache negatively against, so every +// request re-pays that cost. Go's resolver masks all of this, which is why no +// test in this repo caught it before these. +// +// The zone gets the rcodes right with three `template` blocks in +// cmd/atenet/internal/dns/corefile.go: the `IN A` block that answers actor +// names, a regex-matched `template ANY ANY` returning NOERROR plus an SOA +// authority (NODATA) for other qtypes on a well-formed actor name, and a +// terminal `template ANY ANY` returning NXDOMAIN plus an SOA authority for +// everything else in the zone. The first two carry a bare `fallthrough`, which +// is load-bearing: the plugin walks past a class or qtype mismatch by itself, +// but a *regex* miss returns SERVFAIL immediately unless the block declares it. +// The two subtests below are what keeps that from being collapsed back into a +// single block. +// +// This test is family-agnostic and is expected to run, not skip, on a +// single-stack cluster. +func TestActorDNSZone(t *testing.T) { + ctx := context.Background() + dns := mustDNSClient(t, ctx) + + routerV4, _, err := e2e.RouterClusterIPs(ctx) + if err != nil { + t.Fatalf("reading atenet-router ClusterIPs: %v", err) + } + + name := probeActorDNSName() + + t.Run("A answers with the router ClusterIP", func(t *testing.T) { + if routerV4 == "" { + // A v6-only cluster: there is no IPv4 ClusterIP to answer with, and + // emitting an A record at all would be the bug. + t.Skip("atenet-router has no IPv4 ClusterIP") + } + addrs, rcode, err := dns.Lookup(ctx, "ip4", name) + if rcode != e2e.DNSAnswered { + t.Fatalf("A %s: %v (%v); want the router ClusterIP %s", name, rcode, err, routerV4) + } + if !slices.Contains(addrs, routerV4) { + t.Fatalf("A %s = %v; want it to contain the atenet-router ClusterIP %s", name, addrs, routerV4) + } + }) + + t.Run("AAAA is not a server failure", func(t *testing.T) { + // The name is well-formed, so NODATA is the answer owed here on a + // single-stack cluster: it exists, it just has no address in this + // family. That needs a block that matches the qtype. Were the `IN A` + // template the zone's only one, a qtype mismatch would fall through to + // a plugin chain with nothing after it, and plugin.NextOrFailure with a + // nil Next returns SERVFAIL. + _, rcode, err := dns.Lookup(ctx, "ip6", name) + if rcode == e2e.DNSFailed { + t.Fatalf("AAAA %s: %v (%v); want NODATA. A SERVFAIL on a non-A qtype in this "+ + "zone breaks musl-based clients on IPv4-only clusters too, because it takes "+ + "their parallel A query down with it", name, rcode, err) + } + }) + + t.Run("a name outside the actor pattern is not a server failure", func(t *testing.T) { + // A single-label name inside the zone: the zone matches, the qtype + // matches, the actor regex does not. This is the case that depends on + // both halves of the corefile fix at once. A regex miss is the one kind + // of non-match the template plugin does not walk past on its own -- it + // consults fall.Through() and, absent a bare `fallthrough`, answers + // SERVFAIL without evaluating any later block. So the two regex-matched + // templates each need `fallthrough` to decline the name, and the + // terminal catch-all `template ANY ANY` is what turns it into NXDOMAIN. + // Drop either piece and this subtest goes red. + bogus := "not-an-actor." + resources.ActorDNSSuffix + _, rcode, err := dns.Lookup(ctx, "ip4", bogus) + if rcode == e2e.DNSFailed { + t.Fatalf("A %s: %v (%v); want NXDOMAIN", bogus, rcode, err) + } + }) +} + +// TestActorDNSAAAA asserts the zone publishes the router's IPv6 ClusterIP. +// +// Skipped unless the atenet-router Service actually has one, which is the +// steady state on every single-stack cluster: a Service with no +// ipFamilyPolicy is SingleStack and never gets a second ClusterIP, so there is +// nothing an AAAA could correctly point at. +// +// Deliberately separate from TestActorDNSZone: it is the only assertion here +// whose expected result changes when the cluster becomes dual-stack, so keeping +// it its own function lets a dual-stack CI job exclude it while the AAAA +// generator is still in flight. +func TestActorDNSAAAA(t *testing.T) { + ctx := context.Background() + + _, routerV6, err := e2e.RouterClusterIPs(ctx) + if err != nil { + t.Fatalf("reading atenet-router ClusterIPs: %v", err) + } + if routerV6 == "" { + t.Skip("atenet-router has no IPv6 ClusterIP; single-stack cluster, nothing to publish") + } + + dns := mustDNSClient(t, ctx) + name := probeActorDNSName() + + addrs, rcode, err := dns.Lookup(ctx, "ip6", name) + if rcode != e2e.DNSAnswered { + t.Fatalf("AAAA %s: %v (%v); want the router IPv6 ClusterIP %s", name, rcode, err, routerV6) + } + if !slices.Contains(addrs, routerV6) { + t.Fatalf("AAAA %s = %v; want it to contain the atenet-router IPv6 ClusterIP %s", name, addrs, routerV6) + } +} diff --git a/internal/e2e/suites/networking/ingress_family_test.go b/internal/e2e/suites/networking/ingress_family_test.go new file mode 100644 index 000000000..4a2e50d1d --- /dev/null +++ b/internal/e2e/suites/networking/ingress_family_test.go @@ -0,0 +1,280 @@ +// Copyright 2026 Google LLC +// +// 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 networking + +import ( + "context" + "encoding/json" + "fmt" + "maps" + "net" + "os/exec" + "slices" + "strconv" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/agent-substrate/substrate/internal/e2e" + "github.com/agent-substrate/substrate/internal/portforward" + "github.com/agent-substrate/substrate/internal/resources" +) + +const ( + routerAppLabel = "app=atenet-router" + // envoyAdminPort is the admin listener in the router pod's envoy container. + // It is not published by the Service, so the pod proxy subresource is the + // only way at it from a test. + envoyAdminPort = 9901 + + // Listener names from cmd/atenet/internal/router/xds.go. They cannot be + // imported: that package is under cmd/atenet/internal, so only cmd/atenet + // may import it. + ingressHTTPListener = "ingress_http_listener" + ingressHTTPSListener = "ingress_https_listener" + connectTerminateListener = "connect_terminate" + connectTerminateTLSListener = "connect_terminate_tls" + + // Same digest-pinned image the networkpolicy suite probes with. BusyBox's + // wget handles bracketed IPv6 URLs and honors a user-supplied Host header + // instead of adding its own, which is exactly what is needed here. + probeImage = "busybox@sha256:1487d0af5f52b4ba31c7e465126ee2123fe3f2305d638e7827681e7cf6c83d5e" +) + +// envoyListeners is the subset of Envoy's admin /listeners?format=json response +// this test reads. additional_local_addresses is how a listener with +// Listener.additional_addresses reports its extra sockets. +type envoyListeners struct { + ListenerStatuses []struct { + Name string `json:"name"` + LocalAddress struct { + SocketAddress envoySocketAddress `json:"socket_address"` + } `json:"local_address"` + AdditionalLocalAddresses []struct { + SocketAddress envoySocketAddress `json:"socket_address"` + } `json:"additional_local_addresses"` + } `json:"listener_statuses"` +} + +type envoySocketAddress struct { + Address string `json:"address"` + PortValue int `json:"port_value"` +} + +// TestRouterListenerAddresses asserts, from Envoy's own view of itself, that +// each of the router's dataplane listeners — the two ingress ones and the two +// CONNECT ones — bound both an IPv4 and an IPv6 socket. +// +// This is the cheap half of the ingress coverage and the one that runs +// everywhere: the "::" socket binds on a single-stack IPv4 cluster too, it just +// carries no traffic there. It is also the only assertion in the suite that can +// fail when someone removes the IPv6 socket, because every other path a test has +// into the router — a port-forward, the pods/proxy subresource, the +// services/proxy subresource — is mediated by the API server and reaches the +// pod over whatever family the *kubelet or apiserver* chooses. None of them let +// the test select an address family, so none of them can select a listener +// socket. +func TestRouterListenerAddresses(t *testing.T) { + ctx := context.Background() + clients := e2e.GetClients() + pod := mustRouterPodName(t, ctx) + + raw, err := clients.K8s.CoreV1().RESTClient().Get(). + Namespace(e2e.RouterNamespace). + Resource("pods"). + Name(pod+":"+strconv.Itoa(envoyAdminPort)). + SubResource("proxy"). + Suffix("listeners"). + Param("format", "json"). + DoRaw(ctx) + if err != nil { + // The pods/proxy subresource reaches the pod on its primary-family + // PodIP, so this hop used to be family-sensitive. It no longer is: the + // admin listener binds "::" with ipv4_compat + // (manifests/ate-install/atenet-router.yaml), which accepts connections + // from either family. A failure here means the admin interface is not + // answering — the container is not up, or the proxy path is blocked. + t.Fatalf("reading Envoy admin /listeners from %s/%s: %v", e2e.RouterNamespace, pod, err) + } + + var listeners envoyListeners + if err := json.Unmarshal(raw, &listeners); err != nil { + t.Fatalf("decoding /listeners response %q: %v", raw, err) + } + if len(listeners.ListenerStatuses) == 0 { + t.Fatalf("Envoy reports no listeners at all; xDS has not converged. Body: %s", raw) + } + + // One entry per listener name: every address it is bound on. + bound := map[string][]string{} + for _, ls := range listeners.ListenerStatuses { + addrs := []string{ls.LocalAddress.SocketAddress.Address} + for _, extra := range ls.AdditionalLocalAddresses { + addrs = append(addrs, extra.SocketAddress.Address) + } + bound[ls.Name] = addrs + } + + // Only the plain HTTP listener is unconditional. The other three exist + // only when --port-https, --port-connect and --port-connect-tls are set; + // all are set in the shipped manifest, but do not make this test the thing + // that fails if that changes. + for _, l := range []struct { + name string + required bool + }{ + {ingressHTTPListener, true}, + {ingressHTTPSListener, false}, + {connectTerminateListener, false}, + {connectTerminateTLSListener, false}, + } { + name := l.name + t.Run(name, func(t *testing.T) { + addrs, ok := bound[name] + if !ok { + if !l.required { + t.Skipf("router has no %s; listeners present: %v", name, slices.Sorted(maps.Keys(bound))) + } + t.Fatalf("router has no %s; listeners present: %v", name, slices.Sorted(maps.Keys(bound))) + } + // The IPv4 socket is the one that carries all production traffic + // today; losing it is the expensive regression, so assert it first. + if !slices.Contains(addrs, "0.0.0.0") { + t.Errorf("%s is bound on %v; want an IPv4 wildcard socket (0.0.0.0)", name, addrs) + } + if !slices.Contains(addrs, "::") { + t.Errorf("%s is bound on %v; want an IPv6 wildcard socket (::) as well. "+ + "Envoy binds it on a single-stack cluster too, so this failing means the "+ + "listener lost its additional_addresses entry", name, addrs) + } + }) + } +} + +// TestActorIngressPerFamily reaches an actor through the router over each of the +// router Service's ClusterIPs, from a pod inside the cluster. +// +// The client has to be in-cluster: e2e.RouterClient port-forwards to +// 127.0.0.1, which tunnels through the API server to the kubelet, so its own +// address family says nothing about which of the router's sockets served the +// request. +// +// Skipped unless the router Service is dual-stack. On a single-stack cluster +// there is exactly one ClusterIP and TestActorDirectAccess already covers it. +func TestActorIngressPerFamily(t *testing.T) { + ctx := context.Background() + + routerV4, routerV6, err := e2e.RouterClusterIPs(ctx) + if err != nil { + t.Fatalf("reading atenet-router ClusterIPs: %v", err) + } + if routerV4 == "" || routerV6 == "" { + t.Skipf("atenet-router is single-stack (v4=%q v6=%q); nothing to compare", routerV4, routerV6) + } + + actorName, _ := createAndResumeActor(t, ctx, "family", e2e.CounterFixture()) + dnsName := resources.ActorDNSName(resources.ActorRef{Atespace: networkingAtespace, Name: actorName}) + + probeNS := e2e.CreateNamespace(t) + probePod := startProbePod(t, ctx, probeNS.Name) + + // Both families, in one test: the point of dual-stack is that both work, + // and a change that turns the IPv4 socket off is the costly failure. + for _, tc := range []struct{ family, clusterIP string }{ + {"ipv4", routerV4}, + {"ipv6", routerV6}, + } { + t.Run(tc.family, func(t *testing.T) { + // The request must carry the actor's DNS name as the Host: it is + // the only routing key the router's ext_proc has. Only the + // *connection* goes to the literal. + url := fmt.Sprintf("http://%s/readyz", net.JoinHostPort(tc.clusterIP, "80")) + out, err := execInPod(probeNS.Name, probePod, + "wget", "-q", "-T", "10", "-O", "-", "--header", "Host: "+dnsName, url) + if err != nil { + t.Fatalf("GET %s (Host: %s) from %s/%s over %s failed: %v; output: %s", + url, dnsName, probeNS.Name, probePod, tc.family, err, out) + } + t.Logf("actor reached over %s via %s; body: %s", tc.family, tc.clusterIP, strings.TrimSpace(out)) + }) + } +} + +func mustRouterPodName(t *testing.T, ctx context.Context) string { + t.Helper() + pods, err := e2e.GetClients().K8s.CoreV1().Pods(e2e.RouterNamespace).List(ctx, metav1.ListOptions{ + LabelSelector: routerAppLabel, + }) + if err != nil { + t.Fatalf("listing atenet-router pods: %v", err) + } + for i := range pods.Items { + if portforward.IsPodReady(&pods.Items[i]) { + return pods.Items[i].Name + } + } + t.Fatalf("no ready atenet-router pod in %s", e2e.RouterNamespace) + return "" +} + +func startProbePod(t *testing.T, ctx context.Context, namespace string) string { + t.Helper() + clients := e2e.GetClients() + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "ingress-probe", Namespace: namespace}, + Spec: corev1.PodSpec{ + RestartPolicy: corev1.RestartPolicyNever, + Containers: []corev1.Container{{ + Name: "probe", + Image: probeImage, + Command: []string{"/bin/sleep", "3600"}, + }}, + }, + } + if _, err := clients.K8s.CoreV1().Pods(namespace).Create(ctx, pod, metav1.CreateOptions{}); err != nil { + t.Fatalf("creating probe pod %s/%s: %v", namespace, pod.Name, err) + } + + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + got, err := clients.K8s.CoreV1().Pods(namespace).Get(ctx, pod.Name, metav1.GetOptions{}) + if err == nil && got.Status.Phase == corev1.PodRunning { + return pod.Name + } + time.Sleep(time.Second) + } + t.Fatalf("timed out waiting for probe pod %s/%s to run", namespace, pod.Name) + return "" +} + +// execInPod runs a command in a pod. It shells out to kubectl, matching what +// the networkpolicy suite already does, rather than pulling in client-go's +// remotecommand plumbing for two calls. +func execInPod(namespace, pod string, command ...string) (string, error) { + args := []string{} + if e2e.KubeConfig != "" { + args = append(args, "--kubeconfig="+e2e.KubeConfig) + } + if e2e.KubeContext != "" { + args = append(args, "--context="+e2e.KubeContext) + } + args = append(args, "exec", "-n", namespace, pod, "--") + args = append(args, command...) + out, err := exec.Command("kubectl", args...).CombinedOutput() + return string(out), err +}