refactored storage layer

benchmarks/TimescaleBenchmark/Program.cs

@@ -10,7 +10,7 @@ public class TimescaleDBReportingExample
 {
      private const string CleanDbSql = $"""
          DROP TABLE IF EXISTS {SqlQueries.StepStatsTable};
-         DROP TABLE IF EXISTS {SqlQueries.ClusterStatsTableName};
+         DROP TABLE IF EXISTS {SqlQueries.SessionsTable};
      """;
 
     public void Run()
@@ -20,12 +20,12 @@ public void Run()
             .UsePostgreSql();
 
         const string connectionString = "Host=localhost;Port=5432;Database=metricsdb;Username=timescaledb;Password=timescaledb;Pooling=true;Maximum Pool Size=300;";
-        
+
         using var connection = new NpgsqlConnection(connectionString);
 
         connection.ExecuteNonQuery(CleanDbSql);
 
-        connection.ExecuteNonQuery(SqlQueries.CreateStepStatsTable + SqlQueries.CreateClusterStatsTable);
+        connection.ExecuteNonQuery(SqlQueries.CreateStepStatsTable + SqlQueries.CreateSessionsTable);
 
         var writeScenario = new WriteScenario().Create(connectionString);
         var readScenario = new ReadScenario().Create(connectionString, DateTime.UtcNow, sessionId: "0");

benchmarks/TimescaleBenchmark/ReadScenario.cs

@@ -108,8 +108,6 @@ public ScenarioProps Create(string connectionString, DateTime startTime, string
                     SimulationValue = fakePoint.SimulationValue,
                     SessionId = sessionId,
                     CurrentOperation = fakePoint.CurrentOperation,
-                    TestSuite = fakePoint.TestSuite,
-                    TestName = fakePoint.TestName,
 
                     OkStatusCodes = JsonSerializer.Serialize(fakePoint.OkStatusCodes),
                     OkLatencyCount = JsonSerializer.Serialize(fakePoint.OkLatencyCount),

benchmarks/TimescaleBenchmark/WriteScenario.cs

@@ -26,7 +26,8 @@ public ScenarioProps Create(string connectionString)
 
                 fakePoint.Time = curTime;
                 fakePoint.SessionId = ctx.ScenarioInfo.InstanceNumber.ToString();
-
+                fakePoint.CurrentOperation = NBomber.Contracts.Stats.OperationType.Bombing;
+
                 await connection.BinaryBulkInsertAsync(SqlQueries.StepStatsTable, Enumerable.Repeat(fakePoint, 5));
 
                 return Response.Ok();

examples/Demo/infra-config.json

@@ -1,5 +1,5 @@
 {
-  "TimescaleDbSink":{
+  "TimescaleDbSink": {
     "ConnectionString": "Host=localhost;Port=5432;Database=metricsdb;Username=timescaledb;Password=timescaledb;Pooling=true;Maximum Pool Size=300;"
   }
 }

src/NBomber.Sinks.Timescale/Contracts/PointDbRecord.cs

@@ -11,8 +11,7 @@ public class PointDbRecord
     [Column(ColumnNames.SessionId)] public string SessionId { get; set; }
     [Column(ColumnNames.CurrentOperation)] public OperationType CurrentOperation { get; set; }
 
-    [Column(ColumnNames.TestSuite)] public string TestSuite { get; set; }
-    [Column(ColumnNames.TestName)] public string TestName { get; set; }
+
     [Column(ColumnNames.Scenario)] public string Scenario { get; set; }
     [Column(ColumnNames.Step)] public string Step { get; set; }
 
@@ -68,5 +67,8 @@ public class NodeInfoDbRecord
 {
     [Column(ColumnNames.Time)] public DateTime Time { get; set; }
     [Column(ColumnNames.SessionId)] public string SessionId { get; set; }
+    [Column(ColumnNames.CurrentOperation)] public OperationType CurrentOperation { get; set; }
+    [Column(ColumnNames.TestSuite)] public string TestSuite { get; set; }
+    [Column(ColumnNames.TestName)] public string TestName { get; set; }
     [Column(ColumnNames.NodeInfo)] public string NodeInfo { get; set; }
 }

src/NBomber.Sinks.Timescale/SqlQueries.cs

@@ -6,8 +6,8 @@ namespace NBomber.Sinks.Timescale;
 
 public static class SqlQueries
 {
-    public const string StepStatsTable = "step_stats";
-    public const string ClusterStatsTableName = "cluster_stats";
+    public const string StepStatsTable = "nb_step_stats";
+    public const string SessionsTable = "nb_sessions";
 
     public static string CreateStepStatsTable => $@"
         CREATE TABLE IF NOT EXISTS ""{StepStatsTable}""
@@ -16,8 +16,6 @@ CREATE TABLE IF NOT EXISTS ""{StepStatsTable}""
             ""{ColumnNames.SessionId}"" TEXT NOT NULL,
             ""{ColumnNames.CurrentOperation}"" TEXT,            
 
-            ""{ColumnNames.TestSuite}"" TEXT,
-            ""{ColumnNames.TestName}"" TEXT,
             ""{ColumnNames.Scenario}"" TEXT,
             ""{ColumnNames.Step}"" TEXT,            
 
@@ -72,14 +70,15 @@ CREATE TABLE IF NOT EXISTS ""{StepStatsTable}""
         CREATE INDEX IF NOT EXISTS {ColumnNames.SessionId}_index ON {StepStatsTable} ({ColumnNames.SessionId});
    ";
 
-    public static string CreateClusterStatsTable => $@"
-        CREATE TABLE IF NOT EXISTS ""{ClusterStatsTableName}""
+    public static string CreateSessionsTable => $@"
+        CREATE TABLE IF NOT EXISTS ""{SessionsTable}""
         (            
             ""{ColumnNames.Time}"" TIMESTAMPTZ NOT NULL,
-            ""{ColumnNames.SessionId}"" TEXT NOT NULL,            
+            ""{ColumnNames.SessionId}"" TEXT PRIMARY KEY,
+            ""{ColumnNames.CurrentOperation}"" TEXT, 
+            ""{ColumnNames.TestSuite}"" TEXT,
+            ""{ColumnNames.TestName}"" TEXT,
             ""{ColumnNames.NodeInfo}"" JSONB            
-        );
-        SELECT create_hypertable('{ClusterStatsTableName}', by_range('{ColumnNames.Time}', INTERVAL '1 day'), if_not_exists => TRUE);
-        CREATE INDEX IF NOT EXISTS {ColumnNames.SessionId}_index ON {ClusterStatsTableName} ({ColumnNames.SessionId});
+        );    
        ";
-}   
+}

src/NBomber.Sinks.Timescale/TimescaleDbSink.cs

@@ -9,6 +9,7 @@
 using NBomber.Contracts;
 using NBomber.Contracts.Stats;
 using NBomber.Sinks.Timescale.Contracts;
+using System.Data;
 
 namespace NBomber.Sinks.Timescale;
 
@@ -64,7 +65,7 @@ public async Task Init(IBaseContext context, IConfiguration infraConfig)
 
         await _mainConnection.ExecuteNonQueryAsync(
             SqlQueries.CreateStepStatsTable
-            + SqlQueries.CreateClusterStatsTable
+            + SqlQueries.CreateSessionsTable
         );
     }
 
