Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 8 additions & 7 deletions cmd/atenet/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
atenet is a combined daemon for all networking functionality.

* Envoy control plane for programming ATE resolution. `atenet router`
* SDS server that mints the egress gateway's leaf certificates. `atenet sdsmint`

This is built as a single binary for convenience in the prototyping.

Expand All @@ -23,13 +24,13 @@ likely be split in the future for better scalability.)
reaches a port on the actor other than its default (80) with an HTTP
`CONNECT` request naming the port in the authority. Only HTTP(S) traffic
over the tunnel is supported today, not raw TCP.
* Upstream: Envoy's `ORIGINAL_DST` actor cluster dials the actor's in-worker
`atunnel` ingress server on the worker pod's port 443 over mTLS, using the
address `atenet router`'s ext_proc resolves into dynamic metadata rather
than a header. `atunnel` can't read that metadata directly, though -- it's
a separate process on the worker pod, not part of Envoy -- so the port to
reach on the actor itself (its default port, or an arbitrary one for
CONNECT) still travels as a real header, `TargetPortHeader` in `internal/atunnel/ingress`.
* Upstream: Envoy reaches the actor's in-worker `atunnel` at the address
`atenet router`'s ext_proc resolves into dynamic metadata rather than a
header. By default (`--upstream-connect`) it opens an mTLS CONNECT tunnel to
the worker pod's port 8443, and the port to reach on the actor rides in the
CONNECT authority. With `--upstream-connect=false` it dials `atunnel`'s
ingress server on port 443 over mTLS instead, and the port travels as the
`TargetPortHeader` header (`internal/atunnel/ingress`).
`ate-target-actor: <atespace>/<actor>` identifies the Actor independently of
`:authority`/`Host`.
* Termination: the router drains gracefully on SIGTERM (readiness flip →
Expand Down
138 changes: 65 additions & 73 deletions cmd/atenet/internal/router/drain_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,12 @@ import (
"net/http"
"net/http/httptest"
"os"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"testing/synctest"
"time"

"go.ate.dev/substrate/internal/serverboot"
Expand Down Expand Up @@ -76,66 +78,68 @@ func (r *orderRecorder) get() []string {
// order and that an in-flight ext_proc stream releasing promptly lets the
// drain complete without a force stop.
func TestDrainOnShutdownOrdering(t *testing.T) {
rec := &orderRecorder{}
stopper := newFakeStopper()
readiness := &serverboot.Readiness{}

// Release the "parked request" shortly after GracefulStop begins.
go func() {
for !stopper.gracefulCalled.Load() {
time.Sleep(5 * time.Millisecond)
}
rec.add("extproc-drain-started")
time.Sleep(20 * time.Millisecond)
close(stopper.release)
}()

ctx, cancel := context.WithCancel(context.Background())
done := drainOnShutdown(ctx, drainParams{
readiness: readiness,
delay: 0,
dataplane: drainFunc(func(context.Context) error {
if readiness.Ready() {
t.Error("dataplane drain ran before the readiness flip")
synctest.Test(t, func(t *testing.T) {
rec := &orderRecorder{}
stopper := newFakeStopper()
readiness := &serverboot.Readiness{}

// Release the "parked request" shortly after GracefulStop begins.
go func() {
for !stopper.gracefulCalled.Load() {
time.Sleep(5 * time.Millisecond)
}
rec.add("dataplane")
return nil
}),
dataplaneWindow: time.Second,
extproc: stopper,
timeout: 5 * time.Second,
stopRest: func() {
if !stopper.gracefulCalled.Load() {
t.Error("stopRest ran before the ext_proc drain")
}
rec.add("stopRest")
},
})

cancel() // simulate SIGTERM

select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("drain did not complete")
}
rec.add("extproc-drain-started")
time.Sleep(20 * time.Millisecond)
close(stopper.release)
}()

ctx, cancel := context.WithCancel(context.Background())
done := drainOnShutdown(ctx, drainParams{
readiness: readiness,
delay: 0,
dataplane: drainFunc(func(context.Context) error {
if readiness.Ready() {
t.Error("dataplane drain ran before the readiness flip")
}
rec.add("dataplane")
return nil
}),
dataplaneWindow: time.Second,
extproc: stopper,
timeout: 5 * time.Second,
stopRest: func() {
if !stopper.gracefulCalled.Load() {
t.Error("stopRest ran before the ext_proc drain")
}
rec.add("stopRest")
},
})

cancel() // simulate SIGTERM

select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("drain did not complete")
}

want := []string{"dataplane", "extproc-drain-started", "stopRest"}
got := rec.get()
if len(got) != len(want) {
t.Fatalf("drain steps = %v, want %v", got, want)
}
for i := range want {
if got[i] != want[i] {
want := []string{"dataplane", "extproc-drain-started", "stopRest"}
got := rec.get()
if len(got) != len(want) {
t.Fatalf("drain steps = %v, want %v", got, want)
}
}
if stopper.stopCalled.Load() {
t.Error("force Stop fired although the graceful drain completed in time")
}
if readiness.Ready() {
t.Error("readiness should be not-ready after drain")
}
for i := range want {
if got[i] != want[i] {
t.Fatalf("drain steps = %v, want %v", got, want)
}
}
if stopper.stopCalled.Load() {
t.Error("force Stop fired although the graceful drain completed in time")
}
if readiness.Ready() {
t.Error("readiness should be not-ready after drain")
}
})
}

// TestDrainOnShutdownForceStopsAfterTimeout asserts a stream that never
Expand Down Expand Up @@ -230,24 +234,12 @@ func (f *fakeEnvoyAdmin) handler() http.HandlerFunc {
// Includes an admin line that must be excluded from the sum.
w.Write([]byte(
"listener.admin.downstream_cx_active: 1\n" +
"listener.0.0.0.0_8080.downstream_cx_active: " + itoa(active) + "\n" +
"listener.0.0.0.0_8080.downstream_cx_active: " + strconv.Itoa(active) + "\n" +
"http.ingress_http.downstream_cx_active_unrelated: 99\n"))
}
}
}

func itoa(n int) string {
if n == 0 {
return "0"
}
var b []byte
for n > 0 {
b = append([]byte{byte('0' + n%10)}, b...)
n /= 10
}
return string(b)
}

// TestEnvoyDrainerDrainsToZero asserts the drainer POSTs the two admin calls
// and polls stats until active connections reach zero, excluding admin
// listeners from the count.
Expand Down Expand Up @@ -282,9 +274,9 @@ func TestEnvoyDrainerDrainsToZero(t *testing.T) {
}

// TestEnvoyDrainerReachesIPv6OnlyAdmin guards the loopback coupling behind
// --envoy-admin-address: the gateway admin sockets bind "::", and a drainer
// pinned to the IPv4 loopback would find nothing listening, read that as an
// exited Envoy, and report a drain it never performed -- silently, since that
// --envoy-admin-address: a drainer pinned to the IPv4 loopback would find
// nothing listening on an admin bound to the IPv6 one, read that as an exited
// Envoy, and report a drain it never performed -- silently, since that
// path returns nil. Hence the assertion on the POSTs rather than on the error.
func TestEnvoyDrainerReachesIPv6OnlyAdmin(t *testing.T) {
ln, err := net.Listen("tcp6", "[::1]:0")
Expand All @@ -298,7 +290,7 @@ func TestEnvoyDrainerReachesIPv6OnlyAdmin(t *testing.T) {
srv.Start()
defer srv.Close()

d := newEnvoyDrainer(net.JoinHostPort("localhost", itoa(ln.Addr().(*net.TCPAddr).Port)))
d := newEnvoyDrainer(net.JoinHostPort("localhost", strconv.Itoa(ln.Addr().(*net.TCPAddr).Port)))
d.pollInterval = 5 * time.Millisecond

ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ func TestProcessReturnsWebSocketDenialAsImmediateResponse(t *testing.T) {
},
}
stream := &websocketProcessStream{ctx: context.Background(), requests: []*extprocv3.ProcessingRequest{request}}
server := extproc.NewServer(0, nil, extproc.Handlers{
server := extproc.NewServer(nil, extproc.Handlers{
extproc.DirectionEgress: egress.New(nil, nil, 0, nil, ""),
})

Expand Down
14 changes: 3 additions & 11 deletions cmd/atenet/internal/router/extproc/extproc.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,6 @@ type Handlers map[Direction]Handler
// Server implements the external processing gRPC server, dispatching each
// request to the Handler for the direction it arrived on.
type Server struct {
port int
handlers Handlers
recorder *QueryRecorder
routeDuration metric.Float64Histogram
Expand All @@ -54,23 +53,16 @@ type Server struct {
// NewServer builds the ext_proc mux serving the given handlers. Passing a
// subset of the directions is how --mode restricts an instance to the traffic
// its deployment fronts.
func NewServer(port int, routeDuration metric.Float64Histogram, handlers Handlers) *Server {
func NewServer(routeDuration metric.Float64Histogram, handlers Handlers) *Server {
return &Server{
port: port,
handlers: handlers,
recorder: NewQueryRecorder(100),
routeDuration: routeDuration,
}
}

// Queries returns the most recently processed requests, newest first, for the
// /statusz page.
func (s *Server) Queries() []RecordedQuery {
return s.recorder.Get()
}

// Recorder exposes the query ring buffer so tests and the status page can seed
// or read it.
// Recorder is the ring of recently processed requests behind the /statusz
// page.
func (s *Server) Recorder() *QueryRecorder { return s.recorder }

// NewGRPCServer builds the gRPC server with the ext_proc service registered.
Expand Down
2 changes: 1 addition & 1 deletion cmd/atenet/internal/router/extproc/extproc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ func TestProcessRequestHeadersDispatchesByMode(t *testing.T) {
handlers[d] = stubs[d]
}

s := NewServer(50051, nil, handlers)
s := NewServer(nil, handlers)
req := connectRequest("envoy.filters.http.ext_proc", tc.chain)
resp := s.processRequestHeaders(context.Background(), req, req.GetRequestHeaders())

Expand Down
4 changes: 2 additions & 2 deletions cmd/atenet/internal/router/extproc/metrics_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ func TestRecordRouteDuration_Attributes(t *testing.T) {
t.Fatalf("failed to create histogram: %v", err)
}

s := NewServer(50051, h, nil)
s := NewServer(h, nil)
s.recordRouteDuration(context.Background(), 10*time.Millisecond, "team-a-ns", "tmpl-a", classifyOutcome(nil), ateattr.RouterResumeTriggered)

var rm metricdata.ResourceMetrics
Expand Down Expand Up @@ -159,7 +159,7 @@ func TestRecordRouteDuration_NormalizesEmptyTemplateDimensions(t *testing.T) {
t.Fatalf("failed to create histogram: %v", err)
}

s := NewServer(50051, h, nil)
s := NewServer(h, nil)
s.recordRouteDuration(context.Background(), 5*time.Millisecond, "", "", classifyOutcome(errors.New("fail")), ateattr.RouterResumeUnknown)

var rm metricdata.ResourceMetrics
Expand Down
13 changes: 3 additions & 10 deletions cmd/atenet/internal/router/extproc/record.go
Original file line number Diff line number Diff line change
Expand Up @@ -76,18 +76,11 @@ func (qr *QueryRecorder) Get() []RecordedQuery {
return nil
}

// index is 0 until the ring is full, so one walk covers both cases.
res := make([]RecordedQuery, n)
if n < qr.size {
for i := 0; i < n; i++ {
res[i] = qr.queries[n-1-i]
}
} else {
for i := 0; i < n; i++ {
pos := (qr.index - 1 - i + n) % n
res[i] = qr.queries[pos]
}
for i := range res {
res[i] = qr.queries[(qr.index-1-i+n)%n]
}

return res
}

Expand Down
51 changes: 51 additions & 0 deletions cmd/atenet/internal/router/extproc/record_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
// 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 extproc

import (
"slices"
"testing"
)

// The ring returns newest first, whether or not it has wrapped.
func TestQueryRecorderNewestFirst(t *testing.T) {
tests := []struct {
name string
size int
add int
want []string
}{
{name: "empty", size: 3, add: 0},
{name: "partly filled", size: 3, add: 2, want: []string{"2", "1"}},
{name: "exactly full", size: 3, add: 3, want: []string{"3", "2", "1"}},
{name: "wrapped", size: 3, add: 5, want: []string{"5", "4", "3"}},
{name: "wrapped twice", size: 2, add: 5, want: []string{"5", "4"}},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
qr := NewQueryRecorder(tc.size)
for i := 1; i <= tc.add; i++ {
qr.Add(RecordedQuery{Target: string(rune('0' + i))})
}
var got []string
for _, q := range qr.Get() {
got = append(got, q.Target)
}
if !slices.Equal(got, tc.want) {
t.Errorf("Get() targets = %v, want %v", got, tc.want)
}
})
}
}
Loading
Loading