lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Observability/OtOpcUaTelemetryHookTests.cs

using System.Diagnostics;
using System.Diagnostics.Metrics;
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Observability;

/// <summary>
/// F13d — verifies the instrumentation sites actually emit on the central
/// <see cref="OtOpcUaTelemetry"/> meter + activity source. Each test attaches a one-shot
/// listener, exercises the instrumented path, then asserts the recorded measurement matches.
/// </summary>
public sealed class OtOpcUaTelemetryHookTests : RuntimeActorTestBase
{
    /// <summary>Verifies that VirtualTagActor evaluation emits the otopcua_virtualtag_eval counter.</summary>
    [Fact]
    public void VirtualTagActor_evaluation_emits_otopcua_virtualtag_eval_counter()
    {
        using var recorder = new MeterRecorder("otopcua.virtualtag.eval");
        var parent = CreateTestProbe();
        var evaluator = new ConstEval(42);

        var actor = parent.ChildActorOf(VirtualTagActor.Props("vt-tel-1", "expr", evaluator: evaluator));
        actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 1, DateTime.UtcNow));
        parent.ExpectMsg<VirtualTagActor.EvaluationResult>();

        recorder.Total.ShouldBeGreaterThanOrEqualTo(1);
        recorder.WithTag("outcome", "ok").ShouldBeGreaterThanOrEqualTo(1);
    }

    /// <summary>Verifies that OpcUaPublishActor AttributeValueUpdate emits the sink write counter.</summary>
    [Fact]
    public void OpcUaPublishActor_AttributeValueUpdate_emits_sink_write_counter()
    {
        using var recorder = new MeterRecorder("otopcua.opcua.sink.write");
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
            sink: sink,
            serviceLevel: NullServiceLevelPublisher.Instance,
            subscribeRedundancyTopic: false,
            localNode: Commons.Types.NodeId.Parse("test-node")));

        actor.Tell(new OpcUaPublishActor.AttributeValueUpdate(
            NodeId: "ns=2;s=tag-1",
            Value: 42,
            Quality: OpcUaQuality.Good,
            TimestampUtc: DateTime.UtcNow));

        AwaitAssertion(() =>
        {
            recorder.Total.ShouldBeGreaterThanOrEqualTo(1);
            recorder.WithTag("kind", "value").ShouldBeGreaterThanOrEqualTo(1);
        });
    }

    /// <summary>Verifies that RebuildAddressSpace starts an address space rebuild span.</summary>
    [Fact]
    public void RebuildAddressSpace_starts_an_address_space_rebuild_span()
    {
        using var spanRecorder = new ActivityRecorder("otopcua.opcua.address_space_rebuild");
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
            sink: sink,
            serviceLevel: NullServiceLevelPublisher.Instance,
            subscribeRedundancyTopic: false,
            localNode: Commons.Types.NodeId.Parse("test-node")));

        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(Commons.Types.CorrelationId.NewId()));

        AwaitAssertion(() => spanRecorder.Activities.ShouldContain(a => a.OperationName == "otopcua.opcua.address_space_rebuild"));
    }

    private void AwaitAssertion(Action assertion)
    {
        var deadline = DateTime.UtcNow.AddSeconds(2);
        Exception? last = null;
        while (DateTime.UtcNow < deadline)
        {
            try { assertion(); return; }
            catch (Exception ex) { last = ex; Thread.Sleep(25); }
        }
        if (last is not null) throw last;
    }

    /// <summary>Listens to a single instrument by name and tallies the values + tags.</summary>
    private sealed class MeterRecorder : IDisposable
    {
        private readonly string _name;
        private readonly MeterListener _listener;
        private long _total;
        private readonly List<KeyValuePair<string, object?>[]> _tagSets = new();
        private readonly object _gate = new();

        /// <summary>Initializes the recorder for the specified instrument name.</summary>
        /// <param name="instrumentName">Name of the instrument to listen for.</param>
        public MeterRecorder(string instrumentName)
        {
            _name = instrumentName;
            _listener = new MeterListener
            {
                InstrumentPublished = (instrument, listener) =>
                {
                    if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == _name)
                        listener.EnableMeasurementEvents(instrument);
                }
            };
            _listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
            {
                lock (_gate)
                {
                    _total += value;
                    _tagSets.Add(tags.ToArray());
                }
            });
            _listener.Start();
        }

        /// <summary>Gets the total value recorded.</summary>
        public long Total { get { lock (_gate) return _total; } }

        /// <summary>Gets count of measurements with the specified tag key-value pair.</summary>
        /// <param name="key">Tag key.</param>
        /// <param name="value">Tag value.</param>
        public int WithTag(string key, string value)
        {
            lock (_gate)
            {
                return _tagSets.Count(set => set.Any(t => t.Key == key && Equals(t.Value, value)));
            }
        }

    /// <summary>Releases the listener.</summary>
    public void Dispose() => _listener.Dispose();
    }

    /// <summary>Listens to a single ActivitySource by name and stores started Activities.</summary>
    private sealed class ActivityRecorder : IDisposable
    {
        private readonly string _operationName;
        private readonly ActivityListener _listener;
        private readonly List<Activity> _activities = new();
        private readonly object _gate = new();

        /// <summary>Initializes the recorder for the specified operation name.</summary>
        /// <param name="operationName">Name of the operation to listen for.</param>
        public ActivityRecorder(string operationName)
        {
            _operationName = operationName;
            _listener = new ActivityListener
            {
                ShouldListenTo = source => source.Name == OtOpcUaTelemetry.ActivitySourceName,
                Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllData,
                ActivityStarted = activity =>
                {
                    if (activity.OperationName == _operationName)
                    {
                        lock (_gate) _activities.Add(activity);
                    }
                }
            };
            ActivitySource.AddActivityListener(_listener);
        }

        /// <summary>Gets the list of recorded activities.</summary>
        public IReadOnlyList<Activity> Activities { get { lock (_gate) return _activities.ToArray(); } }

        /// <summary>Releases the listener.</summary>
        public void Dispose() => _listener.Dispose();
    }

    private sealed class ConstEval(object? value) : IVirtualTagEvaluator
    {
        /// <summary>Evaluates the virtual tag with a constant value.</summary>
        /// <param name="virtualTagId">Virtual tag ID.</param>
        /// <param name="expression">Expression to evaluate.</param>
        /// <param name="dependencies">Dependency values.</param>
        public VirtualTagEvalResult Evaluate(string virtualTagId, string expression, IReadOnlyDictionary<string, object?> dependencies)
            => VirtualTagEvalResult.Ok(value);
    }

    private sealed class RecordingSink : IOpcUaAddressSpaceSink
    {
        /// <summary>Gets the write count.</summary>
        public int Writes { get; private set; }
        /// <summary>Records a value write.</summary>
        /// <param name="nodeId">The OPC UA node identifier.</param>
        /// <param name="value">The value being written.</param>
        /// <param name="quality">The OPC UA quality status.</param>
        /// <param name="sourceTimestampUtc">The source timestamp in UTC.</param>
        public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) => Writes++;
        /// <summary>Records an alarm state write.</summary>
        /// <param name="alarmNodeId">The alarm node identifier.</param>
        /// <param name="active">Whether the alarm is active.</param>
        /// <param name="acknowledged">Whether the alarm is acknowledged.</param>
        /// <param name="occurredUtc">The time the alarm occurred in UTC.</param>
        public void WriteAlarmState(string alarmNodeId, bool active, bool acknowledged, DateTime occurredUtc) => Writes++;
        /// <summary>Ensures folder exists (stub implementation).</summary>
        /// <param name="folderNodeId">The folder node identifier.</param>
        /// <param name="parentNodeId">The parent folder node identifier.</param>
        /// <param name="displayName">The display name for the folder.</param>
        public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { }
        /// <summary>Ensures variable exists (stub implementation).</summary>
        /// <param name="variableNodeId">The variable node identifier.</param>
        /// <param name="parentFolderNodeId">The parent folder node identifier.</param>
        /// <param name="displayName">The display name for the variable.</param>
        /// <param name="dataType">The OPC UA built-in type name.</param>
        public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType) { }
        /// <summary>Rebuilds address space (recorded via span).</summary>
        public void RebuildAddressSpace() { /* recorded via span */ }
    }
}