// Package event extracts the fields logarchiver needs (host, timestamp) from a // raw log event as it flows through the centralized logging JetStream stream // (argocd-apps #296). Events are one JSON object per NATS message and are NOT // normalized — logarchiver persists them raw, so extraction must be tolerant of // the two shapes that share the logs.> subject space: // // - k8s pod logs (Vector kubernetes_logs): host lives at .kubernetes.pod_node_name, // timestamp at .timestamp (RFC3339). // - VM logs (vm-ingest): host at .host (fallback .hostname), timestamp at // .timestamp (fallback .ts). // // The NATS subject itself is authoritative for partitioning and is supplied by // the consumer, not read from the payload. package event import ( "encoding/json" "strings" "time" ) // Meta is the minimal, index-relevant projection of a raw log event. type Meta struct { // Host is the best-effort source host/node for the event, or "" if none // could be determined. Host string // Timestamp is the event time. Ok reports whether a timestamp field was // found and parsed; when false callers should fall back to ingest time. Timestamp time.Time Ok bool } // hostPaths and tsPaths are tried in order. Dotted paths descend into nested // objects (only .kubernetes.pod_node_name is nested today). var ( hostPaths = [][]string{ {"host"}, {"hostname"}, {"kubernetes", "pod_node_name"}, } tsPaths = [][]string{ {"timestamp"}, {"ts"}, {"@timestamp"}, } ) // Extract parses raw (a single JSON log event) and returns its host/timestamp // projection. It never errors: malformed or field-less events yield a zero-value // Meta (Host=="", Ok==false) so the archiver still stores the raw bytes and the // caller can fall back to ingest time. Only the fields of interest are decoded. func Extract(raw []byte) Meta { var doc map[string]json.RawMessage if err := json.Unmarshal(raw, &doc); err != nil { return Meta{} } m := Meta{} m.Host = firstString(doc, hostPaths) if ts, ok := firstTime(doc, tsPaths); ok { m.Timestamp = ts m.Ok = true } return m } // firstString walks each path and returns the first value that decodes to a // non-empty string. func firstString(doc map[string]json.RawMessage, paths [][]string) string { for _, p := range paths { if v, ok := lookup(doc, p); ok { var s string if json.Unmarshal(v, &s) == nil && s != "" { return s } } } return "" } // firstTime walks each path and returns the first value that parses as a // timestamp (RFC3339/RFC3339Nano string, or a numeric unix seconds/millis). func firstTime(doc map[string]json.RawMessage, paths [][]string) (time.Time, bool) { for _, p := range paths { v, ok := lookup(doc, p) if !ok { continue } var s string if json.Unmarshal(v, &s) == nil && s != "" { if t, err := parseTimeString(s); err == nil { return t, true } } var n json.Number if json.Unmarshal(v, &n) == nil { if t, ok := parseNumericTime(n); ok { return t, true } } } return time.Time{}, false } // lookup descends doc following path. Intermediate elements must be JSON objects. func lookup(doc map[string]json.RawMessage, path []string) (json.RawMessage, bool) { cur := doc for i, key := range path { v, ok := cur[key] if !ok { return nil, false } if i == len(path)-1 { return v, true } var next map[string]json.RawMessage if json.Unmarshal(v, &next) != nil { return nil, false } cur = next } return nil, false } var timeLayouts = []string{ time.RFC3339Nano, time.RFC3339, "2006-01-02T15:04:05.999999999Z0700", "2006-01-02 15:04:05.999999999Z07:00", "2006-01-02 15:04:05", } func parseTimeString(s string) (time.Time, error) { var lastErr error for _, l := range timeLayouts { t, err := time.Parse(l, s) if err == nil { return t.UTC(), nil } lastErr = err } return time.Time{}, lastErr } // parseNumericTime interprets n as unix seconds, milliseconds, microseconds, or // nanoseconds based on magnitude. Fractional seconds are supported. func parseNumericTime(n json.Number) (time.Time, bool) { f, err := n.Float64() if err != nil || f <= 0 { return time.Time{}, false } switch { case f >= 1e18: // nanoseconds return time.Unix(0, int64(f)).UTC(), true case f >= 1e15: // microseconds return time.Unix(0, int64(f*1e3)).UTC(), true case f >= 1e12: // milliseconds return time.Unix(0, int64(f*1e6)).UTC(), true default: // seconds (possibly fractional) sec := int64(f) nsec := int64((f - float64(sec)) * 1e9) return time.Unix(sec, nsec).UTC(), true } } // SubjectToken sanitizes a NATS subject into a filesystem/object-key-safe token, // matching the logging stack's convention of replacing [^a-zA-Z0-9_.-] with '_'. // Dots are preserved because subjects are dot-delimited. func SubjectToken(subject string) string { if subject == "" { return "_" } var b strings.Builder for _, r := range subject { switch { case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '_', r == '.', r == '-': b.WriteRune(r) default: b.WriteRune('_') } } return b.String() }