AI-3190 feat(stream): OTLP as first-class source type with per-signal sink routing

[Matovidlo]

Summary

• Introduces SourceTypeOTLP / OTLPSource as a distinct, first-class source type (parallel to SourceTypeHTTP) with its own secret-based URL, validation, and API representation
• Adds AllowedSignals []string to sink definitions so individual sinks can be wired to only logs, metrics, or traces — a sink with an empty list accepts all signals (backwards compatible)
• Adds Signal() string to recordctx.Context so the router can filter records without inspecting the body; HTTP records return ""
• Fixes OTLP spec gaps: application/protobuf accepted as alias, Retry-After: 1 header on 429, proper HTTP status escalation for auth failures (4xx) and transient failures (5xx/429)
• Extends the dispatcher's in-memory mirror to include OTLP sources alongside HTTP sources, so data is no longer lost on restart
• Exposes the new source type through the Stream REST API (Goa DSL + regenerated artefacts)
• Integration tests using real pdata marshalers covering CORS, error cases, all three signals, gzip encoding, application/protobuf alias, per-signal routing, and partial success

Test plan

• go test ./internal/pkg/service/stream/source/type/otlpsource/... — all unit + integration tests pass
• go test ./internal/pkg/service/stream/mapping/recordctx/... — Signal() tests pass
• Verify creating a source with type: otlp via API returns an OTLP URL
• Verify a sink with allowedSignals: ["logs"] only receives log records
• Verify wrong secret returns 404, disabled source returns 404, gzip payload decodes correctly

🤖 Generated with Claude Code

[jordanrburger]

@claude - Can you fix the unit tests on this one?

[Matovidlo]

@jordanrburger it's ok there are flaky test

[Copilot]

Pull request overview

Adds OTLP as a first-class Stream source type and introduces per-signal sink routing so OTLP logs, metrics, and traces can be routed independently while preserving HTTP-source compatibility.

Changes:

• Adds OTLP source definitions, API/schema exposure, source URL formatting, dispatcher mirroring, and OTLP handler updates.
• Adds allowedSignals to sinks and routes records based on recordctx.Context.Signal().
• Adds OTLP flattening/test coverage updates, API fixtures, and a setup helper script.

Reviewed changes

Copilot reviewed 64 out of 69 changed files in this pull request and generated 18 comments.

Show a summary per file

