Telemetry Sinks

The RIC's own metrics (Observability) describe the platform. An xApp's own output — the throughput it measured, the handover it decided, the SRS estimate it computed — flows out through the SDK's sinks: a push-telemetry layer configured entirely in xapp.yml. The Python and Go bindings implement it identically — the same xapp.yml produces the same wire traffic from either. The C client ships no sink layer, so a C xApp emits nothing through sinks.

Two kinds of telemetry

Kind	Cardinality	Shape	Typical sinks
timeseries	aggregated	numeric samples + tags	Redis Pub/Sub, InfluxDB v2, file / JSONL
structured	per-event	typed record + nested fields	Redis Pub/Sub, file / JSONL

The split is deliberate: timeseries are rates and distributions that are safe to drop and downsample; structured records are discrete events the consumer expects to land intact (subject to bounded-queue overflow).

Configuration

airpuls-sdk xapp add sink writes these blocks for you

The CLI knows the keys each destination needs, validates the manifest before saving, and refuses a secret passed as a literal — see the CLI. The reference below is what it produces, and what to write by hand.

Sinks live under a top-level sinks: block, keyed sinks.{kind}.{type}. Every sink defaults to enabled: false; turning one on requires an explicit enabled: true. If the sinks: block is omitted entirely, no push sinks start.

sinks:
  timeseries:
    redis_pubsub:
      enabled: true
      url: redis://redis:6379/0
      channel: airpuls.xapp.timeseries.{xapp_instance}
      password_env: REDIS_PASSWORD       # env var name — see "Secrets" below
    influxdb_v2:
      enabled: false
      url: http://influx:8086
      org: airpuls
      bucket: xapp_metrics
      token_env: INFLUX_TOKEN
    file:
      enabled: false
      path: ./data/metrics.jsonl        # cwd-relative; default ./xapp-metrics.jsonl
      rotate_mb: 100
      keep: 5

  structured:
    file:
      enabled: true
      path: ./data/events.jsonl          # cwd-relative; default ./xapp-events.jsonl
      rotate_mb: 100
      keep: 5
    redis_pubsub:
      enabled: true
      url: redis://redis:6379/0
      channel: airpuls.xapp.events.{xapp_instance}
      password_env: REDIS_PASSWORD
      queue_size: 1024                   # per-sink hand-off depth (default 1024)
      queue_max_bytes: 67108864          # per-sink memory limit (default 64 MiB)

Those are the built-in sinks — the complete set of type values a binding will start:

type	timeseries	structured	Transport
redis_pubsub	✅	✅	one PUBLISH per item
influxdb_v2	✅	—	one HTTP POST per sample
file	✅	✅	JSONL, size-rotated

XAPP_SINKS.md §5 also specifies otlp, kafka, stdout, and websocket. None of them is implemented in either binding. Sink boot walks the registry of registered types, so a type no binding registers is never instantiated — the block sits inert in xapp.yml and nothing reports it.

Sinks fan out: one metrics.sample(...) reaches every enabled timeseries sink; one events.record(...) reaches every enabled structured sink. Each configured sink runs its own writer thread and its own bounded hand-off queue.

Channel templating

channel: is a template — {xapp_instance}, {xapp_type}, and {xapp_version} are substituted once, when the sink is constructed at boot, from the identity the SDK fills in after the RIC connect. A known placeholder whose identity field is empty becomes the literal unknown; an unknown placeholder is left intact, so a typo routes to a visibly broken channel rather than a half-resolved one. This gives each xApp deployment its own channel out of the box; set a placeholder-free channel to share one.

Secrets: prefer the environment, but inline is accepted

Each secret has two keys — password_env / token_env name an environment variable, password / token carry the value inline. A non-empty env var wins; the inline field is the fallback and logs a warning on use. For Redis, credentials already embedded in url: rank last. Prefer the env form: the inline one puts a live secret into a file the operator may copy or share.

A destination that works locally may not exist once deployed

Every value naming a place — a file path, a host in a URL — is resolved against a different world in a deployed container than on the machine the configuration was written on. A relative path: resolves against the image's working directory, which is root-owned while the container runs as the invoking user; 127.0.0.1 names the xApp's own container. Combined with fire-and-forget, the result is an xApp that looks healthy and produces nothing.

