From cf3a9223b8cef89139b13aea2a6a3c0d0cdda51f Mon Sep 17 00:00:00 2001 From: YehoshuaSagron Date: Fri, 14 Aug 2026 19:49:24 +0300 Subject: [PATCH 1/2] Feat: Add the lineage-telemetry plugin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Emits two facts-only OTel spans per HTTP exchange crossing the sidecar: a request span when the request is seen, a response span at stream end, joined by lineage.exchange.id (the request span's own id). Span names are "{self_id} {protocol} {operation}", with the response span appending " response". The facts are lineage.role / direction / self.id / peer.host / protocol / principal.{sub,client} / outcome / denied_by / parent.source, plus url.scheme and url.path. With capture_io the parsed message content rides along as input.value and output.value, so a trace viewer shows the actual A2A message, MCP tool arguments or LLM prompt inline. capture_io is off by default — payloads may carry user messages and model output. The producer records facts, not meaning: no hop classification, no trust vocabulary, no identity guessing. Interpretation belongs to whatever consumes the spans, which is what keeps this package small and lets the vocabulary change without touching Go. Cross-pod parenting rides a single tracestate member: parent from dg-parent when present, else the wire parent, then re-stamp that member with this span's id. The forwarded traceparent is never modified, so an app with its own tracing keeps its chain intact toward its own backend. Nothing guesses a parent — missing data degrades to an explicit unknown. Config decodes with DisallowUnknownFields so a typo'd knob is a boot error rather than a silent default. self_id falls back to self_id_file, defaulting to the operator-mounted /shared/client-id.txt. bypass_paths and bypass_hosts keep agent-card discovery, health probes and telemetry backends out of the graph. Known limit, documented at plugin.go:22: this plugin orders itself after the gate plugins and the pipeline short-circuits on a request-phase reject, so an exchange denied by a gate before OnRequest ran emits no spans at all. Denials after that point are captured as outcome=denied with denied_by. Signed-off-by: YehoshuaSagron --- authbridge/authlib/plugins/lineage/config.go | 79 ++ authbridge/authlib/plugins/lineage/plugin.go | 746 ++++++++++++++ .../authlib/plugins/lineage/plugin_test.go | 912 ++++++++++++++++++ 3 files changed, 1737 insertions(+) create mode 100644 authbridge/authlib/plugins/lineage/config.go create mode 100644 authbridge/authlib/plugins/lineage/plugin.go create mode 100644 authbridge/authlib/plugins/lineage/plugin_test.go diff --git a/authbridge/authlib/plugins/lineage/config.go b/authbridge/authlib/plugins/lineage/config.go new file mode 100644 index 000000000..c030c7e05 --- /dev/null +++ b/authbridge/authlib/plugins/lineage/config.go @@ -0,0 +1,79 @@ +package lineage + +import ( + "bytes" + "encoding/json" + "fmt" + "strings" +) + +// Config holds the per-plugin configuration decoded from the pipeline YAML. +type Config struct { + // OTelEndpoint is the OTLP gRPC endpoint (host:port or http://host:port). + // Default: "localhost:4317" + OTelEndpoint string `json:"otel_endpoint"` + + // CaptureIO when true attaches parsed request/response content as + // input.value (request span) and output.value (response span) + // attributes, enabling Phoenix to display message content inline. + // + // For A2A (inbound agent calls): input = user message parts, output = artifact. + // For MCP tools/call: input = tool params JSON, output = tool result JSON. + // For Inference (LLM): input = messages array JSON, output = completion text. + // + // Off by default — enable only if traces do not contain PII or the + // OTel backend enforces appropriate access controls. + CaptureIO bool `json:"capture_io"` + + // BypassPaths lists URL path prefixes that should not generate lineage + // hops. Useful for suppressing infrastructure polling (agent-card + // discovery, health checks) that would otherwise flood the lineage graph. + // Default: ["/.well-known/", "/healthz", "/readyz", "/health"] + BypassPaths []string `json:"bypass_paths"` + + // BypassHosts lists target host substrings (matched against pctx.Host) + // that should not generate lineage hops. Useful for suppressing + // infrastructure outbound calls such as OTel trace exports. + // Default: ["otel-collector", "jaeger", "zipkin", "prometheus"] + BypassHosts []string `json:"bypass_hosts"` + + // SelfID is the agent's own stable identifier, emitted as the + // lineage.self.id fact on every span. Typically the Keycloak client ID + // of this workload. If empty, SelfIDFile is consulted instead. + SelfID string `json:"self_id"` + + // SelfIDFile is the path to a file containing the agent's own client ID. + // Defaults to /shared/client-id.txt (the operator-mounted credential). + // Ignored when SelfID is set. + SelfIDFile string `json:"self_id_file"` +} + +func defaultConfig() Config { + return Config{ + OTelEndpoint: "localhost:4317", + BypassPaths: []string{"/.well-known/", "/healthz", "/readyz", "/health"}, + BypassHosts: []string{"otel-collector", "jaeger", "zipkin", "prometheus"}, + SelfIDFile: "/shared/client-id.txt", + } +} + +func decodeConfig(raw json.RawMessage) (Config, error) { + cfg := defaultConfig() + if len(raw) == 0 { + return cfg, nil + } + // Unknown keys are a boot error: a typo'd knob (capture-io, selfid_file) + // must not silently run with defaults. + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.DisallowUnknownFields() + if err := dec.Decode(&cfg); err != nil { + return Config{}, fmt.Errorf("lineage-telemetry config: %w", err) + } + if cfg.OTelEndpoint == "" { + cfg.OTelEndpoint = "localhost:4317" + } + // Strip http:// or https:// prefix — gRPC NewClient expects host:port only. + cfg.OTelEndpoint = strings.TrimPrefix(cfg.OTelEndpoint, "https://") + cfg.OTelEndpoint = strings.TrimPrefix(cfg.OTelEndpoint, "http://") + return cfg, nil +} diff --git a/authbridge/authlib/plugins/lineage/plugin.go b/authbridge/authlib/plugins/lineage/plugin.go new file mode 100644 index 000000000..d31169f54 --- /dev/null +++ b/authbridge/authlib/plugins/lineage/plugin.go @@ -0,0 +1,746 @@ +// Package lineage provides the lineage-telemetry authbridge plugin. +// +// Two-span model (see docs/sidecar-wire-contract.md in the lab-data-governance +// repo, the consumer side — the law this file implements). Each HTTP exchange through the sidecar produces TWO OTLP spans: +// +// - a request span, emitted as soon as the request has been seen and +// forwarded, carrying caller-side facts + input.value; and +// - a response span, emitted at stream end (even when no response was +// produced), carrying status/outcome facts + output.value. +// +// Both spans are ended immediately at emission — no span is held open across +// the wait. lineage.exchange.id (= the request span's own span id) is echoed +// on both so the consumer pairs them. The plugin emits FACTS ONLY (direction, +// protocol, endpoints, parsed payloads); all vocabulary — hop kinds, entity +// kinds, caller/callee — lives in the consumer's classify(). See the "removed +// vs today" migration map in the contract for the attrs this no longer emits. +// +// The plugin implements Finisher so the response span is emitted at stream +// end whatever the outcome — including denials that happen AFTER the request +// span was recorded (a response-phase deny, or a request-phase deny by a +// plugin ordered after this one); pctx.Outcome() is available at that point +// and maps to lineage.outcome=denied + lineage.denied_by. LIMITATION: the +// pipeline YAML places this plugin after the gate plugins (ordering is not +// soft-declared under this capabilities model — position in the list is the +// contract), and the pipeline short-circuits on a request-phase Reject, +// so an exchange denied by a gate BEFORE OnRequest ran emits NO spans at all — +// it is invisible to lineage. Moving lineage ahead of the gates (spans for +// denied traffic too) is a named follow-up, not current behavior. +package lineage + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "os" + "strings" + "sync/atomic" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + semconv "go.opentelemetry.io/otel/semconv/v1.26.0" + "go.opentelemetry.io/otel/trace" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/rossoctl/cortex/authbridge/authlib/pipeline" + "github.com/rossoctl/cortex/authbridge/authlib/plugins" +) + +const pluginName = "lineage-telemetry" + +// tracestateStampKey is the W3C tracestate member that carries the sidecar +// parent chain — the single channel every lineage element reads its parent +// from and writes its own request span id into (wire contract v1.5). Inbound +// stamps the request it forwards to its own app; the app's propagate-only shim +// carries tracestate through its per-request causal chain (contextvars), so +// the member surfaces on exactly the outbound calls that inbound caused. +// Outbound re-stamps the request it forwards to the peer, whose inbound +// sidecar reads it as its parent. Parent precedence is stamp > wire parent in +// BOTH directions, and the chosen source is recorded as the +// lineage.parent.source fact. The forwarded traceparent is never modified: +// the sidecar chain lives entirely in this member, so an app that emits its +// own spans keeps an intact traceparent chain toward its own backend while +// the sidecar chain stays self-consistent in ours. (Until v1.4 the outbound +// instead rewrote the forwarded traceparent — the splice; v1.5 removed it.) +// +// The key names the consuming data-governance system (W3C convention: the key +// identifies the owner of the entry) and is deliberately platform-neutral — +// it was `kglin` until 2026-08-04; the name never lands in stored data, so +// renaming is wire-only. +// +// A trace-keyed map (one entry per trace, "the last inbound seen") used to sit +// between the two. It was removed: its answer is correct only while exactly one +// inbound of that trace is in flight — a precondition it never checked and could +// not verify — and when it was wrong it produced a real, exported, walkable +// parent that was simply untrue. Un-stamped traffic falls to the wire +// parent, which is an app-internal span this pipeline never exported: the +// interaction still derives in full, but as a trace entry rather than a child. +// A visibly missing edge is recoverable; a silently wrong one is not. +const tracestateStampKey = "dg-parent" + +func init() { + plugins.RegisterPlugin(pluginName, func() pipeline.Plugin { return NewLineageTelemetry() }) +} + +// exchangeState carries what OnFinish needs to emit the response span as the +// twin of the request span emitted in OnRequest. +type exchangeState struct { + // reqCtx is the (already-ended) request span's context — the parent of + // the response span. An ended span's SpanContext is a valid parent. + reqCtx trace.SpanContext + // common holds the attributes shared by both spans (lineage.direction, + // self.id, peer.*, protocol, exchange.id) — NOT lineage.role, which + // differs per span. Computed once so both spans agree byte-for-byte. + common []attribute.KeyValue + spanKind trace.SpanKind + // spanName is the request span's name; the response span appends " response". + spanName string + // protocol is the request span's lineage.protocol fact; the response + // span's output.value must be read through the SAME protocol's parser + // (parsers are precedence-ordered, not mutually exclusive — mcp-parser + // also matches any JSON-RPC body, including every a2a exchange). + protocol string +} + +// LineageTelemetry emits OTel spans for each request hop observed by authbridge. +type LineageTelemetry struct { + cfg Config + tp *sdktrace.TracerProvider + tracer trace.Tracer + ready atomic.Bool + propagator propagation.TextMapPropagator + selfID string // agent's own client ID for the lineage.self.id fact +} + +// NewLineageTelemetry constructs an unconfigured plugin. Configure + Init must +// run before it serves traffic (guarded by Ready()). +func NewLineageTelemetry() *LineageTelemetry { + return &LineageTelemetry{ + propagator: propagation.TraceContext{}, + } +} + +func (p *LineageTelemetry) Name() string { return pluginName } + +func (p *LineageTelemetry) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{ + // At least one protocol parser must be present and earlier in the + // chain: the protocol fact and both payload reductions read the + // parsers' Extensions. A chain that misorders lineage before its + // parsers (or has none) fails at startup instead of silently + // emitting lineage.protocol="http" on everything. jwt-validation + // ordering (for the principal facts) cannot be soft-declared under + // this capabilities model — list it before lineage in the YAML. + RequiresAny: []string{"a2a-parser", "mcp-parser", "inference-parser"}, + // The contract is cited major.minor only, deliberately: patch + // revisions (v1.5.x) clarify prose and never change span semantics, + // so a patch bump must not imply a producer change. + Description: "Emits two facts-only lineage spans per HTTP exchange (wire contract v1.5).", + } +} + +func (p *LineageTelemetry) Configure(raw json.RawMessage) error { + cfg, err := decodeConfig(raw) + if err != nil { + return err + } + p.cfg = cfg + return nil +} + +func (p *LineageTelemetry) Init(ctx context.Context) error { + endpoint := p.cfg.OTelEndpoint + conn, err := grpc.NewClient(endpoint, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err != nil { + return fmt.Errorf("lineage-telemetry: gRPC dial %s: %w", endpoint, err) + } + + exporter, err := otlptracegrpc.New(ctx, + otlptracegrpc.WithGRPCConn(conn), + ) + if err != nil { + return fmt.Errorf("lineage-telemetry: OTLP exporter: %w", err) + } + + res, err := resource.New(ctx, + resource.WithAttributes( + semconv.ServiceNameKey.String("authbridge"), + attribute.String("authbridge.component", pluginName), + ), + ) + if err != nil { + slog.Warn("lineage-telemetry: resource detection failed, using default", "error", err) + res = resource.Default() + } + + p.tp = sdktrace.NewTracerProvider( + sdktrace.WithBatcher(exporter), + sdktrace.WithResource(res), + ) + p.tracer = p.tp.Tracer("authbridge/" + pluginName) + + // Resolve self identity for the lineage.self.id fact. Every span this + // plugin emits is a claim of the form "X did Y"; with no X there is no + // claim to make, so an unresolvable identity refuses to start rather + // than serving traffic under a plausible-but-wrong label ("no mechanism + // may guess", contract v1.3). Note the asymmetry with this file's other + // unknowns: a missing status, payload or parent anchor is a missing PART + // of a fact and degrades honestly (abandoned / NULL / parent.source=wire). + // Identity is the fact's subject — it has no degraded form, and a shared + // placeholder would collapse every unidentified pod onto one entity row + // (entity id = uuid5("{kind}:{self.id}"), and entities is upsert-only). + if p.cfg.SelfID != "" { + p.selfID = p.cfg.SelfID + } else if p.cfg.SelfIDFile != "" { + raw, err := os.ReadFile(p.cfg.SelfIDFile) + if err != nil { + return fmt.Errorf("lineage-telemetry: no inline self_id and self_id_file unreadable: %w", err) + } + p.selfID = strings.TrimSpace(string(raw)) + } + if p.selfID == "" { + return fmt.Errorf("lineage-telemetry: self identity unresolved (empty self_id and self_id_file %q)", p.cfg.SelfIDFile) + } + + p.ready.Store(true) + slog.Info("lineage-telemetry: initialized", "endpoint", endpoint, "self_id", p.selfID) + return nil +} + +func (p *LineageTelemetry) Shutdown(ctx context.Context) error { + if p.tp == nil { + return nil + } + return p.tp.Shutdown(ctx) +} + +func (p *LineageTelemetry) Ready() bool { return p.ready.Load() } + +func (p *LineageTelemetry) OnRequest(ctx context.Context, pctx *pipeline.Context) pipeline.Action { + if !p.ready.Load() { + pctx.Skip("not_ready") + return pipeline.Action{Type: pipeline.Continue} + } + + // Skip infrastructure paths (health checks, agent-card discovery, etc.) + for _, prefix := range p.cfg.BypassPaths { + if strings.HasPrefix(pctx.Path, prefix) { + pctx.Skip("bypass_path") + return pipeline.Action{Type: pipeline.Continue} + } + } + + // Skip infrastructure outbound targets (OTel exporters, metrics scrapers, etc.) + for _, substr := range p.cfg.BypassHosts { + if strings.Contains(pctx.Host, substr) { + pctx.Skip("bypass_host") + return pipeline.Action{Type: pipeline.Continue} + } + } + + // Extract remote trace context from the incoming W3C traceparent header. + // HeaderCarrier wraps http.Header and uses case-insensitive Get/Keys so + // canonical-form keys ("Traceparent") match the propagator's lowercase + // lookups. + remoteCtx := p.propagator.Extract(ctx, propagation.HeaderCarrier(pctx.Headers)) + + protocol := protocolOf(pctx) + self := serviceLabel(p.selfID) + spanKind := spanKindFor(pctx.Direction) + spanName := requestSpanName(self, protocol, spanOp(pctx, protocol)) + + // Facts shared by both spans (exchange.id is appended once the request + // span exists, since it IS the request span id). + base := baseAttrs(pctx, self, protocol) + + // Request-span attributes: role + shared facts + request-only facts. + reqAttrs := make([]attribute.KeyValue, 0, len(base)+8) + reqAttrs = append(reqAttrs, attribute.String("lineage.role", "request")) + reqAttrs = append(reqAttrs, base...) + reqAttrs = p.appendRequestFacts(reqAttrs, pctx, protocol) + + // (3) parent · (4) emit · (5) re-stamp — wire contract v1.5. The emit is + // unconditional; the two calls around it are the stamp machinery. + // + // >>> OPTION-4 DELETION POINT <<< + // A pure read-only sidecar deletes exactly the selectParent and + // restampTracestate calls below (and the parent.source fact), parenting on + // remoteCtx alone — wire-parent-only propagation. The emit stays. The + // trade-off to weigh first: without the stamp, two sidecarred pods can + // only be joined through the app's own propagation, so cross-pod + // parenting degrades to whatever the wire parent happens to carry. + parent, parentSource := selectParent(ctx, remoteCtx) + reqAttrs = append(reqAttrs, attribute.String("lineage.parent.source", parentSource)) + reqCtx := p.emitRequestSpan(parent, spanName, spanKind, reqAttrs) + exchangeID := reqCtx.SpanID().String() + restampTracestate(pctx, remoteCtx, exchangeID) + + common := make([]attribute.KeyValue, 0, len(base)+1) + common = append(common, base...) + common = append(common, attribute.String("lineage.exchange.id", exchangeID)) + + pipeline.SetState(pctx, pluginName, &exchangeState{ + reqCtx: reqCtx, + common: common, + spanKind: spanKind, + spanName: spanName, + protocol: protocol, + }) + pctx.Observe("recorded_request") + return pipeline.Action{Type: pipeline.Continue} +} + +// selectParent is step (3) of the single-channel parenting mechanism (wire +// contract v1.5): the parent is the tracestate stamp — the previous sidecar +// element in the chain (the caller's outbound for an inbound, this pod's +// inbound for an outbound) — else the wire parent. Same precedence in both +// directions. There is deliberately no third option: guessing an attribution +// is worse than declining to give one. Returns the parent context and the +// source label the caller emits as the lineage.parent.source fact. +func selectParent(ctx, remoteCtx context.Context) (context.Context, string) { + rsc := trace.SpanContextFromContext(remoteCtx) + if rsc.IsValid() { + if psc, ok := stampedParent(rsc); ok { + return trace.ContextWithRemoteSpanContext(ctx, psc), "tracestate" + } + } + return remoteCtx, "wire" +} + +// emitRequestSpan is step (4): emit the request span under parent and end it +// immediately — no span is held open across the exchange. lineage.exchange.id +// is the span's OWN id, so it can only be set after Start. Returns the span's +// context; an ended span's SpanContext remains a valid parent for the response +// span, and its span id is the exchange id. +func (p *LineageTelemetry) emitRequestSpan( + parent context.Context, + spanName string, + spanKind trace.SpanKind, + reqAttrs []attribute.KeyValue, +) trace.SpanContext { + _, span := p.tracer.Start(parent, spanName, + trace.WithSpanKind(spanKind), + trace.WithAttributes(reqAttrs...), + ) + sc := span.SpanContext() + span.SetAttributes(attribute.String("lineage.exchange.id", sc.SpanID().String())) + span.End() + return sc +} + +// restampTracestate is step (5): rewrite the forwarded request's tracestate +// member with this exchange id — both directions. Inbound: the app's +// propagate-only shim couriers it to exactly the outbound calls this inbound +// caused. Outbound: the peer sidecar's inbound reads it as its parent. The +// forwarded traceparent is never modified (see tracestateStampKey). A valid +// wire traceparent is required — without one the app's shim starts a fresh +// root trace and drops the tracestate anyway, so there is nothing to stamp. +// The listener is responsible for propagating this header mutation (ext_proc +// emits a SetHeaders diff). +func restampTracestate(pctx *pipeline.Context, remoteCtx context.Context, exchangeID string) { + rsc := trace.SpanContextFromContext(remoteCtx) + if !rsc.IsValid() { + return + } + ts, err := rsc.TraceState().Insert(tracestateStampKey, exchangeID) + if err != nil { + // Stamp attempted and refused (tracestate full or a member malformed, + // W3C caps at 32 members / 512 bytes). Without this line the outcome + // is indistinguishable from "app has no shim". + slog.Warn("lineage-telemetry: tracestate stamp rejected; the next element will attribute as wire", + "exchange_id", exchangeID, "error", err) + return + } + pctx.Headers.Set("tracestate", ts.String()) +} + +// stampedParent resolves the tracestate stamp on an outbound wire context: +// the inbound exchange id this pod's sidecar wrote into tracestate on the +// forwarded request, carried back by the app's shim. Returns ok=false when +// the member is absent or malformed (caller falls back to the wire parent). +func stampedParent(rsc trace.SpanContext) (trace.SpanContext, bool) { + raw := rsc.TraceState().Get(tracestateStampKey) + if raw == "" { + return trace.SpanContext{}, false + } + sid, err := trace.SpanIDFromHex(raw) + if err != nil { + return trace.SpanContext{}, false + } + psc := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: rsc.TraceID(), + SpanID: sid, + TraceFlags: rsc.TraceFlags(), + Remote: true, + }) + return psc, psc.IsValid() +} + +// OnResponse is a no-op. The response span is emitted in OnFinish (which fires +// on every finished exchange, including denials and abandonments), not here. +// The method exists only to satisfy the base pipeline.Plugin interface, which +// mandates OnResponse; it carries no logic in the two-span model. +func (p *LineageTelemetry) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} + +// OnFinish emits the response span — the twin of the request span, parented +// under it and echoing the same exchange.id — carrying outcome/status/output. +// Always fires at stream end, so a bodyless or failed exchange still completes +// as a first-class pair. Runs under a recover so an unexpected state never +// crashes the pipeline. +func (p *LineageTelemetry) OnFinish(ctx context.Context, pctx *pipeline.Context) { + defer func() { + if r := recover(); r != nil { + slog.Warn("lineage-telemetry: OnFinish panic recovered", "recover", r) + } + }() + + state := pipeline.GetState[exchangeState](pctx, pluginName) + if state == nil || !state.reqCtx.IsValid() { + return + } + + outcome, status, hasStatus, deniedBy := lineageOutcome(pctx.Outcome()) + + attrs := make([]attribute.KeyValue, 0, len(state.common)+5) + attrs = append(attrs, attribute.String("lineage.role", "response")) + attrs = append(attrs, state.common...) + attrs = append(attrs, attribute.String("lineage.outcome", outcome)) + if hasStatus { + attrs = append(attrs, attribute.Int("http.status_code", status)) + } + if deniedBy != "" { + attrs = append(attrs, attribute.String("lineage.denied_by", deniedBy)) + } + if p.cfg.CaptureIO { + if v := ioOutputValue(pctx, state.protocol); v != "" { + attrs = append(attrs, attribute.String("output.value", v)) + } + } + + parent := trace.ContextWithRemoteSpanContext(ctx, state.reqCtx) + _, span := p.tracer.Start(parent, state.spanName+" response", + trace.WithSpanKind(state.spanKind), + trace.WithAttributes(attrs...), + ) + span.End() +} + +// lineageOutcome maps the pipeline's 3-value Outcome (allow/deny/error, nil +// outside OnFinish) onto the contract's lineage.outcome vocabulary +// (ok|denied|error|abandoned) plus the http.status_code fact. A terminal state +// with no status written (upstream reset, client disconnect, listener death) +// is "abandoned" — the row completes as in-flight-turned-failed rather than +// dangling. hasStatus is false when no status code was produced. +func lineageOutcome(o *pipeline.Outcome) (outcome string, status int, hasStatus bool, deniedBy string) { + if o == nil { + return "abandoned", 0, false, "" + } + switch o.FinalAction { + case pipeline.OutcomeAllow: + return "ok", o.StatusCode, o.StatusCode > 0, "" + case pipeline.OutcomeDeny: + return "denied", o.StatusCode, o.StatusCode > 0, o.DenyingPlugin + case pipeline.OutcomeError: + if o.StatusCode > 0 { + return "error", o.StatusCode, true, "" + } + return "abandoned", 0, false, "" + default: + return "error", o.StatusCode, o.StatusCode > 0, "" + } +} + +// protocolOf reports which parser populated Extensions — the lineage.protocol +// fact. "http" means no parser matched. +func protocolOf(pctx *pipeline.Context) string { + switch { + case pctx.Extensions.A2A != nil: + return "a2a" + case pctx.Extensions.MCP != nil: + return "mcp" + case pctx.Extensions.Inference != nil: + return "inference" + default: + return "http" + } +} + +// spanKindFor maps direction to OTel SpanKind: inbound is SERVER, outbound is +// CLIENT. The response span reuses its request span's kind. +func spanKindFor(dir pipeline.Direction) trace.SpanKind { + if dir == pipeline.Inbound { + return trace.SpanKindServer + } + return trace.SpanKindClient +} + +// baseAttrs returns the facts carried on BOTH spans except exchange.id (added +// once the request span id is known) and role (differs per span). +func baseAttrs(pctx *pipeline.Context, self, protocol string) []attribute.KeyValue { + attrs := []attribute.KeyValue{ + attribute.String("lineage.direction", pctx.Direction.String()), + attribute.String("lineage.self.id", self), + attribute.String("lineage.protocol", protocol), + } + if pctx.Host != "" { + attrs = append(attrs, attribute.String("lineage.peer.host", pctx.Host)) + } + return attrs +} + +// appendRequestFacts adds the request-only facts: HTTP method/path/scheme, the +// protocol-specific parsed facts, validated-JWT principal (inbound only), and +// input.value when capture_io is on. protocolOf guarantees the matching +// extension pointer is non-nil. +func (p *LineageTelemetry) appendRequestFacts(attrs []attribute.KeyValue, pctx *pipeline.Context, protocol string) []attribute.KeyValue { + if pctx.Method != "" { + attrs = append(attrs, attribute.String("http.method", pctx.Method)) + } + if pctx.Path != "" { + attrs = append(attrs, attribute.String("url.path", pctx.Path)) + } + if pctx.Scheme != "" { + attrs = append(attrs, attribute.String("url.scheme", pctx.Scheme)) + } + switch protocol { + case "a2a": + a := pctx.Extensions.A2A + if a.Method != "" { + attrs = append(attrs, attribute.String("a2a.method", a.Method)) + } + if a.SessionID != "" { + attrs = append(attrs, attribute.String("a2a.session_id", a.SessionID)) + } + case "mcp": + m := pctx.Extensions.MCP + if m.Method != "" { + attrs = append(attrs, attribute.String("mcp.method", m.Method)) + } + if t := mcpTool(pctx); t != "" { + attrs = append(attrs, attribute.String("mcp.tool", t)) + } + case "inference": + if model := pctx.Extensions.Inference.Model; model != "" { + attrs = append(attrs, attribute.String("inference.model", model)) + } + } + // Principal facts: request span, inbound only, and only from a validated + // JWT (pctx.Identity non-nil). + if pctx.Direction == pipeline.Inbound && pctx.Identity != nil { + if s := pctx.Identity.Subject(); s != "" { + attrs = append(attrs, attribute.String("lineage.principal.sub", s)) + } + if c := pctx.Identity.ClientID(); c != "" { + attrs = append(attrs, attribute.String("lineage.principal.client", c)) + } + } + if p.cfg.CaptureIO { + if v := ioInputValue(pctx, protocol); v != "" { + attrs = append(attrs, attribute.String("input.value", v)) + } + } + return attrs +} + +// requestSpanName builds "{self} {protocol} {op}", dropping the trailing op +// when it is empty. The response span appends " response". +func requestSpanName(self, protocol, op string) string { + if op == "" { + return self + " " + protocol + } + return self + " " + protocol + " " + op +} + +// spanOp picks the operation label for the span name per protocol: +// mcp.tool / a2a.method / inference.model, falling back to url.path. +func spanOp(pctx *pipeline.Context, protocol string) string { + var op string + switch protocol { + case "a2a": + if pctx.Extensions.A2A != nil { + op = pctx.Extensions.A2A.Method + } + case "mcp": + op = mcpTool(pctx) + if op == "" && pctx.Extensions.MCP != nil { + op = pctx.Extensions.MCP.Method + } + case "inference": + if pctx.Extensions.Inference != nil { + op = pctx.Extensions.Inference.Model + } + } + if op == "" { + op = pctx.Path + } + return op +} + +// mcpTool returns the tool name for an MCP tools/call, or "" otherwise. +func mcpTool(pctx *pipeline.Context) string { + m := pctx.Extensions.MCP + if m == nil || m.Method != "tools/call" || m.Params == nil { + return "" + } + if name, ok := m.Params["name"].(string); ok { + return name + } + return "" +} + +// serviceLabel reduces a SPIFFE ID to its last path segment, or returns +// selfID as-is if it is not a SPIFFE URI. Used for the lineage.self.id fact +// and span names. +// +// "spiffe://trust-domain/ns/team1/sa/weather-service" → "weather-service" +// "weather-service" → "weather-service" +// +// selfID is never empty at the only call site: Init refuses to start without +// a resolved identity (v1.3). There is deliberately no empty-string fallback — +// inventing a label is the guess that rule exists to forbid. +func serviceLabel(selfID string) string { + parts := strings.Split(selfID, "/") + for i := len(parts) - 1; i >= 0; i-- { + if parts[i] != "" { + return parts[i] + } + } + return selfID +} + +// ioInputValue returns the input.value for a request span: the parsed request +// content for *protocol* — the hop's lineage.protocol fact — or "" if that +// parser produced nothing meaningful. Only that protocol's extension is read: +// parsers are precedence-ordered, not mutually exclusive (mcp-parser matches +// any JSON-RPC body, including every a2a exchange), so falling through to +// another parser's output would attach a mislabeled protocol envelope. A hop +// whose own parser yields nothing keeps a NULL payload — the contract's +// "interactions are independent of payloads". +func ioInputValue(pctx *pipeline.Context, protocol string) string { + ext := pctx.Extensions + switch { + case protocol == "a2a" && ext.A2A != nil && len(ext.A2A.Parts) > 0: + // Collect all text parts; fall back to JSON if non-text parts present. + var texts []string + for _, p := range ext.A2A.Parts { + if p.Content != "" { + texts = append(texts, p.Content) + } + } + if len(texts) > 0 { + return strings.Join(texts, "\n") + } + if b, err := json.Marshal(ext.A2A.Parts); err == nil { + return string(b) + } + case protocol == "inference" && ext.Inference != nil && len(ext.Inference.Messages) > 0: + if b, err := json.Marshal(ext.Inference.Messages); err == nil { + return string(b) + } + case protocol == "mcp" && ext.MCP != nil && ext.MCP.Params != nil: + // For tools/call, surface just the arguments (the semantically + // meaningful part) rather than the full {"name":…,"arguments":…} wrapper. + if ext.MCP.Method == "tools/call" { + if args, ok := ext.MCP.Params["arguments"]; ok { + if b, err := json.Marshal(args); err == nil { + return string(b) + } + } + } + if b, err := json.Marshal(ext.MCP.Params); err == nil { + return string(b) + } + } + return "" +} + +// isA2AProtocolEvent returns true when s is a JSON object carrying an A2A +// transport-level "kind" field (status-update, task-status-update, etc.) +// rather than actual content. Used to avoid surfacing protocol metadata +// as output.value when the a2a-parser captures a protocol event as the +// artifact instead of the real agent response text. +func isA2AProtocolEvent(s string) bool { + var obj map[string]json.RawMessage + if json.Unmarshal([]byte(s), &obj) != nil { + return false + } + var kind string + if raw, ok := obj["kind"]; ok { + _ = json.Unmarshal(raw, &kind) + } + return strings.Contains(kind, "status") || strings.Contains(kind, "artifact-update") || + strings.Contains(kind, "Status") || kind == "working" || kind == "canceled" +} + +// ioOutputValue returns the output.value for a response span: the parsed +// response content for *protocol* — the REQUEST span's lineage.protocol fact — +// or "" if that parser produced nothing. Only that protocol's extension is +// read, for the same reason as ioInputValue: mcp-parser also parses every a2a +// response (any JSON-RPC body), and falling through to it would emit the raw +// JSON-RPC envelope — including the protocol events isA2AProtocolEvent exists +// to suppress — as an a2a hop's payload. +func ioOutputValue(pctx *pipeline.Context, protocol string) string { + ext := pctx.Extensions + switch { + case protocol == "a2a" && ext.A2A != nil && ext.A2A.Artifact != "" && !isA2AProtocolEvent(ext.A2A.Artifact): + return ext.A2A.Artifact + case protocol == "a2a" && ext.A2A != nil && ext.A2A.ErrorMessage != "": + return ext.A2A.ErrorMessage + case protocol == "inference" && ext.Inference != nil && ext.Inference.Completion != "": + return ext.Inference.Completion + case protocol == "inference" && ext.Inference != nil && len(ext.Inference.ToolCalls) > 0: + if b, err := json.Marshal(ext.Inference.ToolCalls); err == nil { + return string(b) + } + case protocol == "mcp" && ext.MCP != nil && ext.MCP.Result != nil: + // For tools/call results, extract the text content from the MCP + // content array rather than returning the full {"content":[…],"_meta":…} + // envelope, so the output matches what Phoenix shows for the tool span. + if ext.MCP.Method == "tools/call" { + if content, ok := ext.MCP.Result["content"]; ok { + if items, ok := content.([]any); ok { + var texts []string + for _, item := range items { + if m, ok := item.(map[string]any); ok { + if m["type"] == "text" { + if t, ok := m["text"].(string); ok && t != "" { + texts = append(texts, t) + } + } + } + } + if len(texts) > 0 { + return strings.Join(texts, "\n") + } + } + } + } + if b, err := json.Marshal(ext.MCP.Result); err == nil { + return string(b) + } + case protocol == "mcp" && ext.MCP != nil && ext.MCP.Err != nil: + if b, err := json.Marshal(ext.MCP.Err); err == nil { + return string(b) + } + } + return "" +} + +// Compile-time interface assertions. +var ( + _ pipeline.Plugin = (*LineageTelemetry)(nil) + _ pipeline.Configurable = (*LineageTelemetry)(nil) + _ pipeline.Initializer = (*LineageTelemetry)(nil) + _ pipeline.Shutdowner = (*LineageTelemetry)(nil) + _ pipeline.Finisher = (*LineageTelemetry)(nil) + _ pipeline.Readier = (*LineageTelemetry)(nil) +) diff --git a/authbridge/authlib/plugins/lineage/plugin_test.go b/authbridge/authlib/plugins/lineage/plugin_test.go new file mode 100644 index 000000000..59de55f05 --- /dev/null +++ b/authbridge/authlib/plugins/lineage/plugin_test.go @@ -0,0 +1,912 @@ +package lineage + +import ( + "context" + "encoding/json" + "maps" + "net/http" + "os" + "strings" + "testing" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/propagation" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" + + "github.com/rossoctl/cortex/authbridge/authlib/pipeline" +) + +// newTestPlugin creates a LineageTelemetry wired to an in-memory span exporter +// (synchronous, so a span appears the instant it is ended) and marks it ready +// so Init is not needed. +func newTestPlugin(t *testing.T) (*LineageTelemetry, *tracetest.InMemoryExporter) { + t.Helper() + exp := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exp)) + p := NewLineageTelemetry() + p.tp = tp + p.tracer = tp.Tracer("test") + p.selfID = "weather-service" + p.ready.Store(true) + return p, exp +} + +// run drives a full exchange (request pass + finish) through a single-plugin +// pipeline. Spans are read from the caller's exporter. +func run(t *testing.T, p *LineageTelemetry, pctx *pipeline.Context, outcome pipeline.Outcome) { + t.Helper() + pl, err := pipeline.New([]pipeline.Plugin{p}) + if err != nil { + t.Fatalf("pipeline.New: %v", err) + } + pl.Run(context.Background(), pctx) + pl.RunFinish(context.Background(), pctx, outcome) +} + +// allow is the ordinary success outcome. +func allow(status int) pipeline.Outcome { + return pipeline.Outcome{FinalAction: pipeline.OutcomeAllow, StatusCode: status} +} + +// fakeContext mirrors what the real listeners supply. Method is populated +// because every listener now supplies it (reverseproxy/forwardproxy from +// r.Method, ext_proc from the :method pseudo-header) — if a listener stops, +// the fixture must change with it rather than keep asserting a fiction. +func fakeContext(dir pipeline.Direction, headers http.Header) *pipeline.Context { + return &pipeline.Context{ + Direction: dir, + Method: "POST", + Host: "test-service:8000", + Path: "/test", + Headers: headers, + } +} + +// traceparent builds a header carrier naming traceID/spanID as the wire parent. +func traceparent(traceID, spanID string) http.Header { + h := http.Header{} + h.Set("traceparent", "00-"+traceID+"-"+spanID+"-01") + return h +} + +// extractParent decodes the span context named by the headers' traceparent. +func extractParent(h http.Header) trace.SpanContext { + ctx := propagation.TraceContext{}.Extract(context.Background(), propagation.HeaderCarrier(h)) + return trace.SpanContextFromContext(ctx) +} + +// roleSplit returns the request and response spans from an exported set, +// asserting exactly one of each. +func roleSplit(t *testing.T, spans tracetest.SpanStubs) (req, resp tracetest.SpanStub) { + t.Helper() + var gotReq, gotResp bool + for _, s := range spans { + switch attrStr(s, "lineage.role") { + case "request": + if gotReq { + t.Fatal("more than one request span") + } + req, gotReq = s, true + case "response": + if gotResp { + t.Fatal("more than one response span") + } + resp, gotResp = s, true + default: + t.Fatalf("span %q has no lineage.role", s.Name) + } + } + if !gotReq || !gotResp { + t.Fatalf("want one request + one response span, got %d spans (req=%v resp=%v)", len(spans), gotReq, gotResp) + } + return req, resp +} + +// ---- identifiers, pairing, parenting ---- + +func TestExchange_TwoSpansPairedAndParented(t *testing.T) { + p, exp := newTestPlugin(t) + pctx := fakeContext(pipeline.Inbound, http.Header{}) + + pl, err := pipeline.New([]pipeline.Plugin{p}) + if err != nil { + t.Fatalf("pipeline.New: %v", err) + } + + // Emit on sight: the request span exists after the request pass, before finish. + pl.Run(context.Background(), pctx) + if got := len(exp.GetSpans()); got != 1 { + t.Fatalf("after request pass: want 1 span (request), got %d", got) + } + + pl.RunFinish(context.Background(), pctx, allow(200)) + spans := exp.GetSpans() + if len(spans) != 2 { + t.Fatalf("after finish: want 2 spans, got %d", len(spans)) + } + req, resp := roleSplit(t, spans) + + // exchange.id == request span id, echoed on both. + wantID := req.SpanContext.SpanID().String() + if got := attrStr(req, "lineage.exchange.id"); got != wantID { + t.Errorf("request exchange.id = %q, want %q", got, wantID) + } + if got := attrStr(resp, "lineage.exchange.id"); got != wantID { + t.Errorf("response exchange.id = %q, want %q", got, wantID) + } + + // Response span's parent is the request span (same trace). + if resp.Parent.SpanID() != req.SpanContext.SpanID() { + t.Errorf("response parent span = %s, want request span %s", resp.Parent.SpanID(), req.SpanContext.SpanID()) + } + if resp.SpanContext.TraceID() != req.SpanContext.TraceID() { + t.Errorf("response trace = %s, want request trace %s", resp.SpanContext.TraceID(), req.SpanContext.TraceID()) + } + + // Both spans share the same SpanKind (SERVER for inbound). + if req.SpanKind != trace.SpanKindServer || resp.SpanKind != trace.SpanKindServer { + t.Errorf("span kinds = %v/%v, want server/server", req.SpanKind, resp.SpanKind) + } +} + +// ---- the stamp (single-channel parenting, wire contract v1.5) ---- + +// TestStamp_OutboundRewritesStampNotTraceparent: the outbound reads its +// parent from the inbound's stamp, then re-stamps the forwarded tracestate +// with its OWN request span id for the peer sidecar's inbound to read. The +// forwarded traceparent is NOT modified (v1.5 removed the splice) — an app +// chain riding traceparent toward its own backend stays intact. +func TestStamp_OutboundRewritesStampNotTraceparent(t *testing.T) { + p, exp := newTestPlugin(t) + const traceID, wireParent = "4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7" + const inboundID = "1111111111111111" + h := traceparent(traceID, wireParent) + h.Set("tracestate", tracestateStampKey+"="+inboundID) + pctx := fakeContext(pipeline.Outbound, h) + pctx.Extensions.MCP = &pipeline.MCPExtension{Method: "tools/call", Params: map[string]any{"name": "get_weather"}} + + run(t, p, pctx, allow(200)) + + req, _ := roleSplit(t, exp.GetSpans()) + // Parent comes from the inbound's stamp. + if got := req.Parent.SpanID().String(); got != inboundID { + t.Errorf("parent = %s, want stamped inbound %s", got, inboundID) + } + // The forwarded traceparent is untouched — still the wire parent. + forwarded := extractParent(pctx.Headers) + if got := forwarded.SpanID().String(); got != wireParent { + t.Errorf("forwarded traceparent parent = %s, want untouched wire parent %s", got, wireParent) + } + if got := forwarded.TraceID().String(); got != traceID { + t.Errorf("forwarded trace = %s, want %s", got, traceID) + } + // The forwarded tracestate now stamps THIS outbound request span, + // replacing the inbound's stamp it consumed. + want := tracestateStampKey + "=" + req.SpanContext.SpanID().String() + if got := pctx.Headers.Get("tracestate"); got != want { + t.Errorf("tracestate = %q, want re-stamp %q", got, want) + } +} + +// TestStamp_InboundParentsOnPeerStamp is the cross-pod link: the caller +// sidecar's outbound stamped tracestate with its request span id, and this +// inbound must parent on that stamp — not on the wire traceparent, whose +// span id may belong to an app chain this pipeline never exports. +func TestStamp_InboundParentsOnPeerStamp(t *testing.T) { + p, exp := newTestPlugin(t) + const traceID, wireParent = "4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7" + const peerOutbound = "2222222222222222" + h := traceparent(traceID, wireParent) + h.Set("tracestate", tracestateStampKey+"="+peerOutbound) + pctx := fakeContext(pipeline.Inbound, h) + + run(t, p, pctx, allow(200)) + + req, _ := roleSplit(t, exp.GetSpans()) + if got := req.Parent.SpanID().String(); got != peerOutbound { + t.Errorf("parent = %s, want peer outbound stamp %s", got, peerOutbound) + } + if got := attrStr(req, "lineage.parent.source"); got != "tracestate" { + t.Errorf("lineage.parent.source = %q, want tracestate", got) + } + // The forwarded stamp now names THIS inbound request span — the app's + // shim couriers it to exactly the outbound calls this inbound causes. + want := tracestateStampKey + "=" + req.SpanContext.SpanID().String() + if got := pctx.Headers.Get("tracestate"); got != want { + t.Errorf("tracestate = %q, want re-stamp %q", got, want) + } +} + +func TestStamp_InboundHeadersUntouchedExceptStamp(t *testing.T) { + p, exp := newTestPlugin(t) + h := traceparent("4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7") + before := http.Header{} + maps.Copy(before, h) + pctx := fakeContext(pipeline.Inbound, h) + + run(t, p, pctx, allow(200)) + + // The ONLY inbound mutation is the tracestate stamp; traceparent and + // everything else are forwarded as they arrived. + req, _ := roleSplit(t, exp.GetSpans()) + want := tracestateStampKey + "=" + req.SpanContext.SpanID().String() + if got := pctx.Headers.Get("tracestate"); got != want { + t.Errorf("tracestate = %q, want stamp %q", got, want) + } + after := http.Header{} + maps.Copy(after, pctx.Headers) + after.Del("tracestate") + if !headersEqual(before, after) { + t.Errorf("inbound headers beyond tracestate mutated: before=%v after=%v", before, after) + } + // No stamp arrived, so the parent is the wire traceparent — recorded as such. + if got := attrStr(req, "lineage.parent.source"); got != "wire" { + t.Errorf("lineage.parent.source = %q, want wire", got) + } +} + +func TestStamp_PreservesForeignTracestateMembers(t *testing.T) { + p, exp := newTestPlugin(t) + h := traceparent("4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7") + h.Set("tracestate", "vendor=abc") + pctx := fakeContext(pipeline.Inbound, h) + + run(t, p, pctx, allow(200)) + + req, _ := roleSplit(t, exp.GetSpans()) + got := pctx.Headers.Get("tracestate") + wantStamp := tracestateStampKey + "=" + req.SpanContext.SpanID().String() + if !strings.Contains(got, wantStamp) || !strings.Contains(got, "vendor=abc") { + t.Errorf("tracestate = %q, want both %q and vendor=abc", got, wantStamp) + } +} + +func TestStamp_NoWireTraceparentNoStamp(t *testing.T) { + p, _ := newTestPlugin(t) + pctx := fakeContext(pipeline.Inbound, http.Header{}) + + run(t, p, pctx, allow(200)) + + if got := pctx.Headers.Get("tracestate"); got != "" { + t.Errorf("tracestate stamped without a wire traceparent: %q", got) + } +} + +// TestStamp_OutboundPrefersStampOverMap is the same-trace fan-in case in +// miniature: two concurrent inbound exchanges on ONE trace (the trace-keyed +// map can only hold the later one), then an outbound whose tracestate stamp +// names the EARLIER inbound. Without the stamp this outbound would collapse +// onto the map entry — the 1/N misattribution the fanin-test.sh e2e proves. +func TestStamp_OutboundUsesTheStampedInbound(t *testing.T) { + p, exp := newTestPlugin(t) + const traceID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + + // Two concurrent inbounds on the SAME trace — the case no trace-keyed + // structure can disambiguate, and the reason the stamp exists. + run(t, p, fakeContext(pipeline.Inbound, traceparent(traceID, "1111111111111111")), allow(200)) + in1, _ := roleSplit(t, exp.GetSpans()) + exp.Reset() + run(t, p, fakeContext(pipeline.Inbound, traceparent(traceID, "2222222222222222")), allow(200)) + in2, _ := roleSplit(t, exp.GetSpans()) + + // Outbound couriered in1's stamp back through the app. It must parent + // under in1 specifically — not in2, not the wire parent. + exp.Reset() + h := traceparent(traceID, "3333333333333333") + h.Set("tracestate", tracestateStampKey+"="+in1.SpanContext.SpanID().String()) + run(t, p, fakeContext(pipeline.Outbound, h), allow(200)) + outReq, _ := roleSplit(t, exp.GetSpans()) + + if outReq.Parent.SpanID() != in1.SpanContext.SpanID() { + t.Errorf("parent = %s, want stamped inbound %s (the other in-flight inbound was %s)", + outReq.Parent.SpanID(), in1.SpanContext.SpanID(), in2.SpanContext.SpanID()) + } + if got := attrStr(outReq, "lineage.parent.source"); got != "tracestate" { + t.Errorf("lineage.parent.source = %q, want tracestate", got) + } +} + +func TestStamp_MalformedFallsBackToWire(t *testing.T) { + p, exp := newTestPlugin(t) + const traceID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + const wireParent = "3333333333333333" + + // An inbound on this trace exists — and must NOT be used, because a + // malformed stamp means "unknown", not "guess for me". + run(t, p, fakeContext(pipeline.Inbound, traceparent(traceID, "1111111111111111")), allow(200)) + in1, _ := roleSplit(t, exp.GetSpans()) + + exp.Reset() + h := traceparent(traceID, wireParent) + h.Set("tracestate", tracestateStampKey+"=nothex") + run(t, p, fakeContext(pipeline.Outbound, h), allow(200)) + outReq, _ := roleSplit(t, exp.GetSpans()) + + if got := outReq.Parent.SpanID().String(); got != wireParent { + t.Errorf("parent = %s, want wire parent %s", got, wireParent) + } + if outReq.Parent.SpanID() == in1.SpanContext.SpanID() { + t.Error("malformed stamp silently inherited this pod's inbound span") + } + if got := attrStr(outReq, "lineage.parent.source"); got != "wire" { + t.Errorf("lineage.parent.source = %q, want wire", got) + } +} + +func TestStamp_ParentSourceWireWhenUnstamped(t *testing.T) { + p, exp := newTestPlugin(t) + out := fakeContext(pipeline.Outbound, traceparent("cccccccccccccccccccccccccccccccc", "1111111111111111")) + run(t, p, out, allow(200)) + outReq, _ := roleSplit(t, exp.GetSpans()) + if got := attrStr(outReq, "lineage.parent.source"); got != "wire" { + t.Errorf("lineage.parent.source = %q, want wire", got) + } +} + +// TestStamp_UnstampedOutboundNeverInheritsInbound is the regression guard for +// the removal of the trace-keyed map. An outbound with no stamp must fall to the +// wire parent EVEN WHEN this pod has an inbound span for the same trace. The old +// map answered such cases from "the last inbound seen", which is correct only +// while exactly one inbound is in flight — a precondition it never checked. A +// missing edge is recoverable; a confidently wrong one is not. +func TestStamp_UnstampedOutboundNeverInheritsInbound(t *testing.T) { + p, exp := newTestPlugin(t) + const traceID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + const wireParent = "3333333333333333" + + run(t, p, fakeContext(pipeline.Inbound, traceparent(traceID, "2222222222222222")), allow(200)) + inReq, _ := roleSplit(t, exp.GetSpans()) + + exp.Reset() + run(t, p, fakeContext(pipeline.Outbound, traceparent(traceID, wireParent)), allow(200)) + outReq, _ := roleSplit(t, exp.GetSpans()) + + if outReq.Parent.SpanID() == inReq.SpanContext.SpanID() { + t.Fatal("un-stamped outbound inherited this pod's inbound span — the map is back") + } + if got := outReq.Parent.SpanID().String(); got != wireParent { + t.Errorf("parent = %s, want wire parent %s", got, wireParent) + } + if got := attrStr(outReq, "lineage.parent.source"); got != "wire" { + t.Errorf("lineage.parent.source = %q, want wire", got) + } +} + +func TestStamp_ConcurrentTracesNeverCross(t *testing.T) { + p, exp := newTestPlugin(t) + const traceA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + const traceB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + + // One inbound on each trace. + run(t, p, fakeContext(pipeline.Inbound, traceparent(traceA, "1111111111111111")), allow(200)) + inA, _ := roleSplit(t, exp.GetSpans()) + exp.Reset() + run(t, p, fakeContext(pipeline.Inbound, traceparent(traceB, "2222222222222222")), allow(200)) + inB, _ := roleSplit(t, exp.GetSpans()) + + // Each outbound couriers its own trace's stamp back. + exp.Reset() + hA := traceparent(traceA, "3333333333333333") + hA.Set("tracestate", tracestateStampKey+"="+inA.SpanContext.SpanID().String()) + run(t, p, fakeContext(pipeline.Outbound, hA), allow(200)) + outA, _ := roleSplit(t, exp.GetSpans()) + if outA.Parent.SpanID() != inA.SpanContext.SpanID() { + t.Errorf("outbound A parent = %s, want inbound A %s", outA.Parent.SpanID(), inA.SpanContext.SpanID()) + } + if outA.Parent.SpanID() == inB.SpanContext.SpanID() { + t.Error("outbound A crossed into inbound B's span") + } + if outA.SpanContext.TraceID().String() != traceA { + t.Errorf("outbound A trace = %s, want %s", outA.SpanContext.TraceID(), traceA) + } + + exp.Reset() + hB := traceparent(traceB, "4444444444444444") + hB.Set("tracestate", tracestateStampKey+"="+inB.SpanContext.SpanID().String()) + run(t, p, fakeContext(pipeline.Outbound, hB), allow(200)) + outB, _ := roleSplit(t, exp.GetSpans()) + if outB.Parent.SpanID() != inB.SpanContext.SpanID() { + t.Errorf("outbound B parent = %s, want inbound B %s", outB.Parent.SpanID(), inB.SpanContext.SpanID()) + } +} + +// ---- bodyless / unparsed completeness ---- + +func TestBodyless_UnparsedNoCaptureStillEmitsBothSpans(t *testing.T) { + p, exp := newTestPlugin(t) + // capture_io defaults false; no parser extensions → protocol http. + pctx := fakeContext(pipeline.Outbound, http.Header{}) + + run(t, p, pctx, allow(200)) + + req, resp := roleSplit(t, exp.GetSpans()) + if got := attrStr(req, "lineage.protocol"); got != "http" { + t.Errorf("protocol = %q, want http", got) + } + // Complete: both carry the shared facts and the exchange is paired. + if attrStr(req, "lineage.exchange.id") == "" || attrStr(resp, "lineage.exchange.id") == "" { + t.Error("exchange.id missing on a bodyless span") + } + if got := attrStr(resp, "lineage.outcome"); got != "ok" { + t.Errorf("outcome = %q, want ok", got) + } + // No payloads captured. + if _, ok := findAttr(req, "input.value"); ok { + t.Error("input.value present with capture_io off") + } + if _, ok := findAttr(resp, "output.value"); ok { + t.Error("output.value present with capture_io off") + } +} + +// ---- outcomes ---- + +func TestOutcome_Denied(t *testing.T) { + p, exp := newTestPlugin(t) + pctx := fakeContext(pipeline.Inbound, http.Header{}) + + run(t, p, pctx, pipeline.Outcome{ + FinalAction: pipeline.OutcomeDeny, + StatusCode: 401, + DenyingPlugin: "jwt-validation", + }) + + _, resp := roleSplit(t, exp.GetSpans()) + if got := attrStr(resp, "lineage.outcome"); got != "denied" { + t.Errorf("outcome = %q, want denied", got) + } + if got := attrStr(resp, "lineage.denied_by"); got != "jwt-validation" { + t.Errorf("denied_by = %q, want jwt-validation", got) + } + if got, ok := intAttr(resp, "http.status_code"); !ok || got != 401 { + t.Errorf("http.status_code = %d (ok=%v), want 401", got, ok) + } +} + +func TestOutcome_AbandonedHasNoStatus(t *testing.T) { + p, exp := newTestPlugin(t) + pctx := fakeContext(pipeline.Outbound, http.Header{}) + + // Terminal error with no response written (upstream reset / disconnect). + run(t, p, pctx, pipeline.Outcome{FinalAction: pipeline.OutcomeError, StatusCode: 0}) + + _, resp := roleSplit(t, exp.GetSpans()) + if got := attrStr(resp, "lineage.outcome"); got != "abandoned" { + t.Errorf("outcome = %q, want abandoned", got) + } + if _, ok := findAttr(resp, "http.status_code"); ok { + t.Error("http.status_code present on an abandoned exchange (none was produced)") + } +} + +// ---- request facts + capture_io + span names ---- + +func TestRequestFacts_MCPWithCapture(t *testing.T) { + p, exp := newTestPlugin(t) + p.cfg.CaptureIO = true + pctx := fakeContext(pipeline.Outbound, http.Header{}) + pctx.Host = "weather-tool-mcp.team1.svc:8000" + pctx.Path = "/mcp" + pctx.Scheme = "http" + pctx.Extensions.MCP = &pipeline.MCPExtension{ + Method: "tools/call", + Params: map[string]any{"name": "get_weather", "arguments": map[string]any{"city": "Tokyo"}}, + Result: map[string]any{"content": []any{map[string]any{"type": "text", "text": "sunny"}}}, + } + + run(t, p, pctx, allow(200)) + req, resp := roleSplit(t, exp.GetSpans()) + + checkAttr(t, req, "lineage.protocol", "mcp") + checkAttr(t, req, "mcp.method", "tools/call") + checkAttr(t, req, "mcp.tool", "get_weather") + checkAttr(t, req, "http.method", "POST") + checkAttr(t, req, "url.path", "/mcp") + checkAttr(t, req, "url.scheme", "http") + checkAttr(t, req, "lineage.self.id", "weather-service") + checkAttr(t, req, "lineage.peer.host", "weather-tool-mcp.team1.svc:8000") + checkAttr(t, req, "lineage.direction", "outbound") + checkAttr(t, req, "input.value", `{"city":"Tokyo"}`) + checkAttr(t, resp, "output.value", "sunny") + + if req.Name != "weather-service mcp get_weather" { + t.Errorf("request span name = %q", req.Name) + } + if resp.Name != "weather-service mcp get_weather response" { + t.Errorf("response span name = %q", resp.Name) + } + if req.SpanKind != trace.SpanKindClient { + t.Errorf("outbound request kind = %v, want client", req.SpanKind) + } +} + +func TestRequestFacts_A2AAndInference(t *testing.T) { + p, exp := newTestPlugin(t) + // A2A. + a := fakeContext(pipeline.Outbound, http.Header{}) + a.Extensions.A2A = &pipeline.A2AExtension{Method: "message/send", SessionID: "sess-123"} + run(t, p, a, allow(200)) + areq, _ := roleSplit(t, exp.GetSpans()) + checkAttr(t, areq, "lineage.protocol", "a2a") + checkAttr(t, areq, "a2a.method", "message/send") + checkAttr(t, areq, "a2a.session_id", "sess-123") + if _, ok := findAttr(areq, "url.scheme"); ok { + t.Error("url.scheme present although the context carried no scheme") + } + if areq.Name != "weather-service a2a message/send" { + t.Errorf("a2a span name = %q", areq.Name) + } + + // Inference. + exp.Reset() + i := fakeContext(pipeline.Outbound, http.Header{}) + i.Extensions.Inference = &pipeline.InferenceExtension{Model: "qwen2.5:7b"} + run(t, p, i, allow(200)) + ireq, _ := roleSplit(t, exp.GetSpans()) + checkAttr(t, ireq, "lineage.protocol", "inference") + checkAttr(t, ireq, "inference.model", "qwen2.5:7b") + if ireq.Name != "weather-service inference qwen2.5:7b" { + t.Errorf("inference span name = %q", ireq.Name) + } +} + +// mcp-parser attaches to ANY JSON-RPC body — including every a2a exchange — +// so on an a2a hop both extensions are populated. The payload read is keyed by +// the protocol fact: when the a2a parser yields nothing (no text parts, a +// protocol-event artifact), the payload stays ABSENT rather than falling +// through to the co-populated MCP parse of the same bytes (which would emit +// the raw JSON-RPC envelope on an lineage.protocol=a2a span). +func TestCaptureIO_A2ANeverFallsThroughToCoPopulatedMCP(t *testing.T) { + p, exp := newTestPlugin(t) + p.cfg.CaptureIO = true + pctx := fakeContext(pipeline.Outbound, http.Header{}) + pctx.Extensions.A2A = &pipeline.A2AExtension{ + Method: "message/send", + // A status-update captured as the artifact — a protocol event, filtered. + Artifact: `{"kind":"status-update","taskId":"t-1"}`, + } + pctx.Extensions.MCP = &pipeline.MCPExtension{ + Method: "message/send", + Params: map[string]any{"message": map[string]any{"role": "user"}}, + Result: map[string]any{"artifacts": []any{map[string]any{"artifactId": "a-1"}}}, + } + + run(t, p, pctx, allow(200)) + req, resp := roleSplit(t, exp.GetSpans()) + + checkAttr(t, req, "lineage.protocol", "a2a") + if v, ok := findAttr(req, "input.value"); ok { + t.Errorf("input.value = %q on an a2a hop with no a2a parts — leaked from the co-populated MCP parse", v.Emit()) + } + if v, ok := findAttr(resp, "output.value"); ok { + t.Errorf("output.value = %q on an a2a hop whose artifact is a protocol event — leaked from the co-populated MCP parse", v.Emit()) + } + // mcp.* facts belong to mcp hops only; the a2a label must keep them off. + if v, ok := findAttr(req, "mcp.method"); ok { + t.Errorf("mcp.method = %q emitted on an a2a hop", v.Emit()) + } +} + +func TestPrincipalFacts_InboundRequestOnly(t *testing.T) { + p, exp := newTestPlugin(t) + pctx := fakeContext(pipeline.Inbound, http.Header{}) + pctx.Identity = fakeIdentity{sub: "alice", client: "weather-ui"} + + run(t, p, pctx, allow(200)) + req, resp := roleSplit(t, exp.GetSpans()) + + checkAttr(t, req, "lineage.principal.sub", "alice") + checkAttr(t, req, "lineage.principal.client", "weather-ui") + // Principal facts are request-only. + if _, ok := findAttr(resp, "lineage.principal.sub"); ok { + t.Error("lineage.principal.sub leaked onto the response span") + } +} + +func TestPrincipalFacts_OutboundNeverEmitsPrincipal(t *testing.T) { + p, exp := newTestPlugin(t) + pctx := fakeContext(pipeline.Outbound, http.Header{}) + pctx.Identity = fakeIdentity{sub: "alice", client: "weather-ui"} + + run(t, p, pctx, allow(200)) + req, _ := roleSplit(t, exp.GetSpans()) + if _, ok := findAttr(req, "lineage.principal.sub"); ok { + t.Error("outbound span carried a principal fact (inbound-only)") + } +} + +// ---- the forbidden-keys guard ---- + +// TestForbiddenKeysNeverEmitted scans every attribute of every span emitted +// across a spread of exchange shapes and asserts none carries a key from a +// removed vocabulary. The contract deleted these; this test is the tripwire +// that keeps them deleted. +func TestForbiddenKeysNeverEmitted(t *testing.T) { + forbidden := []string{"trust.", "lineage.hop.kind", "enduser.id", "openinference.", "source", "authbridge.proxy"} + + shapes := []func() *pipeline.Context{ + func() *pipeline.Context { + c := fakeContext(pipeline.Inbound, http.Header{}) + c.Identity = fakeIdentity{sub: "alice", client: "weather-ui"} + return c + }, + func() *pipeline.Context { + c := fakeContext(pipeline.Outbound, http.Header{}) + c.Extensions.MCP = &pipeline.MCPExtension{Method: "tools/call", Params: map[string]any{"name": "get_weather"}} + return c + }, + func() *pipeline.Context { + c := fakeContext(pipeline.Outbound, http.Header{}) + c.Extensions.A2A = &pipeline.A2AExtension{Method: "message/send"} + return c + }, + func() *pipeline.Context { + c := fakeContext(pipeline.Outbound, http.Header{}) + c.Extensions.Inference = &pipeline.InferenceExtension{Model: "qwen2.5:7b"} + return c + }, + } + + for _, mk := range shapes { + p, exp := newTestPlugin(t) + p.cfg.CaptureIO = true + run(t, p, mk(), allow(200)) + for _, s := range exp.GetSpans() { + for _, kv := range s.Attributes { + key := string(kv.Key) + for _, bad := range forbidden { + if key == bad || strings.HasPrefix(key, bad) { + t.Errorf("span %q emitted forbidden attribute %q", s.Name, key) + } + } + } + } + } +} + +// ---- robustness ---- + +func TestOnFinish_NoStateDoesNotPanicOrEmit(t *testing.T) { + p, exp := newTestPlugin(t) + pctx := fakeContext(pipeline.Inbound, http.Header{}) + // OnFinish without OnRequest having run — no exchangeState stored. + p.OnFinish(context.Background(), pctx) + if got := len(exp.GetSpans()); got != 0 { + t.Errorf("OnFinish with no state emitted %d spans, want 0", got) + } +} + +func TestNotReady_SkipsSpan(t *testing.T) { + p := NewLineageTelemetry() + // Do NOT set ready — Init never called. + pctx := fakeContext(pipeline.Inbound, http.Header{}) + action := p.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Continue { + t.Fatalf("expected Continue, got %v", action.Type) + } + if pipeline.GetState[exchangeState](pctx, pluginName) != nil { + t.Error("exchangeState should not be set when plugin is not ready") + } +} + +// ---- config ---- + +func TestConfigure_Defaults(t *testing.T) { + p := NewLineageTelemetry() + if err := p.Configure(nil); err != nil { + t.Fatalf("Configure(nil): %v", err) + } + if p.cfg.OTelEndpoint != "localhost:4317" { + t.Errorf("default endpoint = %q, want localhost:4317", p.cfg.OTelEndpoint) + } + if p.cfg.SelfIDFile != "/shared/client-id.txt" { + t.Errorf("default self_id_file = %q", p.cfg.SelfIDFile) + } +} + +func TestConfigure_DecodesKeptKeys(t *testing.T) { + p := NewLineageTelemetry() + raw := json.RawMessage(`{"otel_endpoint":"http://collector:4317","capture_io":true,"self_id":"weather-service"}`) + if err := p.Configure(raw); err != nil { + t.Fatalf("Configure: %v", err) + } + if p.cfg.OTelEndpoint != "collector:4317" { + t.Errorf("endpoint = %q, want collector:4317 (scheme stripped)", p.cfg.OTelEndpoint) + } + if !p.cfg.CaptureIO { + t.Error("capture_io should be true") + } + if p.cfg.SelfID != "weather-service" { + t.Errorf("self_id = %q", p.cfg.SelfID) + } +} + +// ---- helpers ---- + +type fakeIdentity struct { + sub, client string + scopes []string +} + +func (f fakeIdentity) Subject() string { return f.sub } +func (f fakeIdentity) ClientID() string { return f.client } +func (f fakeIdentity) Scopes() []string { return f.scopes } + +func findAttr(span tracetest.SpanStub, key string) (attribute.Value, bool) { + for _, kv := range span.Attributes { + if string(kv.Key) == key { + return kv.Value, true + } + } + return attribute.Value{}, false +} + +func attrStr(span tracetest.SpanStub, key string) string { + if v, ok := findAttr(span, key); ok { + return v.AsString() + } + return "" +} + +func intAttr(span tracetest.SpanStub, key string) (int64, bool) { + if v, ok := findAttr(span, key); ok { + return v.AsInt64(), true + } + return 0, false +} + +// checkAttr asserts a span contains attribute key with the given string value. +func checkAttr(t *testing.T, span tracetest.SpanStub, key, want string) { + t.Helper() + got, ok := findAttr(span, key) + if !ok { + t.Errorf("attribute %q not found in span %q", key, span.Name) + return + } + if got.AsString() != want { + t.Errorf("attr %q = %q, want %q", key, got.AsString(), want) + } +} + +func headersEqual(a, b http.Header) bool { + if len(a) != len(b) { + return false + } + for k, av := range a { + bv, ok := b[k] + if !ok || len(av) != len(bv) { + return false + } + for i := range av { + if av[i] != bv[i] { + return false + } + } + } + return true +} + +// TestInit_RefusesToStartWithoutIdentity locks the v1.3 rule at the identity +// boundary: a pod whose self identity cannot be resolved must fail at boot, +// never serve traffic under a plausible-but-wrong label (the old behavior +// emitted lineage.self.id="agent" from the empty-string serviceLabel). +func TestInit_RefusesToStartWithoutIdentity(t *testing.T) { + cases := []struct { + name string + cfg Config + wantErr bool + }{ + {"inline self_id starts", Config{OTelEndpoint: "localhost:4317", SelfID: "weather-service"}, false}, + {"missing self_id_file refuses", Config{OTelEndpoint: "localhost:4317", SelfIDFile: t.TempDir() + "/absent.txt"}, true}, + {"no identity source refuses", Config{OTelEndpoint: "localhost:4317"}, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + p := NewLineageTelemetry() + p.cfg = tc.cfg + err := p.Init(context.Background()) + if p.tp != nil { + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + _ = p.tp.Shutdown(ctx) + cancel() + } + if tc.wantErr && err == nil { + t.Fatal("Init succeeded without a resolvable identity") + } + if !tc.wantErr && err != nil { + t.Fatalf("Init failed with a valid inline self_id: %v", err) + } + if tc.wantErr && p.Ready() { + t.Error("plugin reports Ready after a refused Init") + } + }) + } +} + +// TestInit_ReadsSelfIDFile covers the operator-injected path (file, not inline). +func TestInit_ReadsSelfIDFile(t *testing.T) { + dir := t.TempDir() + path := dir + "/client-id.txt" + if err := os.WriteFile(path, []byte("weather-service\n"), 0o600); err != nil { + t.Fatal(err) + } + p := NewLineageTelemetry() + p.cfg = Config{OTelEndpoint: "localhost:4317", SelfIDFile: path} + if err := p.Init(context.Background()); err != nil { + t.Fatalf("Init: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + _ = p.tp.Shutdown(ctx) + cancel() + if p.selfID != "weather-service" { + t.Errorf("selfID = %q, want trimmed file content", p.selfID) + } +} + +// TestConfig_UnknownKeysRefused: a typo'd knob must be a boot error, not a +// silent run-with-defaults. +func TestConfig_UnknownKeysRefused(t *testing.T) { + if _, err := decodeConfig([]byte(`{"capture-io": true}`)); err == nil { + t.Fatal("unknown config key accepted silently") + } + if _, err := decodeConfig([]byte(`{"capture_io": true, "self_id": "x"}`)); err != nil { + t.Fatalf("valid config rejected: %v", err) + } +} + +// ---- bypass config ---- +// The one failure mode of bypass_paths / bypass_hosts produces NO signal +// anywhere: a matched hop is simply absent from the graph. So both directions +// are pinned — a match emits nothing, a near-miss emits the full pair. + +func TestBypassPaths_PrefixMatchEmitsNothing(t *testing.T) { + cases := []struct { + name string + path string + spans int // spans expected from the exchange + }{ + {"prefix match skipped", "/health/live", 0}, + {"exact prefix skipped", "/health", 0}, + {"non-matching path emits", "/api/health-report", 2}, + {"prefix is anchored, not substring", "/v1/health", 2}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + p, exp := newTestPlugin(t) + p.cfg.BypassPaths = []string{"/health"} + pctx := fakeContext(pipeline.Inbound, http.Header{}) + pctx.Path = tc.path + run(t, p, pctx, allow(200)) + if got := len(exp.GetSpans()); got != tc.spans { + t.Fatalf("path %q: got %d spans, want %d", tc.path, got, tc.spans) + } + }) + } +} + +func TestBypassHosts_SubstringMatchEmitsNothing(t *testing.T) { + cases := []struct { + name string + host string + spans int + }{ + {"substring match skipped", "otel-collector.rossoctl-system:4317", 0}, + {"bare name match skipped", "otel-collector:4317", 0}, + {"unrelated host emits", "weather-tool:8000", 2}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + p, exp := newTestPlugin(t) + p.cfg.BypassHosts = []string{"otel-collector"} + pctx := fakeContext(pipeline.Outbound, http.Header{}) + pctx.Host = tc.host + run(t, p, pctx, allow(200)) + if got := len(exp.GetSpans()); got != tc.spans { + t.Fatalf("host %q: got %d spans, want %d", tc.host, got, tc.spans) + } + }) + } +} From d43523f2e851f33e6b0d13b0ec0e8fcc735b05e8 Mon Sep 17 00:00:00 2001 From: YehoshuaSagron Date: Fri, 14 Aug 2026 19:49:24 +0300 Subject: [PATCH 2/2] Feat: Register lineage-telemetry behind an exclude_plugin build tag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follows the one-tag-file-per-plugin convention: five lines per binary in plugins_lineage.go, gated by //go:build !exclude_plugin_lineage, so main.go imports no plugin package directly. A build carrying the exclude tags links neither the plugin nor its OTel dependency subtree. go.mod changes are go mod tidy output. Four direct dependencies, three of them promotions of modules already present as indirect (otel, otel/sdk, otel/trace); the fourth is the OTLP/gRPC trace exporter. Five new indirect. Licences are Apache-2.0 for the OpenTelemetry modules and genproto, MIT for backoff/v5, BSD-3-Clause for grpc-gateway/v2. No go.sum change is needed — the existing sums already cover these modules. Signed-off-by: YehoshuaSagron --- authbridge/authlib/go.mod | 12 +++++++++--- authbridge/cmd/authbridge-envoy/go.mod | 6 ++++++ authbridge/cmd/authbridge-envoy/plugins_lineage.go | 5 +++++ authbridge/cmd/authbridge-proxy/go.mod | 6 ++++++ authbridge/cmd/authbridge-proxy/plugins_lineage.go | 5 +++++ 5 files changed, 31 insertions(+), 3 deletions(-) create mode 100644 authbridge/cmd/authbridge-envoy/plugins_lineage.go create mode 100644 authbridge/cmd/authbridge-proxy/plugins_lineage.go diff --git a/authbridge/authlib/go.mod b/authbridge/authlib/go.mod index a2ca3e191..90c3d0a2a 100644 --- a/authbridge/authlib/go.mod +++ b/authbridge/authlib/go.mod @@ -12,6 +12,10 @@ require ( github.com/open-policy-agent/opa v1.18.2 github.com/rossoctl/context-guru v0.0.0-20260720181432-8fc7c7b36563 github.com/spiffe/go-spiffe/v2 v2.8.1 + go.opentelemetry.io/otel v1.44.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 + go.opentelemetry.io/otel/sdk v1.44.0 + go.opentelemetry.io/otel/trace v1.44.0 golang.org/x/net v0.57.0 golang.org/x/sys v0.47.0 google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa @@ -45,6 +49,7 @@ require ( github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic v1.15.1 // indirect github.com/bytedance/sonic/loader v0.5.1 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 // indirect @@ -56,6 +61,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/goccy/go-json v0.10.6 // indirect github.com/google/uuid v1.6.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect github.com/huandu/go-clone v1.7.3 // indirect github.com/huandu/go-sqlbuilder v1.41.0 // indirect @@ -108,10 +114,9 @@ require ( github.com/yashtewari/glob-intersection v0.2.0 // indirect github.com/yosida95/uritemplate/v3 v3.0.2 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect - go.opentelemetry.io/otel/sdk v1.44.0 // indirect - go.opentelemetry.io/otel/trace v1.44.0 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.starlark.net v0.0.0-20260708150628-5395d018f003 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect @@ -120,6 +125,7 @@ require ( golang.org/x/sync v0.22.0 // indirect golang.org/x/text v0.40.0 // indirect golang.org/x/time v0.15.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect gopkg.in/ini.v1 v1.67.3 // indirect oras.land/oras-go/v2 v2.6.2 // indirect diff --git a/authbridge/cmd/authbridge-envoy/go.mod b/authbridge/cmd/authbridge-envoy/go.mod index 066430ba1..823d7f6e8 100644 --- a/authbridge/cmd/authbridge-envoy/go.mod +++ b/authbridge/cmd/authbridge-envoy/go.mod @@ -40,6 +40,7 @@ require ( github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic v1.15.1 // indirect github.com/bytedance/sonic/loader v0.5.1 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 // indirect @@ -54,6 +55,7 @@ require ( github.com/gobwas/glob v0.2.3 // indirect github.com/goccy/go-json v0.10.6 // indirect github.com/google/uuid v1.6.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect github.com/huandu/go-clone v1.7.3 // indirect github.com/huandu/go-sqlbuilder v1.41.0 // indirect @@ -111,9 +113,12 @@ require ( github.com/yosida95/uritemplate/v3 v3.0.2 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect go.opentelemetry.io/otel/sdk v1.44.0 // indirect go.opentelemetry.io/otel/trace v1.44.0 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.starlark.net v0.0.0-20260708150628-5395d018f003 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect @@ -124,6 +129,7 @@ require ( golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.40.0 // indirect golang.org/x/time v0.15.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect gopkg.in/ini.v1 v1.67.3 // indirect diff --git a/authbridge/cmd/authbridge-envoy/plugins_lineage.go b/authbridge/cmd/authbridge-envoy/plugins_lineage.go new file mode 100644 index 000000000..76ea1e644 --- /dev/null +++ b/authbridge/cmd/authbridge-envoy/plugins_lineage.go @@ -0,0 +1,5 @@ +//go:build !exclude_plugin_lineage + +package main + +import _ "github.com/rossoctl/cortex/authbridge/authlib/plugins/lineage" diff --git a/authbridge/cmd/authbridge-proxy/go.mod b/authbridge/cmd/authbridge-proxy/go.mod index 1d0a6f8f4..4a665a863 100644 --- a/authbridge/cmd/authbridge-proxy/go.mod +++ b/authbridge/cmd/authbridge-proxy/go.mod @@ -33,6 +33,7 @@ require ( github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic v1.15.1 // indirect github.com/bytedance/sonic/loader v0.5.1 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 // indirect @@ -45,6 +46,7 @@ require ( github.com/gobwas/glob v0.2.3 // indirect github.com/goccy/go-json v0.10.6 // indirect github.com/google/uuid v1.6.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect github.com/huandu/go-clone v1.7.3 // indirect github.com/huandu/go-sqlbuilder v1.41.0 // indirect @@ -101,9 +103,12 @@ require ( github.com/yosida95/uritemplate/v3 v3.0.2 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect go.opentelemetry.io/otel/sdk v1.44.0 // indirect go.opentelemetry.io/otel/trace v1.44.0 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.starlark.net v0.0.0-20260708150628-5395d018f003 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect @@ -114,6 +119,7 @@ require ( golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.40.0 // indirect golang.org/x/time v0.15.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/grpc v1.82.0 // indirect google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect diff --git a/authbridge/cmd/authbridge-proxy/plugins_lineage.go b/authbridge/cmd/authbridge-proxy/plugins_lineage.go new file mode 100644 index 000000000..76ea1e644 --- /dev/null +++ b/authbridge/cmd/authbridge-proxy/plugins_lineage.go @@ -0,0 +1,5 @@ +//go:build !exclude_plugin_lineage + +package main + +import _ "github.com/rossoctl/cortex/authbridge/authlib/plugins/lineage"