File	Description
api/stream/design.go	Adds OTLP source and allowedSignals to Goa API design.
internal/pkg/service/stream/api/gen/http/stream/server/encode_decode.go	Regenerates HTTP encoding for OTLP/allowedSignals fields.
internal/pkg/service/stream/api/gen/http/stream/server/types.go	Regenerates HTTP request/response structs and validation.
internal/pkg/service/stream/api/gen/stream/service.go	Regenerates service API types.
internal/pkg/service/stream/api/mapper/aggregation.go	Maps OTLP URL for aggregated sources.
internal/pkg/service/stream/api/mapper/sink_allowed_signals_test.go	Adds mapper tests for sink signal filters.
internal/pkg/service/stream/api/mapper/sink_request.go	Maps allowedSignals from create/update payloads.
internal/pkg/service/stream/api/mapper/sink_response.go	Exposes allowedSignals in sink responses.
internal/pkg/service/stream/api/mapper/source_request.go	Creates/updates OTLP source entities.
internal/pkg/service/stream/api/mapper/source_response.go	Exposes OTLP source URLs.
internal/pkg/service/stream/api/openapi/openapi.yaml	Regenerates OpenAPI v2 schema.
internal/pkg/service/stream/api/openapi/openapi3.yaml	Regenerates OpenAPI v3 schema.
internal/pkg/service/stream/api/service/source.go	Handles OTLP source test requests as flat OTLP records.
internal/pkg/service/stream/definition/sink_base.go	Adds AllowedSignals to sink definitions.
internal/pkg/service/stream/definition/source_base.go	Adds OTLP-specific source config.
internal/pkg/service/stream/definition/source_otlp.go	Defines OTLP source type, signals, and URL formatting.
internal/pkg/service/stream/mapping/recordctx/fasthttp.go	Adds empty signal for HTTP records.
internal/pkg/service/stream/mapping/recordctx/http.go	Adds empty signal for HTTP records.
internal/pkg/service/stream/mapping/recordctx/otlp.go	Adds OTLP signal storage and test-request context creation.
internal/pkg/service/stream/mapping/recordctx/otlp_test.go	Updates OTLP context tests for signal support.
internal/pkg/service/stream/mapping/recordctx/recordctx.go	Adds Signal() to record context interface.
internal/pkg/service/stream/sink/router/collection.go	Mirrors sink allowed-signal filters.
internal/pkg/service/stream/sink/router/router.go	Applies per-signal sink routing.
internal/pkg/service/stream/source/dispatcher/dispatcher.go	Mirrors OTLP sources in dispatcher lookup.
internal/pkg/service/stream/source/type/otlpsource/decode.go	Accepts application/protobuf alias.
internal/pkg/service/stream/source/type/otlpsource/dispatch.go	Passes OTLP signal into record context and tracks sink failures.
internal/pkg/service/stream/source/type/otlpsource/flatten_common.go	Adds conversion helpers for flattened metric slices.
internal/pkg/service/stream/source/type/otlpsource/flatten_logs.go	Adjusts flattened log numeric types.
internal/pkg/service/stream/source/type/otlpsource/flatten_logs_test.go	Updates log flattening expectations.
internal/pkg/service/stream/source/type/otlpsource/flatten_metrics.go	Adjusts flattened metric numeric/slice types.
internal/pkg/service/stream/source/type/otlpsource/flatten_metrics_test.go	Updates metric flattening expectations.
internal/pkg/service/stream/source/type/otlpsource/flatten_traces.go	Adjusts trace flattening loops and flags type.
internal/pkg/service/stream/source/type/otlpsource/flatten_traces_test.go	Updates trace flattening loop style.
internal/pkg/service/stream/source/type/otlpsource/handler.go	Uses signal constants, validates empty batches, adds Retry-After for 429.
internal/pkg/service/stream/source/type/otlpsource/otlpsource_test.go	Adds OTLP integration-style tests.
internal/pkg/service/stream/source/type/otlpsource/response.go	Escalates all-rejected batches to top-level status.
internal/pkg/service/stream/storage/test/source.go	Adds OTLP source test factory.
stream-otlp-setup.sh	Adds setup helper for OTLP source and sinks.
test/stream/api/source/create-005-invalid-type-400/001-create-source/expected-response.json	Updates valid source type error expectation.
test/stream/api/source/create-otlp-source/001-create-source/expected-http-code	Adds OTLP source create fixture.
test/stream/api/source/create-otlp-source/001-create-source/expected-response.json	Adds OTLP source create expected task.
test/stream/api/source/create-otlp-source/001-create-source/request.json	Adds OTLP source create request.
test/stream/api/source/create-otlp-source/002-poll-source/expected-http-code	Adds OTLP source task polling fixture.
test/stream/api/source/create-otlp-source/002-poll-source/expected-response.json	Adds OTLP source task polling response.
test/stream/api/source/create-otlp-source/002-poll-source/request.json	Adds OTLP source task polling request.
test/stream/api/source/create-otlp-source/003-get-source/expected-http-code	Adds OTLP source get fixture.
test/stream/api/source/create-otlp-source/003-get-source/expected-response.json	Adds OTLP source get expected response.
test/stream/api/source/create-otlp-source/003-get-source/request.json	Adds OTLP source get request.
test/stream/api/source/create-otlp-source/expected-etcd-kvs.txt	Adds expected persisted OTLP source state.
test/stream/api/source/otlp-test-mapping/001-create-source/expected-http-code	Adds OTLP mapping test source create code.
test/stream/api/source/otlp-test-mapping/001-create-source/expected-response.json	Adds OTLP mapping test source create response.
test/stream/api/source/otlp-test-mapping/001-create-source/request.json	Adds OTLP mapping test source create request.
test/stream/api/source/otlp-test-mapping/002-poll-source/expected-http-code	Adds source poll expected code.
test/stream/api/source/otlp-test-mapping/002-poll-source/expected-response.json	Adds source poll expected response.
test/stream/api/source/otlp-test-mapping/002-poll-source/request.json	Adds source poll request.
test/stream/api/source/otlp-test-mapping/003-create-sink/expected-http-code	Adds sink create expected code.
test/stream/api/source/otlp-test-mapping/003-create-sink/expected-response.json	Adds sink create expected response.
test/stream/api/source/otlp-test-mapping/003-create-sink/request.json	Adds sink with OTLP flat-record mappings.
test/stream/api/source/otlp-test-mapping/004-poll-sink/expected-http-code	Adds sink poll expected code.
test/stream/api/source/otlp-test-mapping/004-poll-sink/expected-response.json	Adds sink poll expected response.
test/stream/api/source/otlp-test-mapping/004-poll-sink/request.json	Adds sink poll request.
test/stream/api/source/otlp-test-mapping/005-test-mapping-logs/expected-http-code	Adds successful mapping test expected code.
test/stream/api/source/otlp-test-mapping/005-test-mapping-logs/expected-response.json	Adds successful mapping expected response.
test/stream/api/source/otlp-test-mapping/005-test-mapping-logs/request.json	Adds OTLP flat log test request.
test/stream/api/source/otlp-test-mapping/006-test-mapping-resource/expected-http-code	Adds failing mapping expected code.
test/stream/api/source/otlp-test-mapping/006-test-mapping-resource/expected-response.json	Adds missing-resource mapping error expectation.
test/stream/api/source/otlp-test-mapping/006-test-mapping-resource/request.json	Adds invalid flat OTLP test request.

