feat(logging): carry ExecutionId across actor hops onto other actors' log lines (ScadaBridge#56)

LogContext is an AsyncLocal and does not cross an Akka message hop, so the run's id now rides the message: Commons.Observability.IExecutionCorrelated (explicit interface member, no wire change) on SetStaticAttributeCommand, AttributeValueChanged (not for an inferred writer), ScriptCallRequest and the four RouteTo*Requests, plus a new additive SourceExecutionId on GetAttributeRequest, WaitForAttributeRequest, WriteAttributeBatchRequest, WriteTagRequest and WriteTagBatchRequest. ScriptRuntimeContext stamps its id on the requests it sends, the Instance Actor hands it on to the DCL, and the Instance, Script, Alarm, Deployment Manager and both DCL actors re-push it per message in AroundReceive (ExecutionLogContext.PushFor: one type test when the message names no execution). The inbound executor pushes the request's ExecutionId for the method execution. Tests (TestKit): Instance Actor line carries the message's id, no leak from the sender's ambient context, scope ends with the message, per-write ids inside a batched SetAttributes, ids forwarded on DCL writes, ScriptRuntimeContext stamps; DCL adapter lines on single/batch writes carry the id; inbound executor lines carry the request id; Commons mapping + no serialized member. Each mutation- proved. Still open on #56: a queryable per-execution sink and a Flow Debugger section. The DCL actors' own Akka _log lines bypass Serilog (filed #120). Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-10 05:28:20 -05:00
parent 1ccd6cfe33
commit edd0ca672f
28 changed files with 905 additions and 33 deletions
@@ -0,0 +1,122 @@
+using System.Collections.Concurrent;
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using NSubstitute;
+using Serilog;
+using Serilog.Core;
+using Serilog.Events;
+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.HealthMonitoring;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
+
+/// <summary>
+/// ScadaBridge#56: a device write the DCL performs for an execution runs the protocol adapter
+/// inside that execution's log scope, so whatever the adapter logs while writing carries the
+/// execution's id - though the write crossed two actor hops (script, Instance Actor, DCL).
+/// </summary>
+public class ExecutionLogEnrichmentTests : TestKit
+{
+    private sealed class CaptureSink : ILogEventSink
+    {
+        public ConcurrentQueue<LogEvent> Events { get; } = new();
+        public void Emit(LogEvent logEvent) => Events.Enqueue(logEvent);
+    }
+
+    private readonly CaptureSink _sink = new();
+    private readonly Serilog.ILogger _adapterLog;
+    private readonly IDataConnection _adapter = Substitute.For<IDataConnection>();
+
+    /// <summary>Creates the adapter double; its writes log through a LogContext-enriched Serilog logger.</summary>
+    public ExecutionLogEnrichmentTests()
+    {
+        _adapterLog = new LoggerConfiguration().Enrich.FromLogContext().WriteTo.Sink(_sink).CreateLogger();
+        _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(ci =>
+            {
+                _adapterLog.Information("adapter wrote {Tag}", ci.ArgAt<string>(0));
+                return Task.FromResult(new WriteResult(true, null));
+            });
+        _adapter.WriteBatchAsync(Arg.Any<IDictionary<string, object?>>(), Arg.Any<CancellationToken>())
+            .Returns(ci =>
+            {
+                _adapterLog.Information("adapter batch-wrote");
+                return Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(
+                    ci.ArgAt<IDictionary<string, object?>>(0).ToDictionary(kv => kv.Key, _ => new WriteResult(true, null)));
+            });
+    }
+
+    private async Task<IActorRef> ConnectedActor(string name)
+    {
+        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(
+            name, _adapter, options, Substitute.For<ISiteHealthCollector>(),
+            Substitute.For<IDataConnectionFactory>(), "OpcUa")), name);
+        AwaitCondition(() => _adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "ConnectAsync"),
+            TimeSpan.FromSeconds(3));
+        await Task.Delay(200);
+        return actor;
+    }
+
+    private Guid? ExecutionIdOf(string fragment)
+    {
+        var evt = _sink.Events.First(e => e.RenderMessage().Contains(fragment, StringComparison.Ordinal));
+        return evt.Properties.TryGetValue(ExecutionLogContext.ExecutionIdProperty, out var v)
+            ? Guid.Parse(v.ToString().Trim('"'))
+            : null;
+    }
+
+    /// <summary>The adapter's line for a single write carries the execution id the request names.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task SingleWrite_AdapterLinesCarryTheExecutionId()
+    {
+        var actor = await ConnectedActor("enrich-single");
+        var executionId = Guid.NewGuid();
+
+        actor.Tell(new WriteTagRequest("c1", "enrich-single", "tagA", 1, DateTimeOffset.UtcNow, executionId));
+        Assert.True(ExpectMsg<WriteTagResponse>(TimeSpan.FromSeconds(5)).Success);
+
+        Assert.Equal(executionId, ExecutionIdOf("tagA"));
+    }
+
+    /// <summary>The adapter's line for a batch write carries the execution id the request names.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task BatchWrite_AdapterLinesCarryTheExecutionId()
+    {
+        var actor = await ConnectedActor("enrich-batch");
+        var executionId = Guid.NewGuid();
+
+        actor.Tell(new WriteTagBatchRequest("c1", "enrich-batch",
+            new Dictionary<string, object?> { ["tagB"] = 2 }, null, null, DateTimeOffset.UtcNow, executionId));
+        Assert.True(ExpectMsg<WriteTagBatchResponse>(TimeSpan.FromSeconds(5)).Success);
+
+        Assert.Equal(executionId, ExecutionIdOf("batch-wrote"));
+    }
+
+    /// <summary>A write that names no execution logs untagged (control for the two tests above).</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task UncorrelatedWrite_AdapterLinesAreUntagged()
+    {
+        var actor = await ConnectedActor("enrich-none");
+
+        actor.Tell(new WriteTagRequest("c1", "enrich-none", "tagC", 1, DateTimeOffset.UtcNow));
+        Assert.True(ExpectMsg<WriteTagResponse>(TimeSpan.FromSeconds(5)).Success);
+
+        Assert.Null(ExecutionIdOf("tagC"));
+    }
+}