332 lines
9.7 KiB
Go
332 lines
9.7 KiB
Go
// 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
|
|
}
|