-
Notifications
You must be signed in to change notification settings - Fork 40
Feat: Lineage telemetry plugin — two facts-only spans per exchange #761
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
JoshSag
wants to merge
7
commits into
rossoctl:main
Choose a base branch
from
s-and-p-team:lane/lineage-telemetry-plugin
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
7 commits
Select commit
Hold shift + click to select a range
cf3a922
Feat: Add the lineage-telemetry plugin
JoshSag d43523f
Feat: Register lineage-telemetry behind an exclude_plugin build tag
JoshSag 0071332
Fix: Address lineage-telemetry review feedback + resolve go.mod conflict
faraway7 d4bc0a2
Merge remote-tracking branch 'origin/main' into lane/lineage-telemetr…
faraway7 4f4e31c
Fix: Address lineage-telemetry review — conn close, TLS, payload cap
faraway7 87fdac5
Fix: Address CodeRabbit follow-ups — UTF-8 truncate, scheme allowlist…
faraway7 db68cfb
Fix: Reset lineage-telemetry readiness on Shutdown
faraway7 File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Some comments aren't visible on the classic Files Changed page.
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| package lineage | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/json" | ||
| "fmt" | ||
| "net/url" | ||
| "strings" | ||
| ) | ||
|
|
||
| // defaultOTelEndpoint is the OTLP gRPC target used when otel_endpoint is unset: | ||
| // an in-pod collector reached over plaintext loopback. | ||
| const defaultOTelEndpoint = "localhost:4317" | ||
|
|
||
| // defaultMaxPayloadBytes bounds a captured input.value / output.value as a | ||
| // deliberate producer-side cap. It is NOT a mirror of any SDK limit: the OTel | ||
| // SDK's default attribute-value length limit is unlimited (-1) and Init sets no | ||
| // SpanLimits, so an oversized value is not dropped or truncated downstream. This | ||
| // bound is our own guard against unbounded spans (and against any backend value | ||
| // limit); anything longer is cut here with an explicit marker so the loss is | ||
| // visible in the span. 4096 is a conservative default, not a hard requirement. | ||
| const defaultMaxPayloadBytes = 4096 | ||
|
|
||
| // Config holds the per-plugin configuration decoded from the pipeline YAML. | ||
| type Config struct { | ||
| // OTelEndpoint is the OTLP gRPC endpoint (host:port, http://host:port, or | ||
| // https://host:port). An https:// scheme implies OTelTLS=true. Any other | ||
| // URL scheme is rejected at decode (see decodeConfig). | ||
| // Default: "localhost:4317" | ||
| OTelEndpoint string `json:"otel_endpoint"` | ||
|
|
||
| // OTelTLS selects the OTLP transport. False (the default) dials plaintext, | ||
| // which is correct for the in-pod loopback collector but sends spans — | ||
| // including lineage.principal.* on every inbound request, and full payloads | ||
| // under CaptureIO — in cleartext. Set true for any collector off-pod: it | ||
| // dials with TLS against the system root CAs. An https:// otel_endpoint | ||
| // turns this on automatically; a plaintext otel_endpoint with otel_tls:true | ||
| // is honoured (TLS to a host:port). The one rejected combination is an | ||
| // https:// endpoint with an explicit otel_tls:false (see decodeConfig): a | ||
| // contradiction that would otherwise silently downgrade to cleartext. | ||
| OTelTLS bool `json:"otel_tls"` | ||
|
|
||
| // 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"` | ||
|
|
||
| // MaxPayloadBytes caps the size of the input.value / output.value | ||
| // attributes attached under CaptureIO. A payload longer than this is cut on | ||
| // a UTF-8 boundary and suffixed with a truncation marker, so the loss is | ||
| // explicit in the span. This is a deliberate producer-side bound; the OTel | ||
| // SDK does not itself drop or truncate an oversized value (its default | ||
| // attribute-value limit is unlimited and Init sets no SpanLimits), so | ||
| // without this cap the whole payload would be emitted. Zero (or unset) uses | ||
| // defaultMaxPayloadBytes; a negative value disables the cap (attach whole). | ||
| // Ignored when CaptureIO is false. | ||
| // Default: 4096 | ||
| MaxPayloadBytes int `json:"max_payload_bytes"` | ||
|
|
||
| // 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: defaultOTelEndpoint, | ||
| MaxPayloadBytes: defaultMaxPayloadBytes, | ||
| 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 = defaultOTelEndpoint | ||
| } | ||
| // Zero means "unset" → the safe default; a negative value is the explicit | ||
| // opt-out (no cap). This keeps an omitted key and an explicit 0 identical. | ||
| if cfg.MaxPayloadBytes == 0 { | ||
| cfg.MaxPayloadBytes = defaultMaxPayloadBytes | ||
| } | ||
| // gRPC NewClient expects host:port only, so reduce a URL form (e.g. | ||
| // http://collector:4317/v1/traces) to its host — TrimPrefix left any path | ||
| // behind and produced an invalid dial target. A URL scheme also carries an | ||
| // intent about transport: https:// asks for TLS. Honour it (or fail on a | ||
| // contradiction) rather than silently dropping to cleartext. | ||
| if strings.Contains(cfg.OTelEndpoint, "://") { | ||
| u, err := url.Parse(cfg.OTelEndpoint) | ||
| if err != nil || u.Host == "" { | ||
| return Config{}, fmt.Errorf("lineage-telemetry config: invalid otel_endpoint %q", cfg.OTelEndpoint) | ||
| } | ||
| // Only http/https carry a meaningful OTLP transport intent. Reject any | ||
| // other scheme (ftp://, ftps://, …) rather than strip it and dial the | ||
| // bare host:port insecurely — that would silently send principal facts | ||
| // and payloads in cleartext. Fail closed, matching this package's | ||
| // DisallowUnknownFields / https+otel_tls:false posture. | ||
| if u.Scheme != "http" && u.Scheme != "https" { | ||
| return Config{}, fmt.Errorf("lineage-telemetry config: unsupported otel_endpoint scheme %q (want http or https)", u.Scheme) | ||
| } | ||
| if u.Scheme == "https" { | ||
| // An explicit otel_tls:false alongside an https:// endpoint is a | ||
| // contradiction: one asks for encryption, the other for cleartext. | ||
| // Fail closed rather than pick one, consistent with the | ||
| // DisallowUnknownFields fail-on-ambiguity choice this package makes. | ||
| if tlsExplicitlyFalse(raw) { | ||
| return Config{}, fmt.Errorf("lineage-telemetry config: otel_endpoint %q is https but otel_tls is false", cfg.OTelEndpoint) | ||
| } | ||
| cfg.OTelTLS = true | ||
| } | ||
| cfg.OTelEndpoint = u.Host | ||
| } | ||
| return cfg, nil | ||
| } | ||
|
|
||
| // tlsExplicitlyFalse reports whether the raw config carries otel_tls set to a | ||
| // literal false, as opposed to being absent (whose decoded value is also false | ||
| // but carries no intent). Used only to reject the https:// + otel_tls:false | ||
| // contradiction; a decode failure here is treated as "not explicitly false" | ||
| // since the DisallowUnknownFields pass above already validated the shape. | ||
| func tlsExplicitlyFalse(raw json.RawMessage) bool { | ||
| var probe struct { | ||
| OTelTLS *bool `json:"otel_tls"` | ||
| } | ||
| if err := json.Unmarshal(raw, &probe); err != nil { | ||
| return false | ||
| } | ||
| return probe.OTelTLS != nil && !*probe.OTelTLS | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.