From f70a5d1f3101fa7c550606310f0383a3faa836cd Mon Sep 17 00:00:00 2001 From: Chuang Wang Date: Fri, 7 Aug 2026 17:48:29 -0700 Subject: [PATCH] atenet-router: raise actor-cluster circuit breakers The actor cluster configured no breakers, leaving Envoy's 1,024 default, which caps concurrent requests through the router far below what it can carry. Connections, pending and requests rise to 20,000, kept under the ~28k source-port budget so overload trips a counted breaker instead of the kernel's opaque EADDRNOTAVAIL; max_retries keeps its default since nothing here retries. The ext_proc breaker now applies its ceiling to max_pending_requests as well as max_requests. The manifest also enables dispatcher stats and sets router pod resources: small requests so the pod schedules anywhere, 8-core CPU limits as the real cap. --- cmd/atenet/internal/router/cmd.go | 2 +- cmd/atenet/internal/router/xds.go | 51 +++++++++++++++++------ cmd/atenet/internal/router/xds_test.go | 53 ++++++++++++++++++++++++ manifests/ate-install/atenet-router.yaml | 25 +++++++++++ 4 files changed, 117 insertions(+), 14 deletions(-) diff --git a/cmd/atenet/internal/router/cmd.go b/cmd/atenet/internal/router/cmd.go index 018fb18d0..1d4aa0b38 100644 --- a/cmd/atenet/internal/router/cmd.go +++ b/cmd/atenet/internal/router/cmd.go @@ -79,7 +79,7 @@ func NewRouterCmd() *cobra.Command { cmd.Flags().DurationVar(&cfg.ParkedRequest.RetryInterval, "parked-request-retry-interval", ingress.DefaultParkedRequestRetryInterval, "Delay before a parked request's first resume retry") cmd.Flags().Float64Var(&cfg.ParkedRequest.RetryFactor, "parked-request-retry-factor", ingress.DefaultParkedRequestRetryFactor, "Multiplier applied to the retry delay after each attempt; must be >= 1") cmd.Flags().Float64Var(&cfg.ParkedRequest.RetryJitter, "parked-request-retry-jitter", ingress.DefaultParkedRequestRetryJitter, "Random fraction in [0, 1) added to each retry delay to de-synchronize parked requests") - cmd.Flags().IntVar(&cfg.ExtProcMaxRequests, "extproc-max-requests", 0, "Circuit-breaker max_requests for Envoy's ext_proc cluster; 0 (the default) derives it as twice --parked-request-max (minimum 1024). Explicit values must be >= --parked-request-max: every parked request holds one slot for its full wait, and the excess is fast-path headroom") + cmd.Flags().IntVar(&cfg.ExtProcMaxRequests, "extproc-max-requests", 0, "Circuit-breaker ceiling for Envoy's ext_proc cluster, applied to max_requests and max_pending_requests alike; 0 (the default) derives it as twice --parked-request-max (minimum 1024). Explicit values must be >= --parked-request-max: every parked request holds one slot for its full wait, and the excess is fast-path headroom") // Graceful shutdown knobs. The router sits behind a Service, so // route-drain window is needed: after SIGTERM the readiness flip // must propagate to the Service endpoints before the drain starts. diff --git a/cmd/atenet/internal/router/xds.go b/cmd/atenet/internal/router/xds.go index d2ec58dec..7690c1d6c 100644 --- a/cmd/atenet/internal/router/xds.go +++ b/cmd/atenet/internal/router/xds.go @@ -89,10 +89,11 @@ const ( // otherwise Envoy abandons a parked request (500) long before the router does. const defaultExtProcMessageTimeout = 5 * time.Second -// defaultExtProcMaxRequests is the circuit-breaker max_requests set on the -// ext_proc cluster: ingress.DefaultParkedRequestMax plus equal fast-path headroom, so a -// full parking lot cannot starve the millisecond-scale header exchanges of -// requests to already-running actors. See buildCluster. +// defaultExtProcMaxRequests is the circuit-breaker ceiling set on the ext_proc +// cluster, applied to both max_requests and max_pending_requests: +// ingress.DefaultParkedRequestMax plus equal fast-path headroom, so a full +// parking lot cannot starve the millisecond-scale header exchanges of requests +// to already-running actors. See buildCluster. const defaultExtProcMaxRequests = 2048 // defaultRouteTimeout is Envoy's end-to-end route timeout for workload traffic: @@ -121,6 +122,12 @@ const defaultRouteTimeout = 10 * time.Second // routeIdleTimeout. const envoyDefaultStreamIdleTimeout = 5 * time.Minute +// actorClusterMaxConcurrency replaces Envoy's 1024 default, which is far below +// what the router can carry. Kept under the ~28,232 ephemeral-port budget +// (each in-flight HTTP/1.1 request holds one source port) so the breaker's +// counted overflow trips before the kernel's opaque EADDRNOTAVAIL. +const actorClusterMaxConcurrency = 20000 + // XdsServer implements an aggregated discovery service server for dynamic Envoy router nodes. type XdsServer struct { xdsPort int @@ -160,10 +167,11 @@ type XdsServer struct { // response. Must be >= the parking budget so parked requests aren't cut short. extProcMessageTimeout time.Duration - // extProcMaxRequests is the circuit-breaker max_requests on the ext_proc - // cluster — the hard ceiling on concurrent requests held open against the - // router's processing server, parked requests included. Must be >= the - // parking lot size (enforced at startup in Run). + // extProcMaxRequests is the circuit-breaker ceiling on the ext_proc cluster + // — the hard limit on concurrent requests held open against the router's + // processing server, parked requests included. Applied to max_requests and + // max_pending_requests alike. Must be >= the parking lot size (enforced at + // startup in Run). extProcMaxRequests uint32 // routeTimeout is Envoy's end-to-end timeout on the workload route. Actors @@ -209,9 +217,10 @@ func (x *XdsServer) SetExtProcMessageTimeout(d time.Duration) { } } -// SetExtProcMaxRequests sets the circuit-breaker max_requests on the ext_proc -// cluster. Size it to the parking lot plus fast-path headroom (validated in -// Run()); a non-positive value leaves the default unchanged. +// SetExtProcMaxRequests sets the circuit-breaker ceiling on the ext_proc +// cluster, for max_requests and max_pending_requests together. Size it to the +// parking lot plus fast-path headroom (validated in Run()); a non-positive +// value leaves the default unchanged. func (x *XdsServer) SetExtProcMaxRequests(n int) { x.mu.Lock() defer x.mu.Unlock() @@ -475,10 +484,14 @@ func (x *XdsServer) buildCluster() *clusterv3.Cluster { Type: clusterv3.Cluster_STATIC, }, LbPolicy: clusterv3.Cluster_ROUND_ROBIN, + // max_pending_requests rises with max_requests: a request is pending + // until the pool hands it a stream, and a shallower pending queue just + // moves the rejection. CircuitBreakers: &clusterv3.CircuitBreakers{ Thresholds: []*clusterv3.CircuitBreakers_Thresholds{{ - Priority: corev3.RoutingPriority_DEFAULT, - MaxRequests: wrapperspb.UInt32(x.extProcMaxRequests), + Priority: corev3.RoutingPriority_DEFAULT, + MaxRequests: wrapperspb.UInt32(x.extProcMaxRequests), + MaxPendingRequests: wrapperspb.UInt32(x.extProcMaxRequests), }}, }, LoadAssignment: &endpointv3.ClusterLoadAssignment{ @@ -639,6 +652,18 @@ func (x *XdsServer) buildOriginalDstCluster() *clusterv3.Cluster { HttpHeaderName: ingress.OriginalDstHeader, }, }, + // Connections, pending and requests are lifted together: the upstream + // hop is HTTP/1.1, so capping any one below the others just moves + // where the queue forms. max_retries keeps Envoy's default — no route + // to this cluster sets a retry policy. + CircuitBreakers: &clusterv3.CircuitBreakers{ + Thresholds: []*clusterv3.CircuitBreakers_Thresholds{{ + Priority: corev3.RoutingPriority_DEFAULT, + MaxConnections: wrapperspb.UInt32(actorClusterMaxConcurrency), + MaxPendingRequests: wrapperspb.UInt32(actorClusterMaxConcurrency), + MaxRequests: wrapperspb.UInt32(actorClusterMaxConcurrency), + }}, + }, } if ts := x.buildUpstreamTransportSocket(); ts != nil { diff --git a/cmd/atenet/internal/router/xds_test.go b/cmd/atenet/internal/router/xds_test.go index 98287eae8..7f1be3530 100644 --- a/cmd/atenet/internal/router/xds_test.go +++ b/cmd/atenet/internal/router/xds_test.go @@ -564,6 +564,59 @@ func TestXdsServer_ExtProcCircuitBreaker(t *testing.T) { t.Errorf("max_requests after SetExtProcMaxRequests(0) = %d, want default %d", got, defaultExtProcMaxRequests) } }) + + // A request is pending until the pool hands it a stream, so the shallower + // of the two breakers is the one a burst actually meets. + t.Run("PendingTracksMaxRequests", func(t *testing.T) { + for _, set := range []int{0, 4096, 20000} { + x := NewXdsServer(0) + x.SetExtProcMaxRequests(set) + th := x.buildCluster().GetCircuitBreakers().GetThresholds()[0] + req, pend := th.GetMaxRequests().GetValue(), th.GetMaxPendingRequests().GetValue() + if pend != req { + t.Errorf("SetExtProcMaxRequests(%d): max_pending_requests = %d, max_requests = %d; the shallower one is the real ceiling", + set, pend, req) + } + } + }) +} + +func TestXdsServer_ActorClusterCircuitBreaker(t *testing.T) { + thresholds := NewXdsServer(0).buildOriginalDstCluster().GetCircuitBreakers().GetThresholds() + if len(thresholds) != 1 { + t.Fatalf("got %d thresholds, want 1 (on the default priority)", len(thresholds)) + } + th := thresholds[0] + if got := th.GetPriority(); got != corev3.RoutingPriority_DEFAULT { + t.Errorf("priority = %v, want DEFAULT: actor traffic carries no priority header, so a HIGH threshold would never apply", got) + } + + // Connections, pending and requests rise together; any one left at the + // 1024 default just relocates the queue. max_retries stays unset: no + // route to this cluster retries, and a 20k retry budget would be noise. + for _, tc := range []struct { + name string + got uint32 + }{ + {"max_connections", th.GetMaxConnections().GetValue()}, + {"max_pending_requests", th.GetMaxPendingRequests().GetValue()}, + {"max_requests", th.GetMaxRequests().GetValue()}, + } { + if tc.got != uint32(actorClusterMaxConcurrency) { + t.Errorf("%s = %d, want %d", tc.name, tc.got, actorClusterMaxConcurrency) + } + } + if th.GetMaxRetries() != nil { + t.Errorf("max_retries = %d, want unset (Envoy default): nothing on this cluster retries", th.GetMaxRetries().GetValue()) + } + + // Each in-flight HTTP/1.1 request holds one ephemeral port; a breaker + // above the port budget would let the kernel fail first, uncounted. + const defaultEphemeralPorts = 60999 - 32768 + 1 + if actorClusterMaxConcurrency >= defaultEphemeralPorts { + t.Errorf("actorClusterMaxConcurrency (%d) >= the default ephemeral port range (%d): the kernel, not the breaker, becomes the binding limit", + actorClusterMaxConcurrency, defaultEphemeralPorts) + } } func TestXdsServer_RouteTimeout(t *testing.T) { diff --git a/manifests/ate-install/atenet-router.yaml b/manifests/ate-install/atenet-router.yaml index 1dc2e3fe1..9928aa7a9 100644 --- a/manifests/ate-install/atenet-router.yaml +++ b/manifests/ate-install/atenet-router.yaml @@ -89,6 +89,11 @@ data: address: 0.0.0.0 port_value: 9901 + # Per-worker event-loop duration histograms; the documented overhead is + # accepted for the observability. + enable_dispatcher_stats: true + + node: id: substrate-envoy-node cluster: substrate-router-cluster @@ -195,6 +200,16 @@ spec: # turns to survive a shutdown, raise --drain-timeout and # terminationGracePeriodSeconds alongside it. # - "--route-timeout=5m" + resources: + # Small requests so the pod schedules anywhere (including CI kind + # nodes); the limits are the real cap. Scheduling considers only + # requests, so on smaller nodes the limit simply never binds. + requests: + cpu: 250m + memory: 256Mi + limits: + cpu: "8" + memory: 2Gi env: - name: POD_NAME valueFrom: @@ -269,6 +284,16 @@ spec: preStop: exec: command: ["sh", "-c", "while [ ! -f /var/run/atenet/drain-complete ]; do sleep 0.5; done"] + resources: + # Small requests, real cap in limits (see the sidecar's note). Do + # not raise the CPU limit past 8 cores: measured capacity per core + # falls steeply beyond that — scale by adding replicas instead. + requests: + cpu: 250m + memory: 256Mi + limits: + cpu: "8" + memory: 4Gi ports: - name: http containerPort: 8080