Kepler.Sink behaviour (Kepler v0.1.0)

Copy Markdown View Source

Where events go.

Sinks are named, and a watch routes to them by name:

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

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

A watch that names no sink goes to all of them.

Kepler ships two implementations. Kepler.Sink.Webhook is the one to use; Kepler.Sink.Callback hands the event to your own code, which is also how you get durable delivery today. Three adapters with no real users would mean getting this behaviour wrong in three directions at once with no signal about which one is right — so the third gets written the day it is pointed at something real.

Egress classes

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

  • :best_effort — bounded buffer, drop on backpressure, count the drops. Correct for ops signals, and the only class implemented.
  • :at_least_once — durable local buffer, ack tracking, replay. Required for security signals, where a dropped 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, warns at boot, and behaves as best-effort. Declaring it now means your watches do not change when it lands.

Writing one

init/1 runs once at boot; returning {:error, reason} fails the boot loudly rather than leaving you with a Kepler that delivers nowhere.

format/1 is optional and turns the event into whatever deliver/2 needs — JSON for a webhook, CEF for a SIEM. Without it, deliver/2 receives the Kepler.Event itself.

deliver/2 runs in a supervised task, off the poller, so it may block. It is not retried: returning {:error, _} counts a failed delivery and drops the event, which is the documented behaviour for a wedged endpoint.

Summary

Types

installed()

An initialised sink, ready to receive events.

spec()

A sink as configured: a module, or a module with options.

state()

Whatever init/1 returned; passed back to every deliver/2.

Callbacks

deliver(payload, state)

Delivers one formatted event. Runs in a supervised task, off the poller.

format(event)

Turns an event into what deliver/2 takes. Optional; defaults to the event.

init(opts)

Validates options and builds the sink's state. Runs once, at boot.

Functions

build(sinks)

Initialises every configured sink.

route(sinks, names)

The installed sinks a watch routes to.

Types

installed()

@type installed() :: %{
  name: atom(),
  module: module(),
  state: state(),
  formats?: boolean()
}

An initialised sink, ready to receive events.

spec()

@type spec() :: module() | {module(), keyword()}

A sink as configured: a module, or a module with options.

state()

@type state() :: term()

Whatever init/1 returned; passed back to every deliver/2.

Callbacks

deliver(payload, state)

@callback deliver(payload :: term(), state :: state()) :: :ok | {:error, term()}

Delivers one formatted event. Runs in a supervised task, off the poller.

format(event)

(optional) 
@callback format(event :: Kepler.Event.t()) :: term()

Turns an event into what deliver/2 takes. Optional; defaults to the event.

init(opts)

@callback init(opts :: keyword()) :: {:ok, state()} | {:error, term()}

Validates options and builds the sink's state. Runs once, at boot.

Functions

build(sinks)

@spec build(keyword(spec())) :: {:ok, [installed()]} | {:error, String.t()}

Initialises every configured sink.

Takes the :sinks keyword list — [name: spec] — and returns installed sinks in declaration order. Returns {:error, message} naming the sink that failed, so a typo in a webhook URL is a boot failure with a useful message rather than silence.

route(sinks, names)

@spec route([installed()], :all | [atom()]) :: [installed()]

The installed sinks a watch routes to.