Sinks and payloads

Where events go, what they look like, and how to trust them.

Configuring sinks

Sinks are named, and watches route to them by name.

config :kepler,
  sinks: [
    ops: {Kepler.Sink.Webhook,
      url: System.fetch_env!("KEPLER_OPS_URL"),
      headers: [{"x-team", "payments"}]},
    siem: {Kepler.Sink.Webhook,
      url: System.fetch_env!("KEPLER_SIEM_URL"),
      secret: System.fetch_env!("KEPLER_SIEM_SECRET")}
  ]

watch :ratelimit_anomaly do
  source telemetry: [:my_app, :ratelimit, :reject]
  measure :rate
  sink :siem, guarantee: :at_least_once
  fire when: value > 50, sustained: :timer.seconds(10)
end

A watch that names no sink goes to all of them. Naming a sink that is not configured is a boot failure, as is a sink that cannot initialise — a typo in a webhook URL is a startup error, not a silent non-delivery.

Two implementations, deliberately. Three adapters with no real users would mean getting the behaviour wrong in three directions at once with no signal about which one is right; the third gets written the day it is pointed at something real. Sinks are cheap — about a hundred lines each. Sources are where the cost is.

Egress classes

Delivery semantics are not a formatting concern, so they are declared separately from the sink.

The distinction is not academic: a dropped security event is a detection blind spot, and an attacker who can generate volume can wash out their own trail.

guarantee: :at_least_once is accepted today, warns at boot, and behaves as best-effort. Declaring it now means your watches do not change when the durable buffer lands. Until then, route those watches through Kepler.Sink.Callback into a queue you already run.

[kepler] these watches ask for guarantee: :at_least_once, which is not
implemented yet and will be delivered best-effort: ratelimit_anomaly -> :siem.
Route them through Kepler.Sink.Callback into a durable queue if you cannot
afford drops.

The payload

The struct is split in two, and the split is the most important decision in the schema.

The core is small and fully required. id, node, watch, timestamp, severity, and state are always present and always populated. Route on any of them without a nil check, forever.

context is explicitly best-effort. Everything consumer-specific lives there, and keys that do not apply are absent rather than null.

{
  "schema": "kepler.event/1",
  "id": "01920f3c-6a1b-7c4e-9f00-3d2c1b0a9e8f",
  "node": "app@10.0.0.1",
  "watch": "checkout_latency",
  "timestamp": "2026-08-06T12:34:56.789Z",
  "severity": "critical",
  "state": "firing",
  "context": {
    "source": {"type": "telemetry", "event": "my_app.checkout.stop"},
    "tier": 1,
    "condition": "value > 2000",
    "debounce": {"sustained_ms": 30000, "cooldown_ms": 0},
    "measurement": {
      "key": "duration", "aggregate": "p99", "unit": "millisecond",
      "value": 2431, "previous": 1980, "delta": 451, "rate": 451.0,
      "window_ms": 1000
    },
    "meta": {"team": "payments", "runbook": "https://runbooks.example/checkout"},
    "kepler": {"version": "0.1.0", "share": 0.0008, "level": 0}
  }
}

Without the split, the core would have to satisfy a SIEM, an autoscaler, and someone debugging an incident simultaneously — and would degrade into a union of optional fields where nothing is guaranteed and every consumer writes nil checks forever.

The core

• watch — the stable event id. Route on this. Renaming a watch is a breaking change for whoever receives it.
• severity — "info", "warning", "error", or "critical".
• state — "firing" or "resolved".
• id — a UUIDv7, unique per event and sortable by time. Use it to deduplicate if your receiver can deliver twice.
• node — which node observed it. Each node fires independently, so a cluster-wide condition produces one event per node.
• timestamp — ISO 8601, UTC, millisecond precision.

The context

Everything here is conditional on what the watch was and what happened.

schema is versioned. A breaking change to the payload changes it and is called out in the changelog.

Terms JSON cannot hold

context carries whatever your application put there, which will not always be JSON-shaped — a crash reason is a tuple, a process_state is whatever your GenServer was holding. Rather than fail to encode, which would drop the event, Kepler renders them: atoms and tuples become strings and lists, pids and refs become their inspect/1 form, keyword lists become objects, invalid binaries become their inspect/1 form, and nesting deeper than eight levels becomes "...".

Verifying the signature

With a :secret configured, every request carries:

kepler-signature: t=1754481296,v1=8f2a...c31d

where the hex is HMAC-SHA256(secret, "<t>.<raw body>").

Verify against the raw body, before any JSON parsing — a re-serialized body will not match. Compare in constant time, and reject timestamps far from now so a captured request cannot be replayed indefinitely.

In Elixir

defmodule MyApp.KeplerWebhook do
  @tolerance 300

  def verify(secret, signature_header, raw_body) do
    with ["t=" <> timestamp, "v1=" <> digest] <- String.split(signature_header, ","),
         true <- fresh?(timestamp) do
      expected = Kepler.Sink.Webhook.signature(secret, timestamp, raw_body)
      Plug.Crypto.secure_compare(digest, expected)
    else
      _other -> false
    end
  end

  defp fresh?(timestamp) do
    case Integer.parse(timestamp) do
      {sent, ""} -> abs(System.system_time(:second) - sent) <= @tolerance
      _other -> false
    end
  end
end

In a Phoenix endpoint you need the raw body, which means a custom body reader:

# endpoint.ex
plug Plug.Parsers,
  parsers: [:json],
  json_decoder: JSON,
  body_reader: {MyApp.CachingBodyReader, :read_body, []}

# caching_body_reader.ex
defmodule MyApp.CachingBodyReader do
  def read_body(conn, opts) do
    {:ok, body, conn} = Plug.Conn.read_body(conn, opts)
    {:ok, body, Plug.Conn.assign(conn, :raw_body, body)}
  end
