Add metric on extractor upload queue length (#874)

* Add metric on extractor upload queue length

* Review comment

Author: einarmo

Extractor/Utils/AsyncBlockingQueue.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using Nito.AsyncEx;
+using Prometheus;
 
 namespace Cognite.OpcUa.Utils
 {
@@ -32,6 +33,9 @@ public class AsyncBlockingQueue<T>
 
         private ILogger log;
 
+        private static readonly Gauge queueLength = Metrics
+            .CreateGauge("opcua_extractor_queue_length", "Length of the upload queues", "type");
+
         /// <summary>
         /// Constructor.
         /// </summary>
@@ -52,6 +56,11 @@ private void NotifyOverflow()
             OnQueueOverflow?.Invoke(this, new EventArgs());
         }
 
+        private void UpdateMetrics()
+        {
+            queueLength.WithLabels(Name).Set(queue.Count);
+        }
+
         /// <summary>
         /// Enqeueue an item, blocking until the queue has capacity to accept the item.
         /// </summary>
@@ -67,6 +76,7 @@ public void Enqueue(T item, CancellationToken token = default)
                     queueNotFull.Wait(token);
                 }
                 queue.Enqueue(item);
+                UpdateMetrics();
                 queueNotEmpty.Notify();
                 if (Capacity > 0 && queue.Count >= Capacity) NotifyOverflow();
             }
@@ -86,13 +96,14 @@ public void Enqueue(IEnumerable<T> items, CancellationToken token = default)
                     while (Capacity > 0 && queue.Count >= Capacity)
                     {
                         log.LogTrace("{} queue is full", Name);
+                        UpdateMetrics();
                         queueNotFull.Wait(token);
                     }
                     queue.Enqueue(item);
                     queueNotEmpty.Notify();
                     if (Capacity > 0 && queue.Count >= Capacity) NotifyOverflow();
                 }
-
+                UpdateMetrics();
             }
         }
 
@@ -111,6 +122,7 @@ public async Task EnqueueAsync(T item, CancellationToken token = default)
                     await queueNotFull.WaitAsync(token);
                 }
                 queue.Enqueue(item);
+                UpdateMetrics();
                 queueNotEmpty.Notify();
                 if (Capacity > 0 && queue.Count >= Capacity) NotifyOverflow();
             }
@@ -130,12 +142,14 @@ public async Task EnqueueAsync(IEnumerable<T> items, CancellationToken token = d
                     while (Capacity > 0 && queue.Count >= Capacity)
                     {
                         log.LogTrace("{} queue is full", Name);
+                        UpdateMetrics();
                         await queueNotFull.WaitAsync(token);
                     }
                     queue.Enqueue(item);
                     queueNotEmpty.Notify();
                     if (Capacity > 0 && queue.Count >= Capacity) NotifyOverflow();
                 }
+                UpdateMetrics();
             }
         }
 
@@ -152,6 +166,7 @@ public async IAsyncEnumerable<T> DrainAsync([EnumeratorCancellation] Cancellatio
                 {
                     yield return item;
                 }
+                UpdateMetrics();
                 queueNotFull.NotifyAll();
             }
         }
@@ -169,6 +184,7 @@ public IEnumerable<T> Drain(CancellationToken token = default)
                 {
                     yield return item;
                 }
+                UpdateMetrics();
                 queueNotFull.NotifyAll();
             }
         }
@@ -186,6 +202,7 @@ public bool TryDequeue(out T? item, CancellationToken token = default)
             using (queueMutex.Lock(token))
             {
                 var r = queue.TryDequeue(out item);
+                UpdateMetrics();
                 if (r) queueNotFull.Notify();
                 return r;
             }
@@ -202,6 +219,7 @@ public bool TryDequeue(out T? item, CancellationToken token = default)
             using (await queueMutex.LockAsync(token))
             {
                 var r = queue.TryDequeue(out var item);
+                UpdateMetrics();
                 if (r) queueNotFull.Notify();
                 else return default;
                 return item;
@@ -221,6 +239,7 @@ public T Dequeue(CancellationToken token = default)
                 {
                     token.ThrowIfCancellationRequested();
                     var r = queue.TryDequeue(out var item);
+                    UpdateMetrics();
                     if (r)
                     {
                         queueNotFull.Notify();
@@ -244,6 +263,7 @@ public async Task<T> DequeueAsync(CancellationToken token = default)
                 {
                     token.ThrowIfCancellationRequested();
                     var r = queue.TryDequeue(out var item);
+                    UpdateMetrics();
                     if (r)
                     {
                         queueNotFull.Notify();
@@ -263,6 +283,7 @@ public async Task Clear(CancellationToken token = default)
             using (await queueMutex.LockAsync(token))
             {
                 queue.Clear();
+                UpdateMetrics();
                 queueNotFull.NotifyAll();
             }
         }

manifest.yml

@@ -67,7 +67,12 @@ schema:
    - "https://raw.githubusercontent.com/"
 
 versions:
-  "2.28.2":
+  "2.39.0":
+    description: Add a metric on upload queue length.
+    changelog:
+      added:
+        - Added a metric opcua_extractor_queue_length to monitor the length of the upload queues. This metric is labeled either "Events" or "Datapoints".
+  "2.38.2":
     description: Avoid restarting history after rebrowse if no new nodes were discovered.
     changelog:
       fixed: