fathom/internal/ingest/otlp.go

// Package ingest decodes OTLP/HTTP JSON into Fathom's model.
//
// OTLP is the only accepted wire format. That is a deliberate constraint: it
// means anything already speaking OpenTelemetry works against Fathom with an
// endpoint change, and it means there is exactly one decoder to keep correct.
package ingest

import (
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strconv"

	"git.z.gd/zzg/fathom/internal/model"
)

// The OTLP/HTTP JSON wire types. Only the fields Fathom stores are declared;
// encoding/json ignores the rest, which is what makes this resilient to OTLP
// adding fields.
type otlpTraceRequest struct {
	ResourceSpans []struct {
		Resource   otlpResource `json:"resource"`
		ScopeSpans []struct {
			Scope otlpScope  `json:"scope"`
			Spans []otlpSpan `json:"spans"`
		} `json:"scopeSpans"`
	} `json:"resourceSpans"`
}

type otlpLogsRequest struct {
	ResourceLogs []struct {
		Resource  otlpResource `json:"resource"`
		ScopeLogs []struct {
			Scope      otlpScope `json:"scope"`
			LogRecords []otlpLog `json:"logRecords"`
		} `json:"scopeLogs"`
	} `json:"resourceLogs"`
}

type otlpResource struct {
	Attributes []otlpKeyValue `json:"attributes"`
}

type otlpScope struct {
	Name    string `json:"name"`
	Version string `json:"version"`
}

type otlpSpan struct {
	TraceID           string         `json:"traceId"`
	SpanID            string         `json:"spanId"`
	ParentSpanID      string         `json:"parentSpanId"`
	Name              string         `json:"name"`
	Kind              int            `json:"kind"`
	StartTimeUnixNano json.Number    `json:"startTimeUnixNano"`
	EndTimeUnixNano   json.Number    `json:"endTimeUnixNano"`
	Attributes        []otlpKeyValue `json:"attributes"`
	Status            struct {
		Code    int    `json:"code"`
		Message string `json:"message"`
	} `json:"status"`
	Events []struct {
		TimeUnixNano json.Number    `json:"timeUnixNano"`
		Name         string         `json:"name"`
		Attributes   []otlpKeyValue `json:"attributes"`
	} `json:"events"`
}

type otlpLog struct {
	TimeUnixNano         json.Number    `json:"timeUnixNano"`
	ObservedTimeUnixNano json.Number    `json:"observedTimeUnixNano"`
	SeverityNumber       int            `json:"severityNumber"`
	EventName            string         `json:"eventName"`
	Body                 otlpValue      `json:"body"`
	Attributes           []otlpKeyValue `json:"attributes"`
	TraceID              string         `json:"traceId"`
	SpanID               string         `json:"spanId"`
}

type otlpKeyValue struct {
	Key   string    `json:"key"`
	Value otlpValue `json:"value"`
}

// otlpValue is OTLP's AnyValue. Exactly one field is set. Integers arrive as
// JSON strings in OTLP/JSON (the spec mandates it for 64-bit safety), which is
// why IntValue is a json.Number rather than an int64.
type otlpValue struct {
	StringValue *string      `json:"stringValue"`
	BoolValue   *bool        `json:"boolValue"`
	IntValue    *json.Number `json:"intValue"`
	DoubleValue *float64     `json:"doubleValue"`
	ArrayValue  *struct {
		Values []otlpValue `json:"values"`
	} `json:"arrayValue"`
	KvlistValue *struct {
		Values []otlpKeyValue `json:"values"`
	} `json:"kvlistValue"`
	BytesValue *string `json:"bytesValue"`
}

