ate-setup: settings for the router's route timeout, parked budget and ext_proc breaker

cmd/ate-setup/commands.md

@@ -25,6 +25,9 @@ a pre-scan pass, so they may appear anywhere on its command line.
 |---|---|---|
 | `--kind` | `hack/install-ate-kind.sh`, or `ATE_INSTALL_KIND=true` | Kind overlays, the local registry, and host-architecture image builds |
 | `--atenet-dataplane envoy\|agentgateway` | `--atenet-dataplane envoy\|agentgateway` | atenet ingress and egress dataplane (default `envoy`) |
+| `--atenet-route-timeout DURATION` | `ATE_ATENET_ROUTE_TIMEOUT=DURATION` | New. The router's `--route-timeout` (default `5m`). Envoy dataplane only |
+| `--atenet-parked-request-budget DURATION` | `ATE_ATENET_PARKED_REQUEST_BUDGET=DURATION` | New. The router's `--parked-request-budget` (default `5s`); the router pod's `terminationGracePeriodSeconds` grows with it. Envoy dataplane only |
+| `--atenet-extproc-max-requests N` | `ATE_ATENET_EXTPROC_MAX_REQUESTS=N` | New. The router's `--extproc-max-requests`, the circuit breaker on Envoy's ext_proc cluster (default: twice `--parked-request-max`, at least `1024`). Envoy dataplane only |
 | `--rollout-timeout DURATION` | `--rollout-timeout DURATION` | Readiness timeout for workloads (default `60s`). Unlike the shell flag it also governs the podcertificate-controller and CSI waits, which stay at their 120s default until it is passed |
 | `--podcert-workers-per-signer N` | `--podcert-workers-per-signer N` | Concurrent workers per podcertificate-controller signer |
 | `--cluster-size size0\|size10` | `--cluster-size size0\|size10` | Footprint profile (default `size0`). `size10` assumes a dedicated PostgreSQL node: it resizes the bundled StatefulSet and its `postgresql.conf`, pins the apiserver's connection pool, and raises the podcertificate-controller's API rate limits. `ATE_INSTALL_CLUSTER_SIZE` when the flag is absent |

cmd/ate-setup/internal/config/config.go

