Add periodic stats

src/Configuration/CliOptions.cs

@@ -77,6 +77,7 @@ public static (PlcSimulation PlcSimulationInstance, List<string> ExtraArgs) Init
             { "otlpee|otlpendpoint=", $"the endpoint URI to which the OTLP exporter is going to send information.\nDefault: '{config.OtlpEndpointUri}'", (s) => config.OtlpEndpointUri = s },
             { "otlpei|otlpexportinterval=", $"the interval for exporting OTLP information in seconds.\nDefault: {config.OtlpExportInterval.TotalSeconds}", (uint i) => config.OtlpExportInterval = TimeSpan.FromSeconds(i) },
             { "otlpep|otlpexportprotocol=", $"the protocol for exporting OTLP information.\n(allowed values: grpc, protobuf).\nDefault: {config.OtlpExportProtocol}", (string s) => config.OtlpExportProtocol = s },
+            { "otlpub|otlpublishmetrics=", $"how to handle metrics for publish requests.\n(allowed values: disable=Always disabled, enable=Always enabled, auto=Auto-disable when sessions > 40 or monitored items > 500).\nDefault: {config.OtlpPublishMetrics}", (string s) => config.OtlpPublishMetrics = s },
 
             { "lr|ldsreginterval=", $"the LDS(-ME) registration interval in ms. If 0, then the registration is disabled.\nDefault: {config.OpcUa.LdsRegistrationInterval}", (int i) => {
                     if (i >= 0)

src/Configuration/Configuration.cs

@@ -51,6 +51,15 @@ public class OpcPlcConfiguration
     /// </summary>
     public string OtlpExportProtocol { get; set; } = "grpc";
 
+    /// <summary>
+    /// Gets or sets how to handle metrics for publish requests.
+    /// Allowed values:
+    /// disable=Always disabled,
+    /// enable=Always enabled,
+    /// auto=Auto-disable when sessions > 40 or monitored items > 500.
+    /// </summary>
+    public string OtlpPublishMetrics { get; set; } = "auto";
+
     /// <summary>
     /// Show OPC Publisher configuration file using IP address as EndpointUrl.
     /// </summary>

src/Helpers/MetricsHelper.cs

@@ -153,32 +153,13 @@ public static void AddMonitoredItemCount(int delta = 1)
     /// <summary>
     /// Add a published count.
     /// </summary>
-    public static void AddPublishedCount(string sessionId, string subscriptionId, NotificationMessage notificationMessage, ILogger logger)
+    public static void AddPublishedCount(string sessionId, string subscriptionId, int dataChanges, int events)
     {
         if (!IsEnabled)
         {
             return;
         }
 
-        int events = 0;
-        int dataChanges = 0;
-        int diagnostics = 0;
-        notificationMessage.NotificationData.ForEach(x => {
-            if (x.Body is DataChangeNotification changeNotification)
-            {
-                dataChanges += changeNotification.MonitoredItems.Count;
-                diagnostics += changeNotification.DiagnosticInfos.Count;
-            }
-            else if (x.Body is EventNotificationList eventNotification)
-            {
-                events += eventNotification.Events.Count;
-            }
-            else
-            {
-                logger.LogDebug("Unknown notification type: {NotificationType}", x.Body.GetType().Name);
-            }
-        });
-
         if (dataChanges > 0)
         {
             var dataPointsDimensions = MergeWithBaseDimensions(

src/PlcServer.cs

@@ -23,6 +23,7 @@ namespace OpcPlc;
 public partial class PlcServer : StandardServer
 {
     private const uint PlcShutdownWaitSeconds = 10;
+    private const int PeriodicLoggingTimerSeconds = 60;
 
     public PlcNodeManager PlcNodeManager { get; set; }
 
@@ -41,7 +42,13 @@ public partial class PlcServer : StandardServer
     private readonly ILogger _logger;
     private readonly Timer _periodicLoggingTimer;
 
-    private bool _disablePublishMetrics;
+    private bool _autoDisablePublishMetrics;
+    private uint _countCreateSession;
+    private uint _countCreateSubscription;
+    private uint _countCreateMonitoredItems;
+    private uint _countPublish;
+    private uint _countRead;
+    private uint _countWrite;
 
     public PlcServer(OpcPlcConfiguration config, PlcSimulation plcSimulation, TimeService timeService, ImmutableList<IPluginNodes> pluginNodes, ILogger logger)
     {
@@ -63,27 +70,56 @@ public PlcServer(OpcPlcConfiguration config, PlcSimulation plcSimulation, TimeSe
                     IList<Subscription> subscriptions = ServerInternal.SubscriptionManager.GetSubscriptions();
                     int monitoredItemsCount = subscriptions.Sum(s => s.MonitoredItemCount);
 
+                    _autoDisablePublishMetrics = sessionCount > 40 || monitoredItemsCount > 500;
+
                     LogPeriodicInfo(
                         sessionCount,
                         subscriptions.Count,
-                        monitoredItemsCount,
+                        monitoredItemsCount,
                         curProc.WorkingSet64 / 1024 / 1024,
                         availWorkerThreads,
                         availCompletionPortThreads,
-                        curProc.Threads.Count);
-
-                    _disablePublishMetrics = sessionCount > 40 || monitoredItemsCount > 500;
+                        curProc.Threads.Count,
+                        PeriodicLoggingTimerSeconds,
+                        _countCreateSession,
+                        _countCreateSubscription,
+                        _countCreateMonitoredItems,
+                        _countPublish,
+                        _countRead,
+                        _countWrite,
+                        PublishMetricsEnabled);
+
+                    _countCreateSession = 0;
+                    _countCreateSubscription = 0;
+                    _countCreateMonitoredItems = 0;
+                    _countPublish = 0;
+                    _countRead = 0;
+                    _countWrite = 0;
                 }
                 catch
                 {
                     // Ignore error during logging.
                 }
             },
-            state: null, dueTime: TimeSpan.FromSeconds(60), period: TimeSpan.FromSeconds(60));
+            state: null, dueTime: TimeSpan.FromSeconds(PeriodicLoggingTimerSeconds), period: TimeSpan.FromSeconds(PeriodicLoggingTimerSeconds));
 
         MetricsHelper.IsEnabled = Config.OtlpEndpointUri is not null;
     }
 
+    /// <summary>
+    /// Enable publish requests metrics only if the following apply:
+    /// 1) Metrics are enabled by specifying OtlpEndpointUri,
+    /// 2) OtlpPublishMetrics is "enable",
+    /// 3) OtlpPublishMetrics is not "disable",
+    /// 4) When OtlpPublishMetrics is "auto": sessions <= 40 and monitored items <= 500.
+    /// </summary>
+    private bool PublishMetricsEnabled =>
+        MetricsHelper.IsEnabled &&
+        (
+         (Config.OtlpPublishMetrics == "enable" && Config.OtlpPublishMetrics != "disable") ||
+         (Config.OtlpPublishMetrics == "auto" && !_autoDisablePublishMetrics)
+        );
+
     public override ResponseHeader CreateSession(
         RequestHeader requestHeader,
         ApplicationDescription clientDescription,
@@ -103,7 +139,9 @@ public override ResponseHeader CreateSession(
         out SignedSoftwareCertificateCollection serverSoftwareCertificates,
         out SignatureData serverSignature,
         out uint maxRequestMessageSize)
-    {
+    {
+        _countCreateSession++;
+
         try
         {
             var responseHeader = base.CreateSession(requestHeader, clientDescription, serverUri, endpointUrl, sessionName, clientNonce, clientCertificate, requestedSessionTimeout, maxResponseMessageSize, out sessionId, out authenticationToken, out revisedSessionTimeout, out serverNonce, out serverCertificate, out serverEndpoints, out serverSoftwareCertificates, out serverSignature, out maxRequestMessageSize);
@@ -136,6 +174,8 @@ public override ResponseHeader CreateSubscription(
         out uint revisedLifetimeCount,
         out uint revisedMaxKeepAliveCount)
     {
+        _countCreateSubscription++;
+
         try
         {
             OperationContext context = ValidateRequest(requestHeader, RequestType.CreateSubscription);
@@ -168,6 +208,8 @@ public override ResponseHeader CreateMonitoredItems(
         out MonitoredItemCreateResultCollection results,
         out DiagnosticInfoCollection diagnosticInfos)
     {
+        _countCreateMonitoredItems += (uint)itemsToCreate.Count;
+
         results = default;
         diagnosticInfos = default;
 
@@ -206,6 +248,8 @@ public override ResponseHeader Publish(
         out StatusCodeCollection results,
         out DiagnosticInfoCollection diagnosticInfos)
     {
+        _countPublish++;
+
         subscriptionId = default;
         availableSequenceNumbers = default;
         moreNotifications = default;
@@ -219,9 +263,29 @@ public override ResponseHeader Publish(
 
             var responseHeader = base.Publish(requestHeader, subscriptionAcknowledgements, out subscriptionId, out availableSequenceNumbers, out moreNotifications, out notificationMessage, out results, out diagnosticInfos);
 
-            if (!_disablePublishMetrics)
+            if (PublishMetricsEnabled)
             {
-                MetricsHelper.AddPublishedCount(context.SessionId.ToString(), subscriptionId.ToString(), notificationMessage, _logger);
+                int events = 0;
+                int dataChanges = 0;
+                int diagnostics = 0;
+
+                notificationMessage.NotificationData.ForEach(x => {
+                    if (x.Body is DataChangeNotification changeNotification)
+                    {
+                        dataChanges += changeNotification.MonitoredItems.Count;
+                        diagnostics += changeNotification.DiagnosticInfos.Count;
+                    }
+                    else if (x.Body is EventNotificationList eventNotification)
+                    {
+                        events += eventNotification.Events.Count;
+                    }
+                    else
+                    {
+                        LogUnknownNotification(x.Body.GetType().Name);
+                    }
+                });
+
+                MetricsHelper.AddPublishedCount(context.SessionId.ToString(), subscriptionId.ToString(), dataChanges, events);
             }
 
             LogSuccessWithSessionIdAndSubscriptionId(
@@ -281,6 +345,8 @@ public override ResponseHeader Read(
         out DataValueCollection results,
         out DiagnosticInfoCollection diagnosticInfos)
     {
+        _countRead++;
+
         results = default;
         diagnosticInfos = default;
 
@@ -303,6 +369,8 @@ public override ResponseHeader Read(
 
     public override ResponseHeader Write(RequestHeader requestHeader, WriteValueCollection nodesToWrite, out StatusCodeCollection results, out DiagnosticInfoCollection diagnosticInfos)
     {
+        _countWrite++;
+
         try
         {
             var responseHeader = base.Write(requestHeader, nodesToWrite, out results, out diagnosticInfos);
@@ -530,12 +598,35 @@ protected override void OnServerStopping()
         Level = LogLevel.Information,
         Message = "\n\t# Open sessions: {Sessions}\n" +
                   "\t# Open subscriptions: {Subscriptions}\n" +
-                  "\t# Monitored items: {MonitoredItems:N0}\n" +
+                  "\t# Monitored items: {MonitoredItems:N0}\n" +
                   "\t# Working set: {WorkingSet:N0} MB\n" +
                   "\t# Available worker threads: {AvailWorkerThreads:N0}\n" +
                   "\t# Available completion port threads: {AvailCompletionPortThreads:N0}\n" +
-                  "\t# Thread count: {ThreadCount:N0}")]
-    partial void LogPeriodicInfo(int sessions, int subscriptions, int monitoredItems, long workingSet, int availWorkerThreads, int availCompletionPortThreads, int threadCount);
+                  "\t# Thread count: {ThreadCount:N0}\n" +
+                  "\t# Statistics for the last {PeriodicLoggingTimerSeconds} s\n" +
+                  "\t# Sessions created: {CountCreateSession}\n" +
+                  "\t# Subscriptions created: {CountCreateSubscription}\n" +
+                  "\t# Monitored items created: {CountCreateMonitoredItems}\n" +
+                  "\t# Publish requests: {CountPublish}\n" +
+                  "\t# Read requests: {CountRead}\n" +
+                  "\t# Write requests: {CountWrite}\n" +
+                  "\t# Publish metrics enabled: {PublishMetricsEnabled:N0}")]
+    partial void LogPeriodicInfo(
+        int sessions,
+        int subscriptions,
+        int monitoredItems,
+        long workingSet,
+        int availWorkerThreads,
+        int availCompletionPortThreads,
+        int threadCount,
+        int periodicLoggingTimerSeconds,
+        uint countCreateSession,
+        uint countCreateSubscription,
+        uint countCreateMonitoredItems,
+        uint countPublish,
+        uint countRead,
+        uint countWrite,
+        bool publishMetricsEnabled);
 
     [LoggerMessage(
         Level = LogLevel.Debug,
@@ -571,4 +662,9 @@ protected override void OnServerStopping()
         Level = LogLevel.Error,
         Message = "{message}")]
     partial void LogErrorMessage(string message);
+
+    [LoggerMessage(
+        Level = LogLevel.Debug,
+        Message = "Unknown notification type: {NotificationType}")]
+    partial void LogUnknownNotification(string notificationType);
 }

tests/MetricsTests.cs

@@ -2,7 +2,6 @@
 
 using FluentAssertions;
 using NUnit.Framework;
-using Opc.Ua;
 using System;
 using System.Collections.Generic;
 using System.Diagnostics.Metrics;
@@ -20,6 +19,7 @@
     {
         _metrics = new Dictionary<string, object>();
         _meterListener = new MeterListener();
+
         _meterListener.InstrumentPublished = (instrument, listener) => {
             if (instrument.Meter.Name == MetricsHelper.Meter.Name)
             {
@@ -28,13 +28,13 @@
         };
 
         _meterListener.SetMeasurementEventCallback(
             (Instrument instrument, long measurement, ReadOnlySpan<KeyValuePair<string, object?>> tags, object? state) => _metrics.Add(instrument.Name, measurement));
 
         _meterListener.SetMeasurementEventCallback(
             (Instrument instrument, double measurement, ReadOnlySpan<KeyValuePair<string, object?>> tags, object? state) => _metrics.Add(instrument.Name, measurement));
 
         _meterListener.SetMeasurementEventCallback(
             (Instrument instrument, int measurement, ReadOnlySpan<KeyValuePair<string, object?>> tags, object? state) => _metrics.Add(instrument.Name, measurement));
 
         _meterListener.Start();
     }
@@ -80,11 +80,21 @@
         counter.Should().Be(1);
     }
 
+    [Test]
+    public void TestAddPublishedCount()
+    {
+        var sessionId = Guid.NewGuid().ToString();
+        var subscriptionId = Guid.NewGuid().ToString();
+        MetricsHelper.AddPublishedCount(sessionId, subscriptionId, 1, 0);
+        _metrics.TryGetValue("opc_plc_published_count_with_type", out var counter).Should().BeTrue();
+        counter.Should().Be(1);
+    }
+
     [Test]
     public void TestRecordTotalErrors()
     {
         MetricsHelper.RecordTotalErrors("operation");
-        _metrics.TryGetValue("opc_plc_total_errors", out var counter).Should().BeTrue(); ;
+        _metrics.TryGetValue("opc_plc_total_errors", out var counter).Should().BeTrue();
         counter.Should().Be(1);
     }
 }

version.json

@@ -1,6 +1,6 @@
 {
   "$schema": "https://raw.githubusercontent.com/AArnott/Nerdbank.GitVersioning/master/src/NerdBank.GitVersioning/version.schema.json",
-  "version": "2.12.21",
+  "version": "2.12.22",
   "versionHeightOffset": -1,
   "publicReleaseRefSpec": [
     "^refs/heads/main$",