Funnel Ingestion API

Send your first metric. Replace st_YOUR_API_KEY with a key from your project's API Keys page.

curl -X POST http://localhost:4000/v1/metrics \
  -H "authorization: Bearer st_YOUR_API_KEY" \
  -H "content-type: application/json" \
  -d '{"metrics":[{"attributes":{"method":"GET","route":"/users","service":"api"},"kind":"gauge","name":"http.server.duration_ms","time":"2026-05-11T17:00:00Z","value":182.3},{"attributes":{"service":"api"},"kind":"counter","name":"http.server.requests","time":"2026-05-11T17:00:00Z","value":1.0}]}'

Successful response:

{
  "accepted": 2
}

Every request requires a project API key as a Bearer token:

Authorization: Bearer st_AbCdEf1234...
Content-Type: application/json

• · Keys are scoped to a single project. All three pillars (metrics/traces/logs) share the same key.
• · Keys are bcrypt-hashed at rest; the plaintext is shown exactly once on creation.
• · RUM beacons may use ?token=… in the URL instead — required for navigator.sendBeacon.
• · Revoke a key from the API Keys page; future requests return 401.

User-deployed JavaScript handlers served at /w/<org_slug>/<project_slug>/<worker_slug>. No API key on the public URL — these are your published endpoints. Two runtimes are available:

• Node JS — real JavaScript in a per-invocation V8 sandbox (vm.runInContext) with no globals beyond request, env, console, and (when an outbound allow-list is set) funnel. Spawned with --max-old-space-size=<mb> and --disallow-code-generation-from-strings. Wall-clock kill enforced from the BEAM side via Port close.
• Safe templates — a tiny non-Turing DSL (status:, header:, body:) with {{request.query.name}} placeholders. Cannot infinite-loop. Useful for static / redirect / health-check endpoints when JS isn't justified.

Hello world

// Available in scope: request, env, console, funnel.
// Return a Response-shaped object or a JSON-able value.
if (request.method === "GET") {
  return {
    status: 200,
    headers: { "content-type": "application/json" },
    body: JSON.stringify({
      hello: request.query.name || "world",
      path: request.path
    })
  };
}
return { status: 405, body: "method not allowed" };

Request shape

Response shape

Return a Response-like object: { status, headers, body }. Or return a string (becomes plain-text 200), or any JSON-serializable value (becomes JSON 200). Throwing yields a 500 with the error message captured.

Auth

Workers are public by default. Toggle auth_required and rotate a token from the dashboard's Settings → Authentication tab. Callers must then send Authorization: Bearer <token>; the plaintext is shown to you exactly once on rotation and never persisted in cleartext (only a SHA-256 hash).

Env vars (secrets at rest)

Set per-worker secrets from Settings → Environment variables. Each value is encrypted with AES-256-GCM under a key derived from the application secret via HKDF before being written to the DB. The dashboard only ever shows a masked fingerprint; the worker sees the decrypted value at request time. Names must match [A-Z_][A-Z0-9_]*; max 64 keys, 8KB per value.

CORS

Set cors_origins from Settings → CORS allow-list. Without an entry, browser callers are blocked (preflight returns 403). Matching origin → response carries Access-Control-Allow-Origin + Vary: Origin; OPTIONS preflight returns 204 with the Access-Control-Allow-Methods/Headers/Max-Age echo. Wildcard * mirrors the request origin so the worker stays credential-compatible.

Staging environment

Every worker has two version slots: prod (active_version_id) and staging (staging_version_id). The dashboard's Deploy to staging button writes a new version to the staging slot without touching prod; Promote → prod swaps it in.

Staging is reachable via a hidden URL segment:

# Prod
curl https://YOUR-FUNNEL/w/acme/api/hello

# Staging — same worker, separate version
curl https://YOUR-FUNNEL/w/acme/api/hello/_staging

The _staging segment is consumed by the dispatcher; your code sees the remaining path as request.path. Invocations carry an env attribute ("prod"/"staging") for filtering.

Outbound HTTP via funnel.fetch

User code has no fetch / http / net. The single egress primitive is funnel.fetch(url, opts), which round-trips through an Elixir-side proxy that enforces three defenses on every call:

• Allow-list. The URL's host[:port] must appear in outbound_allow_list, or the list must include *. Empty list = throws immediately.
• SSRF guard. After DNS resolution, refuses private (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16), loopback (127.0.0.0/8), link-local (169.254/16 — blocks the cloud metadata service), and multicast addresses.
• Resource caps. 1MB response cap, 5s default timeout (override via timeoutMs, clamped 50–30,000ms), Host/Connection/Transfer-Encoding etc. headers blocked.

// Configure outbound_allow_list = ["api.stripe.com"] in the dashboard.
const resp = await funnel.fetch("https://api.stripe.com/v1/charges", {
  method: "POST",
  headers: {
    authorization: `Bearer ${env.STRIPE_KEY}`,
    "content-type": "application/x-www-form-urlencoded"
  },
  body: "amount=1000&currency=usd",
  timeoutMs: 3000
});

return {
  status: resp.status,
  headers: { "content-type": "application/json" },
  body: resp.body
};

Security model

• No ambient capabilities. fetch, require, process, setTimeout, the filesystem, the network — all undefined. typeof require returns the string "undefined". The only egress primitive is the allow-list-gated funnel.fetch.
• Per-project warm pool. Sandboxes are reused within the same project (skipping ~30–60ms Node start-up per request) but never across projects. Each invocation creates a fresh vm.createContext, so two back-to-back requests can't share V8 state. Pool caps: 32 in-flight global, 8 per project, 4 warm per project. Idle sandboxes self-terminate after 30s.
• Replace-on-failure. Timeout, OOM, or crash kills the sandbox; the next acquire spawns fresh. The Pool traps exits so abnormal sandbox deaths never crash the host.
• Wall-clock kill. timeout_ms is enforced on the BEAM side by closing the Port — V8's own vm timeout is a second layer.
• Memory cap. --max-old-space-size=<memory_mb>. OOM kills the Node process cleanly; the invocation is recorded as crashed.
• Quotas + per-project concurrency. daily_invocation_cap per worker, accounted via an ETS-backed atomic counter (seeded once per day from SQL), plus a per-project sandbox cap. Over-cap returns 429.
• Isolation driver. The Node spawn goes through a pluggable Funnel.Workers.Runtime.Driver. The default driver is in-process. On Linux you can opt-in to the Hardened driver (FUNNEL_WORKERS_DRIVER=Elixir.Funnel.Workers.Runtime.Driver.Hardened), which wraps Node in firejail (caps drop, seccomp, --net=none) or unshare (user/PID/net/mount namespaces) when available.

HTTP status codes

Observability

Each invocation writes a row to worker_invocations (partitioned by month, BRIN-indexed on time) and emits two metric points into the regular pipeline:

• workers.invocations — counter, tags worker, status, warmth.
• workers.latency_ms — gauge, same tags.

The invocation row also stores a JSONB attributes column. Today it carries:

• trace_id + parent_span_id — extracted from W3C traceparent, when the caller sent one. Lets the invocation row join to upstream traces.
• warmth — "cold" (fresh sandbox spawn) or "warm" (reused from pool). Useful for tuning the warm-pool sizing.
• env — "prod" or "staging".