// toModel converts an AnyValue, reporting whether anything was set at all.
// Arrays and maps are flattened to their JSON text: Fathom's attribute values
// are scalars, and preserving the structure as text beats dropping it.
func (v otlpValue) toModel() (model.Value, bool) {
	switch {
	case v.StringValue != nil:
		return model.StringValue(*v.StringValue), true
	case v.BoolValue != nil:
		return model.BoolValue(*v.BoolValue), true
	case v.IntValue != nil:
		if n, err := v.IntValue.Int64(); err == nil {
			return model.IntValue(n), true
		}
		return model.StringValue(v.IntValue.String()), true
	case v.DoubleValue != nil:
		return model.FloatValue(*v.DoubleValue), true
	case v.BytesValue != nil:
		return model.StringValue(*v.BytesValue), true
	case v.ArrayValue != nil:
		// Flatten to the scalar values. Marshalling the AnyValue structs
		// verbatim produces
		// [{"stringValue":"tool_use","boolValue":null,"intValue":null,...}],
		// which is the wire format leaking into the UI where ["tool_use"]
		// was meant.
		out := make([]any, 0, len(v.ArrayValue.Values))
		for _, e := range v.ArrayValue.Values {
			ev, ok := e.toModel()
			if !ok {
				continue
			}
			if f, isNum := ev.Float(); isNum && ev.Type != model.ValString {
				out = append(out, f)
				continue
			}
			out = append(out, ev.Str)
		}
		b, err := json.Marshal(out)
		if err != nil {
			return model.Value{}, false
		}
		return model.StringValue(string(b)), true
	case v.KvlistValue != nil:
		m := make(map[string]string, len(v.KvlistValue.Values))
		for _, kv := range v.KvlistValue.Values {
			if mv, ok := kv.Value.toModel(); ok {
				m[kv.Key] = mv.Str
			}
		}
		b, err := json.Marshal(m)
		if err != nil {
			return model.Value{}, false
		}
		return model.StringValue(string(b)), true
	}
	return model.Value{}, false
}

// Decoder turns OTLP payloads into model records.
type Decoder struct {
	// DropAttrs are attribute keys discarded at ingest, before anything reaches
	// disk. Identity attributes are the motivating case: a single-user store
	// gains nothing from them, and keeping them writes the operator's email
	// into every shard.
	DropAttrs map[string]bool

	// MaxBlobBytes truncates oversized input, output, and body text. An agent
	// that dumps a megabyte tool result should cost a megabyte once, not wedge
	// the UI when someone opens that trace.
	MaxBlobBytes int
}

// DefaultDropAttrs is the identity deny-list applied unless configured
// otherwise.
func DefaultDropAttrs() map[string]bool {
	return map[string]bool{
		"user.email":        true,
		"user.id":           true,
		"user.account_uuid": true,
		"user.account_id":   true,
		"organization.id":   true,
	}
}

func NewDecoder() *Decoder {
	return &Decoder{DropAttrs: DefaultDropAttrs(), MaxBlobBytes: 1 << 20}
}

func (d *Decoder) attrs(kvs []otlpKeyValue) model.Attrs {
	if len(kvs) == 0 {
		return nil
	}
	a := make(model.Attrs, len(kvs))
	for _, kv := range kvs {
		if d.DropAttrs[kv.Key] {
			continue
		}
		if v, ok := kv.Value.toModel(); ok {
			a[kv.Key] = v
		}
	}
	if len(a) == 0 {
		return nil
	}
	return a
}

func (d *Decoder) truncate(s string) string {
	if d.MaxBlobBytes > 0 && len(s) > d.MaxBlobBytes {
		return s[:d.MaxBlobBytes]
	}
	return s
}

// DecodeTraces parses an OTLP/HTTP JSON trace export.
func (d *Decoder) DecodeTraces(body []byte) ([]model.Span, error) {
	var req otlpTraceRequest
	if err := json.Unmarshal(body, &req); err != nil {
		return nil, fmt.Errorf("parsing OTLP trace JSON: %w", err)
	}

	var out []model.Span
	for _, rs := range req.ResourceSpans {
		res := d.attrs(rs.Resource.Attributes)
		service := "unknown"
		if v, ok := res["service.name"]; ok {
			service = v.Str
		}
		for _, ss := range rs.ScopeSpans {
			for _, s := range ss.Spans {
				sp := model.Span{
					TraceID:      normalizeID(s.TraceID),
					SpanID:       normalizeID(s.SpanID),
					ParentSpanID: normalizeID(s.ParentSpanID),
					Name:         s.Name,
					Service:      service,
					StartTS:      nanoToMicro(s.StartTimeUnixNano),
					EndTS:        nanoToMicro(s.EndTimeUnixNano),
					Status:       otlpStatus(s.Status.Code),
					SampleRate:   1,
					Attrs:        d.attrs(s.Attributes),
				}
				sp.Kind = classify(sp.Name, sp.Attrs)
				sp.SampleRate = sampleRate(sp.Attrs)

				// Span events carry payload text in some instrumentations.
				// Folding them into input/output keeps the trace view useful
				// for emitters that never send log records at all.
				sp.Input, sp.Output = d.eventPayloads(&s)

				if sp.SpanID == "" {
					continue // unaddressable; nothing could ever link to it
				}
				out = append(out, sp)
			}
		}
	}
	return out, nil
}

