ocsf_ingest

Backend-agnostic ingestion pipeline for the ocsf Elixir library. Accepts OCSF events on a non-blocking hot path, buffers them durably, batches them, and writes them through any OCSF.Sink implementation — without depending on the underlying database.

• Non-blocking dispatch — OCSF.Ingest.dispatch/2 enqueues and returns immediately; writes happen asynchronously in batches.
• Durable buffer — an ETS staging area with a two-phase (:pending/in-flight) record model survives process crashes and graceful restarts.
• Back-pressure, not memory blowup — GenStage demand absorbs slow sinks; a bounded buffer drops explicitly (with telemetry) instead of growing without limit.
• Idempotent end-to-end — at-least-once delivery combined with sinks that write idempotently on metadata.uid means retries never duplicate rows.
• Per-tenant isolation that scales — events are partitioned per tenant (by key and counters, not one table per tenant), so one noisy tenant can only fill and overflow its own quota, never another's — even with thousands of tenants.
• Adapts to the sink — batch size, timeout, queue size, and concurrency are resolved per sink from configuration.

Why

Emission code on an authentication or security hot path must never block on a slow or degraded database. ocsf_ingest sits between the core library and the sink, turning a synchronous OCSF.Sink.write/1 into a buffered, batched, crash-safe pipeline. It is the layer that relieves the database: instead of one synchronous INSERT per event, the database receives a steady stream of batched bulk writes, rate-limited by back-pressure (see Database load).

When to use it

Use ocsf_ingest when emission is on a hot path, volume is non-trivial, or you want batching, back-pressure, crash-safety, and per-tenant isolation — i.e. most production emission.

You can also skip it and call a sink directly (e.g. OCSF.Ecto.Sink.write([event])) when:

• volume is low and the calling context can afford a synchronous DB round-trip;
• you need a synchronous success/failure result at the call site;
• it's a one-off backfill or a script.

Direct writes are synchronous and block until the database returns; dispatch/2 never blocks and never surfaces sink errors at the call site.

Architecture

flowchart LR
  E[OCSF.Event] -->|dispatch/2| B[("ETS buffer\nper tenant")]
  B --> P[GenStage producer pool]
  P --> BW[Broadway pipeline]
  BW -->|handle_batch| S[[OCSF.Sink.write/1]]
  S --> J[Janitor]

ocsf_ingest depends only on ocsf, broadway, and telemetry. The concrete sink (e.g. ocsf_ecto for Postgres) is injected by configuration and resolved at runtime — this library never depends on a database driver.

Adoption

1. Add the dependencies

You need the core, this pipeline, and at least one sink:

# mix.exs
def deps do
  [
    {:ocsf, "~> 0.2"},
    {:ocsf_ingest, "~> 0.1"},
    {:ocsf_ecto, "~> 0.1"}
  ]
end

2. Configure the sink(s)

Each configured sink gets its own pipeline tree. Declare them under config :ocsf_ingest, OCSF.Ingest:

# config/config.exs
config :ocsf_ingest, OCSF.Ingest,
  # how to attribute an event to a tenant (its isolation boundary)
  tenant_path: [:user, :org, :uid],
  default_tenant: :system,
  sinks: [
    {OCSF.Ecto.Sink,
     batch_size: 500,
     batch_timeout_ms: 100,
     max_queue_size: 30_000,            # per-tenant pending cap
     max_global_queue_size: 1_000_000}  # node-level pending cap for this sink
  ]

The tenant is resolved per event from tenant_path (falling back to default_tenant), or overridden per call with dispatch(event, tenant: ...).

3. Start it

The library starts itself through its OTP application (OCSF.Ingest.Application) as soon as it is a dependency — there is nothing to add to your supervision tree. Booting reads the sinks config and starts one supervised tree per sink. (Configure the sink's own runtime too — e.g. point ocsf_ecto at a Repo and run its migration; see step 4.)

4. Set up the sink's storage

Each sink owns its storage setup. For ocsf_ecto: configure the Repo + Cloak vault and run the migration (OCSF.Ecto.Migration) — see the ocsf_ecto README.

5. Dispatch events

# Build an event via the core library
{:ok, event} =
  OCSF.Events.Authentication.logon(
    user: %{uid: "u1", org: %{uid: "acme"}},
    service: %{name: "My Auth"},
    status: :Success,
    severity: :Informational
  )

# Hand it to the pipeline — returns immediately
:ok = OCSF.Ingest.dispatch(event)

# Batch dispatch
:ok = OCSF.Ingest.dispatch(many_events)

# Target a specific sink and/or override the tenant
:ok = OCSF.Ingest.dispatch(event, sink: OCSF.Ecto.Sink, tenant: "acme")

dispatch/2 returns {:error, :overflow} when the target tenant's buffer is full, or {:error, :no_sink} when no sink can be resolved. All other conditions (sink errors, transient outages) are handled asynchronously downstream and retried. All emission is async — there is no blocking variant.

Per-sink knobs

Knobs are resolved per sink, later sources winning:

1. the options in sinks: config (deployment);
2. the sink module's ingest_defaults/0, if it exports one;
3. the library defaults below.

Database load & tuning

The whole point of the pipeline is to relieve the database:

• Batched bulk writes — events are grouped (batch_size / batch_timeout_ms) and written with a single OCSF.Sink.write/1 per batch (e.g. one multi-row INSERT via Repo.insert_all/3), not one INSERT per event.
• Back-pressure — the pipeline only pulls what it can write; if the database slows, demand drops and events accumulate in the ETS buffer instead of saturating the connection pool.
• Burst smoothing — spikes are absorbed in memory and drained at a steady, database-friendly rate.
• Bounded concurrency — concurrency caps simultaneous writes/connections.
• Overflow, not collapse — past max_queue_size / max_global_queue_size, the newest events are dropped (with telemetry) rather than overwhelming the DB.

Tune batch_size up (and batch_timeout_ms to taste) to trade latency for fewer, larger writes when the database is the bottleneck. For very high volume, add a columnar sink (e.g. ClickHouse) alongside Postgres — the pipeline fans out to multiple sinks unchanged.

Telemetry

tenant is in event metadata for routing/sampling but must not be exported as a metric label by default — thousands of tenants would explode metric series. The [:ocsf, :ingest, :buffer] gauge is aggregated per sink. Attach a module-based handler (never an anonymous function — :telemetry warns about it).

Durability

Events are staged in ETS as :pending, marked in-flight when taken by the pipeline, and deleted only after the sink confirms the write. A crash mid-flight leaves events recoverable: the in-flight records of a dead producer are reclaimed to :pending and re-delivered. Because sinks write idempotently on metadata.uid, retries are no-ops at the row level.

The buffer is in-memory (ETS): it survives process crashes, but not a full node crash or power loss, and it drops the newest events of a tenant whose queue overflows. This at-least-once, in-memory guarantee is the v0 durability level.

Links

• ocsf — core library
• ocsf_ecto — Postgres sink
• OCSF 1.9 Schema
• Broadway — batching pipeline

License

Apache-2.0