Skip to content

Commit 19084f8

Browse files
perf(bench): end-to-end overhead harness and Phase 0 gate results (#50)
* perf: add phase 0 end-to-end benchmark Record the missed p99 gate without production tuning. Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * perf(bench): measure fixed-arrival overhead Keep scheduled timestamps so queueing and late sends remain visible. Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent d740dc6 commit 19084f8

10 files changed

Lines changed: 1197 additions & 1 deletion

File tree

‎.gitignore‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,3 +28,6 @@ vendor/
2828
# Internal planning docs (not OSS-ready)
2929
/PRD-*.md
3030
dist/
31+
32+
# Local benchmark data, profiles, and Go caches
33+
/.cache/

‎Makefile‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
.PHONY: build build-verify run test docker lint clean
1+
.PHONY: build build-verify run test docker lint clean bench-e2e bench-e2e-open-loop
22

33
build:
44
CGO_ENABLED=0 go build -o bin/agentgate ./cmd/agentgw
@@ -12,6 +12,17 @@ run:
1212
test:
1313
CGO_ENABLED=0 go test -v -count=1 ./...
1414

15+
BENCHTIME ?= 20000x
16+
BENCH_FILTER ?= ^BenchmarkHandleActE2E$$
17+
18+
bench-e2e-open-loop: BENCH_FILTER = ^BenchmarkHandleActOpenLoop$$
19+
20+
bench-e2e bench-e2e-open-loop:
21+
mkdir -p .cache/tmp .cache/go-tmp .cache/go-build
22+
TMPDIR="$(CURDIR)/.cache/tmp" GOTMPDIR="$(CURDIR)/.cache/go-tmp" \
23+
GOCACHE="$(CURDIR)/.cache/go-build" GOMAXPROCS=4 \
24+
go test -tags=bench ./bench -run '^$$' -bench '$(BENCH_FILTER)' -benchtime=$(BENCHTIME) -count=1
25+
1526
docker:
1627
docker-compose up --build
1728

‎bench/RESULTS.md‎

Lines changed: 366 additions & 0 deletions
Large diffs are not rendered by default.

‎bench/e2e_test.go‎

Lines changed: 359 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,359 @@
1+
//go:build bench
2+
3+
package bench
4+
5+
import (
6+
"bytes"
7+
"context"
8+
"crypto/rand"
9+
"database/sql"
10+
"encoding/json"
11+
"fmt"
12+
"io"
13+
"log/slog"
14+
"math"
15+
"net/http"
16+
"net/http/httptest"
17+
"path/filepath"
18+
"runtime"
19+
"slices"
20+
"strings"
21+
"sync"
22+
"testing"
23+
"time"
24+
25+
"github.com/Clawdlinux/agentgate/internal/auth"
26+
agentgatedb "github.com/Clawdlinux/agentgate/internal/db"
27+
"github.com/Clawdlinux/agentgate/internal/gateway"
28+
"github.com/Clawdlinux/agentgate/internal/ratelimit"
29+
"github.com/Clawdlinux/agentgate/internal/receipt"
30+
"github.com/Clawdlinux/agentgate/internal/registry"
31+
"github.com/Clawdlinux/agentgate/internal/signer"
32+
"github.com/Clawdlinux/agentgate/internal/vault"
33+
)
34+
35+
const warmupRequests = 256
36+
37+
type stack struct {
38+
db *sql.DB
39+
ledger *receipt.Ledger
40+
client *http.Client
41+
url string
42+
key string
43+
upstream []chan time.Duration
44+
}
45+
46+
type stackOptions struct {
47+
timingSlots int
48+
spinUpstream bool
49+
}
50+
51+
type benchmarkLogWriter struct {
52+
b *testing.B
53+
}
54+
55+
func (w benchmarkLogWriter) Write(data []byte) (int, error) {
56+
if bytes.Contains(data, []byte(`"level":"ERROR"`)) {
57+
w.b.Logf("gateway error: %s", data)
58+
}
59+
return len(data), nil
60+
}
61+
62+
func newStack(b *testing.B, concurrency int, delay time.Duration) *stack {
63+
b.Helper()
64+
return newStackWithOptions(b, concurrency, delay, stackOptions{})
65+
}
66+
67+
func newStackWithOptions(b *testing.B, concurrency int, delay time.Duration, options stackOptions) *stack {
68+
b.Helper()
69+
database, err := agentgatedb.Open(filepath.Join(b.TempDir(), "agentgate.db"))
70+
must(b, err)
71+
b.Cleanup(func() { database.Close() })
72+
must(b, agentgatedb.RunMigrations(database))
73+
74+
masterKey := make([]byte, 32)
75+
_, err = rand.Read(masterKey)
76+
must(b, err)
77+
pepper, err := auth.DeriveKeyPepper(masterKey)
78+
must(b, err)
79+
keys, err := auth.NewKeyStore(database, pepper)
80+
must(b, err)
81+
_, plaintext, err := keys.Create(context.Background(), "e2e-bench", []string{"mock"}, []string{"bench-user"})
82+
must(b, err)
83+
store, err := vault.NewSQLiteStore(database, masterKey)
84+
must(b, err)
85+
cached, err := vault.NewCachedStore(store)
86+
must(b, err)
87+
token := make([]byte, 32)
88+
_, err = rand.Read(token)
89+
must(b, err)
90+
credential := fmt.Sprintf("%x", token)
91+
must(b, cached.Put("bench-user", "mock", vault.Token{AccessToken: credential, TokenType: "Bearer"}))
92+
93+
signers, err := signer.NewStore(database, masterKey)
94+
must(b, err)
95+
b.Cleanup(signers.Close)
96+
_, signingKey, err := signers.LoadOrCreateActive(1)
97+
clear(signingKey)
98+
must(b, err)
99+
ledger := receipt.NewLedger(database, signers)
100+
b.Cleanup(ledger.Close)
101+
102+
timings := make([]chan time.Duration, max(concurrency, options.timingSlots))
103+
for i := range timings {
104+
timings[i] = make(chan time.Duration, 1)
105+
}
106+
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
107+
start := time.Now()
108+
var params struct {
109+
Worker int `json:"worker"`
110+
}
111+
if r.Method != http.MethodPost || r.Header.Get("Authorization") != "Bearer "+credential {
112+
http.Error(w, "unexpected upstream request", http.StatusBadRequest)
113+
return
114+
}
115+
if err := json.NewDecoder(r.Body).Decode(&params); err != nil || params.Worker < 0 || params.Worker >= len(timings) {
116+
http.Error(w, "invalid worker", http.StatusBadRequest)
117+
return
118+
}
119+
// Closed-loop workers serialize requests. Open-loop requests each have their own slot.
120+
defer func() { timings[params.Worker] <- time.Since(start) }()
121+
if delay > 0 {
122+
if options.spinUpstream {
123+
until := time.Now().Add(delay)
124+
for time.Now().Before(until) {
125+
}
126+
} else {
127+
time.Sleep(delay)
128+
}
129+
}
130+
w.Header().Set("Content-Type", "application/json")
131+
io.WriteString(w, `{"ok":true}`)
132+
}))
133+
b.Cleanup(upstream.Close)
134+
135+
reg := registry.New()
136+
must(b, reg.LoadBytes([]byte(fmt.Sprintf(`services:
137+
mock:
138+
base_url: %s
139+
auth:
140+
type: bearer
141+
rate_limit:
142+
requests_per_second: 1000000
143+
burst: 1000000
144+
actions:
145+
echo:
146+
method: POST
147+
path: /echo
148+
params:
149+
worker: int
150+
`, upstream.URL))))
151+
limiter := ratelimit.New(reg.RateLimits(), ratelimit.WithSettings(ratelimit.Settings{
152+
KeyRPS: 1e6, KeyBurst: 1e6, TenantRPS: 1e6, TenantBurst: 1e6, TenantMaxInflight: 1024,
153+
}))
154+
handler := gateway.New(gateway.Config{
155+
Registry: reg, Vault: cached, Authorizer: keys, Receipts: ledger, Limiter: limiter,
156+
Logger: slog.New(slog.NewJSONHandler(benchmarkLogWriter{b}, &slog.HandlerOptions{Level: slog.LevelInfo})),
157+
})
158+
b.Cleanup(handler.CloseIdleConnections)
159+
server := httptest.NewServer(handler)
160+
b.Cleanup(server.Close)
161+
transport := &http.Transport{
162+
MaxIdleConns: concurrency, MaxIdleConnsPerHost: concurrency, MaxConnsPerHost: concurrency,
163+
}
164+
b.Cleanup(transport.CloseIdleConnections)
165+
return &stack{
166+
db: database, ledger: ledger, client: &http.Client{Transport: transport, Timeout: 10 * time.Second},
167+
url: server.URL + "/v1/act", key: plaintext, upstream: timings,
168+
}
169+
}
170+
171+
type sample struct {
172+
total time.Duration
173+
upstream time.Duration
174+
dispatchLag time.Duration
175+
}
176+
177+
func (s *stack) act(ctx context.Context, worker int, body string) (sample, error) {
178+
return s.actAt(ctx, worker, body, time.Time{})
179+
}
180+
181+
func (s *stack) actAt(ctx context.Context, worker int, body string, scheduled time.Time) (sample, error) {
182+
req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.url, strings.NewReader(body))
183+
if err != nil {
184+
return sample{}, err
185+
}
186+
req.Header.Set("Authorization", "Bearer "+s.key)
187+
req.Header.Set("Content-Type", "application/json")
188+
start := time.Now()
189+
resp, err := s.client.Do(req)
190+
if err != nil {
191+
return sample{}, err
192+
}
193+
data, err := io.ReadAll(resp.Body)
194+
resp.Body.Close()
195+
finished := time.Now()
196+
total := finished.Sub(start)
197+
if err != nil {
198+
return sample{}, err
199+
}
200+
if resp.StatusCode != http.StatusOK {
201+
return sample{}, fmt.Errorf("act status %d: %s", resp.StatusCode, data)
202+
}
203+
var envelope struct {
204+
Status int `json:"status"`
205+
Body struct {
206+
OK bool `json:"ok"`
207+
} `json:"body"`
208+
}
209+
if err := json.Unmarshal(data, &envelope); err != nil {
210+
return sample{}, err
211+
}
212+
if envelope.Status != http.StatusOK || !envelope.Body.OK {
213+
return sample{}, fmt.Errorf("unexpected act response: %s", data)
214+
}
215+
select {
216+
case duration := <-s.upstream[worker]:
217+
if duration > total {
218+
return sample{}, fmt.Errorf("upstream time %s exceeds total %s", duration, total)
219+
}
220+
return timedSample(scheduled, start, finished, duration), nil
221+
case <-ctx.Done():
222+
return sample{}, ctx.Err()
223+
}
224+
}
225+
226+
func timedSample(scheduled, started, finished time.Time, upstream time.Duration) sample {
227+
if scheduled.IsZero() {
228+
scheduled = started
229+
}
230+
return sample{
231+
total: finished.Sub(scheduled), upstream: upstream, dispatchLag: started.Sub(scheduled),
232+
}
233+
}
234+
235+
func (s *stack) run(count, concurrency int, samples []sample) error {
236+
ctx, cancel := context.WithCancel(context.Background())
237+
defer cancel()
238+
errors := make(chan error, concurrency)
239+
start := make(chan struct{})
240+
var ready, done sync.WaitGroup
241+
for worker := range min(count, concurrency) {
242+
ready.Add(1)
243+
done.Add(1)
244+
go func() {
245+
defer done.Done()
246+
body := fmt.Sprintf(`{"service":"mock","action":"echo","on_behalf_of":"bench-user","params":{"worker":%d}}`, worker)
247+
ready.Done()
248+
<-start
249+
for i := worker; i < count; i += concurrency {
250+
result, err := s.act(ctx, worker, body)
251+
if err != nil {
252+
errors <- err
253+
cancel()
254+
return
255+
}
256+
if samples != nil {
257+
samples[i] = result
258+
}
259+
}
260+
}()
261+
}
262+
ready.Wait()
263+
close(start)
264+
done.Wait()
265+
close(errors)
266+
for err := range errors {
267+
return err
268+
}
269+
return nil
270+
}
271+
272+
func BenchmarkHandleActE2E(b *testing.B) {
273+
if runtime.GOMAXPROCS(0) != 4 {
274+
b.Fatal("run with GOMAXPROCS=4 for the Phase 0 comparison")
275+
}
276+
for _, delay := range []time.Duration{0, 5 * time.Millisecond} {
277+
b.Run("delay="+delay.String(), func(b *testing.B) {
278+
for _, concurrency := range []int{1, 16, 64} {
279+
b.Run(fmt.Sprintf("c=%d", concurrency), func(b *testing.B) {
280+
benchmarkAct(b, concurrency, delay)
281+
})
282+
}
283+
})
284+
}
285+
}
286+
287+
func benchmarkAct(b *testing.B, concurrency int, delay time.Duration) {
288+
s := newStack(b, concurrency, delay)
289+
must(b, s.run(warmupRequests, concurrency, nil))
290+
samples := make([]sample, b.N)
291+
b.ResetTimer()
292+
start := time.Now()
293+
err := s.run(b.N, concurrency, samples)
294+
elapsed := time.Since(start)
295+
b.StopTimer()
296+
must(b, err)
297+
s.checkReceipts(b, warmupRequests+b.N)
298+
reportSamples(b, samples, elapsed)
299+
}
300+
301+
func (s *stack) checkReceipts(b *testing.B, expected int) {
302+
b.Helper()
303+
var count, invalid int
304+
must(b, s.db.QueryRow(`SELECT COUNT(*), COALESCE(SUM(
305+
CASE WHEN status_code != 200 OR policy_decision != 'allow' OR error_code != ''
306+
OR length(signature) != 64 THEN 1 ELSE 0 END), 0) FROM receipts`).Scan(&count, &invalid))
307+
head, _, err := s.ledger.Head(context.Background())
308+
must(b, err)
309+
if count != expected || head != uint64(count) || invalid != 0 {
310+
b.Fatalf("receipt mismatch: count=%d head=%d invalid=%d expected=%d", count, head, invalid, expected)
311+
}
312+
}
313+
314+
func reportSamples(b *testing.B, samples []sample, elapsed time.Duration) {
315+
b.Helper()
316+
overheads := make([]time.Duration, b.N)
317+
totals := make([]time.Duration, b.N)
318+
var upstreamTotal time.Duration
319+
for i, result := range samples {
320+
overheads[i] = result.total - result.upstream
321+
totals[i] = result.total
322+
upstreamTotal += result.upstream
323+
}
324+
slices.Sort(overheads)
325+
slices.Sort(totals)
326+
for _, p := range []float64{50, 90, 99, 99.9} {
327+
b.ReportMetric(float64(percentile(overheads, p))/float64(time.Microsecond), fmt.Sprintf("p%g-overhead-us", p))
328+
}
329+
b.ReportMetric(float64(percentile(totals, 99))/float64(time.Microsecond), "p99-total-us")
330+
b.ReportMetric(float64(upstreamTotal)/float64(b.N)/float64(time.Microsecond), "upstream-mean-us")
331+
b.ReportMetric(float64(b.N)/elapsed.Seconds(), "req/s")
332+
}
333+
334+
// percentile uses the nearest rank. Subtract upstream time before sorting, never between independently sorted percentiles.
335+
func percentile(sorted []time.Duration, p float64) time.Duration {
336+
return sorted[max(0, int(math.Ceil(p*float64(len(sorted))/100))-1)]
337+
}
338+
339+
func must(b *testing.B, err error) {
340+
b.Helper()
341+
if err != nil {
342+
b.Fatal(err)
343+
}
344+
}
345+
346+
func TestPercentile(t *testing.T) {
347+
samples := make([]time.Duration, 1000)
348+
for i := range samples {
349+
samples[i] = time.Duration(i + 1)
350+
}
351+
for _, p := range []float64{50, 90, 99, 99.9} {
352+
if got, want := percentile(samples, p), time.Duration(math.Round(p*10)); got != want {
353+
t.Errorf("p%g: got %d, want %d", p, got, want)
354+
}
355+
}
356+
if got := percentile([]time.Duration{42}, 99.9); got != 42 {
357+
t.Errorf("single sample: got %d, want 42", got)
358+
}
359+
}

0 commit comments

Comments
 (0)