airpuls-sdk xapp deploy resolves what it can in the uploaded copy alone, leaving the project's xapp.yml untouched: a relative sink path moves onto /xapp/data, the directory the launcher mounts from the deployed entry and the only one the container can write. Keep file sinks under the project's data/ directory and the same setting works on both sides. An absolute path and a loopback URL are the operator's own to resolve, so the deploy reports them instead of rewriting them.

A payload the destination cannot encode is reported, not skipped

Fire-and-forget covers transport failures: a broker that is down or a disk that errors loses the item quietly, because it will likely work again shortly. A payload with no encoded form is different — it will fail identically forever — so it is counted on unserializable_total and the first one logs a warning naming the sink. Two cases produce it: a structured Record whose fields are not JSON-serializable (raw bytes are the usual cause — base64 them, or use a shape JSON can carry), and a Sample with no measurement name or no fields, which InfluxDB line protocol cannot express. It is deliberately not counted as a drop: a drop means slow down or resize, this means fix the payload.

Fire-and-forget, drops are sink-local

The hot path does a non-blocking try-send into each sink's bounded queue. If a sink's queue is full the arriving item is dropped for that sink only — a slow or broken sink never blocks the dispatch thread and never affects other sinks. There is no retry or durability guarantee; queue_size caps the burst a sink can absorb (set 1 for strict drop-newest).

Large payloads are bounded by memory, not by item count

queue_size bounds items, not bytes. An xApp whose records carry large arrays — E2SM-AIR channel estimates run to several megabytes each — can pin gigabytes behind one lagging sink while staying far inside its item count. queue_max_bytes (default 64 MiB, 0 disables) is the second bound: the dispatcher estimates each item's retained size and drops the arriving item once a sink's queued bytes would exceed it. Whichever bound is reached first wins, so ordinary small records are still governed by queue_size. Every drop, from either bound, is counted on the sink's xapp_sink metric as dropped_total, and the first one logs a warning naming the sink — a non-zero dropped_total means that sink is not keeping up with what the xApp emits.

A sink that fails to start is retried; one that breaks later is not

xapp.reconnect.* — the same block that governs the RIC connection — also gates the per-sink start() retry loop: enabled by default, 250 ms initial backoff, ×2 up to a 60 s cap, 20% jitter, unlimited attempts. Its scope ends there. A transport that fails after start() returned does not re-trigger reconnect and is not retried per item; it shows up as state=0 on the sink's xapp_sink self-metric below.

What you get for free

One emission, and only one: the dispatcher's own health. Every 10 s it sends an xapp_sink sample per configured sink to the timeseries sinks —

xapp_sink,sink=<type>,kind=<kind> state=<0|1>,start_attempts_total=N,start_failures_total=N,dropped_total=N,unserializable_total=N <ts>

— where state=0 marks a sink that never started or was dropped from the live set, dropped_total counts the items that sink rejected because its queue was full (by item count or by memory), and unserializable_total counts the payloads it had no encoded form for. Since writes are fire-and-forget, these are the only signals a wedged, misconfigured or overwhelmed transport gives you; a dashboard that ingests the timeseries channel gets them without any xApp code.

Standard xApp telemetry is specified but not implemented

XAPP_SINKS.md §6 defines a set of SDK-populated emissions (xapp_uptime_s, e2_nodes_connected, indications_total, xapp_started, subscription_added, control_outcome_received, …). No binding emits them today. Apart from xapp_sink, every sample and record on the wire is one the xApp emitted itself — do not build a dashboard expecting the §6 set to arrive.

Emitting your own telemetry

Two calls — one per kind — fan out to every matching sink:

PythonGo

class MyXApp(BaseXApp):
    def on_indication(self, client, sub_id, node, ran_func_id, header, data):
        metrics = client.metrics()
        for block in data.blocks:
            for m in block.measurements:
                # timeseries: numeric sample + tags
                metrics.sample(
                    "ue_throughput",
                    tags={"ue": block.ue_id.display, "meas": m.name},
                    fields={"value": m.value},
                )
        # structured: one discrete record
        client.events().record(
            "indication_seen",
            tags={"sm": "KPM"},
            fields={"blocks": data.block_count},
        )

metrics() / events() are methods on the client, not attributes of BaseXApp.

