add healthcheck and remove redundant private modifier

RaaLabs · Oct 21, 2024 · b33b685 · b33b685
1 parent 096ea21
commit b33b685
Show file tree

Hide file tree

Showing 5 changed files with 65 additions and 18 deletions.
diff --git a/Source/DataReader.cs b/Source/DataReader.cs
@@ -14,12 +14,12 @@ namespace RaaLabs.Edge.Connectors.OPCUA;
 
 public class DataReader : IRetrieveData
 {
-    private readonly TimeSpan _publishInterval;
-    private readonly List<(NodeId, TimeSpan)> _subscribeNodes;
-    private readonly List<(NodeId, TimeSpan)> _readNodes;
-    private readonly ICanSubscribeToNodes _subscriber;
-    private readonly ICanReadNodes _reader;
-    private readonly ILogger _logger;
+    readonly TimeSpan _publishInterval;
+    readonly List<(NodeId, TimeSpan)> _subscribeNodes;
+    readonly List<(NodeId, TimeSpan)> _readNodes;
+    readonly ICanSubscribeToNodes _subscriber;
+    readonly ICanReadNodes _reader;
+    readonly ILogger _logger;
 
     public DataReader(ConnectorConfiguration config, ICanSubscribeToNodes subscriber, ICanReadNodes reader, ILogger logger)
     {
@@ -74,21 +74,21 @@ public async Task ReadDataForever(ISession connection, Func<NodeValue, Task> han
         }
     }
 
-    private Task SubscribeOrSleep(ISession connection, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken) =>
+    Task SubscribeOrSleep(ISession connection, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken) =>
         _subscribeNodes.Count switch
         {
             0 => Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken),
             _ => Task.Run(() => _subscriber.SubscribeToChangesFor(connection, _publishInterval, _subscribeNodes, handleValue, cancellationToken), cancellationToken)
         };
 
-    private Task ReadOrSleep(ISession connection, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken) =>
+    Task ReadOrSleep(ISession connection, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken) =>
         _readNodes.Count switch
         {
             0 => Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken),
             _ => Task.Run(() => _reader.ReadNodesForever(connection, _readNodes, handleValue, cancellationToken), cancellationToken)
         };
 
