Improve metric sink concept

src/Promitor.Agents.Scraper/Dockerfile

@@ -8,7 +8,6 @@ COPY Promitor.Core.Telemetry/* Promitor.Core.Telemetry/
 COPY Promitor.Core.Telemetry.Metrics/* Promitor.Core.Telemetry.Metrics/
 COPY Promitor.Integrations.AzureMonitor/* Promitor.Integrations.AzureMonitor/
 COPY Promitor.Integrations.AzureStorage/* Promitor.Integrations.AzureStorage/
-COPY Promitor.Integrations.Sinks.Core/* Promitor.Integrations.Sinks.Core/
 COPY Promitor.Integrations.Sinks.Statsd/* Promitor.Integrations.Sinks.Statsd/
 COPY Promitor.Agents.Scraper/* Promitor.Agents.Scraper/
 RUN dotnet publish Promitor.Agents.Scraper/Promitor.Agents.Scraper.csproj --configuration release --output app

src/Promitor.Agents.Scraper/Extensions/IServiceCollectionExtensions.cs

@@ -7,6 +7,7 @@
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Diagnostics.HealthChecks;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -51,7 +52,7 @@ public static IServiceCollection ScheduleMetricScraping(this IServiceCollection
             var metrics = metricsProvider.Get(applyDefaults: true);
 
             var loggerFactory = serviceProviderToCreateJobsWith.GetService<ILoggerFactory>();
-            var metricSink = serviceProviderToCreateJobsWith.GetRequiredService<IMetricSink>();
+            var metricSinkWriter = serviceProviderToCreateJobsWith.GetRequiredService<MetricSinkWriter>();
             var azureMonitorLoggingConfiguration = serviceProviderToCreateJobsWith.GetService<IOptions<AzureMonitorLoggingConfiguration>>();
             var configuration = serviceProviderToCreateJobsWith.GetService<IConfiguration>();
             var runtimeMetricCollector = serviceProviderToCreateJobsWith.GetService<IRuntimeMetricsCollector>();
@@ -72,7 +73,7 @@ public static IServiceCollection ScheduleMetricScraping(this IServiceCollection
                         builder.AddJob(jobServices =>
                         {
                             return new MetricScrapingJob(jobName, scrapeDefinition,
-                                metricSink,
+                                metricSinkWriter,
                                 jobServices.GetService<IPrometheusMetricWriter>(),
                                 jobServices.GetService<MetricScraperFactory>(),
                                 azureMonitorClient,
@@ -145,12 +146,14 @@ public static IServiceCollection UseMetricSinks(this IServiceCollection services
                 AddStatsdMetricSink(services, metricSinkConfiguration.Statsd);
             }
 
+            services.TryAddSingleton<MetricSinkWriter>();
+
             return services;
         }
 
         private static void AddStatsdMetricSink(IServiceCollection services, StatsdSinkConfiguration statsdConfiguration)
         {
-            services.AddSingleton<IMetricSink, StatsdMetricSink>();
+            services.AddTransient<IMetricSink, StatsdMetricSink>();
             services.AddStatsD(provider =>
             {
                 var loggerFactory = provider.GetRequiredService<ILoggerFactory>();

src/Promitor.Agents.Scraper/Promitor.Agents.Scraper.csproj

@@ -55,7 +55,6 @@
     <ProjectReference Include="..\Promitor.Core\Promitor.Core.csproj" />
     <ProjectReference Include="..\Promitor.Integrations.AzureMonitor\Promitor.Integrations.AzureMonitor.csproj" />
     <ProjectReference Include="..\Promitor.Integrations.AzureStorage\Promitor.Integrations.AzureStorage.csproj" />
-    <ProjectReference Include="..\Promitor.Integrations.Sinks.Core\Promitor.Integrations.Sinks.Core.csproj" />
     <ProjectReference Include="..\Promitor.Integrations.Sinks.Statsd\Promitor.Integrations.Sinks.Statsd.csproj" />
   </ItemGroup>
 

src/Promitor.Agents.Scraper/Scheduling/MetricScrapingJob.cs

@@ -17,14 +17,14 @@ public class MetricScrapingJob : IScheduledJob
         private readonly ScrapeDefinition<IAzureResourceDefinition> _metricScrapeDefinition;
         private readonly IPrometheusMetricWriter _prometheusMetricWriter;
         private readonly AzureMonitorClient _azureMonitorClient;
-        private readonly IMetricSink _metricSink;
+        private readonly MetricSinkWriter _metricSinkWriter;
         private readonly ILogger _logger;
 
         private readonly MetricScraperFactory _metricScraperFactory;
 
         public MetricScrapingJob(string jobName,
             ScrapeDefinition<IAzureResourceDefinition> metricScrapeDefinition,
-            IMetricSink metricSink,
+            MetricSinkWriter metricSinkWriter,
             IPrometheusMetricWriter prometheusMetricWriter,
             MetricScraperFactory metricScraperFactory,
             AzureMonitorClient azureMonitorClient,
@@ -41,7 +41,7 @@ public MetricScrapingJob(string jobName,
 
             _metricScrapeDefinition = metricScrapeDefinition;
             _prometheusMetricWriter = prometheusMetricWriter;
-            _metricSink = metricSink;
+            _metricSinkWriter = metricSinkWriter;
             _logger = logger;
 
             _metricScraperFactory = metricScraperFactory;
@@ -68,7 +68,7 @@ private async Task ScrapeMetric(ScrapeDefinition<IAzureResourceDefinition> metri
         {
             _logger.LogInformation("Scraping {MetricName} for resource type {ResourceType}", metricDefinitionDefinition.PrometheusMetricDefinition.Name, metricDefinitionDefinition.Resource.ResourceType);
 
-            var scraper = _metricScraperFactory.CreateScraper(metricDefinitionDefinition.Resource.ResourceType, _metricSink, _prometheusMetricWriter, _azureMonitorClient);
+            var scraper = _metricScraperFactory.CreateScraper(metricDefinitionDefinition.Resource.ResourceType, _metricSinkWriter, _prometheusMetricWriter, _azureMonitorClient);
             await scraper.ScrapeAsync(metricDefinitionDefinition);
         }
     }

src/Promitor.Core.Scraping/Factories/MetricScraperFactory.cs

@@ -26,12 +26,12 @@ public MetricScraperFactory(ILogger<MetricScraperFactory> logger)
         ///     Creates a scraper that is capable of scraping a specific resource type
         /// </summary>
         /// <param name="metricDefinitionResourceType">Resource type to scrape</param>
-        /// <param name="sink">Sink to write all metrics to</param>
+        /// <param name="metricSinkWriter">Writer to send metrics to all sinks</param>
         /// <param name="prometheusMetricWriter">Metrics collector for our Prometheus scraping endpoint</param>
         /// <param name="azureMonitorClient">Client to interact with Azure Monitor</param>
-        public IScraper<IAzureResourceDefinition> CreateScraper(ResourceType metricDefinitionResourceType, IMetricSink sink, IPrometheusMetricWriter prometheusMetricWriter, AzureMonitorClient azureMonitorClient)
+        public IScraper<IAzureResourceDefinition> CreateScraper(ResourceType metricDefinitionResourceType, MetricSinkWriter metricSinkWriter, IPrometheusMetricWriter prometheusMetricWriter, AzureMonitorClient azureMonitorClient)
         {
-            var scraperConfiguration = new ScraperConfiguration(azureMonitorClient, sink, prometheusMetricWriter, _logger);
+            var scraperConfiguration = new ScraperConfiguration(azureMonitorClient, metricSinkWriter, prometheusMetricWriter, _logger);
 
             switch (metricDefinitionResourceType)
             {

src/Promitor.Core.Scraping/Scraper.cs

@@ -20,7 +20,7 @@ public abstract class Scraper<TResourceDefinition> : IScraper<IAzureResourceDefi
       where TResourceDefinition : class, IAzureResourceDefinition
     {
         private readonly IPrometheusMetricWriter _prometheusMetricWriter;
-        private readonly IMetricSink _metricSink;
+        private readonly MetricSinkWriter _metricSinkWriter;
         private readonly ILogger _logger;
 
         /// <summary>
@@ -31,7 +31,7 @@ protected Scraper(ScraperConfiguration scraperConfiguration)
             Guard.NotNull(scraperConfiguration, nameof(scraperConfiguration));
 
             _logger = scraperConfiguration.Logger;
-            _metricSink = scraperConfiguration.Sink;
+            _metricSinkWriter = scraperConfiguration.MetricSinkWriter;
             _prometheusMetricWriter = scraperConfiguration.PrometheusMetricWriter;
 
             AzureMonitorClient = scraperConfiguration.AzureMonitorClient;
@@ -73,7 +73,7 @@ public async Task ScrapeAsync(ScrapeDefinition<IAzureResourceDefinition> scrapeD
 
                 LogMeasuredMetrics(scrapeDefinition, scrapedMetricResult, aggregationInterval);
 
-                _metricSink.ReportMetric(scrapeDefinition.PrometheusMetricDefinition.Name, scrapeDefinition.PrometheusMetricDefinition.Description, scrapedMetricResult);
+                await _metricSinkWriter.ReportMetricAsync(scrapeDefinition.PrometheusMetricDefinition.Name, scrapeDefinition.PrometheusMetricDefinition.Description, scrapedMetricResult);
                 _prometheusMetricWriter.ReportMetric(scrapeDefinition.PrometheusMetricDefinition, scrapedMetricResult);
             }
             catch (ErrorResponseException errorResponseException)

src/Promitor.Core.Scraping/ScraperConfiguration.cs

@@ -19,9 +19,9 @@ public class ScraperConfiguration
         public IPrometheusMetricWriter PrometheusMetricWriter { get; }
 
         /// <summary>
-        ///     Sink to write all metrics to
+        ///     Writer to send metrics to all configured sinks
         /// </summary>
-        public IMetricSink Sink { get; }
+        public MetricSinkWriter MetricSinkWriter { get; }
 
         /// <summary>
         ///     Logger used for telemetry
@@ -32,20 +32,20 @@ public class ScraperConfiguration
         /// Constructor
         /// </summary>
         /// <param name="azureMonitorClient">Client to communicate with Azure Monitor</param>
-        /// <param name="sink">Sink to write all metrics to</param>
+        /// <param name="metricSinkWriter">Writer to send metrics to all configured sinks</param>
         /// <param name="prometheusMetricWriter">Metrics collector for our Prometheus scraping endpoint</param>
         /// <param name="logger">General logger</param>
-        public ScraperConfiguration(AzureMonitorClient azureMonitorClient, IMetricSink sink, IPrometheusMetricWriter prometheusMetricWriter, ILogger logger)
+        public ScraperConfiguration(AzureMonitorClient azureMonitorClient, MetricSinkWriter metricSinkWriter, IPrometheusMetricWriter prometheusMetricWriter, ILogger logger)
         {
             Guard.NotNull(azureMonitorClient, nameof(azureMonitorClient));
             Guard.NotNull(prometheusMetricWriter, nameof(prometheusMetricWriter));
             Guard.NotNull(logger, nameof(logger));
-            Guard.NotNull(sink, nameof(sink));
+            Guard.NotNull(metricSinkWriter, nameof(metricSinkWriter));
 
             AzureMonitorClient = azureMonitorClient;
             PrometheusMetricWriter = prometheusMetricWriter;
             Logger = logger;
-            Sink = sink;
+            MetricSinkWriter = metricSinkWriter;
         }
     }
 }

src/Promitor.Core.Scraping/Sinks/IMetricSink.cs

@@ -1,9 +1,12 @@
-namespace Promitor.Core.Scraping.Sinks
+using System.Threading.Tasks;
+using Promitor.Integrations.AzureMonitor;
+
+namespace Promitor.Core.Scraping.Sinks
 {
     public interface IMetricSink
     {
-        MetricSinkType SinkType { get; }
+        MetricSinkType Type { get; }
 
-        void ReportMetric(string metricName, string metricDescription, ScrapeResult scrapedMetricResult);
+        Task ReportMetricAsync(string metricName, string metricDescription, MeasuredMetric measuredMetric);
     }
 }

src/Promitor.Core.Scraping/Sinks/MetricSinkWriter.cs

@@ -0,0 +1,60 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using GuardNet;
+using Microsoft.Extensions.Logging;
+
+namespace Promitor.Core.Scraping.Sinks
+{
+    public class MetricSinkWriter
+    {
+        private readonly List<IMetricSink> _configuredSinks;
+        private ILogger Logger { get; }
+
+        public MetricSinkWriter(IEnumerable<IMetricSink> configuredSinks, ILogger<MetricSinkWriter> logger)
+        {
+            var metricSinks = configuredSinks?.ToList();
+            Guard.NotNull(metricSinks, nameof(configuredSinks));
+            Guard.NotNull(logger, nameof(logger));
+
+            Logger = logger;
+            _configuredSinks = metricSinks;
+        }
+
+        public async Task ReportMetricAsync(string metricName, string metricDescription, ScrapeResult scrapedMetricResult)
+        {
+            Guard.NotNullOrWhitespace(metricName, nameof(metricName));
+            Guard.NotNull(scrapedMetricResult, nameof(scrapedMetricResult));
+
+            var reportTasks = new List<Task>();
+            foreach (var sink in _configuredSinks)
+            {
+                var reportTask = ReportMetricAsync(sink, metricName, metricDescription, scrapedMetricResult);
+                reportTasks.Add(reportTask);
+            }
+
+            await Task.WhenAll(reportTasks);
+        }
+
+        private async Task ReportMetricAsync(IMetricSink sink, string metricName, string metricDescription, ScrapeResult scrapedMetricResult)
+        {
+            Guard.NotNull(sink, nameof(sink));
+            Guard.NotNullOrWhitespace(metricName, nameof(metricName));
+            Guard.NotNull(scrapedMetricResult, nameof(scrapedMetricResult));
+            Guard.NotNull(scrapedMetricResult.MetricValues, nameof(scrapedMetricResult.MetricValues));
+
+            foreach (var measuredMetric in scrapedMetricResult.MetricValues)
+            {
+                try
+                {
+                    await sink.ReportMetricAsync(metricName, metricDescription, measuredMetric);
+                }
+                catch (Exception ex)
+                {
+                    Logger.LogCritical(ex, "Failed to write {MetricName} metric for sink {SinkType}", metricName, sink.Type);
+                }
+            }
+        }
+    }
+}

src/Promitor.Integrations.Sinks.Statsd/Promitor.Integrations.Sinks.Statsd.csproj

@@ -18,7 +18,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\Promitor.Integrations.Sinks.Core\Promitor.Integrations.Sinks.Core.csproj" />
+    <ProjectReference Include="..\Promitor.Core.Scraping\Promitor.Core.Scraping.csproj" />
   </ItemGroup>
 
 </Project>

src/Promitor.Integrations.Sinks.Statsd/StatsdMetricSink.cs

@@ -1,30 +1,39 @@
-using GuardNet;
+using System.Threading.Tasks;
+using GuardNet;
 using JustEat.StatsD;
 using Microsoft.Extensions.Logging;
 using Promitor.Core.Scraping.Sinks;
 using Promitor.Integrations.AzureMonitor;
-using Promitor.Integrations.Sinks.Core;
 
 namespace Promitor.Integrations.Sinks.Statsd
 {
-    public class StatsdMetricSink : MetricSink
+    public class StatsdMetricSink : IMetricSink
     {
+        private readonly ILogger<StatsdMetricSink> _logger;
         private readonly IStatsDPublisher _statsDPublisher;
 
         public StatsdMetricSink(IStatsDPublisher statsDPublisher, ILogger<StatsdMetricSink> logger)
-        : base(logger)
         {
             Guard.NotNull(statsDPublisher, nameof(statsDPublisher));
+            Guard.NotNull(logger, nameof(logger));
 
             _statsDPublisher = statsDPublisher;
+            _logger = logger;
         }
 
-        public override MetricSinkType SinkType => MetricSinkType.StatsD;
+        public MetricSinkType Type => MetricSinkType.StatsD;
 
-        public override void ReportMetric(string metricName, string metricDescription, MeasuredMetric measuredMetric)
+        public Task ReportMetricAsync(string metricName, string metricDescription, MeasuredMetric measuredMetric)
         {
+            Guard.NotNullOrEmpty(metricName, nameof(metricName));
+            Guard.NotNull(measuredMetric, nameof(measuredMetric));
+
             var metricValue = measuredMetric.Value ?? 0;
             _statsDPublisher.Gauge(metricValue, metricName);
+
+            _logger.LogTrace("Metric {MetricName} with value {MetricValue} was written to StatsD server", metricName, metricValue);
+
+            return Task.CompletedTask;
         }
     }
 }