From 8b6d326824acfd864033fb4308e2819131d32346 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 12 Nov 2026 23:42:26 -0500 Subject: [PATCH] chore(docker-env2): seed-data-path.sh - opc-plc + read-only Galaxy instances and a store-and-forward sink for the #116 soak Adds an env2-owned HTTP sink container (scadabridge-env2-sfsink) whose 503 mode makes ExternalSystem.CachedCall buffer, and an EnvSoak template whose 2 s SoakTick sends one tick-numbered cached call, so loss and duplicates are countable at the sink. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi --- docker-env2/seed-data-path.sh | 117 ++++++++++++++++++++++++++++++++++ docker-env2/sfsink/sink.py | 87 +++++++++++++++++++++++++ 2 files changed, 204 insertions(+) create mode 100755 docker-env2/seed-data-path.sh create mode 100644 docker-env2/sfsink/sink.py diff --git a/docker-env2/seed-data-path.sh b/docker-env2/seed-data-path.sh new file mode 100755 index 00000000..d5f2058a --- /dev/null +++ b/docker-env2/seed-data-path.sh @@ -0,0 +1,117 @@ +#!/bin/bash +set -euo pipefail + +# Seed env2's site-x with live data and a store-and-forward path, so the +# leader-auto-down soak (ScadaBridge#116) exercises real data, not just cluster +# membership. Run after docker-env2/seed-sites.sh (needs Site-X and the +# "ScadaBridge Site X" MxGateway connection). +# +# What it creates: +# - Container scadabridge-env2-sfsink (python:3.13-alpine on scadabridge-net, +# docker-env2/sfsink/sink.py): an HTTP sink that records every POST it accepts +# and answers 503 while /data/mode says "fail" — the switch that makes cached +# calls buffer in store-and-forward. env2-owned; no shared infra is touched. +# - Data connection "OPC PLC Simulator" on site-x -> opc.tcp://scadabridge-opcua:50000 +# (the shared opc-plc, used as a client only). +# - External system EnvSink -> http://scadabridge-env2-sfsink:5300, method Record = +# POST /record, retry delay 5 s, retry budget 100000 (so a long fail window parks nothing). +# - Template EnvSoak: Spike (Double, ns=3;s=SpikeData) and RandomInt (Int32, +# ns=3;s=RandomSignedInt32) — opc-plc's self-changing nodes — plus a 2 s Interval +# script SoakTick that sends ExternalSystem.CachedCall("EnvSink","Record") with +# tick = unix-seconds / 2. One row per tick at the sink, so a gap is a missed or +# lost tick and a repeated tick is a duplicate. +# - Instance env-soak-1 bound to the OPC UA connection, deployed. +# - Template EnvGalaxyProbe: GalaxyInt (Int32) <- TestMachine_003.TestChangingInt, +# READ ONLY (no script writes, no write binding), and instance env-galaxy-1 bound to +# the MxGateway connection, deployed. +# +# Not idempotent (names are unique). To re-seed, delete the instances and templates. +# +# Watch it: +# docker exec scadabridge-env2-sfsink wget -qO- localhost:5300/stats +# scadabridge --url http://localhost:9100 debug snapshot --id + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" +CLI="${CLI:-$PROJECT_ROOT/src/ZB.MOM.WW.ScadaBridge.CLI/bin/Debug/net10.0/scadabridge}" +A=(--url "${SCADABRIDGE_URL:-http://localhost:9100}" --format=json --username multi-role --password password) +SINK=scadabridge-env2-sfsink +GALAXY_TAG="${GALAXY_TAG:-TestMachine_003.TestChangingInt}" + +json_id() { python3 -c 'import json,sys; print(json.load(sys.stdin)["id"])'; } +find_id() { python3 -c 'import json,sys; k,v=sys.argv[1],sys.argv[2]; print(next(x["id"] for x in json.load(sys.stdin) if x.get(k)==v))' "$1" "$2"; } + +[ -x "$CLI" ] || dotnet build "$PROJECT_ROOT/src/ZB.MOM.WW.ScadaBridge.CLI" -v q > /dev/null + +SITE_ID=$("$CLI" site list "${A[@]}" | find_id siteIdentifier site-x) +MXGW_ID=$("$CLI" data-connection list "${A[@]}" | find_id name "ScadaBridge Site X") +echo "site-x id=$SITE_ID, MxGateway connection id=$MXGW_ID" + +echo "1/7 Store-and-forward sink container $SINK..." +if ! docker ps -a --format '{{.Names}}' | grep -qx "$SINK"; then + docker create --name "$SINK" --network scadabridge-net --restart unless-stopped \ + python:3.13-alpine python -u /sink/sink.py > /dev/null + docker cp "$SCRIPT_DIR/sfsink" "$SINK:/sink" +fi +docker start "$SINK" > /dev/null + +echo "2/7 OPC UA data connection on site-x..." +OPC_CONFIG='{"endpointUrl":"opc.tcp://scadabridge-opcua:50000","securityMode":"none","autoAcceptUntrustedCerts":true}' +OPC_ID=$("$CLI" data-connection create "${A[@]}" --site-id "$SITE_ID" --name "OPC PLC Simulator" \ + --protocol OpcUa --primary-config "$OPC_CONFIG" | json_id) +echo " id=$OPC_ID" + +echo "3/7 External system EnvSink..." +ES_ID=$("$CLI" external-system create "${A[@]}" --name EnvSink --endpoint-url "http://$SINK:5300" \ + --auth-type ApiKey --auth-config '{"key":"env2-sink-not-a-secret"}' --timeout-seconds 5 \ + --max-retries 100000 --retry-delay-seconds 5 | json_id) +"$CLI" external-system method create "${A[@]}" --external-system-id "$ES_ID" --name Record \ + --http-method POST --path /record > /dev/null +echo " id=$ES_ID" + +echo "4/7 Template EnvSoak (opc-plc tags + SoakTick cached call)..." +T_ID=$("$CLI" template create "${A[@]}" --name EnvSoak \ + --description "env2 soak: opc-plc SpikeData/RandomSignedInt32 + a 2 s store-and-forward cached call (#116)" | json_id) +"$CLI" template attribute add "${A[@]}" --template-id "$T_ID" --name Spike --data-type Double \ + --data-source "ns=3;s=SpikeData" --description "opc-plc SpikeData, changes ~1/s" > /dev/null +"$CLI" template attribute add "${A[@]}" --template-id "$T_ID" --name RandomInt --data-type Int32 \ + --data-source "ns=3;s=RandomSignedInt32" --description "opc-plc RandomSignedInt32, changes ~1/s" > /dev/null +read -r -d '' TICK_CODE <<'CS' || true +// One cached call per 2 s tick. tick is wall-clock derived, so it survives a failover: +// a gap at the sink = a tick nobody ran (or a lost message); a repeat = a duplicate. +var tick = System.DateTimeOffset.UtcNow.ToUnixTimeSeconds() / 2; +var spike = await Instance.GetAttribute("Spike"); +var rnd = await Instance.GetAttribute("RandomInt"); +await ExternalSystem.CachedCall("EnvSink", "Record", + new System.Collections.Generic.Dictionary { + ["tick"] = tick, + ["spike"] = spike?.ToString(), + ["randomInt"] = rnd?.ToString(), + ["sentAtUtc"] = System.DateTime.UtcNow.ToString("O") }); +CS +"$CLI" template script add "${A[@]}" --template-id "$T_ID" --name SoakTick \ + --trigger-type Interval --trigger-config '{"intervalMs":2000}' --code "$TICK_CODE" > /dev/null +echo " id=$T_ID" + +echo "5/7 Template EnvGalaxyProbe (read-only Galaxy attribute $GALAXY_TAG)..." +G_ID=$("$CLI" template create "${A[@]}" --name EnvGalaxyProbe \ + --description "env2 soak: one READ-ONLY windev Galaxy DEV attribute via MxGateway (#116)" | json_id) +"$CLI" template attribute add "${A[@]}" --template-id "$G_ID" --name GalaxyInt --data-type Int32 \ + --data-source "$GALAXY_TAG" --description "Galaxy DEV, read only" > /dev/null +echo " id=$G_ID" + +echo "6/7 Pushing artifacts (OPC UA connection + external system) to site-x..." +"$CLI" site deploy-artifacts "${A[@]}" --site-id "$SITE_ID" > /dev/null + +echo "7/7 Instances..." +I1=$("$CLI" instance create "${A[@]}" --name env-soak-1 --template-id "$T_ID" --site-id "$SITE_ID" | json_id) +"$CLI" instance set-bindings "${A[@]}" --id "$I1" --bindings "[[\"Spike\",$OPC_ID],[\"RandomInt\",$OPC_ID]]" > /dev/null +"$CLI" instance deploy "${A[@]}" --id "$I1" > /dev/null +I2=$("$CLI" instance create "${A[@]}" --name env-galaxy-1 --template-id "$G_ID" --site-id "$SITE_ID" | json_id) +"$CLI" instance set-bindings "${A[@]}" --id "$I2" --bindings "[[\"GalaxyInt\",$MXGW_ID]]" > /dev/null +"$CLI" instance deploy "${A[@]}" --id "$I2" > /dev/null +echo " env-soak-1 id=$I1, env-galaxy-1 id=$I2" + +echo +echo "Done. Check: docker exec $SINK wget -qO- localhost:5300/stats" +echo " $CLI ${A[*]:0:2} debug snapshot --id $I1" diff --git a/docker-env2/sfsink/sink.py b/docker-env2/sfsink/sink.py new file mode 100644 index 00000000..002ecd9a --- /dev/null +++ b/docker-env2/sfsink/sink.py @@ -0,0 +1,87 @@ +"""Tiny HTTP sink for the env2 store-and-forward soak (ScadaBridge#116). + +POST/PUT -> 200 and the body is appended to /data/received.jsonl + (one line: receive time, client IP, path, body), or + 503 without recording while /data/mode contains "fail". +GET /stats -> {"mode": ..., "received": } +GET /received -> the raw received.jsonl + +The mode switch is how the soak makes ScadaBridge buffer: a 503 is a transient +failure, so ExternalSystem.CachedCall parks the call in store-and-forward and +retries it; setting the mode back to "ok" lets the active node drain the buffer. + + docker exec scadabridge-env2-sfsink sh -c 'echo fail > /data/mode' + docker exec scadabridge-env2-sfsink sh -c 'echo ok > /data/mode' +""" +import json +import os +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +DATA = "/data" +LOG = os.path.join(DATA, "received.jsonl") +MODE = os.path.join(DATA, "mode") +LOCK = threading.Lock() + + +def mode(): + try: + with open(MODE) as f: + return f.read().strip() or "ok" + except OSError: + return "ok" + + +class Handler(BaseHTTPRequestHandler): + def _reply(self, code, obj): + body = json.dumps(obj).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _record(self): + length = int(self.headers.get("Content-Length") or 0) + raw = self.rfile.read(length).decode("utf-8", "replace") if length else "" + if mode() == "fail": + self._reply(503, {"error": "sink in fail mode"}) + return + try: + body = json.loads(raw) if raw else None + except ValueError: + body = raw + line = json.dumps({"at": time.time(), "client": self.client_address[0], + "path": self.path, "body": body}) + with LOCK, open(LOG, "a") as f: + f.write(line + "\n") + self._reply(200, {"ok": True}) + + do_POST = _record + do_PUT = _record + + def do_GET(self): + if self.path == "/stats": + n = 0 + if os.path.exists(LOG): + with open(LOG) as f: + n = sum(1 for _ in f) + self._reply(200, {"mode": mode(), "received": n}) + elif self.path == "/received": + data = open(LOG).read() if os.path.exists(LOG) else "" + self.send_response(200) + self.send_header("Content-Type", "application/x-ndjson") + self.send_header("Content-Length", str(len(data.encode()))) + self.end_headers() + self.wfile.write(data.encode()) + else: + self._reply(404, {"error": "not found"}) + + def log_message(self, fmt, *args): # keep the container log quiet + pass + + +if __name__ == "__main__": + os.makedirs(DATA, exist_ok=True) + ThreadingHTTPServer(("0.0.0.0", 5300), Handler).serve_forever()