Files not reviewed (3)

• internal/pkg/service/stream/api/gen/http/stream/server/encode_decode.go: Language not supported
• internal/pkg/service/stream/api/gen/http/stream/server/types.go: Language not supported
• internal/pkg/service/stream/api/gen/stream/service.go: Language not supported

Comments suppressed due to low confidence (4)

internal/pkg/service/stream/source/type/otlpsource/flatten_metrics.go:129

• pt.Count() is a uint64; converting it directly to int64 silently overflows for valid OTLP exponential-histogram counts greater than math.MaxInt64, corrupting the flattened record. Please handle overflow explicitly or use a lossless representation for unsigned counts.
  internal/pkg/service/stream/source/type/otlpsource/flatten_metrics.go:159
• pt.Count() is a uint64; converting it directly to int64 silently overflows for valid OTLP summary counts greater than math.MaxInt64, corrupting the flattened record. Please handle overflow explicitly or use a lossless representation for unsigned counts.
  internal/pkg/service/stream/source/type/otlpsource/otlpsource_test.go:398
• This test also relies on the final shutdown statements running. A failing assertion before these lines leaves the Stream components running; use t.Cleanup right after StartComponents succeeds so cleanup runs on failures too.
  internal/pkg/service/stream/source/type/otlpsource/otlpsource_test.go:451
• If the assertions above fail, these shutdown calls are skipped and the source components can keep running after the test fails. Register cleanup with t.Cleanup immediately after startup instead of relying on reaching the end of the test.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[linear]

AI-3190

[jordanrburger]