// eventPayloads pulls prompt/completion text out of span events.
func (d *Decoder) eventPayloads(s *otlpSpan) (input, output string) {
	for _, ev := range s.Events {
		for _, kv := range ev.Attributes {
			v, ok := kv.Value.toModel()
			if !ok {
				continue
			}
			switch kv.Key {
			case "gen_ai.prompt", "input", "input.value", "llm.prompts":
				if input == "" {
					input = d.truncate(v.Str)
				}
			case "gen_ai.completion", "output", "output.value", "llm.completions":
				if output == "" {
					output = d.truncate(v.Str)
				}
			}
		}
	}
	return input, output
}

// DecodeLogs parses an OTLP/HTTP JSON logs export.
func (d *Decoder) DecodeLogs(body []byte) ([]model.LogRecord, error) {
	var req otlpLogsRequest
	if err := json.Unmarshal(body, &req); err != nil {
		return nil, fmt.Errorf("parsing OTLP logs JSON: %w", err)
	}

	var out []model.LogRecord
	for _, rl := range req.ResourceLogs {
		for _, sl := range rl.ScopeLogs {
			for _, r := range sl.LogRecords {
				attrs := d.attrs(r.Attributes)

				ts := nanoToMicro(r.TimeUnixNano)
				if ts == 0 {
					ts = nanoToMicro(r.ObservedTimeUnixNano)
				}

				rec := model.LogRecord{
					TraceID:   normalizeID(r.TraceID),
					SpanID:    normalizeID(r.SpanID),
					EventName: eventName(r, attrs),
					Severity:  uint8(clampSeverity(r.SeverityNumber)),
					TS:        ts,
					Attrs:     attrs,
				}
				if v, ok := r.Body.toModel(); ok {
					rec.Body = d.truncate(v.Str)
				}
				out = append(out, rec)
			}
		}
	}
	return out, nil
}

// eventName resolves the record's event name. OTLP has a dedicated field, but
// emitters disagree about where to put it: some use the field, some put it in
// an `event.name` attribute, some only set the body. Checking all three is the
// difference between a readable timeline and a list of "unknown".
func eventName(r otlpLog, attrs model.Attrs) string {
	if r.EventName != "" {
		return r.EventName
	}
	if v, ok := attrs["event.name"]; ok && v.Str != "" {
		return v.Str
	}
	if v, ok := r.Body.toModel(); ok && v.Type == model.ValString && v.Str != "" {
		// Bodies like "claude_code.user_prompt" name the event; longer bodies
		// are content and must not be mistaken for a name.
		if len(v.Str) < 64 {
			return v.Str
		}
	}
	return "log"
}

func clampSeverity(n int) int {
	if n < 0 {
		return 0
	}
	if n > 255 {
		return 255
	}
	return n
}

// sampleRate reads an emitter-provided sample rate so aggregates can compensate
// the way Scuba does. Absent or nonsensical values mean "not sampled".
func sampleRate(a model.Attrs) float64 {
	v, ok := a["sample_rate"]
	if !ok {
		return 1
	}
	f, ok := v.Float()
	if !ok || f <= 0 {
		return 1
	}
	return f
}

func otlpStatus(code int) model.Status {
	switch code {
	case 1:
		return model.StatusOK
	case 2:
		return model.StatusError
	}
	return model.StatusUnset
}

// nanoToMicro converts an OTLP nanosecond timestamp. OTLP/JSON sends these as
// strings because they overflow a float64's integer range; a number that
// arrived as a float has already lost precision, so parsing the text is the
// only correct path.
func nanoToMicro(n json.Number) int64 {
	if n == "" {
		return 0
	}
	if v, err := n.Int64(); err == nil {
		return v / 1000
	}
	if f, err := strconv.ParseFloat(n.String(), 64); err == nil {
		return int64(f / 1000)
	}
	return 0
}

// normalizeID renders a trace or span ID as lowercase hex.
//
// OTLP/JSON specifies hex, but the protobuf-to-JSON mappings in several SDKs
// emit base64 instead. Accepting both and normalising to hex means IDs from
// different emitters are comparable, which matters because span linkage is
// string equality on these values.
func normalizeID(s string) string {
	if s == "" {
		return ""
	}
	if isHex(s) {
		return lower(s)
	}
	if b, err := base64.StdEncoding.DecodeString(s); err == nil && len(b) > 0 {
		return hex.EncodeToString(b)
	}
	return s
}

func isHex(s string) bool {
	for i := 0; i < len(s); i++ {
		c := s[i]
		if (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F') {
			continue
		}
		return false
	}
	return len(s)%2 == 0
}

func lower(s string) string {
	b := []byte(s)
	for i := range b {
		if b[i] >= 'A' && b[i] <= 'Z' {
			b[i] += 'a' - 'A'
		}
	}
	return string(b)
}