Add batch size configuration

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;
         }
     }
-}
+}

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);

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();

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<ReportedValue> 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<IReadOnlyList<MetricReport>> Batches { get; } = new List<IReadOnlyList<MetricReport>>();
+            public bool IsDisposed { get; private set; }
+
+            public Task ReportMetrics(IEnumerable<MetricReport> reports)
+            {
+                Attempts++;
+                if (_failures > 0)
+                {
+                    _failures--;
+                    throw new InvalidOperationException("Test failure");
+                }
+
+                Batches.Add(reports.ToList());
+                return Task.CompletedTask;
+            }
+
+            public void Dispose()
+            {
+                IsDisposed = true;
+            }
+        }
+    }
+}

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<ReportedValue> 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<MetricReport> Reports(params DateTime[] reportTimes)
+        {
+            foreach (var reportTime in reportTimes)
+            {
+                yield return new MetricReport(reportTime, Values().ToList());
+            }
+        }
+
         public IEnumerable<ReportedValue> ReadAllSensors() => Values();
 
         public void Start()

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);

OhmGraphite/App.config

@@ -4,5 +4,6 @@
     <add key="host" value="localhost" />
     <add key="port" value="2003" />
     <add key="interval" value="5" />
+    <add key="batch_size" value="1" />
   </appSettings>
-</configuration>
+</configuration>