diff --git a/cmd/atenet/internal/router/README.md b/cmd/atenet/internal/router/README.md index 71854a86a..eebdefc0c 100644 --- a/cmd/atenet/internal/router/README.md +++ b/cmd/atenet/internal/router/README.md @@ -2,10 +2,12 @@ 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 +* Serves Envoy xDS configuration when `--atenet-router=envoy` (the default). + Unless `--standalone` is set, it also manages the Envoy Deployment and + Services in Kubernetes. + With `--atenet-router=agentgateway`, the sidecar uses a static ConfigMap and + atenet does not start an xDS server. +* ext_proc server for the proxy. 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 @@ -27,4 +29,4 @@ Contents: * Global flags values * Command line args * Last 100 queries served -* Build tag \ No newline at end of file +* Build tag diff --git a/cmd/atenet/internal/router/cmd.go b/cmd/atenet/internal/router/cmd.go index 10f6563fd..233f717ee 100644 --- a/cmd/atenet/internal/router/cmd.go +++ b/cmd/atenet/internal/router/cmd.go @@ -42,6 +42,7 @@ func NewRouterCmd() *cobra.Command { cmd.Flags().StringVar(&cfg.LogLevel, "log-level", "info", "Log level: debug, info, warn, error") cmd.Flags().StringVar(&cfg.MetricsAddr, "metrics-listen-addr", ":9090", "Address and port the prometheus metrics server should listen on.") cmd.Flags().BoolVar(&cfg.Standalone, "standalone", false, "Run in standalone mode, bypassing creation of managed deployment and services in Kubernetes cluster") + cmd.Flags().StringVar(&cfg.AtenetRouter, "atenet-router", string(atenetRouterEnvoy), "Router dataplane: envoy or agentgateway") 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", "dns:///api.ate-system.svc:443", "gRPC dial target for the cluster ateapi Control instance.") diff --git a/cmd/atenet/internal/router/config.go b/cmd/atenet/internal/router/config.go index 08fef3f4e..7b784f8e4 100644 --- a/cmd/atenet/internal/router/config.go +++ b/cmd/atenet/internal/router/config.go @@ -19,6 +19,13 @@ import ( "time" ) +type atenetRouter string + +const ( + atenetRouterEnvoy atenetRouter = "envoy" + atenetRouterAgentgateway atenetRouter = "agentgateway" +) + // authConfig holds the router's client-auth settings for dialing ateapi. // AteapiCAFile always verifies ateapi's serving cert (the servicedns trust // bundle in-cluster). By default the router presents AteapiClientCertPath @@ -36,6 +43,7 @@ type authConfig struct { // routerConfig holds deployment setup and endpoint options for the router node instance. type routerConfig struct { Standalone bool + AtenetRouter string Namespace string Kubeconfig string AteapiAddr string @@ -86,6 +94,13 @@ type routerConfig struct { ExtProcMaxRequests int } +func (c routerConfig) atenetRouter() atenetRouter { + if c.AtenetRouter == "" { + return atenetRouterEnvoy + } + return atenetRouter(c.AtenetRouter) +} + // extProcMaxRequestsFloor is the minimum derived circuit breaker — Envoy's own // default max_requests — so a small (or disabled) parking lot still leaves // ordinary fast-path capacity. @@ -109,9 +124,15 @@ func (c routerConfig) extProcMaxRequests() int { // validate rejects flag combinations that would make the router misbehave // rather than merely differ. func (c routerConfig) validate() error { + switch c.atenetRouter() { + case atenetRouterEnvoy, atenetRouterAgentgateway: + default: + return fmt.Errorf("--atenet-router must be %q or %q, got %q", atenetRouterEnvoy, atenetRouterAgentgateway, c.AtenetRouter) + } if err := c.ParkedRequest.validate(); err != nil { return err } + if c.ExtProcMaxRequests < 0 { return fmt.Errorf("--extproc-max-requests must not be negative, got %d (0 derives it from --parked-request-max)", c.ExtProcMaxRequests) } diff --git a/cmd/atenet/internal/router/config_test.go b/cmd/atenet/internal/router/config_test.go index 14ecaf48c..85de49d9d 100644 --- a/cmd/atenet/internal/router/config_test.go +++ b/cmd/atenet/internal/router/config_test.go @@ -26,9 +26,22 @@ func TestRouterConfigValidate(t *testing.T) { wantErr string // substring; empty means valid }{ { - name: "defaults are valid (auto breaker)", + name: "atenet-router defaults to envoy", cfg: routerConfig{ExtProcMaxRequests: 0, ParkedRequest: ParkedRequestConfig{Max: defaultParkedRequestMax}}, }, + { + name: "atenet-router set to envoy is valid", + cfg: routerConfig{AtenetRouter: string(atenetRouterEnvoy), ParkedRequest: ParkedRequestConfig{Max: defaultParkedRequestMax}}, + }, + { + name: "atenet-router set to agentgateway is valid", + cfg: routerConfig{AtenetRouter: string(atenetRouterAgentgateway), ParkedRequest: ParkedRequestConfig{Max: defaultParkedRequestMax}}, + }, + { + name: "unknown router rejected", + cfg: routerConfig{AtenetRouter: "blah"}, + wantErr: "--atenet-router must be", + }, { name: "negative extproc-max-requests rejected", cfg: routerConfig{ExtProcMaxRequests: -1, ParkedRequest: ParkedRequestConfig{Max: 0}}, @@ -64,6 +77,25 @@ func TestRouterConfigValidate(t *testing.T) { } } +func TestRouterConfigAtenetRouter(t *testing.T) { + tests := []struct { + name string + cfg routerConfig + want atenetRouter + }{ + {name: "default", cfg: routerConfig{}, want: atenetRouterEnvoy}, + {name: "explicit envoy", cfg: routerConfig{AtenetRouter: string(atenetRouterEnvoy)}, want: atenetRouterEnvoy}, + {name: "agentgateway", cfg: routerConfig{AtenetRouter: string(atenetRouterAgentgateway)}, want: atenetRouterAgentgateway}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := tc.cfg.atenetRouter(); got != tc.want { + t.Fatalf("atenetRouter() = %q, want %q", got, tc.want) + } + }) + } +} + func TestRouterConfigExtProcMaxRequests(t *testing.T) { tests := []struct { name string diff --git a/cmd/atenet/internal/router/dashboard.html b/cmd/atenet/internal/router/dashboard.html index 6d5c170a2..004447124 100644 --- a/cmd/atenet/internal/router/dashboard.html +++ b/cmd/atenet/internal/router/dashboard.html @@ -257,7 +257,7 @@

atenet Router Status

Component Network Allocation
- Workload Port (Http Envoy) + Workload Port (HTTP Dataplane) {{ .HttpPort }}
@@ -298,8 +298,8 @@

atenet Router Status

System Component Health Checks
diff --git a/cmd/atenet/internal/router/dataplane.go b/cmd/atenet/internal/router/dataplane.go new file mode 100644 index 000000000..7af7bbcca --- /dev/null +++ b/cmd/atenet/internal/router/dataplane.go @@ -0,0 +1,90 @@ +// 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" + "time" + + "golang.org/x/sync/errgroup" +) + +type dataplaneHealthCheck struct { + url string + expectedBody string +} + +func (r atenetRouter) routeViaAuthority() bool { + return r == atenetRouterAgentgateway +} + +func (r atenetRouter) healthCheck() dataplaneHealthCheck { + switch r { + case atenetRouterEnvoy: + return dataplaneHealthCheck{url: "http://127.0.0.1:9901/ready", expectedBody: "LIVE"} + case atenetRouterAgentgateway: + return dataplaneHealthCheck{url: "http://127.0.0.1:15021/healthz/ready", expectedBody: "ready"} + default: + return dataplaneHealthCheck{} + } +} + +func (s *RouterServer) startDataplane(ctx context.Context, g *errgroup.Group, parkCfg ParkedRequestConfig) error { + switch s.cfg.atenetRouter() { + case atenetRouterEnvoy: + s.startEnvoyDataplane(ctx, g, parkCfg) + case atenetRouterAgentgateway: + // Agentgateway receives all routing configuration from its static file. + default: + return fmt.Errorf("unsupported atenet router %q", s.cfg.atenetRouter()) + } + return nil +} + +func (s *RouterServer) startEnvoyDataplane(ctx context.Context, g *errgroup.Group, parkCfg ParkedRequestConfig) { + xdsSrv := NewXdsServer(s.cfg.XdsPort) + xdsSrv.SetConfig(s.cfg.HttpPort, s.cfg.ExtprocPort, s.cfg.ExtprocAddr) + setOtlpCollector(ctx, xdsSrv, s.cfg.OtlpCollectorAddress) + + xdsSrv.SetExtProcMaxRequests(s.cfg.extProcMaxRequests()) + if parkCfg.enabled() { + // Envoy must keep a parked request open at least as long as the router + // will hold it; add a margin so the router surfaces its own 503 first. + xdsSrv.SetExtProcMessageTimeout(parkCfg.Budget + 5*time.Second) + } + + xdsSrv.SetTlsConfig(s.cfg.HttpsPort, s.cfg.EnvoyCertPath) + xdsSrv.SetUpstreamTls(s.cfg.UpstreamCredentialBundlePath, s.cfg.UpstreamTrustBundlePath, s.cfg.UpstreamSpiffePrefix) + ctrl := NewController(s.k8sClient, s.clientset, s.cfg, xdsSrv, s.extprocSrv) + + // Envoy receives all routing configuration from the local xDS server. + g.Go(func() error { + slog.InfoContext(ctx, "Starting ActorTemplate controller") + return ctrl.Start(ctx) + }) + 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) + }) +} diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index 795afd583..96cc22238 100644 --- a/cmd/atenet/internal/router/extproc.go +++ b/cmd/atenet/internal/router/extproc.go @@ -40,22 +40,24 @@ import ( // ExtProcServer implements the Envoy external processing gRPC server // to dynamically manage actor activations based on request traffic. type ExtProcServer struct { - port int - apiClient ateapipb.ControlClient - recorder *QueryRecorder - resumer *ActorResumer - routeDuration metric.Float64Histogram - parking *parkingLot + port int + apiClient ateapipb.ControlClient + recorder *QueryRecorder + resumer *ActorResumer + routeDuration metric.Float64Histogram + parking *parkingLot + routeViaAuthority bool } -func NewExtProcServer(port int, apiClient ateapipb.ControlClient, routeDuration metric.Float64Histogram, parkCfg ParkedRequestConfig, parkMetrics *parkingMetrics) *ExtProcServer { +func NewExtProcServer(port int, apiClient ateapipb.ControlClient, routeDuration metric.Float64Histogram, parkCfg ParkedRequestConfig, parkMetrics *parkingMetrics, routeViaAuthority bool) *ExtProcServer { return &ExtProcServer{ - port: port, - apiClient: apiClient, - recorder: NewQueryRecorder(100), - resumer: NewActorResumer(apiClient, withParking(parkCfg)), - routeDuration: routeDuration, - parking: newParkingLot(parkCfg, parkMetrics), + port: port, + apiClient: apiClient, + recorder: NewQueryRecorder(100), + resumer: NewActorResumer(apiClient, withParking(parkCfg)), + routeDuration: routeDuration, + parking: newParkingLot(parkCfg, parkMetrics), + routeViaAuthority: routeViaAuthority, } } @@ -202,7 +204,7 @@ func (s *ExtProcServer) handleRequestHeaders( // dial, without touching :authority — atunnel authorizes the actor by the // original Host (actor DNS name). mutation := &extprocv3.HeaderMutation{} - addOriginalDstMutation(targetAddr, mutation) + addRoutingMutations(targetAddr, metadata.host, s.routeViaAuthority, mutation) return &extprocv3.HeadersResponse{ Response: &extprocv3.CommonResponse{ diff --git a/cmd/atenet/internal/router/extproc_in.go b/cmd/atenet/internal/router/extproc_in.go index 459b26b92..6ab2d2a6d 100644 --- a/cmd/atenet/internal/router/extproc_in.go +++ b/cmd/atenet/internal/router/extproc_in.go @@ -22,6 +22,8 @@ import ( corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" ) +const authorityHeader = ":authority" + type requestMetadata struct { headers map[string]string path string @@ -44,7 +46,7 @@ func newRequestMetadata(headers []*corev3.HeaderValue) *requestMetadata { if k == ":path" { path = val } - if k == ":authority" || k == "host" { + if k == authorityHeader || k == "host" { host = val } } diff --git a/cmd/atenet/internal/router/extproc_out.go b/cmd/atenet/internal/router/extproc_out.go index 964a1a4f7..a6759daf6 100644 --- a/cmd/atenet/internal/router/extproc_out.go +++ b/cmd/atenet/internal/router/extproc_out.go @@ -15,6 +15,7 @@ package router import ( + "github.com/agent-substrate/substrate/internal/atunnel" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" extproc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3" @@ -53,6 +54,31 @@ func addOriginalDstMutation(dst string, mut *extproc.HeaderMutation) { ) } +// addRoutingMutations overwrites all routing metadata derived from the +// control-plane result. Envoy dials OriginalDstHeader while preserving +// :authority. Agentgateway v1.4.1's static dynamic backend instead dials the +// request :authority, so that mode rewrites it to the worker atunnel address. +// OriginalHostHeader lets atunnel restore and authorize the actor authority. +func addRoutingMutations(dst, actorHost string, routeViaAuthority bool, mut *extproc.HeaderMutation) { + addOriginalDstMutation(dst, mut) + mut.SetHeaders = append(mut.SetHeaders, &corev3.HeaderValueOption{ + AppendAction: corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD, + Header: &corev3.HeaderValue{ + Key: atunnel.OriginalHostHeader, + RawValue: []byte(actorHost), + }, + }) + if routeViaAuthority { + mut.SetHeaders = append(mut.SetHeaders, &corev3.HeaderValueOption{ + AppendAction: corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD, + Header: &corev3.HeaderValue{ + Key: authorityHeader, + RawValue: []byte(dst), + }, + }) + } +} + func immediateResponse(statusCode envoy_type.StatusCode, message string) *extproc.ProcessingResponse { return &extproc.ProcessingResponse{ Response: &extproc.ProcessingResponse_ImmediateResponse{ diff --git a/cmd/atenet/internal/router/extproc_test.go b/cmd/atenet/internal/router/extproc_test.go index ab08d0c59..4c9f12466 100644 --- a/cmd/atenet/internal/router/extproc_test.go +++ b/cmd/atenet/internal/router/extproc_test.go @@ -24,6 +24,7 @@ import ( "testing" "time" + "github.com/agent-substrate/substrate/internal/atunnel" "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" @@ -58,7 +59,7 @@ func TestHandleRequestHeadersDoesNotLogSensitiveData(t *testing.T) { 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, ParkedRequestConfig{}, nil) + }, nil, ParkedRequestConfig{}, nil, false) reqHeaders := &extprocv3.HttpHeaders{ Headers: &corev3.HeaderMap{ @@ -195,7 +196,7 @@ func TestExtProcHeadersEvaluation(t *testing.T) { // Parking disabled: these cases assert fail-fast mapping of resume // errors (e.g. FailedPrecondition -> immediate 503). Parking behavior // is covered separately in TestExtProc_ParkingLotFull and resumer_test.go. - s := NewExtProcServer(50051, clientMock, nil, ParkedRequestConfig{}, nil) + s := NewExtProcServer(50051, clientMock, nil, ParkedRequestConfig{}, nil, false) reqHeaders := &extprocv3.HttpHeaders{ Headers: &corev3.HeaderMap{ @@ -236,17 +237,19 @@ func TestExtProcHeadersEvaluation(t *testing.T) { } mutation := res.Response.GetHeaderMutation() - if len(mutation.GetSetHeaders()) != 1 { - t.Fatalf("expected exactly one Header option set, found: %v", mutation.GetSetHeaders()) + if len(mutation.GetSetHeaders()) != 2 { + t.Fatalf("expected exactly two header options, found: %v", mutation.GetSetHeaders()) } - headerOption := mutation.GetSetHeaders()[0] - if strings.ToLower(headerOption.Header.Key) != OriginalDstHeader { - t.Errorf("invalid resulting dynamic parameter key: %s", headerOption.Header.Key) + gotMutations := map[string]string{} + for _, headerOption := range mutation.GetSetHeaders() { + gotMutations[strings.ToLower(headerOption.Header.Key)] = string(headerOption.Header.RawValue) } - - if string(headerOption.Header.RawValue) != tc.expectedTarget { - t.Errorf("invalid destination mapping found: %s, expected: %s", headerOption.Header.RawValue, tc.expectedTarget) + if got := gotMutations[OriginalDstHeader]; got != tc.expectedTarget { + t.Errorf("destination mutation = %q, want %q", got, tc.expectedTarget) + } + if got := gotMutations[strings.ToLower(atunnel.OriginalHostHeader)]; got != tc.authority { + t.Errorf("original host mutation = %q, want %q", got, tc.authority) } // Confirm that query logs recorded metric trace details @@ -274,7 +277,7 @@ func TestExtProc_ParkingLotFull(t *testing.T) { // A 1-slot lot with the slot already occupied deterministically simulates a // full lot without needing a concurrent in-flight request. - s := NewExtProcServer(50051, clientMock, nil, ParkedRequestConfig{Budget: time.Second, Max: 1}, nil) + s := NewExtProcServer(50051, clientMock, nil, ParkedRequestConfig{Budget: time.Second, Max: 1}, nil, false) release, ok := s.parking.enter(context.Background()) if !ok { t.Fatal("priming enter should be admitted") @@ -393,7 +396,7 @@ func TestRecordRouteDuration_Attributes(t *testing.T) { t.Fatalf("failed to create histogram: %v", err) } - s := NewExtProcServer(50051, nil, h, ParkedRequestConfig{}, nil) + s := NewExtProcServer(50051, nil, h, ParkedRequestConfig{}, nil, false) s.recordRouteDuration(context.Background(), 10*time.Millisecond, "team-a-ns", "tmpl-a", classifyOutcome(nil), string(ResumeOutcomeTriggered)) var rm metricdata.ResourceMetrics @@ -418,3 +421,25 @@ func TestRecordRouteDuration_Attributes(t *testing.T) { } } } + +func TestAddRoutingMutationsViaAuthority(t *testing.T) { + mutation := &extprocv3.HeaderMutation{} + addRoutingMutations("10.0.0.52:443", "actor-1.team-a.actors.resources.substrate.ate.dev", true, mutation) + + got := map[string]string{} + for _, option := range mutation.GetSetHeaders() { + if option.GetAppendAction() != corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD { + t.Errorf("mutation %q append action = %v, want overwrite", option.GetHeader().GetKey(), option.GetAppendAction()) + } + got[strings.ToLower(option.GetHeader().GetKey())] = string(option.GetHeader().GetRawValue()) + } + if got[OriginalDstHeader] != "10.0.0.52:443" { + t.Errorf("%s = %q", OriginalDstHeader, got[OriginalDstHeader]) + } + if got[strings.ToLower(atunnel.OriginalHostHeader)] != "actor-1.team-a.actors.resources.substrate.ate.dev" { + t.Errorf("%s = %q", atunnel.OriginalHostHeader, got[strings.ToLower(atunnel.OriginalHostHeader)]) + } + if got[authorityHeader] != "10.0.0.52:443" { + t.Errorf("%s = %q", authorityHeader, got[authorityHeader]) + } +} diff --git a/cmd/atenet/internal/router/health.go b/cmd/atenet/internal/router/health.go index 7d735e2e6..49ee002a0 100644 --- a/cmd/atenet/internal/router/health.go +++ b/cmd/atenet/internal/router/health.go @@ -43,9 +43,9 @@ type ComponentHealth struct { } type RouterHealthReport struct { - Envoy ComponentHealth `json:"envoy"` - K8sAPI ComponentHealth `json:"k8s_api"` - AteAPI ComponentHealth `json:"ate_api"` + Dataplane ComponentHealth `json:"dataplane"` + K8sAPI ComponentHealth `json:"k8s_api"` + AteAPI ComponentHealth `json:"ate_api"` } type componentHealthCheckResult struct { @@ -61,11 +61,11 @@ 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 + dataplaneClient *http.Client } func newRouterHealth(interval time.Duration, clientset kubernetes.Interface, apiClient ateapipb.ControlClient, cfg routerConfig) *routerHealth { @@ -73,11 +73,11 @@ 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, + dataplaneClient: &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}, } } @@ -104,12 +104,12 @@ func (rh *routerHealth) check(ctx context.Context) { // Run network checks concurrently and without holding the report mutex, so // the cycle is bounded by the slowest dependency and status requests can // continue serving the last completed report. - var envoyResult, k8sResult, ateResult componentHealthCheckResult + var dataplaneResult, k8sResult, ateResult componentHealthCheckResult var wg sync.WaitGroup wg.Add(3) go func() { defer wg.Done() - envoyResult = runComponentHealthCheck(ctx, "Envoy health check failed", rh.checkEnvoy) + dataplaneResult = runComponentHealthCheck(ctx, "Router dataplane health check failed", rh.checkDataplane) }() go func() { defer wg.Done() @@ -123,7 +123,7 @@ func (rh *routerHealth) check(ctx context.Context) { rh.mu.Lock() defer rh.mu.Unlock() - updateComponentHealth(&rh.report.Envoy, envoyResult.healthy, envoyResult.message, envoyResult.checkedAt) + updateComponentHealth(&rh.report.Dataplane, dataplaneResult.healthy, dataplaneResult.message, dataplaneResult.checkedAt) updateComponentHealth(&rh.report.K8sAPI, k8sResult.healthy, k8sResult.message, k8sResult.checkedAt) updateComponentHealth(&rh.report.AteAPI, ateResult.healthy, ateResult.message, ateResult.checkedAt) } @@ -156,16 +156,17 @@ func updateComponentHealth(health *ComponentHealth, healthy bool, msg string, ch } } -func (rh *routerHealth) checkEnvoy(ctx context.Context) (bool, string) { +func (rh *routerHealth) checkDataplane(ctx context.Context) (bool, string) { timeoutCtx, cancel := context.WithTimeout(ctx, dependencyHealthCheckTimeout) defer cancel() - req, err := http.NewRequestWithContext(timeoutCtx, "GET", "http://127.0.0.1:9901/ready", nil) + check := rh.cfg.atenetRouter().healthCheck() + req, err := http.NewRequestWithContext(timeoutCtx, "GET", check.url, nil) if err != nil { return false, err.Error() } - resp, err := rh.envoyClient.Do(req) + resp, err := rh.dataplaneClient.Do(req) if err != nil { return false, err.Error() } @@ -181,11 +182,11 @@ func (rh *routerHealth) checkEnvoy(ctx context.Context) (bool, string) { } bodyStr := strings.TrimSpace(string(bodyBytes)) - if bodyStr != "LIVE" { - return false, fmt.Sprintf("expected LIVE but got %q", bodyStr) + if bodyStr != check.expectedBody { + return false, fmt.Sprintf("expected %s but got %q", check.expectedBody, bodyStr) } - return true, "LIVE" + return true, check.expectedBody } func (rh *routerHealth) checkK8s(ctx context.Context) (bool, string) { diff --git a/cmd/atenet/internal/router/health_test.go b/cmd/atenet/internal/router/health_test.go index ef91a5158..e1b4c0393 100644 --- a/cmd/atenet/internal/router/health_test.go +++ b/cmd/atenet/internal/router/health_test.go @@ -59,8 +59,8 @@ func newHealthTestClientset(t *testing.T, server *httptest.Server) kubernetes.In return clientset } -func setHealthyEnvoyClient(rh *routerHealth) { - rh.envoyClient = &http.Client{Transport: healthRoundTripFunc(func(*http.Request) (*http.Response, error) { +func setHealthyDataplaneClient(rh *routerHealth) { + rh.dataplaneClient = &http.Client{Transport: healthRoundTripFunc(func(*http.Request) (*http.Response, error) { return &http.Response{ StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader("LIVE")), @@ -68,6 +68,50 @@ func setHealthyEnvoyClient(rh *routerHealth) { })} } +func TestCheckDataplane(t *testing.T) { + tests := []struct { + name string + router atenetRouter + wantURL string + response string + wantMessage string + }{ + { + name: "envoy", + router: atenetRouterEnvoy, + wantURL: "http://127.0.0.1:9901/ready", + response: "LIVE", + wantMessage: "LIVE", + }, + { + name: "agentgateway", + router: atenetRouterAgentgateway, + wantURL: "http://127.0.0.1:15021/healthz/ready", + response: "ready\n", + wantMessage: "ready", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + rh := newRouterHealth(time.Second, nil, nil, routerConfig{AtenetRouter: string(tc.router)}) + rh.dataplaneClient = &http.Client{Transport: healthRoundTripFunc(func(req *http.Request) (*http.Response, error) { + if req.URL.String() != tc.wantURL { + t.Errorf("health URL = %q, want %q", req.URL.String(), tc.wantURL) + } + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(tc.response)), + }, nil + })} + + healthy, message := rh.checkDataplane(context.Background()) + if !healthy || message != tc.wantMessage { + t.Errorf("checkDataplane() = (%v, %q), want (true, %q)", healthy, message, tc.wantMessage) + } + }) + } +} + func TestCheckK8sTimesOut(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) { <-req.Context().Done() @@ -120,7 +164,7 @@ func TestHealthCheckDoesNotBlockReportOrStatusz(t *testing.T) { defer server.Close() rh := newRouterHealth(time.Second, newHealthTestClientset(t, server), nil, routerConfig{}) - setHealthyEnvoyClient(rh) + setHealthyDataplaneClient(rh) checkDone := make(chan struct{}) go func() { rh.check(context.Background()) @@ -166,8 +210,8 @@ func TestHealthCheckDoesNotBlockReportOrStatusz(t *testing.T) { } report := rh.Report() - if !report.Envoy.Healthy || report.Envoy.SuccessCount != 1 || report.Envoy.LastSuccess.IsZero() { - t.Errorf("Envoy health = %+v, want one successful check", report.Envoy) + if !report.Dataplane.Healthy || report.Dataplane.SuccessCount != 1 || report.Dataplane.LastSuccess.IsZero() { + t.Errorf("dataplane health = %+v, want one successful check", report.Dataplane) } if !report.K8sAPI.Healthy || report.K8sAPI.SuccessCount != 1 || report.K8sAPI.LastSuccess.IsZero() { t.Errorf("Kubernetes health = %+v, want one successful check", report.K8sAPI) @@ -209,8 +253,8 @@ func TestHealthChecksRunConcurrently(t *testing.T) { }, } rh := newRouterHealth(time.Second, newHealthTestClientset(t, server), apiClient, routerConfig{}) - rh.envoyClient = &http.Client{Transport: healthRoundTripFunc(func(*http.Request) (*http.Response, error) { - started <- "envoy" + rh.dataplaneClient = &http.Client{Transport: healthRoundTripFunc(func(*http.Request) (*http.Response, error) { + started <- "dataplane" <-release return &http.Response{ StatusCode: http.StatusOK, @@ -236,7 +280,7 @@ func TestHealthChecksRunConcurrently(t *testing.T) { case dependency := <-started: seen[dependency] = true case <-timer.C: - t.Fatalf("started health checks = %v, want envoy, k8s, and ateapi before any check finishes", seen) + t.Fatalf("started health checks = %v, want dataplane, k8s, and ateapi before any check finishes", seen) } } @@ -248,7 +292,7 @@ func TestHealthChecksRunConcurrently(t *testing.T) { } report := rh.Report() - if !report.Envoy.Healthy || !report.K8sAPI.Healthy || !report.AteAPI.Healthy { + if !report.Dataplane.Healthy || !report.K8sAPI.Healthy || !report.AteAPI.Healthy { t.Errorf("health report = %+v, want all dependencies healthy", report) } } @@ -262,7 +306,7 @@ func TestHealthStartStopsWhenK8sCheckIsCanceled(t *testing.T) { defer server.Close() rh := newRouterHealth(time.Hour, newHealthTestClientset(t, server), nil, routerConfig{}) - setHealthyEnvoyClient(rh) + setHealthyDataplaneClient(rh) ctx, cancel := context.WithCancel(context.Background()) startDone := make(chan struct{}) go func() { diff --git a/cmd/atenet/internal/router/router.go b/cmd/atenet/internal/router/router.go index b989969e2..13d633660 100644 --- a/cmd/atenet/internal/router/router.go +++ b/cmd/atenet/internal/router/router.go @@ -24,7 +24,6 @@ import ( "os" "os/signal" "syscall" - "time" "github.com/spf13/cobra" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" @@ -181,23 +180,12 @@ func (s *RouterServer) Run(ctx context.Context) error { slog.InfoContext(ctx, "Connecting to ateapi", slog.String("address", s.cfg.AteapiAddr), slog.Bool("use-api-token-auth", s.cfg.Auth.AteapiUseTokenAuth)) 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), + slog.String("atenet_router", string(s.cfg.atenetRouter()))) g, ctx := errgroup.WithContext(ctx) - xdsSrv := NewXdsServer(s.cfg.XdsPort) - xdsSrv.SetConfig(s.cfg.HttpPort, s.cfg.ExtprocPort, s.cfg.ExtprocAddr) - setOtlpCollector(ctx, xdsSrv, s.cfg.OtlpCollectorAddress) - - xdsSrv.SetExtProcMaxRequests(s.cfg.extProcMaxRequests()) - if parkCfg.enabled() { - // Envoy must keep a parked request open at least as long as the router - // will hold it; add a margin so the router surfaces its own 503 first. - xdsSrv.SetExtProcMessageTimeout(parkCfg.Budget + 5*time.Second) - } - - xdsSrv.SetTlsConfig(s.cfg.HttpsPort, s.cfg.EnvoyCertPath) - xdsSrv.SetUpstreamTls(s.cfg.UpstreamCredentialBundlePath, s.cfg.UpstreamTrustBundlePath, s.cfg.UpstreamSpiffePrefix) if s.extprocSrv == nil { routeDuration, err := newRouteDurationHistogram() if err != nil { @@ -207,17 +195,13 @@ func (s *RouterServer) Run(ctx context.Context) error { if err != nil { return fmt.Errorf("failed to create parking metrics: %w", err) } - s.extprocSrv = NewExtProcServer(s.cfg.ExtprocPort, s.apiClient, routeDuration, parkCfg, parkMetrics) + s.extprocSrv = NewExtProcServer(s.cfg.ExtprocPort, s.apiClient, routeDuration, parkCfg, parkMetrics, s.cfg.atenetRouter().routeViaAuthority()) } - ctrl := NewController(s.k8sClient, s.clientset, s.cfg, xdsSrv, s.extprocSrv) - s.health = newRouterHealth(s.cfg.HealthInterval, s.clientset, s.apiClient, s.cfg) - // Start Controller / Watcher - g.Go(func() error { - slog.InfoContext(ctx, "Starting ActorTemplate controller") - return ctrl.Start(ctx) - }) + if err := s.startDataplane(ctx, g, parkCfg); err != nil { + return err + } // Start periodic service checking logic g.Go(func() error { @@ -226,18 +210,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)) diff --git a/cmd/atenet/internal/router/status.go b/cmd/atenet/internal/router/status.go index 0cdca8286..6a7625373 100644 --- a/cmd/atenet/internal/router/status.go +++ b/cmd/atenet/internal/router/status.go @@ -119,7 +119,7 @@ func (qr *QueryRecorder) AddRouterRequest( ) { qr.Add(RecordedQuery{ Timestamp: start, - Client: m.headers[":authority"], + Client: m.headers[authorityHeader], Host: m.host, Path: redactPath(m.path), Method: m.headers[":method"], diff --git a/cmd/atenet/internal/router/status_test.go b/cmd/atenet/internal/router/status_test.go index 9b72daf9e..2f6e97cbd 100644 --- a/cmd/atenet/internal/router/status_test.go +++ b/cmd/atenet/internal/router/status_test.go @@ -79,7 +79,7 @@ func TestStatuszEndpoint(t *testing.T) { t.Fatalf("Failed generating router server: %v", err) } - srv.extprocSrv = NewExtProcServer(cfg.ExtprocPort, &mockClient{}, nil, defaultParkedRequestConfig(), nil) + srv.extprocSrv = NewExtProcServer(cfg.ExtprocPort, &mockClient{}, nil, defaultParkedRequestConfig(), nil, false) ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/docs/request-parking.md b/docs/request-parking.md index aa851a545..35ed4b6ed 100644 --- a/docs/request-parking.md +++ b/docs/request-parking.md @@ -134,4 +134,3 @@ bounds the wait. **Status page** (`/statusz`): a "Request Parking" card shows whether parking is enabled, the current vs. maximum parked count, and the max wait. - diff --git a/hack/install-ate.sh b/hack/install-ate.sh index 871350a0e..d7eafb457 100755 --- a/hack/install-ate.sh +++ b/hack/install-ate.sh @@ -64,7 +64,8 @@ function usage() { echo " --deploy-ate-system Deploy core system (CRDs, atelet, apiserver)" echo " --delete-ate-system Delete core system" echo " --delete-all Delete core system and all registered demos" - echo " --ateapi-client-auth=cert|token Select how in-cluster clients authenticate to ateapi for --deploy-ate-system (default: cert; the server always accepts both)" + echo " --ateapi-client-auth=cert|token Select how in-cluster clients authenticate to ateapi for --deploy-ate-system (default: cert; the server always accepts both)" + echo " --atenet-router=envoy|agentgateway Select the atenet router dataplane (default: envoy)" echo "" echo "Infrastructure components:" echo "" @@ -145,9 +146,38 @@ ateapi_client_auth() { esac } +atenet_router() { + case "${ATE_ATENET_ROUTER:-envoy}" in + envoy|agentgateway) + echo "${ATE_ATENET_ROUTER:-envoy}" + ;; + *) + echo "Error: --atenet-router must be envoy or agentgateway, got '${ATE_ATENET_ROUTER}'" >&2 + exit 1 + ;; + esac +} + render_ate_system_manifests() { local client_auth="" + local router="" client_auth="$(ateapi_client_auth)" + router="$(atenet_router)" + + if [[ "${router}" == "agentgateway" ]]; then + local overlay="manifests/ate-install/agentgateway" + if [[ "${client_auth}" == "token" ]]; then + overlay="manifests/ate-install/agentgateway-token-client" + fi + if [[ "${ATE_INSTALL_KIND:-false}" == "true" ]]; then + overlay="manifests/ate-install/kind-agentgateway" + if [[ "${client_auth}" == "token" ]]; then + overlay="manifests/ate-install/kind-agentgateway-token-client" + fi + fi + kubectl kustomize "${overlay}" --load-restrictor LoadRestrictionsNone | run_ko resolve -f - + return + fi if [[ "${client_auth}" == "token" ]]; then local overlay="manifests/ate-install/token-client" @@ -167,6 +197,15 @@ render_ate_system_manifests() { fi } +render_atenet_router_manifest() { + if [[ "$(atenet_router)" == "agentgateway" ]]; then + kubectl kustomize manifests/ate-install/agentgateway-router \ + --load-restrictor LoadRestrictionsNone | run_ko resolve -f - + else + run_ko resolve -f manifests/ate-install/atenet-router.yaml + fi +} + # Extract a CA pool secret's RootCertificateDER and emit it as a PEM certificate. ca_pool_root_pem() { local secret="$1" @@ -389,7 +428,10 @@ deploy_atenet() { run_kubectl apply -f manifests/ate-install/ate-system-namespace.yaml \ && run_kubectl wait --for=jsonpath='{.status.phase}'=Active namespace/ate-system --timeout=60s - run_ko apply -f manifests/ate-install/atenet-router.yaml + local router_manifest="" + router_manifest="$(render_atenet_router_manifest)" + echo "${router_manifest}" | run_kubectl apply -f - + run_ko apply -f manifests/ate-install/atenet-dns.yaml run_kubectl rollout status deployment/atenet-router -n ate-system --timeout=120s # The Deployment in atenet-dns.yaml is named "dns"; every other resource in @@ -503,12 +545,16 @@ 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/components/agentgateway/configmap.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/components/agentgateway/configmap.yaml } deploy_benchmarks() { @@ -562,6 +608,14 @@ for ((i = 0; i < ${#prescan_args[@]}; i++)); do fi ATE_ATEAPI_CLIENT_AUTH="${prescan_args[$((i + 1))]}" ;; + --atenet-router=*) ATE_ATENET_ROUTER="${prescan_args[i]#*=}" ;; + --atenet-router) + if (( i + 1 >= ${#prescan_args[@]} )); then + echo "Error: --atenet-router requires envoy or agentgateway" >&2 + exit 1 + fi + ATE_ATENET_ROUTER="${prescan_args[$((i + 1))]}" + ;; --benchmark-worker-count) BENCHMARK_WORKER_COUNT="${prescan_args[i+1]:-1}" ;; @@ -570,6 +624,7 @@ for ((i = 0; i < ${#prescan_args[@]}; i++)); do ;; esac done +atenet_router >/dev/null while [[ "$#" -gt 0 ]]; do # Run ${demo}_cmdline if it exists. If it returns 0, then we successfully @@ -594,6 +649,15 @@ while [[ "$#" -gt 0 ]]; do fi ATE_ATEAPI_CLIENT_AUTH="$1" ;; + --atenet-router=*) ATE_ATENET_ROUTER="${1#*=}" ;; + --atenet-router) + shift + if [[ "$#" -eq 0 ]]; then + echo "Error: --atenet-router requires envoy or agentgateway" >&2 + exit 1 + fi + ATE_ATENET_ROUTER="$1" + ;; --deploy-ate-system) deploy_ate_system ;; --delete-ate-system) delete_ate_system ;; diff --git a/internal/atunnel/server.go b/internal/atunnel/server.go index e65ac58bc..e58cf6832 100644 --- a/internal/atunnel/server.go +++ b/internal/atunnel/server.go @@ -39,6 +39,11 @@ const ( // StaleAssignmentHeader distinguishes an atunnel routing rejection from a // 421 returned by the actor application itself. StaleAssignmentHeader = "X-Ate-Assignment-Stale" + // OriginalHostHeader carries the actor authority across router dataplanes + // that must use :authority to select the worker as their dynamic backend. + // atunnel only accepts mTLS-authenticated router clients, and the router's + // ext_proc server overwrites this header before every request. + OriginalHostHeader = "X-Ate-Original-Host" ) // Config configures an ingress Server. @@ -229,7 +234,11 @@ func (s *Server) closeIdleUpstreamConnections() { // ServeHTTP validates the actor hostname on every request before proxying it. func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { - host, err := requestHostname(r.Host) + actorHost := r.Header.Get(OriginalHostHeader) + if actorHost == "" { + actorHost = r.Host + } + host, err := requestHostname(actorHost) if err != nil { s.reject(w) return @@ -258,8 +267,14 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { cancel() }() + // Do not expose the router-only routing header to actor code. Restore Host + // so dataplanes that route dynamically on worker IP still give the actor its + // stable actor DNS name. + r.Header.Del(OriginalHostHeader) + r.Host = actorHost + // ReverseProxy changes the URL destination but intentionally retains Host, - // allowing the actor application to observe its stable mesh hostname. + // allowing the actor application to observe its stable actor DNS name. s.proxy.ServeHTTP(w, r.WithContext(requestCtx)) } diff --git a/internal/atunnel/server_test.go b/internal/atunnel/server_test.go index 1b75a836a..9c55a16f5 100644 --- a/internal/atunnel/server_test.go +++ b/internal/atunnel/server_test.go @@ -35,7 +35,7 @@ import ( ) func TestServeHTTP(t *testing.T) { - upstreamHost := make(chan string, 3) + upstreamHost := make(chan string, 4) upstreamURL, err := url.Parse("http://actor.internal:80") if err != nil { t.Fatal(err) @@ -55,23 +55,28 @@ func TestServeHTTP(t *testing.T) { } tests := []struct { - name string - host string - wantStatus int + name string + host string + originalHost string + wantStatus int }{ - {"active actor", "actor-1.team-a.actors.resources.substrate.ate.dev", http.StatusNoContent}, - {"active actor with port", "actor-1.team-a.actors.resources.substrate.ate.dev:443", http.StatusNoContent}, - {"DNS case insensitive", "ACTOR-1.TEAM-A.ACTORS.RESOURCES.SUBSTRATE.ATE.DEV", http.StatusNoContent}, - {"wrong actor", "actor-2.team-a.actors.resources.substrate.ate.dev", http.StatusMisdirectedRequest}, - {"wrong atespace", "actor-1.team-b.actors.resources.substrate.ate.dev", http.StatusMisdirectedRequest}, - {"suffix confusion", "actor-1.team-a.actors.resources.substrate.ate.dev.example.com", http.StatusMisdirectedRequest}, - {"malformed port", "actor-1.team-a.actors.resources.substrate.ate.dev:nope", http.StatusMisdirectedRequest}, - {"empty", "", http.StatusMisdirectedRequest}, + {name: "active actor", host: "actor-1.team-a.actors.resources.substrate.ate.dev", wantStatus: http.StatusNoContent}, + {name: "active actor with port", host: "actor-1.team-a.actors.resources.substrate.ate.dev:443", wantStatus: http.StatusNoContent}, + {name: "DNS case insensitive", host: "ACTOR-1.TEAM-A.ACTORS.RESOURCES.SUBSTRATE.ATE.DEV", wantStatus: http.StatusNoContent}, + {name: "router original host", host: "10.0.0.52:443", originalHost: "actor-1.team-a.actors.resources.substrate.ate.dev", wantStatus: http.StatusNoContent}, + {name: "wrong actor", host: "actor-2.team-a.actors.resources.substrate.ate.dev", wantStatus: http.StatusMisdirectedRequest}, + {name: "wrong atespace", host: "actor-1.team-b.actors.resources.substrate.ate.dev", wantStatus: http.StatusMisdirectedRequest}, + {name: "suffix confusion", host: "actor-1.team-a.actors.resources.substrate.ate.dev.example.com", wantStatus: http.StatusMisdirectedRequest}, + {name: "malformed port", host: "actor-1.team-a.actors.resources.substrate.ate.dev:nope", wantStatus: http.StatusMisdirectedRequest}, + {name: "empty", wantStatus: http.StatusMisdirectedRequest}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "https://worker/hello", nil) req.Host = tt.host + if tt.originalHost != "" { + req.Header.Set(OriginalHostHeader, tt.originalHost) + } rec := httptest.NewRecorder() s.ServeHTTP(rec, req) if rec.Code != tt.wantStatus { @@ -83,7 +88,7 @@ func TestServeHTTP(t *testing.T) { }) } - for range 3 { + for range 4 { if got := <-upstreamHost; got != "actor-1.team-a.actors.resources.substrate.ate.dev" && got != "actor-1.team-a.actors.resources.substrate.ate.dev:443" && got != "ACTOR-1.TEAM-A.ACTORS.RESOURCES.SUBSTRATE.ATE.DEV" { t.Errorf("upstream Host = %q", got) } diff --git a/manifests/ate-install/agentgateway-router/kustomization.yaml b/manifests/ate-install/agentgateway-router/kustomization.yaml new file mode 100644 index 000000000..f31e2d085 --- /dev/null +++ b/manifests/ate-install/agentgateway-router/kustomization.yaml @@ -0,0 +1,22 @@ +# 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. + +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../atenet-router.yaml + +components: + - ../components/agentgateway diff --git a/manifests/ate-install/agentgateway-token-client/kustomization.yaml b/manifests/ate-install/agentgateway-token-client/kustomization.yaml new file mode 100644 index 000000000..5a7c58b0c --- /dev/null +++ b/manifests/ate-install/agentgateway-token-client/kustomization.yaml @@ -0,0 +1,22 @@ +# 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. + +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../token-client + +components: + - ../components/agentgateway diff --git a/manifests/ate-install/agentgateway/kustomization.yaml b/manifests/ate-install/agentgateway/kustomization.yaml new file mode 100644 index 000000000..79cebcbe9 --- /dev/null +++ b/manifests/ate-install/agentgateway/kustomization.yaml @@ -0,0 +1,22 @@ +# 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. + +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../base + +components: + - ../components/agentgateway diff --git a/manifests/ate-install/base/kustomization.yaml b/manifests/ate-install/base/kustomization.yaml new file mode 100644 index 000000000..ac1623f83 --- /dev/null +++ b/manifests/ate-install/base/kustomization.yaml @@ -0,0 +1,25 @@ +# 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. + +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../ate-api-server.yaml + - ../ate-controller.yaml + - ../atelet.yaml + - ../atenet-dns.yaml + - ../atenet-router.yaml + - ../valkey.yaml + - ../pod-certificate-controller.yaml diff --git a/manifests/ate-install/components/agentgateway/configmap.yaml b/manifests/ate-install/components/agentgateway/configmap.yaml new file mode 100644 index 000000000..9876b6c8a --- /dev/null +++ b/manifests/ate-install/components/agentgateway/configmap.yaml @@ -0,0 +1,76 @@ +# 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. + +apiVersion: v1 +kind: ConfigMap +metadata: + name: atenet-router-agentgateway-config + namespace: ate-system +data: + config.yaml: | + # yaml-language-server: $schema=https://agentgateway.dev/schema/config + config: + # Actor sandboxes behind a worker IP are replaced between requests. Do + # not retain an idle connection that may belong to the previous actor. + backend: + poolMaxSize: 0 + + frontendPolicies: + tracing: + host: $AGENTGATEWAY_OTLP_ADDRESS + protocol: grpc + randomSampling: true + + gateways: + http: + port: 8080 + protocol: HTTP + https: + port: 8443 + protocol: HTTPS + tls: + cert: /run/servicedns.podcert.ate.dev/credential-bundle.pem + key: /run/servicedns.podcert.ate.dev/credential-bundle.pem + + routes: + - name: substrate-actors + gateways: + - http + - https + matches: + - path: + pathPrefix: / + policies: + extProc: + host: 127.0.0.1:50051 + failureMode: failClosed + processingOptions: + requestHeaderMode: send + responseHeaderMode: skip + requestBodyMode: none + responseBodyMode: none + requestTrailerMode: skip + responseTrailerMode: skip + backends: + - dynamic: {} + policies: + # atunnel serves HTTPS on each worker pod IP. Verify its certificate + # against the podidentity CA and present the router's podidentity + # credential. Worker SPIFFE IDs vary with their workload namespace, + # so skip DNS/IP hostname matching while retaining CA verification. + backendTLS: + cert: /run/podidentity.podcert.ate.dev/credential-bundle.pem + key: /run/podidentity.podcert.ate.dev/credential-bundle.pem + root: /run/podidentity.podcert.ate.dev/trust-bundle.pem + insecureHost: true diff --git a/manifests/ate-install/components/agentgateway/kustomization.yaml b/manifests/ate-install/components/agentgateway/kustomization.yaml new file mode 100644 index 000000000..5ce179141 --- /dev/null +++ b/manifests/ate-install/components/agentgateway/kustomization.yaml @@ -0,0 +1,93 @@ +# 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. + +apiVersion: kustomize.config.k8s.io/v1alpha1 +kind: Component + +resources: + - configmap.yaml + +patches: + - patch: |- + apiVersion: v1 + kind: ConfigMap + metadata: + name: atenet-router-envoy-config + namespace: ate-system + $patch: delete + - target: + group: apps + version: v1 + kind: Deployment + name: atenet-router + namespace: ate-system + patch: |- + - op: test + path: /spec/template/spec/containers/0/args/4 + value: --port-xds=18000 + - op: remove + path: /spec/template/spec/containers/0/args/4 + - op: test + path: /spec/template/spec/containers/0/args/8 + value: --envoy-cert-path=/run/servicedns.podcert.ate.dev/credential-bundle.pem + - op: remove + path: /spec/template/spec/containers/0/args/8 + - op: test + path: /spec/template/spec/containers/0/ports/0/name + value: xds + - op: remove + path: /spec/template/spec/containers/0/ports/0 + - op: add + path: /spec/template/spec/containers/0/args/- + value: --atenet-router=agentgateway + - op: replace + path: /spec/template/spec/containers/1 + value: + name: agentgateway + image: cr.agentgateway.dev/agentgateway:v1.4.1 + args: + - -f + - /etc/agentgateway/config.yaml + env: + - name: AGENTGATEWAY_OTLP_ADDRESS + value: opentelemetry-collector.gke-managed-otel.svc.cluster.local:4317 + ports: + - name: http + containerPort: 8080 + - name: https + containerPort: 8443 + - name: readiness + containerPort: 15021 + - name: stats + containerPort: 15020 + volumeMounts: + - name: envoy-config + mountPath: /etc/agentgateway + - name: servicedns + mountPath: /run/servicedns.podcert.ate.dev + - name: podidentity + mountPath: /run/podidentity.podcert.ate.dev + - patch: |- + apiVersion: apps/v1 + kind: Deployment + metadata: + name: atenet-router + namespace: ate-system + spec: + template: + spec: + volumes: + - name: envoy-config + configMap: + name: atenet-router-agentgateway-config diff --git a/manifests/ate-install/kind-agentgateway-token-client/kustomization.yaml b/manifests/ate-install/kind-agentgateway-token-client/kustomization.yaml new file mode 100644 index 000000000..a0d876f3c --- /dev/null +++ b/manifests/ate-install/kind-agentgateway-token-client/kustomization.yaml @@ -0,0 +1,38 @@ +# 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. + +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../kind-token-client + +components: + - ../components/agentgateway + +patches: + - patch: |- + apiVersion: apps/v1 + kind: Deployment + metadata: + name: atenet-router + namespace: ate-system + spec: + template: + spec: + containers: + - name: agentgateway + env: + - name: AGENTGATEWAY_OTLP_ADDRESS + value: opentelemetry-collector.otel-system.svc:4317 diff --git a/manifests/ate-install/kind-agentgateway/kustomization.yaml b/manifests/ate-install/kind-agentgateway/kustomization.yaml new file mode 100644 index 000000000..96cdc1975 --- /dev/null +++ b/manifests/ate-install/kind-agentgateway/kustomization.yaml @@ -0,0 +1,38 @@ +# 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. + +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../kind + +components: + - ../components/agentgateway + +patches: + - patch: |- + apiVersion: apps/v1 + kind: Deployment + metadata: + name: atenet-router + namespace: ate-system + spec: + template: + spec: + containers: + - name: agentgateway + env: + - name: AGENTGATEWAY_OTLP_ADDRESS + value: opentelemetry-collector.otel-system.svc:4317 diff --git a/manifests/ate-install/token-client/kustomization.yaml b/manifests/ate-install/token-client/kustomization.yaml index fc004748d..3b172a17b 100644 --- a/manifests/ate-install/token-client/kustomization.yaml +++ b/manifests/ate-install/token-client/kustomization.yaml @@ -19,13 +19,7 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: - - ../ate-api-server.yaml - - ../ate-controller.yaml - - ../atelet.yaml - - ../atenet-dns.yaml - - ../atenet-router.yaml - - ../valkey.yaml - - ../pod-certificate-controller.yaml + - ../base patches: - path: patches.yaml