Added additional metrics

vostok · Feb 2, 2022 · b2b6704 · b2b6704
1 parent a423dde
commit b2b6704
Show file tree

Hide file tree

Showing 4 changed files with 39 additions and 10 deletions.
diff --git a/Vostok.Hercules.Consumers/BatchesStreamConsumer.cs b/Vostok.Hercules.Consumers/BatchesStreamConsumer.cs
@@ -52,11 +52,11 @@ public BatchesStreamConsumer([NotNull] BatchesStreamConsumerSettings<T> settings
             var bufferPool = new BufferPool(settings.MaxPooledBufferSize, settings.MaxPooledBuffersPerBucket);
             client = new StreamApiRequestSender(settings.StreamApiCluster, log /*.WithErrorsTransformedToWarns()*/, bufferPool, settings.StreamApiClientAdditionalSetup);
 
-            var metricContext = settings.MetricContext ?? new DevNullMetricContext();
-            eventsMetric = metricContext.CreateIntegerGauge("events", "type", new IntegerGaugeConfig {ResetOnScrape = true});
-            iterationMetric = metricContext.CreateSummary("iteration", "type", new SummaryConfig {Quantiles = new[] {0.5, 0.75, 1}});
-            metricContext.CreateFuncGauge("events", "type").For("remaining").SetValueProvider(() => CountStreamRemainingEvents());
-            metricContext.CreateFuncGauge("buffer", "type").For("rented_reader").SetValueProvider(() => BufferPool.Rented);
+            var instanceMetricContext = settings.InstanceMetricContext ?? new DevNullMetricContext();
+            eventsMetric = instanceMetricContext.CreateIntegerGauge("events", "type", new IntegerGaugeConfig {ResetOnScrape = true});
+            iterationMetric = instanceMetricContext.CreateSummary("iteration", "type", new SummaryConfig {Quantiles = new[] {0.5, 0.75, 1}});
+            instanceMetricContext.CreateFuncGauge("events", "type").For("remaining").SetValueProvider(() => CountStreamRemainingEvents());
+            instanceMetricContext.CreateFuncGauge("buffer", "type").For("rented_reader").SetValueProvider(() => BufferPool.Rented);
         }
 
         public async Task RunAsync(CancellationToken cancellationToken)

diff --git a/Vostok.Hercules.Consumers/BatchesStreamConsumerSettings.cs b/Vostok.Hercules.Consumers/BatchesStreamConsumerSettings.cs
@@ -62,7 +62,10 @@ public BatchesStreamConsumerSettings(
         public ClusterClientSetup StreamApiClientAdditionalSetup { get; set; }
 
         [CanBeNull]
-        public IMetricContext MetricContext { get; set; }
+        public IMetricContext ApplicationMetricContext { get; set; }
+
+        [CanBeNull]
+        public IMetricContext InstanceMetricContext { get; set; }
 
         [CanBeNull]
         public ITracer Tracer { get; set; }

diff --git a/Vostok.Hercules.Consumers/Helpers/HistogramHelper.cs b/Vostok.Hercules.Consumers/Helpers/HistogramHelper.cs
@@ -0,0 +1,19 @@
+using Vostok.Metrics.Primitives.Timer;
+
+namespace Vostok.Hercules.Consumers.Helpers
+{
+    internal static class HistogramHelper
+    {
+        public static HistogramBuckets CreateDefaultBuckets()
+        {
+            var upperBounds = new double[30];
+
+            upperBounds[0] = 0;
+            upperBounds[1] = 1;
+            for (var i = 2; i < upperBounds.Length; i++)
+                upperBounds[i] = upperBounds[i - 1] * 2;
+
+            return new HistogramBuckets(upperBounds);
+        }
+    }
+}
diff --git a/Vostok.Hercules.Consumers/WindowedStreamConsumer.cs b/Vostok.Hercules.Consumers/WindowedStreamConsumer.cs
@@ -26,8 +26,8 @@ public class WindowedStreamConsumer<T, TKey> : BatchesStreamConsumer<T>
         private readonly ILog log;
         private readonly ITracer tracer;
         private readonly IMetricGroup1<IIntegerGauge> stateMetric;
+        private readonly ITimer eventLagMetric;
         private readonly Dictionary<TKey, Windows<T, TKey>> windows;
-
         private volatile StreamCoordinates leftCoordinates;
 
         public WindowedStreamConsumer([NotNull] WindowedStreamConsumerSettings<T, TKey> settings, [CanBeNull] ILog log)
@@ -39,6 +39,12 @@ public WindowedStreamConsumer([NotNull] WindowedStreamConsumerSettings<T, TKey>
 
             windows = new Dictionary<TKey, Windows<T, TKey>>();
 
+            var applicationMetricContext = settings.ApplicationMetricContext ?? new DevNullMetricContext();
+            eventLagMetric = applicationMetricContext.CreateHistogram("eventLag", new HistogramConfig {Buckets = HistogramHelper.CreateDefaultBuckets(), Unit = WellKnownUnits.Milliseconds});
+
+            var instanceMetricContext = settings.InstanceMetricContext ?? new DevNullMetricContext();
+            stateMetric = instanceMetricContext.CreateIntegerGauge("state", "type");
+
             var settingsOnBatchEnd = settings.OnBatchEnd;
             settings.OnBatchEnd = c =>
             {
@@ -60,9 +66,6 @@ public WindowedStreamConsumer([NotNull] WindowedStreamConsumerSettings<T, TKey>
                 Restart(c).GetAwaiter().GetResult();
                 settingsOnRestart?.Invoke(c);
             };
-
-            var metricContext = settings.MetricContext ?? new DevNullMetricContext();
-            stateMetric = metricContext.CreateIntegerGauge("state", "type");
         }
 
         private async Task Restart(StreamCoordinates rightCoordinates)
@@ -147,6 +150,10 @@ private async Task<long> RestartPartition(int partition, int partitionsCount, lo
         private void AddEvent(T @event, StreamCoordinates queryCoordinates)
         {
             var key = settings.KeyProvider(@event);
+
+            var lag = DateTime.Now - settings.TimestampProvider(@event);
+            eventLagMetric.Report(lag);
+
             if (!windows.ContainsKey(key))
                 windows[key] = new Windows<T, TKey>(key, settings);
             if (!windows[key].AddEvent(@event, queryCoordinates) && !restart)