Configuration Reference

Siphon uses a strict, declarative YAML configuration.

Example config: mekops-labs/siphon/blob/main/configs/example.yaml

Env variables substitution

Whole configuration file is first parsed and all %%ENV_NAME%% references are substituted by values of environmental variables seen by siphon.

The Global Block

The root of the document requires the V2 schema declaration.

1version: 2

Collectors

Collectors define how Siphon ingests data. Each collector is given a unique alias and maps raw input parameters to internal Event Bus topics.

mqtt

Used to subscribe to one or more topics on a remote or local MQTT broker.

Parameters (params):

• url (required): The connection string (e.g., tcp://127.0.0.1:1883 or ssl://broker.example.com:8883).
• user: Username for authentication.
• pass: Password for authentication.

Topics (topics): A map where the key is the internal Siphon bus topic (alias) and the value is the actual MQTT topic string.

file

Reads the content of local system files, typically used for monitoring hardware sensors or system metrics.

Parameters (params):

• interval: Frequency of polling in seconds (default: 10).

Topics (topics): A map where the key is the internal bus topic and the value is the absolute path to the file.

shell

Executes an OS command periodically and captures its stdout.

Parameters (params):

• interval: Frequency of execution in seconds (default: 10).

Topics (topics): A map where the key is the internal bus topic and the value is the shell command to run.

hass

A specialized collector for Home Assistant. It polls entity states via the REST API. When running as an official Add-on, it automatically handles authentication using the $SUPERVISOR_TOKEN.

Parameters (params):

• interval: Polling frequency in seconds (default: 30)

Topics (topics): A map where the key is the internal bus topic and the value is the Home Assistant entity ID (e.g., sensor.living_room_temp). Supports wildcard * value - in this case returns all the entities in JSON format.

rest

Polls a remote HTTP endpoint via GET request periodically and captures the response body.

Parameters (params):

• interval: Frequency of polling in seconds (default: 10).

Topics (topics): A map where the key is the internal bus topic and the value is the full URL to poll.

1collectors:
2  weather:
3    type: rest
4    params:
5      interval: 60
6    topics:
7      current_weather: "https://wttr.in/?format=%C"

webhook

Starts a high-performance, hardened HTTP listener that allows external systems (GitHub, Grafana, custom scripts) to push data directly into Siphon. The server is designed for public internet exposure with multi-layer defense against DDoS attacks, memory exhaustion, and unauthorized access.

Parameters (params):

• port (required): The port to listen on.
• token: If set, rejects any request missing a matching Authorization: Bearer <token> header.
• rps: Sustained requests per second allowed by the rate limiter (default: unlimited).
• burst: Maximum concurrent burst of requests allowed (default: unlimited).
• max_body_mb: Hard limit on incoming payload size in megabytes (default: unlimited).
• dedupe_ttl: Duration in seconds during which identical payloads (by SHA-256 hash) are dropped to prevent duplicate pipeline processing. Returns 409 Conflict to the sender so the caller can distinguish a processed request from a deduplicated one. Only authenticated callers can observe this response.

Topics (topics): A map where the key is the internal bus topic and the value is the HTTP path to expose.

 1collectors:
 2  secure_api:
 3    type: webhook
 4    params:
 5      port: 8080
 6      token: "%%MY_WEBHOOK_SECRET%%"
 7      rps: 5.0
 8      burst: 10
 9      max_body_mb: 2
10      dedupe_ttl: 300
11    topics:
12      github_hooks: "/webhooks/github"
13      sensor_push: "/data/sensor01"

Best Practice: Siphon does not handle TLS natively. Place Siphon behind a reverse proxy (Nginx, Caddy, Traefik) to provide HTTPS encryption. Store secrets via %%ENV_VAR%% substitution rather than hardcoding them.

Example collector definition

1collectors:
2  local_ha:
3    type: hass
4    params:
5      interval: 60
6    topics:
7      outdoor_temp: "sensor.backyard_temperature"

Sinks

Sinks define the egress points for your telemetry data. They are defined globally in the sinks block and can be referenced by name within multiple pipelines.

bus

Publishes the processed payloads back to the internal bus.

Parameters (params):

• topic (required): The internal bus topic to publish to.

gotify

Sends push notifications to a Gotify server.

Parameters (params):

• url (required): The base URL of your Gotify instance.
• token (required): The application token.
• title: The notification title. This fields uses template rendering. now function is defined as alias to time.Now().Format() (default "Siphon Alert")
• priority: The notification priority (default 0)

ntfy

Sends notifications to ntfy.sh or a self-hosted instance.

Parameters (params):

• url (required): The ntfy instance URL (e.g., https://ntfy.sh).
• topic (required): The ntfy topic
• token: The ntfy token (default: "")
• title: The notification title. This fields uses template rendering. now function is defined as alias to time.Now().Format() (default: "")
• priority: The notification priority (default: 3)

windy

Updates a Personal Weather Station (PWS) on Windy.com.

Parameters (params):

• id (required): Your Station ID.
• password (required): Your Windy PWS password from station's page.

iotplotter

Dispatches data to the IoTPlotter service.

Parameters (params):

• url: The API endpoint (default: http://iotplotter.com).
• apikey (required): Your account API key.
• feed (required): The specific feed ID to update.

stdout

Prints the final payload to the Siphon logs. Useful for debugging.

Parameters (params):

• None.

mqtt

Publishes the final pipeline payload to a topic on an MQTT broker. Useful for sending processed data, aggregations, or alerts back to Home Assistant or other IoT dashboards.

Parameters (params):

• url (required): The broker connection string (e.g., tcp://192.168.1.100:1883).
• user: Username for authentication. Supports %%ENV_VAR%% substitution.
• pass: Password for authentication. Supports %%ENV_VAR%% substitution.
• topic (required): The MQTT topic to publish to.
• qos: Quality of Service level — 0, 1, or 2 (default: 0).
• retained: Whether the broker should retain the message (default: false).

hass

Integrates with Home Assistant's MQTT Auto-Discovery feature. When Siphon starts, this sink automatically registers a device named "Siphon ETL Engine" and creates the configured entity under it in Home Assistant.

The sink is component-aware: the component field determines which MQTT topics are included in the discovery payload and where Send() routes its output. All topics are auto-generated from the base path {discovery_prefix}/{component}/{object_id} and can be individually overridden.

Availability is managed automatically via MQTT Last Will and Testament (LWT): Home Assistant marks the entity "Unavailable" if Siphon disconnects unexpectedly, and "Online" upon reconnection.

When running as a Home Assistant Add-on, MQTT connection details are injected automatically.

Parameters (params):

• url (required): The MQTT broker connection string (e.g., tcp://192.168.1.100:1883). Auto-injected as an Add-on.
• user: MQTT username. Supports %%ENV_VAR%% substitution.
• pass: MQTT password. Supports %%ENV_VAR%% substitution.
• object_id (required): The entity ID in Home Assistant (e.g., aggregated_power → sensor.aggregated_power).
• name: The friendly display name of the entity (default: derived from object_id).
• component: The Home Assistant component type (default: sensor). Controls topic selection — see table above.
• device_class: The HA device class (e.g., power, temperature).
• state_class: The HA state class (e.g., measurement, total_increasing).
• unit_of_measurement: The unit displayed in HA (e.g., W, °C).
• icon: The MDI icon (e.g., mdi:flash).
• state_topic: Override the auto-generated state topic.
• trigger_type: HA trigger type (for device_automation).
• subtype: HA trigger subtype (for device_automation).
• payload: Optional exact-match payload for the trigger.
• availability_topic: Override the auto-generated availability topic.

Sensor example:

 1sinks:
 2  custom_ha_sensor:
 3    type: hass
 4    params:
 5      url: "%%MQTT_HOST%%"
 6      user: "%%MQTT_USER%%"
 7      pass: "%%MQTT_PASS%%"
 8      object_id: "aggregated_power"
 9      name: "Total House Power Draw"
10      component: "sensor"
11      device_class: "power"
12      state_class: "measurement"
13      unit_of_measurement: "W"
14      icon: "mdi:flash"
15      # state_topic auto-generated: homeassistant/sensor/aggregated_power/state

Device Trigger example (e.g., Garmin watch trigger):

 1sinks:
 2  garmin_trigger:
 3    type: hass
 4    params:
 5      url: "%%MQTT_HOST%%"
 6      object_id: "garmin_button"
 7      component: "device_automation"
 8      trigger_type: "button_short_press"
 9      subtype: "button_1"
10      icon: "mdi:watch"
11      # topic auto-generated: homeassistant/device_automation/garmin_button/action

Example sink definition

 1sinks:
 2  local_debug:
 3    type: stdout
 4  mobile_alerts:
 5    type: gotify
 6    params:
 7      url: "https://gotify.example.com"
 8      token: "%%GOTIFY_TOKEN%%"
 9  broker_out:
10    type: mqtt
11    params:
12      url: "tcp://192.168.1.100:1883"
13      user: "%%MQTT_USER%%"
14      pass: "%%MQTT_PASS%%"
15      topic: "siphon/processed_data"
16      qos: 1
17      retained: false

Pipelines & Parsers

Pipelines are the primary processing units in Siphon. They act as consumers of the internal Event Bus, defining how raw data is transformed and where it is eventually dispatched.

Pipeline Fields

• name (required): A unique identifier for the pipeline.
• type: Defines the trigger mechanism.
  • event (default): Executes as soon as new data arrives on a subscribed topic.
  • cron: Executes based on a temporal schedule.
• topics (required): A list of internal bus topic aliases this pipeline subscribes to.
• schedule: Required if type is cron. Uses standard cron syntax (e.g., */5 * * * * *).
• stateful: (boolean) When true, Siphon caches the last known variables from this pipeline's parser. This is essential for merging data from multiple sources in a cron pipeline.
• bus_mode:
  • volatile (default): High-performance in-memory processing.
  • durable: (WIP) Persists events to a Write-Ahead Log to prevent data loss during restarts.

The parser Block

The parser converts raw bytes from the collector into structured variables.

• type: The engine used for extraction.
  • jsonpath: For structured JSON payloads.
  • regex: For extracting values from unstructured strings or logs.
• vars: A map where the key is the variable name and the value is the path or pattern.

The transform Block (Optional)

An optional ordered array of expr expressions to further process variables before they hit the sinks. The array form guarantees that transformations are applied in the order they are defined, which is important when one transformation depends on the result of a previous one.

1transform:
2  - temp_f: "(temp_c * 9/5) + 32"
3  - temp_f_str: "string(temp_f) + '°F'"

Note: The transform block was changed from a map to an array in order to guarantee evaluation order. Please update any existing configurations accordingly.

Topic-Namespaced Variables

In stateful pipelines that subscribe to multiple upstream topics, variables from each source pipeline are accessible via their pipeline name as a namespace, using the ?. safe-navigation operator.

For example, if a stateful cron pipeline subscribes to process_temp and process_uptime:

1spec: |
2  {
3    "temp": process_temp?.temp ?? 0,
4    "uptime": process_uptime?.up ?? 0
5  }

This prevents panics when a source pipeline hasn't yet reported data.

The sinks Array

Defines the egress routing for the pipeline.

• name: The alias of a globally defined sink.
• format: The rendering engine used for the payload.
  • expr: Uses the expression engine to build complex JSON or logic.
  • template: Uses standard Go text/template syntax.
• spec: The actual expression or template string defining the output.

Example pipeline definition

 1pipelines:
 2  - name: process_garage
 3    type: event # Triggers instantly when new data arrives
 4    topics: ["garage_telemetry"]
 5    parser:
 6      type: jsonpath
 7      vars:
 8        door_state: "$.door.open"
 9        battery_v: "$.voltage"
10    transform:
11      - door_state: "string(door_state)"
12      - battery_v: "battery_v / 1000"
13    sinks:
14      - name: alert_phone
15        format: expr
16        spec: |
17          {
18            "title": "Garage Update",
19            "door": door_state,
20            "battery": battery_v
21          }

Advanced: Stateful Cron Merging

Because the Event Bus caches the parsed state, a cron pipeline can wake up on a schedule, fetch the latest states from completely different collectors, and merge them into a single payload using the expr engine. Variables from each source pipeline are accessed using the pipeline name as a namespace prefix.

 1pipelines:
 2  - name: outdoor_temp
 3    topics: ["outdoor"]
 4    parser:
 5      type: jsonpath
 6      vars:
 7        state: "$.state"
 8
 9  - name: garage_telemetry
10    topics: ["garage"]
11    parser:
12      type: jsonpath
13      vars:
14        battery_v: "$.voltage"
15
16  - name: unified_dispatcher
17    type: cron
18    stateful: true
19    schedule: "*/30 * * * * *"  # Run every 30 seconds
20    topics: ["outdoor_temp", "garage_telemetry"]
21    sinks:
22      - name: my_backend
23        format: expr
24        spec: |
25          {
26            "merged_telemetry": {
27              "temp": outdoor_temp?.state ?? 0,
28              "garage_v": garage_telemetry?.battery_v ?? 0
29            }
30          }

Note the use of the ?? null-coalescing operator and ?. safe-navigation in the expr block to prevent panics if a collector hasn't reported data yet.