ml

internal/collector/collector_test.go

@@ -392,6 +392,31 @@ func TestHandleIssuesAndByID(t *testing.T) {
 	}
 }
 
+func TestHandleMLDetectorTest(t *testing.T) {
+	coll := New(nil, 10, nil)
+	req := httptest.NewRequest(http.MethodPost, "/api/ml/test-spike", nil)
+	rec := httptest.NewRecorder()
+	coll.HandleMLDetectorTest(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("expected detector test success, got %d: %s", rec.Code, rec.Body.String())
+	}
+	var result struct {
+		Passed bool `json:"passed"`
+		Row    struct {
+			LabelKnown        bool    `json:"labelKnown"`
+			Label             bool    `json:"label"`
+			BaselineErrorRate float64 `json:"baselineErrorRate"`
+			FutureErrorCount  int     `json:"futureErrorCount"`
+		} `json:"row"`
+	}
+	if err := json.NewDecoder(rec.Body).Decode(&result); err != nil {
+		t.Fatal(err)
+	}
+	if !result.Passed || !result.Row.LabelKnown || !result.Row.Label || result.Row.BaselineErrorRate != .1 || result.Row.FutureErrorCount != 5 {
+		t.Fatalf("unexpected detector verification response: %+v", result)
+	}
+}
+
 func TestHandleIssuesFiltersAndPaginates(t *testing.T) {
 	coll := New(noopBroadcaster{}, 10, nil)
 	coll.issues.Record("span_error", "nginx", "GET /.env", issues.Sample{Timestamp: 300, SourceIP: "203.0.113.10", HTTPRoute: "/.env"})

internal/collector/handlers.go

@@ -10,10 +10,12 @@ import (
 	"io"
 	"log"
 	"net/http"
+	"os"
 	"strconv"
 	"strings"
 	"time"
 
+	"git.tornberg.me/go-otel/internal/features"
 	"git.tornberg.me/go-otel/internal/issues"
 	"git.tornberg.me/go-otel/internal/storage"
 	"git.tornberg.me/go-otel/internal/websocket"
@@ -669,6 +671,41 @@ func (c *Collector) HandleHealth(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// HandleMLDetectorTest runs a deterministic, in-memory spike through the
+// production feature and label builder. It stores no telemetry and exists so
+// an operator can verify the detector wiring from the authenticated UI.
+func (c *Collector) HandleMLDetectorTest(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	const service = "ml-verification"
+	base := time.Date(2026, 1, 8, 12, 0, 0, 0, time.UTC)
+	events := make([]features.Event, 0, 100)
+	for day := 1; day <= 7; day++ {
+		for i := 0; i < 10; i++ {
+			events = append(events, features.Event{Timestamp: base.AddDate(0, 0, -day).Add(time.Duration(i) * time.Second), Service: service, Kind: "span", IsError: i == 0, DurationMS: float64(i + 1)})
+		}
+	}
+	for i := 0; i < 10; i++ {
+		events = append(events, features.Event{Timestamp: base.Add(time.Duration(i) * time.Second), Service: service, Release: "verification", Kind: "span", DurationMS: float64(i + 1)})
+		events = append(events, features.Event{Timestamp: base.Add(5 * time.Minute).Add(time.Duration(i) * time.Second), Service: service, Release: "verification", Kind: "log", IsError: i < 5, IssueFingerprint: "verification-spike"})
+	}
+	rows := features.Build(events, features.Config{ObservedUntil: base.Add(20 * time.Minute)})
+	for _, row := range rows {
+		if row.WindowStart.Equal(base) && row.Service == service && row.Release == "verification" {
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode(map[string]interface{}{
+				"passed":   row.LabelKnown && row.Label,
+				"scenario": "five errors in ten future events against a seven-day 10% baseline",
+				"row":      row,
+			})
+			return
+		}
+	}
+	jsonError(w, http.StatusInternalServerError, "detector test did not generate its verification row")
+}
+
 // statsEnabled reports whether an aggregate-capable store is configured.
 func (c *Collector) statsEnabled() bool { return c.statsQuerier != nil }
 
@@ -767,3 +804,107 @@ func (c *Collector) HandleStatsServices(w http.ResponseWriter, r *http.Request)
 	w.Header().Set("Content-Type", "application/json")
 	_ = json.NewEncoder(w).Encode(map[string]interface{}{"services": services})
 }
+
+// needleAddr returns the configured or default address for the Needle 3 sidecar.
+func needleAddr() string {
+	addr := os.Getenv("OTEL_NEEDLE_ADDR")
+	if addr == "" {
+		addr = "http://127.0.0.1:5005"
+	}
+	return strings.TrimRight(addr, "/")
+}
+
+// HandleNeedleStatus checks if the Needle 3 sidecar is reachable and reports its health.
+func (c *Collector) HandleNeedleStatus(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	target := needleAddr() + "/health"
+	client := &http.Client{Timeout: 2 * time.Second}
+	resp, err := client.Get(target)
+	w.Header().Set("Content-Type", "application/json")
+	if err != nil {
+		w.WriteHeader(http.StatusServiceUnavailable)
+		_ = json.NewEncoder(w).Encode(map[string]interface{}{
+			"available": false,
+			"error":     fmt.Sprintf("Needle sidecar unreachable at %s: %v", target, err),
+		})
+		return
+	}
+	defer resp.Body.Close()
+	w.WriteHeader(resp.StatusCode)
+	_, _ = io.Copy(w, resp.Body)
+}
+
+// HandleNeedleClassify proxies telemetry triage requests to the Needle 3 sidecar.
+func (c *Collector) HandleNeedleClassify(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, "failed to read request body", http.StatusBadRequest)
+		return
+	}
+
+	// If the payload specifies an issue_id and no text, look up the issue
+	var reqMap map[string]interface{}
+	if err := json.Unmarshal(body, &reqMap); err == nil {
+		if issueID, ok := reqMap["issue_id"].(string); ok && issueID != "" {
+			if issue, ok := c.GetIssue(issueID); ok {
+				var promptParts []string
+				promptParts = append(promptParts, fmt.Sprintf("Service: %s", issue.Service))
+				promptParts = append(promptParts, fmt.Sprintf("Title: %s", issue.Title))
+				promptParts = append(promptParts, fmt.Sprintf("Kind: %s", issue.Kind))
+				if len(issue.Samples) > 0 {
+					latest := issue.Samples[len(issue.Samples)-1]
+					if latest.Message != "" {
+						promptParts = append(promptParts, fmt.Sprintf("Message: %s", latest.Message))
+					}
+					if latest.URL != "" {
+						promptParts = append(promptParts, fmt.Sprintf("URL: %s", latest.URL))
+					}
+					if latest.HTTPStatus > 0 {
+						promptParts = append(promptParts, fmt.Sprintf("Status: %d", latest.HTTPStatus))
+					}
+					if latest.HTTPRoute != "" {
+						promptParts = append(promptParts, fmt.Sprintf("Route: %s", latest.HTTPRoute))
+					}
+				}
+				reqMap["text"] = strings.Join(promptParts, " | ")
+				reqMap["service"] = issue.Service
+				reqMap["title"] = issue.Title
+				body, _ = json.Marshal(reqMap)
+			}
+		}
+	}
+
+	target := needleAddr() + "/classify"
+	client := &http.Client{Timeout: 60 * time.Second}
+	proxyReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost, target, strings.NewReader(string(body)))
+	if err != nil {
+		http.Error(w, "failed to create proxy request", http.StatusInternalServerError)
+		return
+	}
+	proxyReq.Header.Set("Content-Type", "application/json")
+
+	resp, err := client.Do(proxyReq)
+	if err != nil {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusServiceUnavailable)
+		_ = json.NewEncoder(w).Encode(map[string]interface{}{
+			"success": false,
+			"error":   fmt.Sprintf("Needle sidecar unreachable at %s: %v", target, err),
+		})
+		return
+	}
+	defer resp.Body.Close()
+
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(resp.StatusCode)
+	_, _ = io.Copy(w, resp.Body)
+}

