From 91017032487ae24914a8bd1958a78bd07a9c403d Mon Sep 17 00:00:00 2001 From: Jet Chiang Date: Wed, 1 Jul 2026 12:29:24 -0400 Subject: [PATCH] initial poc Signed-off-by: Jet Chiang agentgateway Signed-off-by: Jet Chiang extproc docs Signed-off-by: Jet Chiang revised: docs and manifests Signed-off-by: Jet Chiang revised: docs and manifests Signed-off-by: Jet Chiang remove default envoy and scripts Signed-off-by: Jet Chiang cleanup a bit Signed-off-by: Jet Chiang cleanup old files from main Signed-off-by: Jet Chiang --- cmd/atenet/internal/dns.go | 21 +- cmd/atenet/internal/dns/dns.go | 27 +- cmd/atenet/internal/dns/dns_test.go | 4 +- cmd/atenet/internal/router.go | 12 +- cmd/atenet/internal/router/README.md | 17 +- cmd/atenet/internal/router/controller.go | 28 +- cmd/atenet/internal/router/dashboard.html | 27 - cmd/atenet/internal/router/envoyrunner.go | 258 -------- cmd/atenet/internal/router/errors.go | 92 --- cmd/atenet/internal/router/errors_test.go | 194 ------ cmd/atenet/internal/router/extproc.go | 83 ++- cmd/atenet/internal/router/extproc_test.go | 260 ++++---- cmd/atenet/internal/router/health.go | 76 +-- cmd/atenet/internal/router/route.go | 98 +++ cmd/atenet/internal/router/route_errors.go | 84 +++ cmd/atenet/internal/router/route_resolve.go | 78 +++ .../internal/router/route_resolve_test.go | 256 ++++++++ cmd/atenet/internal/router/router.go | 88 +-- cmd/atenet/internal/router/status.go | 4 - cmd/atenet/internal/router/status_test.go | 4 +- cmd/atenet/internal/router/xds.go | 606 ------------------ cmd/atenet/internal/router/xds_test.go | 212 ------ cmd/benchmarking/boomer-glutton/main.go | 2 +- docs/dev/extproc-ingress-contract.md | 145 +++++ docs/dev/ingress.md | 90 +++ hack/install-ate.sh | 13 + hack/poc-pluggable-ingress.sh | 120 ++++ internal/e2e/router_client.go | 9 +- manifests/ate-install/atenet-dns.yaml | 1 + .../ate-install/atenet-router-monitoring.yaml | 35 - manifests/ate-install/atenet-router.yaml | 116 +--- .../atenet-gateway-agw-gatewayapi.yaml | 62 ++ .../examples/atenet-gateway-agw.yaml | 123 ++++ .../examples/atenet-gateway-envoy.yaml | 200 ++++++ 34 files changed, 1520 insertions(+), 1925 deletions(-) delete mode 100644 cmd/atenet/internal/router/envoyrunner.go delete mode 100644 cmd/atenet/internal/router/errors.go delete mode 100644 cmd/atenet/internal/router/errors_test.go create mode 100644 cmd/atenet/internal/router/route.go create mode 100644 cmd/atenet/internal/router/route_errors.go create mode 100644 cmd/atenet/internal/router/route_resolve.go create mode 100644 cmd/atenet/internal/router/route_resolve_test.go delete mode 100644 cmd/atenet/internal/router/xds.go delete mode 100644 cmd/atenet/internal/router/xds_test.go create mode 100644 docs/dev/extproc-ingress-contract.md create mode 100644 docs/dev/ingress.md create mode 100755 hack/poc-pluggable-ingress.sh delete mode 100644 manifests/ate-install/atenet-router-monitoring.yaml create mode 100644 manifests/ate-install/examples/atenet-gateway-agw-gatewayapi.yaml create mode 100644 manifests/ate-install/examples/atenet-gateway-agw.yaml create mode 100644 manifests/ate-install/examples/atenet-gateway-envoy.yaml diff --git a/cmd/atenet/internal/dns.go b/cmd/atenet/internal/dns.go index 471e93679..88109d027 100644 --- a/cmd/atenet/internal/dns.go +++ b/cmd/atenet/internal/dns.go @@ -33,10 +33,11 @@ import ( ) type DnsConfig struct { - LogLevel string - Kubeconfig string - ReconcileInterval time.Duration - CorefilePath string + LogLevel string + Kubeconfig string + ReconcileInterval time.Duration + CorefilePath string + IngressServiceName string } func NewDnsCmd() *cobra.Command { @@ -87,10 +88,11 @@ func NewDnsCmd() *cobra.Command { } dnsController := &dns.Controller{ - Client: k8sClient, - Interval: cfg.ReconcileInterval, - CorefilePath: cfg.CorefilePath, - Reloader: dns.NewConfigReloader(), + Client: k8sClient, + Interval: cfg.ReconcileInterval, + CorefilePath: cfg.CorefilePath, + Reloader: dns.NewConfigReloader(), + IngressServiceName: cfg.IngressServiceName, } slog.InfoContext(ctx, "Starting DNS Controller subsystem") @@ -102,6 +104,9 @@ func NewDnsCmd() *cobra.Command { cmd.Flags().StringVar(&cfg.Kubeconfig, "kubeconfig", "", "Absolute path to the kubeconfig configuration file") cmd.Flags().DurationVar(&cfg.ReconcileInterval, "interval", 10*time.Second, "Interval for reconciling DNS configurations") cmd.Flags().StringVar(&cfg.CorefilePath, "corefile-path", "/etc/coredns/Corefile", "Path to the local Corefile configuration on shared volume") + cmd.Flags().StringVar(&cfg.IngressServiceName, "ingress-service-name", dns.DefaultIngressServiceName, + "Kubernetes Service in ate-system whose ClusterIP actor DNS resolves to. "+ + "Set this to whatever Service fronts your gateway if you didn't name it \"atenet-gateway\".") return cmd } diff --git a/cmd/atenet/internal/dns/dns.go b/cmd/atenet/internal/dns/dns.go index cf2db99b6..e59909ae9 100644 --- a/cmd/atenet/internal/dns/dns.go +++ b/cmd/atenet/internal/dns/dns.go @@ -37,6 +37,13 @@ const ( // serviceName is the name of the CoreDNS service. serviceName = "dns" systemNamespace = "ate-system" + + // DefaultIngressServiceName is the Kubernetes Service whose ClusterIP + // CoreDNS resolves actor hostnames to. atenet-router only serves ext_proc + // gRPC, not HTTP, so this must be the Service fronting whatever gateway + // you bring (see docs/dev/ingress.md); atenet-gateway is the name used by + // the reference configs there. + DefaultIngressServiceName = "atenet-gateway" ) // Controller manages the DNS configuration for the ATE. @@ -45,6 +52,10 @@ type Controller struct { Interval time.Duration CorefilePath string Reloader ConfigReloader + // IngressServiceName is the Kubernetes Service in ate-system whose + // ClusterIP is written into the CoreDNS Corefile as the ingress target. + // Defaults to DefaultIngressServiceName ("atenet-gateway") when empty. + IngressServiceName string } // Run the DNS orchestration loop until ctx is canceled. @@ -71,19 +82,25 @@ func (c *Controller) Run(ctx context.Context) error { func (c *Controller) reconcile(ctx context.Context) error { slog.DebugContext(ctx, "Reconciling DNS orchestration configuration...") - // 1. Get the ClusterIP of atenet-router in ate-system namespace + // 1. Get the ClusterIP of the ingress service (your gateway) in ate-system. + ingressSvcName := c.IngressServiceName + if ingressSvcName == "" { + ingressSvcName = DefaultIngressServiceName + } routerSvc := &corev1.Service{} - if err := c.Client.Get(ctx, types.NamespacedName{Name: "atenet-router", Namespace: systemNamespace}, routerSvc); err != nil { + if err := c.Client.Get(ctx, types.NamespacedName{Name: ingressSvcName, Namespace: systemNamespace}, routerSvc); err != nil { if errors.IsNotFound(err) { - slog.WarnContext(ctx, "atenet-router service not found, skipping until it is available") + slog.WarnContext(ctx, "ingress service not found, skipping until it is available", + slog.String("service", ingressSvcName)) return nil } - return fmt.Errorf("failed to get atenet-router service: %w", err) + return fmt.Errorf("failed to get ingress service %q: %w", ingressSvcName, err) } routerIP := routerSvc.Spec.ClusterIP if routerIP == "" || routerIP == "None" { - slog.WarnContext(ctx, "atenet-router service has no ClusterIP yet, waiting...") + slog.WarnContext(ctx, "ingress service has no ClusterIP yet, waiting...", + slog.String("service", ingressSvcName)) return nil } diff --git a/cmd/atenet/internal/dns/dns_test.go b/cmd/atenet/internal/dns/dns_test.go index 34116db28..ae257616b 100644 --- a/cmd/atenet/internal/dns/dns_test.go +++ b/cmd/atenet/internal/dns/dns_test.go @@ -46,7 +46,7 @@ func TestReconcile(t *testing.T) { // 1. Create mock services routerSvc := &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ - Name: "atenet-router", + Name: "atenet-gateway", Namespace: "ate-system", }, Spec: corev1.ServiceSpec{ @@ -151,7 +151,7 @@ func TestReconcileKubeDNSNotFound(t *testing.T) { routerSvc := &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ - Name: "atenet-router", + Name: "atenet-gateway", Namespace: "ate-system", }, Spec: corev1.ServiceSpec{ diff --git a/cmd/atenet/internal/router.go b/cmd/atenet/internal/router.go index c58f31a7e..7f58ee1a9 100644 --- a/cmd/atenet/internal/router.go +++ b/cmd/atenet/internal/router.go @@ -29,7 +29,7 @@ func NewRouterCmd() *cobra.Command { cmd := &cobra.Command{ Use: "router", - Short: "Router components including xDS server and Envoy ExtProc gateway processing server", + Short: "Substrate routing core and ExtProc server", RunE: func(cmd *cobra.Command, args []string) error { srv, err := router.NewRouterServer(cfg) if err != nil { @@ -47,17 +47,11 @@ func NewRouterCmd() *cobra.Command { cmd.Flags().StringVar(&cfg.Namespace, "namespace", "default", "Target operations namespace") cmd.Flags().StringVar(&cfg.Kubeconfig, "kubeconfig", "", "Absolute path to the kubeconfig configuration file") cmd.Flags().StringVar(&cfg.AteapiAddr, "ateapi-address", "api.ate-system.svc:443", "gRPC host address of the cluster ateapi Control instance") - cmd.Flags().IntVar(&cfg.HttpPort, "port-http", 8080, "TCP port for workload traffic entering through the Envoy Router") - cmd.Flags().IntVar(&cfg.XdsPort, "port-xds", 18000, "TCP port listening for the xDS dynamic Envoy connections") - cmd.Flags().IntVar(&cfg.ExtprocPort, "port-extproc", 50051, "Listen port for the Envoy dynamic External Processing (ext_proc) server") - cmd.Flags().StringVar(&cfg.ExtprocAddr, "extproc-address", "127.0.0.1", "Host IP or address of the Envoy External Processing (ext_proc) server") - cmd.Flags().StringVar(&cfg.EnvoyImage, "envoy-image", "envoyproxy/envoy:v1.30-latest", "Image URI used for dynamically launched router instances") + cmd.Flags().IntVar(&cfg.ExtprocPort, "port-extproc", 50051, "Listen port for the ext_proc server") + cmd.Flags().StringVar(&cfg.ExtprocAddr, "extproc-address", "127.0.0.1", "Host IP or address of the ext_proc server") cmd.Flags().StringVar(&cfg.TemplatesFile, "actor-templates-file", "", "Path to offline YAML configuration file listing ActorTemplates") cmd.Flags().IntVar(&cfg.StatusPort, "status-port", 4040, "Port to serve /statusz on (set <= 0 to disable serving status)") cmd.Flags().DurationVar(&cfg.HealthInterval, "health-interval", 1*time.Second, "Interval for checking health of dependent services") - cmd.Flags().IntVar(&cfg.HttpsPort, "port-https", 8443, "TCP port for HTTPS workload traffic entering through the Envoy Router") - cmd.Flags().StringVar(&cfg.EnvoyCertPath, "envoy-cert-path", "", "Path to the Envoy certificate file (if empty, a self-signed cert will be generated for testing)") - cmd.Flags().StringVar(&cfg.OtlpCollectorAddress, "otlp-collector-address", "", "host:port of the OTLP gRPC collector that Envoy reports tracing spans to (empty disables Envoy tracing)") cmd.Flags().StringVar(&cfg.AteapiAuthMode, "ateapi-auth", "mtls", "Client auth to ateapi: mtls|jwt. 'mtls' (default) dials with insecure TLS and relies on pod-projected mTLS credentials for identity. 'jwt' verifies the server cert and sends a Bearer SA token.") cmd.Flags().StringVar(&cfg.AteapiCAFile, "ateapi-ca-file", ateapiauth.DefaultServiceAccountCAFile, "PEM file with CAs trusted to verify the ateapi server cert. Required for jwt.") cmd.Flags().StringVar(&cfg.AteapiServerName, "ateapi-server-name", "", "SNI / hostname expected on the ateapi server cert. Optional.") diff --git a/cmd/atenet/internal/router/README.md b/cmd/atenet/internal/router/README.md index 4f05ae839..78bf7b26f 100644 --- a/cmd/atenet/internal/router/README.md +++ b/cmd/atenet/internal/router/README.md @@ -2,21 +2,18 @@ Router has several responsibilities: -* (Optional) manages a Deployment of Envoy to function as a router for ATE requests. - * This is optional to enable testing the router component in a standalone mode without managing the Kubernetes objects. - * Envoy will be configured to send traffic to via xDS served by the Router. -* ext_proc server for the Envoy. To make the deployment and debugging easier, we will run this component together - with the router, but this will be split later into its own component. - * ext_proc will call into the ATE gRPC API to get the set of relevant backends (specific the worker IP) and - route the traffic accordingly +* Runs the actor-aware routing core: resolve actor identity from a request, + resume/locate the actor via the ATE gRPC API, and pick a worker endpoint. * Make sure the interface with ATE API is pluggable so that we can test with a mock ATE API. -* Runs an xDS server for the Envoy deployment that defines the Cluster information for the ATEs. - * the xDS configuration will configure Envoy to send traffic to ext_proc +* Exposes that routing core over ext_proc, so any proxy that supports the + ExtProc protocol (and dynamic forwarding) can call it. Substrate does not + ship or manage a proxy — see [`docs/dev/ingress.md`](../../../../docs/dev/ingress.md) + for reference gateway configs. * Watches the ActorTemplates to get out the definitions for how to route the session IDs. ## status page -Serve a `/statusz` page on port 8080. +Serve a `/statusz` page on the router's status port. Contents: diff --git a/cmd/atenet/internal/router/controller.go b/cmd/atenet/internal/router/controller.go index 1b0a1ed18..75d2821d4 100644 --- a/cmd/atenet/internal/router/controller.go +++ b/cmd/atenet/internal/router/controller.go @@ -23,28 +23,22 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" ) -// Controller monitors ActorTemplates and coordinates configuration updates -// for the Envoy xDS and external processing servers. +// Controller monitors ActorTemplates for the external processing server. type Controller struct { k8sClient client.Client clientset kubernetes.Interface cfg RouterConfig - xdsSrv *XdsServer extprocSrv *ExtProcServer - atStore atStore - envoyRunner *envoyrunner + atStore atStore } func NewController( k8sClient client.Client, clientset kubernetes.Interface, cfg RouterConfig, - xdsSrv *XdsServer, extprocSrv *ExtProcServer, ) *Controller { - xdsSrv.SetConfig(cfg.HttpPort, cfg.ExtprocPort, cfg.ExtprocAddr) - var store atStore if cfg.TemplatesFile != "" { store = newFileATStore(cfg.TemplatesFile) @@ -56,11 +50,9 @@ func NewController( k8sClient: k8sClient, clientset: clientset, cfg: cfg, - xdsSrv: xdsSrv, extprocSrv: extprocSrv, - atStore: store, - envoyRunner: newEnvoyRunner(k8sClient, cfg), + atStore: store, } } @@ -92,19 +84,5 @@ func (c *Controller) reconcile(ctx context.Context) error { return err } - if err := c.xdsSrv.UpdateSnapshot(); err != nil { - slog.ErrorContext(ctx, "xDS Configuration generation problem", slog.String("err", err.Error())) - return err - } - - if !c.cfg.Standalone && c.cfg.TemplatesFile == "" { - // Reconcile Envoy router Deployment and Kubernetes cluster entities - err := c.envoyRunner.reconcile(ctx) - if err != nil { - slog.ErrorContext(ctx, "Error during Envoy router reconciliation", slog.String("err", err.Error())) - return err - } - } - return nil } diff --git a/cmd/atenet/internal/router/dashboard.html b/cmd/atenet/internal/router/dashboard.html index 41f3d93af..03d94e652 100644 --- a/cmd/atenet/internal/router/dashboard.html +++ b/cmd/atenet/internal/router/dashboard.html @@ -256,14 +256,6 @@

atenet Router Status

Component Network Allocation
-
- Workload Port (Http Envoy) - {{ .HttpPort }} -
-
- xDS Interface Port - {{ .XdsPort }} -
External Processing Port {{ .ExtprocPort }} @@ -277,25 +269,6 @@

atenet Router Status

System Component Health Checks
-
- Envoy Health - {{ if .Health.Envoy.Healthy }} - Healthy - {{ else }} - Degraded - {{ end }} -
-
- Message / Err: - {{ .Health.Envoy.Message }} -
-
- Ok: {{ .Health.Envoy.SuccessCount }} | Err: {{ .Health.Envoy.FailureCount }} - LKG: {{ if not .Health.Envoy.LastSuccess.IsZero }}{{ .Health.Envoy.LastSuccess.Format "15:04:05" }}{{ else }}N/A{{ end }} -
- -
-
Kubernetes API {{ if .Health.K8sAPI.Healthy }} diff --git a/cmd/atenet/internal/router/envoyrunner.go b/cmd/atenet/internal/router/envoyrunner.go deleted file mode 100644 index df8765c62..000000000 --- a/cmd/atenet/internal/router/envoyrunner.go +++ /dev/null @@ -1,258 +0,0 @@ -// 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 router - -import ( - "context" - "fmt" - "log/slog" - - appsv1 "k8s.io/api/apps/v1" - corev1 "k8s.io/api/core/v1" - k8serrors "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/util/intstr" - "sigs.k8s.io/controller-runtime/pkg/client" -) - -const ( - EnvoyDeploymentName = "atenet-router-envoy" - EnvoyServiceName = "atenet-router-envoy" - EnvoyConfigMapName = "atenet-router-envoy-config" -) - -// envoyrunner manages the dynamic deployment and lifecycle of the underlying -// Envoy proxy instance running inside Kubernetes. -type envoyrunner struct { - k8sClient client.Client - cfg RouterConfig -} - -func newEnvoyRunner(k8sClient client.Client, cfg RouterConfig) *envoyrunner { - return &envoyrunner{ - k8sClient: k8sClient, - cfg: cfg, - } -} - -func (r *envoyrunner) reconcile(ctx context.Context) error { - if err := r.reconcileEnvoyConfigMap(ctx); err != nil { - return fmt.Errorf("failed configmap reconciliation: %w", err) - } - - if err := r.reconcileEnvoyDeployment(ctx); err != nil { - return fmt.Errorf("failed deployment reconciliation: %w", err) - } - - if err := r.reconcileEnvoyService(ctx); err != nil { - return fmt.Errorf("failed service reconciliation: %w", err) - } - - return nil -} - -func (r *envoyrunner) reconcileEnvoyConfigMap(ctx context.Context) error { - envoyYaml := fmt.Sprintf(`admin: - address: - socket_address: - address: 0.0.0.0 - port_value: 9901 - -node: - id: %s - cluster: substrate-router-cluster - -dynamic_resources: - lds_config: - resource_api_version: V3 - ads: {} - cds_config: - resource_api_version: V3 - ads: {} - ads_config: - api_type: GRPC - transport_api_version: V3 - grpc_services: - - envoy_grpc: - cluster_name: xds_cluster - -static_resources: - clusters: - - name: xds_cluster - connect_timeout: 0.25s - type: STRICT_DNS - lb_policy: ROUND_ROBIN - typed_extension_protocol_options: - envoy.extensions.upstreams.http.v3.HttpProtocolOptions: - "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions - explicit_http_config: - http2_protocol_options: {} - load_assignment: - cluster_name: xds_cluster - endpoints: - - lb_endpoints: - - endpoint: - address: - socket_address: - address: atenet-router - port_value: %d -`, NodeID, r.cfg.XdsPort) - - cm := &corev1.ConfigMap{ - ObjectMeta: metav1.ObjectMeta{ - Name: EnvoyConfigMapName, - Namespace: r.cfg.Namespace, - }, - Data: map[string]string{ - "envoy.yaml": envoyYaml, - }, - } - - var existing corev1.ConfigMap - err := r.k8sClient.Get(ctx, client.ObjectKey{Namespace: r.cfg.Namespace, Name: EnvoyConfigMapName}, &existing) - if err != nil { - if k8serrors.IsNotFound(err) { - slog.InfoContext(ctx, "Creating Envoy bootstrap ConfigMap", - slog.String("namespace", r.cfg.Namespace), - slog.String("name", EnvoyConfigMapName)) - return r.k8sClient.Create(ctx, cm) - } - return err - } - - existing.Data = cm.Data - return r.k8sClient.Update(ctx, &existing) -} - -func (r *envoyrunner) reconcileEnvoyDeployment(ctx context.Context) error { - replicas := int32(1) - - dep := &appsv1.Deployment{ - ObjectMeta: metav1.ObjectMeta{ - Name: EnvoyDeploymentName, - Namespace: r.cfg.Namespace, - Labels: map[string]string{ - "app": "atenet-router-envoy", - }, - }, - Spec: appsv1.DeploymentSpec{ - Replicas: &replicas, - Selector: &metav1.LabelSelector{ - MatchLabels: map[string]string{ - "app": "atenet-router-envoy", - }, - }, - Template: corev1.PodTemplateSpec{ - ObjectMeta: metav1.ObjectMeta{ - Labels: map[string]string{ - "app": "atenet-router-envoy", - }, - }, - Spec: corev1.PodSpec{ - Containers: []corev1.Container{ - { - Name: "envoy", - Image: r.cfg.EnvoyImage, - Command: []string{ - "/usr/local/bin/envoy", - "-c", - "/etc/envoy/envoy.yaml", - "--component-log-level", - "upstream:debug,router:debug,ext_proc:debug", - }, - Ports: []corev1.ContainerPort{ - { - Name: "http", - ContainerPort: int32(r.cfg.HttpPort), - }, - { - Name: "admin", - ContainerPort: 9901, - }, - }, - VolumeMounts: []corev1.VolumeMount{ - { - Name: "envoy-config", - MountPath: "/etc/envoy", - }, - }, - }, - }, - Volumes: []corev1.Volume{ - { - Name: "envoy-config", - VolumeSource: corev1.VolumeSource{ - ConfigMap: &corev1.ConfigMapVolumeSource{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: EnvoyConfigMapName, - }, - }, - }, - }, - }, - }, - }, - }, - } - - var existing appsv1.Deployment - err := r.k8sClient.Get(ctx, client.ObjectKey{Namespace: r.cfg.Namespace, Name: EnvoyDeploymentName}, &existing) - if err != nil { - if k8serrors.IsNotFound(err) { - slog.InfoContext(ctx, "Creating managed Envoy router Deployment", slog.String("namespace", r.cfg.Namespace)) - return r.k8sClient.Create(ctx, dep) - } - return err - } - - existing.Spec = dep.Spec - return r.k8sClient.Update(ctx, &existing) -} - -func (r *envoyrunner) reconcileEnvoyService(ctx context.Context) error { - svc := &corev1.Service{ - ObjectMeta: metav1.ObjectMeta{ - Name: EnvoyServiceName, - Namespace: r.cfg.Namespace, - }, - Spec: corev1.ServiceSpec{ - Type: corev1.ServiceTypeClusterIP, - Selector: map[string]string{ - "app": "atenet-router-envoy", - }, - Ports: []corev1.ServicePort{ - { - Name: "http", - Port: int32(r.cfg.HttpPort), - TargetPort: intstr.FromString("http"), - }, - }, - }, - } - - var existing corev1.Service - err := r.k8sClient.Get(ctx, client.ObjectKey{Namespace: r.cfg.Namespace, Name: EnvoyServiceName}, &existing) - if err != nil { - if k8serrors.IsNotFound(err) { - slog.InfoContext(ctx, "Creating managed Envoy router ClusterIP service", slog.String("namespace", r.cfg.Namespace)) - return r.k8sClient.Create(ctx, svc) - } - return err - } - - existing.Spec.Ports = svc.Spec.Ports - existing.Spec.Selector = svc.Spec.Selector - return r.k8sClient.Update(ctx, &existing) -} diff --git a/cmd/atenet/internal/router/errors.go b/cmd/atenet/internal/router/errors.go deleted file mode 100644 index 25279661e..000000000 --- a/cmd/atenet/internal/router/errors.go +++ /dev/null @@ -1,92 +0,0 @@ -// 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 router - -import ( - "fmt" - - envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" -) - -// newReqError builds a reqError whose body is the formatted message and no -// wrapped cause. Set the cause field directly when one is available. -func newReqError(code envoy_type.StatusCode, format string, args ...any) error { - return &reqError{ - msg: fmt.Sprintf(format, args...), - statusCode: int(code), - } -} - -// actorNotFoundErr returns a 404 reqError identifying the missing actor. -func actorNotFoundErr(actorID string) error { - return newReqError(envoy_type.StatusCode_NotFound, "actor %q not found", actorID) -} - -// invalidHostErr returns a 404 reqError explaining why the request host was -// rejected. The cause is preserved for log inspection via Unwrap. -func invalidHostErr(host string, cause error) error { - return &reqError{ - msg: fmt.Sprintf("invalid host %q: %v", host, cause), - cause: cause, - statusCode: int(envoy_type.StatusCode_NotFound), - } -} - -// mapResumeError translates an ActorResumer error into a client-facing -// reqError. It maps gRPC status codes to appropriate HTTP status codes and -// short, human-readable bodies. The original error is preserved via Unwrap -// so callers can still inspect it via errors.Is / errors.As when logging. -// -// Unrecognized errors collapse to 500 with a generic body to avoid leaking -// server-side detail (stack traces, internal IDs) to clients. -func mapResumeError(actorID string, err error) error { - if err == nil { - return nil - } - - re := &reqError{cause: err} - switch status.Code(err) { - case codes.NotFound: - re.statusCode = int(envoy_type.StatusCode_NotFound) - re.msg = fmt.Sprintf("actor %q not found", actorID) - case codes.FailedPrecondition: - // Preserve the gRPC description for FailedPrecondition only: it carries - // actionable client-facing context (e.g. "no free workers available") - // and is not security-sensitive. - re.statusCode = int(envoy_type.StatusCode_ServiceUnavailable) - re.msg = fmt.Sprintf("actor %q unavailable: %s", actorID, status.Convert(err).Message()) - case codes.Unavailable: - re.statusCode = int(envoy_type.StatusCode_ServiceUnavailable) - re.msg = fmt.Sprintf("actor %q unavailable", actorID) - case codes.DeadlineExceeded: - re.statusCode = int(envoy_type.StatusCode_GatewayTimeout) - re.msg = fmt.Sprintf("actor %q request timed out", actorID) - case codes.PermissionDenied: - re.statusCode = int(envoy_type.StatusCode_Forbidden) - re.msg = fmt.Sprintf("actor %q access denied", actorID) - case codes.Unauthenticated: - re.statusCode = int(envoy_type.StatusCode_Unauthorized) - re.msg = fmt.Sprintf("actor %q authentication required", actorID) - case codes.ResourceExhausted: - re.statusCode = int(envoy_type.StatusCode_TooManyRequests) - re.msg = fmt.Sprintf("actor %q rate limited", actorID) - default: - re.statusCode = int(envoy_type.StatusCode_InternalServerError) - re.msg = fmt.Sprintf("error resuming actor %q", actorID) - } - return re -} diff --git a/cmd/atenet/internal/router/errors_test.go b/cmd/atenet/internal/router/errors_test.go deleted file mode 100644 index 9b4c2f146..000000000 --- a/cmd/atenet/internal/router/errors_test.go +++ /dev/null @@ -1,194 +0,0 @@ -// 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 router - -import ( - "errors" - "testing" - - envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" -) - -func TestNewReqError(t *testing.T) { - t.Parallel() - - err := newReqError(envoy_type.StatusCode_BadRequest, "actor %q is %s", "abc", "bad") - if err == nil { - t.Fatal("newReqError returned nil") - } - var reqErr *reqError - if !errors.As(err, &reqErr) { - t.Fatalf("errors.As(*reqError) = false, want true; err type = %T", err) - } - if reqErr.statusCode != int(envoy_type.StatusCode_BadRequest) { - t.Errorf("statusCode = %d, want %d", reqErr.statusCode, envoy_type.StatusCode_BadRequest) - } - if got, want := err.Error(), `actor "abc" is bad`; got != want { - t.Errorf("Error() = %q, want %q", got, want) - } -} - -func TestActorNotFoundErr(t *testing.T) { - t.Parallel() - - err := actorNotFoundErr("ctr6") - var reqErr *reqError - if !errors.As(err, &reqErr) { - t.Fatalf("errors.As(*reqError) = false, want true; err type = %T", err) - } - if reqErr.statusCode != int(envoy_type.StatusCode_NotFound) { - t.Errorf("statusCode = %d, want %d", reqErr.statusCode, envoy_type.StatusCode_NotFound) - } - if got, want := err.Error(), `actor "ctr6" not found`; got != want { - t.Errorf("Error() = %q, want %q", got, want) - } -} - -func TestInvalidHostErr(t *testing.T) { - t.Parallel() - - cause := errors.New("missing suffix") - err := invalidHostErr("foo.example.com", cause) - - var reqErr *reqError - if !errors.As(err, &reqErr) { - t.Fatalf("errors.As(*reqError) = false, want true; err type = %T", err) - } - if reqErr.statusCode != int(envoy_type.StatusCode_NotFound) { - t.Errorf("statusCode = %d, want %d", reqErr.statusCode, envoy_type.StatusCode_NotFound) - } - if got, want := err.Error(), `invalid host "foo.example.com": missing suffix`; got != want { - t.Errorf("Error() = %q, want %q", got, want) - } - if !errors.Is(err, cause) { - t.Errorf("errors.Is(err, cause) = false, want true (cause should be wrapped for logging)") - } -} - -func TestMapResumeError(t *testing.T) { - t.Parallel() - - const actorID = "ctr6" - - tests := []struct { - name string - err error - wantCode envoy_type.StatusCode - wantBody string - }{ - { - name: "NotFound maps to 404", - err: status.Error(codes.NotFound, "actor not found"), - wantCode: envoy_type.StatusCode_NotFound, - wantBody: `actor "ctr6" not found`, - }, - { - name: "FailedPrecondition maps to 503 and preserves desc", - err: status.Error(codes.FailedPrecondition, "no free workers available"), - wantCode: envoy_type.StatusCode_ServiceUnavailable, - wantBody: `actor "ctr6" unavailable: no free workers available`, - }, - { - name: "Unavailable maps to 503", - err: status.Error(codes.Unavailable, "control-plane down"), - wantCode: envoy_type.StatusCode_ServiceUnavailable, - wantBody: `actor "ctr6" unavailable`, - }, - { - name: "DeadlineExceeded maps to 504", - err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), - wantCode: envoy_type.StatusCode_GatewayTimeout, - wantBody: `actor "ctr6" request timed out`, - }, - { - name: "PermissionDenied maps to 403", - err: status.Error(codes.PermissionDenied, "denied"), - wantCode: envoy_type.StatusCode_Forbidden, - wantBody: `actor "ctr6" access denied`, - }, - { - name: "Unauthenticated maps to 401", - err: status.Error(codes.Unauthenticated, "no creds"), - wantCode: envoy_type.StatusCode_Unauthorized, - wantBody: `actor "ctr6" authentication required`, - }, - { - name: "ResourceExhausted maps to 429", - err: status.Error(codes.ResourceExhausted, "quota"), - wantCode: envoy_type.StatusCode_TooManyRequests, - wantBody: `actor "ctr6" rate limited`, - }, - { - name: "unknown gRPC code maps to 500 without leaking desc", - err: status.Error(codes.Internal, "stack trace: foo bar"), - wantCode: envoy_type.StatusCode_InternalServerError, - wantBody: `error resuming actor "ctr6"`, - }, - { - name: "non-gRPC error maps to 500 without leaking message", - err: errors.New("raw error with secret"), - wantCode: envoy_type.StatusCode_InternalServerError, - wantBody: `error resuming actor "ctr6"`, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - - got := mapResumeError(actorID, tc.err) - if got == nil { - t.Fatal("mapResumeError returned nil") - } - var reqErr *reqError - if !errors.As(got, &reqErr) { - t.Fatalf("errors.As(*reqError) = false, want true; err type = %T", got) - } - if reqErr.statusCode != int(tc.wantCode) { - t.Errorf("statusCode = %d, want %d", reqErr.statusCode, tc.wantCode) - } - if got.Error() != tc.wantBody { - t.Errorf("Error() = %q, want %q", got.Error(), tc.wantBody) - } - if !errors.Is(got, tc.err) { - t.Errorf("errors.Is(result, original) = false, want true (original must be preserved for logs)") - } - }) - } -} - -func TestMapResumeError_NilError(t *testing.T) { - t.Parallel() - - // Guard against accidental nil-error calls. Returning nil keeps the - // happy path explicit at callsites instead of constructing a bogus 500. - if got := mapResumeError("ctr6", nil); got != nil { - t.Errorf("mapResumeError(_, nil) = %v, want nil", got) - } -} - -// Ensures mapResumeError result satisfies the reqError contract so the -// existing handleRequestHeaders branch (errors.As(err, &reqErr)) keeps working. -func TestMapResumeError_IsReqError(t *testing.T) { - t.Parallel() - - err := mapResumeError("x", status.Error(codes.NotFound, "x")) - var reqErr *reqError - if !errors.As(err, &reqErr) { - t.Fatalf("errors.As(*reqError) = false, want true; err type = %T", err) - } -} diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index ac5b5d5b8..396f97b24 100644 --- a/cmd/atenet/internal/router/extproc.go +++ b/cmd/atenet/internal/router/extproc.go @@ -31,26 +31,24 @@ import ( "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/propagation" "google.golang.org/grpc" - - "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) -// ExtProcServer implements the Envoy external processing gRPC server -// to dynamically manage actor activations based on request traffic. +// ExtProcServer implements the Envoy external processing gRPC server. +// It is a thin adapter: it translates Envoy ExtProc protocol to/from the +// gateway-neutral RouteResolver and records metrics. type ExtProcServer struct { port int - apiClient ateapipb.ControlClient + resolver RouteResolver recorder *QueryRecorder - resumer *ActorResumer routeDuration metric.Float64Histogram } -func NewExtProcServer(port int, apiClient ateapipb.ControlClient, routeDuration metric.Float64Histogram) *ExtProcServer { +// NewExtProcServer creates an ExtProcServer backed by the given resolver. +func NewExtProcServer(port int, resolver RouteResolver, routeDuration metric.Float64Histogram) *ExtProcServer { return &ExtProcServer{ port: port, - apiClient: apiClient, + resolver: resolver, recorder: NewQueryRecorder(100), - resumer: NewActorResumer(apiClient), routeDuration: routeDuration, } } @@ -92,7 +90,7 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer switch reqType := req.Request.(type) { case *extprocv3.ProcessingRequest_RequestHeaders: start := time.Now() - hResponse, rqm, target, tmplNs, tmplName, err := s.handleRequestHeaders(stream.Context(), reqType.RequestHeaders) + hResponse, rqm, res, err := s.handleRequestHeaders(stream.Context(), reqType.RequestHeaders) elapsed := time.Since(start) if err != nil { slog.ErrorContext(stream.Context(), "Error during ext_proc RequestHeaders processing", slog.String("err", err.Error())) @@ -102,10 +100,18 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer } else { resp = immediateResponse(envoy_type.StatusCode_InternalServerError, err.Error()) } + var tmplNs, tmplName string + if res != nil && res.Denial != nil { + // outcome already captured; labels stay empty for denials from parse failures + _ = res + } s.recordRouteDuration(stream.Context(), elapsed, tmplNs, tmplName, classifyOutcome(err)) s.recorder.AddRouterRequest(start, elapsed, "Error", "-", rqm) } else { resp.Response = &extprocv3.ProcessingResponse_RequestHeaders{RequestHeaders: hResponse} + tmplNs := res.Success.TemplateRef.Namespace + tmplName := res.Success.TemplateRef.Name + target := net.JoinHostPort(res.Success.Backend.IP, fmt.Sprintf("%d", res.Success.Backend.Port)) s.recordRouteDuration(stream.Context(), elapsed, tmplNs, tmplName, "ok") s.recorder.AddRouterRequest(start, elapsed, "Route ok", target, rqm) } @@ -127,10 +133,16 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer } } +// handleRequestHeaders is the ExtProc adapter for a single request. It +// extracts trace context, builds a RouteRequest, calls the resolver, and +// maps the RouteResolution to an extproc HeadersResponse. +// +// On denial, it returns a non-nil *reqError so Process() can build an +// ImmediateResponse. The RouteResolution is returned alongside for label access. func (s *ExtProcServer) handleRequestHeaders( ctx context.Context, reqHeaders *extprocv3.HttpHeaders, -) (*extprocv3.HeadersResponse, *requestMetadata, string, string, string, error) { +) (*extprocv3.HeadersResponse, *requestMetadata, *RouteResolution, error) { metadata := newRequestMetadata(reqHeaders.Headers.GetHeaders()) slog.InfoContext(ctx, "Request", slog.String("host", metadata.host)) @@ -142,41 +154,28 @@ func (s *ExtProcServer) handleRequestHeaders( ctx, span := otel.Tracer(routerServiceName).Start(ctx, "ExtProc.RequestHeaders") defer span.End() - atespace, actorID, err := parseActorRef(metadata.host) - if err != nil { - // Host is invalid, respond with 404. - return nil, metadata, "", "", "", invalidHostErr(metadata.host, err) + req := RouteRequest{ + Authority: metadata.host, + Path: metadata.path, + Headers: metadata.headers, + Adapter: "extproc", } - slog.InfoContext(ctx, "ResumeActor", slog.String("atespace", atespace), slog.String("actorID", actorID)) - actor, err := s.resumer.ResumeActor(ctx, atespace, actorID) - if err != nil { - return nil, metadata, "", "", "", mapResumeError(actorID, err) - } - - // Actor template identity, used as low-cardinality route-latency metric - // attributes (see recordRouteDuration). - tmplNs := actor.GetActorTemplateNamespace() - tmplName := actor.GetActorTemplateName() - - workerIP := actor.GetAteomPodIp() - slog.InfoContext(ctx, "ResumeActor result", - slog.String("atespace", atespace), - slog.String("actorID", actorID), - slog.String("status", actor.GetStatus().String()), - slog.String("workerIP", workerIP)) - - if ip := net.ParseIP(workerIP); ip == nil { - return nil, metadata, "", tmplNs, tmplName, newReqError(envoy_type.StatusCode_InternalServerError, - "actor %q routing failed", actorID) + res := s.resolver.ResolveRoute(ctx, req) + if res.Denial != nil { + d := res.Denial + re := &reqError{ + msg: d.Message, + cause: d.Cause, + statusCode: d.HTTPStatus, + } + return nil, metadata, &res, re } - // TODO(bowei) -- handle more than port 80 on the actor. - targetAddr := net.JoinHostPort(workerIP, "80") - - slog.InfoContext(ctx, "Route ok", slog.String("actorID", actorID), slog.String("targetAddr", targetAddr)) + success := res.Success + targetAddr := net.JoinHostPort(success.Backend.IP, fmt.Sprintf("%d", success.Backend.Port)) + slog.InfoContext(ctx, "Route ok", slog.String("actorID", success.ActorID), slog.String("targetAddr", targetAddr)) - // Route by rewriting the :authority header. mutation := &extprocv3.HeaderMutation{} addAuthorityMutation(targetAddr, mutation) @@ -184,7 +183,7 @@ func (s *ExtProcServer) handleRequestHeaders( Response: &extprocv3.CommonResponse{ HeaderMutation: mutation, }, - }, metadata, targetAddr, tmplNs, tmplName, nil + }, metadata, &res, nil } func (s *ExtProcServer) recordRouteDuration(ctx context.Context, d time.Duration, tmplNs, tmplName, outcome string) { diff --git a/cmd/atenet/internal/router/extproc_test.go b/cmd/atenet/internal/router/extproc_test.go index bfd988691..e8dbb588f 100644 --- a/cmd/atenet/internal/router/extproc_test.go +++ b/cmd/atenet/internal/router/extproc_test.go @@ -18,21 +18,31 @@ import ( "bytes" "context" "encoding/json" - "errors" "log/slog" + "net/http" "strings" "testing" "time" - "github.com/agent-substrate/substrate/pkg/proto/ateapipb" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" ) +// mockResolver is a test double for RouteResolver. +type mockResolver struct { + result RouteResolution +} + +func (m *mockResolver) ResolveRoute(_ context.Context, _ RouteRequest) RouteResolution { + return m.result +} + +// mockClient satisfies ateapipb.ControlClient for use in resolver construction +// tests that still exercise the full ActorRouteResolver path. type mockClient struct { ateapipb.ControlClient resumeFn func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) @@ -42,6 +52,23 @@ func (m *mockClient) ResumeActor(ctx context.Context, in *ateapipb.ResumeActorRe return m.resumeFn(ctx, in, opts...) } +func makeReqHeaders(authority, path string, extra ...[2]string) *extprocv3.HttpHeaders { + headers := []*corev3.HeaderValue{ + {Key: ":path", Value: path}, + {Key: ":authority", Value: authority}, + {Key: ":method", Value: "POST"}, + } + for _, kv := range extra { + headers = append(headers, &corev3.HeaderValue{Key: kv[0], Value: kv[1]}) + } + return &extprocv3.HttpHeaders{ + Headers: &corev3.HeaderMap{Headers: headers}, + } +} + +// TestHandleRequestHeadersDoesNotLogSensitiveData verifies that the ExtProc +// adapter layer does not leak secrets (auth tokens, cookies, query params) to +// logs or the query recorder, while still logging routing context (actor ID). func TestHandleRequestHeadersDoesNotLogSensitiveData(t *testing.T) { const testUUID = "123e4567-e89b-12d3-a456-426614174000" const secret = "do-not-log-me" @@ -51,11 +78,12 @@ func TestHandleRequestHeadersDoesNotLogSensitiveData(t *testing.T) { slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, nil))) t.Cleanup(func() { slog.SetDefault(prev) }) - s := NewExtProcServer(50051, &mockClient{ - resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { - return &ateapipb.ResumeActorResponse{Actor: &ateapipb.Actor{AteomPodIp: "10.0.0.52"}}, nil - }, - }, nil) + resolver := &mockResolver{result: RouteResolution{Success: &RouteSuccess{ + ActorID: testUUID, + Backend: Backend{IP: "10.0.0.52", Port: 80}, + TemplateRef: ActorTemplateRef{}, + }}} + s := NewExtProcServer(50051, resolver, nil) reqHeaders := &extprocv3.HttpHeaders{ Headers: &corev3.HeaderMap{ @@ -69,7 +97,7 @@ func TestHandleRequestHeadersDoesNotLogSensitiveData(t *testing.T) { }, } - _, metadata, target, _, _, err := s.handleRequestHeaders(context.Background(), reqHeaders) + _, metadata, res, err := s.handleRequestHeaders(context.Background(), reqHeaders) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -78,10 +106,8 @@ func TestHandleRequestHeadersDoesNotLogSensitiveData(t *testing.T) { if strings.Contains(out, secret) { t.Errorf("router log leaked sensitive value: %s", out) } - if !strings.Contains(out, testUUID) { - t.Errorf("router log missing actor/host routing context: %s", out) - } + target := res.Success.Backend.IP s.recorder.AddRouterRequest(time.Now(), time.Millisecond, "Route ok", target, metadata) for _, q := range s.recorder.Get() { if blob, _ := json.Marshal(q); strings.Contains(string(blob), secret) { @@ -90,165 +116,115 @@ func TestHandleRequestHeadersDoesNotLogSensitiveData(t *testing.T) { } } -func TestExtProcHeadersEvaluation(t *testing.T) { - const testUUID = "123e4567-e89b-12d3-a456-426614174000" +// TestExtProcAdapterMapping verifies that the ExtProc adapter correctly maps +// RouteResolver outcomes to the extproc protocol (header mutation on success, +// immediate response on denial). +func TestExtProcAdapterMapping(t *testing.T) { + const workerIP = "10.0.0.52" + const actorID = "123e4567-e89b-12d3-a456-426614174000" tests := []struct { name string - authority string - resumeResp *ateapipb.ResumeActorResponse - resumeErr error - expectErr bool - expectedErrStr string - expectedStatus envoy_type.StatusCode - expectedTarget string + resolution RouteResolution + wantErr bool + wantStatus envoy_type.StatusCode + wantMsgContain string + wantTarget string }{ { - name: "invalid host returns 404 identifying the host", - authority: "invalid-host.com", - expectErr: true, - expectedErrStr: `invalid host "invalid-host.com": invalid actor DNS name: must end with actors.resources.substrate.ate.dev, got "invalid-host.com"`, - expectedStatus: envoy_type.StatusCode_NotFound, - }, - { - name: "non-gRPC resume error collapses to 500 without leaking detail", - authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", - resumeErr: errors.New("resume failed with sensitive detail"), - expectErr: true, - expectedErrStr: `error resuming actor "123e4567-e89b-12d3-a456-426614174000"`, - expectedStatus: envoy_type.StatusCode_InternalServerError, - }, - { - name: "FailedPrecondition maps to 503 with preserved desc", - authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", - resumeErr: status.Error(codes.FailedPrecondition, "no free workers available"), - expectErr: true, - expectedErrStr: `actor "123e4567-e89b-12d3-a456-426614174000" unavailable: no free workers available`, - expectedStatus: envoy_type.StatusCode_ServiceUnavailable, - }, - { - name: "NotFound maps to 404", - authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", - resumeErr: status.Error(codes.NotFound, "actor missing"), - expectErr: true, - expectedErrStr: `actor "123e4567-e89b-12d3-a456-426614174000" not found`, - expectedStatus: envoy_type.StatusCode_NotFound, + name: "success maps to :authority header mutation", + resolution: RouteResolution{Success: &RouteSuccess{ + ActorID: actorID, + Backend: Backend{IP: workerIP, Port: 80}, + TemplateRef: ActorTemplateRef{Namespace: "ns", Name: "tmpl"}, + }}, + wantTarget: workerIP + ":80", }, { - name: "Unavailable maps to 503", - authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", - resumeErr: status.Error(codes.Unavailable, "control-plane down"), - expectErr: true, - expectedErrStr: `actor "123e4567-e89b-12d3-a456-426614174000" unavailable`, - expectedStatus: envoy_type.StatusCode_ServiceUnavailable, + name: "404 denial becomes reqError with correct status", + resolution: RouteResolution{Denial: &RouteDenial{ + HTTPStatus: http.StatusNotFound, + Message: "actor not found", + OutcomeCode: "not_found", + }}, + wantErr: true, + wantStatus: envoy_type.StatusCode_NotFound, + wantMsgContain: "actor not found", }, { - name: "DeadlineExceeded maps to 504", - authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", - resumeErr: status.Error(codes.DeadlineExceeded, "deadline"), - expectErr: true, - expectedErrStr: `actor "123e4567-e89b-12d3-a456-426614174000" request timed out`, - expectedStatus: envoy_type.StatusCode_GatewayTimeout, + name: "503 denial becomes reqError with correct status", + resolution: RouteResolution{Denial: &RouteDenial{ + HTTPStatus: http.StatusServiceUnavailable, + Message: "actor unavailable: no free workers", + OutcomeCode: "error", + }}, + wantErr: true, + wantStatus: envoy_type.StatusCode_ServiceUnavailable, + wantMsgContain: "no free workers", }, { - name: "Bad Actor IP from resume returns 500 without leaking IP", - authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", - resumeResp: &ateapipb.ResumeActorResponse{ - Actor: &ateapipb.Actor{ - AteomPodIp: "invalid-ip", - }, - }, - expectErr: true, - expectedErrStr: `actor "123e4567-e89b-12d3-a456-426614174000" routing failed`, - expectedStatus: envoy_type.StatusCode_InternalServerError, - }, - { - name: "Successful resume", - authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", - resumeResp: &ateapipb.ResumeActorResponse{ - Actor: &ateapipb.Actor{ - AteomPodIp: "10.0.0.52", - }, - }, - expectErr: false, - expectedTarget: "10.0.0.52:80", + name: "500 denial becomes reqError with correct status", + resolution: RouteResolution{Denial: &RouteDenial{ + HTTPStatus: http.StatusInternalServerError, + Message: "actor routing failed", + OutcomeCode: "error", + }}, + wantErr: true, + wantStatus: envoy_type.StatusCode_InternalServerError, + wantMsgContain: "routing failed", }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - clientMock := &mockClient{ - resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { - if in.GetActorRef().GetName() != testUUID { - t.Errorf("unexpected identifier parsed in test context: %s", in.GetActorRef().GetName()) - } - if tc.resumeErr != nil { - return nil, tc.resumeErr - } - return tc.resumeResp, nil - }, - } - - s := NewExtProcServer(50051, clientMock, nil) + s := NewExtProcServer(50051, &mockResolver{result: tc.resolution}, nil) + reqHeaders := makeReqHeaders(actorID+".team-a.actors.resources.substrate.ate.dev", "/v1/invoke") - reqHeaders := &extprocv3.HttpHeaders{ - Headers: &corev3.HeaderMap{ - Headers: []*corev3.HeaderValue{ - {Key: ":path", Value: "/v1/actors/invoke"}, - {Key: ":authority", Value: tc.authority}, - {Key: ":method", Value: "POST"}, - }, - }, - } - - res, metadata, target, _, _, err := s.handleRequestHeaders(context.Background(), reqHeaders) - if tc.expectErr { + hResp, _, res, err := s.handleRequestHeaders(context.Background(), reqHeaders) + if tc.wantErr { if err == nil { - t.Fatalf("expected error but got nil") - } - if tc.expectedErrStr != "" && err.Error() != tc.expectedErrStr { - t.Errorf("client body mismatch:\n got: %q\n want: %q", err.Error(), tc.expectedErrStr) + t.Fatalf("expected error, got nil") } - var reqErr *reqError - if !errors.As(err, &reqErr) { - t.Fatalf("expected *reqError, got %T (%v)", err, err) + var re *reqError + if !isReqError(err, &re) { + t.Fatalf("expected *reqError, got %T: %v", err, err) } - if got, want := reqErr.statusCode, int(tc.expectedStatus); got != want { - t.Errorf("HTTP status code = %d, want %d", got, want) + if re.statusCode != int(tc.wantStatus) { + t.Errorf("statusCode = %d, want %d", re.statusCode, tc.wantStatus) } - if tc.resumeErr != nil && !errors.Is(err, tc.resumeErr) { - t.Errorf("original resume error must be preserved in chain for logs; errors.Is(err, resumeErr) = false") + if !strings.Contains(err.Error(), tc.wantMsgContain) { + t.Errorf("error message %q does not contain %q", err.Error(), tc.wantMsgContain) } + _ = res return } - if err != nil { - t.Fatalf("ext_proc processing error: %v", err) - } - if target != tc.expectedTarget { - t.Errorf("expected target %q, got %q", tc.expectedTarget, target) + t.Fatalf("unexpected error: %v", err) } - - mutation := res.Response.GetHeaderMutation() + mutation := hResp.Response.GetHeaderMutation() if len(mutation.GetSetHeaders()) != 1 { - t.Fatalf("expected exactly one Header option set, found: %v", mutation.GetSetHeaders()) + t.Fatalf("expected 1 header mutation, got %d", len(mutation.GetSetHeaders())) } - - headerOption := mutation.GetSetHeaders()[0] - if strings.ToLower(headerOption.Header.Key) != ":authority" { - t.Errorf("invalid resulting dynamic parameter key: %s", headerOption.Header.Key) + hv := mutation.GetSetHeaders()[0].Header + if strings.ToLower(hv.Key) != ":authority" { + t.Errorf("mutation key = %q, want :authority", hv.Key) } - - if string(headerOption.Header.RawValue) != tc.expectedTarget { - t.Errorf("invalid destination mapping found: %s, expected: %s", headerOption.Header.RawValue, tc.expectedTarget) - } - - // Confirm that query logs recorded metric trace details - s.recorder.AddRouterRequest(time.Now(), 10*time.Millisecond, "Route ok", tc.expectedTarget, metadata) - queries := s.recorder.Get() - if len(queries) != 1 { - t.Errorf("expected query trace entries, got: %v", queries) + if string(hv.RawValue) != tc.wantTarget { + t.Errorf("mutation value = %q, want %q", hv.RawValue, tc.wantTarget) } }) } } + +func isReqError(err error, out **reqError) bool { + if err == nil { + return false + } + if re, ok := err.(*reqError); ok { + if out != nil { + *out = re + } + return true + } + return false +} diff --git a/cmd/atenet/internal/router/health.go b/cmd/atenet/internal/router/health.go index d9e1fddab..172c4715d 100644 --- a/cmd/atenet/internal/router/health.go +++ b/cmd/atenet/internal/router/health.go @@ -17,15 +17,11 @@ package router import ( "context" "fmt" - "io" "log/slog" - "net/http" - "strings" "sync" "time" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" - "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "k8s.io/client-go/kubernetes" ) @@ -39,7 +35,6 @@ type ComponentHealth struct { } type RouterHealthReport struct { - Envoy ComponentHealth `json:"envoy"` K8sAPI ComponentHealth `json:"k8s_api"` AteAPI ComponentHealth `json:"ate_api"` } @@ -51,11 +46,10 @@ type routerHealth struct { report RouterHealthReport - interval time.Duration - clientset kubernetes.Interface - apiClient ateapipb.ControlClient - cfg RouterConfig - envoyClient *http.Client + interval time.Duration + clientset kubernetes.Interface + apiClient ateapipb.ControlClient + cfg RouterConfig } func newRouterHealth(interval time.Duration, clientset kubernetes.Interface, apiClient ateapipb.ControlClient, cfg RouterConfig) *routerHealth { @@ -63,11 +57,10 @@ func newRouterHealth(interval time.Duration, clientset kubernetes.Interface, api interval = time.Second } return &routerHealth{ - interval: interval, - clientset: clientset, - apiClient: apiClient, - cfg: cfg, - envoyClient: &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}, + interval: interval, + clientset: clientset, + apiClient: apiClient, + cfg: cfg, } } @@ -94,24 +87,7 @@ func (rh *routerHealth) check(ctx context.Context) { slog.InfoContext(ctx, "Checking health") - // 1. Check Envoy - { - healthy, msg := rh.checkEnvoy(ctx) - if healthy { - rh.report.Envoy.Healthy = true - rh.report.Envoy.Message = msg - rh.report.Envoy.LastSuccess = time.Now() - rh.report.Envoy.SuccessCount++ - } else { - rh.report.Envoy.Healthy = false - rh.report.Envoy.Message = msg - rh.report.Envoy.LastFailure = time.Now() - rh.report.Envoy.FailureCount++ - slog.ErrorContext(ctx, "Envoy health check failed", slog.String("msg", msg)) - } - } - - // 2. Check Kubernetes API + // 1. Check Kubernetes API { healthy, msg := rh.checkK8s() if healthy { @@ -128,7 +104,7 @@ func (rh *routerHealth) check(ctx context.Context) { } } - // 3. Check ATE API gRPC + // 2. Check ATE API gRPC { healthy, msg := rh.checkAteAPI(ctx) if healthy { @@ -146,38 +122,6 @@ func (rh *routerHealth) check(ctx context.Context) { } } -func (rh *routerHealth) checkEnvoy(ctx context.Context) (bool, string) { - timeoutCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond) - defer cancel() - - req, err := http.NewRequestWithContext(timeoutCtx, "GET", "http://127.0.0.1:9901/ready", nil) - if err != nil { - return false, err.Error() - } - - resp, err := rh.envoyClient.Do(req) - if err != nil { - return false, err.Error() - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return false, fmt.Sprintf("unexpected status code %d", resp.StatusCode) - } - - bodyBytes, err := io.ReadAll(resp.Body) - if err != nil { - return false, err.Error() - } - - bodyStr := strings.TrimSpace(string(bodyBytes)) - if bodyStr != "LIVE" { - return false, fmt.Sprintf("expected LIVE but got %q", bodyStr) - } - - return true, "LIVE" -} - func (rh *routerHealth) checkK8s() (bool, string) { if rh.clientset == nil { return true, "Skipped (standalone/file store)" diff --git a/cmd/atenet/internal/router/route.go b/cmd/atenet/internal/router/route.go new file mode 100644 index 000000000..72ce6643e --- /dev/null +++ b/cmd/atenet/internal/router/route.go @@ -0,0 +1,98 @@ +// 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 router + +import "context" + +// RouteRequest is a gateway-neutral description of an inbound request that +// needs actor-aware routing. Adapters (ExtProc, future xDS, gRPC) fill this +// from their own protocol types and hand it to a RouteResolver. +type RouteRequest struct { + // Authority is the HTTP :authority / Host header value. + // The actor ID is parsed from the subdomain suffix. + Authority string + + // Path is the HTTP request path (passed through; not used for routing in M1). + Path string + + // Headers carries trace-propagation headers (e.g. traceparent). + // Not used for actor identity. + Headers map[string]string + + // SourceActorID is the verified identity of the calling actor. + // Reserved for M4 (pre-resume authorization); ignored in M1. + SourceActorID string + + // Adapter is an opaque label identifying which adapter is calling the + // resolver (e.g. "extproc"). Used for metrics and logging. + Adapter string +} + +// RouteSuccess is the happy-path result of a route resolution. +type RouteSuccess struct { + // ActorID is the resolved actor identifier. + ActorID string + + // Backend is the IP/port of the worker pod to forward traffic to. + Backend Backend + + // TemplateRef identifies the ActorTemplate that owns this actor, + // used as low-cardinality metric attributes. + TemplateRef ActorTemplateRef +} + +// Backend is the network endpoint of a live actor worker pod. +type Backend struct { + IP string + Port int +} + +// ActorTemplateRef identifies an ActorTemplate by namespace and name. +type ActorTemplateRef struct { + Namespace string + Name string +} + +// RouteDenial is a gateway-neutral error result. HTTP status and message are +// client-safe; Cause is preserved for log inspection only. +type RouteDenial struct { + // HTTPStatus is the HTTP status code to return to the client (e.g. 404, 503). + HTTPStatus int + + // Message is a client-safe human-readable explanation. + Message string + + // Cause is the underlying error preserved for logs. Never sent to clients. + Cause error + + // OutcomeCode is a low-cardinality label for metrics (ok/not_found/cancelled/error). + OutcomeCode string +} + +func (d *RouteDenial) Error() string { return d.Message } +func (d *RouteDenial) Unwrap() error { return d.Cause } + +// RouteResolution holds either a successful resolution or a denial. +// Exactly one of Success or Denial is non-nil. +type RouteResolution struct { + Success *RouteSuccess + Denial *RouteDenial +} + +// RouteResolver resolves an inbound RouteRequest to a backend endpoint or denial. +// Implementations must be safe for concurrent use. +type RouteResolver interface { + ResolveRoute(ctx context.Context, req RouteRequest) RouteResolution +} diff --git a/cmd/atenet/internal/router/route_errors.go b/cmd/atenet/internal/router/route_errors.go new file mode 100644 index 000000000..6de22d01a --- /dev/null +++ b/cmd/atenet/internal/router/route_errors.go @@ -0,0 +1,84 @@ +// 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 router + +import ( + "fmt" + "net/http" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// invalidHostDenial returns a 404 RouteDenial for a host that cannot be parsed +// as an actor address. The parse error is preserved for log inspection. +func invalidHostDenial(host string, cause error) *RouteDenial { + return &RouteDenial{ + HTTPStatus: http.StatusNotFound, + Message: fmt.Sprintf("invalid host %q: %v", host, cause), + Cause: cause, + OutcomeCode: "not_found", + } +} + +// mapResumeDenial translates an ActorResumer error into a gateway-neutral +// RouteDenial. gRPC status codes are mapped to HTTP equivalents; the original +// error is preserved via Cause so callers can inspect the full chain. +// +// Unrecognised codes collapse to 500 to avoid leaking server-side detail. +func mapResumeDenial(actorID string, err error) *RouteDenial { + if err == nil { + return nil + } + d := &RouteDenial{Cause: err} + switch status.Code(err) { + case codes.NotFound: + d.HTTPStatus = http.StatusNotFound + d.Message = fmt.Sprintf("actor %q not found", actorID) + d.OutcomeCode = "not_found" + case codes.FailedPrecondition: + // Preserve the gRPC description for FailedPrecondition only: it carries + // actionable client-facing context (e.g. "no free workers available") + // and is not security-sensitive. + d.HTTPStatus = http.StatusServiceUnavailable + d.Message = fmt.Sprintf("actor %q unavailable: %s", actorID, status.Convert(err).Message()) + d.OutcomeCode = "error" + case codes.Unavailable: + d.HTTPStatus = http.StatusServiceUnavailable + d.Message = fmt.Sprintf("actor %q unavailable", actorID) + d.OutcomeCode = "error" + case codes.DeadlineExceeded: + d.HTTPStatus = http.StatusGatewayTimeout + d.Message = fmt.Sprintf("actor %q request timed out", actorID) + d.OutcomeCode = "cancelled" + case codes.PermissionDenied: + d.HTTPStatus = http.StatusForbidden + d.Message = fmt.Sprintf("actor %q access denied", actorID) + d.OutcomeCode = "error" + case codes.Unauthenticated: + d.HTTPStatus = http.StatusUnauthorized + d.Message = fmt.Sprintf("actor %q authentication required", actorID) + d.OutcomeCode = "error" + case codes.ResourceExhausted: + d.HTTPStatus = http.StatusTooManyRequests + d.Message = fmt.Sprintf("actor %q rate limited", actorID) + d.OutcomeCode = "error" + default: + d.HTTPStatus = http.StatusInternalServerError + d.Message = fmt.Sprintf("error resuming actor %q", actorID) + d.OutcomeCode = "error" + } + return d +} diff --git a/cmd/atenet/internal/router/route_resolve.go b/cmd/atenet/internal/router/route_resolve.go new file mode 100644 index 000000000..f54fde395 --- /dev/null +++ b/cmd/atenet/internal/router/route_resolve.go @@ -0,0 +1,78 @@ +// 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 router + +import ( + "context" + "fmt" + "log/slog" + "net" + "net/http" +) + +// ActorRouteResolver implements RouteResolver using the actor control plane. +// It parses the actor ID from the request authority, resumes the actor via the +// control plane, and resolves the worker pod backend. +type ActorRouteResolver struct { + resumer *ActorResumer +} + +// NewActorRouteResolver constructs an ActorRouteResolver backed by the given resumer. +func NewActorRouteResolver(resumer *ActorResumer) *ActorRouteResolver { + return &ActorRouteResolver{resumer: resumer} +} + +// ResolveRoute implements RouteResolver. It is safe for concurrent use. +func (r *ActorRouteResolver) ResolveRoute(ctx context.Context, req RouteRequest) RouteResolution { + atespace, actorID, err := parseActorRef(req.Authority) + if err != nil { + return RouteResolution{Denial: invalidHostDenial(req.Authority, err)} + } + + slog.InfoContext(ctx, "ResumeActor", slog.String("atespace", atespace), slog.String("actorID", actorID)) + actor, err := r.resumer.ResumeActor(ctx, atespace, actorID) + if err != nil { + return RouteResolution{Denial: mapResumeDenial(actorID, err)} + } + + tmplNs := actor.GetActorTemplateNamespace() + tmplName := actor.GetActorTemplateName() + workerIP := actor.GetAteomPodIp() + + slog.InfoContext(ctx, "ResumeActor result", + slog.String("actorID", actorID), + slog.String("status", actor.GetStatus().String()), + slog.String("workerIP", workerIP)) + + if net.ParseIP(workerIP) == nil { + return RouteResolution{Denial: &RouteDenial{ + HTTPStatus: http.StatusInternalServerError, + Message: fmt.Sprintf("actor %q routing failed", actorID), + OutcomeCode: "error", + }} + } + + return RouteResolution{Success: &RouteSuccess{ + ActorID: actorID, + Backend: Backend{ + IP: workerIP, + Port: 80, // TODO(bowei): handle more than port 80 on the actor. + }, + TemplateRef: ActorTemplateRef{ + Namespace: tmplNs, + Name: tmplName, + }, + }} +} diff --git a/cmd/atenet/internal/router/route_resolve_test.go b/cmd/atenet/internal/router/route_resolve_test.go new file mode 100644 index 000000000..ece8128b0 --- /dev/null +++ b/cmd/atenet/internal/router/route_resolve_test.go @@ -0,0 +1,256 @@ +// 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 router + +import ( + "context" + "errors" + "net/http" + "testing" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const testActorUUID = "123e4567-e89b-12d3-a456-426614174000" + +func newTestResolver(fn func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error)) *ActorRouteResolver { + client := &mockClient{resumeFn: fn} + return NewActorRouteResolver(NewActorResumer(client)) +} + +func TestResolveRoute_InvalidHost(t *testing.T) { + t.Parallel() + r := newTestResolver(nil) + res := r.ResolveRoute(context.Background(), RouteRequest{Authority: "invalid-host.com"}) + if res.Denial == nil { + t.Fatal("expected Denial, got Success") + } + if res.Denial.HTTPStatus != http.StatusNotFound { + t.Errorf("HTTPStatus = %d, want 404", res.Denial.HTTPStatus) + } + want := `invalid host "invalid-host.com": invalid actor DNS name: must end with actors.resources.substrate.ate.dev, got "invalid-host.com"` + if res.Denial.Message != want { + t.Errorf("Message = %q, want %q", res.Denial.Message, want) + } + if res.Denial.OutcomeCode != "not_found" { + t.Errorf("OutcomeCode = %q, want not_found", res.Denial.OutcomeCode) + } +} + +func TestResolveRoute_InvalidHostPreservesCause(t *testing.T) { + t.Parallel() + r := newTestResolver(nil) + res := r.ResolveRoute(context.Background(), RouteRequest{Authority: "no-suffix.example.com"}) + if res.Denial == nil { + t.Fatal("expected Denial") + } + if res.Denial.Cause == nil { + t.Error("Cause should be non-nil for log inspection") + } +} + +func TestResolveRoute_ResumeErrors(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + resumeErr error + wantStatus int + wantMsgHas string + wantCode string + causeIsOrig bool + }{ + { + name: "NotFound maps to 404", + resumeErr: status.Error(codes.NotFound, "actor missing"), + wantStatus: http.StatusNotFound, + wantMsgHas: `actor "` + testActorUUID + `" not found`, + wantCode: "not_found", + causeIsOrig: true, + }, + { + name: "FailedPrecondition maps to 503 with preserved desc", + resumeErr: status.Error(codes.FailedPrecondition, "no free workers available"), + wantStatus: http.StatusServiceUnavailable, + wantMsgHas: `actor "` + testActorUUID + `" unavailable: no free workers available`, + wantCode: "error", + causeIsOrig: true, + }, + { + name: "Unavailable maps to 503", + resumeErr: status.Error(codes.Unavailable, "control-plane down"), + wantStatus: http.StatusServiceUnavailable, + wantMsgHas: `actor "` + testActorUUID + `" unavailable`, + wantCode: "error", + causeIsOrig: true, + }, + { + name: "DeadlineExceeded maps to 504", + resumeErr: status.Error(codes.DeadlineExceeded, "deadline"), + wantStatus: http.StatusGatewayTimeout, + wantMsgHas: `actor "` + testActorUUID + `" request timed out`, + wantCode: "cancelled", + causeIsOrig: true, + }, + { + name: "PermissionDenied maps to 403", + resumeErr: status.Error(codes.PermissionDenied, "denied"), + wantStatus: http.StatusForbidden, + wantMsgHas: `actor "` + testActorUUID + `" access denied`, + wantCode: "error", + causeIsOrig: true, + }, + { + name: "Unauthenticated maps to 401", + resumeErr: status.Error(codes.Unauthenticated, "no creds"), + wantStatus: http.StatusUnauthorized, + wantMsgHas: `actor "` + testActorUUID + `" authentication required`, + wantCode: "error", + causeIsOrig: true, + }, + { + name: "ResourceExhausted maps to 429", + resumeErr: status.Error(codes.ResourceExhausted, "quota"), + wantStatus: http.StatusTooManyRequests, + wantMsgHas: `actor "` + testActorUUID + `" rate limited`, + wantCode: "error", + causeIsOrig: true, + }, + { + name: "non-gRPC error collapses to 500 without leaking detail", + resumeErr: errors.New("resume failed with sensitive detail"), + wantStatus: http.StatusInternalServerError, + wantMsgHas: `error resuming actor "` + testActorUUID + `"`, + wantCode: "error", + causeIsOrig: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + r := newTestResolver(func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + if in.GetActorRef().GetName() != testActorUUID { + t.Errorf("unexpected actor ID: %q", in.GetActorRef().GetName()) + } + if in.GetActorRef().GetAtespace() != "team-a" { + t.Errorf("unexpected atespace: %q", in.GetActorRef().GetAtespace()) + } + return nil, tc.resumeErr + }) + + res := r.ResolveRoute(context.Background(), RouteRequest{ + Authority: testActorUUID + ".team-a.actors.resources.substrate.ate.dev", + }) + + if res.Denial == nil { + t.Fatal("expected Denial, got Success") + } + d := res.Denial + if d.HTTPStatus != tc.wantStatus { + t.Errorf("HTTPStatus = %d, want %d", d.HTTPStatus, tc.wantStatus) + } + if tc.wantMsgHas != "" && d.Message != tc.wantMsgHas { + t.Errorf("Message = %q, want %q", d.Message, tc.wantMsgHas) + } + if d.OutcomeCode != tc.wantCode { + t.Errorf("OutcomeCode = %q, want %q", d.OutcomeCode, tc.wantCode) + } + if tc.causeIsOrig && !errors.Is(d, tc.resumeErr) { + t.Errorf("errors.Is(denial, original) = false; original error must be preserved for logs") + } + }) + } +} + +func TestResolveRoute_BadWorkerIP(t *testing.T) { + t.Parallel() + r := newTestResolver(func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + return &ateapipb.ResumeActorResponse{Actor: &ateapipb.Actor{AteomPodIp: "invalid-ip"}}, nil + }) + + res := r.ResolveRoute(context.Background(), RouteRequest{ + Authority: testActorUUID + ".team-a.actors.resources.substrate.ate.dev", + }) + + if res.Denial == nil { + t.Fatal("expected Denial for invalid IP, got Success") + } + if res.Denial.HTTPStatus != http.StatusInternalServerError { + t.Errorf("HTTPStatus = %d, want 500", res.Denial.HTTPStatus) + } + want := `actor "` + testActorUUID + `" routing failed` + if res.Denial.Message != want { + t.Errorf("Message = %q, want %q", res.Denial.Message, want) + } +} + +func TestResolveRoute_Success(t *testing.T) { + t.Parallel() + const workerIP = "10.0.0.52" + const tmplNs = "ate-system" + const tmplName = "my-template" + + r := newTestResolver(func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + return &ateapipb.ResumeActorResponse{Actor: &ateapipb.Actor{ + AteomPodIp: workerIP, + ActorTemplateNamespace: tmplNs, + ActorTemplateName: tmplName, + }}, nil + }) + + res := r.ResolveRoute(context.Background(), RouteRequest{ + Authority: testActorUUID + ".team-a.actors.resources.substrate.ate.dev", + }) + + if res.Success == nil { + t.Fatalf("expected Success, got Denial: %+v", res.Denial) + } + s := res.Success + if s.ActorID != testActorUUID { + t.Errorf("ActorID = %q, want %q", s.ActorID, testActorUUID) + } + if s.Backend.IP != workerIP { + t.Errorf("Backend.IP = %q, want %q", s.Backend.IP, workerIP) + } + if s.Backend.Port != 80 { + t.Errorf("Backend.Port = %d, want 80", s.Backend.Port) + } + if s.TemplateRef.Namespace != tmplNs { + t.Errorf("TemplateRef.Namespace = %q, want %q", s.TemplateRef.Namespace, tmplNs) + } + if s.TemplateRef.Name != tmplName { + t.Errorf("TemplateRef.Name = %q, want %q", s.TemplateRef.Name, tmplName) + } +} + +func TestMapResumeDenial_NilError(t *testing.T) { + t.Parallel() + if got := mapResumeDenial("x", nil); got != nil { + t.Errorf("mapResumeDenial(_, nil) = %v, want nil", got) + } +} + +func TestMapResumeDenial_PreservesCause(t *testing.T) { + t.Parallel() + orig := status.Error(codes.NotFound, "missing") + d := mapResumeDenial("actor-1", orig) + if !errors.Is(d, orig) { + t.Error("errors.Is(denial, original) = false; cause must be preserved for log inspection") + } +} diff --git a/cmd/atenet/internal/router/router.go b/cmd/atenet/internal/router/router.go index 7ed77df73..0a7f20b7a 100644 --- a/cmd/atenet/internal/router/router.go +++ b/cmd/atenet/internal/router/router.go @@ -16,15 +16,9 @@ package router import ( "context" - "crypto/rand" - "crypto/rsa" - "crypto/x509" - "crypto/x509/pkix" - "encoding/pem" "errors" "fmt" "log/slog" - "math/big" "net" "net/http" "os" @@ -68,21 +62,13 @@ type RouterConfig struct { Namespace string Kubeconfig string AteapiAddr string - HttpPort int - XdsPort int ExtprocPort int ExtprocAddr string - EnvoyImage string TemplatesFile string StatusPort int HealthInterval time.Duration - HttpsPort int - EnvoyCertPath string LogLevel string MetricsAddr string - // OtlpCollectorAddress is the host:port of the OTLP gRPC collector that - // Envoy reports tracing spans to. Empty disables Envoy-side tracing. - OtlpCollectorAddress string AteapiAuthMode string AteapiCAFile string @@ -217,35 +203,21 @@ func (s *RouterServer) Run(ctx context.Context) error { slog.InfoContext(ctx, "Connecting to ateapi", slog.String("address", s.cfg.AteapiAddr), slog.String("auth", string(authMode))) s.apiClient = ateapipb.NewControlClient(conn) - slog.InfoContext(ctx, "Starting substrate router subsystem", slog.Bool("standalone", s.cfg.Standalone)) + slog.InfoContext(ctx, "Starting substrate router subsystem", + slog.Bool("standalone", s.cfg.Standalone)) g, ctx := errgroup.WithContext(ctx) - xdsSrv := NewXdsServer(s.cfg.XdsPort) - xdsSrv.SetConfig(s.cfg.HttpPort, s.cfg.ExtprocPort, s.cfg.ExtprocAddr) - if err := xdsSrv.SetOtlpCollector(s.cfg.OtlpCollectorAddress); err != nil { - return fmt.Errorf("configure OTLP collector: %w", err) - } - - var certContent, keyContent string - if s.cfg.EnvoyCertPath == "" { - slog.InfoContext(ctx, "No Envoy certificate path provided, generating self-signed certificate for testing") - var err error - certContent, keyContent, err = generateSelfSignedCert() - if err != nil { - return fmt.Errorf("failed to generate self-signed cert: %w", err) - } - } - - xdsSrv.SetTlsConfig(s.cfg.HttpsPort, s.cfg.EnvoyCertPath, certContent, keyContent) if s.extprocSrv == nil { routeDuration, err := newRouteDurationHistogram() if err != nil { return fmt.Errorf("failed to create route-duration histogram: %w", err) } - s.extprocSrv = NewExtProcServer(s.cfg.ExtprocPort, s.apiClient, routeDuration) + resumer := NewActorResumer(s.apiClient) + resolver := NewActorRouteResolver(resumer) + s.extprocSrv = NewExtProcServer(s.cfg.ExtprocPort, resolver, routeDuration) } - ctrl := NewController(s.k8sClient, s.clientset, s.cfg, xdsSrv, s.extprocSrv) + ctrl := NewController(s.k8sClient, s.clientset, s.cfg, s.extprocSrv) s.health = newRouterHealth(s.cfg.HealthInterval, s.clientset, s.apiClient, s.cfg) @@ -262,18 +234,6 @@ func (s *RouterServer) Run(ctx context.Context) error { return nil }) - // Start xDS Server - g.Go(func() error { - slog.InfoContext(ctx, "Starting Envoy xDS Server", slog.Int("port", s.cfg.XdsPort)) - lis, err := net.Listen("tcp", fmt.Sprintf(":%d", s.cfg.XdsPort)) - if err != nil { - return fmt.Errorf("failed to listen on port %d: %w", s.cfg.XdsPort, err) - } - defer lis.Close() - - return xdsSrv.Serve(ctx, lis) - }) - // Start ExtProc Server g.Go(func() error { slog.InfoContext(ctx, "Starting ExtProc Server", slog.Int("port", s.cfg.ExtprocPort)) @@ -313,39 +273,3 @@ func (s *RouterServer) Run(ctx context.Context) error { return g.Wait() } - -func generateSelfSignedCert() (string, string, error) { - priv, err := rsa.GenerateKey(rand.Reader, 2048) - if err != nil { - return "", "", err - } - - template := x509.Certificate{ - SerialNumber: big.NewInt(1), - Subject: pkix.Name{ - Organization: []string{"Substrate Local Test"}, - }, - NotBefore: time.Now(), - NotAfter: time.Now().Add(time.Hour * 24 * 365), - - KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, - ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, - BasicConstraintsValid: true, - DNSNames: []string{"localhost"}, - } - - derBytes, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv) - if err != nil { - return "", "", err - } - - certPem := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: derBytes}) - - privBytes, err := x509.MarshalPKCS8PrivateKey(priv) - if err != nil { - return "", "", err - } - keyPem := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: privBytes}) - - return string(certPem), string(keyPem), nil -} diff --git a/cmd/atenet/internal/router/status.go b/cmd/atenet/internal/router/status.go index eab4c8183..ca76fe44d 100644 --- a/cmd/atenet/internal/router/status.go +++ b/cmd/atenet/internal/router/status.go @@ -138,8 +138,6 @@ type DashboardContext struct { BuildTag string `json:"build_tag"` RouterClusterIP string `json:"router_cluster_ip"` Namespace string `json:"namespace"` - HttpPort int `json:"port_http"` - XdsPort int `json:"port_xds"` ExtprocPort int `json:"port_extproc"` StatusPort int `json:"status_port"` Args string `json:"args"` @@ -248,8 +246,6 @@ func (s *RouterServer) handleStatusz(w http.ResponseWriter, req *http.Request) { BuildTag: buildInfo, RouterClusterIP: routerIP, Namespace: s.cfg.Namespace, - HttpPort: s.cfg.HttpPort, - XdsPort: s.cfg.XdsPort, ExtprocPort: s.cfg.ExtprocPort, StatusPort: s.cfg.StatusPort, Args: argsStr, diff --git a/cmd/atenet/internal/router/status_test.go b/cmd/atenet/internal/router/status_test.go index a72011848..8ddb81efb 100644 --- a/cmd/atenet/internal/router/status_test.go +++ b/cmd/atenet/internal/router/status_test.go @@ -51,8 +51,6 @@ func TestStatuszEndpoint(t *testing.T) { Standalone: true, Namespace: "default", StatusPort: httpPort, - HttpPort: 8080, - XdsPort: 18000, ExtprocPort: 50051, TemplatesFile: tmpFile.Name(), MetricsAddr: "127.0.0.1:0", @@ -64,7 +62,7 @@ func TestStatuszEndpoint(t *testing.T) { t.Fatalf("Failed generating router server: %v", err) } - srv.extprocSrv = NewExtProcServer(cfg.ExtprocPort, &mockClient{}, nil) + srv.extprocSrv = NewExtProcServer(cfg.ExtprocPort, &mockResolver{}, nil) ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/cmd/atenet/internal/router/xds.go b/cmd/atenet/internal/router/xds.go deleted file mode 100644 index e7e1602a9..000000000 --- a/cmd/atenet/internal/router/xds.go +++ /dev/null @@ -1,606 +0,0 @@ -// 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 router - -import ( - "context" - "fmt" - "log/slog" - "net" - "strconv" - "sync" - "time" - - "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" - "google.golang.org/grpc" - "google.golang.org/protobuf/types/known/anypb" - "google.golang.org/protobuf/types/known/durationpb" - "google.golang.org/protobuf/types/known/wrapperspb" - - accesslogv3 "github.com/envoyproxy/go-control-plane/envoy/config/accesslog/v3" - clusterv3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3" - mutationrulesv3 "github.com/envoyproxy/go-control-plane/envoy/config/common/mutation_rules/v3" - corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" - endpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3" - listenerv3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" - routev3 "github.com/envoyproxy/go-control-plane/envoy/config/route/v3" - tracev3 "github.com/envoyproxy/go-control-plane/envoy/config/trace/v3" - streamaccesslogv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/access_loggers/stream/v3" - dfpclusterv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/dynamic_forward_proxy/v3" - dfpcommonv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/common/dynamic_forward_proxy/v3" - dfpv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/dynamic_forward_proxy/v3" - extprocv3filter "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3" - routerv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/router/v3" - hcmv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3" - getaddrinfov3 "github.com/envoyproxy/go-control-plane/envoy/extensions/network/dns_resolver/getaddrinfo/v3" - tlsv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3" - httpv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/upstreams/http/v3" - clustergrpc "github.com/envoyproxy/go-control-plane/envoy/service/cluster/v3" - discoverygrpc "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" - endpointgrpc "github.com/envoyproxy/go-control-plane/envoy/service/endpoint/v3" - listenergrpc "github.com/envoyproxy/go-control-plane/envoy/service/listener/v3" - routegrpc "github.com/envoyproxy/go-control-plane/envoy/service/route/v3" - typev3 "github.com/envoyproxy/go-control-plane/envoy/type/v3" - "github.com/envoyproxy/go-control-plane/pkg/cache/types" - cachev3 "github.com/envoyproxy/go-control-plane/pkg/cache/v3" - resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" - serverv3 "github.com/envoyproxy/go-control-plane/pkg/server/v3" -) - -const ( - NodeID = "substrate-envoy-node" - IngressHTTPListener = "ingress_http_listener" - IngressHTTPSListener = "ingress_https_listener" - RouteName = "substrate_routes" - ClusterName = "ate-cluster" - OtlpClusterName = "otel_collector_cluster" -) - -// XdsServer implements an aggregated discovery service server for dynamic Envoy router nodes. -type XdsServer struct { - xdsPort int - extprocPort int - extprocAddr string - ingressPort int - snapshot cachev3.SnapshotCache - srv serverv3.Server - versionCount int64 - - mu sync.Mutex - - httpsPort int - certPath string - certContent string - keyContent string - - otlpHost string - otlpPort uint32 -} - -func NewXdsServer(xdsPort int) *XdsServer { - cache := cachev3.NewSnapshotCache(true, cachev3.IDHash{}, nil) - srv := serverv3.NewServer(context.Background(), cache, nil) - - return &XdsServer{ - xdsPort: xdsPort, - snapshot: cache, - srv: srv, - extprocPort: 50051, // matches default extproc port - extprocAddr: "127.0.0.1", - ingressPort: 8080, - } -} - -func (x *XdsServer) SetConfig(ingressPort int, extprocPort int, extprocAddr string) { - x.mu.Lock() - defer x.mu.Unlock() - x.ingressPort = ingressPort - x.extprocPort = extprocPort - x.extprocAddr = extprocAddr -} - -func (x *XdsServer) SetTlsConfig(httpsPort int, certPath string, certContent string, keyContent string) { - x.mu.Lock() - defer x.mu.Unlock() - x.httpsPort = httpsPort - x.certPath = certPath - x.certContent = certContent - x.keyContent = keyContent -} - -// SetOtlpCollector enables Envoy-side tracing pointed at the OTLP gRPC -// collector at host:port. addr empty disables tracing. port defaults to -// 4317 if omitted. -func (x *XdsServer) SetOtlpCollector(addr string) error { - x.mu.Lock() - defer x.mu.Unlock() - if addr == "" { - x.otlpHost = "" - x.otlpPort = 0 - return nil - } - host, portStr, err := net.SplitHostPort(addr) - if err != nil { - host = addr - portStr = "4317" - } - port, err := strconv.ParseUint(portStr, 10, 32) - if err != nil { - return fmt.Errorf("parse OTLP collector port from %q: %w", addr, err) - } - x.otlpHost = host - x.otlpPort = uint32(port) - return nil -} - -func (x *XdsServer) UpdateSnapshot() error { - x.mu.Lock() - defer x.mu.Unlock() - - x.versionCount++ - ver := strconv.FormatInt(x.versionCount, 10) - - // Clusters - clusters := []types.Resource{ - x.buildCluster(), - x.buildDynamicForwardProxyCluster(), - } - if x.otlpHost != "" { - clusters = append(clusters, x.buildOtlpCollectorCluster()) - } - - // Routes - routes := []types.Resource{ - x.buildRoutes(), - } - - // Listeners - listeners := []types.Resource{ - x.buildListener(), - } - if x.httpsPort > 0 { - listeners = append(listeners, x.buildHttpsListener()) - } - - // Snapshot - snapshot, err := cachev3.NewSnapshot(ver, map[resourcev3.Type][]types.Resource{ - resourcev3.ClusterType: clusters, - resourcev3.RouteType: routes, - resourcev3.ListenerType: listeners, - }) - - if err != nil { - return fmt.Errorf("failed to build xDS Snapshot: %w", err) - } - - if err := snapshot.Consistent(); err != nil { - return fmt.Errorf("snapshot evaluation failed integrity check: %w", err) - } - - slog.Info("Deploying updated xDS configuration snapshot", slog.String("version", ver)) - return x.snapshot.SetSnapshot(context.Background(), NodeID, snapshot) -} - -func (x *XdsServer) Serve(ctx context.Context, lis net.Listener) error { - // Ensure a first snapshot is deployed - if err := x.UpdateSnapshot(); err != nil { - slog.ErrorContext(ctx, "Warning - initial xDS setup update failed", slog.String("err", err.Error())) - } - - grpcServer := grpc.NewServer( - grpc.StatsHandler(otelgrpc.NewServerHandler()), - ) - discoverygrpc.RegisterAggregatedDiscoveryServiceServer(grpcServer, x.srv) - clustergrpc.RegisterClusterDiscoveryServiceServer(grpcServer, x.srv) - endpointgrpc.RegisterEndpointDiscoveryServiceServer(grpcServer, x.srv) - listenergrpc.RegisterListenerDiscoveryServiceServer(grpcServer, x.srv) - routegrpc.RegisterRouteDiscoveryServiceServer(grpcServer, x.srv) - - errChan := make(chan error, 1) - go func() { - if err := grpcServer.Serve(lis); err != nil { - errChan <- err - } - }() - - select { - case <-ctx.Done(): - grpcServer.GracefulStop() - return nil - case err := <-errChan: - return err - } -} - -func (x *XdsServer) buildCluster() *clusterv3.Cluster { - h2Opts, _ := anypb.New(&httpv3.HttpProtocolOptions{ - UpstreamProtocolOptions: &httpv3.HttpProtocolOptions_ExplicitHttpConfig_{ - ExplicitHttpConfig: &httpv3.HttpProtocolOptions_ExplicitHttpConfig{ - ProtocolConfig: &httpv3.HttpProtocolOptions_ExplicitHttpConfig_Http2ProtocolOptions{}, - }, - }, - }) - - return &clusterv3.Cluster{ - Name: ClusterName, - ConnectTimeout: durationpb.New(250 * time.Millisecond), - ClusterDiscoveryType: &clusterv3.Cluster_Type{ - Type: clusterv3.Cluster_STATIC, - }, - LbPolicy: clusterv3.Cluster_ROUND_ROBIN, - LoadAssignment: &endpointv3.ClusterLoadAssignment{ - ClusterName: ClusterName, - Endpoints: []*endpointv3.LocalityLbEndpoints{ - { - LbEndpoints: []*endpointv3.LbEndpoint{ - { - HostIdentifier: &endpointv3.LbEndpoint_Endpoint{ - Endpoint: &endpointv3.Endpoint{ - Address: &corev3.Address{ - Address: &corev3.Address_SocketAddress{ - SocketAddress: &corev3.SocketAddress{ - Address: x.extprocAddr, - PortSpecifier: &corev3.SocketAddress_PortValue{ - PortValue: uint32(x.extprocPort), - }, - }, - }, - }, - }, - }, - }, - }, - }, - }, - }, - TypedExtensionProtocolOptions: map[string]*anypb.Any{ - "envoy.extensions.upstreams.http.v3.HttpProtocolOptions": h2Opts, - }, - } -} - -func buildDnsCacheConfig() *dfpcommonv3.DnsCacheConfig { - resolverConfigAny, _ := anypb.New(&getaddrinfov3.GetAddrInfoDnsResolverConfig{}) - return &dfpcommonv3.DnsCacheConfig{ - Name: "dynamic_forward_proxy_cache_config", - DnsLookupFamily: clusterv3.Cluster_V4_ONLY, - TypedDnsResolverConfig: &corev3.TypedExtensionConfig{ - Name: "envoy.network.dns_resolver.getaddrinfo", - TypedConfig: resolverConfigAny, - }, - } -} - -// buildOtlpCollectorCluster builds a STRICT_DNS HTTP/2 cluster that -// targets the OTLP gRPC collector. Required when HCM tracing is enabled -// so Envoy has somewhere to ship spans. -func (x *XdsServer) buildOtlpCollectorCluster() *clusterv3.Cluster { - h2Opts, _ := anypb.New(&httpv3.HttpProtocolOptions{ - UpstreamProtocolOptions: &httpv3.HttpProtocolOptions_ExplicitHttpConfig_{ - ExplicitHttpConfig: &httpv3.HttpProtocolOptions_ExplicitHttpConfig{ - ProtocolConfig: &httpv3.HttpProtocolOptions_ExplicitHttpConfig_Http2ProtocolOptions{}, - }, - }, - }) - - return &clusterv3.Cluster{ - Name: OtlpClusterName, - ConnectTimeout: durationpb.New(1 * time.Second), - ClusterDiscoveryType: &clusterv3.Cluster_Type{ - Type: clusterv3.Cluster_STRICT_DNS, - }, - LbPolicy: clusterv3.Cluster_ROUND_ROBIN, - LoadAssignment: &endpointv3.ClusterLoadAssignment{ - ClusterName: OtlpClusterName, - Endpoints: []*endpointv3.LocalityLbEndpoints{ - { - LbEndpoints: []*endpointv3.LbEndpoint{ - { - HostIdentifier: &endpointv3.LbEndpoint_Endpoint{ - Endpoint: &endpointv3.Endpoint{ - Address: &corev3.Address{ - Address: &corev3.Address_SocketAddress{ - SocketAddress: &corev3.SocketAddress{ - Address: x.otlpHost, - PortSpecifier: &corev3.SocketAddress_PortValue{ - PortValue: x.otlpPort, - }, - }, - }, - }, - }, - }, - }, - }, - }, - }, - }, - TypedExtensionProtocolOptions: map[string]*anypb.Any{ - "envoy.extensions.upstreams.http.v3.HttpProtocolOptions": h2Opts, - }, - } -} - -func (x *XdsServer) buildDynamicForwardProxyCluster() *clusterv3.Cluster { - dfpClusterConfig := &dfpclusterv3.ClusterConfig{ - ClusterImplementationSpecifier: &dfpclusterv3.ClusterConfig_DnsCacheConfig{ - DnsCacheConfig: buildDnsCacheConfig(), - }, - } - - clusterConfigAny, _ := anypb.New(dfpClusterConfig) - - return &clusterv3.Cluster{ - Name: "dynamic_forward_proxy_cluster", - LbPolicy: clusterv3.Cluster_CLUSTER_PROVIDED, - ClusterDiscoveryType: &clusterv3.Cluster_ClusterType{ - ClusterType: &clusterv3.Cluster_CustomClusterType{ - Name: "envoy.clusters.dynamic_forward_proxy", - TypedConfig: clusterConfigAny, - }, - }, - } -} - -func (x *XdsServer) buildRoutes() *routev3.RouteConfiguration { - return &routev3.RouteConfiguration{ - Name: RouteName, - VirtualHosts: []*routev3.VirtualHost{ - { - Name: "local_service", - Domains: []string{"*"}, - Routes: []*routev3.Route{ - { - Match: &routev3.RouteMatch{ - PathSpecifier: &routev3.RouteMatch_Prefix{ - Prefix: "/", - }, - }, - Action: &routev3.Route_Route{ - Route: &routev3.RouteAction{ - ClusterSpecifier: &routev3.RouteAction_Cluster{ - Cluster: "dynamic_forward_proxy_cluster", - }, - Timeout: durationpb.New(10 * time.Second), - }, - }, - }, - }, - }, - }, - } -} - -func (x *XdsServer) buildHcm(statPrefix string) *anypb.Any { - extProcConfig, _ := anypb.New(&extprocv3filter.ExternalProcessor{ - GrpcService: &corev3.GrpcService{ - TargetSpecifier: &corev3.GrpcService_EnvoyGrpc_{ - EnvoyGrpc: &corev3.GrpcService_EnvoyGrpc{ - ClusterName: ClusterName, - }, - }, - Timeout: durationpb.New(5 * time.Second), - }, - MutationRules: &mutationrulesv3.HeaderMutationRules{ - AllowAllRouting: &wrapperspb.BoolValue{Value: true}, - }, - // Explicitly configure the message timeout to avoid the 200ms default - MessageTimeout: durationpb.New(5 * time.Second), - ProcessingMode: &extprocv3filter.ProcessingMode{ - RequestHeaderMode: extprocv3filter.ProcessingMode_SEND, - ResponseHeaderMode: extprocv3filter.ProcessingMode_SKIP, - RequestBodyMode: extprocv3filter.ProcessingMode_NONE, - ResponseBodyMode: extprocv3filter.ProcessingMode_NONE, - RequestTrailerMode: extprocv3filter.ProcessingMode_SKIP, - ResponseTrailerMode: extprocv3filter.ProcessingMode_SKIP, - }, - }) - - dfpFilterConfig, _ := anypb.New(&dfpv3.FilterConfig{ - ImplementationSpecifier: &dfpv3.FilterConfig_DnsCacheConfig{ - DnsCacheConfig: buildDnsCacheConfig(), - }, - }) - - routerAny, _ := anypb.New(&routerv3.Router{}) - - accessLogConfig, _ := anypb.New(&streamaccesslogv3.StdoutAccessLog{}) - - hcm, _ := anypb.New(&hcmv3.HttpConnectionManager{ - StatPrefix: statPrefix, - GenerateRequestId: &wrapperspb.BoolValue{Value: true}, - Tracing: x.buildTracing(), - AccessLog: []*accesslogv3.AccessLog{ - { - Name: "envoy.access_loggers.stdout", - ConfigType: &accesslogv3.AccessLog_TypedConfig{ - TypedConfig: accessLogConfig, - }, - }, - }, - HttpFilters: []*hcmv3.HttpFilter{ - { - Name: "envoy.filters.http.ext_proc", - ConfigType: &hcmv3.HttpFilter_TypedConfig{ - TypedConfig: extProcConfig, - }, - }, - { - Name: "envoy.filters.http.dynamic_forward_proxy", - ConfigType: &hcmv3.HttpFilter_TypedConfig{ - TypedConfig: dfpFilterConfig, - }, - }, - { - Name: "envoy.filters.http.router", - ConfigType: &hcmv3.HttpFilter_TypedConfig{ - TypedConfig: routerAny, - }, - }, - }, - RouteSpecifier: &hcmv3.HttpConnectionManager_Rds{ - Rds: &hcmv3.Rds{ - RouteConfigName: RouteName, - ConfigSource: &corev3.ConfigSource{ - ResourceApiVersion: corev3.ApiVersion_V3, - ConfigSourceSpecifier: &corev3.ConfigSource_Ads{ - Ads: &corev3.AggregatedConfigSource{}, - }, - }, - }, - }, - }) - return hcm -} - -// buildTracing returns the HCM Tracing block that points Envoy at the -// configured OTLP gRPC collector. Returns nil when no collector is set, -// in which case Envoy emits no spans on its own. -// -// `RandomSampling: 100%` makes Envoy ALWAYS sample requests that arrive -// with no parent traceparent. We rely on upstream clients (locust, etc.) -// to gate sampling: requests without a sampled parent are still tagged -// `sampled` here but downstream services in this repo use -// `ParentBased(NeverSample)` so unsampled-by-client requests stay -// unsampled overall. -func (x *XdsServer) buildTracing() *hcmv3.HttpConnectionManager_Tracing { - if x.otlpHost == "" { - return nil - } - otelConfig, _ := anypb.New(&tracev3.OpenTelemetryConfig{ - GrpcService: &corev3.GrpcService{ - TargetSpecifier: &corev3.GrpcService_EnvoyGrpc_{ - EnvoyGrpc: &corev3.GrpcService_EnvoyGrpc{ - ClusterName: OtlpClusterName, - }, - }, - }, - ServiceName: "atenet-router-envoy", - }) - return &hcmv3.HttpConnectionManager_Tracing{ - RandomSampling: &typev3.Percent{Value: 100}, - Provider: &tracev3.Tracing_Http{ - Name: "envoy.tracers.opentelemetry", - ConfigType: &tracev3.Tracing_Http_TypedConfig{ - TypedConfig: otelConfig, - }, - }, - } -} - -func (x *XdsServer) buildListener() *listenerv3.Listener { - hcm := x.buildHcm("ingress_http") - - return &listenerv3.Listener{ - Name: IngressHTTPListener, - Address: &corev3.Address{ - Address: &corev3.Address_SocketAddress{ - SocketAddress: &corev3.SocketAddress{ - Address: "0.0.0.0", - PortSpecifier: &corev3.SocketAddress_PortValue{ - PortValue: uint32(x.ingressPort), - }, - }, - }, - }, - FilterChains: []*listenerv3.FilterChain{ - { - Filters: []*listenerv3.Filter{ - { - Name: "envoy.filters.network.http_connection_manager", - ConfigType: &listenerv3.Filter_TypedConfig{ - TypedConfig: hcm, - }, - }, - }, - }, - }, - } -} - -func (x *XdsServer) buildHttpsListener() *listenerv3.Listener { - hcm := x.buildHcm("ingress_https") - - tlsConfig := &tlsv3.DownstreamTlsContext{ - CommonTlsContext: &tlsv3.CommonTlsContext{ - TlsCertificates: []*tlsv3.TlsCertificate{ - x.buildTlsCertificate(), - }, - }, - } - tlsConfigAny, _ := anypb.New(tlsConfig) - - return &listenerv3.Listener{ - Name: IngressHTTPSListener, - Address: &corev3.Address{ - Address: &corev3.Address_SocketAddress{ - SocketAddress: &corev3.SocketAddress{ - Address: "0.0.0.0", - PortSpecifier: &corev3.SocketAddress_PortValue{ - PortValue: uint32(x.httpsPort), - }, - }, - }, - }, - FilterChains: []*listenerv3.FilterChain{ - { - Filters: []*listenerv3.Filter{ - { - Name: "envoy.filters.network.http_connection_manager", - ConfigType: &listenerv3.Filter_TypedConfig{ - TypedConfig: hcm, - }, - }, - }, - TransportSocket: &corev3.TransportSocket{ - Name: "envoy.transport_sockets.tls", - ConfigType: &corev3.TransportSocket_TypedConfig{ - TypedConfig: tlsConfigAny, - }, - }, - }, - }, - } -} - -func (x *XdsServer) buildTlsCertificate() *tlsv3.TlsCertificate { - if x.certPath != "" { - return &tlsv3.TlsCertificate{ - CertificateChain: &corev3.DataSource{ - Specifier: &corev3.DataSource_Filename{ - Filename: x.certPath, - }, - }, - PrivateKey: &corev3.DataSource{ - Specifier: &corev3.DataSource_Filename{ - Filename: x.certPath, // Assuming combined file - }, - }, - } - } - - return &tlsv3.TlsCertificate{ - CertificateChain: &corev3.DataSource{ - Specifier: &corev3.DataSource_InlineString{ - InlineString: x.certContent, - }, - }, - PrivateKey: &corev3.DataSource{ - Specifier: &corev3.DataSource_InlineString{ - InlineString: x.keyContent, - }, - }, - } -} diff --git a/cmd/atenet/internal/router/xds_test.go b/cmd/atenet/internal/router/xds_test.go deleted file mode 100644 index 92e347648..000000000 --- a/cmd/atenet/internal/router/xds_test.go +++ /dev/null @@ -1,212 +0,0 @@ -// 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 router - -import ( - "context" - "net" - "strings" - "testing" - "time" - - clusterv3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3" - listenerv3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" - routev3 "github.com/envoyproxy/go-control-plane/envoy/config/route/v3" - cachev3 "github.com/envoyproxy/go-control-plane/pkg/cache/v3" - resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" -) - -func TestXdsServer_UpdateSnapshot(t *testing.T) { - server := NewXdsServer(18000) - server.SetConfig(8081, 50052, "10.0.0.1") - - err := server.UpdateSnapshot() - if err != nil { - t.Fatalf("UpdateSnapshot failed: %v", err) - } - - res, err := server.snapshot.GetSnapshot(NodeID) - if err != nil { - t.Fatalf("Failed to get generated snapshot: %v", err) - } - - snap, ok := res.(*cachev3.Snapshot) - if !ok { - t.Fatalf("Snapshot doesn't conform to type *cachev3.Snapshot, got %T", res) - } - - // Check consistent snapshot - if err := snap.Consistent(); err != nil { - t.Fatalf("Integrity check failed on snapshot: %v", err) - } - - // Verify clusters generated - clustersMap := snap.GetResources(resourcev3.ClusterType) - if len(clustersMap) != 2 { - t.Errorf("Expected 2 cluster definitions, got %d", len(clustersMap)) - } - - if raw, exists := clustersMap["ate-cluster"]; !exists { - t.Error("Static 'ate-cluster' is missing from clusters") - } else { - c := raw.(*clusterv3.Cluster) - if c.GetName() != "ate-cluster" { - t.Errorf("Expected name 'ate-cluster', got %s", c.GetName()) - } - - // Validate Endpoint address mapped from Server parameters - eps := c.GetLoadAssignment().GetEndpoints()[0].GetLbEndpoints()[0].GetEndpoint().GetAddress().GetSocketAddress() - if eps.GetAddress() != "10.0.0.1" { - t.Errorf("Expected address '10.0.0.1', got %s", eps.GetAddress()) - } - if eps.GetPortValue() != 50052 { - t.Errorf("Expected port 50052, got %d", eps.GetPortValue()) - } - } - - if raw, exists := clustersMap["dynamic_forward_proxy_cluster"]; !exists { - t.Error("'dynamic_forward_proxy_cluster' is missing from clusters") - } else { - c := raw.(*clusterv3.Cluster) - if c.GetName() != "dynamic_forward_proxy_cluster" { - t.Errorf("Expected 'dynamic_forward_proxy_cluster', got %s", c.GetName()) - } - } - - // Verify Virtual Hosts generated inside Route configuration - routesMap := snap.GetResources(resourcev3.RouteType) - if len(routesMap) != 1 { - t.Fatalf("Expected 1 route configuration object, got %d", len(routesMap)) - } - - if raw, exists := routesMap[RouteName]; !exists { - t.Errorf("Route name '%s' is missing from snapshot routes configuration", RouteName) - } else { - rc := raw.(*routev3.RouteConfiguration) - if rc.GetName() != RouteName { - t.Errorf("Expected route name '%s', got %s", RouteName, rc.GetName()) - } - - if len(rc.GetVirtualHosts()) != 1 { - t.Fatalf("Expected 1 VirtualHost definition for static routes case, got %d", len(rc.GetVirtualHosts())) - } - - vh := rc.GetVirtualHosts()[0] - if len(vh.GetDomains()) != 1 || vh.GetDomains()[0] != "*" { - t.Errorf("Expected domain '*', got %v", vh.GetDomains()) - } - - if len(vh.GetRoutes()) != 1 { - t.Fatalf("Expected 1 route in fallback VirtualHost, got %d", len(vh.GetRoutes())) - } - - fallbackRoute := vh.GetRoutes()[0] - if fallbackRoute.GetMatch().GetPrefix() != "/" { - t.Errorf("Expected path mapping prefix '/', got '%s'", fallbackRoute.GetMatch().GetPrefix()) - } - } - - // Verify listeners generated - listenersMap := snap.GetResources(resourcev3.ListenerType) - if len(listenersMap) != 1 { - t.Fatalf("Expected 1 listener definition, got %d", len(listenersMap)) - } - - 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()) - } - } -} - -func TestXdsServer_UpdateSnapshot_WithHttps(t *testing.T) { - server := NewXdsServer(18000) - server.SetConfig(8085, 50053, "127.0.0.1") - server.SetTlsConfig(8443, "", "dummy-cert", "dummy-key") - - err := server.UpdateSnapshot() - if err != nil { - t.Fatalf("UpdateSnapshot failed: %v", err) - } - - res, err := server.snapshot.GetSnapshot(NodeID) - if err != nil { - t.Fatalf("Failed to get snapshot: %v", err) - } - - snap, ok := res.(*cachev3.Snapshot) - if !ok { - t.Fatalf("Snapshot doesn't conform to type *cachev3.Snapshot, got %T", res) - } - - listenersMap := snap.GetResources(resourcev3.ListenerType) - if len(listenersMap) != 2 { - t.Fatalf("Expected 2 listener definitions, got %d", len(listenersMap)) - } - - if raw, exists := listenersMap[IngressHTTPSListener]; !exists { - 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()) - } - - // Verify TLS config - fc := l.GetFilterChains()[0] - ts := fc.GetTransportSocket() - if ts.GetName() != "envoy.transport_sockets.tls" { - t.Errorf("Expected transport socket 'envoy.transport_sockets.tls', got '%s'", ts.GetName()) - } - } -} - -func TestXdsServer_Serve_Shutdown(t *testing.T) { - server := NewXdsServer(18000) - server.SetConfig(8085, 50053, "127.0.0.1") - - lis, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatalf("Failed to create tcp listener: %v", err) - } - defer lis.Close() - - ctx, cancel := context.WithCancel(context.Background()) - errChan := make(chan error, 1) - - go func() { - errChan <- server.Serve(ctx, lis) - }() - - // Cancel the context to trigger graceful stop - cancel() - - select { - case err := <-errChan: - if err != nil && !strings.Contains(err.Error(), "use of closed network connection") { - t.Errorf("Serve error returned: %v", err) - } - case <-time.After(2 * time.Second): - t.Error("Timeout exceeded waiting for Serve to finish graceful closure") - } -} diff --git a/cmd/benchmarking/boomer-glutton/main.go b/cmd/benchmarking/boomer-glutton/main.go index 9a5817bfd..c6adde578 100644 --- a/cmd/benchmarking/boomer-glutton/main.go +++ b/cmd/benchmarking/boomer-glutton/main.go @@ -37,7 +37,7 @@ import ( func main() { var ( apiEndpoint = flag.String("api-endpoint", "api.ate-system.svc.cluster.local:443", "ateapi gRPC endpoint host:port.") - routerURL = flag.String("router-url", "http://atenet-router.ate-system.svc.cluster.local", "atenet HTTP router base URL (no trailing slash).") + routerURL = flag.String("router-url", "http://atenet-gateway.ate-system.svc.cluster.local", "atenet HTTP gateway base URL (no trailing slash). atenet-router itself only serves ExtProc gRPC, not HTTP.") atespace = flag.String("atespace", "benchmark", "Atespace every actor this worker creates lives in. Ensured (CreateAtespace, AlreadyExists is ok) at startup.") promAddr = flag.String("prometheus-addr", ":8001", "Address for the Prometheus /metrics endpoint.") configJSON = flag.String("config-json", "", "Initial dynconfig as a JSON object (keys: trace_probability, min_wait_time, max_wait_time in seconds). Unset fields keep their built-in defaults.") diff --git a/docs/dev/extproc-ingress-contract.md b/docs/dev/extproc-ingress-contract.md new file mode 100644 index 000000000..49f93cc61 --- /dev/null +++ b/docs/dev/extproc-ingress-contract.md @@ -0,0 +1,145 @@ +# ExtProc Ingress Integration Contract + +This document describes the contract that any ExtProc-capable proxy must satisfy to integrate with Substrate's actor-aware routing core. + +## Overview + +The request flow is: + +``` +Client + → external proxy (Envoy / gateway) + → ExtProc gRPC → atenet-router:50051 + → RouteResolver (ResumeActor → worker pod IP) + ← ExtProc response (header mutation or denial) + → external proxy forwards to worker-ip:80 + → actor pod +``` + +atenet-router exposes a standard Envoy `ExternalProcessor` gRPC service. Any proxy that supports the `ext_proc` filter can call it. + +--- + +## Required Request Metadata + +The proxy **must** forward the following HTTP headers verbatim to ExtProc: + +| Header | Requirement | Notes | +|--------|-------------|-------| +| `:authority` / `Host` | **Required** | Must end with `.actors.resources.substrate.ate.dev`. The actor ID is the subdomain preceding that suffix. | +| `traceparent` | Recommended | W3C Trace Context. Substrate extracts this to link its internal span to the proxy's ingress span. | +| `tracestate` | Recommended | Forwarded alongside `traceparent`. | + +No other headers are required. Additional headers are passed through unchanged. + +### Actor ID format + +Given `:authority: ..actors.resources.substrate.ate.dev`: + +- Valid actor IDs satisfy the `resources.ActorIDRegexPattern` regex (UUID-like identifiers). +- Port numbers in the authority header (e.g. `actor.actors...ate.dev:80`) are stripped before parsing. + +--- + +## Success Behavior + +When the actor is found and a worker pod is ready, Substrate returns an ExtProc `HeadersResponse` with a **header mutation**: + +``` +SetHeaders: + :authority = :80 +``` + +The proxy should then forward the request to `:80` using the rewritten authority. The recommended mechanism is Envoy's **dynamic_forward_proxy** cluster, which resolves the authority as a host:port at request time. + +--- + +## Denial Behavior + +When routing cannot complete, Substrate returns an ExtProc **ImmediateResponse**: + +| Condition | HTTP Status | Example body | +|-----------|-------------|--------------| +| Host doesn't match actor DNS suffix | 404 | `invalid host "foo.example.com": …` | +| Actor not found in control plane | 404 | `actor "abc-123" not found` | +| No free workers (FailedPrecondition) | 503 | `actor "abc-123" unavailable: no free workers available` | +| Control plane unreachable | 503 | `actor "abc-123" unavailable` | +| Resume timed out | 504 | `actor "abc-123" request timed out` | +| Invalid worker IP returned | 500 | `actor "abc-123" routing failed` | +| Unknown error | 500 | `error resuming actor "abc-123"` | + +All denial responses carry `Content-Type: text/plain`. + +The body is **client-safe**: internal error detail (stack traces, internal IDs) is never included. + +--- + +## ExtProc Filter Configuration + +The following ExtProc filter configuration is required on the proxy side: + +```yaml +name: envoy.filters.http.ext_proc +typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor + grpc_service: + envoy_grpc: + cluster_name: extproc_cluster # must point to atenet-router:50051 + timeout: 5s + mutation_rules: + allow_all_routing: true # required: lets ExtProc rewrite :authority + message_timeout: 5s + processing_mode: + request_header_mode: SEND # only request headers are processed + response_header_mode: SKIP + request_body_mode: NONE + response_body_mode: NONE + request_trailer_mode: SKIP + response_trailer_mode: SKIP +``` + +`allow_all_routing: true` is mandatory. Without it Envoy silently drops the `:authority` mutation and the proxy routes to the wrong upstream. + +### ExtProc cluster + +```yaml +name: extproc_cluster +connect_timeout: 0.25s +type: STRICT_DNS +typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} # gRPC requires HTTP/2 +load_assignment: + cluster_name: extproc_cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: atenet-router.ate-system.svc + port_value: 50051 +``` + +> **TLS:** ExtProc gRPC is plaintext for the PoC. atenet-router and the proxy are cluster-internal; mTLS hardening is deferred. + +--- + +## DNS Integration + +The DNS controller must resolve actor hostnames to your **gateway's** ClusterIP, not atenet-router's (atenet-router only serves ExtProc gRPC, not HTTP): + +``` +atenet dns --ingress-service-name=atenet-gateway ... +``` + +This writes the `atenet-gateway` Service ClusterIP into the CoreDNS Corefile so that `*.actors.resources.substrate.ate.dev` resolves to the external proxy. + +--- + +## Security Notes + +- The ExtProc endpoint (port 50051) should only be reachable by the proxy within the cluster. Use NetworkPolicy to restrict access if needed. +- Substrate never forwards client secrets (Authorization headers, cookies, query parameters) to logs. The `traceparent` header is the only header value logged. +- Denial message bodies are client-safe: they do not include internal stack traces, IP addresses of failed backends, or gRPC internal error strings (except for `FailedPrecondition`, whose description is considered actionable and non-sensitive). diff --git a/docs/dev/ingress.md b/docs/dev/ingress.md new file mode 100644 index 000000000..9d40b5cec --- /dev/null +++ b/docs/dev/ingress.md @@ -0,0 +1,90 @@ +# Ingress integration: gateway reference configs + +Substrate does not ship or manage a proxy. `atenet-router` exposes an ext_proc +gRPC endpoint (port 50051) that any dynamic-forwarding proxy can call to +resolve actor routes — see [the contract](./extproc-ingress-contract.md). You +bring and configure your own gateway in front of it. + +This doc covers the reference configs in `manifests/ate-install/examples/`: +a basic path (raw Envoy, scripted) and power-user paths (agentgateway, plus +Gateway API variants of both). Substrate does not generate any of these +resources — apply the one you want yourself. + +--- + +## Proxy Configurations + +### Standalone Envoy Deployment + +`manifests/ate-install/examples/atenet-gateway-envoy.yaml` is a standalone +Envoy deployment configured directly (no Gateway API / gateway controller). +`hack/poc-pluggable-ingress.sh deploy` applies it on a kind cluster end to +end — this is the fastest way to see the pipeline working. + +``` +Client + → Envoy (static config) + ext_proc filter → atenet-router:50051 + resolves actor → rewrites :authority to :80 (or denies) + dynamic_forward_proxy cluster + re-resolves target from :authority, connects to :80 + → worker pod (ateom) +``` + +Envoy's `dynamic_forward_proxy` cluster re-resolves the request authority +**after** the ext_proc filter has run and dials that address directly — so the +actor-aware decision made by `atenet-router` becomes the upstream target with +no per-actor config. Denials work unchanged: `atenet-router` returns an +ext_proc `ImmediateResponse` (404/503/etc.), which Envoy honors directly. + +DNS: point actor hostnames at the gateway's Service. `atenet-gateway` is +already the DNS controller's default ingress target +(`atenet dns --ingress-service-name`, see +[extproc-ingress-contract.md](./extproc-ingress-contract.md#dns-integration)), +so no flag is needed if you keep that Service name. + +For anything past the basic path — agentgateway, or a Gateway API–driven +control plane — see the power-user section below. + +#### Standalone AgentGateway Deployment + +`manifests/ate-install/examples/atenet-gateway-agw.yaml` is the agentgateway +(https://agentgateway.dev) equivalent of the basic Envoy path: a hand-written +config file, not Gateway API. Deploy it **instead of** +`atenet-gateway-envoy.yaml` (they share the `atenet-gateway` Service name). +Same mapping as the basic path: ext_proc rewrites `:authority`, agentgateway's +`dynamicForwardProxy` backend reads it after ext_proc runs. + +### AgentGateway (Gateway API) + +`manifests/ate-install/examples/atenet-gateway-agw-gatewayapi.yaml` drives the +same pipeline through the Kubernetes Gateway API instead of a hand-written +config file: + +``` +Client + → agentgateway (Gateway API) + AgentgatewayPolicy.extProc → atenet-router:50051 + resolves actor → rewrites :authority to :80 (or denies) + HTTPRoute → AgentgatewayBackend (dynamicForwardProxy) + re-resolves target from :authority, connects to :80 + → worker pod (ateom) +``` + +Prerequisites: +- agentgateway installed, providing a `GatewayClass` (match the class name to + your install; the manifest uses `agentgateway`). You can follow [this official guide](https://agentgateway.dev/docs/kubernetes/main/quickstart/install/) to install agentgateway. +- All resources live in `ate-system` so the ext_proc `backendRef` and the + route/backend stay in one namespace (no `ReferenceGrant` needed). + +> **Why a gateway-specific config and not one portable Gateway API config?** +> There is no portable Gateway API way to express this pipeline. It needs two +> things the standard CRDs don't expose: permission for ext_proc to rewrite +> `:authority`, and a dynamic-forward-proxy backend. Each proxy that supports +> them does so through its own extension CRDs. + +Validate on kind: +``` +curl -v -H "Host: ..actors.resources.substrate.ate.dev" \ + http:/// +``` diff --git a/hack/install-ate.sh b/hack/install-ate.sh index c4f0880f9..0aec3e982 100755 --- a/hack/install-ate.sh +++ b/hack/install-ate.sh @@ -292,10 +292,16 @@ deploy_ate_system() { manifests="$(render_ate_system_manifests)" echo "${manifests}" | run_kubectl apply -f - + # atenet-router only serves ExtProc gRPC now; deploy the basic reference + # Envoy gateway in front of it so actor traffic has somewhere to land. + # See docs/dev/ingress.md for other (power-user) gateway options. + run_kubectl apply -f manifests/ate-install/examples/atenet-gateway-envoy.yaml + log_step "Waiting for ATE system components to be ready..." run_kubectl rollout status deployment/ate-api-server-deployment -n ate-system --timeout=120s run_kubectl rollout status deployment/ate-controller -n ate-system --timeout=120s run_kubectl rollout status deployment/atenet-router -n ate-system --timeout=120s + run_kubectl rollout status deployment/atenet-gateway -n ate-system --timeout=120s run_kubectl rollout status statefulset/valkey-cluster -n ate-system --timeout=120s run_kubectl rollout status daemonset/atelet -n ate-system --timeout=120s } @@ -360,8 +366,13 @@ deploy_atenet() { run_ko apply -f manifests/ate-install/atenet-router.yaml run_ko apply -f manifests/ate-install/atenet-dns.yaml + # atenet-router only serves ExtProc gRPC now; deploy the basic reference + # Envoy gateway in front of it so actor traffic has somewhere to land. + # See docs/dev/ingress.md for other (power-user) gateway options. + run_kubectl apply -f manifests/ate-install/examples/atenet-gateway-envoy.yaml run_kubectl rollout status deployment/atenet-router -n ate-system --timeout=120s run_kubectl rollout status deployment/atenet-dns -n ate-system --timeout=120s + run_kubectl rollout status deployment/atenet-gateway -n ate-system --timeout=120s } # get_actor_status echoes the actor's status enum (e.g. STATUS_SUSPENDED). @@ -467,12 +478,14 @@ delete_ate_system() { else run_kubectl delete --ignore-not-found -f manifests/ate-install fi + run_kubectl delete --ignore-not-found -f manifests/ate-install/examples/atenet-gateway-envoy.yaml run_kubectl delete --ignore-not-found -f manifests/ate-install/generated } delete_atenet() { log_step "delete_atenet" run_kubectl delete --ignore-not-found -f manifests/ate-install/atenet-router.yaml + run_kubectl delete --ignore-not-found -f manifests/ate-install/examples/atenet-gateway-envoy.yaml } deploy_benchmarks() { diff --git a/hack/poc-pluggable-ingress.sh b/hack/poc-pluggable-ingress.sh new file mode 100755 index 000000000..8f6470f38 --- /dev/null +++ b/hack/poc-pluggable-ingress.sh @@ -0,0 +1,120 @@ +#!/usr/bin/env bash + +# 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. + +# Deploys the basic reference gateway: a standalone external Envoy that calls +# atenet-router's ExtProc endpoint. atenet-router itself is deployed by +# hack/install-ate-kind.sh as part of the normal install (it ships no proxy), +# so this script only needs to add the gateway in front of it. +# +# NOTE: Before running this script, you must have installed the ATE using hack/install-ate-kind.sh +# To test the actors, you can use a demo setup such as hack/install-ate-kind.sh --deploy-demo-counter +# +# For other gateways (agentgateway, Gateway API), see docs/dev/ingress.md and +# manifests/ate-install/examples/ — this script only covers the basic Envoy path. + +set -o errexit -o nounset -o pipefail + +ROOT="$(git rev-parse --show-toplevel)" +cd "${ROOT}" + +PHASE="${1:-deploy}" +NAMESPACE="ate-system" +KO_DOCKER_REPO="${KO_DOCKER_REPO:-localhost:5002}" +KIND_CLUSTER_NAME="${KIND_CLUSTER_NAME:-kind}" +KUBECTL_CONTEXT="${KUBECTL_CONTEXT:-kind-${KIND_CLUSTER_NAME}}" +KUBECTL="kubectl --context=${KUBECTL_CONTEXT}" + +log() { echo "[poc-pluggable-ingress] $*" >&2; } +die() { log "ERROR: $*"; exit 1; } + +deploy() { + log "=== Deploying pluggable ingress (basic Envoy example) ===" + + # atenet-gateway uses a plain Envoy image (no ko:// reference). + log "Applying atenet-gateway-envoy.yaml ..." + ${KUBECTL} apply -f manifests/ate-install/examples/atenet-gateway-envoy.yaml + + log "Waiting for atenet-router to be ready..." + ${KUBECTL} -n "${NAMESPACE}" rollout status deployment/atenet-router --timeout=120s + + log "Waiting for atenet-gateway to be ready..." + ${KUBECTL} -n "${NAMESPACE}" rollout status deployment/atenet-gateway --timeout=120s + + log "Applying atenet-dns.yaml so DNS resolves actor hostnames to atenet-gateway ..." + export KO_DOCKER_REPO + ko apply -f manifests/ate-install/atenet-dns.yaml -- --context="${KUBECTL_CONTEXT}" + + log "Waiting for DNS controller to be ready..." + ${KUBECTL} -n "${NAMESPACE}" rollout status deployment/dns --timeout=120s + + log "" + log "=== Pluggable ingress deployed successfully ===" + log "" + log "1. Start a port-forward to atenet-gateway:" + log " ${KUBECTL} -n ${NAMESPACE} port-forward svc/atenet-gateway 8080:80 &" + log "" + log "2. Pick an actor and its atespace (e.g. from 'kubectl get actors -A'):" + log " ATESPACE= # e.g. team-a" + log " ACTOR_ID= # e.g. 123e4567-e89b-12d3-a456-426614174000" + log "" + log " The actor hostname format is: ..actors.resources.substrate.ate.dev" + log "" + log "3. Send a request with an explicit Host header (bypasses DNS):" + log " curl -v -H \"Host: \${ACTOR_ID}.\${ATESPACE}.actors.resources.substrate.ate.dev\" \\" + log " http://localhost:8080/your/path" + log "" + log "4. Inspect Envoy access logs to confirm ext_proc was called:" + log " ${KUBECTL} -n ${NAMESPACE} logs -l app=atenet-gateway --tail=20" + log "" + log "5. Inspect atenet-router logs:" + log " ${KUBECTL} -n ${NAMESPACE} logs -l app=atenet-router --tail=20" + log "" + log "6. Send a request (inside the cluster) using the actor hostname (DNS should resolve to the gateway):" + log " kubectl run -it --rm --restart=Never curltest --image=curlimages/curl -- \\" + log " curl -v http://\${ACTOR_ID}.\${ATESPACE}.actors.resources.substrate.ate.dev/your/path" +} + +phase_status() { + log "=== Pluggable ingress status ===" + echo "" + echo "--- atenet-router ---" + ${KUBECTL} -n "${NAMESPACE}" get deploy atenet-router -o wide 2>/dev/null || echo " (not deployed)" + echo "" + echo "--- atenet-gateway ---" + ${KUBECTL} -n "${NAMESPACE}" get deploy atenet-gateway -o wide 2>/dev/null || echo " (not deployed)" + echo "" + echo "--- Services ---" + ${KUBECTL} -n "${NAMESPACE}" get svc atenet-router atenet-gateway 2>/dev/null || true + echo "" + echo "--- DNS controller args ---" + ${KUBECTL} -n "${NAMESPACE}" get deploy -l app=dns -o jsonpath='{.items[*].spec.template.spec.containers[0].args}' 2>/dev/null || echo " (DNS controller not found)" +} + +# --------------------------------------------------------------------------- +# teardown: remove the gateway (atenet-router stays; it's part of the base install) +# --------------------------------------------------------------------------- +phase_teardown() { + log "=== Teardown ===" + ${KUBECTL} delete -f manifests/ate-install/examples/atenet-gateway-envoy.yaml --ignore-not-found + log "Gateway removed. atenet-router is still running (it's part of the base install)." +} + +case "${PHASE}" in + status) phase_status ;; + teardown) phase_teardown ;; + deploy) deploy ;; + *) die "Unknown phase: ${PHASE}. Use: status | teardown | deploy" ;; +esac diff --git a/internal/e2e/router_client.go b/internal/e2e/router_client.go index 48f1fcbb9..f9b98033d 100644 --- a/internal/e2e/router_client.go +++ b/internal/e2e/router_client.go @@ -34,12 +34,15 @@ import ( const ( routerNamespace = "ate-system" - routerService = "atenet-router" + // routerService is the gateway Service that fronts atenet-router — the + // router itself only serves ExtProc gRPC, not HTTP. This must match + // whatever gateway hack/install-ate.sh deploys (see docs/dev/ingress.md). + routerService = "atenet-gateway" ) -// RouterClient sends HTTP requests to actors through the atenet router, the +// RouterClient sends HTTP requests to actors through the atenet gateway, the // same way real traffic arrives (so the request is routed and, if needed, the -// actor is resumed). It port-forwards the router Service, mirroring the +// actor is resumed). It port-forwards the gateway Service, mirroring the // approach in internal/ateclient. type RouterClient struct { baseURL string diff --git a/manifests/ate-install/atenet-dns.yaml b/manifests/ate-install/atenet-dns.yaml index 46968a386..0fdf19e0e 100644 --- a/manifests/ate-install/atenet-dns.yaml +++ b/manifests/ate-install/atenet-dns.yaml @@ -149,6 +149,7 @@ spec: - "--log-level=debug" - "--interval=10s" - "--corefile-path=/etc/coredns/Corefile" + - "--ingress-service-name=atenet-gateway" volumeMounts: - name: dns-config-volume mountPath: /etc/coredns diff --git a/manifests/ate-install/atenet-router-monitoring.yaml b/manifests/ate-install/atenet-router-monitoring.yaml deleted file mode 100644 index 6a42ac904..000000000 --- a/manifests/ate-install/atenet-router-monitoring.yaml +++ /dev/null @@ -1,35 +0,0 @@ -# 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. - -# Scrape the Envoy sidecar's admin /stats/prometheus endpoint so its end-to-end -# request-latency histogram (envoy_http_downstream_rq_time, milliseconds) reaches -# Google Managed Prometheus. This is E2E *context* for the per-stage latency -# dashboard, not an SLI we own (the SLI is the OTLP atenet.router.route.duration -# histogram). Envoy only speaks Prometheus, so it needs an explicit scrape; the -# admin port (9901) is already exposed by the envoy container above. -apiVersion: monitoring.googleapis.com/v1 -kind: PodMonitoring -metadata: - name: atenet-router-envoy - namespace: ate-system - labels: - app: atenet-router -spec: - selector: - matchLabels: - app: atenet-router - endpoints: - - port: admin - path: /stats/prometheus - interval: 30s diff --git a/manifests/ate-install/atenet-router.yaml b/manifests/ate-install/atenet-router.yaml index 967f22c29..5fc233cda 100644 --- a/manifests/ate-install/atenet-router.yaml +++ b/manifests/ate-install/atenet-router.yaml @@ -11,7 +11,14 @@ # 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. - +# +# atenet-router: the actor-aware routing core and ExtProc server. +# +# Substrate does not ship or manage a proxy. atenet-router exposes an ext_proc +# gRPC endpoint on port 50051 that any dynamic-forwarding proxy can call to +# resolve actor routes; you bring and configure your own gateway in front of +# it. See docs/dev/ingress.md for reference gateway configs. +--- apiVersion: v1 kind: ServiceAccount metadata: @@ -47,58 +54,6 @@ roleRef: name: atenet-router apiGroup: rbac.authorization.k8s.io --- -apiVersion: v1 -kind: ConfigMap -metadata: - name: atenet-router-envoy-config - namespace: ate-system -data: - envoy.yaml: | - admin: - address: - socket_address: - address: 0.0.0.0 - port_value: 9901 - - node: - id: substrate-envoy-node - cluster: substrate-router-cluster - - dynamic_resources: - lds_config: - resource_api_version: V3 - ads: {} - cds_config: - resource_api_version: V3 - ads: {} - ads_config: - api_type: GRPC - transport_api_version: V3 - grpc_services: - - envoy_grpc: - cluster_name: xds_cluster - - static_resources: - clusters: - - name: xds_cluster - connect_timeout: 0.25s - type: STRICT_DNS - lb_policy: ROUND_ROBIN - typed_extension_protocol_options: - envoy.extensions.upstreams.http.v3.HttpProtocolOptions: - "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions - explicit_http_config: - http2_protocol_options: {} - load_assignment: - cluster_name: xds_cluster - endpoints: - - lb_endpoints: - - endpoint: - address: - socket_address: - address: 127.0.0.1 - port_value: 18000 ---- apiVersion: apps/v1 kind: Deployment metadata: @@ -127,16 +82,11 @@ spec: - "router" - "--standalone" - "--namespace=ate-system" - - "--port-http=8080" - - "--port-xds=18000" - "--port-extproc=50051" - - "--extproc-address=127.0.0.1" + # Bind ExtProc on all interfaces so external proxies can reach it. + - "--extproc-address=0.0.0.0" - "--ateapi-address=api.ate-system.svc:443" - "--status-port=4040" - - "--port-https=8443" - - "--envoy-cert-path=/run/servicedns.podcert.ate.dev/credential-bundle.pem" - - "--otlp-collector-address=opentelemetry-collector.gke-managed-otel.svc.cluster.local:4317" - - "--ateapi-ca-file=/run/servicedns-ca/trust-bundle.pem" env: - name: POD_NAME valueFrom: @@ -155,45 +105,12 @@ spec: - name: OTEL_EXPORTER_OTLP_ENDPOINT value: http://opentelemetry-collector.gke-managed-otel.svc.cluster.local:4317 ports: - - name: xds - containerPort: 18000 - name: extproc containerPort: 50051 - name: status containerPort: 4040 - name: metrics containerPort: 9090 - - name: envoy - image: envoyproxy/envoy:v1.30-latest - command: - - "/usr/local/bin/envoy" - - "-c" - - "/etc/envoy/envoy.yaml" - - "--component-log-level" - - "upstream:debug,router:debug,ext_proc:debug" - ports: - - name: http - containerPort: 8080 - - name: https - containerPort: 8443 - - name: admin - containerPort: 9901 - volumeMounts: - - name: envoy-config - mountPath: /etc/envoy - - name: "servicedns" - mountPath: "/run/servicedns.podcert.ate.dev" - volumes: - - name: envoy-config - configMap: - name: atenet-router-envoy-config - - name: "servicedns" - projected: - sources: - - podCertificate: - signerName: servicedns.podcert.ate.dev/identity - keyType: ECDSAP256 - credentialBundlePath: credential-bundle.pem --- apiVersion: v1 kind: Service @@ -205,11 +122,12 @@ spec: selector: app: atenet-router ports: - - name: http - port: 80 - targetPort: 8080 + # ExtProc port exposed cluster-internally so an external proxy can call it. + - name: extproc + port: 50051 + targetPort: 50051 protocol: TCP - - name: https - port: 443 - targetPort: 8443 + - name: status + port: 4040 + targetPort: 4040 protocol: TCP diff --git a/manifests/ate-install/examples/atenet-gateway-agw-gatewayapi.yaml b/manifests/ate-install/examples/atenet-gateway-agw-gatewayapi.yaml new file mode 100644 index 000000000..ee626dced --- /dev/null +++ b/manifests/ate-install/examples/atenet-gateway-agw-gatewayapi.yaml @@ -0,0 +1,62 @@ +apiVersion: gateway.networking.k8s.io/v1 +kind: Gateway +metadata: + name: atenet-gateway + namespace: ate-system +spec: + gatewayClassName: agentgateway + listeners: + - name: http + protocol: HTTP + port: 80 + allowedRoutes: + namespaces: + from: Same +--- +apiVersion: agentgateway.dev/v1alpha1 +kind: AgentgatewayPolicy +metadata: + name: atenet-extproc + namespace: ate-system +spec: + targetRefs: + - group: gateway.networking.k8s.io + kind: Gateway + name: atenet-gateway + traffic: + extProc: + backendRef: + name: atenet-router + namespace: ate-system + port: 50051 + processingOptions: + requestHeaderMode: Send + responseHeaderMode: Skip + requestBodyMode: None + responseBodyMode: None + requestTrailerMode: Skip + responseTrailerMode: Skip +--- +apiVersion: agentgateway.dev/v1alpha1 +kind: AgentgatewayBackend +metadata: + name: atenet-dfp + namespace: ate-system +spec: + dynamicForwardProxy: {} +--- +apiVersion: gateway.networking.k8s.io/v1 +kind: HTTPRoute +metadata: + name: atenet-actors + namespace: ate-system +spec: + parentRefs: + - name: atenet-gateway + hostnames: + - "*.actors.resources.substrate.ate.dev" + rules: + - backendRefs: + - name: atenet-dfp + group: agentgateway.dev + kind: AgentgatewayBackend \ No newline at end of file diff --git a/manifests/ate-install/examples/atenet-gateway-agw.yaml b/manifests/ate-install/examples/atenet-gateway-agw.yaml new file mode 100644 index 000000000..741715ee4 --- /dev/null +++ b/manifests/ate-install/examples/atenet-gateway-agw.yaml @@ -0,0 +1,123 @@ +# 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. +# +# atenet-gateway (agentgateway variant) — reference external proxy, power-user +# alternative to atenet-gateway-envoy.yaml. +# +# This is the agentgateway equivalent of atenet-gateway-envoy.yaml (Envoy). It is +# a standalone agentgateway (https://agentgateway.dev) configured by a +# hand-written config file — NOT via the Kubernetes Gateway API / a gateway +# controller. All actor-aware routing is delegated to atenet-router via the +# ext_proc gRPC filter, exactly as in the Envoy variant. +# +# Why direct config and not Gateway API: the standard Gateway API policy CRDs +# (EnvoyExtensionPolicy / AgentgatewayPolicy) do not expose the two knobs this +# pipeline needs — permission for ext_proc to rewrite :authority, and a dynamic +# forward proxy backend. Authoring the proxy config directly is the only way to +# combine them without raw-xDS patches or the Gateway API Inference Extension. +# +# Deploy INSTEAD OF atenet-gateway-envoy.yaml (they share the atenet-gateway +# Service name), alongside manifests/ate-install/atenet-router.yaml. DNS must +# resolve actor hostnames to the atenet-gateway Service ClusterIP — see +# docs/dev/ingress.md. +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: atenet-gateway-agentgateway-config + namespace: ate-system +data: + config.yaml: | + # yaml-language-server: $schema=https://agentgateway.dev/schema/config + binds: + - port: 8080 + listeners: + - name: ingress + protocol: HTTP + routes: + # No `matches`/`hostnames` => matches every request (any host, any path), + # mirroring the Envoy variant's domains:["*"] prefix:"/" route. + - name: actors + policies: + # Step 1: call atenet-router ext_proc to resolve actor -> worker IP. + # On success it rewrites :authority to :80; on denial it + # returns an ImmediateResponse (404/503/etc), which agentgateway honors. + extProc: + host: "atenet-router.ate-system.svc:50051" + failureMode: failClosed + processingOptions: + requestHeaderMode: send + responseHeaderMode: skip + requestBodyMode: none # headers-only; do not stream the body + responseBodyMode: none + requestTrailerMode: skip + responseTrailerMode: skip + backends: + # Step 2: dynamic forward proxy. agentgateway re-resolves the upstream + # target from the :authority the ext_proc set, and connects to that + # :80 directly. + - dynamic: {} +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: atenet-gateway + namespace: ate-system + labels: + app: atenet-gateway +spec: + replicas: 1 + selector: + matchLabels: + app: atenet-gateway + template: + metadata: + labels: + app: atenet-gateway + spec: + containers: + - name: agentgateway + image: ghcr.io/agentgateway/agentgateway:latest + args: + - "-f" + - "/etc/agentgateway/config.yaml" + ports: + - name: http + containerPort: 8080 + - name: admin + containerPort: 15000 + volumeMounts: + - name: config + mountPath: /etc/agentgateway + volumes: + - name: config + configMap: + name: atenet-gateway-agentgateway-config +--- +apiVersion: v1 +kind: Service +metadata: + name: atenet-gateway + namespace: ate-system + labels: + app: atenet-gateway +spec: + type: ClusterIP + selector: + app: atenet-gateway + ports: + - name: http + port: 80 + targetPort: 8080 + protocol: TCP diff --git a/manifests/ate-install/examples/atenet-gateway-envoy.yaml b/manifests/ate-install/examples/atenet-gateway-envoy.yaml new file mode 100644 index 000000000..51b384485 --- /dev/null +++ b/manifests/ate-install/examples/atenet-gateway-envoy.yaml @@ -0,0 +1,200 @@ +# 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. +# +# atenet-gateway — a reference external Envoy proxy that calls atenet-router's +# ExtProc endpoint for actor-aware routing. +# +# This is a standalone Envoy deployment; Substrate does not generate or manage +# it. It is a reference config you apply yourself, alongside +# manifests/ate-install/atenet-router.yaml. DNS must be repointed to the +# atenet-gateway Service ClusterIP — see docs/dev/ingress.md. +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: atenet-gateway-envoy-config + namespace: ate-system +data: + envoy.yaml: | + admin: + address: + socket_address: + address: 0.0.0.0 + port_value: 9901 + + node: + id: atenet-gateway-node + cluster: atenet-gateway-cluster + + static_resources: + listeners: + - name: ingress_http_listener + address: + socket_address: + address: 0.0.0.0 + port_value: 8080 + filter_chains: + - filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: ingress_http + generate_request_id: true + access_log: + - name: envoy.access_loggers.stdout + typed_config: + "@type": type.googleapis.com/envoy.extensions.access_loggers.stream.v3.StdoutAccessLog + http_filters: + # Step 1: call atenet-router ExtProc to resolve actor → worker IP. + # On success ExtProc rewrites :authority to :80. + # On denial ExtProc returns an immediate HTTP response (404/503/etc). + - name: envoy.filters.http.ext_proc + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor + grpc_service: + envoy_grpc: + cluster_name: extproc_cluster + timeout: 5s + mutation_rules: + allow_all_routing: true + message_timeout: 5s + processing_mode: + request_header_mode: SEND + response_header_mode: SKIP + request_body_mode: NONE + response_body_mode: NONE + request_trailer_mode: SKIP + response_trailer_mode: SKIP + # Step 2: dynamic_forward_proxy resolves the :authority set by ExtProc + # (now :80) to the actual upstream endpoint. + - name: envoy.filters.http.dynamic_forward_proxy + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.dynamic_forward_proxy.v3.FilterConfig + dns_cache_config: + name: dynamic_forward_proxy_cache_config + dns_lookup_family: V4_ONLY + typed_dns_resolver_config: + name: envoy.network.dns_resolver.getaddrinfo + typed_config: + "@type": type.googleapis.com/envoy.extensions.network.dns_resolver.getaddrinfo.v3.GetAddrInfoDnsResolverConfig + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + route_config: + name: local_route + virtual_hosts: + - name: local_service + domains: ["*"] + routes: + - match: + prefix: "/" + route: + cluster: dynamic_forward_proxy_cluster + timeout: 10s + + clusters: + # ExtProc cluster: gRPC/H2 to atenet-router's ExtProc port. + # STRICT_DNS so the address resolves through cluster DNS. + - name: extproc_cluster + connect_timeout: 0.25s + type: STRICT_DNS + lb_policy: ROUND_ROBIN + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} + load_assignment: + cluster_name: extproc_cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + # atenet-router binds ExtProc on 0.0.0.0 by default + # (see manifests/ate-install/atenet-router.yaml). + address: atenet-router.ate-system.svc + port_value: 50051 + + # Dynamic forward proxy cluster: Envoy resolves :authority at request time + # to reach the worker pod IP rewritten by ExtProc. + - name: dynamic_forward_proxy_cluster + lb_policy: CLUSTER_PROVIDED + cluster_type: + name: envoy.clusters.dynamic_forward_proxy + typed_config: + "@type": type.googleapis.com/envoy.extensions.clusters.dynamic_forward_proxy.v3.ClusterConfig + dns_cache_config: + name: dynamic_forward_proxy_cache_config + dns_lookup_family: V4_ONLY + typed_dns_resolver_config: + name: envoy.network.dns_resolver.getaddrinfo + typed_config: + "@type": type.googleapis.com/envoy.extensions.network.dns_resolver.getaddrinfo.v3.GetAddrInfoDnsResolverConfig +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: atenet-gateway + namespace: ate-system + labels: + app: atenet-gateway +spec: + replicas: 1 + selector: + matchLabels: + app: atenet-gateway + template: + metadata: + labels: + app: atenet-gateway + spec: + containers: + - name: envoy + image: envoyproxy/envoy:v1.30-latest + command: + - "/usr/local/bin/envoy" + - "-c" + - "/etc/envoy/envoy.yaml" + - "--component-log-level" + - "upstream:debug,router:debug,ext_proc:debug" + ports: + - name: http + containerPort: 8080 + - name: admin + containerPort: 9901 + volumeMounts: + - name: envoy-config + mountPath: /etc/envoy + volumes: + - name: envoy-config + configMap: + name: atenet-gateway-envoy-config +--- +apiVersion: v1 +kind: Service +metadata: + name: atenet-gateway + namespace: ate-system + labels: + app: atenet-gateway +spec: + type: ClusterIP + selector: + app: atenet-gateway + ports: + - name: http + port: 80 + targetPort: 8080 + protocol: TCP