feat: slo context run ado.net

.github/workflows/slo.yml

@@ -58,6 +58,12 @@ jobs:
           language0: '.NET SDK over table-service'
           workload_build_context0: ../..
           workload_build_options0: -f Dockerfile --build-arg SRC_PATH=TableService
+
+          language_id1: 'ado-net'
+          workload_path1: 'slo/src'
+          language1: 'ADO.NET over query-service'
+          workload_build_context1: ../..
+          workload_build_options1: -f Dockerfile --build-arg SRC_PATH=AdoNet
 
       - uses: actions/upload-artifact@v3
         if: always() && env.DOCKER_REPO != null

slo/playground/docker-compose.yml

@@ -106,19 +106,3 @@ services:
     depends_on:
       slo-create:
         condition: service_completed_successfully
-
-  slo-cleanup:
-    build:
-      context: ../..
-      dockerfile: slo/src/Dockerfile
-    command:
-      - 'cleanup'
-      - 'http://ydb:2136'
-      - '/local'
-      - '--table-name'
-      - 'slo-dotnet'
-    networks:
-      - monitor-net
-    depends_on:
-      slo-run:
-        condition: service_completed_successfully

slo/src/AdoNet/AdoNet.csproj

@@ -5,6 +5,7 @@
         <TargetFramework>net8.0</TargetFramework>
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>enable</Nullable>
+        <AssemblyName>slo</AssemblyName>
     </PropertyGroup>
 
     <ItemGroup>

slo/src/AdoNet/Program.cs

@@ -1,6 +1,6 @@
 // See https://aka.ms/new-console-template for more information
 
 using AdoNet;
-using Internal.Cli;
+using Internal;
 
 await Cli.Run(new SloContext(), args);

slo/src/AdoNet/SloContext.cs

@@ -1,17 +1,26 @@
 using Internal;
-using Internal.Cli;
+using Microsoft.Extensions.Logging;
 using Polly;
 using Prometheus;
+using Ydb.Sdk;
 using Ydb.Sdk.Ado;
 using Ydb.Sdk.Value;
 
 namespace AdoNet;
 
 public class SloContext : SloContext<YdbDataSource>
 {
-    private readonly AsyncPolicy _policy = Policy.Handle<YdbException>()
-        .WaitAndRetryAsync(10, attempt => TimeSpan.FromSeconds(attempt),
-            (_, _, retryCount, context) => { context["RetryCount"] = retryCount; });
+    private readonly AsyncPolicy _policy = Policy.Handle<YdbException>(exception => exception.IsTransient)
+        .WaitAndRetryAsync(10, attempt => TimeSpan.FromMilliseconds(attempt * 10),
+            (e, _, _, context) =>
+            {
+                var errorsGauge = (Gauge)context["errorsGauge"];
+
+                Logger.LogWarning(e, "Failed read / write operation");
+                errorsGauge?.WithLabels(((YdbException)e).Code.StatusName(), "retried").Inc();
+            });
+
+    protected override string Job => "workload-ado-net";
 
     protected override async Task Create(YdbDataSource client, string createTableSql, int operationTimeout)
     {
@@ -22,10 +31,16 @@ protected override async Task Create(YdbDataSource client, string createTableSql
             .ExecuteNonQueryAsync();
     }
 
-    protected override async Task<int> Upsert(YdbDataSource dataSource, string upsertSql,
+    protected override async Task<(int, StatusCode)> Upsert(YdbDataSource dataSource, string upsertSql,
         Dictionary<string, YdbValue> parameters, int writeTimeout, Gauge? errorsGauge = null)
     {
-        var policyResult = await _policy.ExecuteAndCaptureAsync(async () =>
+        var context = new Context();
+        if (errorsGauge != null)
+        {
+            context["errorsGauge"] = errorsGauge;
+        }
+
+        var policyResult = await _policy.ExecuteAndCaptureAsync(async _ =>
         {
             await using var ydbConnection = await dataSource.OpenConnectionAsync();
 
@@ -38,22 +53,43 @@ protected override async Task<int> Upsert(YdbDataSource dataSource, string upser
             }
 
             await ydbCommand.ExecuteNonQueryAsync();
-        });
+        }, context);
 
-        return (int)policyResult.Context["RetryCount"];
-    }
 
-    protected override Task<string> Select(string selectSql, Dictionary<string, YdbValue> parameters, int readTimeout)
-    {
-        throw new NotImplementedException();
+        return (policyResult.Context.TryGetValue("RetryCount", out var countAttempts) ? (int)countAttempts : 1,
+            ((YdbException)policyResult.FinalException)?.Code ?? StatusCode.Success);
     }
 
-    protected override Task CleanUp(string dropTableSql, int operationTimeout)
+    protected override async Task<(int, StatusCode, object?)> Select(YdbDataSource dataSource, string selectSql,
+        Dictionary<string, YdbValue> parameters, int readTimeout, Gauge? errorsGauge = null)
     {
-        throw new NotImplementedException();
+        var context = new Context();
+        if (errorsGauge != null)
+        {
+            context["errorsGauge"] = errorsGauge;
+        }
+
+        var attempts = 0;
+        var policyResult = await _policy.ExecuteAndCaptureAsync(async _ =>
+        {
+            attempts++;
+            await using var ydbConnection = await dataSource.OpenConnectionAsync();
+
+            var ydbCommand = new YdbCommand(ydbConnection)
+                { CommandText = selectSql, CommandTimeout = readTimeout };
+
+            foreach (var (key, value) in parameters)
+            {
+                ydbCommand.Parameters.AddWithValue(key, value);
+            }
+
+            return await ydbCommand.ExecuteScalarAsync();
+        }, context);
+
+        return (attempts, ((YdbException)policyResult.FinalException)?.Code ?? StatusCode.Success, policyResult.Result);
     }
 
-    public override Task<YdbDataSource> CreateClient(Config config)
+    protected override Task<YdbDataSource> CreateClient(Config config)
     {
         var splitEndpoint = config.Endpoint.Split("://");
         var useTls = splitEndpoint[0] switch
@@ -67,6 +103,6 @@ public override Task<YdbDataSource> CreateClient(Config config)
         var port = splitEndpoint[1].Split(":")[1];
 
         return Task.FromResult(new YdbDataSource(new YdbConnectionStringBuilder
-            { UseTls = useTls, Host = host, Port = int.Parse(port), Database = config.Db }));
+            { UseTls = useTls, Host = host, Port = int.Parse(port), Database = config.Db, LoggerFactory = Factory }));
     }
 }

