Collector: customer-side sinks

What it does

• Every value of the chosen sources goes to your database too, alongside the live stream and the recordings, whatever each source's record says. Nothing is resampled.
• The values go from the collector straight to your database, in plaintext, on your network. They never pass through Kymoscope, and Kymoscope never reads that database.
• A slow or unreachable database never slows the live stream or the recordings: each sink has its own local log and catches up later.
• The health stream is never written to a sink. Values with a Bad status (gaps) are skipped, as in the live stream.

Sinks are part of Enterprise; ask Kymoscope to turn them on for your organization. They need kymoscope-collector 0.5.0 or later.

Configure

Add a sinks: list to collector.yaml and restart the collector. Secrets take env:NAME or file:PATH, as api_key does.

api_key: env:KYMOSCOPE_API_KEY
state_dir: state
health:
  stream: Customer A collector health
sources:
  - name: Customer A DCS
    type: opcua
    url: opc.tcp://dcs-gw:4840
    tags: tags.csv
  - name: Packaging line
    type: modbus
    transport: tcp
    host: 10.0.4.20
    map: packaging.csv
sinks:
  - name: Customer SQL
    type: sql
    dsn: env:CUSTOMER_SQL_DSN           # postgresql://kymo_writer:password@db.customer.local:5432/customer
    table: kymoscope_values          # created when missing (create_table: true)
    # sources: [Customer A DCS]         # omit to write every source
    # max_local_mb: 2048             # cap of this sink's local log
    # batch_rows: 5000
  - name: Customer Influx
    type: influx
    url: https://influx.customer.local:8086
    api: v2
    token: env:INFLUX_TOKEN
    org: customer
    bucket: kymoscope
    sources: [Packaging line]
    # measurement: kymoscope
    # ca: certs/customer-ca.pem         # a private CA for the InfluxDB certificate

A mistake stops the start with one sentence naming the sink and the key, like any other key in collector.yaml.

SQL databases

Install the SQL drivers with pip install "kymoscope-collector[sql]". SQL Server also needs the Microsoft ODBC Driver 18, and MySQL or MariaDB their own ODBC driver, on the collector host.

# PostgreSQL / TimescaleDB
dsn: postgresql://kymo_writer:password@db.customer.local:5432/customer

# SQL Server (Microsoft ODBC Driver 18)
dsn: "mssql://kymo_writer:password@sql.customer.local:1433/customer?driver=ODBC+Driver+18+for+SQL+Server"

# MySQL / MariaDB (their ODBC driver)
dsn: "DRIVER={MySQL ODBC 8.0 Unicode Driver};SERVER=mysql.customer.local;DATABASE=customer;UID=kymo_writer;PWD=password"

# Oracle (thin mode, no Oracle client needed)
dsn: oracle://kymo_writer:password@ora.customer.local:1521/CUSTOMERPDB

The table