Invocation writes are batched + retried by Funnel.Workers.InvocationWriter (5 attempts with exponential backoff, then a structured log line per row so an external collector can recover the data). The Pool drains gracefully on app shutdown (FUNNEL_WORKERS_DRAIN_MS, default 5s).

A non-Turing alternative to the JS runtime. Pick this when you only need to return a static response, a JSON envelope built from request fields, a redirect, or an env-driven config — and you want a strong guarantee that the worker cannot spin a CPU, eat memory, or call out anywhere.

Syntax

Source is line-oriented. Empty lines and lines starting with # are ignored. Three directives are recognised:

Placeholders

Anything inside {{…}} on any line is replaced before the line is written. Unknown placeholders render as the empty string (never throw).

Auto-escaping

Interpolated values are escaped by default based on the response content-type you set with a header: directive — this is the only thing standing between you and a reflected XSS when echoing query strings into a page.

Header values are not escaped — CRLF (\r / \n) is stripped instead, so a malicious {{request.query.x}} can't smuggle a Set-Cookie via response-splitting.

To opt out of escaping for a specific value (you've already validated it, you're emitting pre-rendered HTML, etc.), prefix the placeholder name with raw:

header: content-type: text/html
# {{request.query.name}}        — escaped: <b>Bob</b> renders as &lt;b&gt;Bob&lt;/b&gt;
# {{raw request.query.name}}    — raw:    <b>Bob</b> renders as <b>Bob</b>
body: <h1>Hi {{request.query.name}}</h1>

Examples

JSON response that echoes a query param

status: 200
header: content-type: application/json
body: {"hello":"{{request.query.name}}","served_at":"{{now}}"}

Redirect

status: 302
header: location: https://example.com/{{request.query.to}}
header: cache-control: no-store

Health check with current time

# Minimal health endpoint — perfect for uptime monitors.
status: 200
header: content-type: text/plain
body: ok {{now}}

Public app config (env-driven)

# Set env_vars on the worker: api_base, feature_flag
status: 200
header: content-type: application/json
header: cache-control: public, max-age=60
body: {"api":"{{env.api_base}}","new_ui":"{{env.feature_flag}}"}

Multi-line HTML body

Lines that don't start with a directive become body lines.

status: 200
header: content-type: text/html
body: <!DOCTYPE html>
<html>
<head><title>Hi {{request.query.name}}</title></head>
<body><h1>Hello, {{request.query.name}}</h1>
<p>Served at {{now}}</p></body>
</html>

Limits & behaviour

• No loops, no branching. Templates are pure substitution. To pick a value based on the request, do it on the client (or upstream).
• Single body field. Either set body: on a single line, or build a multi-line body by writing literal lines beneath it. The output response has exactly one body.
• No nested placeholders. {{…}} cannot reference each other. Each interpolation is a single lookup.
• Headers and body still respect platform caps: the workers_public rate limit, daily invocation cap, and HTTP body size limits all apply just like the Node runtime.
• Every call still records an invocation row and emits workers.invocations & workers.latency_ms metrics — so safe-template workers show up next to JS workers in the dashboard and Metrics Explorer.

The HTTP transport accepts both OTLP encodings: application/json and application/x-protobuf. Either works with the same three endpoints — no separate URL, no separate port. The decoder dispatches on Content-Type.

SDK configuration

Point any OpenTelemetry SDK at Funnel via the standard env vars:

export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4000
export OTEL_EXPORTER_OTLP_HEADERS="authorization=Bearer st_YOUR_API_KEY"
export OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf   # or http/json
export OTEL_EXPORTER_OTLP_COMPRESSION=gzip          # optional but recommended

SDKs append /v1/metrics, /v1/traces, /v1/logs automatically.

Spec compliance

Backpressure

Each pillar has its own GenServer with a bounded buffer. When a buffer crosses the 10k high-water mark, the server returns 503 Service Unavailable with a Retry-After header. Recovery uses hysteresis: shedding clears once the buffer drops below 5k. Buffer state is exposed via Funnel.Ingestion.Pipeline.stats/0.

# Server is shedding — example response
HTTP/1.1 503 Service Unavailable
Retry-After: 12
Content-Type: application/json

{"error":"service overloaded","retry_after_ms":12000}

Verified throughput

Sustained 30,900 events / sec for 30 seconds with full SDS regex redaction, Software Catalog auto-tagging, partition-aware inserts, and PartialSuccess validation — zero drops, zero 5xx. Single-node laptop, single Postgres instance. Run your own benchmark with the included mix funnel.load task:

# 500 req/s × 100 events/req = 50k events/sec target
mix funnel.load \
  --rps 500 --duration 30 --batch 100 --senders 16 \
  --endpoint http://localhost:4000/v1/metrics \
  --token st_xxx --gzip

Real gRPC server on the OTLP-standard port 4317. Implements the three canonical services:

• opentelemetry.proto.collector.metrics.v1.MetricsService/Export
• opentelemetry.proto.collector.trace.v1.TraceService/Export
• opentelemetry.proto.collector.logs.v1.LogsService/Export

Enable it

Boot-gated so dev installs don't pay the listener overhead. Set FUNNEL_GRPC_ENABLED=1 before starting:

export FUNNEL_GRPC_ENABLED=1
export FUNNEL_GRPC_PORT=4317        # optional, this is the default
mix phx.server

SDK configuration

export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
export OTEL_EXPORTER_OTLP_PROTOCOL=grpc
export OTEL_EXPORTER_OTLP_HEADERS="authorization=Bearer st_YOUR_API_KEY"

gRPC status mapping

POST your ExportMetricsServiceRequest / ExportTraceServiceRequest / ExportLogsServiceRequest protobuf bytes to the regular endpoint — no separate URL, no separate scope:

curl -X POST http://localhost:4000/v1/metrics \
  -H "authorization: Bearer st_YOUR_API_KEY" \
  -H "content-type: application/x-protobuf" \
  --data-binary @export-metrics-request.pb

Decoded by a pure-Elixir protobuf wire-format parser (Funnel.Ingestion.OtlpProto). No protoc, no codegen, no extra deps — the decoder targets the OTLP messages directly. Hardened with @max_depth=32 recursion limit and @max_field_bytes=16MB length-prefix cap to prevent OOM via hostile payloads. Combined with gzip, this is the fastest transport Funnel offers.

Every project belongs to a plan tier with two layers of limits: a daily budget (bytes + events; resets at 00:00 UTC) and a burst rate (token-bucket events/sec; protects against runaway senders consuming the daily budget in a few seconds).

Tier defaults

Check order on every request

1. Burst check — token-bucket scoped to the project. Sub-microsecond ETS lookup. If insufficient tokens for the batch: 429 with Retry-After; daily budget is not charged.
2. Daily quota check — bytes + events vs the tier cap. If over: 429 with Retry-After: <seconds-to-midnight-UTC> and JSON body describing what was exceeded.
3. Pipeline acceptance — buffer headroom check. If shedding: 503 with Retry-After.

Response shapes

Burst rate-limited (429)

