mirror of
https://github.com/hanzoai/kafka.git
synced 2026-10-30 14:48:25 +00:00
zeekayandHanzo Dev
13404430a5
The broker named three APIs as the chatty ones and announced every other frame at Info. That is a guess about which calls a client repeats, and it was wrong about the one that matters. Measured in production: Metadata was 824 of 1200 log lines, 69% of everything the process said. Eight minutes of that rotates a pod's boot errors out of reach, which is how the last outage came to need forensics instead of a read. Every received frame is Debug now, so there is no list to keep correct as APIs are added. Info stays for what an operator did not already know. Two races were in the way of proving it, and `go test -race ./...` had never passed here — seven at HEAD, before any of this. Serve published the listener while Addr and Shutdown read it, with nothing ordering the three. The bound port is now settled before the listener is published, so a caller that can see one can see the other. The level itself was a plain global read by every logging goroutine, and the sink was chosen by a log.SetOutput on every single line — a process-global written concurrently from every connection, which also left no way to read back what was written. The level is atomic and the sink is chosen once. TRACE was missing from the severity map and scored zero by absence; saying it is cheaper than rediscovering why the order comes out right. Co-authored-by: Hanzo Dev <[email protected]>
118 lines
3.4 KiB
Go
118 lines
3.4 KiB
Go
package e2e
|
|
|
|
import (
|
|
"bytes"
|
|
"log"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/hanzoai/kafka/logging"
|
|
"github.com/twmb/franz-go/pkg/kgo"
|
|
)
|
|
|
|
// TestAReceivedFrameIsNotNews holds the broker to one line of restraint: a frame
|
|
// that arrived is a trace, not news.
|
|
//
|
|
// The broker used to name three APIs as the chatty ones and announce every other
|
|
// frame at Info. That is a guess about which calls a client repeats, and the
|
|
// guess was wrong about the one that matters. Measured in production: Metadata
|
|
// was 824 of 1200 lines — 69% of everything the process said — and eight minutes
|
|
// of it rotated the boot error explaining an outage out of the pod's history
|
|
// before anyone could read it.
|
|
//
|
|
// The test drives a real client through a real round trip and reads back what
|
|
// the broker said. Two assertions, and the second is why the first can be
|
|
// believed: at Info nothing may announce a frame, and at Debug those same frames
|
|
// must appear — otherwise a broker that logged nothing at all would pass.
|
|
func TestAReceivedFrameIsNotNews(t *testing.T) {
|
|
atInfo := said(t, logging.INFO)
|
|
atDebug := said(t, logging.DEBUG)
|
|
|
|
if n := announced(atInfo, "[INFO]"); n != 0 {
|
|
t.Fatalf("broker announced %d frame(s) at Info; a frame is a trace:\n%s",
|
|
n, firstFew(atInfo, "[INFO]"))
|
|
}
|
|
if n := announced(atDebug, "[DEBUG]"); n == 0 {
|
|
t.Fatal("no frame was traced even at Debug — the test is blind and its " +
|
|
"Info assertion proves nothing")
|
|
}
|
|
}
|
|
|
|
// said runs one produce/consume round trip at the given level and returns
|
|
// everything the broker wrote. The logger's sink is a process global written by
|
|
// every connection goroutine, so the buffer is guarded and both globals are put
|
|
// back the way they were found.
|
|
func said(t *testing.T, level string) string {
|
|
t.Helper()
|
|
|
|
buf := &guarded{}
|
|
priorLevel, priorFlags := logging.LogLevel(), log.Flags()
|
|
logging.SetLogLevel(level)
|
|
log.SetOutput(buf)
|
|
log.SetFlags(0)
|
|
t.Cleanup(func() {
|
|
log.SetOutput(discardTo(t))
|
|
log.SetFlags(priorFlags)
|
|
logging.SetLogLevel(priorLevel)
|
|
})
|
|
|
|
s := newStack(t)
|
|
topic := "quiet-" + strings.ToLower(level)
|
|
s.createTopic(t, topic)
|
|
produceN(t, s.client(t), topic, 0, 5)
|
|
consumeN(t, s.client(t,
|
|
kgo.ConsumeTopics(topic),
|
|
kgo.ConsumeResetOffset(kgo.NewOffset().AtStart())), 5)
|
|
|
|
return buf.String()
|
|
}
|
|
|
|
// discardTo restores logging to somewhere harmless for whatever runs next. The
|
|
// broker outlives the round trip by however long its goroutines take to notice
|
|
// the shutdown, and those writes must not land in the next test's buffer.
|
|
func discardTo(t *testing.T) *guarded {
|
|
t.Helper()
|
|
return &guarded{}
|
|
}
|
|
|
|
func announced(out, level string) int {
|
|
n := 0
|
|
for _, line := range strings.Split(out, "\n") {
|
|
if strings.Contains(line, level) && strings.Contains(line, "Received ") {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
|
|
func firstFew(out, level string) string {
|
|
var keep []string
|
|
for _, line := range strings.Split(out, "\n") {
|
|
if strings.Contains(line, level) && strings.Contains(line, "Received ") {
|
|
if keep = append(keep, line); len(keep) == 5 {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
return strings.Join(keep, "\n")
|
|
}
|
|
|
|
// guarded is a bytes.Buffer every connection goroutine may write at once.
|
|
type guarded struct {
|
|
mu sync.Mutex
|
|
b bytes.Buffer
|
|
}
|
|
|
|
func (g *guarded) Write(p []byte) (int, error) {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
return g.b.Write(p)
|
|
}
|
|
|
|
func (g *guarded) String() string {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
return g.b.String()
|
|
}
|