func (h *myHandler) OnIndication(client *ric.Client, subID uint32,
    node *ric.E2Node, ranFuncID uint16,
    header *ric.IndicationHeader, data *ric.IndicationData) {
    for _, block := range data.Blocks() {
        for _, m := range block.Measurements() {
            client.Metrics().Sample("ue_throughput",
                map[string]string{"meas": m.Name()},
                map[string]any{"value": m.Integer()})
        }
    }
    client.Events().Record("indication_seen",
        map[string]string{"sm": "KPM"},
        map[string]any{"blocks": len(data.Blocks())})
}

No C equivalent

Sinks live in the Python and Go bindings only — libric-client contains no sink implementation. A C xApp parses the same xapp.yml, but the sinks: block has no effect and nothing is published. Emitting telemetry from C means writing the transport yourself.

Preserving an external timestamp

sample() / record() stamp the item at fan-out with the wall clock. When the datum carries its own time — a KPM indication header, a sensing frame — use the Sample / Record form instead, which keeps the timestamp you set:

PythonGo

from airpuls_ric_sdk.sinks import Sample

client.metrics().emit(Sample(
    measurement="kpm",
    tags={"sm": "KPM", "node": node, "ue": ue},
    fields=fields,
    timestamp_ns=header_ts_ns,
))

client.Metrics().Emit(sinks.Sample{
    Measurement: "kpm",
    Tags:        map[string]string{"sm": "KPM", "node": node},
    Fields:      fields,
    Timestamp:   headerTime,
})

events().emit(Record(...)) / Events().Emit(sinks.Record{...}) do the same for structured records. kpm-monitor uses this path so its samples carry the indication's own time rather than the moment they were dispatched.

What lands on the wire

You do not have to guess what a consumer will see. Each sink has one format, and the SDK auto-injects the xApp identity as tags (xapp_instance, xapp_type, xapp_version) on every item that does not already carry them — so a shared channel stays attributable.

Redis Pub/Sub

One PUBLISH per item, on the channel the template resolved to. In a multi-xApp deployment, subscribe to the pattern:

PSUBSCRIBE airpuls.xapp.timeseries.*
PSUBSCRIBE airpuls.xapp.events.*

Timeseries — InfluxDB line protocol, one line, no trailing newline:

kpm,node=gnb-1,sm=KPM,ue=ue:gnb_cu_ue_f1ap_id:1,xapp_instance=kpm-monitor RRU.PrbTotDl=12 1747834510123456789

Tags and fields are emitted in sorted key order, so equivalent samples produce byte-identical lines. Escaping follows the v2 line-protocol rules, and a tag with an empty key or value is omitted — the grammar has no empty form. The trailing integer is nanoseconds since the Unix epoch. Telegraf's inputs.redis_subscriber parses this natively, so no glue code is needed between an xApp and a dashboarding stack.

Structured — a JSON envelope:

{"kind":"record",
 "event":"subscription_added",
 "tags":{"sm":"KPM","xapp_instance":"kpm-monitor"},
 "fields":{"sub_id":1,"node":"gnb-1","measurement_count":7},
 "ts":1747834510123456789}

Redis Pub/Sub is fire-and-forget at the transport too: a subscriber receives only while connected, with no replay and no acknowledgement.

File (JSONL)

The same JSON envelope, one object per line, appended to path: and rotated at rotate_mb keeping keep generations. The directory is created on start if it does not exist.

InfluxDB v2

One HTTP POST per sample to /api/v2/write, carrying the same line protocol as the Redis timeseries channel, authenticated with the token from token_env. There is no client-side batching — one sample, one request — so this sink suits control-plane rates, not per-slot PHY telemetry. For high-rate streams, publish to Redis and let a collector batch into Influx.

Replay mode

Replay is a test mode. With it on, the SDK opens no SCTP association, registers with no RIC and holds no subscriptions — it replays a directory of recordings through the sinks instead, at the cadence they were recorded at. Every consumer downstream of the sinks sees exactly what it would see live, with no RAN and no RIC attached.

replay:
  enabled: false          # off unless set; the live path is untouched
  dir: ./replay           # directory of *.jsonl recordings
  loop: true              # restart the timeline after the last entry
  untimed_interval_ms: 22 # spacing for a stream that carries no cadence