HTTP/1.1 429 Too Many Requests
Retry-After: 1
Content-Type: application/json

{"error":"burst rate limit","retry_after_ms":1000}

Daily quota exceeded (429)

HTTP/1.1 429 Too Many Requests
Retry-After: 14400
Content-Type: application/json

{
  "error":"quota exceeded",
  "kind":"bytes",
  "used":104857600,
  "limit":104857600,
  "resets_at_utc":"2026-05-18T00:00:00Z"
}

Per-project overrides

Operator dial for paying customers that need more headroom than the tier default. Set any of these projects.*_override columns to NULL to inherit the tier value:

• ingest_bytes_per_day_override — override the daily byte cap
• ingest_events_per_day_override — override the daily event cap
• retention_days_override — extend retention without upgrading the tier
• max_workers_override — extra Edge Worker headroom

Live dashboard

The Quotas & limits page (sidebar → Security group) shows live usage, rejection counts, and pipeline buffer headroom. Refreshes every 3 seconds.

Alert rules are evaluated by an Oban cron worker once a minute. When a rule breaches it transitions to firing, writes an alert_events row, and enqueues one webhook delivery per configured destination. When the rule recovers it transitions back to ok and resolves any open events.

Rule kinds

Each rule has a kind discriminator and a free-form config jsonb. The same shape is accepted by the dashboard form and the underlying schema.

{
  "config": {
    "agg": "p95",
    "filters": {
      "service": "api"
    },
    "metric": "http.server.duration_ms",
    "operator": ">",
    "threshold": 500,
    "window_seconds": 300
  },
  "kind": "metric_threshold",
  "name": "API p95 latency",
  "severity": "warning"
}

{
  "config": {
    "service": "api",
    "threshold": 0.05,
    "window_seconds": 300
  },
  "kind": "error_rate",
  "name": "API error rate",
  "severity": "critical"
}

{
  "config": {
    "pattern": "timeout connecting to upstream",
    "severity_min": "error",
    "threshold": 10,
    "window_seconds": 300
  },
  "kind": "log_pattern",
  "name": "Repeated DB timeout",
  "severity": "critical"
}

{
  "config": {
    "k": 3,
    "metric": "http.server.duration_ms",
    "window_seconds": 300
  },
  "kind": "anomaly",
  "name": "Login latency anomaly",
  "severity": "warning"
}

{
  "config": {
    "check_id": 42,
    "consecutive_failures": 3
  },
  "kind": "synthetic_failure",
  "name": "Public homepage down",
  "severity": "critical"
}

State machine

Every minute the evaluator transitions a rule between three states.

Synthetic checks are scheduled HTTP probes that run from your Funnel deployment. Each successful run writes a synthetic.latency_ms metric and a synthetic.success counter through the normal ingestion pipeline, plus a log entry — so synthetic data shows up in the Metrics Explorer, Logs, and Alerts pages just like any other telemetry.

Check configuration

{
  "assert_body_regex": "\"status\":\\s*\"ok\"",
  "enabled": true,
  "expected_status": 200,
  "headers": {
    "x-api-key": "..."
  },
  "interval_seconds": 60,
  "max_latency_ms": 1500,
  "method": "GET",
  "name": "Homepage health",
  "timeout_ms": 10000,
  "url": "https://www.example.com/health"
}

Pass/fail logic

A check is considered successful when all the following hold:

• · HTTP status equals expected_status
• · Response received before timeout_ms elapsed
• · max_latency_ms (if set) is not exceeded
• · assert_body_regex (if set) matches the response body

What gets emitted per run

Pairing with alerts

The cleanest way to alert on a synthetic check is the synthetic_failure rule kind — it tracks consecutive_failures on the check row directly, avoiding flap. For ratio-based alerting across many checks you can also evaluate the emitted synthetic.success counter via a metric_threshold rule with agg = "avg".

The Sensitive Data Scanner inspects every piece of telemetry you ingest — logs, traces, RUM events, session replay strings, and metric attributes — and either alerts, redacts, hashes, or drops matches before they land in storage. SDS sits inline in the ingestion pipeline shard, so there is no batch job to wait for: a credit card in a log line becomes [REDACTED:credit_card] in the same flush that would have persisted it.

Three additional production-grade safeguards run on every scan:

• ReDoS protection — user-supplied regexes are validated by a static heuristic + a 100 ms stress test at config time, then run with a per-call 50 ms wall-clock ceiling. The classic (a+)+$ class of pathological patterns is rejected at the form.
• Per-project CPU budget — token-bucket limit on scan-microseconds per second. A noisy project can't starve scanning for other projects on the same shard.
• Cluster-wide kill switch — operators can pause SDS for a project instantly via the sds_enabled column or the LiveView toggle. Phoenix.PubSub broadcasts the flip; every node updates within milliseconds.

All operations on SDS — rule CRUD, kill-switch flips, bulk category toggles — are mirrored to /v1/findings as sds.rule.* events so the audit trail is unified with the rest of the security pipeline.

The full curated catalogue, grouped by category. Each row shows the rule's stable builtin_id (used when seeding / referencing rules), its default action, and severity. Patterns tagged noisy seed disabled by default — operators opt in deliberately.

Live-loaded from Funnel.Sds.Library.all/0 at request time — what you see here is exactly what runs in production, never a marketing copy.

Plus two NER kinds (person_name, address) wired through the pluggable adapter for free-form PII the regex layer doesn't see.

A rule has a kind that decides how the pattern is executed and an action that decides what happens on match.

Rule kinds

Actions

Validators (built-in only)

A built-in pattern can attach a checksum validator. Matches that pass the regex but fail the validator are silently filtered out — they don't fire a finding AND they aren't redacted. Eliminates the false-positive class where a random 16-digit number looks like a credit card.

• Luhn — credit cards
• ISO 7064 mod-97 — IBAN
• ABA mod-10 — US routing numbers
• Shannon entropy ≥ 3.0 bits/char — generic secrets, AWS secret keys
• JWT shape — three base64-url segments whose first two decode to JSON

Scopes

Every rule has four scope booleans — scope_logs, scope_traces, scope_metrics, scope_rum — controlling which pillars the scanner applies the rule to. At least one must be true. Per-rule scope toggles are also exposed on the SDS LiveView so you can flip them without leaving the dashboard.

Example: create a custom regex rule

# Create a custom regex rule that redacts internal account IDs.
# Pattern is validated by PatternSafety on save — evil regexes
# (`(a+)+\$`, ambiguous alternation under quantifier, …) are
# rejected before reaching the hot path.

curl -X POST $FUNNEL/api/graphql \\
  -H "Authorization: Bearer st_YOUR_KEY" \\
  -H "Content-Type: application/json" \\
  -d '{
    "query": "mutation { ... }"
  }'

# In Elixir code (e.g. via mix task or seed file):
Funnel.Sds.create_rule(project, %{
  "name"        => "Internal account ID",
  "kind"        => "regex",
  "pattern"     => "acct-[a-z0-9]{12}",
  "action"      => "redact",
  "severity"    => "high",
  "scope_logs"  => true,
  "scope_traces"=> true,
  "scope_rum"   => true
}, actor: "[email protected]")