Reviewing through the lens of building the kbc-ui side of this on feat/stream-otlp-ui-phase1 (PR #5589 over there). Architecturally this is solid — OTLP as a first-class source type alongside HTTP, per-signal routing via allowedSignals, OTLP-spec-correct partial-success and retry semantics, full dispatcher mirror coverage so records aren't lost on restart, and decent integration tests. The pieces we hit in production while building the UI all checked out.

Five things I'd push for, prioritized — most importantly, the header-auth feature is half-implemented and the secret rotation / cascade delete story has gaps that will create real ops pain.

🔴 Push for before merge

1. Register the header-auth routes — handler is already ready

handler.go:parseAuthParams falls back to Authorization: Bearer <secret> if no <secret> URL param is present. But httpsource.go only registers the secret-in-URL routes — the secret-less routes are missing. So the bearer-token fallback in the handler is currently dead code.

This isn't a follow-up — the code is already there and not wired. Six lines in httpsource.go:

router.Options("/otlp/<projectID>/<sourceID>/v1/logs", otlpHandler.HandleOptions)
router.Post("/otlp/<projectID>/<sourceID>/v1/logs", otlpHandler.HandleLogs)
router.Options("/otlp/<projectID>/<sourceID>/v1/metrics", otlpHandler.HandleOptions)
router.Post("/otlp/<projectID>/<sourceID>/v1/metrics", otlpHandler.HandleMetrics)
router.Options("/otlp/<projectID>/<sourceID>/v1/traces", otlpHandler.HandleOptions)
router.Post("/otlp/<projectID>/<sourceID>/v1/traces", otlpHandler.HandleTraces)

Without this, the only auth mode is "secret in URL," which will leak into HTTP access logs, CDN logs, and APM URL attributes. We had to actively drop the header-auth tab from the kbc-ui PR because the canary returns 404 on these paths today.

2. Expose the secret separately on the source response

Today FormatOTLPSourceURL returns a single string with the secret embedded:

https://stream-in.../otlp/22/my-source/<48-char-secret>

For the UI to render a clean header-auth flow (once #1 lands), we need otlp.secret exposed separately — or an otlp.baseUrl field that omits the secret. The kbc-ui currently has to parse the URL with a regex to split the secret out, which works but is brittle. A small addition to the OTLPSource Goa type:

var OTLPSource = Type("OTLPSource", func() {
    Attribute("url", String, func() { ... })        // existing — full URL
    Attribute("baseUrl", String, func() { ... })    // new — without secret
    Attribute("secret", String, func() { ... })     // new — 48-char value
    Required("url", "baseUrl", "secret")
})

🟡 Push for as follow-ups (already tracked in Linear)

3. Cascade-delete option on source delete

DELETE /v1/branches/{branchId}/sources/{sourceId}?cascade=true to delete the source's sinks, tables, and bucket atomically. Without this, every deletion leaves orphans behind in Storage, and the recreate-with-same-name flow trips a generic 409 from the backend. Tracked as AI-3191 and assigned to you.

4. Secret rotation endpoint

POST /v1/branches/{branchId}/sources/{sourceId}/rotate-secret, optionally accepting gracePeriodSeconds so the old secret keeps working for N seconds while producers are reconfigured. Without this, leaked secrets force delete-and-recreate, breaking every cached producer config. Tracked as AI-3199.

5. Per-sink rejected-records counter in stats

PartialSuccess.RejectedLogRecords (and the metrics/traces equivalents) are surfaced to the producer in the OTLP response, but SinkStatisticsTotalResult only tracks recordsCount (imported/waiting). The UI has no way to show users "12K records rejected by sink mapping in the last hour." Add a cumulative rejected counter alongside recordsCount. Tracked as AI-3196.

🟢 Smaller things worth flagging

allowedSignals semantics on HTTP sinks

The field accepts any of "logs" | "metrics" | "traces" via validate:"dive,oneof=..." regardless of the parent source's type. The doc-comment says "HTTP sources ignore this field." Two suggestions:

• (a) Add a validation that rejects allowedSignals on a sink whose parent source is HTTP-typed (clearer error than silent ignore).
• (b) Or at least add a unit test pinning the silent-ignore behavior so it doesn't regress.

TestSource OTLP body format is undocumented

TestSource now accepts a "flattened OTLP record" body for OTLP sources, with the inline comment pointing at FlattenLogs/Metrics/Traces. The OpenAPI / TestSourcePayload description doesn't mention this — API callers (including future UI iterations of the payload tester) have to read the Go source to know what to send. Suggest expanding the TestSourcePayload description or adding an example flattened-record body in the Goa DSL.

Document the Body() jsonnet null-default behavior

Out of strict scope for this PR, but adjacent: the no-default form of Body('path') throws and silently rejects the entire record when a key is missing. We hit this hard choosing UI defaults — every Cowork event was getting dropped until we switched every template to Body('path', null). Worth a sentence in the OTLP user docs once they exist: "Use Body('path', null) for any optional field unless you want missing fields to drop the record."

FormatOTLPSourceURL — clarify the "base URL" semantics in the field description

The URL we return doesn't include the signal path — the OTel SDK appends /v1/logs etc. itself, per the OTLP/HTTP spec. Some downstream SDKs (we saw Cowork) require the literal base URL with no signal suffix and auto-strip if you try to append one. Not a code change — just expand the URL field's description in design.go:

Base endpoint URL for the OTLP source. Configure this as the endpoint in your OpenTelemetry SDK. The SDK automatically appends /v1/logs, /v1/metrics, or /v1/traces based on the signal type — do not append a signal path yourself. Most SDK exporters reject or silently strip the suffix.

OTLP/gRPC scope

OTLP/HTTP only. One line in the PR description saying gRPC is out of scope (for now / forever / planned for X) would head off reviewer questions.

stream-otlp-setup.sh

A 369-line bash script at the repo root that sets up a test OTLP source. Useful, but it's the only top-level shell script in the repo — feels out-of-place there. Consider scripts/ or dev/.

---

TL;DR for merge

Block on (cheap to fix in this PR):

1. Register the 6 header-auth routes (handler already supports it).
2. Add otlp.baseUrl + otlp.secret to the source response shape.

Track for next PRs (already filed):

• Cascade delete: AI-3191
• Secret rotation: AI-3199
• Rejected-records counter: AI-3196

Everything else is polish that can land later. Parent epic for full UI context: AI-3190.

[Copilot]

Pull request overview

Copilot reviewed 67 out of 72 changed files in this pull request and generated 10 comments.

Files not reviewed (3)

• internal/pkg/service/stream/api/gen/http/stream/server/encode_decode.go: Language not supported
• internal/pkg/service/stream/api/gen/http/stream/server/types.go: Language not supported
• internal/pkg/service/stream/api/gen/stream/service.go: Language not supported

Comments suppressed due to low confidence (1)

internal/pkg/service/stream/source/type/otlpsource/flatten_traces.go:85

• Span link flags are still stored with the OTLP unsigned integer type inside this loop, while span/log flags were converted to int64 to avoid go-jsonnet's unsupported unsigned values. A trace containing links can still fail mappings that read Body('links') or serialize the full body; convert l.Flags() the same way as span.Flags().

[Copilot]

Pull request overview

Copilot reviewed 67 out of 72 changed files in this pull request and generated 7 comments.

Files not reviewed (3)

• internal/pkg/service/stream/api/gen/http/stream/server/encode_decode.go: Language not supported
• internal/pkg/service/stream/api/gen/http/stream/server/types.go: Language not supported
• internal/pkg/service/stream/api/gen/stream/service.go: Language not supported

[Copilot]

Pull request overview

Copilot reviewed 68 out of 73 changed files in this pull request and generated 10 comments.

Files not reviewed (3)

• internal/pkg/service/stream/api/gen/http/stream/server/encode_decode.go: Language not supported
• internal/pkg/service/stream/api/gen/http/stream/server/types.go: Language not supported
• internal/pkg/service/stream/api/gen/stream/service.go: Language not supported