internal/collector/issue_extract.go

@@ -87,7 +87,7 @@ func spanSample(span ptrace.Span, res pcommon.Resource, timestamp int64, title s
 		SpanID:      idString(span.SpanID()),
 		Message:     title,
 		HTTPMethod:  firstAttr(span.Attributes(), "http.request.method", "http.method"),
-		HTTPRoute:   firstAttr(span.Attributes(), "http.route", "http.target", "url.path"),
+		HTTPRoute:   httpRoute(span.Attributes()),
 		UserAgent:   firstAttr(span.Attributes(), "user_agent.original", "http.user_agent"),
 		SourceIP:    sourceIP(span.Attributes(), res.Attributes()),
 		Release:     firstAttr(span.Attributes(), "service.release"),
@@ -109,7 +109,7 @@ func logSample(record plog.LogRecord, res pcommon.Resource, timestamp int64, tit
 		Message:     title,
 		Stacktrace:  attrString(record.Attributes(), "exception.stacktrace"),
 		HTTPMethod:  firstAttr(record.Attributes(), "http.request.method", "http.method"),
-		HTTPRoute:   firstAttr(record.Attributes(), "http.route", "http.target", "url.path"),
+		HTTPRoute:   httpRoute(record.Attributes()),
 		UserAgent:   firstAttr(record.Attributes(), "user_agent.original", "http.user_agent", "browser.userAgent"),
 		SourceIP:    sourceIP(record.Attributes(), res.Attributes()),
 		Release:     firstAttr(record.Attributes(), "service.release"),
@@ -179,6 +179,18 @@ func firstAttr(attrs pcommon.Map, keys ...string) string {
 	return ""
 }
 
+// httpRoute prefers the concrete URL path when instrumentation reports "/"
+// as a generic router mount. This keeps API endpoints distinct in issue
+// fingerprints and avoids collapsing unrelated requests into one root route.
+func httpRoute(attrs pcommon.Map) string {
+	route := firstAttr(attrs, "http.route")
+	path := firstAttr(attrs, "url.path", "http.target")
+	if path != "" && (route == "" || route == "/" && path != "/") {
+		return path
+	}
+	return route
+}
+
 // getHTTPStatusCode extracts an HTTP status code from attributes, supporting
 // both new (http.response.status_code) and old (http.status_code) semconv keys,
 // stored as integer or string.

internal/features/features.go

@@ -0,0 +1,328 @@
+// Package features derives fixed-width, non-PII service window features from
+// normalized OTLP events. It deliberately has no ClickHouse dependency so the
+// exact same feature logic can be used for exports, offline evaluation and
+// eventual online scoring.
+package features
+
+import (
+	"sort"
+	"time"
+)
+
+// Event is the subset of a normalized event used by the detector. Text fields
+// are already normalized/redacted by the normalization package.
+type Event struct {
+	Timestamp        time.Time
+	Service          string
+	Release          string
+	Kind             string
+	Source           string
+	IsError          bool
+	StatusClass      string
+	DurationMS       float64
+	TraceID          string
+	PatternSignature string
+	IssueFingerprint string
+	IsRetry          bool
+	IsTimeout        bool
+	IsCancellation   bool
+	Message          string
+	SpanName         string
+}
+
+// Config controls feature and label windows. A zero Config receives the
+// documented five-minute observation / fifteen-minute prediction defaults.
+type Config struct {
+	Window time.Duration
+	Step   time.Duration
+	Future time.Duration
+	// ObservedUntil is the exclusive source-export watermark. Supplying it
+	// lets a quiet tail interval be labelled safely; when absent Build falls
+	// back to the latest event timestamp and is deliberately conservative.
+	ObservedUntil      time.Time
+	BaselineDays       int
+	MinBaselineSamples int
+	SpikeMultiple      float64
+	MinFutureErrors    int
+}
+
+// DefaultConfig is intentionally conservative: labels are withheld until all
+// seven same-time-of-day baseline observations are present.
+func DefaultConfig() Config {
+	return Config{
+		Window:             5 * time.Minute,
+		Step:               5 * time.Minute,
+		Future:             15 * time.Minute,
+		BaselineDays:       7,
+		MinBaselineSamples: 7,
+		SpikeMultiple:      3,
+		MinFutureErrors:    5,
+	}
+}
+
+// Row is a numeric, fixed-width model input. Label is only meaningful when
+// LabelKnown is true; this prevents short exports from silently producing
+// negative training examples without a seven-day history.
+type Row struct {
+	WindowStart time.Time `json:"windowStart"`
+	WindowEnd   time.Time `json:"windowEnd"`
+	Service     string    `json:"service"`
+	Release     string    `json:"release,omitempty"`
+
+	TotalEvents int     `json:"totalEvents"`
+	ErrorCount  int     `json:"errorCount"`
+	ErrorRate   float64 `json:"errorRate"`
+	P50Duration float64 `json:"p50DurationMs"`
+	P95Duration float64 `json:"p95DurationMs"`
+
+	SpanCount   int `json:"spanCount"`
+	LogCount    int `json:"logCount"`
+	MetricCount int `json:"metricCount"`
+	Status2xx   int `json:"status2xx"`
+	Status4xx   int `json:"status4xx"`
+	Status5xx   int `json:"status5xx"`
+
+	UniqueTraceCount  int `json:"uniqueTraceCount"`
+	UniqueIssueCount  int `json:"uniqueIssueCount"`
+	NewIssueCount     int `json:"newIssueCount"`
+	FrontendErrors    int `json:"frontendErrors"`
+	BackendErrors     int `json:"backendErrors"`
+	RetryCount        int `json:"retryCount"`
+	TimeoutCount      int `json:"timeoutCount"`
+	CancellationCount int `json:"cancellationCount"`
+
+	BaselineErrorRate float64 `json:"baselineErrorRate"`
+	BaselineSamples   int     `json:"baselineSamples"`
+	FutureEventCount  int     `json:"futureEventCount"`
+	FutureErrorCount  int     `json:"futureErrorCount"`
+	FutureErrorRate   float64 `json:"futureErrorRate"`
+	LabelKnown        bool    `json:"labelKnown"`
+	Label             bool    `json:"label"`
+}
+
+// Build aggregates every service/release over rolling observation windows.
+// The future target and baseline are service-wide as specified by the model
+// target; release is context for the input rather than a way to split alerts.
+func Build(input []Event, cfg Config) []Row {
+	cfg = normalizedConfig(cfg)
+	if len(input) == 0 {
+		return nil
+	}
+	events := append([]Event(nil), input...)
+	for i := range events {
+		events[i].Timestamp = events[i].Timestamp.UTC()
+		if events[i].Service == "" {
+			events[i].Service = "unknown"
+		}
+	}
+	sort.Slice(events, func(i, j int) bool { return events[i].Timestamp.Before(events[j].Timestamp) })
+	firstIssueSeen := make(map[string]time.Time)
+	for _, e := range events {
+		if !e.IsError || issueKey(e) == "" {
+			continue
+		}
+		key := e.Service + "\x00" + issueKey(e)
+		if seen, ok := firstIssueSeen[key]; !ok || e.Timestamp.Before(seen) {
+			firstIssueSeen[key] = e.Timestamp
+		}
+	}
+
+	first := events[0].Timestamp.Truncate(cfg.Step)
+	last := events[len(events)-1].Timestamp
+	observedUntil := cfg.ObservedUntil.UTC()
+	if observedUntil.IsZero() {
+		observedUntil = last
+	}
+	rows := make([]Row, 0)
+	for start := first; !start.After(last); start = start.Add(cfg.Step) {
+		end := start.Add(cfg.Window)
+		groups := map[string]*accumulator{}
+		for _, e := range events {
+			if e.Timestamp.Before(start) || !e.Timestamp.Before(end) {
+				continue
+			}
+			key := e.Service + "\x00" + e.Release
+			a := groups[key]
+			if a == nil {
+				a = newAccumulator(start, end, e.Service, e.Release)
+				groups[key] = a
+			}
+			a.add(e)
+		}
+		for _, a := range groups {
+			row := a.row()
+			for issue := range a.issues {
+				if seen := firstIssueSeen[row.Service+"\x00"+issue]; !seen.Before(start) && seen.Before(end) {
+					row.NewIssueCount++
+				}
+			}
+			futureTotal, futureErrors := count(events, start.Add(cfg.Window), start.Add(cfg.Window).Add(cfg.Future), row.Service)
+			baselineRate, samples := baseline(events, start, cfg.Window, cfg.BaselineDays, row.Service)
+			row.FutureEventCount, row.FutureErrorCount = futureTotal, futureErrors
+			row.FutureErrorRate = rate(futureErrors, futureTotal)
+			row.BaselineErrorRate, row.BaselineSamples = baselineRate, samples
+			// A target is only known once the input has reached its exclusive
+			// end. This deliberately withholds the tail of an export instead of
+			// treating an incomplete future interval as a negative label.
+			row.LabelKnown = samples >= cfg.MinBaselineSamples && !start.Add(cfg.Window).Add(cfg.Future).After(observedUntil)
+			row.Label = row.LabelKnown && futureErrors >= cfg.MinFutureErrors && row.FutureErrorRate >= cfg.SpikeMultiple*baselineRate
+			rows = append(rows, row)
+		}
+	}
+	sort.Slice(rows, func(i, j int) bool {
+		if !rows[i].WindowStart.Equal(rows[j].WindowStart) {
+			return rows[i].WindowStart.Before(rows[j].WindowStart)
+		}
+		if rows[i].Service != rows[j].Service {
+			return rows[i].Service < rows[j].Service
+		}
+		return rows[i].Release < rows[j].Release
+	})
+	return rows
+}
+
+type accumulator struct {
+	value     Row
+	durations []float64
+	traces    map[string]struct{}
+	issues    map[string]struct{}
+}
+
+func newAccumulator(start, end time.Time, service, release string) *accumulator {
+	return &accumulator{value: Row{WindowStart: start, WindowEnd: end, Service: service, Release: release}, traces: map[string]struct{}{}, issues: map[string]struct{}{}}
+}
+
+func (a *accumulator) add(e Event) {
+	a.value.TotalEvents++
+	switch e.Kind {
+	case "span":
+		a.value.SpanCount++
+	case "log":
+		a.value.LogCount++
+	case "metric":
+		a.value.MetricCount++
+	}
+	switch e.StatusClass {
+	case "2xx":
+		a.value.Status2xx++
+	case "4xx":
+		a.value.Status4xx++
+	case "5xx":
+		a.value.Status5xx++
+	}
+	if e.TraceID != "" {
+		a.traces[e.TraceID] = struct{}{}
+	}
+	if e.Kind == "span" && e.DurationMS >= 0 {
+		a.durations = append(a.durations, e.DurationMS)
+	}
+	if !e.IsError {
+		return
+	}
+	a.value.ErrorCount++
+	if e.Source == "frontend" {
+		a.value.FrontendErrors++
+	} else {
+		a.value.BackendErrors++
+	}
+	if issue := issueKey(e); issue != "" {
+		a.issues[issue] = struct{}{}
+	}
+	if e.IsRetry {
+		a.value.RetryCount++
+	}
+	if e.IsTimeout {
+		a.value.TimeoutCount++
+	}
+	if e.IsCancellation {
+		a.value.CancellationCount++
+	}
+}
+
+func issueKey(e Event) string {
+	if e.IssueFingerprint != "" {
+		return e.IssueFingerprint
+	}
+	return e.PatternSignature
+}
+
+func (a *accumulator) row() Row {
+	a.value.ErrorRate = rate(a.value.ErrorCount, a.value.TotalEvents)
+	a.value.UniqueTraceCount, a.value.UniqueIssueCount = len(a.traces), len(a.issues)
+	if len(a.durations) != 0 {
+		sort.Float64s(a.durations)
+		a.value.P50Duration = percentile(a.durations, .50)
+		a.value.P95Duration = percentile(a.durations, .95)
+	}
+	return a.value
+}
+
+func count(events []Event, start, end time.Time, service string) (total, errors int) {
+	for _, e := range events {
+		if e.Service == service && !e.Timestamp.Before(start) && e.Timestamp.Before(end) {
+			total++
+			if e.IsError {
+				errors++
+			}
+		}
+	}
+	return total, errors
+}
+
+// baseline samples the same UTC time window on the preceding calendar days.
+// All intervals end before the current observation start, so no future data
+// leaks into a training label.
+func baseline(events []Event, start time.Time, window time.Duration, days int, service string) (float64, int) {
+	var total, errors, samples int
+	for day := 1; day <= days; day++ {
+		from := start.AddDate(0, 0, -day)
+		to := from.Add(window)
+		countTotal, countErrors := count(events, from, to, service)
+		if countTotal == 0 {
+			continue
+		}
+		total += countTotal
+		errors += countErrors
+		samples++
+	}
+	return rate(errors, total), samples
+}
+
+func percentile(values []float64, p float64) float64 {
+	if len(values) == 0 {
+		return 0
+	}
+	index := int(p * float64(len(values)-1))
+	return values[index]
+}
+func rate(errors, total int) float64 {
+	if total == 0 {
+		return 0
+	}
+	return float64(errors) / float64(total)
+}
+func normalizedConfig(cfg Config) Config {
+	d := DefaultConfig()
+	if cfg.Window <= 0 {
+		cfg.Window = d.Window
+	}
+	if cfg.Step <= 0 {
+		cfg.Step = d.Step
+	}
+	if cfg.Future <= 0 {
+		cfg.Future = d.Future
+	}
+	if cfg.BaselineDays <= 0 {
+		cfg.BaselineDays = d.BaselineDays
+	}
+	if cfg.MinBaselineSamples <= 0 {
+		cfg.MinBaselineSamples = d.MinBaselineSamples
+	}
+	if cfg.SpikeMultiple <= 0 {
+		cfg.SpikeMultiple = d.SpikeMultiple
+	}
+	if cfg.MinFutureErrors <= 0 {
+		cfg.MinFutureErrors = d.MinFutureErrors
+	}
+	return cfg
+}

internal/features/features_test.go

@@ -0,0 +1,56 @@
+package features
+
+import (
+	"testing"
+	"time"
+)
+
+func TestBuildLabelsSpikeUsingOnlyEarlierSameTimeWindows(t *testing.T) {
+	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC)
+	events := make([]Event, 0)
+	// The prior two daily windows have 10 events and one error each.
+	for day := 1; day <= 2; day++ {
+		for i := 0; i < 10; i++ {
+			events = append(events, Event{Timestamp: base.AddDate(0, 0, -day).Add(time.Duration(i) * time.Second), Service: "api", Kind: "span", IsError: i == 0, DurationMS: float64(i)})
+		}
+	}
+	// Current row is non-error telemetry. The following fifteen minutes have
+	// five errors out of ten events: 5x the 10% baseline and above the floor.
+	for i := 0; i < 10; i++ {
+		events = append(events, Event{Timestamp: base.Add(time.Duration(i) * time.Second), Service: "api", Release: "v2", Kind: "span", IsError: false, DurationMS: float64(i)})
+	}
+	for i := 0; i < 10; i++ {
+		events = append(events, Event{Timestamp: base.Add(5 * time.Minute).Add(time.Duration(i) * time.Second), Service: "api", Kind: "log", IsError: i < 5})
+	}
+	// The source export's explicit exclusive watermark proves that the rest of
+	// the future interval was observed even though it was quiet.
+	rows := Build(events, Config{Window: 5 * time.Minute, Step: 5 * time.Minute, Future: 15 * time.Minute, ObservedUntil: base.Add(20 * time.Minute), BaselineDays: 2, MinBaselineSamples: 2, SpikeMultiple: 3, MinFutureErrors: 5})
+	var row *Row
+	for i := range rows {
+		if rows[i].WindowStart.Equal(base) && rows[i].Service == "api" && rows[i].Release == "v2" {
+			row = &rows[i]
+			break
+		}
+	}
+	if row == nil {
+		t.Fatal("missing current api/v2 row")
+	}
+	if !row.LabelKnown || !row.Label {
+		t.Fatalf("expected a known positive label, got %+v", *row)
+	}
+	if row.BaselineErrorRate != .1 || row.FutureErrorRate != .5 || row.FutureErrorCount != 5 {
+		t.Fatalf("unexpected label inputs: %+v", *row)
+	}
+	if row.P50Duration != 4 || row.P95Duration != 8 || row.SpanCount != 10 {
+		t.Fatalf("unexpected fixed-width features: %+v", *row)
+	}
+}
+
+func TestBuildWithholdsLabelWithoutCompleteBaseline(t *testing.T) {
+	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC)
+	events := []Event{{Timestamp: base, Service: "api", Kind: "log"}, {Timestamp: base.Add(20 * time.Minute), Service: "other", Kind: "log"}}
+	rows := Build(events, Config{Window: 5 * time.Minute, Step: 5 * time.Minute, Future: 15 * time.Minute, BaselineDays: 7, MinBaselineSamples: 7})
+	if len(rows) == 0 || rows[0].LabelKnown {
+		t.Fatalf("short history must not be labelled: %+v", rows)
+	}
+}