Sample masking

Findings store a matched_sample that's aggressively masked for critical/high severity rules (***6f (40 chars)) and partially revealed for lower severity (bil...com). The plaintext never leaves the shard — only the mask survives into the database.

Regex catches structured secrets — credit cards, JWTs, API keys. It misses free-form PII: someone's name in a support log, a customer address in a checkout error. The NER (Named Entity Recognition) layer adds a second detector for those cases.

The default detector is a pure-Elixir heuristic engine combining Census gazetteers, structural anchors (street suffix + state code + ZIP), and confidence scoring. No model weights to download, no inference latency. Findings only fire above a configurable threshold (default 0.5).

Confidence model — person names

Confidence model — addresses

Worked example

Input: Approved by Dr. Sarah Johnson at 1234 Main Street, Springfield, IL 62701

Creating a NER rule

Funnel.Sds.create_rule(project, %{
  "name"        => "Customer PII (names + addresses)",
  "kind"        => "ner",
  # entity filter: "person_name" | "address" | "any"
  "pattern"     => "any",
  "action"      => "redact",
  "severity"    => "high",
  "scope_logs"  => true,
  "scope_rum"   => true
}, actor: "[email protected]")

# Hot-path result on:  "Approved by Dr. Sarah Johnson"
# → "Approved by Dr. [REDACTED:rule_42:person_name]"

The pattern field on a NER rule carries the entity kind: person_name, address, or any to match both. Action and severity work the same as on regex rules.

Pluggable ML backend

NER detection goes through Funnel.Sds.NerAdapter — a behaviour with one callback, detect/2. The default adapter is the heuristic engine. For projects that need true ML recall (free-form addresses, non-US names, multi-language PII), implement the behaviour against a Bumblebee NER model and set:

defmodule MyApp.BumblebeeNer do
  @behaviour Funnel.Sds.NerAdapter

  @impl true
  def detect(text, _opts) do
    # Run Bumblebee.Text.token_classification against a BERT NER
    # checkpoint and return entities in the adapter's shape.
    results = Nx.Serving.batched_run(BertNer, text)

    Enum.map(results.entities, fn e ->
      %{
        kind:       map_label(e.label),  # :person_name | :address | :other
        text:       e.phrase,
        offset:     e.start,
        length:     String.length(e.phrase),
        confidence: e.score,
        reasons:    [:ml]
      }
    end)
  end

  defp map_label("PER"), do: :person_name
  defp map_label("LOC"), do: :address
  defp map_label(_),     do: :other
end

# config/runtime.exs
config :funnel, :sds_ner_adapter, MyApp.BumblebeeNer

Funnel's hot path stays unchanged — every other component (rule storage, finding writer, sample masking, webhook fan-out) works identically with either adapter.

Every match is recorded as one row in sds_findings. The row carries snapshots of the rule's name, pattern, and action at match time — so deleting a rule later doesn't orphan history. A stable 16-hex-char fingerprint identifies the matched value (de-duplicated normalisation; rule + match), letting you count distinct secrets without storing them.

Finding row shape

Webhook fan-out

Critical and high-severity findings are pushed to every webhook_destinations row configured on the project (the same table that alerts use — one URL, two event types). The delivery worker retries 5× with exponential backoff and HMAC-signs the body when the destination has a secret.

{
  "type": "sds.finding.created",
  "delivered_at": "2026-05-17T13:43:40.417Z",
  "finding": {
    "id":             "abc123de-...",
    "time":           "2026-05-17T13:43:40Z",
    "project_id":     42,
    "rule_id":        7,
    "builtin_id":     "credit_card",
    "rule_name":      "Built-in: Credit Card Number",
    "severity":       "critical",
    "source_kind":    "logs",
    "source_id":      "msg-001",
    "action_taken":   "redact",
    "matched_sample": "***11 (19 chars)",
    "service_name":   "checkout",
    "field_name":     "message",
    "hits_count":     1,
    "fingerprint":    "129a20e901e6e335"
  }
}

The x-funnel-event-type header is set to sds.finding.created so receivers can route SDS events separately from alert.fired.

Retention

Funnel.Sds.RetentionJob runs at 04:15 UTC daily. For each project it (a) drops any whole monthly partition whose range falls fully below the max-retention across all projects, then (b) batch- DELETEs older rows in the boundary partition using LIMIT 50_000 chunks to avoid long-running locks. Per-tier retention applies — see Quotas & tiers.

A unified read API for everything you've ingested. Same Bearer token as the write paths, same rate-limit primitive (token bucket per project), but exposes the underlying query layer over plain JSON HTTP so you can build custom dashboards, ML pipelines, incident-response runbooks, or exports.

Endpoints

Example: rolling p95 latency

# p95 of http.server.duration_ms over the last 5m, in 10s buckets,
# grouped by service:
curl -s "$URL/api/v1/metrics/query?\
name=http.server.duration_ms&\
from=2026-05-13T17:00:00Z&\
to=2026-05-13T17:05:00Z&\
agg=p95&\
bucket=10&\
group_by=service" \
  -H "authorization: Bearer st_YOUR_API_KEY"

Rate-limit headers

Every response includes:

• X-RateLimit-Limit — bucket capacity
• X-RateLimit-Policy — capacity;w=window (in seconds)
• Retry-After — only on 429, seconds to wait

Project-scoped LLM gateway. Single endpoint sits in front of Anthropic, OpenAI, and a built-in echo provider (for dev/tests). Every call is rate-limited, cost-tracked, recorded in ai_requests, and — uniquely to Funnel — emitted back into the project's own metrics as ai.request.latency_ms, ai.request.cost_cents, and token counters. So your AI usage shows up in the Metrics Explorer next to your app telemetry.

Endpoint

Request

curl -s -X POST $URL/v1/ai/chat \
  -H "authorization: Bearer st_YOUR_API_KEY" \
  -H "content-type: application/json" \
  -d '{
    "messages": [
      {"role": "system", "content": "You are a senior SRE."},
      {"role": "user",   "content": "Summarize the last hour of error logs for service=api."}
    ],
    "max_tokens": 512
  }'

Response shape

{
  "content": "Over the last hour there were 47 errors across 'api' …",
  "cost_cents": 0.299,
  "latency_ms": 1240,
  "model": "claude-3-5-sonnet-latest",
  "provider": "anthropic",
  "usage": {
    "completion_tokens": 174,
    "prompt_tokens": 132,
    "total_tokens": 306
  }
}

Error responses

Provider matrix

Pricing is approximate, used for budget tracking only — always verify against the provider's current rates.

Funnel's load-balancing strategy is straightforward: run multiple BEAM nodes connected via libcluster, fronted by any standard L4/L7 load balancer (nginx, Caddy, AWS ALB, fly.io edge). Phoenix.PubSub is cluster-aware out of the box, so dashboards on one node see telemetry ingested on another within milliseconds.

For workloads that benefit from cache locality (the ETS rate-limit buckets, in-memory anomaly baselines), the Funnel.LoadBalancer module exposes a deterministic consistent-hash router that maps a project_id to a "home node":