One row per value, in long format, so it works the same on every database: ts (the value's time, UTC), source (the source name), channel (the channel name) and value. The primary key is (source, channel, ts). This is what the collector creates:

-- PostgreSQL / TimescaleDB
create table if not exists kymoscope_values (ts timestamptz not null, source text not null,
  channel text not null, value double precision not null, primary key (source, channel, ts));

-- SQL Server
CREATE TABLE kymoscope_values (ts datetime2(6) NOT NULL, source nvarchar(100) NOT NULL,
  channel nvarchar(300) NOT NULL, value float NOT NULL, PRIMARY KEY (source, channel, ts));

-- MySQL / MariaDB
CREATE TABLE IF NOT EXISTS kymoscope_values (ts DATETIME(6) NOT NULL, source VARCHAR(100) NOT NULL,
  channel VARCHAR(300) NOT NULL, value DOUBLE NOT NULL, PRIMARY KEY (source, channel, ts));

-- Oracle
CREATE TABLE kymoscope_values (ts TIMESTAMP(6) NOT NULL, source VARCHAR2(100 CHAR) NOT NULL,
  channel VARCHAR2(300 CHAR) NOT NULL, value BINARY_DOUBLE NOT NULL, PRIMARY KEY (source, channel, ts));

• Times are UTC: timestamptz in PostgreSQL, UTC values without a zone in the others. Convert to the site's zone in your queries.
• Outside PostgreSQL, source names are cut at 100 characters and channel names at 300.
• With create_table: false, give your table the same four columns and a unique key on (source, channel, ts). Without it, PostgreSQL and MySQL can store a replayed value twice.
• The account needs INSERT and SELECT on the table, plus CREATE on the schema while create_table is true.

For TimescaleDB, turn the table into a hypertable once:

SELECT create_hypertable('kymoscope_values', 'ts', migrate_data => true);

-- the last hour of one channel
SELECT ts, value FROM kymoscope_values
WHERE source = 'Customer A DCS' AND channel = 'FIC101.PV' AND ts >= now() - interval '1 hour'
ORDER BY ts;

InfluxDB

  - name: Customer Influx 1.x
    type: influx
    url: https://influx.customer.local:8086
    api: v1
    database: kymoscope
    # rp: autogen                    # retention policy; the database's default when left out
    user: kymo_writer
    password: env:INFLUX_PASSWORD

Each value is one point: the measurement, the tags source and channel, the field value, and the time in microseconds (UTC):

kymoscope,source=Customer\ A\ DCS,channel=FIC101.PV value=12.5 1761400800123456

The collector checks /ping before it writes, so the token needs write access to the bucket only.

Reliability

• Local log. Each sink keeps its own log under state_dir/sinks/<sink>, encrypted with the collector's local key, as the buffers are. A value is in the log before the collector moves on, and leaves it only after the database accepted it, so a database that hangs never makes the collector drop values.
• Outages. While the database is down, values wait in the log (sink_backlog_mb rises) and the collector retries, every 5 seconds at first and every minute at most. After the outage the backlog is written oldest first, at the database's speed. A restart loses nothing: everything is already in each sink's log when the collector stops, and it is written after the next start.
• Cap. The log stops growing at max_local_mb, or when the disk falls below buffer.min_free_mb. New values for that sink are then dropped and counted in sink_dropped; the live stream and the history are not affected. Backfilled values are past data that can't be sent again: they go past max_local_mb and stop only at the disk floor.
• No duplicates. Delivery is at least once, and every write skips a value already stored (same source, channel and time). InfluxDB overwrites the same point. Replays after an outage or a restart never duplicate.
• Rejected values. If the database keeps refusing a batch while it answers SELECT 1, the batch is written value by value. Values it refuses are dropped and counted in sink_rows_rejected, so one bad value never blocks the sink. If it refuses every value, the cause is the database (a full disk, a revoked grant): the batch stays in the log.
• Turning it on. The collector asks Kymoscope whether sinks are on for your organization at start and every hour. A yes is remembered for 7 days, so a Kymoscope outage never stops your copy. Until the first answer arrives, values wait in the log.
• Backfills. Values a running collector backfills (backfill_gaps) go to the sinks too; the separate backfill command does not write to them.

Health channels

With health.stream set, each sink adds these channels to the health stream, named after the sink:

Automation

Chart and alert on the sink_* channels in the app like any other stream. To check from the database side, alert when the newest value is too old:

# check-sink.ps1: run every 15 minutes from Task Scheduler (SQL Server; sqlcmd from the ODBC tools)
$age = sqlcmd -S sql.customer.local -d customer -E -h -1 -W -Q `
  "SET NOCOUNT ON; SELECT DATEDIFF(second, MAX(ts), SYSUTCDATETIME()) FROM kymoscope_values"
if ([int]$age -gt 600) {
  Write-EventLog -LogName Application -Source "Kymoscope" -EventId 1101 -EntryType Warning `
    -Message "Kymoscope sink: the newest value in kymoscope_values is $age s old"
  exit 1
}

#!/bin/sh
# /etc/cron.d/kymoscope-sink: */15 * * * * kymoscope /opt/kymoscope/check-sink.sh  (PostgreSQL)
age=$(psql "$CUSTOMER_SQL_DSN" -Atc "select extract(epoch from now() - max(ts))::int from kymoscope_values")
if [ "${age:-999999}" -gt 600 ]; then
  logger -p user.warning "Kymoscope sink: the newest value in kymoscope_values is ${age}s old"
  exit 1
fi

Troubleshooting