@@ -73,21 +74,25 @@ public async Task Start()
         if (_mainConnection != null)
         {
             var nodeInfo = _context.GetNodeInfo();
-
+            var testInfo = _context.TestInfo;
+
             var record = new NodeInfoDbRecord
             {
                 Time = DateTime.UtcNow,
-                SessionId = _context.TestInfo.SessionId,
+                SessionId = testInfo.SessionId,
+                CurrentOperation = OperationType.Bombing,
+                TestSuite = testInfo.TestSuite,
+                TestName = testInfo.TestName,
                 NodeInfo = Json.serialize(nodeInfo)
             };
 
-            await _mainConnection.BinaryBulkInsertAsync(SqlQueries.ClusterStatsTableName, new [] { record });
+            await _mainConnection.BinaryBulkInsertAsync(SqlQueries.SessionsTable, new [] { record });
         }
     }
 
     public Task SaveRealtimeStats(ScenarioStats[] stats) => SaveScenarioStats(stats);
 
-    public Task SaveFinalStats(NodeStats stats) => SaveScenarioStats(stats.ScenarioStats);
+    public Task SaveFinalStats(NodeStats stats) => SaveScenarioStats(stats.ScenarioStats, isFinal: true);
 
     public Task Stop() => Task.CompletedTask;
 
@@ -97,7 +102,7 @@ public void Dispose()
         _mainConnection.Dispose();
     }
 
-    private async Task SaveScenarioStats(ScenarioStats[] stats)
+    private async Task SaveScenarioStats(ScenarioStats[] stats, bool isFinal = false)
     {
         if (_mainConnection != null)
         {
@@ -107,7 +112,31 @@ private async Task SaveScenarioStats(ScenarioStats[] stats)
                 .SelectMany(step => MapToPoint(step, currentTime))
                 .ToArray();
 
-            await _mainConnection.BinaryBulkInsertAsync(SqlQueries.StepStatsTable, points);
+            if (!isFinal)
+            {
+                await _mainConnection.BinaryBulkInsertAsync(SqlQueries.StepStatsTable, points);
+            }
+            else
+            {
+                using (var transaction = _mainConnection.EnsureOpen().BeginTransaction())
+                {
+                    await _mainConnection.BinaryBulkInsertAsync(SqlQueries.StepStatsTable, points, transaction: (NpgsqlTransaction) transaction);
+
+                    var testInfo = _context.TestInfo;
+
+                    var queryEntity = new NodeInfoDbRecord
+                    {
+                        SessionId = testInfo.SessionId,
+                        CurrentOperation = OperationType.Complete,
+                    };
+
+                    var fields = Field.Parse<NodeInfoDbRecord>(e => e.CurrentOperation);
+
+                    await _mainConnection.UpdateAsync(SqlQueries.SessionsTable, queryEntity, fields: fields, transaction: transaction);
+
+                    transaction.Commit();
+                }
+            }
         }
     }
 
@@ -143,8 +172,6 @@ private PointDbRecord[] MapToPoint(ScenarioStats scnStats, DateTime currentTime)
                 Time = currentTime,
                 SessionId = testInfo.SessionId,
                 CurrentOperation = nodeInfo.CurrentOperation,
-                TestSuite = testInfo.TestSuite,
-                TestName = testInfo.TestName,
                 Scenario = scnStats.ScenarioName,
                 Step = step.StepName,