Merge remote-tracking branch 'upstream/main'

# Conflicts:
#	cmd/atenet/internal/router/config.go
#	cmd/atenet/internal/router/dashboard.html
#	cmd/atenet/internal/router/extproc.go
#	cmd/atenet/internal/router/health.go
#	cmd/atenet/internal/router/health_test.go
#	cmd/atenet/internal/router/router.go
#	hack/install-ate.sh
#	manifests/ate-install/base/kustomization.yaml
#	manifests/ate-install/kind-agentgateway/kustomization.yaml
#	manifests/ate-install/token-client/kustomization.yaml

Author: EItanya

charts/substrate/templates/atenet-router.yaml

@@ -84,14 +84,8 @@ data:
                 responseTrailerMode: skip
           backends:
           - dynamic: {}
-            policies:
-              transformations:
-                request:
-                  set:
-                    ":authority": 'request.headers["x-ate-actor-host"]'
-                  remove:
-                  - x-ate-actor-host
 {{- if eq .Values.auth.mode "mtls" }}
+            policies:
               backendTLS:
                 cert: "/run/podidentity.podcert.ate.dev/cert.pem"
                 key: "/run/podidentity.podcert.ate.dev/key.pem"
@@ -128,14 +122,8 @@ data:
                 responseTrailerMode: skip
           backends:
           - dynamic: {}
-            policies:
-              transformations:
-                request:
-                  set:
-                    ":authority": 'request.headers["x-ate-actor-host"]'
-                  remove:
-                  - x-ate-actor-host
 {{- if eq .Values.auth.mode "mtls" }}
+            policies:
               backendTLS:
                 cert: "/run/podidentity.podcert.ate.dev/cert.pem"
                 key: "/run/podidentity.podcert.ate.dev/key.pem"

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 `--networking-mode=envoy`.
+  Unless `--standalone` is set, it also manages the Envoy Deployment and
+  Services in Kubernetes.
+  With the default `--networking-mode=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
+* Build tag

cmd/atenet/internal/router/agentgateway.go

@@ -140,16 +140,10 @@ func (p agentgatewayProvider) routeBlock(name string) string {
 }
 
 func (p agentgatewayProvider) backendBlock() string {
-	backend := fmt.Sprintf(`- dynamic: {}
-  policies:
-    transformations:
-      request:
-        set:
-          ":authority": 'request.headers["%s"]'
-        remove:
-        - %s`, ActorHostHeader, ActorHostHeader)
+	backend := `- dynamic: {}`
 	if p.cfg.UpstreamCertPath != "" {
 		backend += fmt.Sprintf(`
+  policies:
     backendTLS:
       cert: %q
       key: %q

cmd/atenet/internal/router/cmd.go

@@ -44,6 +44,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.NetworkingMode, "atenet-router", NetworkingModeAgentgateway, "Alias for --networking-mode")
 	cmd.Flags().StringVar(&cfg.Namespace, "namespace", "default", "Target operations namespace")
 	cmd.Flags().StringVar(&cfg.RouterServiceName, "router-service-name", installdefaults.RouterServiceName, "Service name of this atenet-router in the operations namespace. Override when the deployment renames the Service.")
 	cmd.Flags().StringVar(&cfg.Kubeconfig, "kubeconfig", "", "Absolute path to the kubeconfig configuration file")

cmd/atenet/internal/router/config_test.go

