113 lines
4.4 KiB
Go
113 lines
4.4 KiB
Go
// window-features exports fixed-width service/release feature rows and the
|
|
// documented future error-spike labels from a normalized JSONL export.
|
|
package main
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"time"
|
|
|
|
"git.tornberg.me/go-otel/internal/features"
|
|
)
|
|
|
|
type event struct {
|
|
Timestamp time.Time `json:"timestamp"`
|
|
Service string `json:"service"`
|
|
Release string `json:"release"`
|
|
Kind string `json:"kind"`
|
|
Source string `json:"source"`
|
|
IsError uint8 `json:"isError"`
|
|
StatusClass string `json:"statusClass"`
|
|
DurationMS float64 `json:"durationMs"`
|
|
TraceID string `json:"traceId"`
|
|
PatternSignature string `json:"patternSignature"`
|
|
IssueFingerprint string `json:"issueFingerprint"`
|
|
IsRetry uint8 `json:"isRetry"`
|
|
IsTimeout uint8 `json:"isTimeout"`
|
|
IsCancellation uint8 `json:"isCancellation"`
|
|
Message string `json:"messageNormalized"`
|
|
SpanName string `json:"spanName"`
|
|
// Optional trace-resolved fields (internal/tracectx); absent in older exports.
|
|
ClientIP string `json:"clientIp"`
|
|
UAClass string `json:"uaClass"`
|
|
IsDuplicate uint8 `json:"isDuplicate"`
|
|
IsRootCause uint8 `json:"isRootCause"`
|
|
HasUpstream uint8 `json:"hasUpstream"`
|
|
}
|
|
|
|
func main() {
|
|
input := flag.String("input", "", "normalized JSONL input")
|
|
window := flag.Duration("window", 5*time.Minute, "observation window")
|
|
step := flag.Duration("step", 5*time.Minute, "window start interval")
|
|
future := flag.Duration("future", 15*time.Minute, "future prediction interval")
|
|
until := flag.String("until", "", "exclusive source-export watermark, RFC3339; required to label a quiet tail safely")
|
|
baselineDays := flag.Int("baseline-days", 7, "same-time-of-day trailing baseline days")
|
|
minBaselineSamples := flag.Int("min-baseline-samples", 7, "minimum nonempty baseline windows required for a label")
|
|
spikeMultiple := flag.Float64("spike-multiple", 3, "future/baseline error-rate ratio for a positive label")
|
|
minFutureErrors := flag.Int("min-future-errors", 5, "minimum future errors for a positive label")
|
|
includeUnknown := flag.Bool("include-unknown-labels", false, "emit rows lacking a complete baseline (default omits them)")
|
|
flag.Parse()
|
|
if *input == "" || *window <= 0 || *step <= 0 || *future <= 0 || *baselineDays < 1 || *minBaselineSamples < 1 || *spikeMultiple <= 0 || *minFutureErrors < 1 {
|
|
log.Fatal("invalid feature configuration")
|
|
}
|
|
var observedUntil time.Time
|
|
if *until != "" {
|
|
parsed, err := time.Parse(time.RFC3339Nano, *until)
|
|
if err != nil {
|
|
log.Fatalf("invalid -until %q: %v", *until, err)
|
|
}
|
|
observedUntil = parsed.UTC()
|
|
}
|
|
|
|
items := readEvents(*input)
|
|
rows := features.Build(items, features.Config{
|
|
Window: *window, Step: *step, Future: *future, ObservedUntil: observedUntil, BaselineDays: *baselineDays,
|
|
MinBaselineSamples: *minBaselineSamples, SpikeMultiple: *spikeMultiple, MinFutureErrors: *minFutureErrors,
|
|
})
|
|
encoder := json.NewEncoder(os.Stdout)
|
|
count := 0
|
|
for _, row := range rows {
|
|
if !*includeUnknown && !row.LabelKnown {
|
|
continue
|
|
}
|
|
if err := encoder.Encode(row); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
count++
|
|
}
|
|
fmt.Fprintf(os.Stderr, "built %d feature rows from %d events\n", count, len(items))
|
|
}
|
|
|
|
func readEvents(path string) []features.Event {
|
|
file, err := os.Open(path)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer file.Close()
|
|
result := []features.Event{}
|
|
scanner := bufio.NewScanner(file)
|
|
scanner.Buffer(make([]byte, 64*1024), 4*1024*1024)
|
|
for scanner.Scan() {
|
|
var item event
|
|
if err := json.Unmarshal(scanner.Bytes(), &item); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
result = append(result, features.Event{
|
|
Timestamp: item.Timestamp, Service: item.Service, Release: item.Release, Kind: item.Kind, Source: item.Source,
|
|
IsError: item.IsError != 0, StatusClass: item.StatusClass, DurationMS: item.DurationMS, TraceID: item.TraceID,
|
|
PatternSignature: item.PatternSignature, IssueFingerprint: item.IssueFingerprint, IsRetry: item.IsRetry != 0,
|
|
IsTimeout: item.IsTimeout != 0, IsCancellation: item.IsCancellation != 0, Message: item.Message, SpanName: item.SpanName,
|
|
ClientIP: item.ClientIP, UAClass: item.UAClass, IsDuplicate: item.IsDuplicate != 0, IsRootCause: item.IsRootCause != 0, HasUpstream: item.HasUpstream != 0,
|
|
})
|
|
}
|
|
if err := scanner.Err(); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
features.MarkBadActors(result, features.BadActorConfig{})
|
|
return result
|
|
}
|