igwn-connect

A service built on SGN, connecting data between LLDD, Arrakis, and frame files.

Stream gravitational-wave detector timeseries between the low-latency data distribution system (LLDD), the Arrakis timeseries service, and GWF frame files. Pick a --source and a --sink and the connector builds and runs the SGN pipeline between them.

Resources

• Source Code
• Issue Tracker

Installation

pip install igwn-connect

Or from source:

git clone git@git.ligo.org:ngdd/igwn-connect.git
cd igwn-connect
pip install .

Features

• Sources: LLDD (Kafka frame streams), Arrakis, GWF frame file directories, replayed frame sets, and synthetic test signals
• Sinks: Arrakis, LLDD, GWF frame files, and a null sink for testing
• Any source can be paired with any sink
• Live streams are resilient to interruptions: sources reconnect with capped exponential backoff and bridge outages with gap buffers, and sinks drop a file or block that fails to write rather than stopping (--max-write-failures, --max-publish-failures)
• Bounded (--start/--end) or continuous live operation
• Arrakis replay namespace support for both streaming and publishing (--replay-id)
• Replay a fixed set of frame files as live data (--source replay): retimestamped to the present, paced to the wall clock, and looped
• Frame writing options: file retention (--max-files, --retention-time), and skipping files that contain no real data (--skip-all-gap)
• Live monitoring of the running pipeline (--monitor): an HTML dashboard, Prometheus metrics, and a health endpoint

Quickstart

The CLI shows contextual options: pass --source and/or --sink with --help to see the options relevant to that combination, e.g.

igwn-connect --source frames --sink arrakis --help

Publish live DAQ data from LLDD to Arrakis

igwn-connect --source lldd --sink arrakis \
    --ifo L1 --topic Live_LLO_Data \
    --bootstrap-servers kafka1:9092,kafka2:9092 \
    --arrakis-url grpc://arrakis-online1:31206 \
    --publisher-id L1-lldd

The channels to publish are discovered from the publisher's registration on the Arrakis server.

Write an Arrakis stream to frame files

The Arrakis source reads the server URL from the ARRAKIS_SERVER environment variable:

export ARRAKIS_SERVER=grpc://arrakis-online1:31206
igwn-connect --source arrakis --sink frames \
    --channels L1:GDS-CALIB_STRAIN L1:GDS-CALIB_STATE_VECTOR \
    --frame-duration 64 --output-dir /data/frames \
    --skip-all-gap

Without --start and --end this streams live data continuously; press Ctrl+C to stop.

Publish frame files to LLDD

Watch a directory for new frame files and publish them to Kafka:

igwn-connect --source frames --sink lldd \
    --channels H1:GDS-CALIB_STRAIN \
    --watch-dir /data/live/H1 \
    --ifo H1 --topic Live_LHO_Data \
    --bootstrap-servers kafka1:9092

Replay frame files into Arrakis

Replay a frame set as if it were live, publishing into a replay namespace:

igwn-connect --source replay --sink arrakis \
    --frames /data/O3ReplayMDC/L1 \
    --arrakis-url grpc://arrakis-replay1:31206 \
    --publisher-id L1-lldd-O3Replay --replay-id O3Replay

--frames takes directories of .gwf files, LAL .cache files, or individual .gwf files (T050017 names), in any mix. The replay window [replay_start, replay_end) is mapped onto the present by adding a whole number of window durations, the same mapping the Arrakis frames backend uses to serve replays, so every instance replaying the same window stays coherent. With --sink arrakis the window defaults to the server's registered window for --replay-id and the channels come from the publisher's registration; other sinks need --replay-start, --replay-end, and --channels. Data is delivered one second at a time once the wall clock passes it, spans no file covers are delivered as gaps, and the window loops until --end or Ctrl+C. --target-latency N holds each second of data back a further N seconds, emulating the transport latency of a live stream.

Test a pipeline with synthetic data

igwn-connect --source fake --sink null \
    --channels H1:TEST --rate 2048 --signal-type sin \
    --start 0 --end 10

Monitoring

Serve live monitoring for the running pipeline with sgnmon:

igwn-connect --source arrakis --sink frames \
    --channels L1:GDS-CALIB_STRAIN \
    --frame-duration 64 --output-dir /data/frames \
    --monitor

This starts a background web server (default port 9090; --monitor-port changes it and implies --monitor, with 0 picking a free port) exposing:

• / — a live dashboard drawing the pipeline graph with per-channel rates, latencies, and gap fractions
• /metrics — Prometheus metrics for scraping
• /health — a JSON health report (HTTP 503 when unhealthy), usable directly by container orchestration or sgnmon check
• /readyz — readiness probe: 200 while the pipeline's run loop is running (between ready and stopping) and every health check passes, 503 while starting, stopping, or a check fails
• /healthz — liveness probe: 200 while the run loop keeps striding, 503 once it has gone stale
• /status — a JSON snapshot of all probes

Rather than observing every pad, the connector taps each link once at the receiving element, plus post-alignment consumption (adp) to distinguish "data arrived" from "data processed", and the source's output pads so input rate and latency are visible per channel at the origin. The data-freshness health check tolerates three missed output strides (--frame-duration for the frames sink, --delta-t for the LLDD sink) before failing, so slow output cadences do not false-alarm.

--monitor-max-latency N (implies --monitor) adds a publishing-latency check: /health and /readyz fail when the data the sink is publishing or writing trails real time by more than N seconds, per channel, and /health/max_latency reports that check alone. It applies after the same grace period as the freshness check, so a source still catching up at startup does not trip it, and it is only meaningful for live data.

Health reporting under systemd

The pipeline reports its lifecycle through sgn.health: a Type=notify unit (or podman's default --sdnotify=container) sets NOTIFY_SOCKET, and the connector then sends READY=1 once the graph is running, WATCHDOG=1 on every stride, and STOPPING=1 when the run loop exits. This needs no flag; --monitor adds the monitoring server's probes alongside it, driven by the same lifecycle.

[Service]
Type=notify
ExecStart=/usr/bin/igwn-connect --source lldd --sink arrakis ...
WatchdogSec=60s
TimeoutStopSec=120s
Restart=on-failure

• WATCHDOG=1 is only sent as the graph strides, so WatchdogSec= must exceed the longest stall a source can have: the LLDD source's --poll-timeout is 1 s and the frames source's --queue-timeout is 10 s by default, so 60s leaves ample margin.
• On SIGTERM the connector drains the pipeline to end-of-stream, then sends STOPPING=1, which disarms the watchdog and starts TimeoutStopSec=; set it generously enough for the sink to flush.

Usage with Docker

The container image is hosted on containers.ligo.org. Pull the latest version with:

podman pull docker://containers.ligo.org/ngdd/igwn-connect:latest

Run with podman (or docker):

podman run --rm --net=host \
    docker://containers.ligo.org/ngdd/igwn-connect:latest \
    --source lldd --sink arrakis \
    --ifo L1 \
    --arrakis-url grpc://arrakis-online1:31206 \
    --publisher-id L1-lldd \
    --bootstrap-servers kafka1:9092,kafka2:9092 \
    --topic Live_LLO_Data