-    private static void ThrowIfFailedWithError(Task task)
+    static void ThrowIfFailedWithError(Task task)
     {
         if (task.Exception?.GetBaseException() is {} error and not OperationCanceledException)
         {

diff --git a/Source/HealthCheck.cs b/Source/HealthCheck.cs
@@ -0,0 +1,46 @@
+using System;
+using System.Globalization;
+using RaaLabs.Edge.Connectors.OPCUA.Events;
+using RaaLabs.Edge.Modules.Diagnostics.Health;
+using RaaLabs.Edge.Modules.EventHandling;
+using Serilog;
+
+namespace RaaLabs.Edge.Connectors.OPCUA;
+
+public class HealthCheck : IConsumeEvent<OpcuaDatapointOutput>, IExposeHealthStatus
+{
+    readonly TimeSpan _maxTimeBetweenDatapoints;
+    readonly TimeProvider _clock;
+    readonly ILogger _logger;
+    DateTimeOffset _lastTimestampReceived;
+
+    public HealthCheck(TimeProvider clock, ILogger logger)
+    {
+        _maxTimeBetweenDatapoints = TimeSpan.FromSeconds(double.Parse(Environment.GetEnvironmentVariable("HEALTHCHECK_MAX_SECONDS_BETWEEN_DATAPOINTS") ?? "120", CultureInfo.CurrentCulture));
+        _clock = clock;
+        _logger = logger;
+        _lastTimestampReceived = _clock.GetUtcNow();
+    }
+
+    public void Handle(OpcuaDatapointOutput @event)
+    {
+        _lastTimestampReceived = _clock.GetUtcNow();
+        _logger.Debug("Sent data, updating last timestamp received in healthcheck with {LastTimestampReceived}", _lastTimestampReceived);
+    }
+
+    public bool IsReady => true;
+    public bool IsHealthy
+    {
+        get
+        {
+            var timeSinceLastDatapoint = _clock.GetUtcNow() - _lastTimestampReceived;
+            if (timeSinceLastDatapoint > _maxTimeBetweenDatapoints)
+            {
+                _logger.Warning("Time since data was sent is {Since}, returning unhealthy", timeSinceLastDatapoint);
+                return false;
+            }
+            _logger.Debug("Time since data was sent is {Since}, returning healthy", timeSinceLastDatapoint);
+            return true;
+        }
+    }
+}
diff --git a/Source/Program.cs b/Source/Program.cs
@@ -24,6 +24,7 @@ static void Main(string[] args)
             .WithModule<Configuration>()
             .WithModule<EdgeHub>()
             .WithModule<RaaLabsDiagnostics>()
+            .WithHandler<HealthCheck>()
             .WithTask<Connector>()
             .WithManualRegistration(_ =>
             {

diff --git a/Source/Reader.cs b/Source/Reader.cs
@@ -14,8 +14,8 @@ namespace RaaLabs.Edge.Connectors.OPCUA;
 
 public class Reader : ICanReadNodes
 {
-    private readonly ILogger _logger;
-    private readonly IMetricsHandler _metrics;
+    readonly ILogger _logger;
+    readonly IMetricsHandler _metrics;
 
     public Reader(ILogger logger, IMetricsHandler metrics)
     {
@@ -50,7 +50,7 @@ public async Task ReadNodesForever(ISession connection, IEnumerable<(NodeId node
         _logger.Debug("Read operation completed...");
     }
 
-    private async Task ReadNodeForever(ISession connection, NodeId node, TimeSpan readInterval, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken)
+    async Task ReadNodeForever(ISession connection, NodeId node, TimeSpan readInterval, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken)
     {
         using var timer = new PeriodicTimer(readInterval);
         while (!cancellationToken.IsCancellationRequested)

diff --git a/Source/Subscriber.cs b/Source/Subscriber.cs
@@ -15,8 +15,8 @@ namespace RaaLabs.Edge.Connectors.OPCUA;
 
 public class Subscriber : ICanSubscribeToNodes
 {
-    private readonly IMetricsHandler _metrics;
-    private readonly ILogger _logger;
+    readonly IMetricsHandler _metrics;
+    readonly ILogger _logger;
 
     public Subscriber(IMetricsHandler metrics, ILogger logger)
     {
@@ -58,7 +58,7 @@ public async Task SubscribeToChangesFor(ISession connection, TimeSpan publishInt
         }
     }
 
-    private MonitoredItem MonitoringFor(NodeId nodeId, TimeSpan samplingInterval, ChannelWriter<NodeValue> writer)
+    MonitoredItem MonitoringFor(NodeId nodeId, TimeSpan samplingInterval, ChannelWriter<NodeValue> writer)
     {
         var monitored = new MonitoredItem()
         {
@@ -83,14 +83,14 @@ private MonitoredItem MonitoringFor(NodeId nodeId, TimeSpan samplingInterval, Ch
         return monitored;
     }
 
-    private void DeleteSubscriptionWhenCancelled(Subscription subscription, CancellationToken cancellationToken) =>
+    void DeleteSubscriptionWhenCancelled(Subscription subscription, CancellationToken cancellationToken) =>
         cancellationToken.Register(() =>
         {
             _logger.Debug("CancellationToken cancelled, deleting subscription");
             subscription.Delete(true);
         });
 
-    private void CompleteChannelWhenSubscriptionCompletes(Subscription subscription, ChannelWriter<NodeValue> writer)
+    void CompleteChannelWhenSubscriptionCompletes(Subscription subscription, ChannelWriter<NodeValue> writer)
     {
         subscription.PublishStatusChanged += (_, changed) =>
         {
@@ -112,7 +112,7 @@ private void CompleteChannelWhenSubscriptionCompletes(Subscription subscription,
         };
     }
 
-    private static Subscription CreateEmptySubscription(TimeSpan publishInterval) =>
+    static Subscription CreateEmptySubscription(TimeSpan publishInterval) =>
         new()
         {
             PublishingInterval = (int)publishInterval.TotalMilliseconds,