feat(logging): bridge Akka logging to Serilog and copy traced runs' node-log lines into the Flow Debugger (ScadaBridge#120, #56)
#120: Akka ILoggingAdapter lines used to bypass Serilog and go to stdout only. - BuildHocon sets akka.loggers = [Host.Actors.SerilogAkkaLogger] and keeps akka.loglevel = INFO, so Akka Debug chatter is still filtered at the adapter. The INFO+ lines that used to reach stdout now go through the Serilog pipeline instead, with no extra volume. - Lines reach the file sinks with the node enrichers. SourceContext is the logging class's full name, AkkaSource is the actor path, Akka's own timestamp is kept, and format arguments become structured properties. - The Akka logger actor writes on its own thread, after the logging actor's LogContext is gone. To keep the ExecutionId, the two DCL actors log through ExecutionAwareLogging.GetLogger(Context), which records ExecutionLogContext.Current (a new AsyncLocal mirror of the pushed id) in an ExecutionTaggedLogMessage. The logger actor pushes that id again while it writes the line. #56 remainder (coordinator decision: no replicated schema change): - The host pipeline gains ExecutionLogCaptureSink. - While a run is traced, its step logger registers its ExecutionId: the site ScriptLogger from the launcher (plus each inline CallShared child id), and the central InboundScriptLogger from the executor. - Each tagged line becomes a ScriptLog row with LogSource "node" and the new ScriptTraceExtra.LogCategory. - The rows count against the existing MaxLogLinesPerRun cap, are clipped at 2 000 chars, ride the low-priority collector lane and pass the same redactor. Untraced runs cost one volatile read. - Lines are never copied twice. The step loggers' own forwards run inside ExecutionLogCapture.Suppress(), and the sink suppresses itself while it calls a target. The registration is declared inside the launcher's and executor's try, so it ends before the catch blocks log the outcome, and FlushDropped closes it. - FlowGraphBuilder lists node lines in FlowGraph.NodeLogs and never folds them onto a step. - The Flow Debugger shows them in a Node log card (FlowNodeLogPanel): grouped per execution under its run step, with an All | Warn + filter. Live and Replay both read the same rows, so site scope and role rules apply unchanged. - CLI --show-logs prints them as a trailing block. Tests: Commons ExecutionLogCaptureTests, ExecutionLogContext.Current, FlowGraphBuilder node lines. SiteRuntime NodeLogCaptureTests covers cross-actor capture via the launcher, untraced runs capturing nothing, no double capture of script and runtime lines or of the launcher failure line, the shared cap, collector redaction, shared-script child ids and FlushDropped closing capture. InboundNodeLogCaptureTests covers the central path. DCL ExecutionAwareLoggingTests covers the DataConnectionActor write-path Debug line being tagged and an uncorrelated line not being tagged. Host SerilogAkkaLoggerTests covers mapping, the ExecutionId crossing the logger hop, HOCON, ActorSystem boot and the Akka line reaching a traced run's step log end to end. Host LoggerConfigurationTests checks that the sink is wired. bUnit FlowNodeLogPanelTests and a CLI projection test cover the UI and CLI. Every new assertion was mutation-proven. Docs: step-logs design section 12, Component-SiteRuntime, CentralUI, InboundAPI, Host, CLI README, CLAUDE.md. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
@@ -423,6 +423,44 @@ public class FlowDebugCommandsTests
|
||||
Assert.DoesNotContain("at Z", projected);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#56: node lines are not on a step (the logs column does not count them); with
|
||||
/// --show-logs they print as one trailing block, each naming its logger without the namespace.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Projection_ShowLogs_PrintsNodeLinesAsATrailingBlock()
|
||||
{
|
||||
var root = Guid.NewGuid();
|
||||
var t0 = new DateTime(2026, 11, 10, 10, 0, 0, DateTimeKind.Utc);
|
||||
var events = new List<AuditEvent>
|
||||
{
|
||||
ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
||||
occurredAtUtc: t0, target: "Runner", executionId: root, sourceSiteId: "site-a",
|
||||
sourceInstanceId: "M1", sourceScript: "Runner", durationMs: 50),
|
||||
ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptLog, AuditStatus.Delivered,
|
||||
occurredAtUtc: t0.AddMilliseconds(5), target: "Runner", executionId: root, requestSummary: "Writing to tag T1",
|
||||
extra: new ScriptTraceExtra
|
||||
{
|
||||
LogLevel = "warn", LogSource = "node", Seq = 1,
|
||||
LogCategory = "ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors.DataConnectionActor",
|
||||
}.ToJson()),
|
||||
};
|
||||
var json = JsonSerializer.Serialize(
|
||||
new FlowDebugRunResult(root, true, null, null, events, false, new DateTimeOffset(t0), new DateTimeOffset(t0.AddSeconds(1))),
|
||||
CamelCase);
|
||||
|
||||
using var plain = JsonDocument.Parse(FlowDebugTableProjection.ProjectEvents(json));
|
||||
var row = Assert.Single(plain.RootElement.EnumerateArray());
|
||||
Assert.Equal(0, row.GetProperty("logs").GetInt32());
|
||||
|
||||
using var shown = JsonDocument.Parse(FlowDebugTableProjection.ProjectEvents(json, showLogs: true));
|
||||
var rows = shown.RootElement.EnumerateArray().ToList();
|
||||
Assert.Equal(3, rows.Count);
|
||||
Assert.Equal("node log (1 line):", rows[1].GetProperty("note").GetString());
|
||||
Assert.Equal($" {t0.AddMilliseconds(5):HH:mm:ss.fff} WARN [DataConnectionActor] Writing to tag T1",
|
||||
rows[2].GetProperty("note").GetString());
|
||||
}
|
||||
|
||||
/// <summary>Verifies that hiding logs omits the log and exception detail lines.</summary>
|
||||
[Fact]
|
||||
public void Projection_ShowLogsFalse_OmitsLogAndExceptionDetailLines()
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
using Bunit;
|
||||
using ZB.MOM.WW.ScadaBridge.CentralUI.Components.Deployment;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.CentralUI.Tests.Deployment;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#56: the Flow Debugger's node-log section lists the lines other actors wrote for each
|
||||
/// traced execution — grouped per execution under that execution's run step, with level filtering and
|
||||
/// the logger category — and renders nothing when the graph has no node lines.
|
||||
/// </summary>
|
||||
public class FlowNodeLogPanelTests : BunitContext
|
||||
{
|
||||
private static readonly DateTimeOffset T0 = new(2026, 11, 10, 12, 0, 0, TimeSpan.Zero);
|
||||
|
||||
private static FlowStep Run(Guid execution, string title)
|
||||
=> new(Guid.NewGuid(), execution, null, FlowStepKind.ScriptRun, FlowLane.Site, title, null, T0, 10,
|
||||
FlowStepOutcome.Ok, null, null, null, "site-a", "node-a", "M1", title, null);
|
||||
|
||||
private static FlowLogLine Node(Guid execution, string level, string message, int ms,
|
||||
string category = "ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors.DataConnectionActor",
|
||||
string? exceptionMessage = null)
|
||||
=> new(Guid.NewGuid(), execution, T0.AddMilliseconds(ms), level, FlowLogLine.NodeSource, message,
|
||||
exceptionMessage, exceptionMessage is null ? null : "System.Exception: " + exceptionMessage, null, ms, category);
|
||||
|
||||
private static FlowGraph Graph(IReadOnlyList<FlowStep> steps, IReadOnlyList<FlowLogLine>? nodeLogs)
|
||||
=> new(steps, [], Predicted: false, Truncated: false, NodeLogs: nodeLogs);
|
||||
|
||||
/// <summary>No node lines (or no graph): the section is not rendered at all.</summary>
|
||||
[Fact]
|
||||
public void NoNodeLines_RendersNothing()
|
||||
{
|
||||
var execution = Guid.NewGuid();
|
||||
Assert.Empty(Render<FlowNodeLogPanel>(p => p.Add(x => x.Graph, Graph([Run(execution, "Runner")], null)))
|
||||
.FindAll("[data-test=flow-nodelog-panel]"));
|
||||
Assert.Empty(Render<FlowNodeLogPanel>().FindAll("[data-test=flow-nodelog-panel]"));
|
||||
}
|
||||
|
||||
/// <summary>Lines are grouped per execution, titled by its run step, with short category, level and counts.</summary>
|
||||
[Fact]
|
||||
public void Lines_AreGroupedPerExecution_UnderTheRunStep()
|
||||
{
|
||||
var a = Guid.NewGuid();
|
||||
var b = Guid.NewGuid();
|
||||
var graph = Graph([Run(a, "OnRecipe"), Run(b, "SharedHelper")],
|
||||
[
|
||||
Node(a, "debug", "Writing to tag T1", 1),
|
||||
Node(b, "warn", "tag T2 bad quality", 2, category: "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.InstanceActor"),
|
||||
Node(a, "error", "write failed", 3, exceptionMessage: "timeout"),
|
||||
]);
|
||||
|
||||
var cut = Render<FlowNodeLogPanel>(p => p.Add(x => x.Graph, graph));
|
||||
|
||||
Assert.Equal("3 lines · 1 error · 1 warning · 1 info/debug", cut.Find("[data-test=flow-nodelog-count]").TextContent);
|
||||
var groups = cut.FindAll("[data-test=flow-nodelog-execution]");
|
||||
Assert.Equal(2, groups.Count);
|
||||
Assert.Contains("OnRecipe", groups[0].TextContent);
|
||||
Assert.Contains(a.ToString(), groups[0].TextContent);
|
||||
var aRows = groups[0].QuerySelectorAll("tbody tr");
|
||||
Assert.Equal(2, aRows.Length);
|
||||
Assert.Contains("DataConnectionActor", aRows[0].TextContent);
|
||||
Assert.DoesNotContain("ZB.MOM.WW", aRows[0].TextContent);
|
||||
Assert.Contains("timeout", aRows[1].TextContent);
|
||||
Assert.Contains("InstanceActor", groups[1].TextContent);
|
||||
Assert.Contains("tag T2 bad quality", groups[1].TextContent);
|
||||
}
|
||||
|
||||
/// <summary>Warn + hides debug/info lines; an execution left with none says so; All restores them.</summary>
|
||||
[Fact]
|
||||
public void WarnPlusFilter_HidesLowLevels()
|
||||
{
|
||||
var a = Guid.NewGuid();
|
||||
var b = Guid.NewGuid();
|
||||
var graph = Graph([Run(a, "OnRecipe"), Run(b, "Other")],
|
||||
[Node(a, "info", "connected", 1), Node(a, "warn", "slow write", 2), Node(b, "debug", "chatter", 3)]);
|
||||
var cut = Render<FlowNodeLogPanel>(p => p.Add(x => x.Graph, graph));
|
||||
|
||||
cut.Find("[data-test=flow-nodelog-filter-warn]").Click();
|
||||
|
||||
var groups = cut.FindAll("[data-test=flow-nodelog-execution]");
|
||||
Assert.Equal("warn", Assert.Single(groups[0].QuerySelectorAll("tbody tr")).GetAttribute("data-level"));
|
||||
Assert.NotNull(groups[1].QuerySelector("[data-test=flow-nodelog-none-shown]"));
|
||||
|
||||
cut.Find("[data-test=flow-nodelog-filter-all]").Click();
|
||||
Assert.Equal(3, cut.FindAll("[data-test=flow-nodelog-lines] tbody tr").Count);
|
||||
}
|
||||
|
||||
/// <summary>Clicking the execution's title selects its run step.</summary>
|
||||
[Fact]
|
||||
public void ExecutionTitle_SelectsItsRunStep()
|
||||
{
|
||||
var a = Guid.NewGuid();
|
||||
var run = Run(a, "OnRecipe");
|
||||
FlowStep? selected = null;
|
||||
var cut = Render<FlowNodeLogPanel>(p => p
|
||||
.Add(x => x.Graph, Graph([run], [Node(a, "warn", "slow", 1)]))
|
||||
.Add(x => x.OnSelectStep, s => selected = s));
|
||||
|
||||
cut.Find("[data-test=flow-nodelog-step]").Click();
|
||||
|
||||
Assert.Same(run, selected);
|
||||
}
|
||||
}
|
||||
@@ -213,4 +213,46 @@ public class FlowGraphBuilderLogLineTests
|
||||
Assert.Single(call.Logs!);
|
||||
Assert.Null(StepOf(graph, FlowStepKind.ScriptRun).Logs);
|
||||
}
|
||||
|
||||
private static AuditEvent NodeLine(Guid execution, int ms, int? seq, string message, string category = "Acme.Dcl") =>
|
||||
ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptLog, AuditStatus.Delivered,
|
||||
occurredAtUtc: T0.AddMilliseconds(ms), actor: "SomeScript", target: "SomeScript", sourceNode: "node-a",
|
||||
executionId: execution, sourceSiteId: "site-a", sourceInstanceId: "M1", requestSummary: message,
|
||||
extra: new ScriptTraceExtra { LogLevel = "warn", LogSource = "node", LogCategory = category, Seq = seq }.ToJson());
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#56: node lines are listed on the graph in time order (seq breaking ties), never folded
|
||||
/// onto a step — not even the anchor — and never counted as unattached, while script lines still fold.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void NodeLines_AreListedOnTheGraph_NotFoldedOntoSteps()
|
||||
{
|
||||
var execution = Guid.NewGuid();
|
||||
var orphan = Guid.NewGuid(); // an execution with no step at all
|
||||
|
||||
var graph = FlowGraphBuilder.Build(
|
||||
[
|
||||
Run(execution), Line(execution),
|
||||
NodeLine(execution, ms: 30, seq: 2, message: "late"),
|
||||
NodeLine(execution, ms: 20, seq: 9, message: "early-b"),
|
||||
NodeLine(execution, ms: 20, seq: 1, message: "early-a"),
|
||||
NodeLine(orphan, ms: 40, seq: 1, message: "orphan"),
|
||||
]);
|
||||
|
||||
Assert.Equal(["early-a", "early-b", "late", "orphan"], graph.NodeLogs!.Select(l => l.Message));
|
||||
Assert.All(graph.NodeLogs!, l => Assert.True(l.IsNodeLine));
|
||||
Assert.Equal("Acme.Dcl", graph.NodeLogs![0].Category);
|
||||
var anchorLogs = StepOf(graph, FlowStepKind.ScriptRun).Logs!;
|
||||
Assert.Equal("hello", Assert.Single(anchorLogs).Message);
|
||||
Assert.Equal(0, graph.UnattachedLogLines);
|
||||
}
|
||||
|
||||
/// <summary>A trace with no node lines has a null <see cref="FlowGraph.NodeLogs"/>, including a step-less one.</summary>
|
||||
[Fact]
|
||||
public void NoNodeLines_LeavesNodeLogsNull()
|
||||
{
|
||||
var execution = Guid.NewGuid();
|
||||
Assert.Null(FlowGraphBuilder.Build([Run(execution), Line(execution)]).NodeLogs);
|
||||
Assert.Single(FlowGraphBuilder.Build([NodeLine(execution, 1, 1, "only")]).NodeLogs!);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Serilog;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Commons.Tests.Observability;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#56: the per-execution sink copies a line to the target registered for the line's
|
||||
/// <c>ExecutionId</c> — and to nothing else. Each test uses fresh ids, because the registry is
|
||||
/// process-wide and other tests may register concurrently.
|
||||
/// </summary>
|
||||
public class ExecutionLogCaptureTests
|
||||
{
|
||||
private sealed class Target : IExecutionLogTarget
|
||||
{
|
||||
public ConcurrentQueue<CapturedLogLine> Lines { get; } = new();
|
||||
public Action<CapturedLogLine>? OnCapture { get; init; }
|
||||
|
||||
public void Capture(CapturedLogLine line)
|
||||
{
|
||||
Lines.Enqueue(line);
|
||||
OnCapture?.Invoke(line);
|
||||
}
|
||||
}
|
||||
|
||||
private static Serilog.ILogger Node() => new LoggerConfiguration()
|
||||
.MinimumLevel.Information()
|
||||
.Enrich.FromLogContext()
|
||||
.WriteTo.Sink(new ExecutionLogCaptureSink())
|
||||
.CreateLogger();
|
||||
|
||||
/// <summary>A line tagged with a registered id reaches its target with level, category, text, exception and time.</summary>
|
||||
[Fact]
|
||||
public void TaggedLine_ReachesItsTarget()
|
||||
{
|
||||
var node = Node().ForContext("SourceContext", "Acme.InstanceActor");
|
||||
var id = Guid.NewGuid();
|
||||
var target = new Target();
|
||||
using var registration = ExecutionLogCapture.Register(id, target);
|
||||
|
||||
using (ExecutionLogContext.Push(id))
|
||||
node.Warning(new InvalidOperationException("bad"), "wrote {Tag}", "T1");
|
||||
|
||||
var line = Assert.Single(target.Lines);
|
||||
Assert.Equal("warn", line.Level);
|
||||
Assert.Equal("Acme.InstanceActor", line.Category);
|
||||
Assert.Equal("wrote \"T1\"", line.Message);
|
||||
Assert.Equal("bad", line.Exception!.Message);
|
||||
Assert.Equal(DateTimeKind.Utc, line.OccurredAtUtc.Kind);
|
||||
Assert.True(ExecutionLogCapture.HasTargets);
|
||||
}
|
||||
|
||||
/// <summary>Untagged lines, lines of another execution, and lines below the node's level are not captured.</summary>
|
||||
[Fact]
|
||||
public void UntaggedOrOtherOrFilteredLines_AreNotCaptured()
|
||||
{
|
||||
var node = Node();
|
||||
var id = Guid.NewGuid();
|
||||
var target = new Target();
|
||||
using var registration = ExecutionLogCapture.Register(id, target);
|
||||
|
||||
node.Information("untagged");
|
||||
using (ExecutionLogContext.Push(Guid.NewGuid()))
|
||||
node.Information("other execution");
|
||||
using (ExecutionLogContext.Push(id))
|
||||
node.Debug("below the node minimum");
|
||||
|
||||
Assert.Empty(target.Lines);
|
||||
}
|
||||
|
||||
/// <summary>A line written inside <see cref="ExecutionLogCapture.Suppress"/> (already a trace row) is skipped.</summary>
|
||||
[Fact]
|
||||
public void SuppressedLine_IsNotCaptured()
|
||||
{
|
||||
var node = Node();
|
||||
var id = Guid.NewGuid();
|
||||
var target = new Target();
|
||||
using var registration = ExecutionLogCapture.Register(id, target);
|
||||
|
||||
using (ExecutionLogContext.Push(id))
|
||||
{
|
||||
using (ExecutionLogCapture.Suppress())
|
||||
node.Information("already a row");
|
||||
node.Information("captured");
|
||||
}
|
||||
|
||||
Assert.Equal("captured", Assert.Single(target.Lines).Message);
|
||||
}
|
||||
|
||||
/// <summary>A target that logs for the same execution while capturing does not loop back into itself.</summary>
|
||||
[Fact]
|
||||
public void TargetThatLogs_DoesNotRecapture()
|
||||
{
|
||||
var node = Node();
|
||||
var id = Guid.NewGuid();
|
||||
// Logs once only, so a broken guard fails this test instead of overflowing the stack.
|
||||
var logged = 0;
|
||||
var target = new Target { OnCapture = _ => { if (Interlocked.Increment(ref logged) == 1) node.Warning("emit failed"); } };
|
||||
using var registration = ExecutionLogCapture.Register(id, target);
|
||||
|
||||
using (ExecutionLogContext.Push(id))
|
||||
node.Information("first");
|
||||
|
||||
Assert.Equal("first", Assert.Single(target.Lines).Message);
|
||||
}
|
||||
|
||||
/// <summary>A line logged on another thread that re-pushed the id (an actor serving the run) is captured.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task LineOnAnotherThread_WithTheIdRepushed_IsCaptured()
|
||||
{
|
||||
var node = Node();
|
||||
var id = Guid.NewGuid();
|
||||
var target = new Target();
|
||||
using var registration = ExecutionLogCapture.Register(id, target);
|
||||
|
||||
await Task.Factory.StartNew(() =>
|
||||
{
|
||||
using (ExecutionLogContext.Push(id))
|
||||
node.Information("from an actor thread");
|
||||
}, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
|
||||
|
||||
Assert.Equal("from an actor thread", Assert.Single(target.Lines).Message);
|
||||
}
|
||||
|
||||
/// <summary>Disposing removes the registration; a duplicate registration is ignored and its disposal harmless.</summary>
|
||||
[Fact]
|
||||
public void Dispose_Unregisters_AndDuplicateIsIgnored()
|
||||
{
|
||||
var node = Node();
|
||||
var id = Guid.NewGuid();
|
||||
var first = new Target();
|
||||
var second = new Target();
|
||||
var registration = ExecutionLogCapture.Register(id, first);
|
||||
ExecutionLogCapture.Register(id, second).Dispose(); // duplicate: no-op, must not remove the first
|
||||
|
||||
Assert.True(ExecutionLogCapture.TryGetTarget(id, out var found));
|
||||
Assert.Same(first, found);
|
||||
|
||||
registration.Dispose();
|
||||
registration.Dispose();
|
||||
Assert.False(ExecutionLogCapture.TryGetTarget(id, out _));
|
||||
using (ExecutionLogContext.Push(id))
|
||||
node.Information("after");
|
||||
Assert.Empty(first.Lines);
|
||||
Assert.Empty(second.Lines);
|
||||
}
|
||||
}
|
||||
@@ -102,4 +102,29 @@ public class ExecutionLogContextTests
|
||||
Assert.Equal(child.ToString(), events[1].Properties[ExecutionLogContext.ExecutionIdProperty].ToString().Trim('"'));
|
||||
Assert.False(events[2].Properties.ContainsKey(ExecutionLogContext.ExecutionIdProperty));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#120: <see cref="ExecutionLogContext.Current"/> mirrors the pushed id, follows
|
||||
/// nesting, and is restored when each push is disposed (twice is harmless).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Current_TracksPushes_AndIsRestoredOnDispose()
|
||||
{
|
||||
var outer = Guid.NewGuid();
|
||||
var inner = Guid.NewGuid();
|
||||
Assert.Null(ExecutionLogContext.Current);
|
||||
|
||||
var a = ExecutionLogContext.Push(outer);
|
||||
Assert.Equal(outer, ExecutionLogContext.Current);
|
||||
var b = ExecutionLogContext.Push(inner);
|
||||
Assert.Equal(inner, ExecutionLogContext.Current);
|
||||
Assert.Null(ExecutionLogContext.Push(null)); // a null push changes nothing
|
||||
Assert.Equal(inner, ExecutionLogContext.Current);
|
||||
|
||||
b!.Dispose();
|
||||
b.Dispose();
|
||||
Assert.Equal(outer, ExecutionLogContext.Current);
|
||||
a!.Dispose();
|
||||
Assert.Null(ExecutionLogContext.Current);
|
||||
}
|
||||
}
|
||||
|
||||
+131
@@ -0,0 +1,131 @@
|
||||
using Akka.Actor;
|
||||
using Akka.Event;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using NSubstitute;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Logging;
|
||||
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#120: the DCL actors' own Akka <c>_log</c> lines record the execution id of the
|
||||
/// message being served, so the host's Serilog logger actor can tag them although it writes them
|
||||
/// on another thread.
|
||||
/// </summary>
|
||||
public class ExecutionAwareLoggingTests : TestKit
|
||||
{
|
||||
/// <summary>Debug level so the DCL's write-path Debug line is published.</summary>
|
||||
public ExecutionAwareLoggingTests() : base("akka.loglevel = DEBUG")
|
||||
{
|
||||
}
|
||||
|
||||
private sealed record LogUnder(Guid? Id);
|
||||
|
||||
private sealed class Probe : ReceiveActor
|
||||
{
|
||||
private readonly ILoggingAdapter _log = ExecutionAwareLogging.GetLogger(Context);
|
||||
|
||||
public Probe()
|
||||
{
|
||||
Receive<LogUnder>(m =>
|
||||
{
|
||||
var id = m.Id;
|
||||
using (ExecutionLogContext.Push(id))
|
||||
_log.Info("probe {0}", id?.ToString() ?? "none");
|
||||
});
|
||||
Receive<string>(_ => _log.Debug("debug {0}", "x"));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A line logged inside an execution's push is published wrapped with that id; its text is unchanged.</summary>
|
||||
[Fact]
|
||||
public void LineInsideAPush_IsTaggedWithTheExecutionId_TextUnchanged()
|
||||
{
|
||||
var events = CreateTestProbe();
|
||||
Sys.EventStream.Subscribe(events.Ref, typeof(Info));
|
||||
var id = Guid.NewGuid();
|
||||
|
||||
Sys.ActorOf(Props.Create(() => new Probe())).Tell(new LogUnder(id));
|
||||
|
||||
var info = events.FishForMessage<Info>(i => i.Message.ToString()!.StartsWith("probe", StringComparison.Ordinal));
|
||||
var tagged = Assert.IsType<ExecutionTaggedLogMessage>(info.Message);
|
||||
Assert.Equal(id, tagged.ExecutionId);
|
||||
Assert.Equal($"probe {id}", info.Message.ToString());
|
||||
Assert.IsAssignableFrom<LogMessage>(tagged.Message);
|
||||
}
|
||||
|
||||
/// <summary>With no execution current, the line is published exactly as <c>Context.GetLogger()</c> would publish it.</summary>
|
||||
[Fact]
|
||||
public void LineOutsideAPush_IsPublishedUnwrapped()
|
||||
{
|
||||
var events = CreateTestProbe();
|
||||
Sys.EventStream.Subscribe(events.Ref, typeof(Info));
|
||||
|
||||
Sys.ActorOf(Props.Create(() => new Probe())).Tell(new LogUnder(null));
|
||||
|
||||
var info = events.FishForMessage<Info>(i => i.Message.ToString()!.StartsWith("probe", StringComparison.Ordinal));
|
||||
Assert.IsAssignableFrom<LogMessage>(info.Message);
|
||||
Assert.Equal("probe none", info.Message.ToString());
|
||||
}
|
||||
|
||||
/// <summary>The adapter honours <c>akka.loglevel</c>: a Debug line is not published once the bus is at Info.</summary>
|
||||
[Fact]
|
||||
public void DisabledLevel_PublishesNothing()
|
||||
{
|
||||
Sys.EventStream.SetLogLevel(Akka.Event.LogLevel.InfoLevel);
|
||||
var events = CreateTestProbe();
|
||||
Sys.EventStream.Subscribe(events.Ref, typeof(Debug));
|
||||
|
||||
Sys.ActorOf(Props.Create(() => new Probe())).Tell("debug");
|
||||
|
||||
events.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// End to end on the DCL: the DataConnectionActor's own write-path Debug line, logged while it
|
||||
/// serves a correlated WriteTagRequest, is tagged with the request's execution id; an
|
||||
/// uncorrelated write's line is not (control).
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task DataConnectionActor_WritePathLine_IsTagged_UncorrelatedIsNot()
|
||||
{
|
||||
var adapter = Substitute.For<IDataConnection>();
|
||||
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.CompletedTask);
|
||||
adapter.WriteAsync(Arg.Any<string>(), Arg.Any<object?>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.FromResult(new WriteResult(true, null)));
|
||||
var options = new DataConnectionOptions
|
||||
{
|
||||
ReconnectInterval = TimeSpan.FromMilliseconds(100),
|
||||
WriteTimeout = TimeSpan.FromSeconds(5),
|
||||
SeedReadMaxAttempts = 1,
|
||||
SeedReadRetryDelay = TimeSpan.FromMilliseconds(10),
|
||||
};
|
||||
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
||||
"akka-tag", adapter, options, Substitute.For<ISiteHealthCollector>(),
|
||||
Substitute.For<IDataConnectionFactory>(), "OpcUa")), "akka-tag");
|
||||
AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "ConnectAsync"),
|
||||
TimeSpan.FromSeconds(3));
|
||||
await Task.Delay(200);
|
||||
|
||||
var events = CreateTestProbe();
|
||||
Sys.EventStream.Subscribe(events.Ref, typeof(Debug));
|
||||
var id = Guid.NewGuid();
|
||||
|
||||
actor.Tell(new WriteTagRequest("c1", "akka-tag", "tagT", 1, DateTimeOffset.UtcNow, id));
|
||||
Assert.True(ExpectMsg<WriteTagResponse>(TimeSpan.FromSeconds(5)).Success);
|
||||
var tagged = events.FishForMessage<Debug>(d => d.Message.ToString()!.Contains("Writing to tag tagT", StringComparison.Ordinal));
|
||||
Assert.Equal(id, Assert.IsType<ExecutionTaggedLogMessage>(tagged.Message).ExecutionId);
|
||||
|
||||
actor.Tell(new WriteTagRequest("c2", "akka-tag", "tagU", 1, DateTimeOffset.UtcNow));
|
||||
Assert.True(ExpectMsg<WriteTagResponse>(TimeSpan.FromSeconds(5)).Success);
|
||||
var plain = events.FishForMessage<Debug>(d => d.Message.ToString()!.Contains("Writing to tag tagU", StringComparison.Ordinal));
|
||||
Assert.IsNotType<ExecutionTaggedLogMessage>(plain.Message);
|
||||
}
|
||||
}
|
||||
@@ -186,4 +186,32 @@ public class LoggerConfigurationTests
|
||||
// a recognised level, so ParseLevel is silent too.)
|
||||
Assert.Empty(writer.ToString());
|
||||
}
|
||||
|
||||
private sealed class CollectingTarget : ZB.MOM.WW.ScadaBridge.Commons.Observability.IExecutionLogTarget
|
||||
{
|
||||
public List<ZB.MOM.WW.ScadaBridge.Commons.Observability.CapturedLogLine> Lines { get; } = new();
|
||||
public void Capture(ZB.MOM.WW.ScadaBridge.Commons.Observability.CapturedLogLine line) => Lines.Add(line);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#56: the host pipeline carries the per-execution capture sink, so a line tagged with
|
||||
/// a registered (traced) execution reaches that execution's step log.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Build_IncludesTheExecutionCaptureSink()
|
||||
{
|
||||
var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>()).Build();
|
||||
using var logger = LoggerConfigurationFactory.Build(configuration, "Site", "site-a", "node1", new StringWriter())
|
||||
.CreateLogger();
|
||||
var id = Guid.NewGuid();
|
||||
var target = new CollectingTarget();
|
||||
|
||||
using (ZB.MOM.WW.ScadaBridge.Commons.Observability.ExecutionLogCapture.Register(id, target))
|
||||
using (ZB.MOM.WW.ScadaBridge.Commons.Observability.ExecutionLogContext.Push(id))
|
||||
{
|
||||
logger.Warning("tagged line");
|
||||
}
|
||||
|
||||
Assert.Equal("tagged line", Assert.Single(target.Lines).Message);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Akka.Actor;
|
||||
using Akka.Configuration;
|
||||
using Akka.Event;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Serilog;
|
||||
using Serilog.Core;
|
||||
using Serilog.Events;
|
||||
using ZB.MOM.WW.ScadaBridge.ClusterInfrastructure;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Logging;
|
||||
using ZB.MOM.WW.ScadaBridge.Host.Actors;
|
||||
using AkkaLogEvent = Akka.Event.LogEvent;
|
||||
using SerilogLogEvent = Serilog.Events.LogEvent;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#120: Akka <c>ILoggingAdapter</c> lines reach the host's Serilog pipeline, with the
|
||||
/// node enrichers and, for an execution-aware adapter, the <c>ExecutionId</c> of the message the
|
||||
/// actor was serving.
|
||||
/// </summary>
|
||||
public class SerilogAkkaLoggerTests : TestKit
|
||||
{
|
||||
private sealed class CaptureSink : ILogEventSink
|
||||
{
|
||||
public ConcurrentQueue<SerilogLogEvent> Events { get; } = new();
|
||||
public void Emit(SerilogLogEvent logEvent) => Events.Enqueue(logEvent);
|
||||
}
|
||||
|
||||
/// <summary>An actor that logs through the execution-aware adapter while serving a message.</summary>
|
||||
private sealed class LoggingActor : ReceiveActor
|
||||
{
|
||||
private readonly ILoggingAdapter _log = ExecutionAwareLogging.GetLogger(Context);
|
||||
|
||||
public LoggingActor()
|
||||
{
|
||||
Receive<(Guid? Id, string Text)>(m =>
|
||||
{
|
||||
using (ExecutionLogContext.Push(m.Id))
|
||||
_log.Warning("actor line {0}", m.Text);
|
||||
Sender.Tell("done");
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private readonly CaptureSink _sink = new();
|
||||
private readonly Serilog.ILogger _serilog;
|
||||
|
||||
/// <summary>
|
||||
/// The system keeps Akka's default logger (TestKit's); each test that needs the bridge spawns a
|
||||
/// <see cref="SerilogAkkaLogger"/> over a capturing pipeline that mirrors the host's
|
||||
/// <c>LoggerConfigurationFactory</c> (<c>Enrich.FromLogContext</c> plus a node property).
|
||||
/// </summary>
|
||||
public SerilogAkkaLoggerTests() : base("akka.loglevel = INFO")
|
||||
{
|
||||
_serilog = new LoggerConfiguration()
|
||||
.MinimumLevel.Information()
|
||||
.Enrich.WithProperty("NodeRole", "site-a")
|
||||
.Enrich.FromLogContext()
|
||||
.WriteTo.Sink(_sink)
|
||||
.CreateLogger();
|
||||
}
|
||||
|
||||
private void StartBridge()
|
||||
{
|
||||
var bridge = Sys.ActorOf(Props.Create(() => new SerilogAkkaLogger(_serilog)));
|
||||
Sys.EventStream.Subscribe(bridge, typeof(AkkaLogEvent));
|
||||
}
|
||||
|
||||
private SerilogLogEvent AwaitLine(string fragment)
|
||||
{
|
||||
SerilogLogEvent? found = null;
|
||||
AwaitCondition(() =>
|
||||
{
|
||||
found = _sink.Events.FirstOrDefault(e => e.RenderMessage().Contains(fragment, StringComparison.Ordinal));
|
||||
return found is not null;
|
||||
}, TimeSpan.FromSeconds(5));
|
||||
return found!;
|
||||
}
|
||||
|
||||
private static string? Scalar(SerilogLogEvent e, string name)
|
||||
=> e.Properties.TryGetValue(name, out var v) && v is ScalarValue { Value: { } x } ? x.ToString() : null;
|
||||
|
||||
/// <summary>An Akka line reaches Serilog with its level, class, actor path, structured args and node enrichers.</summary>
|
||||
[Fact]
|
||||
public void AkkaLine_ReachesSerilog_WithSourceContextAndEnrichers()
|
||||
{
|
||||
StartBridge();
|
||||
var actor = Sys.ActorOf(Props.Create(() => new LoggingActor()), "plain-logger");
|
||||
|
||||
actor.Tell(((Guid?)null, "alpha"));
|
||||
ExpectMsg("done");
|
||||
|
||||
var line = AwaitLine("actor line");
|
||||
Assert.Equal(LogEventLevel.Warning, line.Level);
|
||||
Assert.Equal(typeof(LoggingActor).FullName, Scalar(line, "SourceContext"));
|
||||
Assert.Contains("plain-logger", Scalar(line, SerilogAkkaLogger.AkkaSourceProperty));
|
||||
Assert.Equal("site-a", Scalar(line, "NodeRole"));
|
||||
Assert.Equal("actor line \"alpha\"", line.RenderMessage());
|
||||
// The adapter's argument is a structured property, not baked into the text.
|
||||
Assert.Equal("alpha", Scalar(line, "0"));
|
||||
Assert.False(line.Properties.ContainsKey(ExecutionLogContext.ExecutionIdProperty));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The line logged while an execution was current carries that id, although Serilog writes it on
|
||||
/// the logger actor's thread, where the logging actor's LogContext is not in effect.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ExecutionAwareLine_CarriesTheExecutionId_AcrossTheLoggerHop()
|
||||
{
|
||||
StartBridge();
|
||||
var actor = Sys.ActorOf(Props.Create(() => new LoggingActor()));
|
||||
var id = Guid.NewGuid();
|
||||
|
||||
actor.Tell(((Guid?)id, "beta"));
|
||||
ExpectMsg("done");
|
||||
|
||||
Assert.Equal(id.ToString(), Scalar(AwaitLine("beta"), ExecutionLogContext.ExecutionIdProperty));
|
||||
}
|
||||
|
||||
/// <summary>The Akka timestamp is kept, not replaced by the time the logger actor wrote the line.</summary>
|
||||
[Fact]
|
||||
public void Write_KeepsTheAkkaTimestamp_AndMapsEveryLevel()
|
||||
{
|
||||
var serilog = new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(_sink).CreateLogger();
|
||||
var then = new DateTime(2026, 1, 2, 3, 4, 5, DateTimeKind.Utc);
|
||||
|
||||
foreach (var (evt, expected) in new (AkkaLogEvent, LogEventLevel)[]
|
||||
{
|
||||
(new Debug("src", typeof(LoggingActor), "d-line"), LogEventLevel.Debug),
|
||||
(new Info("src", typeof(LoggingActor), "i-line"), LogEventLevel.Information),
|
||||
(new Warning("src", typeof(LoggingActor), "w-line"), LogEventLevel.Warning),
|
||||
(new Error(new InvalidOperationException("boom"), "src", typeof(LoggingActor), "e-line"), LogEventLevel.Error),
|
||||
})
|
||||
{
|
||||
typeof(AkkaLogEvent).GetProperty(nameof(AkkaLogEvent.Timestamp))!.SetValue(evt, then);
|
||||
SerilogAkkaLogger.Write(evt, serilog);
|
||||
var line = _sink.Events.Single(e => e.RenderMessage().Contains(evt.Message.ToString()!, StringComparison.Ordinal));
|
||||
Assert.Equal(expected, line.Level);
|
||||
Assert.Equal(new DateTimeOffset(then), line.Timestamp);
|
||||
}
|
||||
|
||||
Assert.Equal("boom", _sink.Events.Single(e => e.Level == LogEventLevel.Error).Exception!.Message);
|
||||
}
|
||||
|
||||
/// <summary>The Serilog minimum level applies to Akka lines as it does to every other line.</summary>
|
||||
[Fact]
|
||||
public void Write_BelowTheSerilogMinimum_IsDropped()
|
||||
{
|
||||
// _serilog's minimum is Information.
|
||||
SerilogAkkaLogger.Write(new Debug("src", typeof(LoggingActor), "debug-only"), _serilog);
|
||||
SerilogAkkaLogger.Write(new Info("src", typeof(LoggingActor), "info-kept"), _serilog);
|
||||
|
||||
Assert.DoesNotContain(_sink.Events, e => e.RenderMessage().Contains("debug-only", StringComparison.Ordinal));
|
||||
Assert.Contains(_sink.Events, e => e.RenderMessage().Contains("info-kept", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The HOCON the host builds names this logger and keeps <c>akka.loglevel</c> at INFO, so Akka's
|
||||
/// Debug lifecycle chatter stays filtered at the adapter (no volume change from the bridge).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void BuildHocon_NamesTheSerilogLogger_AndKeepsInfoLevel()
|
||||
{
|
||||
var hocon = AkkaHostedService.BuildHocon(
|
||||
new NodeOptions { Role = "Site", NodeHostname = "n", RemotingPort = 8082 },
|
||||
new ClusterOptions { SeedNodes = ["akka.tcp://scadabridge@n:8082"], SplitBrainResolverStrategy = "auto-down" },
|
||||
["Site"], TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(15));
|
||||
|
||||
var config = ConfigurationFactory.ParseString(hocon);
|
||||
Assert.Equal([SerilogAkkaLogger.HoconTypeName], config.GetStringList("akka.loggers"));
|
||||
Assert.Equal("INFO", config.GetString("akka.loglevel"));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The type name in <c>akka.loggers</c> resolves and the logger completes Akka's startup
|
||||
/// handshake — a wrong name or a missing <c>LoggerInitialized</c> reply fails system creation.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task ActorSystem_StartsWithTheSerilogLoggerConfigured()
|
||||
{
|
||||
var config = ConfigurationFactory.ParseString(
|
||||
$"akka.loggers = [\"{SerilogAkkaLogger.HoconTypeName}\"]\nakka.logger-startup-timeout = 5s");
|
||||
var system = ActorSystem.Create("serilog-logger-boot", config);
|
||||
try
|
||||
{
|
||||
system.Log.Info("boot line"); // must not throw; goes to the static Serilog logger
|
||||
Assert.False(system.WhenTerminated.IsCompleted);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await system.Terminate();
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class CollectingTarget : IExecutionLogTarget
|
||||
{
|
||||
public ConcurrentQueue<CapturedLogLine> Lines { get; } = new();
|
||||
public void Capture(CapturedLogLine line) => Lines.Enqueue(line);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#56 + #120 end to end: an Akka actor's own line, logged while it serves a traced
|
||||
/// execution, crosses the logger-actor hop and still reaches that execution's step log, with the
|
||||
/// actor's class as its category.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void AkkaLine_ForATracedExecution_ReachesItsStepLog()
|
||||
{
|
||||
var serilog = new LoggerConfiguration()
|
||||
.Enrich.FromLogContext()
|
||||
.WriteTo.Sink(new ExecutionLogCaptureSink())
|
||||
.CreateLogger();
|
||||
var bridge = Sys.ActorOf(Props.Create(() => new SerilogAkkaLogger(serilog)));
|
||||
Sys.EventStream.Subscribe(bridge, typeof(AkkaLogEvent));
|
||||
var id = Guid.NewGuid();
|
||||
var target = new CollectingTarget();
|
||||
using var registration = ExecutionLogCapture.Register(id, target);
|
||||
|
||||
Sys.ActorOf(Props.Create(() => new LoggingActor())).Tell(((Guid?)id, "gamma"));
|
||||
ExpectMsg("done");
|
||||
|
||||
AwaitCondition(() => !target.Lines.IsEmpty, TimeSpan.FromSeconds(5));
|
||||
var line = Assert.Single(target.Lines);
|
||||
Assert.Equal("warn", line.Level);
|
||||
Assert.Equal(typeof(LoggingActor).FullName, line.Category);
|
||||
Assert.Equal("actor line \"gamma\"", line.Message);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
using NSubstitute;
|
||||
using Serilog;
|
||||
using ZB.MOM.WW.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#56 (central half): while a method request is traced, other central node-log lines
|
||||
/// tagged with its <c>ExecutionId</c> become <c>node</c> rows of the request; the script's own
|
||||
/// <c>Log.*</c> line and the executor's outcome line are not copied, and an untraced request copies
|
||||
/// nothing.
|
||||
/// </summary>
|
||||
public class InboundNodeLogCaptureTests
|
||||
{
|
||||
private readonly Serilog.ILogger _node = new LoggerConfiguration()
|
||||
.MinimumLevel.Information()
|
||||
.Enrich.FromLogContext()
|
||||
.WriteTo.Sink(new ExecutionLogCaptureSink())
|
||||
.CreateLogger();
|
||||
|
||||
private readonly InboundScriptExecutor _executor;
|
||||
private readonly ApiMethod _method = new("node", "x") { Id = 1, TimeoutSeconds = 10 };
|
||||
|
||||
/// <summary>Builds an executor whose own logger writes to the node pipeline, like the host's.</summary>
|
||||
public InboundNodeLogCaptureTests()
|
||||
=> _executor = new InboundScriptExecutor(new SerilogMel(_node), Substitute.For<IServiceProvider>());
|
||||
|
||||
private sealed class SerilogMel(Serilog.ILogger inner) : ILogger<InboundScriptExecutor>
|
||||
{
|
||||
private readonly Serilog.ILogger _inner = inner.ForContext("SourceContext", typeof(InboundScriptExecutor).FullName);
|
||||
|
||||
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||
|
||||
public bool IsEnabled(LogLevel logLevel) => logLevel >= LogLevel.Information;
|
||||
|
||||
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
|
||||
Func<TState, Exception?, string> formatter)
|
||||
{
|
||||
if (!IsEnabled(logLevel)) return;
|
||||
var level = logLevel switch
|
||||
{
|
||||
LogLevel.Information => Serilog.Events.LogEventLevel.Information,
|
||||
LogLevel.Warning => Serilog.Events.LogEventLevel.Warning,
|
||||
_ => Serilog.Events.LogEventLevel.Error,
|
||||
};
|
||||
_inner.Write(level, exception, "{Message:l}", formatter(state, exception));
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<List<(AuditDetails Details, ScriptTraceExtra Extra)>> RunAsync(
|
||||
Func<InboundScriptContext, Task<object?>> handler, bool trace)
|
||||
{
|
||||
_executor.RegisterHandler("node", handler);
|
||||
var sink = new List<AuditEvent>();
|
||||
var route = new RouteHelper(Substitute.For<IInstanceLocator>(), Substitute.For<IInstanceRouter>())
|
||||
.WithTraceSink(e => { lock (sink) sink.Add(e); });
|
||||
await _executor.ExecuteAsync(_method, new Dictionary<string, object?>(), route, TimeSpan.FromSeconds(10),
|
||||
parentExecutionId: Guid.NewGuid(), trace: trace);
|
||||
lock (sink)
|
||||
{
|
||||
return sink
|
||||
.Select(e => AuditDetailsCodec.Deserialize(e.DetailsJson))
|
||||
.Where(d => d.Kind == nameof(AuditKind.ScriptLog))
|
||||
.Select(d => (d, ScriptTraceExtra.TryParse(d.Extra)!))
|
||||
.ToList();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A callee's line on the request's flow becomes a node row with its category.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task TracedRequest_CalleeLine_IsANodeRow()
|
||||
{
|
||||
var rows = await RunAsync(async _ =>
|
||||
{
|
||||
await Task.Yield();
|
||||
_node.ForContext("SourceContext", "Acme.InboundDatabaseHelper").Warning("query retried {Count}", 2);
|
||||
return null;
|
||||
}, trace: true);
|
||||
|
||||
var row = Assert.Single(rows);
|
||||
Assert.Equal("node", row.Extra.LogSource);
|
||||
Assert.Equal("warn", row.Extra.LogLevel);
|
||||
Assert.Equal("Acme.InboundDatabaseHelper", row.Extra.LogCategory);
|
||||
Assert.Equal("query retried 2", row.Details.RequestSummary);
|
||||
}
|
||||
|
||||
/// <summary>An untraced request registers nothing, so the same line is not captured.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task UntracedRequest_CapturesNothing()
|
||||
{
|
||||
var rows = await RunAsync(_ =>
|
||||
{
|
||||
_node.Warning("query retried");
|
||||
return Task.FromResult<object?>(null);
|
||||
}, trace: false);
|
||||
|
||||
Assert.Empty(rows);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The script's own Log line is one script row (its node forward is suppressed), and the
|
||||
/// executor's failure line is the inbound row's error, so it is not copied.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ScriptLine_AndExecutorFailureLine_AreNotCopied()
|
||||
{
|
||||
var rows = await RunAsync(ctx =>
|
||||
{
|
||||
ctx.Log.Info("from the script");
|
||||
throw new InvalidOperationException("boom");
|
||||
}, trace: true);
|
||||
|
||||
var row = Assert.Single(rows);
|
||||
Assert.Equal("script", row.Extra.LogSource);
|
||||
Assert.Equal("from the script", row.Details.RequestSummary);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,312 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.CodeAnalysis.CSharp.Scripting;
|
||||
using Microsoft.CodeAnalysis.Scripting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Scripts;
|
||||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Scripts;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#56: while a run is traced, node-log lines that OTHER actors write for it (tagged with
|
||||
/// its <c>ExecutionId</c>) are copied into its <c>ScriptLog</c> rows as source <c>node</c> — under the
|
||||
/// run's line cap, through the same collector (and so the same redaction) as every other row — and
|
||||
/// an untraced run copies nothing.
|
||||
/// </summary>
|
||||
public class NodeLogCaptureTests : TestKit, IDisposable
|
||||
{
|
||||
private readonly ScriptExecutionScheduler _scheduler = new(4);
|
||||
private readonly SharedScriptLibrary _sharedLibrary = new(
|
||||
new ScriptCompilationService(NullLogger<ScriptCompilationService>.Instance),
|
||||
NullLogger<SharedScriptLibrary>.Instance);
|
||||
|
||||
/// <summary>The node's Serilog pipeline as the host builds it: LogContext enrichment plus the capture sink.</summary>
|
||||
private readonly Serilog.ILogger _node = new Serilog.LoggerConfiguration()
|
||||
.MinimumLevel.Information()
|
||||
.Enrich.FromLogContext()
|
||||
.WriteTo.Sink(new ExecutionLogCaptureSink())
|
||||
.CreateLogger();
|
||||
|
||||
/// <inheritdoc />
|
||||
void IDisposable.Dispose()
|
||||
{
|
||||
Shutdown();
|
||||
_scheduler.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>A Microsoft logger over a Serilog logger (the host's <c>UseSerilog</c> shape), with a category.</summary>
|
||||
private sealed class SerilogMel(Serilog.ILogger inner, string category) : ILogger
|
||||
{
|
||||
private readonly Serilog.ILogger _inner = inner.ForContext("SourceContext", category);
|
||||
|
||||
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||
|
||||
public bool IsEnabled(LogLevel logLevel) => logLevel >= LogLevel.Information;
|
||||
|
||||
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
|
||||
Func<TState, Exception?, string> formatter)
|
||||
{
|
||||
if (!IsEnabled(logLevel)) return;
|
||||
var level = logLevel switch
|
||||
{
|
||||
LogLevel.Information => Serilog.Events.LogEventLevel.Information,
|
||||
LogLevel.Warning => Serilog.Events.LogEventLevel.Warning,
|
||||
_ => Serilog.Events.LogEventLevel.Error,
|
||||
};
|
||||
_inner.Write(level, exception, "{Message:l}", formatter(state, exception));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stands in for the Instance Actor: re-pushes the message's execution id (as the real one does in
|
||||
/// AroundReceive), logs a line on ITS thread, and acknowledges the write.
|
||||
/// </summary>
|
||||
private sealed class LoggingInstanceActor : ReceiveActor
|
||||
{
|
||||
public LoggingInstanceActor(Serilog.ILogger node)
|
||||
{
|
||||
var log = node.ForContext("SourceContext", "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.InstanceActor");
|
||||
Receive<SetStaticAttributeCommand>(cmd =>
|
||||
{
|
||||
using (ExecutionLogContext.PushFor(cmd))
|
||||
{
|
||||
log.Warning("instance actor applied {Attribute} token=hunter2", cmd.AttributeName);
|
||||
}
|
||||
|
||||
Sender.Tell(new SetStaticAttributeResponse(
|
||||
cmd.CorrelationId, cmd.InstanceUniqueName, cmd.AttributeName, true, null, DateTimeOffset.UtcNow));
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private static Script<object?> Compile(string code)
|
||||
{
|
||||
var options = ScriptOptions.Default
|
||||
.WithReferences(typeof(object).Assembly, typeof(Enumerable).Assembly, typeof(ScriptGlobals).Assembly)
|
||||
.WithImports("System", "System.Collections.Generic", "System.Linq", "System.Threading.Tasks");
|
||||
var script = CSharpScript.Create<object?>(code, options, typeof(ScriptGlobals));
|
||||
script.Compile();
|
||||
return script;
|
||||
}
|
||||
|
||||
private ScriptCallResult Run(string body, bool trace, int maxLogLines = 200)
|
||||
{
|
||||
var caller = CreateTestProbe();
|
||||
var instance = Sys.ActorOf(Props.Create(() => new LoggingInstanceActor(_node)));
|
||||
ScriptRunLauncher.LaunchScript(
|
||||
scriptName: "Runner", instanceName: "Inst1", compiledScript: Compile(body),
|
||||
parameters: null, callDepth: 0, instanceActor: instance,
|
||||
sharedScriptLibrary: _sharedLibrary,
|
||||
options: new SiteRuntimeOptions
|
||||
{
|
||||
MaxScriptCallDepth = 10, ScriptExecutionTimeoutSeconds = 30, StuckScriptGraceMs = 30000,
|
||||
MaxLogLinesPerRun = maxLogLines,
|
||||
},
|
||||
replyTo: caller.Ref, correlationId: "corr-1", completionTarget: CreateTestProbe().Ref,
|
||||
siteCommunicationActor: null,
|
||||
logger: new SerilogMel(_node, "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.ScriptActor"),
|
||||
scope: ScriptScope.Root, runId: 1, healthCollector: null, serviceProvider: null,
|
||||
parentExecutionId: null, executionTimeoutSeconds: null, scheduler: _scheduler,
|
||||
trace: trace, triggerExtra: null);
|
||||
return caller.ExpectMsg<ScriptCallResult>(TimeSpan.FromSeconds(20));
|
||||
}
|
||||
|
||||
private static List<(AuditEvent Event, ScriptTraceExtra Extra)> LogRows(ScriptCallResult result)
|
||||
=> result.TraceEvents!
|
||||
.Where(e => e.AsRow().Kind == AuditKind.ScriptLog)
|
||||
.Select(e => (e, ScriptTraceExtra.TryParse(e.AsRow().Extra)!))
|
||||
.ToList();
|
||||
|
||||
/// <summary>
|
||||
/// Cross-actor: the Instance Actor's line, written on its own thread while serving the traced
|
||||
/// run's write, comes back inline as a <c>node</c> row of that run, with its category and level.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void TracedRun_OtherActorsLine_IsCapturedAsANodeRow()
|
||||
{
|
||||
var result = Run("await Instance.SetAttribute(\"Setpoint\", \"42\"); return null;", trace: true);
|
||||
|
||||
Assert.True(result.Success, result.ErrorMessage);
|
||||
var node = Assert.Single(LogRows(result), r => r.Extra.LogSource == "node");
|
||||
Assert.Equal("warn", node.Extra.LogLevel);
|
||||
Assert.Equal("ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.InstanceActor", node.Extra.LogCategory);
|
||||
Assert.Contains("instance actor applied \"Setpoint\"", node.Event.AsRow().RequestSummary);
|
||||
Assert.Equal(result.ExecutionId, node.Event.AsRow().ExecutionId);
|
||||
// The registration ends with the run: a later line for the same id is not captured.
|
||||
Assert.False(ExecutionLogCapture.TryGetTarget(result.ExecutionId!.Value, out _));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An inline shared script's write is stamped with the SHARED script's own execution id; its
|
||||
/// node line still lands in the caller's rows (where the shared script's Log.* lines go).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void SharedScriptsWrite_NodeLine_LandsInTheCallersRows()
|
||||
{
|
||||
Assert.True(_sharedLibrary.CompileAndRegister(
|
||||
"WriteHelper", "await Instance.SetAttribute(\"FromShared\", \"1\"); return null;"));
|
||||
|
||||
var result = Run("await Instance.Scripts.CallShared(\"WriteHelper\"); return null;", trace: true);
|
||||
|
||||
Assert.True(result.Success, result.ErrorMessage);
|
||||
var node = Assert.Single(LogRows(result), r => r.Extra.LogSource == "node");
|
||||
Assert.Contains("FromShared", node.Event.AsRow().RequestSummary);
|
||||
Assert.Equal(result.ExecutionId, node.Event.AsRow().ExecutionId);
|
||||
}
|
||||
|
||||
/// <summary>An untraced run never registers, so the same line is not captured (and the reply has no rows).</summary>
|
||||
[Fact]
|
||||
public void UntracedRun_CapturesNothing()
|
||||
{
|
||||
var result = Run("await Instance.SetAttribute(\"Setpoint\", \"42\"); return null;", trace: false);
|
||||
|
||||
Assert.True(result.Success, result.ErrorMessage);
|
||||
Assert.True(result.TraceEvents is null || result.TraceEvents.Count == 0);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// No double capture: the script's own <c>Log.Info</c> goes to the node logger (tagged with the run's
|
||||
/// id) and is a <c>script</c> row exactly once — never a second <c>node</c> row. The launcher's
|
||||
/// own failure line is the <c>ScriptRun</c> row, so it is not copied either.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ScriptLogLine_AndLauncherFailureLine_AreNotCopiedAsNodeRows()
|
||||
{
|
||||
var result = Run("Log.Info(\"hello\"); throw new InvalidOperationException(\"boom\");", trace: true);
|
||||
|
||||
Assert.False(result.Success);
|
||||
var rows = LogRows(result);
|
||||
var script = Assert.Single(rows);
|
||||
Assert.Equal("script", script.Extra.LogSource);
|
||||
Assert.Equal("hello", script.Event.AsRow().RequestSummary);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Node rows share the run's line cap with the script's own lines: past it they are dropped and
|
||||
/// the closing runtime row names the count.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void NodeRows_ShareTheRunsLineCap()
|
||||
{
|
||||
var result = Run(
|
||||
"Log.Info(\"one\"); for (var i = 0; i < 3; i++) await Instance.SetAttribute(\"A\" + i, \"1\"); return null;",
|
||||
trace: true, maxLogLines: 2);
|
||||
|
||||
Assert.True(result.Success, result.ErrorMessage);
|
||||
var rows = LogRows(result);
|
||||
Assert.Equal(3, rows.Count); // "one" + 1 node row + the closing row
|
||||
Assert.Single(rows, r => r.Extra.LogSource == "node");
|
||||
var closing = Assert.Single(rows, r => r.Extra.LogSource == "runtime");
|
||||
Assert.Contains("2 further log line(s)", closing.Event.AsRow().RequestSummary);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Node rows ride the same inline collector as every other row, so its redactor applies: what Live
|
||||
/// shows is what Replay stores.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task NodeRows_PassTheCollectorsRedactor()
|
||||
{
|
||||
var collector = new FlowTraceCollector(redactor: new TokenRedactor());
|
||||
var exec = Guid.NewGuid();
|
||||
var ctx = new ScriptRuntimeContext(
|
||||
Sys.ActorOf(Props.Create(() => new LoggingInstanceActor(_node))), _sharedLibrary,
|
||||
currentCallDepth: 0, maxCallDepth: 10, askTimeout: TimeSpan.FromSeconds(5),
|
||||
instanceName: "Inst1", logger: NullLogger.Instance, siteId: "site-a",
|
||||
sourceScript: "ScriptActor:Runner", executionId: exec, isTraced: true, traceCollector: collector);
|
||||
|
||||
using (ctx.Log.CaptureNodeLines(exec))
|
||||
{
|
||||
await ctx.SetAttribute("Setpoint", "42");
|
||||
}
|
||||
|
||||
var node = Assert.Single(collector.Complete(), e => e.AsRow().Kind == AuditKind.ScriptLog);
|
||||
Assert.DoesNotContain("hunter2", node.DetailsJson);
|
||||
Assert.Contains("token=***", node.AsRow().RequestSummary);
|
||||
}
|
||||
|
||||
/// <summary>Scrubs <c>hunter2</c> from the details, standing in for the host's body-regex redaction.</summary>
|
||||
private sealed class TokenRedactor : IAuditRedactor
|
||||
{
|
||||
public AuditEvent Apply(AuditEvent rawEvent)
|
||||
=> rawEvent with { DetailsJson = rawEvent.DetailsJson?.Replace("hunter2", "***", StringComparison.Ordinal) };
|
||||
}
|
||||
|
||||
private (ScriptLogger Logger, List<(string Level, string Source, string Message, string? Category)> Rows) Traced(Guid id)
|
||||
{
|
||||
var rows = new List<(string, string, string, string?)>();
|
||||
var logger = new ScriptLogger(
|
||||
new SerilogMel(_node, "ScriptActor"),
|
||||
(level, source, message, _, _, category, _) => { lock (rows) rows.Add((level, source, message, category)); },
|
||||
maxLines: 200, "Inst1", "Runner", id);
|
||||
return (logger, rows);
|
||||
}
|
||||
|
||||
/// <summary>An untraced logger never touches the registry.</summary>
|
||||
[Fact]
|
||||
public void UntracedLogger_CaptureNodeLines_DoesNotRegister()
|
||||
{
|
||||
var id = Guid.NewGuid();
|
||||
var untraced = new ScriptLogger(NullLogger.Instance, emit: null, maxLines: 200, "Inst1", "Runner", id);
|
||||
|
||||
using (untraced.CaptureNodeLines(id))
|
||||
{
|
||||
Assert.False(ExecutionLogCapture.TryGetTarget(id, out _));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>FlushDropped (every run-end path) closes the copy, and a capture opened after it is refused.</summary>
|
||||
[Fact]
|
||||
public void FlushDropped_ClosesTheCopy()
|
||||
{
|
||||
var id = Guid.NewGuid();
|
||||
var (logger, rows) = Traced(id);
|
||||
var capture = logger.CaptureNodeLines(id);
|
||||
Assert.True(ExecutionLogCapture.TryGetTarget(id, out _));
|
||||
|
||||
logger.FlushDropped();
|
||||
|
||||
Assert.False(ExecutionLogCapture.TryGetTarget(id, out _));
|
||||
using (logger.CaptureNodeLines(id))
|
||||
using (ExecutionLogContext.Push(id))
|
||||
{
|
||||
_node.Warning("after the run");
|
||||
}
|
||||
|
||||
Assert.Empty(rows);
|
||||
capture.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The runtime decorator's forward is suppressed: a runtime warning logged while traced is one
|
||||
/// <c>runtime</c> row, never a second <c>node</c> row — while a genuine other-actor line still is one.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void RuntimeLine_IsNotCopiedAgain_ButAnotherLoggersLineIs()
|
||||
{
|
||||
var id = Guid.NewGuid();
|
||||
var (logger, rows) = Traced(id);
|
||||
var runtime = new TraceCapturingLogger(new SerilogMel(_node, "ScriptActor"), logger);
|
||||
|
||||
using (logger.CaptureNodeLines(id))
|
||||
using (ExecutionLogContext.Push(id))
|
||||
{
|
||||
runtime.LogWarning("runtime warning");
|
||||
_node.ForContext("SourceContext", "Acme.Other").Warning("other actor");
|
||||
}
|
||||
|
||||
Assert.Equal(
|
||||
[("warn", "runtime", "runtime warning", (string?)null), ("warn", "node", "other actor", "Acme.Other")],
|
||||
rows);
|
||||
}
|
||||
}
|
||||
@@ -54,7 +54,7 @@ public class ScriptLoggerTests
|
||||
var rows = new List<Row>();
|
||||
var logger = new ScriptLogger(
|
||||
node,
|
||||
(level, source, message, exception, stepEventId) => rows.Add(new Row(level, source, message, exception, stepEventId)),
|
||||
(level, source, message, exception, stepEventId, _, _) => rows.Add(new Row(level, source, message, exception, stepEventId)),
|
||||
maxLines,
|
||||
instance: "CvdReactor01",
|
||||
script: "OnHeartbeat",
|
||||
@@ -252,7 +252,7 @@ public class ScriptLoggerTests
|
||||
var node = new RecordingLogger();
|
||||
var logger = new ScriptLogger(
|
||||
node,
|
||||
(_, _, _, _, _) => throw new InvalidOperationException("collector is broken"),
|
||||
(_, _, _, _, _, _, _) => throw new InvalidOperationException("collector is broken"),
|
||||
maxLines: 200,
|
||||
"CvdReactor01",
|
||||
"OnHeartbeat",
|
||||
@@ -271,7 +271,7 @@ public class ScriptLoggerTests
|
||||
var rows = new List<Row>();
|
||||
var logger = new ScriptLogger(
|
||||
new ThrowingLogger(),
|
||||
(level, source, message, exception, stepEventId) => rows.Add(new Row(level, source, message, exception, stepEventId)),
|
||||
(level, source, message, exception, stepEventId, _, _) => rows.Add(new Row(level, source, message, exception, stepEventId)),
|
||||
maxLines: 200,
|
||||
"CvdReactor01",
|
||||
"OnHeartbeat",
|
||||
|
||||
@@ -55,7 +55,7 @@ public class TraceCapturingLoggerTests
|
||||
var rows = new List<Row>();
|
||||
var sink = new ScriptLogger(
|
||||
new RecordingLogger(),
|
||||
(level, source, message, exception, stepEventId) => rows.Add(new Row(level, source, message, exception, stepEventId)),
|
||||
(level, source, message, exception, stepEventId, _, _) => rows.Add(new Row(level, source, message, exception, stepEventId)),
|
||||
maxLines: 200,
|
||||
instance: "CvdReactor01",
|
||||
script: "OnHeartbeat",
|
||||
|
||||
Reference in New Issue
Block a user