implemented simple DB migrator

.github/workflows/build.yml

@@ -12,12 +12,20 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v3
-
+    - uses: actions/checkout@v3   
     - name: Setup .NET
       uses: actions/setup-dotnet@v2
       with:
         dotnet-version: 8.x.x    
 
+    - name: Docker Compose Up
+      run: docker compose --project-directory ./tests/Nbomber.Sinks.Timescale.Tests up -d
+
     - name: Build
-      run: dotnet build NBomber.Sinks.Timescale.sln
+      run: dotnet build NBomber.Sinks.Timescale.sln
+
+    - name: Test
+      run: dotnet test tests/Nbomber.Sinks.Timescale.Tests/Nbomber.Sinks.Timescale.Tests.csproj --configuration Release --verbosity normal
+
+    - name: Docker Compose Down
+      run: docker compose --project-directory ./tests/Nbomber.Sinks.Timescale.Tests down

NBomber.Sinks.Timescale.sln

@@ -1,19 +1,24 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Version 17
+VisualStudioVersion = 17.10.35013.160
+MinimumVisualStudioVersion = 10.0.40219.1
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "solution items", "solution items", "{997E99F2-8640-4A0B-918B-D000900B5F79}"
 	ProjectSection(SolutionItems) = preProject
 		docker-compose.yml = docker-compose.yml
 	EndProjectSection
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "benchmarks", "benchmarks", "{7CF08F85-D062-49F6-9A69-72F7DB9369AB}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TimescaleBenchmark", "benchmarks\TimescaleBenchmark\TimescaleBenchmark.csproj", "{323B3F41-09BF-4B77-BE47-15AF73A1BCA1}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "TimescaleBenchmark", "benchmarks\TimescaleBenchmark\TimescaleBenchmark.csproj", "{323B3F41-09BF-4B77-BE47-15AF73A1BCA1}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "examples", "examples", "{83B64449-B0C6-4C1E-BF2E-D9641C33F1ED}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Demo", "examples\Demo\Demo.csproj", "{2A175FF0-F464-46FC-BBDD-37CC720B212C}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Demo", "examples\Demo\Demo.csproj", "{2A175FF0-F464-46FC-BBDD-37CC720B212C}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NBomber.Sinks.Timescale", "src\NBomber.Sinks.Timescale\NBomber.Sinks.Timescale.csproj", "{9225A528-D8A6-4EDB-9C71-228F4F0B58E2}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NBomber.Sinks.Timescale", "src\NBomber.Sinks.Timescale\NBomber.Sinks.Timescale.csproj", "{9225A528-D8A6-4EDB-9C71-228F4F0B58E2}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Nbomber.Sinks.Timescale.Tests", "tests\Nbomber.Sinks.Timescale.Tests\Nbomber.Sinks.Timescale.Tests.csproj", "{2034EFD1-9463-4292-9DE2-C5DA9610B1BC}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -33,9 +38,19 @@ Global
 		{9225A528-D8A6-4EDB-9C71-228F4F0B58E2}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{9225A528-D8A6-4EDB-9C71-228F4F0B58E2}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{9225A528-D8A6-4EDB-9C71-228F4F0B58E2}.Release|Any CPU.Build.0 = Release|Any CPU