# Decide whether to handle this project locally or RPC to its home node:
case Funnel.LoadBalancer.where_should_handle(project_id) do
  :local ->
    handle_locally(project_id, payload)

  {:remote, node} ->
    :rpc.call(node, MyModule, :handle, [project_id, payload])
end

Probe endpoints

Three endpoints, intentionally unauthenticated, designed for load balancers and orchestration systems.

Suggested LB config (nginx)

upstream funnel_app {
  server funnel-1:4000 max_fails=2 fail_timeout=15s;
  server funnel-2:4000 max_fails=2 fail_timeout=15s;
  keepalive 32;
}

server {
  listen 443 ssl http2;

  # Health checks the upstreams
  location = /lb-check { proxy_pass http://funnel_app/ready; }

  location / {
    proxy_pass http://funnel_app;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_set_header X-Forwarded-For   $remote_addr;
    proxy_set_header X-Forwarded-Proto $scheme;
  }

  # WebSockets (LiveView)
  location /live {
    proxy_pass http://funnel_app;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "Upgrade";
  }
}

Kubernetes probes

livenessProbe:
  httpGet:  { path: /health, port: 4000 }
  initialDelaySeconds: 10
  periodSeconds: 10
readinessProbe:
  httpGet:  { path: /ready, port: 4000 }
  initialDelaySeconds: 5
  periodSeconds: 5
  failureThreshold: 3

When an alert rule fires, Funnel POSTs this body to every enabled webhook destination for the project. Signed with HMAC-SHA256 via X-Funnel-Signature: sha256=… (when a webhook secret is configured).

{
  "fired_at": "2026-05-11T17:03:12.483Z",
  "message": "error_rate for api = 0.084 (threshold 0.05)",
  "project_id": 1,
  "rule": {
    "id": 42,
    "kind": "error_rate",
    "name": "API error rate"
  },
  "severity": "critical",
  "type": "alert.fired",
  "value": 0.084
}

Verifying the signature (Node.js)

const crypto = require("crypto");

function verify(rawBody, signatureHeader, secret) {
  const expected =
    "sha256=" +
    crypto.createHmac("sha256", secret).update(rawBody).digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(signatureHeader),
    Buffer.from(expected)
  );
}

Funnel speaks OpenTelemetry, so anything that can emit OTLP works out of the box. Below is a copy-and-paste recipe for every one of the 33 curated connectors. In every example, replace YOUR-FUNNEL with your host and st_YOUR_KEY with an API key created in API keys.

# values.yaml for the open-telemetry/opentelemetry-collector chart
mode: daemonset
config:
  receivers:
    otlp:        { protocols: { http: {}, grpc: {} } }
    kubeletstats: { collection_interval: 30s, auth_type: serviceAccount }
    filelog:     { include: [/var/log/pods/*/*/*.log] }
  exporters:
    otlphttp:
      endpoint: https://YOUR-FUNNEL/v1
      headers:
        authorization: "Bearer st_YOUR_KEY"
  service:
    pipelines:
      metrics: { receivers: [otlp, kubeletstats], exporters: [otlphttp] }
      traces:  { receivers: [otlp],               exporters: [otlphttp] }
      logs:    { receivers: [otlp, filelog],      exporters: [otlphttp] }

# Install:
#   helm repo add open-telemetry https://open-telemetry.github.io/opentelemetry-helm-charts
#   helm install otel open-telemetry/opentelemetry-collector -f values.yaml

# otel.yaml
receivers:
  docker_stats: { collection_interval: 30s }
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [docker_stats], exporters: [otlphttp] }

# Run:
docker run -d --name otel-collector \
  -v $PWD/otel.yaml:/etc/otelcol/config.yaml \
  -v /var/run/docker.sock:/var/run/docker.sock:ro \
  otel/opentelemetry-collector:latest

# adot-config.yaml
receivers:
  awscloudwatchmetrics:
    region: us-east-1
    metrics:
      namespaces: [AWS/EC2, AWS/RDS, AWS/Lambda]
    collection_interval: 60s
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [awscloudwatchmetrics], exporters: [otlphttp] }

# Run via ECS task definition or:
#   aws-otel-collector --config adot-config.yaml

receivers:
  azuremonitor:
    subscription_id: ${AZURE_SUB}
    tenant_id:       ${AZURE_TENANT}
    client_id:       ${AZURE_CLIENT}
    client_secret:   ${AZURE_SECRET}
    collection_interval: 60s
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [azuremonitor], exporters: [otlphttp] }

receivers:
  googlecloudmonitoring:
    project_id: my-gcp-project
    collection_interval: 60s
    metrics_list:
      - metric_name: compute.googleapis.com/instance/cpu/utilization
      - metric_name: cloudsql.googleapis.com/database/cpu/utilization
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [googlecloudmonitoring], exporters: [otlphttp] }

# /etc/cron.d/funnel-heartbeat — runs every minute
* * * * * root /usr/local/bin/funnel-heartbeat.sh

# /usr/local/bin/funnel-heartbeat.sh
#!/usr/bin/env bash
set -eu
FUNNEL=https://YOUR-FUNNEL
KEY=st_YOUR_KEY
load=$(awk '{print $1}' /proc/loadavg)
mem=$(free | awk '/Mem:/ {printf "%.1f", $3/$2*100}')
curl -sS -X POST "$FUNNEL/v1/hosts/heartbeat" \
  -H "Authorization: Bearer $KEY" \
  -H "Content-Type: application/json" \
  -d "{\"hostname\":\"$(hostname)\",\"load\":$load,\"mem_used_pct\":$mem}"

# mix.exs
defp deps, do: [
  {:funnel_agent, "~> 0.1"}
]

# lib/my_app/application.ex
def start(_type, _args) do
  FunnelAgent.start(
    endpoint:     "https://YOUR-FUNNEL/v1",
    api_key:      System.fetch_env!("FUNNEL_KEY"),
    service_name: "my-app"
  )

  children = [...]
  Supervisor.start_link(children, strategy: :one_for_one, name: MyApp.Supervisor)
end

# Auto-instruments :phoenix, :ecto, :oban, :bandit.

// npm i @opentelemetry/sdk-node \
//   @opentelemetry/auto-instrumentations-node \
//   @opentelemetry/exporter-trace-otlp-http

// instrumentation.js
const { NodeSDK } = require('@opentelemetry/sdk-node');
const { OTLPTraceExporter } = require('@opentelemetry/exporter-trace-otlp-http');
const { getNodeAutoInstrumentations } = require('@opentelemetry/auto-instrumentations-node');

new NodeSDK({
  serviceName: 'my-app',
  traceExporter: new OTLPTraceExporter({
    url: 'https://YOUR-FUNNEL/v1/traces',
    headers: { authorization: 'Bearer st_YOUR_KEY' }
  }),
  instrumentations: [getNodeAutoInstrumentations()]
}).start();

// Run:  node -r ./instrumentation.js app.js

pip install opentelemetry-distro opentelemetry-exporter-otlp
opentelemetry-bootstrap -a install

OTEL_SERVICE_NAME=my-app \
OTEL_EXPORTER_OTLP_ENDPOINT=https://YOUR-FUNNEL \
OTEL_EXPORTER_OTLP_HEADERS="authorization=Bearer st_YOUR_KEY" \
OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf \
opentelemetry-instrument python app.py