internal/issues/tracker.go

@@ -153,7 +153,7 @@ func NewTracker() *Tracker {
 // fingerprint.
 func (t *Tracker) Record(kind, service, title string, sample Sample) {
 	normalizeSample(&sample, title)
-	id := fingerprint(kind, service, NormalizeTitle(title))
+	id := Fingerprint(kind, service, NormalizeTitle(title))
 	bucket := HourBucket(sample.Timestamp)
 
 	t.mu.Lock()
@@ -174,7 +174,7 @@ func (t *Tracker) Record(kind, service, title string, sample Sample) {
 			NormalizedTitle: NormalizeTitle(title),
 		}
 		issue.PatternSignature = patternSignature(kind, service, title, sample)
-		issue.PatternID = fingerprint("pattern", issue.PatternSignature)
+		issue.PatternID = Fingerprint("pattern", issue.PatternSignature)
 		t.issues[id] = issue
 	}
 
@@ -653,7 +653,10 @@ func decodeStatus(encoded string) string {
 	return b.String()
 }
 
-func fingerprint(parts ...string) string {
+// Fingerprint creates the stable opaque identifier used for canonical issue
+// grouping. Normalization uses it for error rows so offline feature exports
+// can count the same issue identity as the live issue tracker.
+func Fingerprint(parts ...string) string {
 	h := fnv.New64a()
 	_, _ = h.Write([]byte(strings.Join(parts, "\x00")))
 	return fmt.Sprintf("%016x", h.Sum64())

internal/normalization/normalizer.go

@@ -0,0 +1,447 @@
+// Package normalization converts raw OTLP envelopes into stable, non-PII
+// event rows suitable for aggregation and later model training.
+package normalization
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"git.tornberg.me/go-otel/internal/issues"
+)
+
+const Version = "v1"
+
+type Event struct {
+	EventID           string  `json:"eventId"`
+	EntryType         string  `json:"entryType"`
+	EventIndex        uint32  `json:"eventIndex"`
+	IngestTimestamp   int64   `json:"ingestTimestamp"`
+	Timestamp         int64   `json:"timestamp"`
+	NormalizerVersion string  `json:"normalizerVersion"`
+	Service           string  `json:"service"`
+	Release           string  `json:"release,omitempty"`
+	Environment       string  `json:"environment,omitempty"`
+	Kind              string  `json:"kind"`
+	Source            string  `json:"source"`
+	IsError           bool    `json:"isError"`
+	StatusClass       string  `json:"statusClass,omitempty"`
+	StatusCode        int     `json:"statusCode,omitempty"`
+	DurationMS        float64 `json:"durationMs,omitempty"`
+	Route             string  `json:"route,omitempty"`
+	SpanName          string  `json:"spanName,omitempty"`
+	MetricName        string  `json:"metricName,omitempty"`
+	MetricType        string  `json:"metricType,omitempty"`
+	MetricUnit        string  `json:"metricUnit,omitempty"`
+	MetricValue       float64 `json:"metricValue,omitempty"`
+	MetricCount       uint64  `json:"metricCount,omitempty"`
+	MetricSum         float64 `json:"metricSum,omitempty"`
+	MetricMin         float64 `json:"metricMin,omitempty"`
+	MetricMax         float64 `json:"metricMax,omitempty"`
+	ExceptionType     string  `json:"exceptionType,omitempty"`
+	MessageNormalized string  `json:"messageNormalized,omitempty"`
+	StackSignature    string  `json:"stackSignature,omitempty"`
+	TraceID           string  `json:"traceId,omitempty"`
+	IssueFingerprint  string  `json:"issueFingerprint,omitempty"`
+	IsRetry           bool    `json:"isRetry"`
+	IsTimeout         bool    `json:"isTimeout"`
+	IsCancellation    bool    `json:"isCancellation"`
+	ExactSignature    string  `json:"exactSignature"`
+	PatternSignature  string  `json:"patternSignature"`
+}
+
+// NormalizeEntry expands one raw OTLP envelope into one row per span, log
+// record, or metric datapoint. It never mutates or stores the input payload.
+func NormalizeEntry(entryType string, ingestTimestamp int64, data []byte) ([]Event, error) {
+	var envelope map[string]json.RawMessage
+	if err := json.Unmarshal(data, &envelope); err != nil {
+		return nil, fmt.Errorf("decode OTLP JSON: %w", err)
+	}
+	var events []Event
+	switch entryType {
+	case "traces":
+		events = normalizeTraces(ingestTimestamp, envelope)
+	case "logs":
+		events = normalizeLogs(ingestTimestamp, envelope)
+	case "metrics":
+		events = normalizeMetrics(ingestTimestamp, envelope)
+	default:
+		return nil, fmt.Errorf("unsupported telemetry type %q", entryType)
+	}
+	for i := range events {
+		events[i].EntryType = entryType
+		events[i].EventIndex = uint32(i)
+		events[i].IngestTimestamp = ingestTimestamp
+		events[i].EventID = signature(entryType, strconv.FormatInt(ingestTimestamp, 10), strconv.Itoa(i), events[i].ExactSignature)
+	}
+	return events, nil
+}
+
+func normalizeTraces(ingest int64, envelope map[string]json.RawMessage) []Event {
+	var resources []struct {
+		Resource   object `json:"resource"`
+		ScopeSpans []struct {
+			Spans []span `json:"spans"`
+		} `json:"scopeSpans"`
+	}
+	_ = json.Unmarshal(envelope["resourceSpans"], &resources)
+	result := []Event{}
+	for _, resource := range resources {
+		for _, scope := range resource.ScopeSpans {
+			for _, item := range scope.Spans {
+				attrs := mergeAttrs(resource.Resource.Attributes, item.Attributes)
+				status := intValue(item.Status.Code)
+				statusCode := intAttr(attrs, "http.response.status_code", "http.status_code")
+				route := normalizeHTTPRoute(attrs)
+				message := item.Status.Message
+				exceptionType, exceptionMessage, stack := exception(item.Events)
+				if exceptionMessage != "" {
+					message = exceptionMessage
+				}
+				if message == "" {
+					message = item.Name
+				}
+				if exceptionType != "" && message != "" {
+					message = exceptionType + ": " + message
+				}
+				isError := status == 2 || statusCode >= 400 || intAttr(attrs, "rpc.grpc.status_code") > 0 || exceptionType != ""
+				e := baseEvent(ingest, attrs, "span", item.TraceID)
+				e.IsError, e.StatusCode, e.StatusClass = isError, statusCode, statusClass(statusCode)
+				e.DurationMS = durationMS(item.Start, item.End)
+				e.Route, e.SpanName, e.ExceptionType, e.MessageNormalized, e.StackSignature = route, item.Name, exceptionType, normalizeMessage(message), firstFrame(stack)
+				e.IsRetry, e.IsTimeout, e.IsCancellation = operationFlags(attrs, statusCode, intAttr(attrs, "rpc.grpc.status_code"))
+				result = append(result, finish(e))
+			}
+		}
+	}
+	return result
+}
+
+func normalizeLogs(ingest int64, envelope map[string]json.RawMessage) []Event {
+	var resources []struct {
+		Resource  object `json:"resource"`
+		ScopeLogs []struct {
+			LogRecords []logRecord `json:"logRecords"`
+		} `json:"scopeLogs"`
+	}
+	_ = json.Unmarshal(envelope["resourceLogs"], &resources)
+	result := []Event{}
+	for _, resource := range resources {
+		for _, scope := range resource.ScopeLogs {
+			for _, item := range scope.LogRecords {
+				attrs := mergeAttrs(resource.Resource.Attributes, item.Attributes)
+				message := bodyString(item.Body)
+				exceptionType := stringAttr(attrs, "exception.type")
+				stack := stringAttr(attrs, "exception.stacktrace")
+				statusCode := intAttr(attrs, "http.response.status_code", "http.status_code")
+				grpcCode := intAttr(attrs, "rpc.grpc.status_code")
+				e := baseEvent(timestampOr(ingest, item.Timestamp), attrs, "log", item.TraceID)
+				e.IsError = item.SeverityNumber >= 17 || strings.EqualFold(item.SeverityText, "ERROR") || strings.EqualFold(item.SeverityText, "FATAL") || exceptionType != "" || statusCode >= 400 || grpcCode > 0
+				e.StatusCode, e.StatusClass = statusCode, statusClass(statusCode)
+				e.Route, e.ExceptionType, e.MessageNormalized, e.StackSignature = normalizeHTTPRoute(attrs), exceptionType, normalizeMessage(message), firstFrame(stack)
+				e.IsRetry, e.IsTimeout, e.IsCancellation = operationFlags(attrs, statusCode, grpcCode)
+				result = append(result, finish(e))
+			}
+		}
+	}
+	return result
+}
+
+func normalizeMetrics(ingest int64, envelope map[string]json.RawMessage) []Event {
+	var resources []struct {
+		Resource     object `json:"resource"`
+		ScopeMetrics []struct {
+			Metrics []metric `json:"metrics"`
+		} `json:"scopeMetrics"`
+	}
+	_ = json.Unmarshal(envelope["resourceMetrics"], &resources)
+	result := []Event{}
+	for _, resource := range resources {
+		attrs := resource.Resource.Attributes
+		for _, scope := range resource.ScopeMetrics {
+			for _, item := range scope.Metrics {
+				for _, point := range item.Points() {
+					e := baseEvent(timestampOr(ingest, point.Timestamp), mergeAttrs(attrs, point.Attributes), "metric", "")
+					e.MetricName, e.MetricType, e.MetricUnit = item.Name, point.Type, item.Unit
+					e.MetricValue, e.MetricCount, e.MetricSum = point.Value, point.Count, point.Sum
+					e.MetricMin, e.MetricMax = point.Min, point.Max
+					result = append(result, finish(e))
+				}
+			}
+		}
+	}
+	return result
+}
+
+type object struct {
+	Attributes []attribute `json:"attributes"`
+}
+type attribute struct {
+	Key   string `json:"key"`
+	Value value  `json:"value"`
+}
+type value struct {
+	StringValue string      `json:"stringValue"`
+	IntValue    json.Number `json:"intValue"`
+	DoubleValue float64     `json:"doubleValue"`
+	BoolValue   bool        `json:"boolValue"`
+}
+type span struct {
+	TraceID    string      `json:"traceId"`
+	Name       string      `json:"name"`
+	Start      string      `json:"startTimeUnixNano"`
+	End        string      `json:"endTimeUnixNano"`
+	Attributes []attribute `json:"attributes"`
+	Events     []spanEvent `json:"events"`
+	Status     struct {
+		Code    json.Number `json:"code"`
+		Message string      `json:"message"`
+	} `json:"status"`
+}
+type spanEvent struct {
+	Name       string      `json:"name"`
+	Attributes []attribute `json:"attributes"`
+}
+type logRecord struct {
+	TraceID        string      `json:"traceId"`
+	Timestamp      string      `json:"timeUnixNano"`
+	Body           value       `json:"body"`
+	Attributes     []attribute `json:"attributes"`
+	SeverityNumber int         `json:"severityNumber"`
+	SeverityText   string      `json:"severityText"`
+}
+type metric struct {
+	Name                 string              `json:"name"`
+	Unit                 string              `json:"unit"`
+	Gauge                metricNumberData    `json:"gauge"`
+	Sum                  metricNumberData    `json:"sum"`
+	Histogram            metricHistogramData `json:"histogram"`
+	ExponentialHistogram metricNumberData    `json:"exponentialHistogram"`
+}
+
+type metricNumberData struct {
+	DataPoints []numberPoint `json:"dataPoints"`
+}
+
+type metricHistogramData struct {
+	DataPoints []histogramPoint `json:"dataPoints"`
+}
+
+type numberPoint struct {
+	Attributes []attribute `json:"attributes"`
+	Time       string      `json:"timeUnixNano"`
+	AsDouble   float64     `json:"asDouble"`
+	AsInt      json.Number `json:"asInt"`
+}
+
+type histogramPoint struct {
+	Attributes []attribute `json:"attributes"`
+	Time       string      `json:"timeUnixNano"`
+	Count      json.Number `json:"count"`
+	Sum        float64     `json:"sum"`
+	Min        float64     `json:"min"`
+	Max        float64     `json:"max"`
+}
+
+type metricPoint struct {
+	Attributes []attribute
+	Timestamp  string
+	Type       string
+	Value      float64
+	Count      uint64
+	Sum        float64
+	Min        float64
+	Max        float64
+}
+
+func (m metric) Points() []metricPoint {
+	points := make([]metricPoint, 0, len(m.Gauge.DataPoints)+len(m.Sum.DataPoints)+len(m.Histogram.DataPoints)+len(m.ExponentialHistogram.DataPoints))
+	for _, point := range m.Gauge.DataPoints {
+		points = append(points, metricPoint{Attributes: point.Attributes, Timestamp: point.Time, Type: "gauge", Value: numberValue(point.AsDouble, point.AsInt)})
+	}
+	for _, point := range m.Sum.DataPoints {
+		points = append(points, metricPoint{Attributes: point.Attributes, Timestamp: point.Time, Type: "sum", Value: numberValue(point.AsDouble, point.AsInt)})
+	}
+	for _, point := range m.Histogram.DataPoints {
+		count, _ := strconv.ParseUint(string(point.Count), 10, 64)
+		points = append(points, metricPoint{Attributes: point.Attributes, Timestamp: point.Time, Type: "histogram", Count: count, Sum: point.Sum, Min: point.Min, Max: point.Max})
+	}
+	for _, point := range m.ExponentialHistogram.DataPoints {
+		points = append(points, metricPoint{Attributes: point.Attributes, Timestamp: point.Time, Type: "exponentialHistogram", Count: numberUint(point.AsInt), Value: point.AsDouble})
+	}
+	return points
+}
+
+func numberValue(double float64, integer json.Number) float64 {
+	if integer != "" {
+		value, _ := strconv.ParseFloat(string(integer), 64)
+		return value
+	}
+	return double
+}
+
+func numberUint(value json.Number) uint64 {
+	n, _ := strconv.ParseUint(string(value), 10, 64)
+	return n
+}
+
+func mergeAttrs(groups ...[]attribute) []attribute {
+	result := []attribute{}
+	for _, group := range groups {
+		result = append(result, group...)
+	}
+	return result
+}
+func baseEvent(timestamp int64, attrs []attribute, kind, traceID string) Event {
+	source := stringAttr(attrs, "source")
+	if source == "" {
+		source = "backend"
+	}
+	return Event{Timestamp: timestamp, NormalizerVersion: Version, Service: defaultString(stringAttr(attrs, "service.name"), "unknown"), Release: stringAttr(attrs, "service.release"), Environment: defaultString(stringAttr(attrs, "deployment.environment.name", "deployment.environment", "service.environment"), ""), Kind: kind, Source: source, TraceID: traceID}
+}
+func finish(e Event) Event {
+	if e.IsError {
+		kind := e.Kind + "_error"
+		if e.Source == "frontend" {
+			kind = "frontend_error"
+		}
+		title := e.MessageNormalized
+		if title == "" {
+			title = e.SpanName
+		}
+		if title != "" {
+			e.IssueFingerprint = issues.Fingerprint(kind, e.Service, title)
+		}
+	}
+	e.ExactSignature = signature(e.Service, e.Kind, e.Source, e.Route, e.SpanName, e.MetricName, e.MetricType, formatFloat(e.MetricValue), strconv.FormatUint(e.MetricCount, 10), formatFloat(e.MetricSum), strconv.Itoa(e.StatusCode), e.ExceptionType, e.MessageNormalized, e.StackSignature)
+	e.PatternSignature = signature(e.Service, e.Kind, e.Source, e.Route, e.SpanName, e.MetricName, e.MetricType, e.StatusClass, e.ExceptionType, e.MessageNormalized, e.StackSignature)
+	return e
+}
+func formatFloat(value float64) string { return strconv.FormatFloat(value, 'g', 17, 64) }
+func signature(parts ...string) string {
+	h := sha256.New()
+	for _, part := range parts {
+		h.Write([]byte(part))
+		h.Write([]byte{0})
+	}
+	return hex.EncodeToString(h.Sum(nil))
+}
+func stringAttr(attrs []attribute, keys ...string) string {
+	for _, key := range keys {
+		for _, attr := range attrs {
+			if attr.Key == key && attr.Value.StringValue != "" {
+				return attr.Value.StringValue
+			}
+		}
+	}
+	return ""
+}
+func intAttr(attrs []attribute, keys ...string) int {
+	for _, key := range keys {
+		for _, attr := range attrs {
+			if attr.Key == key {
+				if attr.Value.IntValue != "" {
+					n, _ := strconv.Atoi(string(attr.Value.IntValue))
+					return n
+				}
+				if attr.Value.StringValue != "" {
+					n, _ := strconv.Atoi(attr.Value.StringValue)
+					return n
+				}
+			}
+		}
+	}
+	return 0
+}
+func intValue(value json.Number) int { n, _ := strconv.Atoi(string(value)); return n }
+func bodyString(body value) string   { return body.StringValue }
+func timestampOr(fallback int64, value string) int64 {
+	n, _ := strconv.ParseInt(value, 10, 64)
+	if n > 0 {
+		return n
+	}
+	return fallback
+}
+func durationMS(start, end string) float64 {
+	a, _ := strconv.ParseInt(start, 10, 64)
+	b, _ := strconv.ParseInt(end, 10, 64)
+	if b <= a {
+		return 0
+	}
+	return float64(b-a) / 1e6
+}
+func statusClass(status int) string {
+	if status >= 500 && status < 600 {
+		return "5xx"
+	}
+	if status >= 400 && status < 500 {
+		return "4xx"
+	}
+	if status >= 300 && status < 400 {
+		return "3xx"
+	}
+	if status > 0 {
+		return "2xx"
+	}
+	return ""
+}
+func defaultString(value, fallback string) string {
+	if value == "" {
+		return fallback
+	}
+	return value
+}
+func normalizeMessage(message string) string { return issues.NormalizeTitle(message) }
+
+func normalizeHTTPRoute(attrs []attribute) string {
+	route := stringAttr(attrs, "http.route")
+	path := stringAttr(attrs, "url.path", "http.target")
+	// Some HTTP instrumentations report the router mount point (often "/")
+	// as http.route while url.path contains the actual endpoint. Prefer the
+	// specific path in that case so unrelated API calls do not collapse into
+	// one pattern.
+	if path != "" && (route == "" || route == "/" && path != "/") {
+		route = path
+	}
+	return issues.NormalizeRoute(route)
+}
+
+func firstFrame(stack string) string {
+	if i := strings.IndexByte(stack, '\n'); i >= 0 {
+		stack = stack[:i]
+	}
+	return strings.TrimSpace(stack)
+}
+
+// operationFlags only reads structured protocol attributes/statuses. Keeping
+// these flags out of message matching avoids making model features depend on
+// arbitrary user-provided text.
+func operationFlags(attrs []attribute, httpStatus, grpcCode int) (retry, timeout, cancellation bool) {
+	retry = intAttr(attrs, "retry.count", "http.request.resend_count", "http.retry_count") > 0 || boolAttr(attrs, "retry", "http.retry")
+	typeValue := strings.ToLower(stringAttr(attrs, "error.type", "exception.type", "rpc.error.type"))
+	timeout = httpStatus == 408 || httpStatus == 504 || grpcCode == 4 || strings.Contains(typeValue, "timeout") || strings.Contains(typeValue, "deadline")
+	cancellation = grpcCode == 1 || strings.Contains(typeValue, "cancel") || boolAttr(attrs, "cancelled", "canceled", "request.cancelled")
+	return retry, timeout, cancellation
+}
+func boolAttr(attrs []attribute, keys ...string) bool {
+	for _, key := range keys {
+		for _, attr := range attrs {
+			if attr.Key == key && (attr.Value.BoolValue || strings.EqualFold(attr.Value.StringValue, "true")) {
+				return true
+			}
+		}
+	}
+	return false
+}
+func exception(events []spanEvent) (string, string, string) {
+	for _, event := range events {
+		if event.Name != "exception" {
+			continue
+		}
+		return stringAttr(event.Attributes, "exception.type"), stringAttr(event.Attributes, "exception.message"), stringAttr(event.Attributes, "exception.stacktrace")
+	}
+	return "", "", ""
+}

internal/normalization/normalizer_test.go

@@ -0,0 +1,108 @@
+package normalization
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestNormalizeTraceRedactsValuesAndCreatesStableSignatures(t *testing.T) {
+	payload := map[string]interface{}{
+		"resourceSpans": []interface{}{map[string]interface{}{
+			"resource": map[string]interface{}{"attributes": []interface{}{
+				map[string]interface{}{"key": "service.name", "value": map[string]string{"stringValue": "checkout"}},
+			}},
+			"scopeSpans": []interface{}{map[string]interface{}{"spans": []interface{}{map[string]interface{}{
+				"traceId": "abc123", "name": "POST /orders/123",
+				"startTimeUnixNano": "1000000000", "endTimeUnixNano": "1250000000",
+				"attributes": []interface{}{
+					map[string]interface{}{"key": "http.route", "value": map[string]string{"stringValue": "/orders/123"}},
+					map[string]interface{}{"key": "http.response.status_code", "value": map[string]string{"stringValue": "504"}},
+				},
+				"status": map[string]string{"code": "2", "message": "database timeout for order 123"},
+			}}}},
+		}},
+	}
+	data, _ := json.Marshal(payload)
+	first, err := NormalizeEntry("traces", 99, data)
+	if err != nil || len(first) != 1 {
+		t.Fatalf("normalize failed: %v %+v", err, first)
+	}
+	second, err := NormalizeEntry("traces", 99, data)
+	if err != nil || first[0].ExactSignature != second[0].ExactSignature || first[0].PatternSignature != second[0].PatternSignature {
+		t.Fatalf("signatures are not deterministic: %+v %+v", first, second)
+	}
+	event := first[0]
+	if event.Service != "checkout" || event.Route != "/orders/{id}" || event.StatusClass != "5xx" || event.DurationMS != 250 {
+		t.Fatalf("unexpected normalized fields: %+v", event)
+	}
+	if event.MessageNormalized != "database timeout for order <num>" {
+		t.Fatalf("unexpected normalized message: %q", event.MessageNormalized)
+	}
+	if event.TraceID == "" || event.ExactSignature == "" || event.PatternSignature == "" {
+		t.Fatalf("missing correlation fields: %+v", event)
+	}
+	if event.IssueFingerprint == "" || !event.IsTimeout {
+		t.Fatalf("missing canonical issue or timeout features: %+v", event)
+	}
+}
+
+func TestNormalizeLogUsesEventTimestampAndFrontendSource(t *testing.T) {
+	data := []byte(`{"resourceLogs":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"web"}}]},"scopeLogs":[{"logRecords":[{"timeUnixNano":"1230000000","severityNumber":17,"body":{"stringValue":"User 123 failed"},"attributes":[{"key":"source","value":{"stringValue":"frontend"}}]}]}]}]}`)
+	events, err := NormalizeEntry("logs", 99, data)
+	if err != nil || len(events) != 1 {
+		t.Fatalf("normalize failed: %v %+v", err, events)
+	}
+	if events[0].Timestamp != 1230000000 || events[0].Source != "frontend" || !events[0].IsError {
+		t.Fatalf("unexpected log event: %+v", events[0])
+	}
+	if events[0].MessageNormalized != "user <num> failed" {
+		t.Fatalf("unexpected normalized log: %+v", events[0])
+	}
+}
+
+func TestNormalizeLogTreatsHTTPAndGRPCFailuresAsErrors(t *testing.T) {
+	data := []byte(`{"resourceLogs":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"api"}}]},"scopeLogs":[{"logRecords":[{"timeUnixNano":"1230000000","body":{"stringValue":"request failed"},"attributes":[{"key":"http.response.status_code","value":{"intValue":"503"}}]},{"timeUnixNano":"1240000000","body":{"stringValue":"rpc failed"},"attributes":[{"key":"rpc.grpc.status_code","value":{"intValue":"14"}}]}]}]}]}`)
+	events, err := NormalizeEntry("logs", 99, data)
+	if err != nil || len(events) != 2 {
+		t.Fatalf("normalize failed: %v %+v", err, events)
+	}
+	if !events[0].IsError || events[0].StatusClass != "5xx" || !events[1].IsError {
+		t.Fatalf("HTTP/gRPC failures must be errors: %+v", events)
+	}
+}
+
+func TestNormalizeOperationFlagsComeFromStructuredAttributes(t *testing.T) {
+	data := []byte(`{"resourceLogs":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"api"}}]},"scopeLogs":[{"logRecords":[{"timeUnixNano":"1230000000","severityNumber":17,"body":{"stringValue":"opaque message"},"attributes":[{"key":"retry.count","value":{"intValue":"1"}},{"key":"request.cancelled","value":{"boolValue":true}}]}]}]}]}`)
+	events, err := NormalizeEntry("logs", 99, data)
+	if err != nil || len(events) != 1 {
+		t.Fatalf("normalize failed: %v %+v", err, events)
+	}
+	if !events[0].IsRetry || !events[0].IsCancellation || events[0].IsTimeout {
+		t.Fatalf("unexpected structured flags: %+v", events[0])
+	}
+}
+
+func TestNormalizeMetricDatapointsAndHistogram(t *testing.T) {
+	data := []byte(`{"resourceMetrics":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"worker"}}]},"scopeMetrics":[{"metrics":[{"name":"queue.depth","unit":"items","gauge":{"dataPoints":[{"timeUnixNano":"1230000000","asInt":"7","attributes":[{"key":"queue","value":{"stringValue":"payments"}}]}]}},{"name":"request.duration","unit":"ms","histogram":{"dataPoints":[{"timeUnixNano":"1240000000","count":"3","sum":12.5,"min":2,"max":6}]}}]}]}]}`)
+	events, err := NormalizeEntry("metrics", 99, data)
+	if err != nil || len(events) != 2 {
+		t.Fatalf("normalize failed: %v %+v", err, events)
+	}
+	if events[0].MetricName != "queue.depth" || events[0].MetricType != "gauge" || events[0].MetricValue != 7 || events[0].Timestamp != 1230000000 {
+		t.Fatalf("unexpected gauge: %+v", events[0])
+	}
+	if events[1].MetricType != "histogram" || events[1].MetricCount != 3 || events[1].MetricSum != 12.5 || events[1].MetricMin != 2 || events[1].MetricMax != 6 {
+		t.Fatalf("unexpected histogram: %+v", events[1])
+	}
+}
+
+func TestNormalizeHTTPRoutePrefersSpecificPathOverRootMount(t *testing.T) {
+	data := []byte(`{"resourceSpans":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"web"}}]},"scopeSpans":[{"spans":[{"name":"GET /","attributes":[{"key":"http.route","value":{"stringValue":"/"}},{"key":"url.path","value":{"stringValue":"/api/controller/instances"}}]}]}]}]}`)
+	events, err := NormalizeEntry("traces", 99, data)
+	if err != nil || len(events) != 1 {
+		t.Fatalf("normalize failed: %v %+v", err, events)
+	}
+	if events[0].Route != "/api/controller/instances" {
+		t.Fatalf("expected specific URL path, got %q", events[0].Route)
+	}
+}