zap-proto/zip
1
0
Fork 0
mirror of https://github.com/zap-proto/zip.git synced 2026-10-19 15:38:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

telemetry: spans and the request record ship over native ZAP

Browse Source 
The request boundary already holds everything a server span and a request
record contain, so both are structs filled from data it has: the trace, span
and parent ids it minted or joined, the route template as the name, start and
duration, an outcome-derived status, and the caller's org when a gateway
parked one. They are appended under one lock, drained once per tick and at
shutdown, and shipped as JSON in a ZAP envelope — internal/o11y declares the
batch shapes itself, no SDK, no protobuf, nothing a consumer inherits.

The keys are the collector's vocabulary, not ours. It reads a path out of
url.path / http.target / http.route / path in that order, so those are the
keys this fills; http.path, which an earlier draft shipped, fills nothing.
The contract/ module proves it the only way that stays true: it starts the
collector's real receivers, serves one real request, and asserts on the
structs those receivers produced. It is its own module because the collector
already requires zip — the proof would otherwise be a cycle.

The address is the switch, spelled O11Y_SPANS_ADDR / O11Y_LOGS_ADDR /
O11Y_METRICS_ADDR — three names because the collector binds one listener per
signal and promises no adjacency between them. The metrics name replaces
O11Y_METRICS_ZAP_ENDPOINT, which deployments had already stopped matching;
the socket preference is deleted because the collector only ever binds TCP,
so that branch could not run. Nothing stated means nothing starts: no
exporter, no node, no dial, no goroutine.