+		{2034EFD1-9463-4292-9DE2-C5DA9610B1BC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{2034EFD1-9463-4292-9DE2-C5DA9610B1BC}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{2034EFD1-9463-4292-9DE2-C5DA9610B1BC}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{2034EFD1-9463-4292-9DE2-C5DA9610B1BC}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
 	EndGlobalSection
 	GlobalSection(NestedProjects) = preSolution
 		{323B3F41-09BF-4B77-BE47-15AF73A1BCA1} = {7CF08F85-D062-49F6-9A69-72F7DB9369AB}
 		{2A175FF0-F464-46FC-BBDD-37CC720B212C} = {83B64449-B0C6-4C1E-BF2E-D9641C33F1ED}
 	EndGlobalSection
+	GlobalSection(ExtensibilityGlobals) = postSolution
+		SolutionGuid = {776A101B-4C02-4D9A-A63C-4D50065DC601}
+	EndGlobalSection
 EndGlobal

benchmarks/TimescaleBenchmark/Program.cs

@@ -25,7 +25,7 @@ public void Run()
 
         connection.ExecuteNonQuery(CleanDbSql);
 
-        connection.ExecuteNonQuery(SqlQueries.CreateStepStatsTable + SqlQueries.CreateSessionsTable);
+        connection.ExecuteNonQuery(SqlQueries.CreateStepStatsTable + SqlQueries.CreateSessionsTable + SqlQueries.CreateDbSchemaVersion);
 
         var writeScenario = new WriteScenario().Create(connectionString);
         var readScenario = new ReadScenario().Create(connectionString, DateTime.UtcNow, sessionId: "0");

benchmarks/TimescaleBenchmark/TimescaleBenchmark.csproj

@@ -7,11 +7,11 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="AutoBogus" Version="2.13.1"/>
-        <PackageReference Include="NBomber" Version="5.7.0"/>
-        <PackageReference Include="Dapper" Version="2.1.35"/>
-        <PackageReference Include="RepoDb.PostgreSql" Version="1.13.2-alpha1"/>
-        <PackageReference Include="RepoDb.PostgreSql.BulkOperations" Version="1.13.2-alpha1"/>
+        <PackageReference Include="AutoBogus" Version="2.13.1" />
+        <PackageReference Include="NBomber" Version="5.7.0" />
+        <PackageReference Include="Dapper" Version="2.1.35" />
+        <PackageReference Include="RepoDb.PostgreSql" Version="1.13.1" />
+        <PackageReference Include="RepoDb.PostgreSql.BulkOperations" Version="1.13.1" />
     </ItemGroup>
 
     <ItemGroup>

global.json

@@ -0,0 +1,7 @@
+{
+  "sdk": {
+    "version": "8.0.0",
+    "rollForward": "latestMajor",
+    "allowPrerelease": false
+  }
+}

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

@@ -1,18 +1,18 @@
 namespace NBomber.Sinks.Timescale.Contracts;
 
 public static class ColumnNames
 {
     public const string Time = "time";
     public const string ScenarioTimestamp = "scenario_timestamp";
     public const string SessionId = "session_id";
     public const string CurrentOperation = "current_operation";
 
     public const string NodeInfo = "node_info";
     public const string TestSuite = "test_suite";
     public const string TestName = "test_name";
 
     public const string Scenario = "scenario";
     public const string Step = "step";
 
     public const string AllReqCount = "all_req_count";
     public const string AllDataAll = "all_data_all";
@@ -60,4 +60,6 @@
     public const string FailLatencyCount = "fail_latency_count";
 
     public const string SimulationValue = "simulation_value";
+
+    public const string Version = "Version";
 }

src/NBomber.Sinks.Timescale/DbMigrations .cs

@@ -0,0 +1,93 @@
+using NBomber.Sinks.Timescale.Contracts;
+using Npgsql;
+using RepoDb;
+using ILogger = Serilog.ILogger;
+
+namespace NBomber.Sinks.Timescale
+{
+    public class DbMigrations
+    {
+        public const int SinkSchemaVersion = 0;
+        private string _connectionString;
+        private ILogger _logger;
+
+        public DbMigrations(string connectionString, ILogger logger)
+        {
+            _connectionString = connectionString;
+            _logger = logger;
+        }
+
+        public async Task Run()
+        {
+            var currentDbVersion = await GetCurrendDBVersion();
+
+            if (currentDbVersion > SinkSchemaVersion)
+            {
+                var errMessage = $@"Your NBomber.Sinks.Timescale schema version: '{SinkSchemaVersion}' is not compatible with DB schema version: '{currentDbVersion}'";
+                _logger.Error(errMessage);
+                throw new PlatformNotSupportedException(errMessage);
+            }
+            else if (currentDbVersion < SinkSchemaVersion) 
+            {
+                for (var v = currentDbVersion + 1; v <= SinkSchemaVersion; v++)
+                {
+                    await ApplyMigration(v);
+                }
+            }
+        }
+
+        private async Task<int> GetCurrendDBVersion()
+        {
+            try
+            {
+                using var connection = new NpgsqlConnection(_connectionString);
+                var result = await connection.ExecuteQueryAsync<int>($@"SELECT ""{ColumnNames.Version}"" FROM {SqlQueries.DbSchemaVersion};");
+                var currentDbVersion = result.FirstOrDefault();
+                return currentDbVersion;
+            }
+            catch (Exception ex) 
+            { 
+                _logger.Error(ex, ex.Message);
+                return -1;
+            }
+        }
+
+        private async Task ApplyMigration(int version)
+        {
+            using var connection = new NpgsqlConnection(_connectionString);
+
+            switch (version) 
+            {
+                case 0:
+                    await connection.ExecuteNonQueryAsync(
+                        SqlQueries.CreateStepStatsTable
+                      + SqlQueries.CreateSessionsTable
+                      + SqlQueries.CreateDbSchemaVersion);
+
+                    await connection.ExecuteNonQueryAsync($@"
+                            INSERT INTO {SqlQueries.DbSchemaVersion} (""{ColumnNames.Version}"")
+                            VALUES ({version})
+                            ;");
+
+                    _logger.Debug("Created initial tables");
+                    break;
+
+                //case 1:
+                //    await connection.ExecuteNonQueryAsync($@"
+                //            ALTER TABLE {SqlQueries.StepStatsTable}
+                //            ADD COLUMN IF NOT EXISTS {ColumnNames.TestCulomn} TEXT
+                //            ;
+
+                //            WITH updated AS (
+                //                UPDATE {SqlQueries.DbSchemaVersion}
+                //                SET ""{ColumnNames.Version}"" = {version}
+                //                RETURNING *
+                //            )
+                //            INSERT INTO {SqlQueries.DbSchemaVersion} (""{ColumnNames.Version}"")
+                //            SELECT 1 
+                //            WHERE NOT EXISTS (SELECT * FROM updated);");
+                //    break;
+            }
+        }
+    }
+}

src/NBomber.Sinks.Timescale/SqlQueries.cs

@@ -8,7 +8,7 @@ public static class SqlQueries
 {
     public const string StepStatsTable = "nb_step_stats";
     public const string SessionsTable = "nb_sessions";
-
+    public const string DbSchemaVersion = "nb_sink_schema_version";
     public static string CreateStepStatsTable => $@"
         CREATE TABLE IF NOT EXISTS ""{StepStatsTable}""
         (
@@ -82,4 +82,11 @@ CREATE TABLE IF NOT EXISTS ""{SessionsTable}""
             ""{ColumnNames.NodeInfo}"" JSONB            
         );    
        ";
+
+    public static string CreateDbSchemaVersion => $@"
+        CREATE TABLE IF NOT EXISTS ""{DbSchemaVersion}""
+        (            
+            ""{ColumnNames.Version}"" INT PRIMARY KEY         
+        );    
+       ";
 }

src/NBomber.Sinks.Timescale/TimescaleDbSink.cs

@@ -60,13 +60,11 @@ public async Task Init(IBaseContext context, IConfiguration infraConfig)
         GlobalConfiguration
             .Setup()
             .UsePostgreSql();
-        
+
         await _mainConnection.OpenAsync();
 
-        await _mainConnection.ExecuteNonQueryAsync(
-            SqlQueries.CreateStepStatsTable
-            + SqlQueries.CreateSessionsTable
-        );
+        var migration = new DbMigrations(_connectionString, _logger);
+        await migration.Run();  
     }
 
     public async Task Start()

tests/Nbomber.Sinks.Timescale.Tests/Infra/EnvContextFixture.cs

@@ -0,0 +1,67 @@
+using Ductus.FluentDocker.Services;
+using Ductus.FluentDocker.Builders;
+using System.Text.Json;
+using NBomber.Sinks.Timescale;
+
+namespace Nbomber.Sinks.Timescale.Tests.Infra
+{
+    public class EnvContextFixture : IDisposable
+    {
+        private readonly ICompositeService _docker;
+        private static readonly bool UseDocker = false;
+        private readonly Func<TimescaleDbSink> _createSincFn;
+
+        public TestHelper TestHelper {  get; private set; }
+        public TimescaleDbSink TimescaleDbSink { get; private set; }
+
+        public EnvContextFixture()
+        {
+            var config = JsonSerializer.Deserialize<Config>(
+                json: File.ReadAllText("config.Development.json")
+            );
+
+            _createSincFn = () => new TimescaleDbSink(config.DBSettings.ConnectionString);
+
+            if (UseDocker)
+            {
+                var dockerCompose = Path.Combine(Directory.GetCurrentDirectory(), "docker-compose.yml");
+
+                _docker = new Builder()
+                    .UseContainer()
+                    .UseCompose()
+                    .FromFile(dockerCompose)
+                    .RemoveOrphans()
+                    .Build();
+
+                _docker.Start();
+            }
+
+            HealthCheck.WaitUntilReady(config.DBSettings.ConnectionString).Wait();
+
+            TestHelper = new TestHelper(config.DBSettings.ConnectionString);
+        }
+
+        public TimescaleDbSink CreateTimescaleDbSinkInstance()
+        {
+            return _createSincFn();
+        }
+        public void Dispose()
+        {
+            if (UseDocker)
+            {
+                _docker.Stop();
+                _docker.Dispose();
+            }
+        }
+    }
+
+    public class Config
+    {
+        public DBSettings DBSettings { get; set; }
+    }
+
+    public class DBSettings
+    {
+        public string ConnectionString { get; set; }
+    }
+}

tests/Nbomber.Sinks.Timescale.Tests/Infra/HealthCheck.cs

@@ -0,0 +1,30 @@
+using Npgsql;
+using RepoDb;
+
+namespace Nbomber.Sinks.Timescale.Tests.Infra
+{
+    static class HealthCheck
+    {
+        public static async Task WaitUntilReady(string connectionString)
+        {
+            while (!await CheckIfDbExist(connectionString))
+            {
+                await Task.Delay(TimeSpan.FromSeconds(1));
+            }
+        }
+
+        private static async Task<bool> CheckIfDbExist(string connectionString)
+        {
+            try
+            {
+                using var connection = new NpgsqlConnection(connectionString);
+
+                return await connection.ExecuteScalarAsync<bool>("SELECT EXISTS (SELECT FROM pg_tables)");
+            }
+            catch
+            {
+                return false;
+            }
+        }
+    }
+}