From 58f5d38d6e507f538b2149b915ce3c96cca28d71 Mon Sep 17 00:00:00 2001 From: Nick Babcock Date: Sat, 15 Aug 2026 11:24:57 -0500 Subject: [PATCH] Add batch size configuration --- OhmGraphite.Test/ConfigTest.cs | 34 ++++++- OhmGraphite.Test/GraphiteTest.cs | 5 +- OhmGraphite.Test/InfluxTest.cs | 12 ++- OhmGraphite.Test/MetricTimerTest.cs | 139 ++++++++++++++++++++++++++ OhmGraphite.Test/TestSensorCreator.cs | 10 ++ OhmGraphite.Test/TimescaleTest.cs | 23 ++++- OhmGraphite/App.config | 3 +- OhmGraphite/GraphiteWriter.cs | 21 ++-- OhmGraphite/IWriteMetrics.cs | 2 +- OhmGraphite/Influx2Writer.cs | 5 +- OhmGraphite/InfluxWriter.cs | 5 +- OhmGraphite/MetricConfig.cs | 14 ++- OhmGraphite/MetricReport.cs | 7 ++ OhmGraphite/MetricTimer.cs | 122 ++++++++++++++++++---- OhmGraphite/TimescaleWriter.cs | 91 ++++++++--------- OhmGraphite/Worker.cs | 19 ++-- README.md | 8 ++ assets/graphite.config | 3 +- 18 files changed, 413 insertions(+), 110 deletions(-) create mode 100644 OhmGraphite.Test/MetricTimerTest.cs create mode 100644 OhmGraphite/MetricReport.cs diff --git a/OhmGraphite.Test/ConfigTest.cs b/OhmGraphite.Test/ConfigTest.cs index 467394b..158377e 100644 --- a/OhmGraphite.Test/ConfigTest.cs +++ b/OhmGraphite.Test/ConfigTest.cs @@ -20,6 +20,7 @@ public void CanParseGraphiteConfig() Assert.Equal("myhost", results.Graphite.Host); Assert.Equal(2004, results.Graphite.Port); Assert.Equal(TimeSpan.FromSeconds(6), results.Interval); + Assert.Equal(3, results.BatchSize); Assert.True(results.Graphite.Tags); } @@ -36,6 +37,7 @@ public void CanParseNullConfig() Assert.Equal("localhost", results.Graphite.Host); Assert.Equal(2003, results.Graphite.Port); Assert.Equal(TimeSpan.FromSeconds(5), results.Interval); + Assert.Equal(1, results.BatchSize); Assert.False(results.Graphite.Tags); Assert.True(results.EnabledHardware.Cpu); @@ -47,6 +49,36 @@ public void CanParseNullConfig() Assert.True(results.EnabledHardware.Storage); } + [Theory] + [InlineData("invalid")] + [InlineData("0")] + [InlineData("-1")] + public void InvalidBatchSizeUsesDefault(string batchSize) + { + var configMap = new ExeConfigurationFileMap { ExeConfigFilename = "assets/default.config" }; + var config = ConfigurationManager.OpenMappedExeConfiguration(configMap, ConfigurationUserLevel.None); + config.AppSettings.Settings.Add("batch_size", batchSize); + + var results = MetricConfig.ParseAppSettings(new CustomConfig(config)); + + Assert.Equal(1, results.BatchSize); + } + + [Theory] + [InlineData("invalid")] + [InlineData("0")] + [InlineData("-1")] + public void InvalidIntervalUsesDefault(string interval) + { + var configMap = new ExeConfigurationFileMap { ExeConfigFilename = "assets/default.config" }; + var config = ConfigurationManager.OpenMappedExeConfiguration(configMap, ConfigurationUserLevel.None); + config.AppSettings.Settings.Add("interval", interval); + + var results = MetricConfig.ParseAppSettings(new CustomConfig(config)); + + Assert.Equal(TimeSpan.FromSeconds(5), results.Interval); + } + [Fact] public void CanParseInfluxDbConfig() { @@ -197,4 +229,4 @@ public void CanInstallCertificateVerification() ServicePointManager.ServerCertificateValidationCallback = null; } } -} \ No newline at end of file +} diff --git a/OhmGraphite.Test/GraphiteTest.cs b/OhmGraphite.Test/GraphiteTest.cs index dc23452..5d2ad9b 100644 --- a/OhmGraphite.Test/GraphiteTest.cs +++ b/OhmGraphite.Test/GraphiteTest.cs @@ -26,11 +26,12 @@ public async Task InsertGraphiteTest() var port = container.GetMappedPublicPort(2003); using var writer = new GraphiteWriter(container.Hostname, port, "my-pc", tags: false); using var client = new HttpClient(); + var reportTime = DateTime.Now; for (int attempts = 0; ; attempts++) { try { - await writer.ReportMetrics(DateTime.Now, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(reportTime.AddSeconds(-1), reportTime)); var resp = await client.GetAsync( $"http://{container.Hostname}:{container.GetMappedPublicPort(80)}/render?format=csv&target=ohm.my-pc.intelcpu.0.temperature.cpucore.1", @@ -72,7 +73,7 @@ public async Task InsertTagGraphiteTest() { try { - await writer.ReportMetrics(DateTime.Now, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(DateTime.Now)); var resp = await client.GetAsync( $"http://{container.Hostname}:{container.GetMappedPublicPort(80)}/render?format=csv&target=seriesByTag('sensor_type=Temperature','hardware_type=CPU')", cancellationToken); diff --git a/OhmGraphite.Test/InfluxTest.cs b/OhmGraphite.Test/InfluxTest.cs index 306544c..288470c 100644 --- a/OhmGraphite.Test/InfluxTest.cs +++ b/OhmGraphite.Test/InfluxTest.cs @@ -31,11 +31,12 @@ public async Task CanInsertIntoInflux() var config = new InfluxConfig(new Uri(baseUrl), "mydb", "my_user", "my_pass"); using var writer = new InfluxWriter(config, "my-pc"); using var client = new HttpClient(); + var reportTime = DateTime.Now; for (int attempts = 0; ; attempts++) { try { - await writer.ReportMetrics(DateTime.Now, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(reportTime.AddSeconds(-1), reportTime)); var resp = await client.GetAsync( $"{baseUrl}/query?pretty=true&db=mydb&q=SELECT%20*%20FROM%20Temperature", @@ -79,7 +80,7 @@ public async Task CanInsertIntoPasswordLessInfluxdb() { try { - await writer.ReportMetrics(DateTime.Now, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(DateTime.Now)); var resp = await client.GetAsync( $"{baseUrl}/query?pretty=true&db=mydb&q=SELECT%20*%20FROM%20Temperature", @@ -128,11 +129,12 @@ public async Task CanInsertIntoInflux2() var config = new Influx2Config(options); using var writer = new Influx2Writer(config, "my-pc"); + var reportTime = DateTime.Now; for (int attempts = 0; ; attempts++) { try { - await writer.ReportMetrics(DateTime.Now, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(reportTime.AddSeconds(-1), reportTime)); var influxDBClient = new InfluxDBClient(options); var flux = "from(bucket:\"mydb\") |> range(start: -1h)"; var queryApi = influxDBClient.GetQueryApi(); @@ -183,7 +185,7 @@ public async Task CanInsertIntoInflux2Token() { try { - await writer.ReportMetrics(DateTime.Now, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(DateTime.Now)); var influxDBClient = new InfluxDBClient(results.Influx2.Options); var flux = "from(bucket:\"mydb\") |> range(start: -1h)"; var queryApi = influxDBClient.GetQueryApi(); @@ -251,7 +253,7 @@ public async Task CanInsertIntoInflux2TokenTls() { try { - await writer.ReportMetrics(DateTime.Now, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(DateTime.Now)); var influxDbClient = new InfluxDBClient(results.Influx2.Options); var flux = "from(bucket:\"mydb\") |> range(start: -1h)"; var queryApi = influxDbClient.GetQueryApi(); diff --git a/OhmGraphite.Test/MetricTimerTest.cs b/OhmGraphite.Test/MetricTimerTest.cs new file mode 100644 index 0000000..1f4dbb9 --- /dev/null +++ b/OhmGraphite.Test/MetricTimerTest.cs @@ -0,0 +1,139 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Xunit; + +namespace OhmGraphite.Test +{ + public class MetricTimerTest + { + [Fact] + public async Task SendsReportsWhenBatchIsFull() + { + var collector = new RecordingCollector(); + var writer = new RecordingWriter(); + using var timer = new MetricTimer(TimeSpan.FromHours(1), 3, collector, writer); + var times = new[] + { + new DateTime(2026, 1, 1, 0, 0, 1, DateTimeKind.Utc), + new DateTime(2026, 1, 1, 0, 0, 2, DateTimeKind.Utc), + new DateTime(2026, 1, 1, 0, 0, 3, DateTimeKind.Utc), + }; + + await timer.CollectMetrics(times[0]); + await timer.CollectMetrics(times[1]); + + Assert.Empty(writer.Batches); + Assert.Equal(2, collector.ReadCount); + + await timer.CollectMetrics(times[2]); + + var batch = Assert.Single(writer.Batches); + Assert.Equal(times, batch.Select(x => x.ReportTime)); + Assert.All(batch, report => Assert.Equal(3, report.Sensors.Count)); + Assert.Equal(3, collector.ReadCount); + } + + [Fact] + public async Task BatchSizeOneSendsImmediately() + { + var writer = new RecordingWriter(); + using var timer = new MetricTimer(TimeSpan.FromHours(1), 1, new RecordingCollector(), writer); + + await timer.CollectMetrics(DateTime.UtcNow); + + Assert.Single(writer.Batches); + } + + [Fact] + public async Task FailedBatchIsDropped() + { + var writer = new RecordingWriter(failures: 1); + using var timer = new MetricTimer(TimeSpan.FromHours(1), 2, new RecordingCollector(), writer); + var first = new DateTime(2026, 1, 1, 0, 0, 1, DateTimeKind.Utc); + var second = first.AddSeconds(1); + var third = second.AddSeconds(1); + var fourth = third.AddSeconds(1); + + await timer.CollectMetrics(first); + await timer.CollectMetrics(second); + await timer.CollectMetrics(third); + + Assert.Equal(1, writer.Attempts); + Assert.Empty(writer.Batches); + + await timer.CollectMetrics(fourth); + + Assert.Equal(2, writer.Attempts); + var batch = Assert.Single(writer.Batches); + Assert.Equal(new[] { third, fourth }, batch.Select(x => x.ReportTime)); + } + + [Fact] + public async Task DisposeFlushesPartialBatch() + { + var writer = new RecordingWriter(); + var timer = new MetricTimer(TimeSpan.FromHours(1), 3, new RecordingCollector(), writer); + var reportTime = new DateTime(2026, 1, 1, 0, 0, 1, DateTimeKind.Utc); + await timer.CollectMetrics(reportTime); + + timer.Dispose(); + + var batch = Assert.Single(writer.Batches); + Assert.Equal(reportTime, Assert.Single(batch).ReportTime); + Assert.True(writer.IsDisposed); + } + + private sealed class RecordingCollector : IGiveSensors + { + public int ReadCount { get; private set; } + + public IEnumerable ReadAllSensors() + { + ReadCount++; + return TestSensorCreator.Values(); + } + + public void Start() + { + } + + public void Dispose() + { + } + } + + private sealed class RecordingWriter : IWriteMetrics + { + private int _failures; + + public RecordingWriter(int failures = 0) + { + _failures = failures; + } + + public int Attempts { get; private set; } + public List> Batches { get; } = new List>(); + public bool IsDisposed { get; private set; } + + public Task ReportMetrics(IEnumerable reports) + { + Attempts++; + if (_failures > 0) + { + _failures--; + throw new InvalidOperationException("Test failure"); + } + + Batches.Add(reports.ToList()); + return Task.CompletedTask; + } + + public void Dispose() + { + IsDisposed = true; + } + } + } +} diff --git a/OhmGraphite.Test/TestSensorCreator.cs b/OhmGraphite.Test/TestSensorCreator.cs index 6fe56fd..0600ebc 100644 --- a/OhmGraphite.Test/TestSensorCreator.cs +++ b/OhmGraphite.Test/TestSensorCreator.cs @@ -1,4 +1,6 @@ +using System; using System.Collections.Generic; +using System.Linq; namespace OhmGraphite.Test { @@ -11,6 +13,14 @@ public static IEnumerable Values() yield return new ReportedValue("/intelcpu/0/temperature/2", "CPU Core #3", 10, SensorType.Temperature, "Intel Core i7-6700K", HardwareType.CPU, "0", 2); } + public static IEnumerable Reports(params DateTime[] reportTimes) + { + foreach (var reportTime in reportTimes) + { + yield return new MetricReport(reportTime, Values().ToList()); + } + } + public IEnumerable ReadAllSensors() => Values(); public void Start() diff --git a/OhmGraphite.Test/TimescaleTest.cs b/OhmGraphite.Test/TimescaleTest.cs index 64175fe..2d2c580 100644 --- a/OhmGraphite.Test/TimescaleTest.cs +++ b/OhmGraphite.Test/TimescaleTest.cs @@ -1,4 +1,5 @@ using System; +using System.Linq; using System.Threading.Tasks; using DotNet.Testcontainers.Builders; using Npgsql; @@ -27,11 +28,25 @@ public async Task CanSetupTimescale() using var writer = new TimescaleWriter(connStr, true, "my-pc"); await using var conn = new NpgsqlConnection(connStr); - await writer.ReportMetrics(epoch, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(epoch, epoch.AddSeconds(1))); conn.Open(); - await using var cmd = new NpgsqlCommand("SELECT COUNT(*) FROM ohm_stats", conn); - Assert.Equal(3, Convert.ToInt32(cmd.ExecuteScalar())); + await using (var cmd = new NpgsqlCommand("SELECT COUNT(*), COUNT(DISTINCT time) FROM ohm_stats", conn)) + await using (var reader = cmd.ExecuteReader()) + { + Assert.True(reader.Read()); + Assert.Equal(6, reader.GetInt64(0)); + Assert.Equal(2, reader.GetInt64(1)); + } + + // A large batch must stay below the Postgres limit of 65535 parameters per + // statement. At 9 parameters per sensor, a single statement can only hold 7281 + // sensors, which a batch of reports can easily exceed. + var many = Enumerable.Range(1, 3000).Select(x => epoch.AddSeconds(x)).ToArray(); + await writer.ReportMetrics(TestSensorCreator.Reports(many)); + + await using var largeCmd = new NpgsqlCommand("SELECT COUNT(*) FROM ohm_stats", conn); + Assert.Equal(9006, Convert.ToInt32(largeCmd.ExecuteScalar())); } [IgnoreOnRemoteDockerFact, Trait("Category", "integration")] @@ -63,7 +78,7 @@ public async Task InsertOnlyTimescale() string connStr = $"Host={container.Hostname};Username=ohm;Password=itsohm;Port={container.GetMappedPublicPort(5432)};Database=timescale_built"; using var writer = new TimescaleWriter(connStr, false, "my-pc"); await using var conn = new NpgsqlConnection(selectStr); - await writer.ReportMetrics(epoch, TestSensorCreator.Values()); + await writer.ReportMetrics(TestSensorCreator.Reports(epoch)); conn.Open(); await using var cmd = new NpgsqlCommand("SELECT COUNT(*) FROM ohm_stats", conn); diff --git a/OhmGraphite/App.config b/OhmGraphite/App.config index 570f614..8fcc911 100644 --- a/OhmGraphite/App.config +++ b/OhmGraphite/App.config @@ -4,5 +4,6 @@ + - \ No newline at end of file + diff --git a/OhmGraphite/GraphiteWriter.cs b/OhmGraphite/GraphiteWriter.cs index 3141cad..44db3e1 100644 --- a/OhmGraphite/GraphiteWriter.cs +++ b/OhmGraphite/GraphiteWriter.cs @@ -31,7 +31,7 @@ public GraphiteWriter(string remoteHost, int remotePort, string localHost, bool _localHost = localHost; } - public async Task ReportMetrics(DateTime reportTime, IEnumerable sensors) + public async Task ReportMetrics(IEnumerable reports) { // Since the graphite writer keeps the same connection open across // writes, we need to ensure that only one thread has access to @@ -47,7 +47,7 @@ public async Task ReportMetrics(DateTime reportTime, IEnumerable try { - await SendGraphite(reportTime, sensors); + await SendGraphite(reports); } finally { @@ -55,7 +55,7 @@ public async Task ReportMetrics(DateTime reportTime, IEnumerable } } - private async Task SendGraphite(DateTime reportTime, IEnumerable sensors) + private async Task SendGraphite(IEnumerable reports) { try { @@ -69,11 +69,6 @@ private async Task SendGraphite(DateTime reportTime, IEnumerable await _client.ConnectAsync(_remoteHost, _remotePort); } - // We don't want to transmit metrics across multiple seconds as they - // are being retrieved so calculate the timestamp of the signaled event - // only once. - long epoch = new DateTimeOffset(reportTime).ToUnixTimeSeconds(); - // Create a stream writer that leaves the underlying stream open // when the writer is closed, as we don't want our TCP connection // closed too. Since this requires the four param constructor for @@ -81,9 +76,13 @@ private async Task SendGraphite(DateTime reportTime, IEnumerable // the C# reference source. using (var writer = new StreamWriter(_client.GetStream(), Utf8NoBom, bufferSize: 1024, leaveOpen: true)) { - foreach (var sensor in sensors) + foreach (var report in reports) { - await writer.WriteLineAsync(FormatGraphiteData(epoch, sensor)); + var epoch = new DateTimeOffset(report.ReportTime).ToUnixTimeSeconds(); + foreach (var sensor in report.Sensors) + { + await writer.WriteLineAsync(FormatGraphiteData(epoch, sensor)); + } } } @@ -158,4 +157,4 @@ public void Dispose() _client?.Dispose(); } } -} \ No newline at end of file +} diff --git a/OhmGraphite/IWriteMetrics.cs b/OhmGraphite/IWriteMetrics.cs index 2485012..f0dfc04 100644 --- a/OhmGraphite/IWriteMetrics.cs +++ b/OhmGraphite/IWriteMetrics.cs @@ -6,6 +6,6 @@ namespace OhmGraphite { public interface IWriteMetrics : IDisposable { - Task ReportMetrics(DateTime reportTime, IEnumerable sensors); + Task ReportMetrics(IEnumerable reports); } } diff --git a/OhmGraphite/Influx2Writer.cs b/OhmGraphite/Influx2Writer.cs index 0a502de..9612219 100644 --- a/OhmGraphite/Influx2Writer.cs +++ b/OhmGraphite/Influx2Writer.cs @@ -22,11 +22,12 @@ public Influx2Writer(Influx2Config config, string localHost) _localHost = localHost; } - public async Task ReportMetrics(DateTime reportTime, IEnumerable sensors) + public async Task ReportMetrics(IEnumerable reports) { + var points = reports.SelectMany(report => + report.Sensors.Select(sensor => NewPoint(report.ReportTime, sensor))).ToList(); var influxDbClient = new InfluxDBClient(_config.Options); var writeApi = influxDbClient.GetWriteApiAsync(); - var points = sensors.Select(x => NewPoint(reportTime, x)).ToList(); await writeApi.WritePointsAsync(points); } diff --git a/OhmGraphite/InfluxWriter.cs b/OhmGraphite/InfluxWriter.cs index 8d72383..649edaf 100644 --- a/OhmGraphite/InfluxWriter.cs +++ b/OhmGraphite/InfluxWriter.cs @@ -21,13 +21,14 @@ public InfluxWriter(InfluxConfig config, string localHost) _localHost = localHost; } - public async Task ReportMetrics(DateTime reportTime, IEnumerable sensors) + public async Task ReportMetrics(IEnumerable reports) { var payload = new LineProtocolPayload(); var password = _config.User != null ? (_config.Password ?? "") : null; var client = new LineProtocolClient(_config.Address, _config.Db, _config.User, password); - foreach (var point in sensors.Select(x => NewPoint(reportTime, x))) + foreach (var point in reports.SelectMany(report => + report.Sensors.Select(sensor => NewPoint(report.ReportTime, sensor)))) { payload.Add(point); } diff --git a/OhmGraphite/MetricConfig.cs b/OhmGraphite/MetricConfig.cs index 1366601..d3cf148 100644 --- a/OhmGraphite/MetricConfig.cs +++ b/OhmGraphite/MetricConfig.cs @@ -15,11 +15,12 @@ public class MetricConfig { private readonly INameResolution _nameLookup; - public MetricConfig(TimeSpan interval, INameResolution nameLookup, GraphiteConfig graphite, InfluxConfig influx, + public MetricConfig(TimeSpan interval, int batchSize, INameResolution nameLookup, GraphiteConfig graphite, InfluxConfig influx, PrometheusConfig prometheus, TimescaleConfig timescale, Dictionary aliases, List hiddenSensors, Influx2Config influx2, EnabledHardware enabledHardware) { _nameLookup = nameLookup; Interval = interval; + BatchSize = batchSize; Graphite = graphite; Influx = influx; Prometheus = prometheus; @@ -32,6 +33,7 @@ public MetricConfig(TimeSpan interval, INameResolution nameLookup, GraphiteConfi public string LookupName() => _nameLookup.LookupName(); public TimeSpan Interval { get; } + public int BatchSize { get; } public GraphiteConfig Graphite { get; } public InfluxConfig Influx { get; } public Influx2Config Influx2 { get; } @@ -43,12 +45,16 @@ public MetricConfig(TimeSpan interval, INameResolution nameLookup, GraphiteConfi public static MetricConfig ParseAppSettings(IAppConfig config) { - if (!int.TryParse(config["interval"], out int seconds)) + if (!int.TryParse(config["interval"], out int seconds) || seconds <= 0) { seconds = 5; } var interval = TimeSpan.FromSeconds(seconds); + if (!int.TryParse(config["batch_size"], out int batchSize) || batchSize <= 0) + { + batchSize = 1; + } INameResolution nameLookup = NameLookup(config["name_lookup"] ?? "netbios"); InstallCertificateVerification(config["certificate_verification"] ?? "True"); @@ -102,7 +108,7 @@ public static MetricConfig ParseAppSettings(IAppConfig config) RegexOptions.IgnoreCase | RegexOptions.Singleline )).ToList(); - return new MetricConfig(interval, nameLookup, gconfig, iconfig, pconfig, timescale, aliases, hiddenSensors, influx2, enabledHardware); + return new MetricConfig(interval, batchSize, nameLookup, gconfig, iconfig, pconfig, timescale, aliases, hiddenSensors, influx2, enabledHardware); } private static EnabledHardware ParseEnabledHardware(IAppConfig config) @@ -163,4 +169,4 @@ public static void InstallCertificateVerification(string type) public bool TryGetAlias(string v, out string alias) => Aliases.TryGetValue(v, out alias); public bool IsHidden(string id) => HiddenSensors.Any(x => x.IsMatch(id)); } -} \ No newline at end of file +} diff --git a/OhmGraphite/MetricReport.cs b/OhmGraphite/MetricReport.cs new file mode 100644 index 0000000..e8991d9 --- /dev/null +++ b/OhmGraphite/MetricReport.cs @@ -0,0 +1,7 @@ +using System; +using System.Collections.Generic; + +namespace OhmGraphite +{ + public record MetricReport(DateTime ReportTime, IReadOnlyList Sensors); +} diff --git a/OhmGraphite/MetricTimer.cs b/OhmGraphite/MetricTimer.cs index b2c73a9..ecf644a 100644 --- a/OhmGraphite/MetricTimer.cs +++ b/OhmGraphite/MetricTimer.cs @@ -1,7 +1,9 @@ using System; +using System.Collections.Generic; using System.Diagnostics; using System.Linq; -using System.Timers; +using System.Threading; +using System.Threading.Tasks; using NLog; namespace OhmGraphite @@ -9,55 +11,139 @@ namespace OhmGraphite public class MetricTimer : IManage { private static readonly Logger Logger = LogManager.GetCurrentClassLogger(); + private static readonly TimeSpan ShutdownTimeout = TimeSpan.FromSeconds(5); + private readonly int _batchSize; private readonly IGiveSensors _collector; - - private readonly Timer _timer; + private readonly TimeSpan _interval; + private readonly List _reports = new List(); private readonly IWriteMetrics _writer; + private CancellationTokenSource _cancellation; + private bool _disposed; + private Task _reportTask; - public MetricTimer(TimeSpan interval, IGiveSensors collector, IWriteMetrics writer) + public MetricTimer(TimeSpan interval, int batchSize, IGiveSensors collector, IWriteMetrics writer) { - _timer = new Timer(interval.TotalMilliseconds) { AutoReset = true }; - _timer.Elapsed += ReportMetrics; + if (interval <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException(nameof(interval)); + } + + if (batchSize <= 0) + { + throw new ArgumentOutOfRangeException(nameof(batchSize)); + } + + _interval = interval; + _batchSize = batchSize; _collector = collector; _writer = writer; } public void Start() { + if (_disposed) + { + throw new ObjectDisposedException(nameof(MetricTimer)); + } + + if (_reportTask != null) + { + return; + } + Logger.LogAction("starting metric timer", () => { _collector.Start(); - _timer.Start(); + _cancellation = new CancellationTokenSource(); + _reportTask = ReportOnInterval(_cancellation.Token); }); } - private async void ReportMetrics(object sender, ElapsedEventArgs e) + private async Task ReportOnInterval(CancellationToken stoppingToken) + { + using var timer = new PeriodicTimer(_interval); + try + { + while (await timer.WaitForNextTickAsync(stoppingToken)) + { + await CollectMetrics(DateTime.Now); + } + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + } + finally + { + await FlushMetrics(); + } + } + + internal async Task CollectMetrics(DateTime reportTime) { - Logger.Debug("Starting to report metrics"); + Logger.Debug("Starting to collect metrics"); try { - // Read all the sensors into a list so that they are only polled once. - // Polling sensors can be relatively expensive so the intermediate - // list cuts down on the number of potential updates. - var stopwatch = Stopwatch.StartNew(); + // Read all sensors into a list so that each sensor is polled one time. var sensors = _collector.ReadAllSensors().ToList(); - await _writer.ReportMetrics(e.SignalTime, sensors); - Logger.Info($"Sent {sensors.Count} metrics in {stopwatch.Elapsed.TotalMilliseconds}ms"); + _reports.Add(new MetricReport(reportTime, sensors)); + if (_reports.Count >= _batchSize) + { + await FlushMetrics(); + } + } + catch (Exception ex) + { + Logger.Error(ex, "Unable to collect metrics"); + } + } + + private async Task FlushMetrics() + { + if (_reports.Count == 0) + { + return; + } + + var reports = _reports.ToList(); + _reports.Clear(); + var metricCount = reports.Sum(x => x.Sensors.Count); + var stopwatch = Stopwatch.StartNew(); + try + { + await _writer.ReportMetrics(reports); + Logger.Info($"Sent {metricCount} metrics from {reports.Count} reports in {stopwatch.Elapsed.TotalMilliseconds}ms"); } catch (Exception ex) { - Logger.Error(ex, "Unable to send metrics"); + Logger.Error(ex, $"Unable to send {metricCount} metrics from {reports.Count} reports"); } } public void Dispose() { + if (_disposed) + { + return; + } + + _disposed = true; Logger.LogAction("stopping metric timer", () => { + _cancellation?.Cancel(); + + if (_reportTask?.Wait(ShutdownTimeout) ?? true) + { + FlushMetrics().GetAwaiter().GetResult(); + } + else + { + Logger.Warn($"Metric timer did not stop within {ShutdownTimeout.TotalSeconds}s"); + } + + _cancellation?.Dispose(); _writer?.Dispose(); _collector?.Dispose(); - _timer?.Dispose(); }); } } -} \ No newline at end of file +} diff --git a/OhmGraphite/TimescaleWriter.cs b/OhmGraphite/TimescaleWriter.cs index 7b95a35..1f2d418 100644 --- a/OhmGraphite/TimescaleWriter.cs +++ b/OhmGraphite/TimescaleWriter.cs @@ -26,14 +26,10 @@ public TimescaleWriter(string connStr, bool setupTable, string localHost) _setupTable = setupTable; } - public Task ReportMetrics(DateTime reportTime, IEnumerable sensors) + public Task ReportMetrics(IEnumerable reports) { try { - // "timestamp with time zone" postgres type is a UTC timestamp so - // we explicitly convert the reported time to UTC to avoid a cast - // exception by npgsql - reportTime = reportTime.ToUniversalTime(); if (_failure) { Logger.Debug("Clearing connection pool"); @@ -96,46 +92,21 @@ public Task ReportMetrics(DateTime reportTime, IEnumerable sensor } } - var values = sensors.ToList(); - using (var cmd = new NpgsqlCommand(BatchedInsertSql(values), conn)) + // Every report becomes its own statement, but they are all sent in a + // single batch, so the whole flush is one round trip and one commit. + using (var transaction = conn.BeginTransaction()) + using (var batch = new NpgsqlBatch(conn, transaction)) { - // Note that all parameters must be set before calling Prepare() - // they are part of the information transmitted to PostgreSQL - // and used to effectively plan the statement. You must also set - // the DbType or NpgsqlDbType on your parameters to unambiguously - // specify the data type (setting the value isn't support) - for (int i = 0; i < values.Count; i++) + foreach (var report in reports) { - cmd.Parameters.Add($"time{i}", NpgsqlDbType.TimestampTz); - cmd.Parameters.Add($"host{i}", NpgsqlDbType.Text); - cmd.Parameters.Add($"hardware{i}", NpgsqlDbType.Text); - cmd.Parameters.Add($"hardware_type{i}", NpgsqlDbType.Text); - cmd.Parameters.Add($"identifier{i}", NpgsqlDbType.Text); - cmd.Parameters.Add($"sensor{i}", NpgsqlDbType.Text); - cmd.Parameters.Add($"sensor_type{i}", NpgsqlDbType.Text); - cmd.Parameters.Add($"value{i}", NpgsqlDbType.Real); - cmd.Parameters.Add($"sensor_index{i}", NpgsqlDbType.Integer); + batch.BatchCommands.Add(InsertCommand(report)); } // A majority of the time, the same number of sensors will be - // reported on, so it's important to prepare the statement - cmd.Prepare(); - - for (int i = 0; i < values.Count; i++) - { - var sensor = values[i]; - cmd.Parameters[$"time{i}"].Value = reportTime; - cmd.Parameters[$"host{i}"].Value = _localHost; - cmd.Parameters[$"hardware{i}"].Value = sensor.Hardware; - cmd.Parameters[$"hardware_type{i}"].Value = Enum.GetName(typeof(HardwareType), sensor.HardwareType); - cmd.Parameters[$"identifier{i}"].Value = sensor.Identifier; - cmd.Parameters[$"sensor{i}"].Value = sensor.Sensor; - cmd.Parameters[$"sensor_type{i}"].Value = Enum.GetName(typeof(SensorType), sensor.SensorType); - cmd.Parameters[$"value{i}"].Value = sensor.Value; - cmd.Parameters[$"sensor_index{i}"].Value = sensor.SensorIndex; - } - - cmd.ExecuteNonQuery(); + // reported on, so it's important to prepare the statements + batch.Prepare(); + batch.ExecuteNonQuery(); + transaction.Commit(); } _failure = false; @@ -152,22 +123,42 @@ public Task ReportMetrics(DateTime reportTime, IEnumerable sensor } } - // Returns a SQL INSERT statement that will insert all the reported values in one go. - // Since there is no batched insert API that is part of Npgsql, we simulate one by - // creating a unique set of sql parameters for each reported value by it's index. - // Sending one insert of 70 values was nearly 10x faster than 70 inserts of 1 value, - // so this circumnavigation around a lack of native batched insert statements is - // worth it. - private static string BatchedInsertSql(IEnumerable values) + // Returns a statement that inserts every sensor of a report in one go. + private NpgsqlBatchCommand InsertCommand(MetricReport report) { - var sqlColumns = values.Select((x, i) => + var sensors = report.Sensors; + var reportTime = report.ReportTime.ToUniversalTime(); + + var sqlColumns = sensors.Select((x, i) => $"(@time{i}, @host{i}, @hardware{i}, @hardware_type{i}, @identifier{i}, @sensor{i}, @sensor_type{i}, @sensor_index{i}, @value{i})"); var columns = string.Join(", ", sqlColumns); - return "INSERT INTO ohm_stats " + + var cmd = new NpgsqlBatchCommand("INSERT INTO ohm_stats " + "(time, host, hardware, hardware_type, identifier, sensor, sensor_type, sensor_index, value) VALUES " + - columns; + columns); + + // You must set the DbType or NpgsqlDbType on the parameters to unambiguously + // specify the data type, as the type is part of the information transmitted to + // PostgreSQL and used to effectively plan the statement. + for (int i = 0; i < sensors.Count; i++) + { + var sensor = sensors[i]; + cmd.Parameters.Add(Param($"time{i}", NpgsqlDbType.TimestampTz, reportTime)); + cmd.Parameters.Add(Param($"host{i}", NpgsqlDbType.Text, _localHost)); + cmd.Parameters.Add(Param($"hardware{i}", NpgsqlDbType.Text, sensor.Hardware)); + cmd.Parameters.Add(Param($"hardware_type{i}", NpgsqlDbType.Text, Enum.GetName(typeof(HardwareType), sensor.HardwareType))); + cmd.Parameters.Add(Param($"identifier{i}", NpgsqlDbType.Text, sensor.Identifier)); + cmd.Parameters.Add(Param($"sensor{i}", NpgsqlDbType.Text, sensor.Sensor)); + cmd.Parameters.Add(Param($"sensor_type{i}", NpgsqlDbType.Text, Enum.GetName(typeof(SensorType), sensor.SensorType))); + cmd.Parameters.Add(Param($"sensor_index{i}", NpgsqlDbType.Integer, sensor.SensorIndex)); + cmd.Parameters.Add(Param($"value{i}", NpgsqlDbType.Real, sensor.Value)); + } + + return cmd; } + private static NpgsqlParameter Param(string name, NpgsqlDbType type, object value) => + new NpgsqlParameter(name, type) { Value = value }; + public void Dispose() { NpgsqlConnection.ClearPool(new NpgsqlConnection(_connStr)); diff --git a/OhmGraphite/Worker.cs b/OhmGraphite/Worker.cs index dac9ff4..0cda765 100644 --- a/OhmGraphite/Worker.cs +++ b/OhmGraphite/Worker.cs @@ -49,16 +49,19 @@ private static IManage CreateOhmGraphite(MetricConfig config) private static IManage CreateManager(MetricConfig config, SensorCollector collector) { var hostname = config.LookupName(); - double seconds = config.Interval.TotalSeconds; + if (config.Prometheus == null) + { + Logger.Info($"Metric interval: {config.Interval.TotalSeconds}s batch size: {config.BatchSize}"); + } + if (config.Graphite != null) { - Logger.Info( - $"Graphite host: {config.Graphite.Host} port: {config.Graphite.Port} interval: {seconds} tags: {config.Graphite.Tags}"); + Logger.Info($"Graphite host: {config.Graphite.Host} port: {config.Graphite.Port} tags: {config.Graphite.Tags}"); var writer = new GraphiteWriter(config.Graphite.Host, config.Graphite.Port, hostname, config.Graphite.Tags); - return new MetricTimer(config.Interval, collector, writer); + return new MetricTimer(config.Interval, config.BatchSize, collector, writer); } else if (config.Prometheus != null) { @@ -70,20 +73,20 @@ private static IManage CreateManager(MetricConfig config, SensorCollector collec else if (config.Timescale != null) { var writer = new TimescaleWriter(config.Timescale.Connection, config.Timescale.SetupTable, hostname); - return new MetricTimer(config.Interval, collector, writer); + return new MetricTimer(config.Interval, config.BatchSize, collector, writer); } else if (config.Influx != null) { Logger.Info($"Influxdb address: {config.Influx.Address} db: {config.Influx.Db}"); var writer = new InfluxWriter(config.Influx, hostname); - return new MetricTimer(config.Interval, collector, writer); + return new MetricTimer(config.Interval, config.BatchSize, collector, writer); } else { Logger.Info($"Influx2 address: {config.Influx2.Options.Url}"); var writer = new Influx2Writer(config.Influx2, hostname); - return new MetricTimer(config.Interval, collector, writer); + return new MetricTimer(config.Interval, config.BatchSize, collector, writer); } } } -} \ No newline at end of file +} diff --git a/README.md b/README.md index 3a5c674..a328eeb 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,8 @@ Any `value` in the config can reference an environment variable using `%NAME%` s When running as a Windows service, set the variable machine-wide (e.g. via `setx /M INFLUX_TOKEN ...` or System Properties → Environment Variables) so the service account inherits it; per-user variables won't be visible to `LocalSystem`. Unset references are left as the literal `%NAME%` placeholder (standard Windows behavior). +For data connectors that push metrics to a destination, `batch_size` can be used to batch `n` number of metric reports. So if the interval is 5 seconds, and the batch size is 12, OhmGraphite will write data once a minute. + ### Graphite Configuration The config below polls our hardware every `5` seconds and sends the results to a graphite server listening on `localhost:2003`. @@ -75,6 +77,7 @@ The config below polls our hardware every `5` seconds and sends the results to a + @@ -103,12 +106,14 @@ Graphite is the default export style, but if you're an InfluxDB user you can cha + @@ -126,6 +131,7 @@ If OhmGraphite will be connecting to InfluxDB 2, the configuration will need to + ``` @@ -242,6 +248,8 @@ One can configure OhmGraphite to send to Timescale / Postgres with the following + + diff --git a/assets/graphite.config b/assets/graphite.config index b90edc8..1fa5e8c 100644 --- a/assets/graphite.config +++ b/assets/graphite.config @@ -4,6 +4,7 @@ + - \ No newline at end of file +