// go get go.opentelemetry.io/otel \
//   go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp \
//   go.opentelemetry.io/otel/sdk/trace

import (
  "context"
  "go.opentelemetry.io/otel"
  "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
  sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

func initTracer(ctx context.Context) (*sdktrace.TracerProvider, error) {
  exp, err := otlptracehttp.New(ctx,
    otlptracehttp.WithEndpoint("YOUR-FUNNEL"),       // no scheme
    otlptracehttp.WithURLPath("/v1/traces"),
    otlptracehttp.WithHeaders(map[string]string{
      "authorization": "Bearer st_YOUR_KEY",
    }),
  )
  if err != nil { return nil, err }
  tp := sdktrace.NewTracerProvider(sdktrace.WithBatcher(exp))
  otel.SetTracerProvider(tp)
  return tp, nil
}

# Download opentelemetry-javaagent.jar from
# https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases

java -javaagent:opentelemetry-javaagent.jar \
     -Dotel.service.name=my-app \
     -Dotel.exporter.otlp.endpoint=https://YOUR-FUNNEL \
     -Dotel.exporter.otlp.headers="authorization=Bearer st_YOUR_KEY" \
     -Dotel.exporter.otlp.protocol=http/protobuf \
     -jar app.jar

# Gemfile
gem 'opentelemetry-sdk'
gem 'opentelemetry-exporter-otlp'
gem 'opentelemetry-instrumentation-all'

# config/initializers/otel.rb (Rails) or boot.rb
require 'opentelemetry/sdk'
require 'opentelemetry/exporter/otlp'
require 'opentelemetry/instrumentation/all'

ENV['OTEL_SERVICE_NAME']         ||= 'my-app'
ENV['OTEL_EXPORTER_OTLP_ENDPOINT'] ||= 'https://YOUR-FUNNEL'
ENV['OTEL_EXPORTER_OTLP_HEADERS']  ||= 'authorization=Bearer st_YOUR_KEY'

OpenTelemetry::SDK.configure { |c| c.use_all }

# Cargo.toml
[dependencies]
opentelemetry      = "0.21"
opentelemetry_sdk  = { version = "0.21", features = ["rt-tokio"] }
opentelemetry-otlp = { version = "0.14", features = ["http-proto", "reqwest-client"] }

// src/main.rs
use opentelemetry_otlp::WithExportConfig;
use std::collections::HashMap;

fn init_tracer() -> Result<(), Box<dyn std::error::Error>> {
  let mut headers = HashMap::new();
  headers.insert("authorization".into(), "Bearer st_YOUR_KEY".into());

  opentelemetry_otlp::new_pipeline()
    .tracing()
    .with_exporter(
      opentelemetry_otlp::new_exporter()
        .http()
        .with_endpoint("https://YOUR-FUNNEL/v1/traces")
        .with_headers(headers),
    )
    .install_batch(opentelemetry_sdk::runtime::Tokio)?;
  Ok(())
}

# .github/workflows/build.yml
jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - id: timer
        run: echo "started=$(date +%s)" >> $GITHUB_OUTPUT
      - run: ./build.sh

      - name: Report to Funnel
        if: always()
        env:
          FUNNEL_KEY: ${{ secrets.FUNNEL_KEY }}
        run: |
          elapsed=$(( $(date +%s) - ${{ steps.timer.outputs.started }} ))
          curl -sS -X POST https://YOUR-FUNNEL/v1/metrics \
            -H "Authorization: Bearer $FUNNEL_KEY" \
            -H "Content-Type: application/json" \
            -d @- <<JSON
          { "metrics": [{
              "name": "ci.github.job.duration_s",
              "kind": "gauge",
              "value": $elapsed,
              "attributes": {
                "workflow": "${{ github.workflow }}",
                "job":      "${{ github.job }}",
                "status":   "${{ job.status }}",
                "repo":     "${{ github.repository }}"
              }
          }]}
          JSON

# .gitlab-ci.yml
.report_to_funnel: &report_to_funnel
  after_script:
    - |
      curl -sS -X POST https://YOUR-FUNNEL/v1/metrics \
        -H "Authorization: Bearer $FUNNEL_KEY" \
        -H "Content-Type: application/json" \
        -d "{
          \"metrics\": [{
            \"name\":  \"ci.gitlab.job.duration_s\",
            \"kind\":  \"gauge\",
            \"value\": $CI_JOB_DURATION,
            \"attributes\": {
              \"job\":    \"$CI_JOB_NAME\",
              \"status\": \"$CI_JOB_STATUS\",
              \"ref\":    \"$CI_COMMIT_REF_NAME\"
            }
          }]
        }"

build:
  <<: *report_to_funnel
  script: ./build.sh

1. Manage Jenkins → Plugins → install "OpenTelemetry".
2. Manage Jenkins → Configuration → OpenTelemetry.
3. Endpoint:     https://YOUR-FUNNEL
4. Headers:      authorization=Bearer st_YOUR_KEY
5. Protocol:     OTLP HTTP/protobuf  (or gRPC if FUNNEL_GRPC_ENABLED=1, port 4317)
6. Save.

Every build now exports a root span "ci.pipeline.run" with
child spans per stage/step, plus build duration and outcome metrics.

# Add a notification webhook to argocd-notifications-cm ConfigMap:
data:
  service.webhook.funnel: |
    url: https://YOUR-FUNNEL/v1/logs
    headers:
      - name: Authorization
        value: "Bearer st_YOUR_KEY"
      - name: Content-Type
        value: application/json
  template.app-sync-status: |
    webhook:
      funnel:
        method: POST
        body: |
          { "logs": [{
            "severity": "info",
            "service_name": "argocd",
            "message": "{{.app.metadata.name}} sync {{.app.status.sync.status}}",
            "attributes": {
              "app":      "{{.app.metadata.name}}",
              "revision": "{{.app.status.sync.revision}}"
            }
          }]}
  trigger.on-sync-status-change: |
    - when: app.status.sync.status in ['Synced', 'OutOfSync']
      send: [app-sync-status]

# Make sure Kafka has JMX exposed (KAFKA_JMX_PORT=9999).
receivers:
  jmx:
    jar_path: /opt/opentelemetry-jmx-metrics.jar
    endpoint: kafka:9999
    target_system: kafka
    collection_interval: 30s
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [jmx], exporters: [otlphttp] }

# Exposes:  kafka.message.count, kafka.partition.lag, kafka.consumer.lag, ...

# Enable plugin: rabbitmq-plugins enable rabbitmq_management
receivers:
  rabbitmq:
    endpoint: http://rabbitmq:15672
    username: ${RABBIT_USER}
    password: ${RABBIT_PASS}
    collection_interval: 30s
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [rabbitmq], exporters: [otlphttp] }

# Exposes:  rabbitmq.consumer.count, rabbitmq.message.published, queue.size, ...

receivers:
  redis:
    endpoint: redis:6379
    password: ${REDIS_PASS}     # omit if disabled
    collection_interval: 30s
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [redis], exporters: [otlphttp] }

