Files
lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs
T
Joseph Doherty c74ccc55b5 feat(mqtt): flatten Sparkplug Template/DataSet metrics to sibling tags (#515)
`SparkplugCodec` handled scalars; `DataSet`, `Template` and `PropertySet` were skipped
with a warning. They now flatten to addressable tags (owner decision, recorded on the
issue):

  Template  → sibling metrics, `/`-joined, recursively: Motor → Motor/Speed, Motor/Stator/Temp
  DataSet   → one metric per cell, {column}/{rowIndex}: Recipe → Recipe/Step/0, Recipe/Target/1
  PropertySet → node metadata, never tags — it describes a metric rather than being one

Flattening was chosen over OPC UA Structures because flattened scalars are consumable by
any client with no type dictionary, and the result mirrors how the `/raw` tree works.

**The row index is always present, including on row 0.** Emitting a bare column name for
a single-row table and an indexed one otherwise would make a tag's name depend on how many
rows the plant published — a table growing from one row to two would rename every tag
already authored against it.

**A DATA metric carries the parent's alias and no name**, so member names cannot be
composed until the alias table has resolved the parent. Expansion therefore happens in
`SparkplugIngestor.PublishStructuredMembers` — after resolution, at the single funnel both
the DATA and birth-value paths already pass through — not at decode time. The alias table
installs a binding per member so browse enumerates them and a DDATA member that omits its
datatype falls back to the birth's (guessing from the CLR type would read a signed Int32 of
-42 out of its unsigned field as 4294967254).

**Skip-and-warn survives, and refusal is whole** — the explicit requirement on the issue.
A template definition, an unnamed member, an extension value, a column/type mismatch, a
short row, or a payload past MaxDepth(5)/MaxMembers(512) refuses the ENTIRE metric. A
silently half-flattened UDT looks exactly like a complete one in the picker.

The browse picker offers the members and deliberately NOT the structured parent, which has
no value and no servable datatype. The rediscovery signature likewise tracks members rather
than the parent, or a UDT gaining a member would leave the signature unmoved.

Member paths contain `/`, which #509 already solved: the slashed name travels intact as
`metricName` while the derived tag Name is coerced to a legal RawPath segment.

Live-gated 3/3 against the Mosquitto fixture with the project-owned simulator extended to
publish a nested Template and a 2x2 DataSet. Falsified at three points: dropping the member
bindings, not expanding the parent, and refusing to flatten templates each turn exactly the
expected tests red.

Also fixes a regression from the #511 commit in the same branch: `AddOtOpcUaDriverFactories`
resolved `IDriverPrimaryRoleSource` with `GetRequiredService` but only `AddOtOpcUaRuntime`
registered it, so a graph wiring driver factories without the full runtime failed at
resolution. Two Host integration suites do exactly that and caught it; both extensions now
TryAdd the same concrete `RedundancyRoleView`, so the registration is order-independent and
the fail-loud resolve is kept.

Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
2026-07-31 08:14:04 -04:00

174 lines
6.9 KiB
C#

