// 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 Route string StatusCode int ExceptionType 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 }