# Grant the read-only monitoring user pg_monitor first:
--   CREATE USER monitoring WITH PASSWORD 'xxx';
--   GRANT pg_monitor TO monitoring;

receivers:
  postgresql:
    endpoint:  postgres:5432
    transport: tcp
    username:  monitoring
    password:  ${PG_PASS}
    databases: [mydb]
    collection_interval: 30s
    tls:       { insecure: true }
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [postgresql], exporters: [otlphttp] }

# Grant the monitoring user:
--   CREATE USER 'monitoring'@'%' IDENTIFIED BY 'xxx';
--   GRANT PROCESS, REPLICATION CLIENT, SELECT ON *.* TO 'monitoring'@'%';

receivers:
  mysql:
    endpoint: mysql:3306
    username: monitoring
    password: ${MYSQL_PASS}
    collection_interval: 30s
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [mysql], exporters: [otlphttp] }

# Create a user with clusterMonitor role:
# use admin
# db.createUser({ user: "monitoring", pwd: "xxx",
#   roles: [{ role: "clusterMonitor", db: "admin" }] })

receivers:
  mongodb:
    hosts:
      - endpoint: mongo:27017
    username: monitoring
    password: ${MONGO_PASS}
    collection_interval: 30s
exporters:
  otlphttp:
    endpoint: https://YOUR-FUNNEL/v1
    headers: { authorization: "Bearer st_YOUR_KEY" }
service:
  pipelines:
    metrics: { receivers: [mongodb], exporters: [otlphttp] }

# .github/workflows/secret-alert-forward.yml
on:
  secret_scanning_alert: { types: [created, resolved] }

jobs:
  forward:
    runs-on: ubuntu-latest
    steps:
      - run: |
          curl -sS -X POST https://YOUR-FUNNEL/v1/findings \
            -H "Authorization: Bearer ${{ secrets.FUNNEL_KEY }}" \
            -H "Content-Type: application/json" \
            -d @- <<JSON
          {
            "source":   "secret_scanner",
            "severity": "high",
            "title":    "${{ github.event.alert.secret_type_display_name }}",
            "target":   "${{ github.repository }}",
            "detail":   "${{ github.event.alert.html_url }}"
          }
          JSON

# Scan, then post each vulnerability:
trivy fs --format json --output trivy.json .

jq -c '.Results[]?.Vulnerabilities[]? | {
  source:   "vuln_scanner",
  severity: (.Severity | ascii_downcase),
  title:    (.VulnerabilityID + " in " + .PkgName),
  target:   .PkgName,
  detail:   (.Title // .Description // "")
}' trivy.json | while read -r body; do
  curl -sS -X POST https://YOUR-FUNNEL/v1/findings \
    -H "Authorization: Bearer $FUNNEL_KEY" \
    -H "Content-Type: application/json" \
    -d "$body"
done

# falco.yaml
json_output: true
json_include_output_property: true

http_output:
  enabled:    true
  url:        "https://YOUR-FUNNEL/v1/findings"
  user_agent: "falco"
  insecure:   false
  headers:
    - "Authorization: Bearer st_YOUR_KEY"
    - "Content-Type: application/json"

# Funnel accepts Falco's native envelope and maps:
#   priority   -> severity
#   rule       -> title
#   output     -> detail
#   output_fields.container.name -> target

# Lambda triggered by S3 PUT on the CloudTrail log bucket.
# Forwards every event with errorCode set as a finding.
import boto3, gzip, json, os, urllib3
http = urllib3.PoolManager()

def lambda_handler(event, _):
    s3 = boto3.client('s3')
    for rec in event['Records']:
        obj  = s3.get_object(
            Bucket=rec['s3']['bucket']['name'],
            Key=rec['s3']['object']['key'])
        body = json.loads(gzip.decompress(obj['Body'].read()))
        for e in body.get('Records', []):
            if not e.get('errorCode'):
                continue
            http.request('POST',
                os.environ['FUNNEL'] + '/v1/findings',
                headers={
                  'Authorization': 'Bearer ' + os.environ['FUNNEL_KEY'],
                  'Content-Type':  'application/json',
                },
                body=json.dumps({
                  'source':   'cloudtrail',
                  'severity': 'medium',
                  'title':    e['errorCode'],
                  'target':   e.get('userIdentity', {}).get('arn', 'unknown'),
                  'detail':   e.get('errorMessage', ''),
                }))

# Run on EventBridge cron(0 6 * * *)
# IAM:  ce:GetCostAndUsage
import boto3, requests, os, datetime as dt

yesterday = (dt.date.today() - dt.timedelta(days=1)).isoformat()
today     = dt.date.today().isoformat()

ce = boto3.client('ce')
resp = ce.get_cost_and_usage(
    TimePeriod={'Start': yesterday, 'End': today},
    Granularity='DAILY',
    Metrics=['UnblendedCost'],
    GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}])

records = [{
    'date':     yesterday,
    'service':  g['Keys'][0],
    'cost_usd': float(g['Metrics']['UnblendedCost']['Amount']),
    'provider': 'aws'
} for g in resp['ResultsByTime'][0]['Groups']]

requests.post(
    os.environ['FUNNEL'] + '/v1/cost/records',
    headers={'Authorization': 'Bearer ' + os.environ['FUNNEL_KEY']},
    json={'records': records})

# Prereq: enable Billing Export to BigQuery.
# Run on Cloud Scheduler daily.
from google.cloud import bigquery
import requests, os

bq = bigquery.Client()
query = '''
SELECT service.description AS service,
       SUM(cost)             AS cost_usd,
       DATE(usage_start_time) AS day
FROM `proj.billing.gcp_billing_export_v1_XXXXXX_*`
WHERE DATE(usage_start_time) = CURRENT_DATE() - 1
GROUP BY service, day
'''
records = [{
  'date':     str(r['day']),
  'service':  r['service'],
  'cost_usd': float(r['cost_usd']),
  'provider': 'gcp'
} for r in bq.query(query).result()]

requests.post(
    os.environ['FUNNEL'] + '/v1/cost/records',
    headers={'Authorization': 'Bearer ' + os.environ['FUNNEL_KEY']},
    json={'records': records})

1. In Slack:  Apps  →  Incoming Webhooks  →  Add to Workspace.
               Pick a channel, copy the URL:
               https://hooks.slack.com/services/T.../B.../...

2. In Funnel:  Alerts  →  New rule  →  Destination  →  Webhook.
               Paste the URL.   Save.

Funnel POSTs a Slack-shaped payload:
  { "text": "Alert: error_rate > 0.05 on service=api",
    "attachments": [{ "color": "danger", "fields": [...] }] }

No transformation required.

1. PagerDuty:  Service  →  Integrations  →  + Events API v2.
                Copy the Integration Key (32 hex chars).

2. Funnel:  Alerts  →  New rule  →  Destination  →  PagerDuty.
             Paste the Integration Key.   Save.

Funnel POSTs to https://events.pagerduty.com/v2/enqueue:
  { "routing_key":  "<INTEGRATION_KEY>",
    "event_action": "trigger" | "resolve",
    "payload": {
      "summary":  "<alert name> fired",
      "severity": "critical" | "warning" | "info",
      "source":   "funnel"
    }
  }