Making a recording

The two file sinks are the recorder. Enable sinks.timeseries.file and sinks.structured.file on a live run, then copy what they wrote into the replay directory — rotated backups included:

$ cp ./data/*.jsonl* ./replay/

The file sinks rotate, so a run of any length leaves a chain rather than a single file:

events.jsonl      <- newest
events.jsonl.1
events.jsonl.2    <- oldest

Replay reads the whole chain, not just the current file: each <name>.jsonl chain — the current file plus its .N backups — is loaded and paced as its own timeline, so rotation boundaries disappear within a chain. Chains are not merged with one another: events.jsonl and metrics.jsonl are independent streams, each paced by its own recorded gaps, because merging them lets one stream's cadence — or its absence — distort the other. Where two entries in a chain carry the same timestamp, rotation order breaks the tie: the highest .N is oldest.

What replay does with a recording

• Cadence comes from the data. Each entry waits the gap between its own timestamp and its predecessor's. Nothing is declared; a recording carries its own clock.
• A stream with no clock is paced by untimed_interval_ms. A producer writing outside the sink dispatcher leaves every record sharing one timestamp; such a stream has no gaps of its own and would publish as fast as the transport allows, so its entries are spaced at this interval instead (default 22 ms; 0 leaves the recorded gaps alone).
• Looping waits at the wrap. A looping timeline pauses between passes — by untimed_interval_ms for a stream with no cadence, else by the middle of the stream's own gaps. A timeline offering neither is played once and not looped, rather than republishing back to back for as long as the xApp lives.
• Timestamps are re-stamped at publication. The recorded timestamp paces the entry but does not travel to the sinks, so a dashboard querying the last few minutes finds a live window rather than a stream dated to whenever the recording was made.
• Every sink sees it. Substitution happens at the dispatcher, so Redis, InfluxDB and the file sinks all carry the same replayed stream and agree about every instant.
• The xApp's own emits are dropped. Everything reaching a sink comes from the recording. The xapp_sink self-metric is unaffected and keeps reporting real sink state.
• Identity resolves from config. xapp_instance is normally settled by the RIC handshake; under replay it comes from xapp.instance, falling back to xapp.deployment_name, so channel templates still resolve.
• A line that will not decode is skipped, logged once with its file and line number. One bad line never costs the whole recording.

Replayed output is indistinguishable from live

Nothing marks a replayed payload. That is deliberate — a consumer under test should see exactly what production sends — but it means a replay run pointed at a shared broker puts recorded data in front of real dashboards. Point it at a broker of its own.

Do not record into the directory you replay from

If a file sink's path sits inside replay.dir, a replay run records onto its own recordings.

Python xApps on BaseXApp need no code change

BaseXApp.run_sync checks for replay after constructing the client and skips SM registration, callback wiring and subscriptions. Flip replay.enabled in the manifest and run the xApp's real binary:

$ python3 xapps/xapp-replay/python/main.py -c xapp.yml
replay: publishing 412 recorded entries from ./replay across 2 stream(s) (loop=True)

Replay requires run_sync. The async run() entry point refuses it: its child-process model re-reads the same manifest, which would bring up a second dispatcher over the same recordings and publish every entry twice.

Everything else — a Python xApp driving RicClient directly, and every Go xApp, since Go has no BaseXApp — checks client.is_replaying (Go: client.IsReplaying()) and parks on client.wait_for_replay_shutdown() (Go: client.WaitForReplayShutdown()) instead of the dispatch loop. A Go xApp that calls Run, Dispatch, RunIteration, PollFd or any subscribe/control call under replay gets ErrClientReplaying back, naming the mode rather than failing with an opaque status.

Custom sinks

To add a new sink transport, register its config keys with the config loader (add_param per sinks.{kind}.{type}.{key}) and register a factory with the sinks subsystem — both before the config loader runs. Built-in type names are reserved; re-registration fails loudly. See src/sdk/XAPP_SINKS.md for the full contract.

Where the reference xApps use sinks

Most reference xApps stream through sinks — kpm-monitor emits kpm samples, ho-cell-block publishes ho_decision / ue_snapshot records, and the sensing xApps (ran-sensing, srs-estimates) push their DSP output to Redis for a live dashboard. See xApps.