slo/src/Internal/Cli/Cli.cs → slo/src/Internal/Cli.cs

@@ -1,6 +1,6 @@
 using System.CommandLine;
 
-namespace Internal.Cli;
+namespace Internal;
 
 public static class Cli
 {
@@ -90,16 +90,6 @@ public static class Cli
         WriteTimeoutOption
     };
 
-    private static readonly Command CleanupCommand = new(
-        "cleanup",
-        "drops table in database")
-    {
-        EndpointArgument,
-        DbArgument,
-        TableOption,
-        WriteTimeoutOption
-    };
-
     private static readonly Command RunCommand = new(
         "run",
         "runs workload (read and write to table with sets RPS)")
@@ -120,7 +110,7 @@ public static class Cli
 
     private static readonly RootCommand RootCommand = new("SLO app")
     {
-        CreateCommand, CleanupCommand, RunCommand
+        CreateCommand, RunCommand
     };
 
     public static async Task<int> Run<T>(SloContext<T> sloContext, string[] args) where T : IDisposable
@@ -129,10 +119,7 @@ public static async Task<int> Run<T>(SloContext<T> sloContext, string[] args) wh
             new CreateConfigBinder(EndpointArgument, DbArgument, TableOption, MinPartitionsCountOption,
                 MaxPartitionsCountOption, PartitionSizeOption, InitialDataCountOption, WriteTimeoutOption));
 
-        CleanupCommand.SetHandler(async cleanUpConfig => { await CliCommands.CleanUp(cleanUpConfig); },
-            new CleanUpConfigBinder(EndpointArgument, DbArgument, TableOption, WriteTimeoutOption));
-
-        RunCommand.SetHandler(async runConfig => { await CliCommands.Run(runConfig); },
+        RunCommand.SetHandler(async runConfig => { await sloContext.Run(runConfig); },
             new RunConfigBinder(EndpointArgument, DbArgument, TableOption, PromPgwOption, ReportPeriodOption,
                 ReadRpsOption, ReadTimeoutOption, WriteRpsOption, WriteTimeoutOption, TimeOption, ShutdownTimeOption));