Resolution events fire automatically when the rule un-breaches.

1. In Teams:  Channel  →  ⋯  →  Connectors  →  Incoming Webhook.
               Name it "Funnel alerts".  Copy the webhook URL.

2. In Funnel:  Alerts  →  New rule  →  Destination  →  Webhook.
                Paste the URL.   Funnel detects the
                outlook.office.com host and posts a
                MessageCard-formatted body automatically.

No transformation required.

# In Funnel UI:
#   Alerts  →  New rule  →  Destination  →  Email.
#   Enter one or more recipient addresses.

# SMTP is configured once, project-wide, in config/runtime.exs:
config :funnel, Funnel.Mailer,
  adapter:  Swoosh.Adapters.SMTP,
  relay:    System.fetch_env!("SMTP_HOST"),
  port:     587,
  username: System.fetch_env!("SMTP_USER"),
  password: System.fetch_env!("SMTP_PASS"),
  tls:      :always,
  auth:     :always,
  retries:  2

# Works out of the box with Mailtrap, SES, Postmark, SendGrid.

Funnel exposes a project-scoped GraphQL API for everything SDS-related: findings (with cursor pagination + multi-filter), rule statistics, per-pillar breakdowns, writer health, CPU budget, and live subscriptions for newly-created findings. Same Bearer-token auth as the REST API; rate-limited at 8 req/sec per project.

Backed by Absinthe 1.7 with a real introspectable schema, enums (uppercase names: CRITICAL, LOGS, etc.), custom DateTime + JSON scalars, and cursor-paginated connections. Subscription delivery uses Phoenix.PubSub over the existing /live socket — no extra server to run.

Authentication

The same Authorization: Bearer st_… header that gates the REST API gates GraphQL. The API key resolves to a project; the FunnelWeb.GraphQL.ContextPlug puts that project on the Absinthe context and every resolver reads it from there. The client cannot pass a projectId argument to query a different project — the API key is the only project selector.

Worked example

# Paginated finding list with multi-filter.
curl -X POST $FUNNEL/api/graphql \\
  -H "Authorization: Bearer st_YOUR_KEY" \\
  -H "Content-Type: application/json" \\
  -d '{
    "query": "query Findings(\$first: Int!, \$sev: Severity) {
      findings(first: \$first, filter: { severity: \$sev }) {
        nodes {
          id time severity ruleName matchedSample
          sourceKind serviceName hitsCount fingerprint
        }
        pageInfo { hasNextPage endCursor }
        totalCount
      }
    }",
    "variables": { "first": 25, "sev": "CRITICAL" }
  }'

Errors come back in the standard { data: …, errors: [{ message }] } shape. The most common error is "unauthenticated" when the API key resolves to no project, and field-level validation errors when an enum value is misspelled (use CRITICAL, not critical).

Single dashboard query

GraphQL's point is composing many queries in one HTTP round-trip. The SDS dashboard fetches everything it needs with one query:

# One round-trip for a custom SDS dashboard.
{
  summary: findings(first: 1) { totalCount }
  byKind: bySourceKind(hours: 24) { sourceKind count }
  services: topServices(hours: 24, limit: 5) { service count }
  rules: topRules(hours: 24, limit: 5) {
    ruleName count severity
  }
  budget { tokens deniedCount }
  writerStats {
    written buffered dropped
    perShard { shard written batches }
  }
}

Cursor pagination

Connection-style pagination using opaque cursors. Decode-on-server: the cursor is a base64-url-encoded (time, uuid) tuple. Pass the previous page's pageInfo.endCursor back as after to fetch the next page.

{
  topRules(hours: 24, limit: 10) {
    ruleName
    builtinId
    severity
    actionTaken
    count
    uniqueFingerprints
    firstSeen
    lastSeen
  }
}

For windows up to 30 days the resolver returns a totalCount; for longer windows the count is null to avoid a slow COUNT(*) against the partitioned table.

GraphQL subscription { findingCreated } delivers every newly-persisted finding to subscribed clients in real time. Each shard flush publishes to Phoenix.PubSub topic sds:findings:<project_id> (plus a per-severity variant), and Absinthe's subscription router fans the message out to every matching client.

JavaScript example

# Connect via the Phoenix Socket — channel topic
# "__absinthe__:control" handles GraphQL subscription frames.

import { Socket } from "phoenix";
import * as AbsintheSocket from "@absinthe/socket";
import { createAbsintheSocketLink } from "@absinthe/socket-apollo-link";

const phx = new Socket("/socket", {
  params: { token: "st_YOUR_KEY" }
});
const absintheSocket = AbsintheSocket.create(phx);

AbsintheSocket.observe(absintheSocket, {
  operation: \`subscription {
    findingCreated(severity: CRITICAL) {
      ruleName matchedSample serviceName sourceKind
    }
  }\`,
  variables: {}
}, {
  onResult: (data) => console.log("finding:", data)
});

Subscriptions can filter by severity at subscription-time: findingCreated(severity: CRITICAL). Filtering happens at the topic level, so a CRITICAL-only subscriber doesn't receive INFO findings over the wire.

Condensed SDL view of the schema. Introspection is enabled, so tools like Insomnia, Apollo DevTools, and GraphiQL can autocomplete fields and validate queries before sending them.

type Query {
  findings(filter: FindingFilter, first: Int, after: String): FindingConnection!
  finding(id: ID!): Finding
  topRules(hours: Int = 24, limit: Int = 10): [RuleStat!]!
  bySourceKind(hours: Int = 24): [SourceKindStat!]!
  topServices(hours: Int = 24, limit: Int = 10): [ServiceStat!]!
  budget: Budget!
  writerStats: WriterStats!
  health: String!
}

type Subscription {
  findingCreated(severity: Severity): Finding!
}

input FindingFilter {
  severity:    Severity
  sourceKind:  SourceKind
  service:     String
  ruleId:      Int
  builtinId:   String
  fingerprint: String
  since:       DateTime
  until:       DateTime
}

enum Severity   { CRITICAL HIGH MEDIUM LOW INFO }
enum SourceKind { LOGS SPANS RUM METRICS SESSIONS }
enum ActionKind { ALERT REDACT HASH DROP }
scalar DateTime
scalar JSON

type Finding {
  id:              ID!
  time:            DateTime!
  ruleId:          Int
  builtinId:       String
  ruleName:        String
  patternSnapshot: String
  severity:        Severity!
  sourceKind:      SourceKind!
  sourceId:        String
  actionTaken:     ActionKind!
  matchedSample:   String
  serviceName:     String
  fieldName:       String
  hitsCount:       Int!
  fingerprint:     String
  attributes:      JSON
}

type FindingConnection {
  edges:      [FindingEdge!]!
  nodes:      [Finding!]!
  pageInfo:   PageInfo!
  totalCount: Int
}

Type details

Operational caveats

• Enum input form: uppercase names. severity: CRITICAL not "critical".
• Date input: ISO-8601 with a Z timezone: since: "2026-05-17T00:00:00Z".
• Max per-page: first: 200; larger values are silently clamped.
• GraphiQL: only mounted in dev/test — production never exposes the introspection-heavy playground at the edge. Use a desktop client.