448 lines
16 KiB
Go
448 lines
16 KiB
Go
// 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 "", "", ""
|
|
}
|