244 lines
8.5 KiB
Go
244 lines
8.5 KiB
Go
package api
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"net/http"
|
|
"runtime"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// sseEvent is one parsed SSE frame: the event name and the JSON payload.
|
|
type sseEvent struct {
|
|
name string
|
|
data string
|
|
}
|
|
|
|
// openSSE opens GET /logs/stream and returns a reader for the frames. The
|
|
// stream stays open until the context the caller passed is cancelled or the
|
|
// body is closed, so every test here has to do one or the other.
|
|
func openSSE(t *testing.T, h *harness, ctx context.Context) (*http.Response, *bufio.Reader) {
|
|
t.Helper()
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.ts.URL+"/api/logs/stream", nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
req.Header.Set("X-API-Key", "rw-key")
|
|
res, err := h.ts.Client().Do(req)
|
|
if err != nil {
|
|
t.Fatalf("open the log stream: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = res.Body.Close() })
|
|
return res, bufio.NewReader(res.Body)
|
|
}
|
|
|
|
// nextEvent reads until a complete frame has arrived. Comment frames (the
|
|
// keep-alive ": ping") and the blank separator line are skipped.
|
|
func nextEvent(t *testing.T, br *bufio.Reader) sseEvent {
|
|
t.Helper()
|
|
var ev sseEvent
|
|
for {
|
|
line, err := br.ReadString('\n')
|
|
if err != nil {
|
|
t.Fatalf("read the log stream: %v", err)
|
|
}
|
|
line = strings.TrimRight(line, "\r\n")
|
|
switch {
|
|
case strings.HasPrefix(line, "event: "):
|
|
ev.name = strings.TrimPrefix(line, "event: ")
|
|
case strings.HasPrefix(line, "data: "):
|
|
ev.data = strings.TrimPrefix(line, "data: ")
|
|
case line == "":
|
|
if ev.name != "" || ev.data != "" {
|
|
return ev
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// freshUnsubSignal gives the SSE test its own one-shot unsubscribe channel.
|
|
// The harness closes a single shared channel through a sync.Once, so without
|
|
// this only the first stream test in the package could observe it.
|
|
func freshUnsubSignal(h *harness) <-chan struct{} {
|
|
h.fake.mu.Lock()
|
|
defer h.fake.mu.Unlock()
|
|
h.fake.unsub = make(chan struct{})
|
|
h.fake.unsubOnce = sync.Once{}
|
|
return h.fake.unsub
|
|
}
|
|
|
|
// TestSSEStreamHeadersAndEvents pins the SSE contract the live log view
|
|
// depends on: the streaming content type, the two headers that stop an
|
|
// intermediary from buffering the stream, a "ready" event so the UI can show it
|
|
// is connected, and exactly one "entry" event per query-log row.
|
|
func TestSSEStreamHeadersAndEvents(t *testing.T) {
|
|
h := newHarness(t, nil)
|
|
unsub := freshUnsubSignal(h)
|
|
// Queued before the stream opens: the fake's channel is buffered, so this
|
|
// is a deterministic ordering and not a race on the subscription.
|
|
want := []LogEntry{
|
|
{ID: 1, Name: "a.example.com.", QTypeStr: "A", RCode: 0, Client: "198.51.100.1"},
|
|
{ID: 2, Name: "b.example.com.", QTypeStr: "AAAA", RCode: 3, Client: "198.51.100.2", Blocked: true},
|
|
{ID: 3, Name: "c.example.com.", QTypeStr: "TXT", RCode: 0, Client: "198.51.100.3"},
|
|
}
|
|
for _, e := range want {
|
|
h.fake.stream <- e
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
res, br := openSSE(t, h, ctx)
|
|
|
|
if got := res.Header.Get("Content-Type"); got != "text/event-stream" {
|
|
t.Errorf("Content-Type = %q, want text/event-stream", got)
|
|
}
|
|
// no-store, not no-cache: this stream is never replayable.
|
|
if got := res.Header.Get("Cache-Control"); got != "no-store" {
|
|
t.Errorf("Cache-Control = %q, want no-store", got)
|
|
}
|
|
// Without this an nginx in front buffers the stream and the live view shows
|
|
// nothing until the buffer fills.
|
|
if got := res.Header.Get("X-Accel-Buffering"); got != "no" {
|
|
t.Errorf("X-Accel-Buffering = %q, want no", got)
|
|
}
|
|
|
|
ready := nextEvent(t, br)
|
|
if ready.name != "ready" {
|
|
t.Fatalf("first event = %q, want ready", ready.name)
|
|
}
|
|
var readyBody map[string]any
|
|
if err := json.Unmarshal([]byte(ready.data), &readyBody); err != nil {
|
|
t.Fatalf("ready payload is not JSON: %v: %q", err, ready.data)
|
|
}
|
|
if readyBody["version"] != "test" {
|
|
t.Errorf("ready payload = %v, want it to carry the version", readyBody)
|
|
}
|
|
for i, e := range want {
|
|
got := nextEvent(t, br)
|
|
if got.name != "entry" {
|
|
t.Fatalf("event %d = %q, want entry", i, got.name)
|
|
}
|
|
var gotEntry LogEntry
|
|
if err := json.Unmarshal([]byte(got.data), &gotEntry); err != nil {
|
|
t.Fatalf("entry payload is not JSON: %v: %q", err, got.data)
|
|
}
|
|
if gotEntry.ID != e.ID || gotEntry.Name != e.Name || gotEntry.Client != e.Client {
|
|
t.Errorf("entry %d = %+v, want %+v", i, gotEntry, e)
|
|
}
|
|
}
|
|
cancel()
|
|
waitFor(t, 2*time.Second, "the subscriber to be released", func() bool {
|
|
return h.fake.subscribers() == 0
|
|
})
|
|
select {
|
|
case <-unsub:
|
|
default:
|
|
t.Error("the handler returned without running the unsubscribe hook")
|
|
}
|
|
}
|
|
|
|
// TestSSEStreamUnsubscribesOnClientDisconnect checks a client that goes away
|
|
// does not leave a handler parked on the log channel: the request context is
|
|
// cancelled when the connection drops, and the handler must return and run its
|
|
// cancel hook, or every reconnect leaks a goroutine and a log subscriber.
|
|
func TestSSEStreamUnsubscribesOnClientDisconnect(t *testing.T) {
|
|
h := newHarness(t, nil)
|
|
unsub := freshUnsubSignal(h)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
_, br := openSSE(t, h, ctx)
|
|
if got := nextEvent(t, br); got.name != "ready" {
|
|
t.Fatalf("first event = %q, want ready", got.name)
|
|
}
|
|
waitFor(t, 2*time.Second, "the handler to subscribe", func() bool {
|
|
return h.fake.subscribers() == 1
|
|
})
|
|
|
|
// Disconnect: dropping the request context closes the connection, which is
|
|
// what the server sees when a browser tab is closed.
|
|
cancel()
|
|
|
|
waitFor(t, 2*time.Second, "the handler to unsubscribe", func() bool {
|
|
return h.fake.subscribers() == 0
|
|
})
|
|
select {
|
|
case <-unsub:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("the unsubscribe hook never ran after the client disconnected")
|
|
}
|
|
}
|
|
|
|
// TestSSEStreamLeavesNoGoroutinesBehind opens and closes the stream repeatedly
|
|
// and asserts the goroutine count stays bounded. The bound, not an exact count:
|
|
// the HTTP client and server keep per-connection goroutines that wind down on
|
|
// their own, and a test that demands an exact number is a flake. The property
|
|
// worth having is that the count does not grow with the number of streams.
|
|
func TestSSEStreamLeavesNoGoroutinesBehind(t *testing.T) {
|
|
h := newHarness(t, nil)
|
|
runtime.GC() // keep the baseline honest: the count is what it is after a GC
|
|
before := runtime.NumGoroutine()
|
|
for i := 0; i < 25; i++ {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
_, br := openSSE(t, h, ctx)
|
|
if got := nextEvent(t, br); got.name != "ready" {
|
|
cancel()
|
|
t.Fatalf("stream %d: first event = %q, want ready", i, got.name)
|
|
}
|
|
cancel()
|
|
}
|
|
waitFor(t, 2*time.Second, "every subscriber to be released", func() bool {
|
|
return h.fake.subscribers() == 0
|
|
})
|
|
const bound = 8 // per-connection goroutines that may still be winding down
|
|
waitFor(t, 3*time.Second, "the goroutine count to settle", func() bool {
|
|
return runtime.NumGoroutine() <= before+bound
|
|
})
|
|
}
|
|
|
|
// TestSSEStreamEndsWhenTheSourceCloses covers the other exit from the handler
|
|
// loop: the hook's channel being closed, which is what a shutting-down log ring
|
|
// does. The response has to end rather than hang, and the unsubscribe hook must
|
|
// still run — a handler that only returns on the request context is a handler
|
|
// that leaks when the producer goes away.
|
|
func TestSSEStreamEndsWhenTheSourceCloses(t *testing.T) {
|
|
h := newHarness(t, nil)
|
|
unsub := freshUnsubSignal(h)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
// Closing the hook's channel is what a shutting-down log ring does.
|
|
close(h.fake.stream)
|
|
_, br := openSSE(t, h, ctx)
|
|
if got := nextEvent(t, br); got.name != "ready" {
|
|
t.Fatalf("first event = %q, want ready", got.name)
|
|
}
|
|
if _, err := br.ReadString('\n'); err == nil {
|
|
t.Error("the stream carried on after its source closed")
|
|
}
|
|
waitFor(t, 2*time.Second, "the handler to unsubscribe", func() bool {
|
|
return h.fake.subscribers() == 0
|
|
})
|
|
select {
|
|
case <-unsub:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("the unsubscribe hook never ran after the source closed")
|
|
}
|
|
}
|
|
|
|
// TestSSEStreamRequiresAuth checks the stream is behind the same auth as the
|
|
// rest of the API: it is a live view of every client address and domain the
|
|
// instance sees, so an unauthenticated subscribe is a data leak, not a
|
|
// cosmetic one.
|
|
func TestSSEStreamRequiresAuth(t *testing.T) {
|
|
h := newHarness(t, nil)
|
|
res := h.do(http.MethodGet, "/api/logs/stream", "")
|
|
if res.status != http.StatusUnauthorized {
|
|
t.Fatalf("GET /api/logs/stream without a key = %d, want 401: %s", res.status, res.body)
|
|
}
|
|
if h.fake.count("SubscribeLog") != 0 {
|
|
t.Error("an unauthenticated request reached the subscribe hook")
|
|
}
|
|
}
|