@@ -158,6 +158,13 @@ type Config struct {
 
 	// Router selects the atenet router dataplane.
 	Router string
+	// RouterRouteTimeout and RouterParkedRequestBudget are the router's
+	// --route-timeout and --parked-request-budget; zero keeps the router default.
+	RouterRouteTimeout        time.Duration
+	RouterParkedRequestBudget time.Duration
+	// RouterExtProcMaxRequests is the router's --extproc-max-requests; zero
+	// keeps the router's derivation from --parked-request-max.
+	RouterExtProcMaxRequests int
 	// The read/write and owner connections can use different login identities.
 	// With one configured connection, both pools use it. Both empty selects
 	// bundled PostgreSQL.

cmd/ate-setup/internal/config/config_test.go

@@ -60,6 +60,9 @@ func loadEnv(t *testing.T) {
 		"ATE_API_POSTGRES_SCHEMA",
 		"ATE_API_POSTGRES_SERVER_CA_FILE",
 		"ATE_ATENET_DATAPLANE",
+		"ATE_ATENET_EXTPROC_MAX_REQUESTS",
+		"ATE_ATENET_PARKED_REQUEST_BUDGET",
+		"ATE_ATENET_ROUTE_TIMEOUT",
 		"ATE_CREDENTIAL_PROVIDER",
 		"ATE_IMAGE_REPO",
 		"ATE_IMAGE_TAG",

cmd/ate-setup/internal/config/load.go

@@ -244,11 +244,14 @@ func buildConfig(root string, env map[string]string, r *Resolved) (*Config, erro
 			Repo: strings.TrimSuffix(r.String("images.repo"), "/"),
 			Tag:  r.String("images.tag"),
 		},
-		Router:               r.String("atenet.dataplane"),
-		PostgresSchema:       r.String("ateapi.postgres.schema"),
-		PostgresPoolMaxConns: r.String("ateapi.postgres.poolMaxConns"),
-		PostgresServerCAFile: r.String("ateapi.postgres.serverCAFile"),
-		PostgresStorageClass: r.String("ateapi.postgres.storageClass"),
+		Router:                    r.String("atenet.dataplane"),
+		RouterRouteTimeout:        r.Duration("atenet.router.routeTimeout"),
+		RouterParkedRequestBudget: r.Duration("atenet.router.parkedRequestBudget"),
+		RouterExtProcMaxRequests:  r.Int("atenet.router.extprocMaxRequests"),
+		PostgresSchema:            r.String("ateapi.postgres.schema"),
+		PostgresPoolMaxConns:      r.String("ateapi.postgres.poolMaxConns"),
+		PostgresServerCAFile:      r.String("ateapi.postgres.serverCAFile"),
+		PostgresStorageClass:      r.String("ateapi.postgres.storageClass"),
 
 		PostgresReadWriteConnectionString: readWriteConnectionString,
 		PostgresOwnerConnectionString:     ownerConnectionString,
@@ -399,6 +402,27 @@ func validateResolved(cfg *Config, r *Resolved) error {
 		return &InvalidError{Value: dataplane, Want: RouterEnvoy + " or " + RouterAgentgateway}
 	}
 
+	// Only the envoy dataplane runs the atenet-router container the flags go on.
+	for _, key := range []string{"atenet.router.routeTimeout", "atenet.router.parkedRequestBudget", "atenet.router.extprocMaxRequests"} {
+		v, _ := r.Value(key)
+		if v.Raw == "" {
+			continue
+		}
+		switch v.Setting.Kind {
+		case KindDuration:
+			if r.Duration(key) <= 0 {
+				return &InvalidError{Value: v, Want: "a positive duration"}
+			}
+		case KindInt:
+			if r.Int(key) <= 0 {
+				return &InvalidError{Value: v, Want: "a positive integer"}
+			}
+		}
+		if cfg.Router != RouterEnvoy {
+			return &ConflictError{A: v, B: dataplane, Why: "the router flags need the envoy dataplane"}
+		}
+	}
+
 	if ipType, _ := r.Value("ateapi.postgres.cloudsql.ipType"); ipType.Raw != "" {
 		switch ipType.Raw {
 		case CloudSQLIPTypePrivate, CloudSQLIPTypePublic, CloudSQLIPTypePSC:

cmd/ate-setup/internal/config/load_test.go

@@ -20,6 +20,7 @@ import (
 	"runtime"
 	"strings"
 	"testing"
+	"time"
 )
 
 // Under --kind the ambient environment is ignored, but a value the caller named
@@ -224,3 +225,76 @@ func errPath(err error) string {
 	}
 	return rest
 }
+
+func TestLoadFlagsRouterSettings(t *testing.T) {
+	loadEnv(t)
+	t.Setenv("ATE_ATENET_ROUTE_TIMEOUT", "30m")
+	t.Setenv("ATE_ATENET_PARKED_REQUEST_BUDGET", "2m")
+	t.Setenv("ATE_ATENET_EXTPROC_MAX_REQUESTS", "4096")
+
+	fs, err := optionsFlagSet(Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cfg, err := LoadFlags(fs, LoadOptions{})
+	if err != nil {
+		t.Fatalf("LoadFlags() error = %v", err)
+	}
+	if cfg.RouterRouteTimeout != 30*time.Minute {
+		t.Errorf("RouterRouteTimeout = %s, want 30m", cfg.RouterRouteTimeout)
+	}
+	if cfg.RouterParkedRequestBudget != 2*time.Minute {
+		t.Errorf("RouterParkedRequestBudget = %s, want 2m", cfg.RouterParkedRequestBudget)
+	}
+	if cfg.RouterExtProcMaxRequests != 4096 {
+		t.Errorf("RouterExtProcMaxRequests = %d, want 4096", cfg.RouterExtProcMaxRequests)
+	}
+}
+
+func TestLoadFlagsRouterSettingsNeedEnvoy(t *testing.T) {
+	for env, value := range map[string]string{
+		"ATE_ATENET_PARKED_REQUEST_BUDGET": "2m",
+		"ATE_ATENET_EXTPROC_MAX_REQUESTS":  "4096",
+	} {
+		t.Run(env, func(t *testing.T) {
+			loadEnv(t)
+			t.Setenv("ATE_ATENET_DATAPLANE", RouterAgentgateway)
+			t.Setenv(env, value)
+
+			fs, err := optionsFlagSet(Options{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			_, err = LoadFlags(fs, LoadOptions{})
+			if err == nil {
+				t.Fatal("LoadFlags() = nil error, want a dataplane conflict")
+			}
+			for _, want := range []string{env, "ATE_ATENET_DATAPLANE"} {
+				if !strings.Contains(err.Error(), want) {
+					t.Errorf("error %q does not name %s", err, want)
+				}
+			}
+		})
+	}
+}
+
+func TestLoadFlagsRouterSettingsMustBePositive(t *testing.T) {
+	for _, tc := range []struct{ env, value, want string }{
+		{"ATE_ATENET_ROUTE_TIMEOUT", "0s", "positive duration"},
+		{"ATE_ATENET_EXTPROC_MAX_REQUESTS", "0", "positive integer"},
+		{"ATE_ATENET_EXTPROC_MAX_REQUESTS", "-1", "positive integer"},
+	} {
+		t.Run(tc.env+"="+tc.value, func(t *testing.T) {
+			loadEnv(t)
+			t.Setenv(tc.env, tc.value)
+
+			fs, err := optionsFlagSet(Options{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if _, err := LoadFlags(fs, LoadOptions{}); err == nil || !strings.Contains(err.Error(), tc.want) {
+				t.Errorf("LoadFlags() error = %v, want a %s error", err, tc.want)
+			}
+		})
+	}
+}

cmd/ate-setup/internal/config/setting.go

@@ -163,6 +163,22 @@ var Registry = []Setting{
 		Default: RouterEnvoy,
 		Usage:   "Atenet ingress and egress dataplane: envoy or agentgateway",
 	},
+	{
+		// Set on the envoy dataplane's atenet-router container; see steps.configureAtenetRouter.
+		Key: "atenet.router.routeTimeout", Env: "ATE_ATENET_ROUTE_TIMEOUT", Flag: "atenet-route-timeout",
+		Kind:  KindDuration,
+		Usage: "The router's --route-timeout; unset keeps the router default (envoy dataplane only)",
+	},
+	{
+		Key: "atenet.router.parkedRequestBudget", Env: "ATE_ATENET_PARKED_REQUEST_BUDGET", Flag: "atenet-parked-request-budget",
+		Kind:  KindDuration,
+		Usage: "The router's --parked-request-budget; unset keeps the router default; the router pod's grace period grows with it (envoy dataplane only)",
+	},
+	{
+		Key: "atenet.router.extprocMaxRequests", Env: "ATE_ATENET_EXTPROC_MAX_REQUESTS", Flag: "atenet-extproc-max-requests",
+		Kind:  KindInt,
+		Usage: "The router's --extproc-max-requests, the circuit breaker on Envoy's ext_proc cluster; unset keeps the router's derivation (envoy dataplane only)",
+	},
 	{
 		Key: "atenet.egress.additionalExtprocService", Env: "ATE_ADDITIONAL_EGRESS_EXTPROC_SERVICE",
 		Flag: "experimental-additional-egress-extproc-service", Kind: KindString,

cmd/ate-setup/internal/steps/apienv.go

@@ -336,15 +336,11 @@ func (e *Env) annotateAPIServerEnvHash(ctx context.Context) error {
 	return e.Kube.PatchDeployment(ctx, e.Namespace(), "ate-api-server", []byte(patch))
 }
 
-// applyWithAPIServerEnvHash applies a manifest that carries the ate-api-server
-// Deployment, with the digest of its current environment on the pod template.
-// A new image and a new environment are then one rollout, and a fresh install
-// starts out stamped.
-func (e *Env) applyWithAPIServerEnvHash(ctx context.Context, manifest []byte) error {
-	objs, err := kube.DecodeManifestBytes(manifest)
-	if err != nil {
-		return err
-	}
+// applyWithAPIServerEnvHash applies a decoded manifest that carries the
+// ate-api-server Deployment, with the digest of its current environment on the
+// pod template. A new image and a new environment are then one rollout, and a
+// fresh install starts out stamped.
+func (e *Env) applyWithAPIServerEnvHash(ctx context.Context, objs []*unstructured.Unstructured) error {
 	hash, err := e.apiServerEnvHash(ctx)
 	if err != nil {
 		return err

cmd/ate-setup/internal/steps/deploy.go

@@ -148,7 +148,14 @@ func (e *Env) DeployAteSystem(ctx context.Context, opts DeployOptions) error {
 	if err := e.removeFakeWorkersync(ctx); err != nil {
 		return err
 	}
-	if err := e.applyWithAPIServerEnvHash(ctx, manifests); err != nil {
+	objs, err := kube.DecodeManifestBytes(manifests)
+	if err != nil {
+		return err
+	}
+	if err := e.configureAtenetRouter(objs); err != nil {
+		return err
+	}
+	if err := e.applyWithAPIServerEnvHash(ctx, objs); err != nil {
 		return err
 	}
 
@@ -332,7 +339,11 @@ func (e *Env) DeployAteAPIServer(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	if err := e.applyWithAPIServerEnvHash(ctx, manifest); err != nil {
+	objs, err := kube.DecodeManifestBytes(manifest)
+	if err != nil {
+		return err
+	}
+	if err := e.applyWithAPIServerEnvHash(ctx, objs); err != nil {
 		return err
 	}
 	// After the manifest, which resets the pod template to the sidecar-free base.
@@ -558,7 +569,14 @@ func (e *Env) DeployAtenet(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	if err := e.Kube.ApplyBytes(ctx, routerManifest); err != nil {
+	routerObjs, err := kube.DecodeManifestBytes(routerManifest)
+	if err != nil {
+		return err
+	}
+	if err := e.configureAtenetRouter(routerObjs); err != nil {
+		return err
+	}
+	if err := e.Kube.Apply(ctx, routerObjs); err != nil {
 		return err
 	}
 	if err := e.EnsureEgressMITMCAPoolSecret(ctx); err != nil {

cmd/ate-setup/internal/steps/router.go

@@ -0,0 +1,130 @@
+// 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 steps
+
+import (
+	"fmt"
+	"math"
+	"strconv"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+
+	"go.ate.dev/substrate/cmd/ate-setup/internal/kube"
+	"go.ate.dev/substrate/cmd/ate-setup/internal/log"
+)
+
+// manifestParkedRequestBudget is the budget the manifest's
+// terminationGracePeriodSeconds is sized for: the router default.
+const manifestParkedRequestBudget = 5 * time.Second
+
+// configureAtenetRouter sets --route-timeout, --parked-request-budget and
+// --extproc-max-requests on the atenet-router container and grows
+// terminationGracePeriodSeconds with the budget, since the router's drain
+// deadline follows it. A no-op when unset.
+func (e *Env) configureAtenetRouter(objs []*unstructured.Unstructured) error {
+	routeTimeout, budget, maxRequests := e.Cfg.RouterRouteTimeout, e.Cfg.RouterParkedRequestBudget, e.Cfg.RouterExtProcMaxRequests
+	if routeTimeout <= 0 && budget <= 0 && maxRequests <= 0 {
+		return nil
+	}
+	var dep *unstructured.Unstructured
+	for _, obj := range objs {
+		if obj.GetKind() == "Deployment" && obj.GetNamespace() == e.Namespace() && obj.GetName() == "atenet-router" {
+			dep = obj
+			break
+		}
+	}
+	if dep == nil {
+		return fmt.Errorf("the router manifest has no deployment/atenet-router in %s", e.Namespace())
+	}
+
+	containers, _, err := unstructured.NestedSlice(dep.Object, "spec", "template", "spec", "containers")
+	if err != nil {
+		return fmt.Errorf("%s: %w", kube.Describe(dep), err)
+	}
+	idx := -1
+	for i, c := range containers {
+		if m, ok := c.(map[string]any); ok && m["name"] == "atenet-router" {
+			idx = i
+			break
+		}
+	}
+	if idx < 0 {
+		return fmt.Errorf("%s has no atenet-router container for the router flags", kube.Describe(dep))
+	}
+	container := containers[idx].(map[string]any)
+	args, _, err := unstructured.NestedSlice(container, "args")
+	if err != nil {
+		return fmt.Errorf("%s: %w", kube.Describe(dep), err)
+	}
+
+	if routeTimeout > 0 {
+		log.Infof("Setting the router's --route-timeout to %s", routeTimeout)
+		args = setFlag(args, "--route-timeout", routeTimeout.String())
+	}
+	if budget > 0 {
+		log.Infof("Setting the router's --parked-request-budget to %s", budget)
+		args = setFlag(args, "--parked-request-budget", budget.String())
+		if extra := budget - manifestParkedRequestBudget; extra > 0 {
+			grace, err := gracePeriod(dep)
+			if err != nil {
+				return err
+			}
+			grace += int64(math.Ceil(extra.Seconds()))
+			if err := unstructured.SetNestedField(dep.Object, grace, "spec", "template", "spec", "terminationGracePeriodSeconds"); err != nil {
+				return fmt.Errorf("%s: %w", kube.Describe(dep), err)
+			}
+		}
+	}
+	if maxRequests > 0 {
+		log.Infof("Setting the router's --extproc-max-requests to %d", maxRequests)
+		args = setFlag(args, "--extproc-max-requests", strconv.Itoa(maxRequests))
+	}
+	container["args"] = args
+	containers[idx] = container
+	return unstructured.SetNestedSlice(dep.Object, containers, "spec", "template", "spec", "containers")
+}
+
+// setFlag replaces the --flag=value argument in args, or appends it.
+func setFlag(args []any, flag, value string) []any {
+	arg := flag + "=" + value
+	for i, a := range args {
+		if s, ok := a.(string); ok && strings.HasPrefix(s, flag+"=") {
+			args[i] = arg
+			return args
+		}
+	}
+	return append(args, arg)
+}
+
+// gracePeriod reads terminationGracePeriodSeconds; YAML decodes it as float64, JSON as int64.
+func gracePeriod(dep *unstructured.Unstructured) (int64, error) {
+	raw, found, err := unstructured.NestedFieldNoCopy(dep.Object, "spec", "template", "spec", "terminationGracePeriodSeconds")
+	if err != nil {
+		return 0, fmt.Errorf("%s: %w", kube.Describe(dep), err)
+	}
+	if !found {
+		return 30, nil // the Kubernetes default
+	}
+	switch v := raw.(type) {
+	case int64:
+		return v, nil
+	case float64:
+		return int64(v), nil
+	default:
+		return 0, fmt.Errorf("%s: terminationGracePeriodSeconds is %T, want a number", kube.Describe(dep), raw)
+	}
+}