feat: durable WAL + native logs + native traces (full ZAP observability stack)

Completes the prometheus-free, Grafana-free observability backend: - wal.go: append-only write-ahead log; every write is logged + replayed on startup, so all three signals survive restart (EnableDurability under DataDir). - store.go: metrics store now WAL-durable. - logstore.go: native log store (Loki-free) — label + time + substring query. - tracestore.go: native span store (Tempo-free) — trace-id waterfall + time range. - mount.go: one subsystem serves /v1/{metrics,logs,traces}/* over zip; durability wired from deps.DataDir. Zero prometheus, zero /api/. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 08:18:13 -07:00
parent c1166837ab
commit bf9993bdd8
5 changed files with 431 additions and 7 deletions
@@ -0,0 +1,103 @@
+package metrics
+
+import (
+	"encoding/json"
+	"strings"
+	"sync"
+)
+
+// LogRecord is one structured log line — the native, prometheus-free, Loki-free
+// log signal. Bodies are stored verbatim; labels are the indexed dimensions.
+type LogRecord struct {
+	TsNs   int64             `json:"t"`
+	Level  string            `json:"level,omitempty"`
+	Body   string            `json:"body"`
+	Labels map[string]string `json:"labels,omitempty"`
+}
+
+// LogStore is the native log store: an append-only bounded ring with label +
+// time-range + case-insensitive substring query. Durable via the shared WAL.
+type LogStore struct {
+	mu      sync.RWMutex
+	records []LogRecord
+	max     int
+	wal     *WAL
+}
+
+// NewLogStore returns an empty store retaining up to 1Mi records in memory.
+func NewLogStore() *LogStore { return &LogStore{max: 1 << 20} }
+
+// EnableDurability opens and replays a WAL at path (e.g. <DataDir>/logs/logs.wal).
+func (s *LogStore) EnableDurability(path string) error {
+	w, err := OpenWAL(path)
+	if err != nil {
+		return err
+	}
+	if err := w.Replay(func(rec []byte) {
+		var lr LogRecord
+		if json.Unmarshal(rec, &lr) == nil {
+			s.appendMem(lr)
+		}
+	}); err != nil {
+		return err
+	}
+	s.mu.Lock()
+	s.wal = w
+	s.mu.Unlock()
+	return nil
+}
+
+// Append stores one log record (and durably logs it when durability is on).
+func (s *LogStore) Append(lr LogRecord) {
+	s.appendMem(lr)
+	if s.wal != nil {
+		if rec, err := json.Marshal(lr); err == nil {
+			_ = s.wal.Append(rec)
+		}
+	}
+}
+
+func (s *LogStore) appendMem(lr LogRecord) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.records = append(s.records, lr)
+	if len(s.records) > s.max {
+		s.records = s.records[len(s.records)-s.max:]
+	}
+}
+
+// Query returns up to limit records (newest first) matching labels, the
+// [startNs,endNs] range, and an optional case-insensitive substring of Body.
+func (s *LogStore) Query(matchers map[string]string, startNs, endNs int64, contains string, limit int) []LogRecord {
+	if limit <= 0 {
+		limit = 100
+	}
+	contains = strings.ToLower(contains)
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	out := make([]LogRecord, 0, limit)
+	for i := len(s.records) - 1; i >= 0 && len(out) < limit; i-- {
+		r := s.records[i]
+		if startNs != 0 && r.TsNs < startNs {
+			continue
+		}
+		if endNs != 0 && r.TsNs > endNs {
+			continue
+		}
+		if !labelsSupersetOf(r.Labels, matchers) {
+			continue
+		}
+		if contains != "" && !strings.Contains(strings.ToLower(r.Body), contains) {
+			continue
+		}
+		out = append(out, r)
+	}
+	return out
+}
+
+// Count reports the number of records held.
+func (s *LogStore) Count() int {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return len(s.records)
+}
@@ -11,6 +11,7 @@ package metrics
 import (
 	"fmt"
 	"net/http"
+	"path/filepath"
 	"strconv"
 	"strings"

@@ -19,12 +20,17 @@ import (
 	metric "github.com/luxfi/metric"
 )

-// Version is surfaced on /v1/metrics/health.
+// Version is surfaced on the /health routes.
 const Version = "0.1.0"

-// store is the process-global in-memory store. A durable per-tenant backend can
-// replace it behind the same Store API without touching the handlers below.
-var store = NewStore()
+// The three native observability signals share the WAL-backed store pattern and
+// mount under one subsystem. A durable per-tenant backend can replace each store
+// behind its existing API without touching the handlers below.
+var (
+	store      = NewStore()      // metrics
+	logStore   = NewLogStore()   // logs   (Loki-free)
+	traceStore = NewTraceStore() // traces (Tempo-free)
+)

 // init registers the subsystem. cloud.MountFunc takes app as `any` (to avoid an
 // import cycle in pkg/cloud), so we assert it to *zip.App and call the typed Mount.
@@ -40,8 +46,24 @@ func init() {

 // Mount registers the native metrics routes on the shared cloud App per HIP-0106.
 func Mount(app *zip.App, deps cloud.Deps) error {
-	log := deps.Logger.New("subsystem", "metrics")
+	log := deps.Logger.New("subsystem", "o11y")

+	// Durability — replay/append WALs under DataDir so all three signals survive
+	// restart. Falls back to in-memory if DataDir is unset or unwritable.
+	if deps.DataDir != "" {
+		dir := filepath.Join(deps.DataDir, "o11y")
+		for name, enable := range map[string]func(string) error{
+			"metrics.wal": store.EnableDurability,
+			"logs.wal":    logStore.EnableDurability,
+			"traces.wal":  traceStore.EnableDurability,
+		} {
+			if err := enable(filepath.Join(dir, name)); err != nil {
+				log.Warn("durability disabled", "wal", name, "err", err)
+			}
+		}
+	}
+
+	// --- Metrics ---
 	// Liveness/readiness — always answers, no auth, no external dep.
 	app.Get("/v1/metrics/health", func(c *zip.Ctx) error {
 		return c.JSON(http.StatusOK, map[string]any{
@@ -88,7 +110,60 @@ func Mount(app *zip.App, deps cloud.Deps) error {
 		return c.JSON(http.StatusOK, map[string]any{"count": len(res), "series": res})
 	})

-	log.Info("mounted native ZAP metrics store", "version", Version, "routes", "/v1/metrics/{health,batch,write,query}")
+	// --- Logs (native, Loki-free) ---
+	app.Get("/v1/logs/health", func(c *zip.Ctx) error {
+		return c.JSON(http.StatusOK, map[string]any{"status": "ok", "service": "logs", "version": Version, "records": logStore.Count()})
+	})
+	app.Post("/v1/logs/write", func(c *zip.Ctx) error {
+		var req struct {
+			Records []LogRecord `json:"records"`
+		}
+		if err := c.Bind(&req); err != nil {
+			return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid logs write"})
+		}
+		for _, r := range req.Records {
+			logStore.Append(r)
+		}
+		return c.JSON(http.StatusOK, map[string]any{"written": len(req.Records)})
+	})
+	app.Get("/v1/logs/query", func(c *zip.Ctx) error {
+		start, _ := strconv.ParseInt(c.Query("start"), 10, 64)
+		end, _ := strconv.ParseInt(c.Query("end"), 10, 64)
+		limit, _ := strconv.Atoi(c.Query("limit"))
+		res := logStore.Query(parseMatchers(c.Query("match")), start, end, c.Query("contains"), limit)
+		return c.JSON(http.StatusOK, map[string]any{"count": len(res), "records": res})
+	})
+
+	// --- Traces (native, Tempo-free) ---
+	app.Get("/v1/traces/health", func(c *zip.Ctx) error {
+		return c.JSON(http.StatusOK, map[string]any{"status": "ok", "service": "traces", "version": Version, "spans": traceStore.Count()})
+	})
+	app.Post("/v1/traces/write", func(c *zip.Ctx) error {
+		var req struct {
+			Spans []Span `json:"spans"`
+		}
+		if err := c.Bind(&req); err != nil {
+			return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid traces write"})
+		}
+		for _, sp := range req.Spans {
+			traceStore.Append(sp)
+		}
+		return c.JSON(http.StatusOK, map[string]any{"written": len(req.Spans)})
+	})
+	app.Get("/v1/traces/trace", func(c *zip.Ctx) error {
+		return c.JSON(http.StatusOK, map[string]any{"spans": traceStore.ByTrace(c.Query("id"))})
+	})
+	app.Get("/v1/traces/query", func(c *zip.Ctx) error {
+		start, _ := strconv.ParseInt(c.Query("start"), 10, 64)
+		end, _ := strconv.ParseInt(c.Query("end"), 10, 64)
+		limit, _ := strconv.Atoi(c.Query("limit"))
+		res := traceStore.Recent(start, end, limit)
+		return c.JSON(http.StatusOK, map[string]any{"count": len(res), "spans": res})
+	})
+
+	log.Info("mounted native ZAP observability store (metrics+logs+traces)",
+		"version", Version, "durable", deps.DataDir != "",
+		"routes", "/v1/{metrics,logs,traces}/*")
 	return nil
 }

@@ -10,6 +10,7 @@
 package metrics

 import (
+	"encoding/json"
 	"sort"
 	"strings"
 	"sync"
@@ -35,6 +36,7 @@ type Store struct {
 	mu           sync.RWMutex
 	series       map[string]*Series
 	maxPerSeries int
+	wal          *WAL // nil = in-memory only
 }

 // NewStore returns an empty store with a default per-series retention of 64Ki
@@ -65,9 +67,50 @@ func seriesKey(name string, labels map[string]string) string {
 	return b.String()
 }

+// EnableDurability opens a write-ahead log at path and replays it into the
+// store so samples survive restart. Replayed samples are not re-logged. Pass a
+// per-deployment path (e.g. <DataDir>/metrics/metrics.wal).
+func (s *Store) EnableDurability(path string) error {
+	w, err := OpenWAL(path)
+	if err != nil {
+		return err
+	}
+	if err := w.Replay(func(rec []byte) {
+		var ws walSample
+		if json.Unmarshal(rec, &ws) == nil {
+			s.appendMem(ws.Name, ws.Labels, ws.Sample)
+		}
+	}); err != nil {
+		return err
+	}
+	s.mu.Lock()
+	s.wal = w
+	s.mu.Unlock()
+	return nil
+}
+
+// walSample is the on-disk record for one appended sample.
+type walSample struct {
+	Name   string            `json:"n"`
+	Labels map[string]string `json:"l,omitempty"`
+	Sample Sample            `json:"s"`
+}
+
 // Append adds one sample to the series identified by (name, labels), creating
-// the series on first write and evicting the oldest sample past retention.
+// the series on first write and evicting the oldest sample past retention. When
+// durability is enabled the sample is also written to the WAL.
 func (s *Store) Append(name string, labels map[string]string, smp Sample) {
+	s.appendMem(name, labels, smp)
+	if s.wal != nil {
+		if rec, err := json.Marshal(walSample{Name: name, Labels: labels, Sample: smp}); err == nil {
+			_ = s.wal.Append(rec)
+		}
+	}
+}
+
+// appendMem applies a sample to the in-memory index only (used by Append and by
+// WAL replay, which must not re-log).
+func (s *Store) appendMem(name string, labels map[string]string, smp Sample) {
 	k := seriesKey(name, labels)
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -0,0 +1,118 @@
+package metrics
+
+import (
+	"encoding/json"
+	"sync"
+)
+
+// Span is one unit of a distributed trace — the native, prometheus-free,
+// Tempo-free trace signal. Times are nanoseconds since the Unix epoch.
+type Span struct {
+	TraceID string            `json:"traceId"`
+	SpanID  string            `json:"spanId"`
+	Parent  string            `json:"parentId,omitempty"`
+	Name    string            `json:"name"`
+	StartNs int64             `json:"startNs"`
+	EndNs   int64             `json:"endNs"`
+	Attrs   map[string]string `json:"attrs,omitempty"`
+}
+
+// TraceStore is the native span store: an append-only bounded ring with a
+// trace-id index for waterfall lookup and time-range listing. Durable via WAL.
+type TraceStore struct {
+	mu      sync.RWMutex
+	spans   []Span
+	byTrace map[string][]int
+	max     int
+	wal     *WAL
+}
+
+// NewTraceStore returns an empty store retaining up to 1Mi spans in memory.
+func NewTraceStore() *TraceStore {
+	return &TraceStore{byTrace: make(map[string][]int), max: 1 << 20}
+}
+
+// EnableDurability opens and replays a WAL at path (e.g. <DataDir>/traces/traces.wal).
+func (s *TraceStore) EnableDurability(path string) error {
+	w, err := OpenWAL(path)
+	if err != nil {
+		return err
+	}
+	if err := w.Replay(func(rec []byte) {
+		var sp Span
+		if json.Unmarshal(rec, &sp) == nil {
+			s.appendMem(sp)
+		}
+	}); err != nil {
+		return err
+	}
+	s.mu.Lock()
+	s.wal = w
+	s.mu.Unlock()
+	return nil
+}
+
+// Append stores one span (and durably logs it when durability is on).
+func (s *TraceStore) Append(sp Span) {
+	s.appendMem(sp)
+	if s.wal != nil {
+		if rec, err := json.Marshal(sp); err == nil {
+			_ = s.wal.Append(rec)
+		}
+	}
+}
+
+func (s *TraceStore) appendMem(sp Span) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.spans = append(s.spans, sp)
+	s.byTrace[sp.TraceID] = append(s.byTrace[sp.TraceID], len(s.spans)-1)
+	if len(s.spans) > s.max {
+		// Drop the oldest half and rebuild the trace-id index (infrequent).
+		s.spans = append([]Span(nil), s.spans[len(s.spans)-s.max:]...)
+		s.byTrace = make(map[string][]int, len(s.spans))
+		for i, sp := range s.spans {
+			s.byTrace[sp.TraceID] = append(s.byTrace[sp.TraceID], i)
+		}
+	}
+}
+
+// ByTrace returns every span belonging to a trace id (the waterfall).
+func (s *TraceStore) ByTrace(traceID string) []Span {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	idxs := s.byTrace[traceID]
+	out := make([]Span, 0, len(idxs))
+	for _, i := range idxs {
+		out = append(out, s.spans[i])
+	}
+	return out
+}
+
+// Recent returns up to limit spans (newest first) starting within [startNs,endNs].
+func (s *TraceStore) Recent(startNs, endNs int64, limit int) []Span {
+	if limit <= 0 {
+		limit = 100
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	out := make([]Span, 0, limit)
+	for i := len(s.spans) - 1; i >= 0 && len(out) < limit; i-- {
+		sp := s.spans[i]
+		if startNs != 0 && sp.StartNs < startNs {
+			continue
+		}
+		if endNs != 0 && sp.StartNs > endNs {
+			continue
+		}
+		out = append(out, sp)
+	}
+	return out
+}
+
+// Count reports the number of spans held.
+func (s *TraceStore) Count() int {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return len(s.spans)
+}
@@ -0,0 +1,85 @@
+package metrics
+
+import (
+	"bufio"
+	"encoding/binary"
+	"io"
+	"os"
+	"path/filepath"
+	"sync"
+)
+
+// WAL is a simple append-only write-ahead log of length-prefixed records. It
+// gives the in-memory stores durability: every write is appended here and the
+// log is replayed on startup. This is intentionally a single flat segment — a
+// production deployment adds rotation/compaction behind the same Append/Replay
+// API without touching callers.
+type WAL struct {
+	mu sync.Mutex
+	f  *os.File
+	w  *bufio.Writer
+}
+
+// OpenWAL opens (creating parent dirs and the file as needed) an append log.
+func OpenWAL(path string) (*WAL, error) {
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		return nil, err
+	}
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644)
+	if err != nil {
+		return nil, err
+	}
+	return &WAL{f: f, w: bufio.NewWriter(f)}, nil
+}
+
+// Replay invokes fn for every record from the start, then positions the file at
+// the end so subsequent Appends extend the log. Call once on startup.
+func (w *WAL) Replay(fn func([]byte)) error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if _, err := w.f.Seek(0, io.SeekStart); err != nil {
+		return err
+	}
+	r := bufio.NewReader(w.f)
+	var hdr [4]byte
+	for {
+		if _, err := io.ReadFull(r, hdr[:]); err != nil {
+			if err == io.EOF {
+				break
+			}
+			return err
+		}
+		rec := make([]byte, binary.LittleEndian.Uint32(hdr[:]))
+		if _, err := io.ReadFull(r, rec); err != nil {
+			return err
+		}
+		fn(rec)
+	}
+	_, err := w.f.Seek(0, io.SeekEnd)
+	return err
+}
+
+// Append writes one length-prefixed record and flushes it to disk.
+func (w *WAL) Append(rec []byte) error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	var hdr [4]byte
+	binary.LittleEndian.PutUint32(hdr[:], uint32(len(rec)))
+	if _, err := w.w.Write(hdr[:]); err != nil {
+		return err
+	}
+	if _, err := w.w.Write(rec); err != nil {
+		return err
+	}
+	return w.w.Flush()
+}
+
+// Close flushes and closes the log.
+func (w *WAL) Close() error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.w != nil {
+		_ = w.w.Flush()
+	}
+	return w.f.Close()
+}