Collector: REST and webhooks

REST and webhooks

Two source types read from systems that speak HTTP and JSON: rest asks an API for new values on a schedule (a historian's REST interface, a vendor cloud, an MES), and webhook is a small receiver for systems that prefer to push. Both turn a JSON body into values with the same mapping options, so a payload you can describe once works either way. Neither needs more than read access, and the collector never writes to the system it reads.

Install the extras you use: pip install "kymoscope-collector[rest]" and "kymoscope-collector[webhook]".

REST poller

sources:
  # Wide: one record per time, one column per tag
  - name: Energy portal
    type: rest
    url: https://portal.example.com/api/v1/readings
    auth: bearer
    token: env:PORTAL_TOKEN
    params: {from: "{since}", to: "{until}", limit: "1000"}
    poll_ms: 60000
    items_path: data[*]
    time_path: timestamp
    values: {Power: kw, Voltage: electrical.volts}

  # Long: one record per value, with the tag name in the record
  - name: MES
    type: rest
    url: https://mes.customer.local/api/samples
    auth: oauth2
    token_url: https://login.customer.local/oauth2/token
    client_id: kymoscope
    client_secret: file:mes-secret.txt
    scope: read
    ca: customer-ca.pem
    method: POST
    body: {from: "{since}", to: "{until}"}
    items_path: samples[*]
    tag_path: tag
    value_path: value
    time_path: ts
    time_format: epoch_ms
    cursor_path: next_cursor
    cursor_param: cursor

Every poll_ms the collector sends one request, follows the pages of the answer, and publishes the values it has not published before. The request is a template: in the URL, in params and in body, {since} is the start of the window, {until} its end, {now} the current time, {tags} the tag names (a list in a JSON body, joined by ; in a URL), and {api_key} / {token} the configured secrets. Times are written as request_time_format (iso, epoch_s or epoch_ms).

Authentication

Set auth to one of the following; secrets accept env:NAME and file:path.

Pages

Use at most one of these when the answer comes in pieces (up to max_pages, default 1000, per poll):

Mapping the answer

Inside a record a path starting with ^ reads from the enclosing element, which is how a nested answer such as {data: [{tagName, samples: [{t, v}]}]} is read: items_path: data[*].samples[*], tag_path: ^.tagName, time_path: t, value_path: v. Text values and records without a readable number are skipped.

Rate limits and errors

A 429 answer (or 503 with a Retry-After) is waited out for the time the server asks, up to five minutes, then retried. Any other failure is logged, counted in poll_errors and retried with a growing pause up to a minute. Nothing is lost meanwhile: the watermark does not move, so the next successful request covers the gap. The watermark and the list of recent values are kept in the collector's state folder, and are saved only once the values up to them have been sent (or are in the stream's disk buffer), so a restart continues where it stopped without duplicates and without a gap. The previous good copy is kept as a .bak file: if the state file is unreadable the collector logs an error and continues from the backup.

discover lists the configured tags (or the keys of values). backfill sends the same request for the window you give it ({since} and {until} are its start and end) and needs a time_path. The collector's proxy setting applies to these requests.

Presets for Canary and GE Proficy Historian

preset fills in the request and the mapping for a historian's REST API; you give the server url, credentials and the tags to read. Any option you set yourself overrides the preset.

sources:
  - name: Canary
    type: rest
    preset: canary
    url: https://canary.customer.local:55236
    api_key: env:CANARY_TOKEN
    tags: ["Customer.Reactor1.Temp", "Customer.Reactor1.Pressure"]
    ca: customer-ca.pem

  - name: Proficy
    type: rest
    preset: proficy
    url: https://historian.customer.local:8443
    auth: oauth2
    token_url: https://historian.customer.local:8443/uaa/oauth/token
    client_id: kymoscope
    client_secret: env:PROFICY_SECRET
    tags: ["Reactor1.Temp"]

canary posts to the Web API's /api/v2/getTagData2 with the API token in the body and follows its continuation token. proficy reads /historian-rest-api/v1/datapoints/raw/<tags>/<start>/<end>/0/0, with the OAuth 2 token of the Proficy authorization service.

Webhook receiver

sources:
  - name: Vendor push
    type: webhook
    bind: 127.0.0.1:8099        # inbound port, see the warning below
    path: /ingest
    hmac_secret: env:WEBHOOK_HMAC
    hmac_header: X-Signature
    items_path: events[*]
    tag_path: sensor
    value_path: reading
    time_path: at
    time_format: epoch_s

One of secret or hmac_secret is required; a receiver without authentication is not accepted by the configuration check. The mapping options above (items_path, tag_path, value_path, time_path, values, quality_path, tags) read the JSON body. A request is answered right after it is checked and parsed: 202 accepted, 400 not JSON, 401 failed authentication, 413 too large, 503 with Retry-After if the queue (up to 100,000 values, counted in values, not requests) is full: the sender should retry.202 means accepted, not durable: the values are published from a queue, so they are lost if the collector stops before it is drained. A sender that must not lose data keeps its own copy until it can verify the data arrived.

# The sender signs the raw body: hex( HMAC-SHA256(secret, body) ), optionally prefixed with "sha256="
SIG=$(printf '%s' "$BODY" | openssl dgst -sha256 -hmac "$SECRET" -hex | sed 's/^.* //')
curl -X POST http://127.0.0.1:8099/ingest -H "X-Signature: sha256=$SIG" -d "$BODY"

# With hmac_timestamp_header: X-Kymoscope-Timestamp, sign "<timestamp>.<body>" (recommended)
TS=$(date +%s)
SIG=$(printf '%s.%s' "$TS" "$BODY" | openssl dgst -sha256 -hmac "$SECRET" -hex | sed 's/^.* //')
curl -X POST http://127.0.0.1:8099/ingest -H "X-Kymoscope-Timestamp: $TS" -H "X-Signature: sha256=$SIG" -d "$BODY"

Health

rest:     requests, http_2xx, http_3xx, http_4xx, http_429, http_5xx, token_requests, rate_limited, next_rejected, tags_dropped,
          polls, poll_errors, overruns, last_poll_s, values_per_s, last_watermark_age_s
webhook:  requests, rejected (failed authentication), bad_requests (too large or not JSON),
          queue_full, queue (values waiting), tags_dropped, values_per_s

last_watermark_age_s is how old the newest published value is; a number that keeps growing while poll_errors stays at 0 means the API answers but has no new data.

Ask the site for

• A read-only API account, token or OAuth client for the system (the collector never writes).
• REST: outbound HTTPS from the collector to the API host and port, and the CA certificate if the server uses a private CA.
• Webhook: the sender's address and an approved firewall rule for the receiver's port; see the warning above.