An app's own logging stays its own. Only zip's request line ships to the log
ear; teeing a program's every log line off the box is one decision for
luxfi/log to make once, for everything that logs.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zooqueen
2026-08-04 21:49:59 -07:00
co-authored by Hanzo Dev
parent ed9616933e
commit e570a16f6a
12 changed files with 1488 additions and 991 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contract/contract_test.go
+276
View File
@@ -0,0 +1,276 @@
package contract_test
import (
"context"
"fmt"
"log/slog"
"net"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"github.com/hanzoai/o11y/pkg/zaplogreceiver"
"github.com/hanzoai/o11y/pkg/zapreceiver"
luxlog "github.com/luxfi/log"
"github.com/zap-proto/zip"
)
// The collector this is proven against is the version cloud runs. Moving the
// pin in go.mod is how this test asks the question again of a newer receiver.
// quiet is the receivers' logger. Their own warnings are this test's noise
// unless something fails, and when something fails the assertion says more.
func quiet() *slog.Logger { return slog.New(slog.DiscardHandler) }
// free returns a loopback address nothing is listening on yet. The receiver
// takes a port out of it and binds; the exporter dials the same host:port.
func free(t *testing.T) string {
t.Helper()
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("port: %v", err)
}
port := l.Addr().(*net.TCPAddr).Port
if err := l.Close(); err != nil {
t.Fatalf("close: %v", err)
}
return fmt.Sprintf("127.0.0.1:%d", port)
}
// held is what a receiver handed its writer, kept so a test can read it.
type held[B any] struct {
mu sync.Mutex
batch *B
seen chan struct{}
}
func (h *held[B]) take(ctx context.Context, b *B) error {
h.mu.Lock()
h.batch = b
h.mu.Unlock()
select {
case h.seen <- struct{}{}:
default:
}
return nil
}
func (h *held[B]) await(t *testing.T) *B {
t.Helper()
select {
case <-h.seen:
case <-time.After(10 * time.Second):
t.Fatal("the receiver decoded nothing")
}
h.mu.Lock()
defer h.mu.Unlock()
return h.batch
}
// serveOneRequest is the whole producing side: an app with two addresses, one
// route, one request, and a shutdown that drains what the boundary collected.
// Nothing in it mentions telemetry, which is the claim.
func serveOneRequest(t *testing.T, spans, logs string, req *http.Request) {
t.Helper()
app := zip.New(zip.Config{
AppName: "proof",
Logger: luxlog.NewNoOpLogger(),
DisableStartupMessage: true,
Telemetry: zip.Telemetry{Spans: spans, Logs: logs},
})
app.Get("/v1/orders/:id", func(c *zip.Ctx) error { return c.String(200, "ok") })
h, err := zip.Serve(app, "http://"+free(t))
if err != nil {
t.Fatalf("serve: %v", err)
}
defer func() { _ = h.Close() }()
resp, err := app.Test(req)
if err != nil {
t.Fatalf("request: %v", err)
}
resp.Body.Close()
// Shutdown drains, so the proof does not wait out an export interval.
if err := app.Shutdown(); err != nil {
t.Fatalf("shutdown: %v", err)
}
}
// TestTheCollectorDecodesWhatZipExports is the proof. Both receivers are the
// collector's own, and everything asserted below is read off the struct THEY
// produced — so a field renamed on either side arrives as a zero value here.
func TestTheCollectorDecodesWhatZipExports(t *testing.T) {
spanAddr, logAddr := free(t), free(t)
gotSpans := &held[zapreceiver.SpanBatch]{seen: make(chan struct{}, 4)}
spanRcv, err := zapreceiver.New(zapreceiver.Config{
Listen: spanAddr, OnBatch: gotSpans.take, Logger: quiet(),
})
if err != nil {
t.Fatalf("span receiver: %v", err)
}
defer spanRcv.Stop()
gotLogs := &held[zaplogreceiver.LogBatch]{seen: make(chan struct{}, 4)}
logRcv, err := zaplogreceiver.New(zaplogreceiver.Config{
Listen: logAddr, OnBatch: gotLogs.take, Logger: quiet(),
})
if err != nil {
t.Fatalf("log receiver: %v", err)
}
defer logRcv.Stop()
req := httptest.NewRequest(http.MethodGet, "/v1/orders/8a3f", nil)
req.Header.Set("X-Org-Id", "acme")
req.Header.Set("X-User-Id", "u-1")
serveOneRequest(t, spanAddr, logAddr, req)
t.Run("span", func(t *testing.T) {
batch := gotSpans.await(t)
if batch.AppName != "proof" {
t.Errorf("appName = %q, want proof — the collector lifts it into service.name", batch.AppName)
}
if len(batch.Spans) != 1 {
t.Fatalf("want one span for one request, got %d", len(batch.Spans))
}
s := batch.Spans[0]
if len(s.TraceID) != 32 || len(s.SpanID) != 16 {
t.Errorf("trace/span ids are %q/%q, want 32/16 hex", s.TraceID, s.SpanID)
}
if s.Name != "/v1/orders/:id" {
t.Errorf("name = %q, want the route template — a trace store groups by it", s.Name)
}
if s.Kind != "server" {
t.Errorf("kind = %q, want server", s.Kind)
}
if s.StatusCode != "ok" {
t.Errorf("statusCode = %q, want ok for a 200", s.StatusCode)
}
if s.EndUnixNs <= s.StartUnixNs {
t.Errorf("span does not span time: start=%d end=%d", s.StartUnixNs, s.EndUnixNs)
}
// The keys the collector reads its columns out of. url.path is first in
// the list it picks a span's path from; a key of our own invention would
// be stored and would leave that column empty.
wantAttrs(t, s.Attributes, map[string]any{
"url.path": "/v1/orders/8a3f",
"http.route": "/v1/orders/:id",
"http.request.method": "GET",
"http.response.status_code": float64(200),
"hanzo.org": "acme",
"hanzo.user": "u-1",
})
// The tenant is the operator of this program, never the caller: the
// collector reads resource["org"] as whose rows these become.
if _, ok := batch.Resource["org"]; ok {
t.Errorf("the batch claims tenant %q from a request it merely answered", batch.Resource["org"])
}
})
t.Run("record", func(t *testing.T) {
batch := gotLogs.await(t)
if batch.AppName != "proof" {
t.Errorf("appName = %q, want proof", batch.AppName)
}
if len(batch.Records) != 1 {
t.Fatalf("want one record for one request, got %d", len(batch.Records))
}
r := batch.Records[0]
if r.Body != "request" {
t.Errorf("body = %q, want request — one line per request, the same line the app logs", r.Body)
}
if r.Severity != 9 || r.SeverityText != "info" {
t.Errorf("severity = %d/%q, want 9/info for a 200", r.Severity, r.SeverityText)
}
if r.TimeUnixNs == 0 {
t.Error("the record has no time, so nothing can order it")
}
// The ids are why a log address is worth setting: they put this line
// inside the span above rather than beside it.
if len(r.TraceID) != 32 || len(r.SpanID) != 16 {
t.Errorf("record ids are %q/%q, want 32/16 hex", r.TraceID, r.SpanID)
}
wantAttrs(t, r.Attributes, map[string]any{
"url.path": "/v1/orders/8a3f",
"http.route": "/v1/orders/:id",
"http.request.method": "GET",
"http.response.status_code": float64(200),
"hanzo.org": "acme",
"hanzo.user": "u-1",
})
if _, ok := r.Attributes["duration_ms"]; !ok {
t.Error("the record has no duration, which is the field the log line exists to carry")
}
})
}
// TestAFailedRequestArrivesAsAFailure pins the one thing an operator filters on.
// A span that says ok for a 500, or a record at info, is worse than no telemetry:
// it answers the question wrongly.
func TestAFailedRequestArrivesAsAFailure(t *testing.T) {
spanAddr, logAddr := free(t), free(t)
gotSpans := &held[zapreceiver.SpanBatch]{seen: make(chan struct{}, 4)}
spanRcv, err := zapreceiver.New(zapreceiver.Config{
Listen: spanAddr, OnBatch: gotSpans.take, Logger: quiet(),
})
if err != nil {
t.Fatalf("span receiver: %v", err)
}
defer spanRcv.Stop()
gotLogs := &held[zaplogreceiver.LogBatch]{seen: make(chan struct{}, 4)}
logRcv, err := zaplogreceiver.New(zaplogreceiver.Config{
Listen: logAddr, OnBatch: gotLogs.take, Logger: quiet(),
})
if err != nil {
t.Fatalf("log receiver: %v", err)
}
defer logRcv.Stop()
app := zip.New(zip.Config{
AppName: "proof",
Logger: luxlog.NewNoOpLogger(),
DisableStartupMessage: true,
Telemetry: zip.Telemetry{Spans: spanAddr, Logs: logAddr},
})
app.Get("/x", func(c *zip.Ctx) error { return c.String(500, "boom") })
h, err := zip.Serve(app, "http://"+free(t))
if err != nil {
t.Fatalf("serve: %v", err)
}
defer func() { _ = h.Close() }()
resp, err := app.Test(httptest.NewRequest(http.MethodGet, "/x", nil))
if err != nil {
t.Fatalf("request: %v", err)
}
resp.Body.Close()
if err := app.Shutdown(); err != nil {
t.Fatalf("shutdown: %v", err)
}
if s := gotSpans.await(t).Spans[0]; s.StatusCode != "error" {
t.Errorf("statusCode = %q, want error for a 500", s.StatusCode)
}
r := gotLogs.await(t).Records[0]
if r.Severity != 17 || r.SeverityText != "error" {
t.Errorf("severity = %d/%q, want 17/error for a 500", r.Severity, r.SeverityText)
}
}
// wantAttrs checks every key the collector reads a column out of, and reports
// the ones that arrived under some other name.
func wantAttrs(t *testing.T, got map[string]any, want map[string]any) {
t.Helper()
for k, v := range want {
if got[k] != v {
t.Errorf("attribute %q = %v (%T), want %v", k, got[k], got[k], v)
}
}
}
contract/doc.go
+30
View File
@@ -0,0 +1,30 @@
// Package contract holds one proof: what a zip app exports is what the
// collector decodes.
//
// # Why it is a module of its own
//
// The proof needs hanzoai/o11y, and zip must not require it. Two reasons, both
// measured rather than assumed. It is circular — o11y already requires zip, so
// zip requiring o11y leaves neither end able to release without the other. And
// it is expensive for everyone else: adding o11y to zip's own module takes its
// go.sum from 147 lines to about 720 and its module graph past a thousand
// modules, a cost paid by every program built on zip in order to run a test that
// belongs to zip alone.
//
// A module here pays neither. Nothing imports this package, `go build ./...` and
// `go test ./...` in zip do not descend into it, and no consumer of zip inherits
// a line of it. CI runs it as its own step, which is where the cost belongs.
//
// # What it proves
//
// The two ends re-declare one contract: zip's internal/o11y package and the
// collector's pkg/zapreceiver and pkg/zaplogreceiver each state the batch shapes
// in their own types, so that a collector can be deployed without rebuilding
// every service that reports to it. Two declarations need something holding them
// together, and transcribing field names into a test only proves that a
// transcription matches itself.
//
// So this test runs the real receivers, has a real zip app answer a real
// request, and reads what the receivers decoded. A field renamed on either side
// fails it, and a field zip invents fails it too — it arrives as nothing.
package contract
contract/go.mod
+63
View File
@@ -0,0 +1,63 @@
// A module of its own so that requiring the collector costs zip's consumers
// nothing, and so that two modules which require each other can still be built.
// See doc.go.
module github.com/zap-proto/zip/contract
go 1.26.5
require (
github.com/hanzoai/o11y v1.5.58
github.com/luxfi/log v1.6.0
github.com/zap-proto/zip v1.24.3
)
require (
github.com/andybalholm/brotli v1.2.1 // indirect
github.com/cenkalti/backoff v2.2.1+incompatible // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/cloudflare/circl v1.6.3 // indirect
github.com/go-logr/logr v1.4.3 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/gofiber/schema v1.7.1 // indirect
github.com/gofiber/utils/v2 v2.0.4 // indirect
github.com/google/uuid v1.6.1-0.20241114170450-2d3c2a9cc518 // indirect
github.com/gorilla/rpc v1.2.1 // indirect
github.com/grandcat/zeroconf v1.0.0 // indirect
github.com/klauspost/compress v1.18.6 // indirect
github.com/luxfi/accel v1.2.4 // indirect
github.com/luxfi/crypto v1.20.2 // indirect
github.com/luxfi/mdns v0.1.1 // indirect
github.com/luxfi/metric v1.9.1 // indirect
github.com/luxfi/zap v1.2.7 // indirect
github.com/mattn/go-colorable v0.1.15 // indirect
github.com/mattn/go-isatty v0.0.22 // indirect
github.com/miekg/dns v1.1.72 // indirect
github.com/philhofer/fwd v1.2.0 // indirect
github.com/tinylib/msgp v1.6.4 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
github.com/valyala/fasthttp v1.72.0 // indirect
github.com/zap-proto/fiber/v3 v3.2.1 // indirect
github.com/zap-proto/go v1.3.0 // indirect
github.com/zap-proto/http v0.3.1 // indirect
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
go.opentelemetry.io/otel v1.44.0 // indirect
go.opentelemetry.io/otel/log v0.20.0 // indirect
go.opentelemetry.io/otel/metric v1.44.0 // indirect
go.opentelemetry.io/otel/sdk v1.44.0 // indirect
go.opentelemetry.io/otel/sdk/log v0.20.0 // indirect
go.opentelemetry.io/otel/sdk/metric v1.44.0 // indirect
go.opentelemetry.io/otel/trace v1.44.0 // indirect
golang.org/x/crypto v0.54.0 // indirect
golang.org/x/mod v0.37.0 // indirect
golang.org/x/net v0.57.0 // indirect
golang.org/x/sync v0.22.0 // indirect
golang.org/x/sys v0.47.0 // indirect
golang.org/x/text v0.40.0 // indirect
golang.org/x/tools v0.47.0 // indirect
google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect
gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
)
// The zip under test is the working tree, never a published tag: a proof of what
// this repository exports has to read what this repository does.
replace github.com/zap-proto/zip => ../
contract/go.sum
+134
View File
@@ -0,0 +1,134 @@
github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8=
github.com/cloudflare/circl v1.6.3/go.mod h1:2eXP6Qfat4O/Yhh8BznvKnJ+uzEoTQ6jVKJRn81BiS4=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ=
github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/gofiber/schema v1.7.1 h1:oSJBKdgP8JeIME4TQSAqlNKTU2iBB+2RNmKi8Nsc+TI=
github.com/gofiber/schema v1.7.1/go.mod h1:A/X5Ffyru4p9eBdp99qu+nzviHzQiZ7odLT+TwxWhbk=
github.com/gofiber/utils/v2 v2.0.4 h1:WwAxUA7L4MW2DjdEHF234lfqvBqd2vYYuBtA9TJq2ec=
github.com/gofiber/utils/v2 v2.0.4/go.mod h1:GGERKU3Vhj5z6hS8YKvxL99A54DjOvTFZ0cjZnG4Lj4=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/uuid v1.6.1-0.20241114170450-2d3c2a9cc518 h1:UBg1xk+oAsIVbFuGg6hdfAm7EvCv3EL80vFxJNsslqw=
github.com/google/uuid v1.6.1-0.20241114170450-2d3c2a9cc518/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/rpc v1.2.1 h1:yC+LMV5esttgpVvNORL/xX4jvTTEUE30UZhZ5JF7K9k=
github.com/gorilla/rpc v1.2.1/go.mod h1:uNpOihAlF5xRFLuTYhfR0yfCTm0WTQSQttkMSptRfGk=
github.com/grandcat/zeroconf v1.0.0 h1:uHhahLBKqwWBV6WZUDAT71044vwOTL+McW0mBJvo6kE=
github.com/grandcat/zeroconf v1.0.0/go.mod h1:lTKmG1zh86XyCoUeIHSA4FJMBwCJiQmGfcP2PdzytEs=
github.com/hanzoai/o11y v1.5.58 h1:xp+lMkBwYRdd3TJeUB83Qq8XXnyLBX+nmOQ2hPEwhJU=
github.com/hanzoai/o11y v1.5.58/go.mod h1:TrDmdeIBBwijyr5dti7+to4gkHOWxlvB0Lu/oRqS9uw=
github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/luxfi/accel v1.2.4 h1:5VbIHyEvvfobn2zBiTFODxDw1CeqxCepZOLlvkuf9yQ=
github.com/luxfi/accel v1.2.4/go.mod h1:ISIwAX+ZfsL/S5nsP2JvfldXN6Nc+QzoWf6Jtaq+xsQ=
github.com/luxfi/crypto v1.20.2 h1:L81WEsU/hs2A76F5PWBusG0yU74QqkDdUqqgexWUxh4=
github.com/luxfi/crypto v1.20.2/go.mod h1:qYHOM0lO4PRh7LEaObxFQUIMjmT1/paVm/WgZkobT1k=
github.com/luxfi/log v1.6.0 h1:BbV7kOAK9nrUcrpr4/yCw0ODaBtQOLR5Cjj/NLWwZqo=
github.com/luxfi/log v1.6.0/go.mod h1:NWkRRwwXPPphCUwC0+UemY0eoedQEFmhSIZXTxdA9VQ=
github.com/luxfi/mdns v0.1.1 h1:g2eRr9AXcziPkkcd24M+Qu9ApEpoKKjfI79QSNqv0rQ=
github.com/luxfi/mdns v0.1.1/go.mod h1:dbp5f3h3aE7CGzwbaWzBM9cwdcekhmSrWhQevgYhhNA=
github.com/luxfi/metric v1.9.1 h1:dyIHgdZ9mvvlppbbvccRoDyOYjXmLWEZYA45rDRm398=
github.com/luxfi/metric v1.9.1/go.mod h1:JhVJPOM10FEpGSSaP8QrGLi7w0D3pjJ2kwWwmGi+9YI=
github.com/luxfi/pq v1.1.0 h1:ADplfUSyirLymSxs3Ix0HeDTyl5oswCNUpXJt/5vLY8=
github.com/luxfi/pq v1.1.0/go.mod h1:KT5rG9ztpzIkT9QSnXK4WFqBBLzKCLjY7l1c/unBi8I=
github.com/luxfi/zap v1.2.7 h1:LzYit+1xLoQRT0ahYsoYcdoDz5I5gc0ltLvBXRkPpwQ=
github.com/luxfi/zap v1.2.7/go.mod h1:cAoQ1g80iSxgWOhXj6YBrWfKJi771FjPe8HdQ2eOLZw=
github.com/mattn/go-colorable v0.1.15 h1:+u9SLTRGnXv73cEsnsmoZBom+dMU88B2M0aDcWy0/jY=
github.com/mattn/go-colorable v0.1.15/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4=
github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4=
github.com/miekg/dns v1.1.27/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
github.com/miekg/dns v1.1.72 h1:vhmr+TF2A3tuoGNkLDFK9zi36F2LS+hKTRW0Uf8kbzI=
github.com/miekg/dns v1.1.72/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs=
github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM=
github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/quic-go/quic-go v0.59.1 h1:0Gmua0HW1Tv7ANR7hUYwRyD0MG5OJfgvYSZasGZzBic=
github.com/quic-go/quic-go v0.59.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
github.com/shamaton/msgpack/v3 v3.1.0 h1:jsk0vEAqVvvS9+fTZ5/EcQ9tz860c9pWxJ4Iwecz8gU=
github.com/shamaton/msgpack/v3 v3.1.0/go.mod h1:DcQG8jrdrQCIxr3HlMYkiXdMhK+KfN2CitkyzsQV4uc=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/tinylib/msgp v1.6.4 h1:mOwYbyYDLPj35mkA2BjjYejgJk9BuHxDdvRnb6v2ZcQ=
github.com/tinylib/msgp v1.6.4/go.mod h1:RSp0LW9oSxFut3KzESt5Voq4GVWyS+PSulT77roAqEA=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
github.com/valyala/fasthttp v1.72.0 h1:R7kYdoWhn1ye1fVpP+cDHDJwYm3NkwLliwgzJ/Abg7M=
github.com/valyala/fasthttp v1.72.0/go.mod h1:zsbLTYqcpIktdQytlVBwIjY9La5d6bs990nBxWg8efk=
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
github.com/zap-proto/fiber/v3 v3.2.1 h1:k45oKyTwySPtGt8sPz2Ao8OUHc7pDEhai8Np2Ym6Jbg=
github.com/zap-proto/fiber/v3 v3.2.1/go.mod h1:eDm2z+ufJrkuE4MeX0Mea4oc/p7/HpXiZjVB+BXCKOA=
github.com/zap-proto/go v1.3.0 h1:S3rMoawwhH/BbSZ4G8zG05hJoQnMSMDPzIq75diCTqE=
github.com/zap-proto/go v1.3.0/go.mod h1:914SNGTH6Rv3Yu1MweWJBPEN8FZlo5C39QyhaB0C7Q0=
github.com/zap-proto/http v0.3.1 h1:A2rCPWYCX866eAsdiWuns0dvWnBmViZtGm4pwX7jwlY=
github.com/zap-proto/http v0.3.1/go.mod h1:UYfGhDDCetgxs65XSev8Lpf65COg5vKQK+cWwZGh4zQ=
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
go.opentelemetry.io/otel/log v0.20.0 h1:/5i0vuHxCLWUfChWG41K9wkM0jafruPw9NU1/RCJirs=
go.opentelemetry.io/otel/log v0.20.0/go.mod h1:wOcMcjsZpG8x7Bak7IhSi/lg8wscV2C1VdrKCLPlt0E=
go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
go.opentelemetry.io/otel/metric/x v0.66.0 h1:YkCrx1zLOChi9ZcZ6euupOcsgzbVlec7D/xoEU1+cTA=
go.opentelemetry.io/otel/metric/x v0.66.0/go.mod h1:d1+BDj9t96do0/1LoU1ayfCv79ZgNE41qbhBvnMOBZk=
go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58=
go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0=
go.opentelemetry.io/otel/sdk/log v0.20.0 h1:vM3xI7TQgKPiSghe6urZtAkyFY7SodrSpC83CffDFuY=
go.opentelemetry.io/otel/sdk/log v0.20.0/go.mod h1:Knej2nmsTUzN79T2eeXdRsjjPcoxoq2pUyUHz9TFyyU=
go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI=
go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA=
go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ=
golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
golang.org/x/tools v0.0.0-20191216052735-49a3e744a425/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q=
golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af h1:+5/Sw3GsDNlEmu7TfklWKPdQ0Ykja5VEmq2i817+jbI=
google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
hanzo.yml
+7
View File
@@ -25,3 +25,10 @@ test:
# every auth gate in every app built on zip), the no-Next abort, empty-leaf
# group roots, static fallback, and the runtime/middleware packages.
run: go test -count=1 -timeout 10m ./...
- name: go-test-collector
# contract/ is a module of its own, so ./... above does not reach it and no
# consumer of zip inherits what it requires. It runs the collector's REAL
# receivers against a real zip app: the one thing proving that two
# deliberately separate declarations of the batch shapes still describe one
# thing. See contract/doc.go.
run: cd contract && go test -count=1 -timeout 5m ./...
internal/o11y/o11y.go
+319
View File
@@ -0,0 +1,319 @@
// Package o11y is the shape of what a zip app sends to the collector, and the
// one place that knows it.
//
// A batch is JSON, the JSON rides in a ZAP envelope tagged with the signal's
// message type, and the envelope goes down a ZAP connection. That is the whole
// transport: no OpenTelemetry SDK, no protobuf, no gRPC, no OTLP. zip is the
// base every Hanzo program is built on, so a dependency taken here is a
// dependency taken in all of them, and an SDK is a large one to take in order
// to send two JSON documents.
//
// # The shapes are copied deliberately, not imported
//
// SpanBatch and LogBatch are field-for-field what hanzoai/o11y decodes in
// pkg/zapreceiver and pkg/zaplogreceiver. They are re-declared here rather than
// imported, and that is the same choice the receiving side already made — its
// own doc says it re-declares shapes rather than importing the producer
// libraries. Importing across the boundary would also be circular: o11y already
// requires zip, so zip requiring o11y would leave neither end able to move
// without the other. Two declarations of one contract is the cost of letting
// the two ends ship independently, and contract/ — a module of its own, so no
// consumer of zip inherits the collector's dependencies — is what holds them
// together, by feeding this encoder's output to the real decoder.
//
// # The address is the switch, on both ends
//
// A signal with no address is a signal this program does not report. That is
// not a convenience: it is the same rule the collector states for receiving —
// "THE ADDRESS IS THE SWITCH ... an empty address means that signal is not
// received in this process" — so one sentence describes both halves, and there
// is no enable flag anywhere that can disagree with an address.
//
// An address is host:port. The collector reads a port out of its listen address
// and binds TCP, so host:port is the only form it can serve and the only form
// worth sending to.
package o11y
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"sync"
"github.com/luxfi/zap"
)
// The signal message types, in the ZAP envelope's flags. They are a fleet-wide
// registry rather than this package's choice: 1 is spans, 2 is metrics, 3 is
// logs, and every deployed collector switches on them. Append only —
// renumbering breaks every receiver at once. Metrics is stated but never sent
// from here: luxfi/metric owns that encoder, and the number is written down so
// nothing invents a second meaning for it.
const (
MsgSpan uint16 = 1
MsgMetric uint16 = 2
MsgLog uint16 = 3
)
// SpanBatch is one shipment of spans. Mirrors o11y zapreceiver.SpanBatch.
//
// Resource holds what is true of the SENDER for every span in the batch. It
// deliberately carries no caller identity: the collector reads resource["org"]
// as the TENANT that owns the telemetry — whose rows these become and who may
// read them — which is the operator of this program, not whoever it answered.
// The request's org is an attribute, where a per-request value belongs.
type SpanBatch struct {
AppName string `json:"appName,omitempty"`
Version string `json:"version,omitempty"`
Resource map[string]string `json:"resource,omitempty"`
Spans []Span `json:"spans"`
}
// Span is one operation. Mirrors o11y zapreceiver.Span.
//
// Kind and StatusCode are read through the collector's normEnum, which
// lowercases and strips a leading "span_kind_" / "status_code_". So "server"
// and "SPAN_KIND_SERVER" are the same value, and the plain lowercase spelling
// is the one used here.
type Span struct {
TraceID string `json:"traceId"`
SpanID string `json:"spanId"`
ParentSpanID string `json:"parentSpanId,omitempty"`
Name string `json:"name"`
Kind string `json:"kind,omitempty"`
StartUnixNs int64 `json:"startUnixNs"`
EndUnixNs int64 `json:"endUnixNs"`
Attributes map[string]any `json:"attributes,omitempty"`
Events []SpanEvent `json:"events,omitempty"`
StatusCode string `json:"statusCode,omitempty"`
StatusMsg string `json:"statusMessage,omitempty"`
}
// SpanEvent is a point in time inside a span. Mirrors o11y zapreceiver.SpanEvent.
type SpanEvent struct {
Name string `json:"name"`
TimeUnixNs int64 `json:"timeUnixNs"`
Attributes map[string]any `json:"attributes,omitempty"`
}
// LogBatch is one shipment of records. Mirrors o11y zaplogreceiver.LogBatch.
type LogBatch struct {
AppName string `json:"appName,omitempty"`
Version string `json:"version,omitempty"`
Resource map[string]string `json:"resource,omitempty"`
Records []LogRecord `json:"records"`
}
// LogRecord is one line. Mirrors o11y zaplogreceiver.LogRecord.
//
// Severity is the severity NUMBER, which the collector clamps to a uint8 and
// stores beside the text: 1 trace, 5 debug, 9 info, 13 warn, 17 error, 21
// fatal. TraceID and SpanID are what join a line to the span it happened under,
// and they are the reason a log address is worth setting at all.
type LogRecord struct {
TimeUnixNs int64 `json:"timeUnixNs"`
ObservedTimeUnixNs int64 `json:"observedTimeUnixNs,omitempty"`
Severity int `json:"severity,omitempty"`
SeverityText string `json:"severityText,omitempty"`
Body string `json:"body,omitempty"`
Attributes map[string]any `json:"attributes,omitempty"`
TraceID string `json:"traceId,omitempty"`
SpanID string `json:"spanId,omitempty"`
EventName string `json:"eventName,omitempty"`
}
// The severity numbers, named so a mapping reads as a mapping.
const (
SeverityDebug = 5
SeverityInfo = 9
SeverityWarn = 13
SeverityError = 17
)
// Signal binds a signal's name to the message type it travels under, so a call
// site cannot pair the two wrongly. Metrics is absent because luxfi/metric
// ships that signal itself.
type Signal struct {
Name string // "spans" or "logs" — as the collector's own config names them
Msg uint16 // the ZAP message type
}
var (
Spans = Signal{Name: "spans", Msg: MsgSpan}
Logs = Signal{Name: "logs", Msg: MsgLog}
)
// Export ships one signal's batches to the collector at one address.
//
// Nothing is dialled until there is something to send, and a failure to send
// costs the caller nothing.
//
// Fire and forget, always. Telemetry must never be able to stop the program it
// describes, so a batch that cannot be delivered is DROPPED rather than held —
// holding it is how a producer grows without bound while nobody is listening.
type Export struct {
sig Signal
addr string
appName string
say func(string, ...any)
mu sync.Mutex
node *zap.Node
peer string
up bool // last known state, so reporting is edge-triggered
said bool // whether a state has been reported at all yet
closed bool
}
// New prepares an exporter for an address the caller has already decided to
// report to. It binds nothing, dials nothing and cannot fail: a collector that
// is absent, or that starts later, is the ordinary case.
//
// say is how it reports, and it is the APP's own logger rather than one of this
// package's choosing — an exporter's two lines are the program telling you
// something about itself, and they belong in the same stream, at the same
// level, with the same fields as everything else the program says.
func New(sig Signal, addr, appName string, say func(string, ...any)) *Export {
if say == nil {
say = func(string, ...any) {}
}
return &Export{sig: sig, addr: addr, appName: appName, say: say}
}
// connect dials the collector, and redials after a connection is lost.
//
// The address is passed to luxfi/zap untouched, which reads its form to choose
// a network — the same discipline zip applies when serving, and the reason
// nothing here inspects the string.
func (e *Export) connect() error {
if e.node == nil {
n := zap.NewNode(zap.NodeConfig{
NodeID: "o11y-" + e.appName + "-" + e.sig.Name,
ServiceType: "_o11y._tcp",
// The node's own chatter — started, peer connected, peer gone — is
// this package's internals, not the program's news. What an operator
// needs is the two edges in report, in the program's own log.
Logger: slog.New(slog.DiscardHandler),
// Discovery would make an exporter announce itself and hunt for
// peers. It has one peer, at the one address it was given.
NoDiscovery: true,
})
if err := n.Start(); err != nil {
return fmt.Errorf("start node: %w", err)
}
e.node = n
}
if err := e.node.ConnectDirect(e.addr); err != nil {
return fmt.Errorf("%s: %w", e.addr, err)
}
peers := e.node.Peers()
if len(peers) == 0 {
return fmt.Errorf("%s: connected but no peer", e.addr)
}
e.peer = peers[0]
return nil
}
// report says when export STARTS working and when it STOPS, and nothing in
// between.
//
// Edge-triggered because the alternative is a line per failed flush forever on
// every box whose collector is down — which trains everyone to ignore the
// stream that is supposed to carry the one line that matters. The transition is
// the event; the steady state is not.
func (e *Export) report(up bool, why error) {
if e.said && up == e.up {
return
}
e.said, e.up = true, up
if up {
e.say("telemetry export working", "signal", e.sig.Name, "addr", e.addr)
return
}
e.say("telemetry export stopped", "signal", e.sig.Name, "addr", e.addr, "reason", why.Error())
}
// Spans ships a batch of spans.
func (e *Export) Spans(ctx context.Context, b SpanBatch) {
if len(b.Spans) == 0 {
return
}
b.AppName = e.appName
e.send(ctx, MsgSpan, &b)
}
// Logs ships a batch of records.
func (e *Export) Logs(ctx context.Context, b LogBatch) {
if len(b.Records) == 0 {
return
}
b.AppName = e.appName
e.send(ctx, MsgLog, &b)
}
// send is the one path a batch takes: dial, marshal, frame, write.
//
// Every failure ends here. A caller cannot act on a telemetry error — there is
// nothing sensible for a request handler to do about a collector being down —
// so none is returned, and the only trace of it is the edge-triggered line.
func (e *Export) send(ctx context.Context, msg uint16, batch any) {
e.mu.Lock()
defer e.mu.Unlock()
if e.closed {
return
}
// Dial lazily, and redial whenever there is no connection: the collector may
// have appeared since the last attempt, which is the normal case when start
// order puts this program first.
if e.peer == "" {
if err := e.connect(); err != nil {
e.report(false, err)
return
}
}
payload, err := json.Marshal(batch)
if err != nil {
e.report(false, err)
return
}
m, err := zap.Parse(frame(msg, payload))
if err != nil {
e.report(false, err)
return
}
if err := e.node.Send(ctx, e.peer, m); err != nil {
// The connection is gone. Forget the peer so the next batch redials,
// rather than writing into a socket that will never drain.
e.peer = ""
e.report(false, err)
return
}
e.report(true, nil)
}
// frame wraps a JSON payload in the ZAP envelope every collector receiver
// expects: a root object whose field 0 is the payload bytes, with the signal's
// message type in the upper 8 bits of the envelope flags.
func frame(msg uint16, payload []byte) []byte {
const envelope = 16
b := zap.NewBuilder(envelope + 64 + len(payload))
root := b.StartObject(envelope)
root.SetBytes(0, payload)
root.FinishAsRoot()
return b.FinishWithFlags(msg << 8)
}
// Close stops the node. Safe to call more than once, because shutdown paths run
// from more than one place.
func (e *Export) Close() {
e.mu.Lock()
defer e.mu.Unlock()
if e.closed {
return
}
e.closed = true
if e.node != nil {
e.node.Stop()
}
}
internal/o11y/o11y_test.go
+281
View File
@@ -0,0 +1,281 @@
package o11y_test
import (
"context"
"encoding/json"
"fmt"
"net"
"sync"
"testing"
"time"
"github.com/luxfi/zap"
"github.com/zap-proto/zip/internal/o11y"
)
// What this package sends is proven against the collector's OWN decoder, in
// contract/ — a module of its own, so no consumer of zip inherits the
// collector's dependencies to run it. That proof is the one that can fail when a
// field is renamed on either side, so there is no transcribed copy of the
// receiver's field names here to drift away from it.
//
// What is left for this package's own tests is everything the contract module
// cannot see: that an unreachable collector costs the caller nothing, that an
// idle service is silent, and that the report about export itself is
// edge-triggered.
// filled builds values with every field set.
func filledSpan() o11y.SpanBatch {
return o11y.SpanBatch{
AppName: "svc",
Version: "v1",
Resource: map[string]string{"service.name": "svc"},
Spans: []o11y.Span{{
TraceID: "0102030405060708090a0b0c0d0e0f10",
SpanID: "1112131415161718",
ParentSpanID: "2122232425262728",
Name: "/v1/x",
Kind: "server",
StartUnixNs: 1700000000000000000,
EndUnixNs: 1700000000000001000,
Attributes: map[string]any{"http.request.method": "GET"},
Events: []o11y.SpanEvent{{Name: "e", TimeUnixNs: 1, Attributes: map[string]any{"k": "v"}}},
StatusCode: "ok",
StatusMsg: "fine",
}},
}
}
func filledLog() o11y.LogBatch {
return o11y.LogBatch{
AppName: "svc",
Version: "v1",
Resource: map[string]string{"service.name": "svc"},
Records: []o11y.LogRecord{{
TimeUnixNs: 1700000000000000000,
ObservedTimeUnixNs: 1700000000000000001,
Severity: o11y.SeverityInfo,
SeverityText: "info",
Body: "request",
Attributes: map[string]any{"http.request.method": "GET"},
TraceID: "0102030405060708090a0b0c0d0e0f10",
SpanID: "1112131415161718",
EventName: "request",
}},
}
}
// TestSignalNumbers pins the fleet-wide registry. These are not this package's
// to choose: every deployed collector switches on them, so a change here is a
// change to every receiver at once.
func TestSignalNumbers(t *testing.T) {
for _, c := range []struct {
what string
got uint16
want uint16
}{
{"span", o11y.MsgSpan, 1},
{"metric", o11y.MsgMetric, 2},
{"log", o11y.MsgLog, 3},
} {
if c.got != c.want {
t.Errorf("%s message type = %d, want %d", c.what, c.got, c.want)
}
}
}
// ear is a ZAP listener that decodes one signal. It reads the envelope the way
// the collector's receivers do — payload out of field 0, message type out of the
// flags' upper byte.
type ear struct {
node *zap.Node
mu sync.Mutex
got [][]byte
seen chan struct{}
}
func listen(t *testing.T, addr string, msg uint16) *ear {
t.Helper()
e := &ear{seen: make(chan struct{}, 8)}
e.node = zap.NewNode(zap.NodeConfig{
NodeID: "ear-" + addr,
ServiceType: "_o11y._tcp",
Address: addr,
NoDiscovery: true,
})
e.node.Handle(msg, func(_ context.Context, _ string, m *zap.Message) (*zap.Message, error) {
cp := append([]byte(nil), m.Root().Bytes(0)...)
e.mu.Lock()
e.got = append(e.got, cp)
e.mu.Unlock()
select {
case e.seen <- struct{}{}:
default:
}
// No reply: the sender does not wait, so answering would be a frame
// nothing reads.
return nil, nil
})
if err := e.node.Start(); err != nil {
t.Fatalf("ear on %s: %v", addr, err)
}
t.Cleanup(e.node.Stop)
return e
}
func (e *ear) await(t *testing.T) []byte {
t.Helper()
select {
case <-e.seen:
case <-time.After(5 * time.Second):
t.Fatal("nothing arrived at the collector")
}
e.mu.Lock()
defer e.mu.Unlock()
return e.got[len(e.got)-1]
}
// free returns a loopback address nothing is listening on yet. host:port is the
// only form the collector serves, so it is the only form tested.
func free(t *testing.T) string {
t.Helper()
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("port: %v", err)
}
port := l.Addr().(*net.TCPAddr).Port
if err := l.Close(); err != nil {
t.Fatalf("close: %v", err)
}
return fmt.Sprintf("127.0.0.1:%d", port)
}
func TestSpansReachTheCollector(t *testing.T) {
addr := free(t)
e := listen(t, addr, o11y.MsgSpan)
x := o11y.New(o11y.Spans, addr, "svc", nil)
defer x.Close()
x.Spans(context.Background(), filledSpan())
var got o11y.SpanBatch
if err := json.Unmarshal(e.await(t), &got); err != nil {
t.Fatalf("collector could not decode: %v", err)
}
if len(got.Spans) != 1 || got.Spans[0].TraceID != filledSpan().Spans[0].TraceID {
t.Fatalf("span did not survive the trip: %+v", got.Spans)
}
if got.AppName != "svc" {
t.Errorf("appName = %q, want svc — the exporter stamps it", got.AppName)
}
}
func TestLogsReachTheCollector(t *testing.T) {
addr := free(t)
e := listen(t, addr, o11y.MsgLog)
x := o11y.New(o11y.Logs, addr, "svc", nil)
defer x.Close()
x.Logs(context.Background(), filledLog())
var got o11y.LogBatch
if err := json.Unmarshal(e.await(t), &got); err != nil {
t.Fatalf("collector could not decode: %v", err)
}
if len(got.Records) != 1 || got.Records[0].Body != "request" {
t.Fatalf("record did not survive the trip: %+v", got.Records)
}
if got.Records[0].TraceID == "" || got.Records[0].SpanID == "" {
t.Error("the record lost its trace ids, which are the reason logs ship at all")
}
}
// TestUnreachableCollectorIsSurvivable is the fire-and-forget claim. An address
// nothing listens on must cost the caller nothing — no error to handle, no
// panic, no block — because telemetry may never be able to stop the program it
// describes.
func TestUnreachableCollectorIsSurvivable(t *testing.T) {
x := o11y.New(o11y.Spans, free(t), "svc", nil)
defer x.Close()
done := make(chan struct{})
go func() {
x.Spans(context.Background(), filledSpan())
x.Logs(context.Background(), filledLog())
close(done)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("exporting to a dead collector blocked the caller")
}
}
// TestEmptyBatchSendsNothing keeps idle services quiet: a tick with no traffic
// must not open a connection or write a frame.
func TestEmptyBatchSendsNothing(t *testing.T) {
addr := free(t)
e := listen(t, addr, o11y.MsgSpan)
x := o11y.New(o11y.Spans, addr, "svc", nil)
defer x.Close()
x.Spans(context.Background(), o11y.SpanBatch{})
select {
case <-e.seen:
t.Fatal("an empty batch was sent")
case <-time.After(300 * time.Millisecond):
}
}
// TestReportIsEdgeTriggered is the "must not spam" claim. A program whose
// collector is down answers requests forever; it must say so once, not once per
// flush.
func TestReportIsEdgeTriggered(t *testing.T) {
var mu sync.Mutex
var lines []string
say := func(msg string, _ ...any) {
mu.Lock()
lines = append(lines, msg)
mu.Unlock()
}
x := o11y.New(o11y.Spans, free(t), "svc", say)
defer x.Close()
for range 10 {
x.Spans(context.Background(), filledSpan())
}
mu.Lock()
defer mu.Unlock()
if len(lines) != 1 {
t.Fatalf("ten failed exports produced %d lines, want exactly 1: %v", len(lines), lines)
}
if lines[0] != "telemetry export stopped" {
t.Errorf("line = %q, want the stopped edge", lines[0])
}
}
// TestReportSaysWhenExportStartsWorking is the other edge, and the one that
// matters for start order: a collector that appears later is picked up, and the
// program says so once.
func TestReportSaysWhenExportStartsWorking(t *testing.T) {
var mu sync.Mutex
var lines []string
say := func(msg string, _ ...any) {
mu.Lock()
lines = append(lines, msg)
mu.Unlock()
}
addr := free(t)
x := o11y.New(o11y.Spans, addr, "svc", say)
defer x.Close()
// Nothing is listening yet.
x.Spans(context.Background(), filledSpan())
// The collector starts AFTER the program did.
listen(t, addr, o11y.MsgSpan)
x.Spans(context.Background(), filledSpan())
mu.Lock()
defer mu.Unlock()
if len(lines) != 2 || lines[0] != "telemetry export stopped" || lines[1] != "telemetry export working" {
t.Fatalf("want stopped then working, got %v", lines)
}
}
internal/otlz/otlz.go
-366
View File
@@ -1,366 +0,0 @@
// Package otlz is the shape of what a zip app sends to o11y, and the one place
// that knows it.
//
// It is telemetry over ZAP and nothing else: a batch is JSON, the JSON rides in
// a ZAP envelope tagged with the signal's message type, and the envelope goes
// down a ZAP connection. There is no OpenTelemetry SDK anywhere in it, no
// protobuf, no gRPC and no OTLP — which is the point. zip is the base every
// Hanzo program is built on, so a dependency here is a dependency in all of
// them, and the OTel SDK is a large one to take for three JSON documents.
//
// # The shapes are copied deliberately, not imported
//
// SpanBatch and LogBatch below are field-for-field what o11y's zapreceiver and
// zaplogreceiver decode. They are re-declared rather than imported, and that is
// the same choice the receiving side already made — zapingest's own doc says it
// re-declares shapes rather than importing the producer libraries. Importing
// across that boundary would make the emitter and the collector one build, so a
// collector could not be deployed without rebuilding every service that talks to
// it. Two declarations of one contract is the cost of letting the two ends ship
// independently, and the test beside this file is what keeps them honest.
//
// # The address chooses the transport, and nothing here branches on it
//
// An address is dialled by its FORM: a filesystem path is a unix socket,
// anything else is host:port TCP. That rule lives in luxfi/zap's Network, so
// this package passes the address through and never inspects it — the same
// discipline zip's own transportFor applies to serving. One consequence is worth
// stating: over a unix socket the kernel answers for the peer, so o11y learns
// WHICH process exported a batch as an attestation rather than a claim, exactly
// as zip documents for a Call. TCP loopback cannot say that — any process on the
// box can open the port and assert anything it likes about itself.
package otlz
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"os"
"strconv"
"strings"
"sync"
"github.com/luxfi/zap"
"github.com/zap-proto/zip/internal/sock"
)
// The signal message types, in the ZAP envelope's flags. They are a fleet-wide
// registry rather than this package's choice: 1 is spans (luxfi/trace), 2 is
// metrics (luxfi/metric), 3 is logs (luxfi/log), and every deployed collector
// switches on them. Append only — renumbering breaks every receiver at once.
const (
MsgSpan uint16 = 1
MsgMetric uint16 = 2
MsgLog uint16 = 3
)
// SpanBatch is one shipment of spans. Mirrors o11y zapreceiver.SpanBatch.
type SpanBatch struct {
AppName string `json:"appName,omitempty"`
Version string `json:"version,omitempty"`
Resource map[string]string `json:"resource,omitempty"`
Spans []Span `json:"spans"`
}
// Span is one operation. Mirrors o11y zapreceiver.Span.
//
// Kind and StatusCode are read through the collector's normEnum, which
// lowercases and strips a leading "span_kind_" / "status_code_", and maps
// "unset" to empty. So "server" and "SPAN_KIND_SERVER" are the same value, and
// the plain lowercase spelling is the one used here.
type Span struct {
TraceID string `json:"traceId"`
SpanID string `json:"spanId"`
ParentSpanID string `json:"parentSpanId,omitempty"`
Name string `json:"name"`
Kind string `json:"kind,omitempty"`
StartUnixNs int64 `json:"startUnixNs"`
EndUnixNs int64 `json:"endUnixNs"`
Attributes map[string]any `json:"attributes,omitempty"`
Events []SpanEvent `json:"events,omitempty"`
StatusCode string `json:"statusCode,omitempty"`
StatusMsg string `json:"statusMessage,omitempty"`
}
// SpanEvent is a point in time inside a span. Mirrors o11y zapreceiver.SpanEvent.
type SpanEvent struct {
Name string `json:"name"`
TimeUnixNs int64 `json:"timeUnixNs"`
Attributes map[string]any `json:"attributes,omitempty"`
}
// LogBatch is one shipment of records. Mirrors o11y zaplogreceiver.LogBatch.
type LogBatch struct {
AppName string `json:"appName,omitempty"`
Version string `json:"version,omitempty"`
Resource map[string]string `json:"resource,omitempty"`
Records []LogRecord `json:"records"`
}
// LogRecord is one line. Mirrors o11y zaplogreceiver.LogRecord.
//
// Severity is the OTel severity NUMBER, which the collector clamps to a uint8
// and stores beside the text: 1 trace, 5 debug, 9 info, 13 warn, 17 error,
// 21 fatal. TraceID and SpanID are what join a line to the span it happened
// under, and they are the reason a log address is worth configuring at all.
type LogRecord struct {
TimeUnixNs int64 `json:"timeUnixNs"`
ObservedTimeUnixNs int64 `json:"observedTimeUnixNs,omitempty"`
Severity int `json:"severity,omitempty"`
SeverityText string `json:"severityText,omitempty"`
Body string `json:"body,omitempty"`
Attributes map[string]any `json:"attributes,omitempty"`
TraceID string `json:"traceId,omitempty"`
SpanID string `json:"spanId,omitempty"`
EventName string `json:"eventName,omitempty"`
}
// The OTel severity numbers, named so a mapping reads as a mapping.
const (
SeverityDebug = 5
SeverityInfo = 9
SeverityWarn = 13
SeverityError = 17
)
// Signal is one of the three things a program reports, and everything that
// differs between them: the message type on the wire, the socket the collector
// serves it on, and the loopback port it falls back to.
//
// The three are separate because o11y receives them separately — one listener
// each, so "a signal must be able to fail, saturate or be firewalled on its
// own", as its config says. Stating them as one table is what keeps the
// producing side from inventing a fourth arrangement.
type Signal struct {
Name string // "spans", "logs", "metrics" — as o11y's own config names them
Msg uint16 // the ZAP message type
Sock string // the well-known service name its socket is reached by
Port int // the conventional loopback port
}
// The three signals, and the conventions that make telemetry need no
// configuration at all.
//
// A socket name is a service NAME in zip's existing scheme, so it resolves the
// way every other name does — <RuntimeDir()>/<name>.sock — and one environment
// variable already moves the whole fleet. The ports are the fleet's documented
// ones, which o11y's own config names in the same breath: spans 4317, logs 4318,
// metrics 4319.
var (
Spans = Signal{Name: "spans", Msg: MsgSpan, Sock: "o11y-spans", Port: 4317}
Logs = Signal{Name: "logs", Msg: MsgLog, Sock: "o11y-logs", Port: 4318}
Metrics = Signal{Name: "metrics", Msg: MsgMetric, Sock: "o11y-metrics", Port: 4319}
)
// Address is where a signal goes, decided the way zip decides everything else:
// by convention, with configuration available and never required.
//
// an override a deployment that states one always wins
// the socket if the collector's socket EXISTS, it is the destination
// loopback otherwise the conventional port on this host
//
// The socket is preferred wherever it is present, and its PRESENCE is the whole
// of the configuration — o11y's ingest running in this pod creates it, and that
// is the fact worth acting on. It is also the better channel: over a unix socket
// the kernel attests which process sent a batch, so a collector learns the
// sender rather than being told, exactly as zip documents for a Call. On TCP
// loopback any process on the box can claim to be any service.
//
// It is re-decided on every attempt rather than resolved once, because plugin
// start order is nobody's business: a program that starts before the collector
// must pick it up when it appears, without a restart.
func Address(s Signal, override string) string {
if o := strings.TrimSpace(override); o != "" {
return o
}
sock := socket(s)
if _, err := os.Stat(sock); err == nil {
return sock
}
return "127.0.0.1:" + strconv.Itoa(s.Port)
}
// socket is the collector's socket path for one signal, in the fleet's one
// scheme — the same function zip.SocketPath is, so a collector serving by name
// and a program looking for it cannot disagree.
func socket(s Signal) string { return sock.Path(s.Sock) }
// Export ships one signal's batches to o11y.
//
// Nothing is dialled until there is something to send, and a failure to send
// costs the caller nothing. A program on a laptop with no collector anywhere
// runs exactly as a program in a pod does: it reports, the report goes nowhere,
// and it never says so more than once.
//
// Fire and forget, always. Telemetry must never be able to stop the program it
// describes, so a batch that cannot be delivered is DROPPED rather than held —
// holding it is how a producer grows without bound while nobody is listening.
type Export struct {
sig Signal
override string
appName string
say func(string, ...any)
mu sync.Mutex
node *zap.Node
peer string
addr string // where the current connection went, for the report
up bool // last known state, so reporting is edge-triggered
said bool // whether a state has been reported at all yet
closed bool
}
// New prepares an exporter. It binds nothing, dials nothing and cannot fail:
// a collector that is absent, or that starts later, is the ordinary case.
//
// say is how it reports, and it is the APP's own logger rather than one of this
// package's choosing — an exporter's two lines are the program telling you
// something about itself, and they belong in the same stream, at the same level,
// with the same fields as everything else the program says.
func New(sig Signal, override, appName string, say func(string, ...any)) *Export {
if say == nil {
say = func(string, ...any) {}
}
return &Export{sig: sig, override: override, appName: appName, say: say}
}
// connect resolves the destination afresh and dials it.
//
// The address is passed to luxfi/zap untouched, which reads its FORM to choose
// unix or tcp — so a socket path and a host:port arrive here identically and
// neither is special-cased. That is the same discipline zip applies when
// serving, and it is why growing a socket ear on the collector needs no change
// on this side.
func (e *Export) connect() error {
if e.node == nil {
n := zap.NewNode(zap.NodeConfig{
NodeID: "otlz-" + e.appName + "-" + e.sig.Name,
ServiceType: "_o11y._tcp",
// The node's own chatter — started, peer connected, peer gone — is
// this package's internals, not the program's news. What an operator
// needs is the two edges below, in the program's own log.
Logger: slog.New(slog.DiscardHandler),
// Discovery would make an exporter announce itself and hunt for peers.
// It has one peer, and convention already names it.
NoDiscovery: true,
})
if err := n.Start(); err != nil {
return fmt.Errorf("start node: %w", err)
}
e.node = n
}
addr := Address(e.sig, e.override)
if err := e.node.ConnectDirect(addr); err != nil {
return fmt.Errorf("%s: %w", addr, err)
}
peers := e.node.Peers()
if len(peers) == 0 {
return fmt.Errorf("%s: connected but no peer", addr)
}
e.peer, e.addr = peers[0], addr
return nil
}
// report says when export STARTS working and when it STOPS, and nothing in
// between.
//
// Edge-triggered because the alternative is a line per failed flush forever on
// every laptop that has no collector — which trains everyone to ignore the
// stream that is supposed to carry the one line that matters. The transition is
// the event; the steady state is not.
func (e *Export) report(up bool, why error) {
if e.said && up == e.up {
return
}
e.said, e.up = true, up
if up {
e.say("telemetry export working", "signal", e.sig.Name, "addr", e.addr)
return
}
e.say("telemetry export stopped", "signal", e.sig.Name, "reason", why.Error())
}
// Spans ships a batch of spans.
func (e *Export) Spans(ctx context.Context, b SpanBatch) {
if len(b.Spans) == 0 {
return
}
b.AppName = e.appName
e.send(ctx, MsgSpan, &b)
}
// Logs ships a batch of records.
func (e *Export) Logs(ctx context.Context, b LogBatch) {
if len(b.Records) == 0 {
return
}
b.AppName = e.appName
e.send(ctx, MsgLog, &b)
}
// send is the one path a batch takes: resolve, dial, marshal, frame, write.
//
// Every failure ends here. A caller cannot act on a telemetry error — there is
// nothing sensible for a request handler to do about a collector being down — so
// none is returned, and the only trace of it is the edge-triggered line.
func (e *Export) send(ctx context.Context, msg uint16, batch any) {
e.mu.Lock()
defer e.mu.Unlock()
if e.closed {
return
}
// Dial lazily, and re-probe whenever there is no connection: the collector
// may have appeared since the last attempt, which is the normal case when
// plugin start order puts this program first.
if e.peer == "" {
if err := e.connect(); err != nil {
e.report(false, err)
return
}
}
payload, err := json.Marshal(batch)
if err != nil {
e.report(false, err)
return
}
m, err := zap.Parse(frame(msg, payload))
if err != nil {
e.report(false, err)
return
}
if err := e.node.Send(ctx, e.peer, m); err != nil {
// The connection is gone. Forget the peer so the next batch re-resolves
// and redials, rather than writing into a socket that will never drain.
e.peer = ""
e.report(false, err)
return
}
e.report(true, nil)
}
// frame wraps a JSON payload in the ZAP envelope every o11y receiver expects:
// a root object whose field 0 is the payload bytes, with the signal's message
// type in the upper 8 bits of the envelope flags.
func frame(msg uint16, payload []byte) []byte {
const envelope = 16
b := zap.NewBuilder(envelope + 64 + len(payload))
root := b.StartObject(envelope)
root.SetBytes(0, payload)
root.FinishAsRoot()
return b.FinishWithFlags(msg << 8)
}
// Close stops the node. Safe to call more than once, because shutdown paths run
// from more than one place.
func (e *Export) Close() {
e.mu.Lock()
defer e.mu.Unlock()
if e.closed {
return
}
e.closed = true
if e.node != nil {
e.node.Stop()
}
}
internal/otlz/otlz_test.go
-457
View File
@@ -1,457 +0,0 @@
package otlz_test
import (
"context"
"encoding/json"
"fmt"
"net"
"os"
"path/filepath"
"sync"
"testing"
"time"
"github.com/luxfi/zap"
"github.com/zap-proto/zip/internal/otlz"
)
// The shapes are pinned as BYTES, not against o11y's Go types.
//
// Decoding through the collector's own structs would be the stronger proof and
// it was measured rather than assumed: importing github.com/hanzoai/o11y for
// tests takes zip's go.sum from 149 lines to 720 and its module graph to 1145
// modules, pulling in Kubernetes apimachinery and klog. zip is the base every
// Hanzo program builds on, and a module graph is shared with every consumer, so
// that cost is paid by all 125 plugins to run a test here.
//
// So the field names below are transcribed from the receiver's own struct tags —
// o11y pkg/zapreceiver (SpanBatch, Span, SpanEvent) and pkg/zaplogreceiver
// (LogBatch, LogRecord) at v1.5.58 — and asserted exactly. If either end renames
// a field, this fails. That is the whole job: the two declarations are
// deliberate copies of one contract, and something has to hold them together.
// spanKeys is every JSON name the span receiver decodes, batch and span.
var spanKeys = struct{ batch, span, event []string }{
batch: []string{"appName", "version", "resource", "spans"},
span: []string{
"traceId", "spanId", "parentSpanId", "name", "kind",
"startUnixNs", "endUnixNs", "attributes", "events",
"statusCode", "statusMessage",
},
event: []string{"name", "timeUnixNs", "attributes"},
}
// logKeys is every JSON name the log receiver decodes.
var logKeys = struct{ batch, record []string }{
batch: []string{"appName", "version", "resource", "records"},
record: []string{
"timeUnixNs", "observedTimeUnixNs", "severity", "severityText",
"body", "attributes", "traceId", "spanId", "eventName",
},
}
// keysOf marshals v and returns its top-level object keys.
func keysOf(t *testing.T, v any) map[string]bool {
t.Helper()
b, err := json.Marshal(v)
if err != nil {
t.Fatalf("marshal: %v", err)
}
var m map[string]json.RawMessage
if err := json.Unmarshal(b, &m); err != nil {
t.Fatalf("unmarshal: %v", err)
}
out := map[string]bool{}
for k := range m {
out[k] = true
}
return out
}
// assertKeys checks the emitted object carries exactly want and nothing else.
// Extra keys matter as much as missing ones: a field this side invented is a
// field the collector drops, and silently.
func assertKeys(t *testing.T, what string, got map[string]bool, want []string) {
t.Helper()
for _, k := range want {
if !got[k] {
t.Errorf("%s: missing %q — the receiver decodes it", what, k)
}
}
for k := range got {
found := false
for _, w := range want {
if k == w {
found = true
break
}
}
if !found {
t.Errorf("%s: emits %q, which the receiver does not decode", what, k)
}
}
}
// filled builds values with every field set, so omitempty cannot hide a rename.
func filledSpan() otlz.SpanBatch {
return otlz.SpanBatch{
AppName: "svc",
Version: "v1",
Resource: map[string]string{"service.name": "svc"},
Spans: []otlz.Span{{
TraceID: "0102030405060708090a0b0c0d0e0f10",
SpanID: "1112131415161718",
ParentSpanID: "2122232425262728",
Name: "/v1/x",
Kind: "server",
StartUnixNs: 1700000000000000000,
EndUnixNs: 1700000000000001000,
Attributes: map[string]any{"http.method": "GET"},
Events: []otlz.SpanEvent{{Name: "e", TimeUnixNs: 1, Attributes: map[string]any{"k": "v"}}},
StatusCode: "Ok",
StatusMsg: "fine",
}},
}
}
func filledLog() otlz.LogBatch {
return otlz.LogBatch{
AppName: "svc",
Version: "v1",
Resource: map[string]string{"service.name": "svc"},
Records: []otlz.LogRecord{{
TimeUnixNs: 1700000000000000000,
ObservedTimeUnixNs: 1700000000000000001,
Severity: otlz.SeverityInfo,
SeverityText: "info",
Body: "request",
Attributes: map[string]any{"http.method": "GET"},
TraceID: "0102030405060708090a0b0c0d0e0f10",
SpanID: "1112131415161718",
EventName: "request",
}},
}
}
func TestSpanShapeMatchesReceiver(t *testing.T) {
b := filledSpan()
assertKeys(t, "SpanBatch", keysOf(t, b), spanKeys.batch)
assertKeys(t, "Span", keysOf(t, b.Spans[0]), spanKeys.span)
assertKeys(t, "SpanEvent", keysOf(t, b.Spans[0].Events[0]), spanKeys.event)
}
func TestLogShapeMatchesReceiver(t *testing.T) {
b := filledLog()
assertKeys(t, "LogBatch", keysOf(t, b), logKeys.batch)
assertKeys(t, "LogRecord", keysOf(t, b.Records[0]), logKeys.record)
}
// TestSignalNumbers pins the fleet-wide registry. These are not this package's
// to choose: every deployed collector switches on them, so a change here is a
// change to every receiver at once.
func TestSignalNumbers(t *testing.T) {
if otlz.MsgSpan != 1 {
t.Errorf("span message type = %d, want 1", otlz.MsgSpan)
}
if otlz.MsgLog != 3 {
t.Errorf("log message type = %d, want 3", otlz.MsgLog)
}
}
// ear is a ZAP listener that decodes one signal, standing in for o11y's
// receiver. It decodes the envelope exactly as the receiver does — payload out
// of field 0, message type out of the flags' upper byte — so what it accepts is
// what the collector accepts.
type ear struct {
node *zap.Node
mu sync.Mutex
got [][]byte
msgs []uint16
seen chan struct{}
}
func listen(t *testing.T, addr string, msg uint16) *ear {
t.Helper()
e := &ear{seen: make(chan struct{}, 8)}
e.node = zap.NewNode(zap.NodeConfig{
NodeID: "ear-" + addr,
ServiceType: "_o11y._tcp",
Address: addr,
NoDiscovery: true,
})
e.node.Handle(msg, func(_ context.Context, _ string, m *zap.Message) (*zap.Message, error) {
// Read the payload exactly as o11y's receivers do: field 0 of the root.
cp := append([]byte(nil), m.Root().Bytes(0)...)
e.mu.Lock()
e.got = append(e.got, cp)
e.msgs = append(e.msgs, msg)
e.mu.Unlock()
select {
case e.seen <- struct{}{}:
default:
}
// No reply: the sender does not wait, so answering would be a frame
// nothing reads.
return nil, nil
})
if err := e.node.Start(); err != nil {
t.Fatalf("ear on %s: %v", addr, err)
}
t.Cleanup(e.node.Stop)
return e
}
func (e *ear) await(t *testing.T) []byte {
t.Helper()
select {
case <-e.seen:
case <-time.After(5 * time.Second):
t.Fatal("nothing arrived at the collector")
}
e.mu.Lock()
defer e.mu.Unlock()
return e.got[len(e.got)-1]
}
// sock returns a short socket path. A unix socket path is capped near 104 bytes
// on darwin, and t.TempDir() alone can exceed it — which fails as "invalid
// argument" and looks like a bug in the code under test.
func sock(t *testing.T, name string) string {
t.Helper()
d, err := os.MkdirTemp("", "otlz")
if err != nil {
t.Fatalf("tempdir: %v", err)
}
t.Cleanup(func() { _ = os.RemoveAll(d) })
return filepath.Join(d, name)
}
// TestSpansOverSocket and its siblings are the transport claim: the SAME
// exporter reaches a collector over a unix socket, chosen by nothing but the
// form of the address. No scheme is parsed here and none is passed.
func TestSpansOverSocket(t *testing.T) {
addr := sock(t, "s.sock")
e := listen(t, addr, otlz.MsgSpan)
x := otlz.New(otlz.Spans, addr, "svc", nil)
defer x.Close()
x.Spans(context.Background(), filledSpan())
var got otlz.SpanBatch
if err := json.Unmarshal(e.await(t), &got); err != nil {
t.Fatalf("collector could not decode: %v", err)
}
if len(got.Spans) != 1 || got.Spans[0].TraceID != filledSpan().Spans[0].TraceID {
t.Fatalf("span did not survive the socket: %+v", got.Spans)
}
if got.AppName != "svc" {
t.Errorf("appName = %q, want svc — the exporter stamps it", got.AppName)
}
}
func TestLogsOverSocket(t *testing.T) {
addr := sock(t, "l.sock")
e := listen(t, addr, otlz.MsgLog)
x := otlz.New(otlz.Logs, addr, "svc", nil)
defer x.Close()
x.Logs(context.Background(), filledLog())
var got otlz.LogBatch
if err := json.Unmarshal(e.await(t), &got); err != nil {
t.Fatalf("collector could not decode: %v", err)
}
if len(got.Records) != 1 || got.Records[0].Body != "request" {
t.Fatalf("record did not survive the socket: %+v", got.Records)
}
if got.Records[0].TraceID == "" || got.Records[0].SpanID == "" {
t.Error("the record lost its trace ids, which are the reason logs ship at all")
}
}
// TestOverTCP proves the same exporter, unchanged, reaches a host:port
// collector. Together with the socket tests this is the whole transport claim:
// one code path, two transports, chosen by the address.
func TestOverTCP(t *testing.T) {
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("port: %v", err)
}
port := l.Addr().(*net.TCPAddr).Port
if err := l.Close(); err != nil {
t.Fatalf("close: %v", err)
}
addr := fmt.Sprintf("127.0.0.1:%d", port)
e := listen(t, addr, otlz.MsgSpan)
x := otlz.New(otlz.Spans, addr, "svc", nil)
defer x.Close()
x.Spans(context.Background(), filledSpan())
var got otlz.SpanBatch
if err := json.Unmarshal(e.await(t), &got); err != nil {
t.Fatalf("collector could not decode: %v", err)
}
if len(got.Spans) != 1 {
t.Fatalf("span did not survive tcp: %+v", got.Spans)
}
}
// TestUnreachableCollectorIsSurvivable is the fire-and-forget claim. An address
// nothing listens on must cost the caller nothing — no error to handle, no
// panic, no block — because telemetry may never be able to stop the program it
// describes.
func TestUnreachableCollectorIsSurvivable(t *testing.T) {
x := otlz.New(otlz.Spans, sock(t, "nobody.sock"), "svc", nil)
defer x.Close()
done := make(chan struct{})
go func() {
x.Spans(context.Background(), filledSpan())
x.Logs(context.Background(), filledLog())
close(done)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("exporting to a dead collector blocked the caller")
}
}
// TestEmptyBatchSendsNothing keeps idle services quiet: a tick with no traffic
// must not open a connection or write a frame.
func TestEmptyBatchSendsNothing(t *testing.T) {
addr := sock(t, "quiet.sock")
e := listen(t, addr, otlz.MsgSpan)
x := otlz.New(otlz.Spans, addr, "svc", nil)
defer x.Close()
x.Spans(context.Background(), otlz.SpanBatch{})
select {
case <-e.seen:
t.Fatal("an empty batch was sent")
case <-time.After(300 * time.Millisecond):
}
}
// --- the conventions ------------------------------------------------------
// TestConventionPrefersThePresentSocket is the zero-configuration claim: with
// the collector's socket on the runtime directory, that is where a report goes,
// and nothing was configured to make it so.
func TestConventionPrefersThePresentSocket(t *testing.T) {
dir, err := os.MkdirTemp("", "rt")
if err != nil {
t.Fatalf("tempdir: %v", err)
}
t.Cleanup(func() { _ = os.RemoveAll(dir) })
t.Setenv("ZIP_RUNTIME_DIR", dir)
want := filepath.Join(dir, "o11y-spans.sock")
if got := otlz.Address(otlz.Spans, ""); got == want {
t.Fatal("the socket was chosen before it existed; presence is the configuration")
}
// Create it, and the same call must now resolve to it.
f, err := os.Create(want)
if err != nil {
t.Fatalf("create: %v", err)
}
_ = f.Close()
if got := otlz.Address(otlz.Spans, ""); got != want {
t.Errorf("address = %q, want the socket %q now that it exists", got, want)
}
}
// TestConventionFallsBackToLoopback pins the second rule: no socket, no
// configuration, and a report still has somewhere to go.
func TestConventionFallsBackToLoopback(t *testing.T) {
dir, err := os.MkdirTemp("", "rt")
if err != nil {
t.Fatalf("tempdir: %v", err)
}
t.Cleanup(func() { _ = os.RemoveAll(dir) })
t.Setenv("ZIP_RUNTIME_DIR", dir)
for _, c := range []struct {
sig otlz.Signal
want string
}{
{otlz.Spans, "127.0.0.1:4317"},
{otlz.Logs, "127.0.0.1:4318"},
{otlz.Metrics, "127.0.0.1:4319"},
} {
if got := otlz.Address(c.sig, ""); got != c.want {
t.Errorf("%s address = %q, want %q", c.sig.Name, got, c.want)
}
}
}
// TestOverrideWins keeps configuration available for the remote collector,
// without it ever being required.
func TestOverrideWins(t *testing.T) {
dir, err := os.MkdirTemp("", "rt")
if err != nil {
t.Fatalf("tempdir: %v", err)
}
t.Cleanup(func() { _ = os.RemoveAll(dir) })
t.Setenv("ZIP_RUNTIME_DIR", dir)
f, _ := os.Create(filepath.Join(dir, "o11y-spans.sock"))
if f != nil {
_ = f.Close()
}
if got := otlz.Address(otlz.Spans, "collector.example:4317"); got != "collector.example:4317" {
t.Errorf("address = %q, want the override even with a socket present", got)
}
}
// TestReportIsEdgeTriggered is the "must not spam" claim. A program with no
// collector answers requests forever; it must say so once, not once per flush.
func TestReportIsEdgeTriggered(t *testing.T) {
var mu sync.Mutex
var lines []string
say := func(msg string, _ ...any) {
mu.Lock()
lines = append(lines, msg)
mu.Unlock()
}
x := otlz.New(otlz.Spans, sock(t, "absent.sock"), "svc", say)
defer x.Close()
for range 10 {
x.Spans(context.Background(), filledSpan())
}
mu.Lock()
defer mu.Unlock()
if len(lines) != 1 {
t.Fatalf("ten failed exports produced %d lines, want exactly 1: %v", len(lines), lines)
}
if lines[0] != "telemetry export stopped" {
t.Errorf("line = %q, want the stopped edge", lines[0])
}
}
// TestReportSaysWhenExportStartsWorking is the other edge, and the one that
// matters for start order: a collector that appears later is picked up, and the
// program says so once.
func TestReportSaysWhenExportStartsWorking(t *testing.T) {
var mu sync.Mutex
var lines []string
say := func(msg string, _ ...any) {
mu.Lock()
lines = append(lines, msg)
mu.Unlock()
}
addr := sock(t, "late.sock")
x := otlz.New(otlz.Spans, addr, "svc", say)
defer x.Close()
// Nothing is listening yet.
x.Spans(context.Background(), filledSpan())
// The collector starts AFTER the program did.
listen(t, addr, otlz.MsgSpan)
x.Spans(context.Background(), filledSpan())
mu.Lock()
defer mu.Unlock()
if len(lines) != 2 || lines[0] != "telemetry export stopped" || lines[1] != "telemetry export working" {
t.Fatalf("want stopped then working, got %v", lines)
}
}
telemetry.go
+169 -109
View File
@@ -14,7 +14,7 @@ import (
luxlog "github.com/luxfi/log"
"github.com/luxfi/metric"
fiber "github.com/zap-proto/fiber/v3"
"github.com/zap-proto/zip/internal/otlz"
"github.com/zap-proto/zip/internal/o11y"
)
// Telemetry is a property of running, not a feature an app switches on.
@@ -39,6 +39,14 @@ import (
// luxlog.Logger and stays one. zip writes the request line through it, so an
// app's fields, level and destination are the app's; the LINE is the
// framework's.
//
// Which is also the answer to "why is only the request line in the store": the
// request line is the only line zip writes, so it is the only one zip can ship.
// Everything a handler logs goes where the app's logger points it, untouched.
// Taking every line off the box is one decision about logging in general, and it
// belongs to luxfi/log — the thing that already sees every line, for every
// program, whether or not that program serves requests. Made here it would be
// zip deciding for 125 programs that a log line leaves the machine.
// The one identifier a request carries between services, and the header it
// travels in.
@@ -62,57 +70,100 @@ const HeaderTrace = "traceparent"
// link is visible, a misparsed one is not.
const traceVersion = "00"
// Telemetry is the override, not the setup. Every field may stay empty forever.
// Telemetry is where this app's three reports go. Every field may stay empty
// forever, and an empty field means that signal is not exported.
//
// A program reports because it exists. It needs no address, no environment
// variable and no line of configuration, because zip already knows where a
// report goes — by the same conventions it already uses for everything else:
// THE ADDRESS IS THE SWITCH, and it is the switch on both ends: the collector
// states exactly this rule for receiving — an empty address means that signal is
// not received in this process — so one sentence describes the producer and the
// consumer, and there is no enable flag on either side that can disagree with an
// address. A program with nothing stated still logs and still counts; it simply
// has nowhere to send, which costs it no goroutine, no connection and no port.
//
// the socket <RuntimeDir()>/o11y-<signal>.sock, if it EXISTS
// loopback 127.0.0.1:4317 spans, :4318 logs, :4319 metrics
// A field is read first, and the matching variable in the environment second, so
// a deployment that states an address in its manifest and a process that
// inherits one from its host agree on which wins — the explicit one:
//
// The socket wins wherever it is present, and its presence is the entire
// configuration: o11y's ingest running in this pod creates it, and that is the
// fact worth acting on rather than a setting somebody has to keep in step with
// it. It is also the better channel — over a unix socket the kernel attests
// which process sent a batch, so a collector learns the sender instead of being
// told, exactly as zip documents for a Call. On TCP loopback any process on the
// box can claim to be any service.
// Spans O11Y_SPANS_ADDR
// Logs O11Y_LOGS_ADDR
// Metrics O11Y_METRICS_ADDR
//
// The destination is re-decided whenever there is no connection, so start order
// is nobody's business: a program that starts before the collector picks it up
// when it appears, with no restart. A program with no collector at all — a
// laptop, a test — reports into nothing, costs nothing, and says so exactly once.
// Three names rather than one base address with the three conventional ports
// derived from it (spans 4317, logs 4318, metrics 4319). The collector binds one
// listener per signal at one address per signal, refuses two signals that name
// the same address, and receives whichever signals have one — so deriving two
// addresses from a third would assert an adjacency it does not promise, and
// would make "logs are off, spans are on" unsayable. Sibling names also state
// which port belongs to which signal at the place a deployment sets it, and a
// batch sent to the wrong ear is dropped by a receiver that decodes one message
// type, silently on both sides.
//
// What is reported is not configurable at all, because a fleet whose members
// report different shapes cannot be read as a fleet.
type Telemetry struct {
// Spans overrides where spans go, for a collector that is not in this pod
// and not on this host. Empty reads OTEL_EXPORTER_ZAP_ENDPOINT — the name the
// fleet already deploys — and then falls back to the conventions above.
// Spans is where this app's server spans go, as host:port.
Spans string
// Logs overrides where the request line goes. Empty reads
// O11Y_LOGS_ZAP_ENDPOINT, then convention.
// Logs is where this app's request line goes, as host:port.
//
// Only zip's OWN request line ships here. An app's logging is its own: its
// luxlog.Logger writes wherever the app pointed it and is never intercepted,
// teed or rewritten by the framework. Shipping a program's every log line off
// the box is a decision that belongs to the program, and taking it here would
// mean zip quietly deciding it for 125 of them.
// Only zip's OWN request line ships here — the same line, with the same
// fields, that the app's logger already writes. An app's other logging is
// its own: its luxlog.Logger writes wherever the app pointed it and is
// never intercepted, teed or rewritten by the framework. Shipping a
// program's every log line off the box is one decision for the logger to
// make, in luxfi/log, once, for everything that logs; taking it here would
// mean zip quietly making it for 125 programs that never asked.
Logs string
// Metrics overrides where the numbers go. Empty reads
// O11Y_METRICS_ZAP_ENDPOINT, then convention.
// Metrics is where the numbers go, as host:port.
Metrics string
// Off stops this app reporting at all: no line, no numbers, no spans, no
// trace context. It is for the app that must not emit — a test double, a
// one-shot command — and it is the only switch, so "is this app
// one-shot command — and it is the only switch of its kind, so "is this app
// instrumented" has one answer and one place to read it.
Off bool
}
// The environment's spelling of the three addresses. A deployment sets these;
// they are the names cloud's manifest already uses for the metric ear, extended
// to its two siblings.
const (
envSpans = "O11Y_SPANS_ADDR"
envLogs = "O11Y_LOGS_ADDR"
envMetrics = "O11Y_METRICS_ADDR"
)
// addresses is where one app reports, resolved once when the app is built.
//
// Once, because there is nothing left to re-decide: a field is fixed at
// construction and the environment of a running process does not change. The
// three are held together in one value so "does this app export anything" is one
// question about one thing.
type addresses struct{ spans, logs, metrics string }
// resolve reads the three addresses: what the program stated, else what its
// environment states, else nothing.
func resolve(cfg Config) addresses {
pick := func(stated, env string) string {
if v := strings.TrimSpace(stated); v != "" {
return v
}
return strings.TrimSpace(os.Getenv(env))
}
t := cfg.Telemetry
return addresses{
spans: pick(t.Spans, envSpans),
logs: pick(t.Logs, envLogs),
metrics: pick(t.Metrics, envMetrics),
}
}
// any reports whether this app has anywhere at all to report to. It is what
// makes "off" cost nothing: with no address there is no exporter, so no node
// starts, no port is taken and no connection is attempted.
func (a addresses) any() bool { return a.spans != "" || a.logs != "" || a.metrics != "" }
// telemetry is one app's instruments. Built once in [newApp], read from every
// serving goroutine, never rebuilt: a generation swap re-installs the handler
// but must not reset the counters, or every composition change would look like
@@ -142,8 +193,13 @@ type telemetry struct {
plugins metric.GaugeVec
dropped metric.Counter
// Where this app reports, resolved once when it was built.
to addresses
// ship says this app has somewhere to send spans and lines, so the boundary
// should collect them.
// should collect them. False when neither address is stated, and then the
// boundary keeps nothing: a buffer nobody will drain is a memory leak with a
// good reason attached.
//
// It is decided WITH THE APP rather than when the export goroutine starts,
// and that is not a micro-optimisation: Serve returns as soon as the serving
@@ -161,8 +217,8 @@ type telemetry struct {
// request the span describes, which is the one place telemetry must never
// be. See [telemetry.keep].
pending sync.Mutex
spans []otlz.Span
records []otlz.LogRecord
spans []o11y.Span
records []o11y.LogRecord
}
// pendingLimit caps what one export interval may accumulate.
@@ -195,11 +251,11 @@ func newTelemetry(cfg Config) *telemetry {
if t.off {
return t
}
// There is always somewhere for a report to go — a socket if the collector
// is in this pod, loopback otherwise — so an app that is not silenced always
// collects. That is what "no configuration" means: nothing to set, and no
// arrangement in which a program is quietly not reporting.
t.ship = true
// An app collects spans and lines when it has somewhere to send them, and
// only then. The numbers are different: they are always measured, because
// /metrics serves them off this registry whether or not anything is pushed.
t.to = resolve(cfg)
t.ship = t.to.spans != "" || t.to.logs != ""
t.registry = metric.NewRegistry()
t.requests = t.registry.NewCounterVec(
"http_requests_total",
@@ -234,7 +290,7 @@ func newTelemetry(cfg Config) *telemetry {
// they describe the same request: dropping the span while keeping the line would
// leave a log record pointing at a trace that has no span in it, which reads as
// a broken trace rather than as a dropped one.
func (t *telemetry) keep(s otlz.Span, r otlz.LogRecord) {
func (t *telemetry) keep(s o11y.Span, r o11y.LogRecord) {
t.pending.Lock()
defer t.pending.Unlock()
if len(t.spans) >= pendingLimit {
@@ -251,7 +307,7 @@ func (t *telemetry) keep(s otlz.Span, r otlz.LogRecord) {
// drain takes everything pending and hands the buffers back empty. The caller
// ships outside the lock, so a slow collector never blocks a request that is
// only trying to append.
func (t *telemetry) drain() ([]otlz.Span, []otlz.LogRecord) {
func (t *telemetry) drain() ([]o11y.Span, []o11y.LogRecord) {
t.pending.Lock()
defer t.pending.Unlock()
s, r := t.spans, t.records
@@ -325,11 +381,19 @@ func (a *App) report() Handler {
// not a metric label. A trace store is built to hold one row per request and is
// searched by exactly this — "show me the request for order 8a3f" — where the
// same value in a metric label would create a time series per order.
//
// The KEYS are the collector's vocabulary, not a spelling of our own. It reads
// a span's path column out of the first of url.path, http.target, http.route,
// path that a batch carries, and reads a log record's path column out of the
// same four — one vocabulary across both signals, so a query written against
// traces answers over logs too. A key it does not know is stored and searchable
// but fills no column, which is how a report can look complete and leave the
// column an operator actually searches empty.
func facts(c *Ctx, method, route, path string, status int) map[string]any {
f := map[string]any{
"http.method": method,
"http.path": path,
"http.status_code": status,
"http.request.method": method,
"url.path": path,
"http.response.status_code": status,
}
if route != "" {
f["http.route"] = route
@@ -353,14 +417,14 @@ func facts(c *Ctx, method, route, path string, status int) map[string]any {
// is bounded: a trace store groups by operation name to answer "how is
// /v1/orders/:id doing", and a name per order id makes that question
// unanswerable. The path is an attribute, where the cardinality belongs.
func span(tr trace, method, route, path string, status int, dur time.Duration, start time.Time, err error, c *Ctx) otlz.Span {
func span(tr trace, method, route, path string, status int, dur time.Duration, start time.Time, err error, c *Ctx) o11y.Span {
name := route
if name == "" {
// Nothing matched, so there is no operation to name. The method alone is
// bounded and honest — the path it was looking for rides as an attribute.
name = method
}
s := otlz.Span{
s := o11y.Span{
TraceID: tr.trace,
SpanID: tr.span,
ParentSpanID: tr.parent,
@@ -372,13 +436,16 @@ func span(tr trace, method, route, path string, status int, dur time.Duration, s
StartUnixNs: start.UnixNano(),
EndUnixNs: start.Add(dur).UnixNano(),
Attributes: facts(c, method, route, path, status),
StatusCode: "Ok",
// Lowercase because that is what is stored: the collector lowercases the
// value and strips a "status_code_" prefix, so sending what a query will
// match saves everyone one translation.
StatusCode: "ok",
}
// A span is in error when THIS service failed. A refusal is a correct answer
// to a bad request, and marking it an error makes every trace view that
// filters on errors mostly noise.
if status >= 500 {
s.StatusCode = "Error"
s.StatusCode = "error"
if err != nil {
s.StatusMsg = err.Error()
}
@@ -388,18 +455,21 @@ func span(tr trace, method, route, path string, status int, dur time.Duration, s
// record renders the same request as the log line that ships.
//
// It carries the trace and span ids, which is the whole reason a log address is
// worth setting: they are what lets a store put this line inside the span above
// it rather than beside it.
func record(tr trace, method, route, path string, status int, dur time.Duration, err error, c *Ctx) otlz.LogRecord {
// It is the line [emit] writes, in the shape the collector decodes: same
// request, same values, same one line for it, so the stream an operator greps
// on the box and the stream they query in the store cannot tell different
// stories. It carries the trace and span ids too, which is the whole reason a
// log address is worth setting: they are what lets a store put this line INSIDE
// the span above it rather than beside it.
func record(tr trace, method, route, path string, status int, dur time.Duration, err error, c *Ctx) o11y.LogRecord {
f := facts(c, method, route, path, status)
f["duration_ms"] = dur.Milliseconds()
if err != nil {
f["error"] = err.Error()
}
r := otlz.LogRecord{
r := o11y.LogRecord{
TimeUnixNs: time.Now().UnixNano(),
Severity: otlz.SeverityInfo,
Severity: o11y.SeverityInfo,
SeverityText: "info",
Body: "request",
Attributes: f,
@@ -407,7 +477,7 @@ func record(tr trace, method, route, path string, status int, dur time.Duration,
SpanID: tr.span,
}
if status >= 500 {
r.Severity, r.SeverityText = otlz.SeverityError, "error"
r.Severity, r.SeverityText = o11y.SeverityError, "error"
}
return r
}
@@ -653,27 +723,6 @@ func (a *App) gather() ([]*metric.MetricFamily, error) {
return t.registry.Gather()
}
// addresses are the three destinations, resolved once.
//
// The environment is consulted only when the config says nothing, so a
// deployment that states an address in its manifest and a process that inherits
// one from its host agree on which wins — the explicit one. Each name belongs
// to exactly one signal, because the ports do: a batch sent to the wrong ear is
// dropped by a receiver that only answers one message type, and the failure is
// silent on both sides.
func addresses(cfg Config) (spans, logs, metrics string) {
pick := func(stated, env string) string {
if v := strings.TrimSpace(stated); v != "" {
return v
}
return strings.TrimSpace(os.Getenv(env))
}
t := cfg.Telemetry
return pick(t.Spans, "OTEL_EXPORTER_ZAP_ENDPOINT"),
pick(t.Logs, "O11Y_LOGS_ZAP_ENDPOINT"),
pick(t.Metrics, "O11Y_METRICS_ZAP_ENDPOINT")
}
// exports reports whether anything is collected for shipment. A process with no
// span or log address still measures and still logs; it just does not fill a
// buffer nothing will drain.
@@ -711,7 +760,7 @@ func (a *App) Metrics() metric.Registerer { return a.telemetry.registry }
// depending on which way the data arrived.
const exportInterval = 15 * time.Second
// export ships this app's telemetry to o11y until it shuts down.
// export ships this app's telemetry to the collector until it shuts down.
//
// It is a PUSH because that is what the receiving side accepts: o11y ingests
// what services send it and scrapes nothing (its own prober checks liveness,
@@ -719,47 +768,54 @@ const exportInterval = 15 * time.Second
// and scraped — is a service registry we would have to run in order to learn
// what we already know at the moment a process starts serving.
//
// Three signals, three addresses, three connections, mirroring the three ears
// o11y binds — a signal that saturates or is firewalled takes only itself down.
// Metrics ride the exporter luxfi/metric already ships, because it is this same
// pattern plus the family translation, and a second implementation here would be
// a second thing to keep in step with the collector.
// One address and one connection per signal, mirroring the ears the collector
// binds — a signal that saturates or is firewalled takes only itself down. A
// signal with no address gets no exporter at all, so an app that reports nowhere
// starts no goroutine, opens no connection and takes no port: it returns right
// here. Metrics ride the exporter luxfi/metric already ships, because that is
// this same shape plus the family translation, and a second implementation here
// would be a second thing to keep in step with the collector.
//
// Nothing here can fail loudly. A collector that is down, moved or not yet
// deployed must not take an app with it, so every failure is a debug line and
// the next tick tries again.
func (a *App) export() {
if a.telemetry.off {
t := a.telemetry
if t.off || !t.to.any() {
return
}
spans, logs, metrics := addresses(a.cfg)
stop := make(chan struct{})
var closers []func(context.Context) error
// Nothing dials here. Each exporter resolves its destination when it first
// has something to send, and re-resolves whenever it has no connection, so a
// collector that starts after this program is picked up without a restart.
spanOut := otlz.New(otlz.Spans, spans, a.cfg.AppName, a.logger.Info)
logOut := otlz.New(otlz.Logs, logs, a.cfg.AppName, a.logger.Info)
closers = append(closers,
func(context.Context) error { spanOut.Close(); return nil },
func(context.Context) error { logOut.Close(); return nil },
)
// Nothing dials here. Each exporter connects when it first has something to
// send, and reconnects whenever it has none, so a collector that starts
// after this program is picked up without a restart.
var spanOut, logOut *o11y.Export
if t.to.spans != "" {
spanOut = o11y.New(o11y.Spans, t.to.spans, a.cfg.AppName, a.logger.Info)
closers = append(closers, func(context.Context) error { spanOut.Close(); return nil })
}
if t.to.logs != "" {
logOut = o11y.New(o11y.Logs, t.to.logs, a.cfg.AppName, a.logger.Info)
closers = append(closers, func(context.Context) error { logOut.Close(); return nil })
}
// Metrics ride the exporter luxfi/metric already ships, which is this same
// pattern plus the family translation. It takes a fixed endpoint rather than
// resolving one, so the convention is applied here instead.
metricOut, err := metric.NewZAPExporter(metric.ZAPExporterConfig{
Endpoint: otlz.Address(otlz.Metrics, metrics),
AppName: a.cfg.AppName,
Resource: map[string]string{"service.name": a.cfg.AppName},
})
if err != nil {
a.logger.Error("zip telemetry metrics export unavailable", "err", err)
metricOut = nil
} else {
closers = append(closers, metricOut.Shutdown)
// shape plus the family translation. It is built only when there is an
// address, because building it starts a node and takes a port immediately.
var metricOut *metric.ZAPExporter
if t.to.metrics != "" {
out, err := metric.NewZAPExporter(metric.ZAPExporterConfig{
Endpoint: t.to.metrics,
AppName: a.cfg.AppName,
})
if err != nil {
a.logger.Error("zip telemetry metrics export unavailable", "err", err)
} else {
metricOut = out
closers = append(closers, out.Shutdown)
}
}
a.OnShutdown(func(ctx context.Context) error {
@@ -797,16 +853,20 @@ func (a *App) export() {
// The buffer is drained ONCE and both batches are built from that one drain, so
// a span and the line describing the same request leave together. Draining per
// signal would let a shutdown deliver the spans and lose the lines.
func (a *App) flush(ctx context.Context, spanOut, logOut *otlz.Export) {
//
// The batch names the app and nothing else. The collector lifts a batch's app
// name into service.name when the resource does not already carry one, so
// saying it twice would be two places to disagree about what this program is
// called.
func (a *App) flush(ctx context.Context, spanOut, logOut *o11y.Export) {
if !a.telemetry.ship {
return
}
spans, records := a.telemetry.drain()
resource := map[string]string{"service.name": a.cfg.AppName}
if spanOut != nil {
spanOut.Spans(ctx, otlz.SpanBatch{Resource: resource, Spans: spans})
spanOut.Spans(ctx, o11y.SpanBatch{Spans: spans})
}
if logOut != nil {
logOut.Logs(ctx, otlz.LogBatch{Resource: resource, Records: records})
logOut.Logs(ctx, o11y.LogBatch{Records: records})
}
}
telemetry_internal_test.go
+118
View File
@@ -0,0 +1,118 @@
package zip
import (
"testing"
luxlog "github.com/luxfi/log"
)
// The address is the switch, so these tests are about one question: what does
// this app export, and what does it cost when the answer is nothing.
// TestEnvironmentNames pins the spelling a deployment sets. A manifest and a
// program have to agree on it letter for letter, and a rename here is invisible
// at runtime — an unread variable and an absent one look identical.
func TestEnvironmentNames(t *testing.T) {
for _, c := range []struct{ got, want string }{
{envSpans, "O11Y_SPANS_ADDR"},
{envLogs, "O11Y_LOGS_ADDR"},
{envMetrics, "O11Y_METRICS_ADDR"},
} {
if c.got != c.want {
t.Errorf("environment name = %q, want %q", c.got, c.want)
}
}
}
// TestResolveReadsTheStatedAddressThenTheEnvironment covers the whole rule: a
// field first, the environment second, nothing otherwise.
func TestResolveReadsTheStatedAddressThenTheEnvironment(t *testing.T) {
t.Run("nothing stated anywhere", func(t *testing.T) {
t.Setenv(envSpans, "")
t.Setenv(envLogs, "")
t.Setenv(envMetrics, "")
got := resolve(Config{})
if got.any() {
t.Errorf("resolved %+v with nothing configured; an unstated signal is not exported", got)
}
})
t.Run("the environment states them", func(t *testing.T) {
t.Setenv(envSpans, "127.0.0.1:4317")
t.Setenv(envLogs, "127.0.0.1:4318")
t.Setenv(envMetrics, "127.0.0.1:4319")
got := resolve(Config{})
want := addresses{spans: "127.0.0.1:4317", logs: "127.0.0.1:4318", metrics: "127.0.0.1:4319"}
if got != want {
t.Errorf("resolved %+v, want %+v", got, want)
}
})
t.Run("the program's own field wins", func(t *testing.T) {
t.Setenv(envSpans, "127.0.0.1:4317")
got := resolve(Config{Telemetry: Telemetry{Spans: "collector:4317"}})
if got.spans != "collector:4317" {
t.Errorf("spans = %q, want the address the program stated", got.spans)
}
})
t.Run("blank is not an address", func(t *testing.T) {
t.Setenv(envSpans, " ")
if got := resolve(Config{Telemetry: Telemetry{Spans: " "}}); got.any() {
t.Errorf("resolved %+v from whitespace; that is a variable somebody meant to unset", got)
}
})
}
// TestNoAddressStartsNothing is the cost of being off, stated where it can be
// measured: export returns before it registers the hook that its goroutine,
// its exporters and its connections all hang from. No hook is therefore no
// goroutine, no node, no port and no dial.
func TestNoAddressStartsNothing(t *testing.T) {
t.Setenv(envSpans, "")
t.Setenv(envLogs, "")
t.Setenv(envMetrics, "")
quiet := New(Config{AppName: "quiet", Logger: luxlog.NewNoOpLogger(), DisableStartupMessage: true})
before := len(quiet.hooks)
quiet.export()
if len(quiet.hooks) != before {
t.Fatalf("an app with nowhere to report registered %d shutdown hooks", len(quiet.hooks)-before)
}
// The same app with one address stated does start, so the test above is
// measuring the switch rather than a broken export.
loud := New(Config{
AppName: "loud",
Logger: luxlog.NewNoOpLogger(),
DisableStartupMessage: true,
Telemetry: Telemetry{Spans: "127.0.0.1:1"},
})
loud.export()
if len(loud.hooks) == 0 {
t.Fatal("an app with a span address registered no shutdown hook, so nothing drains it")
}
if err := loud.Shutdown(); err != nil {
t.Fatalf("shutdown: %v", err)
}
}
// TestOffKeepsNothing proves the buffer follows the address: an app that cannot
// ship must not accumulate spans and records for a drain that will never come.
func TestOffKeepsNothing(t *testing.T) {
t.Setenv(envSpans, "")
t.Setenv(envLogs, "")
t.Setenv(envMetrics, "")
if newTelemetry(Config{}).exports() {
t.Error("an app with no span or log address collects for a shipment that cannot happen")
}
if !newTelemetry(Config{Telemetry: Telemetry{Logs: "127.0.0.1:4318"}}).exports() {
t.Error("an app with a log address does not collect, so its flush has nothing to send")
}
// The metric address alone does not make the boundary collect: the numbers
// come from the registry at flush time, not from a buffer.
if newTelemetry(Config{Telemetry: Telemetry{Metrics: "127.0.0.1:4319"}}).exports() {
t.Error("a metrics-only app buffers spans and records nothing will drain")
}
}
telemetry_test.go
+91 -59
View File
@@ -3,11 +3,11 @@ package zip_test
import (
"context"
"encoding/json"
"fmt"
"io"
"net"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"testing"
@@ -369,22 +369,26 @@ func (c *collector) await(t *testing.T) []byte {
return c.body[len(c.body)-1]
}
func socketAt(t *testing.T, name string) string {
// free returns a loopback address nothing is listening on yet. host:port is the
// only address form the collector serves, so it is the only one tested.
func free(t *testing.T) string {
t.Helper()
d, err := os.MkdirTemp("", "zt")
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("tempdir: %v", err)
t.Fatalf("port: %v", err)
}
t.Cleanup(func() { _ = os.RemoveAll(d) })
return filepath.Join(d, name)
port := l.Addr().(*net.TCPAddr).Port
if err := l.Close(); err != nil {
t.Fatalf("close: %v", err)
}
return fmt.Sprintf("127.0.0.1:%d", port)
}
// TestBoundaryExportsASpanOverASocket is the end-to-end claim, and it is stated
// over a UNIX SOCKET because that is where both ends of a pod-local export meet.
// An app registers one route, answers one request, and a span for that request
// arrives at a collector — with no telemetry code in the app at all.
func TestBoundaryExportsASpanOverASocket(t *testing.T) {
addr := socketAt(t, "spans.sock")
// TestBoundaryExportsASpan is the end-to-end claim: an app registers one route,
// answers one request, and a span for that request arrives at a collector — with
// no telemetry code in the app at all.
func TestBoundaryExportsASpan(t *testing.T) {
addr := free(t)
got := collect(t, addr, 1) // 1 = spans, the fleet-wide message type
app := zip.New(zip.Config{
@@ -395,8 +399,7 @@ func TestBoundaryExportsASpanOverASocket(t *testing.T) {
})
app.Get("/v1/orders/:id", func(c *zip.Ctx) error { return c.String(200, "ok") })
sockPath := socketAt(t, "app.sock")
h, err := zip.Serve(app, sockPath)
h, err := zip.Serve(app, "http://"+free(t))
if err != nil {
t.Fatalf("serve: %v", err)
}
@@ -443,22 +446,23 @@ func TestBoundaryExportsASpanOverASocket(t *testing.T) {
if s.Kind != "server" {
t.Errorf("kind = %q, want server", s.Kind)
}
if s.StatusCode != "Ok" {
t.Errorf("statusCode = %q, want Ok for a 200", s.StatusCode)
if s.StatusCode != "ok" {
t.Errorf("statusCode = %q, want ok for a 200", s.StatusCode)
}
if s.End <= s.Start {
t.Errorf("span does not span time: start=%d end=%d", s.Start, s.End)
}
// The path rides as an ATTRIBUTE, where cardinality belongs.
if s.Attributes["http.path"] != "/v1/orders/8a3f" {
t.Errorf("http.path = %v, want the path that arrived", s.Attributes["http.path"])
// The path rides as an ATTRIBUTE, where cardinality belongs, under the key
// the collector reads a span's path column out of.
if s.Attributes["url.path"] != "/v1/orders/8a3f" {
t.Errorf("url.path = %v, want the path that arrived", s.Attributes["url.path"])
}
}
// TestBoundaryExportsALogOverASocket proves the same for the log leg, and that
// the record carries the ids that join it to its span.
func TestBoundaryExportsALogOverASocket(t *testing.T) {
addr := socketAt(t, "logs.sock")
// TestBoundaryExportsALog proves the same for the log leg, and that the record
// carries the ids that join it to its span.
func TestBoundaryExportsALog(t *testing.T) {
addr := free(t)
got := collect(t, addr, 3) // 3 = logs
app := zip.New(zip.Config{
@@ -469,7 +473,7 @@ func TestBoundaryExportsALogOverASocket(t *testing.T) {
})
app.Get("/x", func(c *zip.Ctx) error { return c.String(500, "boom") })
h, err := zip.Serve(app, socketAt(t, "app2.sock"))
h, err := zip.Serve(app, "http://"+free(t))
if err != nil {
t.Fatalf("serve: %v", err)
}
@@ -513,33 +517,24 @@ func TestBoundaryExportsALogOverASocket(t *testing.T) {
}
}
// TestZeroConfigurationFindsTheSocket is the headline claim: an app that
// configures NOTHING — no address, no environment variable, no Telemetry field —
// exports to the collector because the collector's socket is there.
//
// Presence is the configuration, so the test creates the situation a pod
// creates: o11y's ingest listening on the runtime directory under its
// well-known name.
func TestZeroConfigurationFindsTheSocket(t *testing.T) {
dir, err := os.MkdirTemp("", "rt")
if err != nil {
t.Fatalf("tempdir: %v", err)
}
t.Cleanup(func() { _ = os.RemoveAll(dir) })
t.Setenv("ZIP_RUNTIME_DIR", dir)
// The collector, at the name the convention says — nothing tells the app.
got := collect(t, filepath.Join(dir, "o11y-spans.sock"), 1)
// TestTheEnvironmentStatesTheAddress is how a deployment turns export on: the
// manifest sets a variable, the program sets no field, and the span arrives.
// This is the path cloud's own manifest takes, so the name is asserted here as
// the deployment spells it.
func TestTheEnvironmentStatesTheAddress(t *testing.T) {
addr := free(t)
got := collect(t, addr, 1)
t.Setenv("O11Y_SPANS_ADDR", addr)
app := zip.New(zip.Config{
AppName: "zeroconf",
AppName: "fromenv",
Logger: luxlog.NewWriter(&sink{}),
DisableStartupMessage: true,
// No Telemetry field at all.
// No Telemetry field at all: the environment is the whole configuration.
})
app.Get("/x", func(c *zip.Ctx) error { return c.String(200, "ok") })
h, err := zip.Serve(app, socketAt(t, "zc.sock"))
h, err := zip.Serve(app, "http://"+free(t))
if err != nil {
t.Fatalf("serve: %v", err)
}
@@ -567,16 +562,14 @@ func TestZeroConfigurationFindsTheSocket(t *testing.T) {
}
}
// TestNoCollectorIsFree is the other half of zero configuration: a program on a
// laptop, with no collector anywhere, must serve normally, keep logging, and
// never fail because telemetry has nowhere to go.
func TestNoCollectorIsFree(t *testing.T) {
dir, err := os.MkdirTemp("", "rt")
if err != nil {
t.Fatalf("tempdir: %v", err)
// TestNoAddressIsSilent is the other half of the switch: a program with nothing
// stated serves normally, keeps logging, and says nothing about telemetry —
// because it never tried. A line about export in a program that exports nothing
// would be the symptom of a connection nobody asked for.
func TestNoAddressIsSilent(t *testing.T) {
for _, name := range []string{"O11Y_SPANS_ADDR", "O11Y_LOGS_ADDR", "O11Y_METRICS_ADDR"} {
t.Setenv(name, "")
}
t.Cleanup(func() { _ = os.RemoveAll(dir) })
t.Setenv("ZIP_RUNTIME_DIR", dir) // empty: no socket, and loopback has no ear
s := &sink{}
app := zip.New(zip.Config{
@@ -586,7 +579,7 @@ func TestNoCollectorIsFree(t *testing.T) {
})
app.Get("/x", func(c *zip.Ctx) error { return c.String(200, "ok") })
h, err := zip.Serve(app, socketAt(t, "lonely.sock"))
h, err := zip.Serve(app, "http://"+free(t))
if err != nil {
t.Fatalf("serve: %v", err)
}
@@ -606,12 +599,51 @@ func TestNoCollectorIsFree(t *testing.T) {
t.Fatalf("shutdown: %v", err)
}
// It still logs every request.
if s.find("request") == nil {
t.Error("an app with no collector stopped logging")
}
// And it says the export is not working AT MOST once per signal, not once
// per flush — the difference between a useful line and noise.
s.mu.Lock()
defer s.mu.Unlock()
for _, l := range s.lines {
if msg, _ := l["message"].(string); strings.HasPrefix(msg, "telemetry export") {
t.Errorf("an app with no address said %q; it should have dialled nothing", msg)
}
}
}
// TestTheCollectorBeingDownCostsNothing is the fire-and-forget claim at the app
// level: an address that nothing answers must not fail a request, and must be
// reported once rather than once per flush.
func TestTheCollectorBeingDownCostsNothing(t *testing.T) {
s := &sink{}
app := zip.New(zip.Config{
AppName: "shouting",
Logger: luxlog.NewWriter(s),
DisableStartupMessage: true,
Telemetry: zip.Telemetry{Spans: free(t)},
})
app.Get("/x", func(c *zip.Ctx) error { return c.String(200, "ok") })
h, err := zip.Serve(app, "http://"+free(t))
if err != nil {
t.Fatalf("serve: %v", err)
}
defer func() { _ = h.Close() }()
for range 25 {
resp, err := app.Test(httptest.NewRequest(http.MethodGet, "/x", nil))
if err != nil {
t.Fatalf("request failed with a dead collector: %v", err)
}
if resp.StatusCode != 200 {
t.Fatalf("status %d with a dead collector, want 200", resp.StatusCode)
}
resp.Body.Close()
}
if err := app.Shutdown(); err != nil {
t.Fatalf("shutdown: %v", err)
}
s.mu.Lock()
defer s.mu.Unlock()
stopped := 0
@@ -620,7 +652,7 @@ func TestNoCollectorIsFree(t *testing.T) {
stopped++
}
}
if stopped > 2 {
if stopped != 1 {
t.Errorf("said %d times that export is down; edge-triggered means once per signal", stopped)
}
}