feat: multi-source ingestion#3517
Open
Ziinc wants to merge 6 commits into
Open
Conversation
Ingest requests to POST /, /api/logs, and /api/events can now declare a source per event by setting __LF_SOURCE to the source UUID on each event. The ?source= query param becomes optional and a single batch can fan out to multiple sources. A new VerifyDeclaredSources plug detects multi-source mode by pattern matching on the first event, collects unique declared UUIDs, resolves them via Sources.Cache, and verifies the caller owns each one (including OAuth scope check). VerifyResourceAccess now passes through when the query-param source is nil but declared sources are present. LogController.create dispatches events per declared source, strips __LF_SOURCE before ingest, and aggregates per-source results. Events with malformed/missing __LF_SOURCE and no default source surface as errors in the response while valid events are still ingested. https://claude.ai/code/session_012GiTAWivtpRQfB1bUHnpW2
- group_events_by_source and aggregate_results: use for/reduce instead of Enum.reduce for consistency with codebase idioms - event_source_token: use pattern matching function clauses instead of Map.get - resolve_and_verify token collection: use for/into instead of reduce + filter - log_controller tests: remove unsupported single-event test, add _json fan-out case, use source.token directly, move multi_source_setup to top of helpers, insert test-specific sources at test level, drop reject_context_functions from multi-source describe (test-level sources can't be pre-warmed before Mimic stubs run) - verify_declared_sources tests: insert sources at test level, combine passthrough tests into one parameterized assertion, combine auth failure cases into one parameterized test https://claude.ai/code/session_012GiTAWivtpRQfB1bUHnpW2
djwhitt
reviewed
May 25, 2026
| def call(%{assigns: %{resource_type: :source, user: %User{} = user}} = conn, _opts) do | ||
| case extract_events(conn.body_params) do | ||
| [first | _] = events | ||
| when is_map(first) and (is_map_key(first, :__LF_SOURCE) or is_map_key(first, "__LF_SOURCE")) -> |
Contributor
There was a problem hiding this comment.
Could we make this more explicit by using a header to identify these? I worry about mistakes with the more implicit check on the first event.
djwhitt
reviewed
May 25, 2026
| @spec call(Plug.Conn.t(), opts()) :: Plug.Conn.t() | ||
| def call(%{assigns: %{declared_sources: declared}} = conn, _opts) | ||
| when map_size(declared) > 0 do | ||
| conn |
Contributor
There was a problem hiding this comment.
Is the limiter bypass here intentional?
djwhitt
reviewed
May 25, 2026
| } | ||
| ) | ||
|
|
||
| assert json_response(conn, 406) |
Contributor
There was a problem hiding this comment.
Is 406 right? Seems more like 400.
djwhitt
reviewed
May 25, 2026
| {:ok, n} -> {acc + n, errs} | ||
| :ok -> {acc, errs} | ||
| {:error, more} when is_list(more) -> {acc, errs ++ more} | ||
| {:error, err} -> {acc, errs ++ [err]} |
Contributor
There was a problem hiding this comment.
Nit: pretty sure this O(N) in list length. Could prepend and reverse after.
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
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.
This allows source routing at the payload level, with
__LF_SOURCEbeing provided as a special routing key.This allows separate HTTP pipelines to be merged together into a multi-source request, while maintaining backwards compatability with query parameter source routing.
internally, at LogEvent.make/2 nothing changes. All changes are done at the controller level.
Pre-merge todos:
Closes O11Y-1838
here i am sending a multi-source payload to all sources named loadfest.
for benchmark of 500 events:
there is very marginal performance impact, in the grand scheme of things it would speed things up as less http requests are made overall.
CleanShot.2026-05-22.at.14.13.42.mp4