using Org.Eclipse.Tahu.Protobuf;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator;
// ---------------------------------------------------------------------------
// Standalone entry point for the `sparkplug-sim` fixture container.
//
// Runs one or more simulated edge nodes against the fixture broker, forever: NBIRTH + DBIRTH once,
// then N/DDATA on a cadence, honouring rebirth NCMDs throughout. It exists so an operator (and Task
// 26's AdminUI `/run` verification) has a live Sparkplug plant to browse, NOT so the live tests have
// one — those drive `SparkplugEdgeNode` in process, because the §3.6 matrix needs an edge node it
// can command mid-test.
//
// Every setting comes from the environment; nothing is defaulted to a credential.
// ---------------------------------------------------------------------------
var host = Env("MQTT_SIM_HOST", "mosquitto");
var port = int.TryParse(Env("MQTT_SIM_PORT", "8883"), out var p) ? p : 8883;
var useTls = !string.Equals(Env("MQTT_SIM_USE_TLS", "true"), "false", StringComparison.OrdinalIgnoreCase);
var group = Env("MQTT_SIM_GROUP", "OtOpcUaSim");
var nodes = Env("MQTT_SIM_NODES", "EdgeA,EdgeB").Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
var interval = int.TryParse(Env("MQTT_SIM_INTERVAL_SECONDS", "2"), out var i) && i > 0 ? i : 2;
var username = Environment.GetEnvironmentVariable("MQTT_FIXTURE_USERNAME");
var password = Environment.GetEnvironmentVariable("MQTT_FIXTURE_PASSWORD");
var caPath = Environment.GetEnvironmentVariable("MQTT_SIM_CA_CERT");
var allowUntrusted = string.Equals(Env("MQTT_SIM_ALLOW_UNTRUSTED", "false"), "true", StringComparison.OrdinalIgnoreCase);
if (string.IsNullOrEmpty(password))
{
// Same fail-loud posture as the compose file's `${VAR:?}` guards: the broker runs
// allow_anonymous=false, so a blank password would produce an unexplained connect refusal.
Console.Error.WriteLine(
"sparkplug-sim: MQTT_FIXTURE_PASSWORD is not set. The fixture broker runs allow_anonymous=false; "
+ "supply the password gen-fixture-material.sh was run with.");
return 2;
}
using var stopping = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
e.Cancel = true;
stopping.Cancel();
};
AppDomain.CurrentDomain.ProcessExit += (_, _) => stopping.Cancel();
Console.WriteLine(
$"sparkplug-sim: broker={host}:{port} tls={useTls} group={group} nodes=[{string.Join(", ", nodes)}] "
+ $"interval={interval}s");
var running = new List<Task>();
var edges = new List<SparkplugEdgeNode>();
foreach (var node in nodes)
{
var edge = new SparkplugEdgeNode(
new SparkplugEdgeNodeOptions
{
Host = host,
Port = port,
UseTls = useTls,
CaCertificatePath = caPath,
AllowUntrustedServerCertificate = allowUntrusted,
Username = username,
Password = password,
GroupId = group,
EdgeNodeId = node,
},
Console.WriteLine);
// A small but genuinely heterogeneous catalog: a float, a double, a NEGATIVE int32 (the two's
// complement case), a bool and a string — plus one device, so the picker's
// Group → EdgeNode → Device → Metric tree has every level populated.
edge.SetNodeMetrics(
new SparkplugMetricSeed("Temperature", 1, DataType.Float, 21.5f),
new SparkplugMetricSeed("Pressure", 2, DataType.Double, 101.325d),
new SparkplugMetricSeed("Count", 3, DataType.Int32, -42),
new SparkplugMetricSeed("Running", 4, DataType.Boolean, true),
new SparkplugMetricSeed("Serial", 5, DataType.String, $"{node}-001"));
edge.SetDeviceMetrics(
"Filler1",
new SparkplugMetricSeed("Temperature", 10, DataType.Float, 55.5f),
new SparkplugMetricSeed("FillCount", 11, DataType.Int64, 1000L),
new SparkplugMetricSeed("Jammed", 12, DataType.Boolean, false),
// Structured metrics for the Gitea #515 gate. They live on the DEVICE rather than the node
// deliberately: a DDATA template is the case a naive implementation gets wrong, because its
// metric carries the parent's ALIAS and no name, so the member names cannot be composed until
// the alias table has resolved the parent.
new SparkplugMetricSeed("Motor", 13, DataType.Template, BuildMotorTemplate()),
new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, BuildRecipeDataSet()));
edges.Add(edge);
}
try
{
foreach (var edge in edges)
{
await edge.ConnectAsync(stopping.Token);
running.Add(edge.RunAsync(TimeSpan.FromSeconds(interval), stopping.Token));
}
await Task.WhenAll(running);
}
catch (OperationCanceledException)
{
Console.WriteLine("sparkplug-sim: stopping.");
}
finally
{
foreach (var edge in edges)
{
await edge.DisposeAsync();
}
}
return 0;
static string Env(string name, string fallback) =>
Environment.GetEnvironmentVariable(name) is { Length: > 0 } value ? value : fallback;
// A UDT instance: `Motor` with a `Speed` double, a `Running` bool and a nested `Stator` template.
// Flattens to `Motor/Speed`, `Motor/Running` and `Motor/Stator/Temp` (Gitea #515).
static Payload.Types.Template BuildMotorTemplate()
{
var stator = new Payload.Types.Template { Version = "1", IsDefinition = false };
stator.Metrics.Add(new Payload.Types.Metric
{
Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = 71.5,
});
var motor = new Payload.Types.Template { Version = "1", IsDefinition = false };
motor.Metrics.Add(new Payload.Types.Metric
{
Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = 1500,
});
motor.Metrics.Add(new Payload.Types.Metric
{
Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = true,
});
// Nested one level: proves the recursion, and proves the `/`-joined path composes correctly rather
// than only working for flat templates.
motor.Metrics.Add(new Payload.Types.Metric
{
Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator,
});
return motor;
}
// A two-column, two-row table. Flattens to `Recipe/Step/0`, `Recipe/Target/0`, `Recipe/Step/1`,
// `Recipe/Target/1` — the row index is always present, so a table that grows a row never renames the
// tags already authored against it.
static Payload.Types.DataSet BuildRecipeDataSet()
{
var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL };
dataSet.Columns.Add("Step");
dataSet.Columns.Add("Target");
dataSet.Types_.Add((uint)DataType.Int32);
dataSet.Types_.Add((uint)DataType.Double);
for (var row = 0; row < 2; row++)
{
var r = new Payload.Types.DataSet.Types.Row();
r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row });
r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = 20d + row });
dataSet.Rows.Add(r);
}
return dataSet;
}