end

In another language

The scheme is deliberately boring. In Python:

import hashlib, hmac, time

def verify(secret, header, raw_body, tolerance=300):
    parts = dict(p.split("=", 1) for p in header.split(","))
    if abs(time.time() - int(parts["t"])) > tolerance:
        return False
    expected = hmac.new(
        secret.encode(), f"{parts['t']}.".encode() + raw_body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(parts["v1"], expected)

Routing without parsing

Every webhook request carries these headers, so a consumer can route before it looks at the body:

Delivery, and what happens when it fails

Firing must never block evaluation, and a wedged endpoint must never become a memory leak. So the poller hands the event to Kepler.Emitter and moves on; the emitter holds a fixed-size queue and runs a bounded number of deliveries at a time under Kepler's own task supervisor.

When the queue is full, events are dropped and counted. That is the whole policy:

• No retries. Kepler will not retry into a cluster that is already having a bad day.
• No durable spool. There is no database, so there is nothing to retry from. A delivery lost to a crash is gone.
• A non-2xx response is a failed delivery, counted and logged.

config :kepler,
  buffer: [
    max_size: 500,             # events queued before dropping
    max_in_flight: 4,          # events being delivered at once
    delivery_timeout: 15_000,  # before a wedged sink is killed
    drop: :oldest              # or :newest
  ]

drop: :oldest keeps the freshest picture of what is wrong, which is what an on-call human wants. drop: :newest refuses what just arrived and keeps the first sign of trouble, which is what a post-mortem wants.

Kepler.status().emitter reports queued, in_flight, delivered, dropped, and failed. Kepler also emits its own telemetry — [:kepler, :delivery, :stop], [:kepler, :delivery, :exception], and [:kepler, :event, :dropped] — so you can watch the watcher with whatever you already use for metrics.

Fanning out to several destinations

An event routed to several sinks is delivered to all of them concurrently, each in its own task. One slow destination does not delay the others.

Deliveries are also isolated. A sink that raises, exits, or stops responding is recorded as that sink's failure; its siblings receive the event regardless:

[kepler] sink :siem timed out delivering ratelimit_anomaly; other sinks for this
event were unaffected

A sink that stops responding is killed after :delivery_timeout (15 seconds by default) rather than holding an in-flight slot forever behind an endpoint that accepted the connection and then went quiet. That failure surfaces as [:kepler, :delivery, :exception] with reason: :timeout.

Two bounds compose here:

• :max_in_flight caps how many events are being delivered at once.
• Each of those fans out to the sinks its watch routes to.

So the ceiling on concurrent outbound requests is max_in_flight times the number of sinks a watch names — a small, static number, since sinks are configured rather than discovered. With the defaults and three sinks, at most twelve requests are in flight.

Delivery order across sinks is not guaranteed, and there is no failover: every routed sink gets its own attempt, and one succeeding says nothing about another.

Calling your own code

config :kepler,
  sinks: [alerts: {Kepler.Sink.Callback, handler: {MyApp.Alerts, :handle_kepler_event}}]

defmodule MyApp.Alerts do
  def handle_kepler_event(%Kepler.Event{state: :firing, severity: :critical} = event) do
    MyApp.PagerDuty.trigger(event.watch, Kepler.Event.to_map(event))
  end

  def handle_kepler_event(_event), do: :ok
end

The handler is a one-arity function or a {module, function} pair, and the module and function must exist at boot. It runs in a supervised task off the poller, so it may block — but it is called once per event with no retry, and raising counts as a failed delivery rather than taking anything down.

Handing the event to a queue you already run is the supported way to get at-least-once delivery today. Kepler deliberately has no durable buffer of its own, so this is the seam where you add one.

Using your own HTTP client

The default transport is OTP's :httpc, in a dedicated profile, with TLS verification enabled — :httpc does not verify certificates by default, and Kepler turns that on because the body is a signed description of your production system. This keeps Kepler's runtime dependency list at exactly one entry.

If you already run Finch, Req, or Mint, implement Kepler.Transport:

defmodule MyApp.KeplerTransport do
  @behaviour Kepler.Transport

  @impl true
  def post(url, headers, body, _opts) do
    case Finch.build(:post, url, headers, body) |> Finch.request(MyApp.Finch) do
      {:ok, %{status: status}} -> {:ok, status}
      {:error, reason} -> {:error, reason}
    end
  end
end

config :kepler,
  sinks: [
    ops: {Kepler.Sink.Webhook, url: ..., transport: MyApp.KeplerTransport}
  ]

Do not retry inside a transport. A wedged endpoint should produce drops, not a growing queue.

Writing a sink

defmodule MyApp.PubSubSink do
  @behaviour Kepler.Sink

  @impl true
  def init(opts), do: {:ok, Keyword.fetch!(opts, :topic)}

  @impl true
  def deliver(event, topic) do
    Phoenix.PubSub.broadcast(MyApp.PubSub, topic, {:kepler, event})
  end
end

Kepler.Sink.init/1 runs once at boot; returning {:error, reason} fails the boot loudly.

Kepler.Sink.format/1 is optional and turns the event into whatever Kepler.Sink.deliver/2 takes — JSON for a webhook, CEF for a SIEM. Without it, deliver/2 receives the Kepler.Event itself. Splitting the two means the wire format is replaceable without reimplementing the transport, signing, and failure handling around it:

@impl true
def format(event), do: MyApp.CEF.encode(Kepler.Event.to_map(event))

@impl true
def deliver(line, socket), do: MyApp.Syslog.send(socket, line)

Kepler.Sink.deliver/2 runs in a supervised task and should return :ok or {:error, reason}.