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
This commit is contained in:
Joseph Doherty
Executable
+117
@@ -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 <instance 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<string, object?> {
|
||||
["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"
|
||||
@@ -0,0 +1,87 @@
|
||||
"""Tiny HTTP sink for the env2 store-and-forward soak (ScadaBridge#116).
|
||||
|
||||
POST/PUT <any path> -> 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": <lines>}
|
||||
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()
|
||||
Reference in New Issue
Block a user