@@ -25,10 +25,6 @@ func TestRouterConfigValidate(t *testing.T) {
 		cfg     routerConfig
 		wantErr string // substring; empty means valid
 	}{
-		{
-			name: "defaults are valid (auto breaker)",
-			cfg:  routerConfig{ExtProcMaxRequests: 0, ParkedRequest: ParkedRequestConfig{Max: defaultParkedRequestMax}},
-		},
 		{
 			name:    "negative extproc-max-requests rejected",
 			cfg:     routerConfig{ExtProcMaxRequests: -1, ParkedRequest: ParkedRequestConfig{Max: 0}},

cmd/atenet/internal/router/extproc.go

@@ -48,15 +48,19 @@ type ExtProcServer struct {
 	networkingMode string
 }
 
-func NewExtProcServer(port int, apiClient ateapipb.ControlClient, routeDuration metric.Float64Histogram, parkCfg ParkedRequestConfig, parkMetrics *parkingMetrics) *ExtProcServer {
-	return &ExtProcServer{
+func NewExtProcServer(port int, apiClient ateapipb.ControlClient, routeDuration metric.Float64Histogram, parkCfg ParkedRequestConfig, parkMetrics *parkingMetrics, routeViaAuthority ...bool) *ExtProcServer {
+	s := &ExtProcServer{
 		port:          port,
 		apiClient:     apiClient,
 		recorder:      NewQueryRecorder(100),
 		resumer:       NewActorResumer(apiClient, withParking(parkCfg)),
 		routeDuration: routeDuration,
 		parking:       newParkingLot(parkCfg, parkMetrics),
 	}
+	if len(routeViaAuthority) > 0 && routeViaAuthority[0] {
+		s.networkingMode = NetworkingModeAgentgateway
+	}
+	return s
 }
 
 func (s *ExtProcServer) SetNetworkingMode(mode string) {
@@ -196,17 +200,7 @@ func (s *ExtProcServer) handleRequestHeaders(
 	slog.InfoContext(ctx, "Route ok", slog.Any("actor", actorRef), slog.String("targetAddr", targetAddr))
 
 	mutation := &extprocv3.HeaderMutation{}
-	if s.networkingMode == NetworkingModeAgentgateway {
-		// Agentgateway's dynamic backend selects the upstream from :authority.
-		// Its backend transformation restores the actor Host before the request
-		// reaches atunnel.
-		addAuthorityMutation(targetAddr, mutation)
-		addActorHostMutation(metadata.host, mutation)
-	} else {
-		// Envoy's ORIGINAL_DST cluster reads the target from this header and
-		// leaves :authority as the actor DNS name for atunnel authorization.
-		addOriginalDstMutation(targetAddr, mutation)
-	}
+	addRoutingMutations(targetAddr, metadata.host, s.networkingMode == NetworkingModeAgentgateway, mutation)
 
 	return &extprocv3.HeadersResponse{
 		Response: &extprocv3.CommonResponse{

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

cmd/atenet/internal/router/extproc_out.go

@@ -15,21 +15,12 @@
 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"
 )
 
-const ActorHostHeader = "x-ate-actor-host"
-
-func addAuthorityMutation(authority string, mut *extproc.HeaderMutation) {
-	addHeaderMutation(":authority", authority, mut)
-}
-
-func addActorHostMutation(host string, mut *extproc.HeaderMutation) {
-	addHeaderMutation(ActorHostHeader, host, mut)
-}
-
 func addHeaderMutation(name, value string, mut *extproc.HeaderMutation) {
 	mut.SetHeaders = append(mut.SetHeaders, &corev3.HeaderValueOption{
 		AppendAction: corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
@@ -65,6 +56,31 @@ func addOriginalDstMutation(dst string, mut *extproc.HeaderMutation) {
 	addHeaderMutation(OriginalDstHeader, dst, mut)
 }
 
+// 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{

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
@@ -271,7 +274,7 @@ func TestAgentgatewayRoutingHeaders(t *testing.T) {
 		t.Fatal(err)
 	}
 	headers := res.Response.GetHeaderMutation().GetSetHeaders()
-	if len(headers) != 2 || headers[0].GetHeader().GetKey() != ":authority" || string(headers[0].GetHeader().GetRawValue()) != "10.0.0.52:443" || headers[1].GetHeader().GetKey() != ActorHostHeader || string(headers[1].GetHeader().GetRawValue()) != host {
+	if len(headers) != 3 || headers[2].GetHeader().GetKey() != authorityHeader || string(headers[2].GetHeader().GetRawValue()) != "10.0.0.52:443" {
 		t.Fatalf("Agentgateway mutations = %v", headers)
 	}
 }
@@ -291,7 +294,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")
@@ -410,7 +413,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
@@ -435,3 +438,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])
+	}
+}

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