added slo tasks

Author: KirillKurdyukov

.github/workflows/slo-topic.yaml

@@ -0,0 +1,67 @@
+name: SLO YDB Topics
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+  workflow_dispatch:
+    inputs:
+      github_pull_request_number:
+        required: true
+      slo_workload_duration_seconds:
+        default: '600'
+        required: false
+      slo_workload_read_max_rps:
+        default: '1000'
+        required: false
+      slo_workload_write_max_rps:
+        default: '1000'
+        required: false
+
+jobs:
+  ydb-slo-action:
+    name: SLO test
+    runs-on: ubuntu-latest
+    
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+      - name: Install .NET
+        uses: actions/setup-dotnet@v4
+        with:
+          dotnet-version: 8.0.x
+      - name: Initialize YDB SLO
+        uses: ydb-platform/ydb-slo-action/init@main
+        with:
+          github_pull_request_number: ${{ github.event.inputs.github_pull_request_number }}
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          workload_name: TopicService
+          ydb_database_node_count: 5
+      - name: Prepare SLO Database
+        run: |
+          cd slo/src/TopicService
+          dotnet run create grpc://localhost:2135 /Root/testdb
+      - name: Run SLO Tests
+        run: |
+          cd slo/src/TopicService
+          dotnet run run grpc://localhost:2135 /Root/testdb \
+            --prom-pgw http://localhost:9091  \
+            --report-period 250 \
+            --time ${{inputs.slo_workload_duration_seconds || 600 }} \
+            --write-rps ${{inputs.slo_workload_write_max_rps || 1000 }} \
+            --write-timeout 1000
+
+      - if: always()
+        name: Store ydb chaos testing logs
+        run: |
+          docker logs ydb-chaos > chaos-ydb.log
+
+      - if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: TopicService-chaos-ydb.log
+          path: ./chaos-ydb.log
+          retention-days: 1

slo/src/TopicService/SloTopicContext.cs

@@ -1,7 +1,10 @@
 using System.Collections.Concurrent;
+using System.Diagnostics;
 using System.Text;
+using System.Threading.RateLimiting;
 using Internal;
 using Microsoft.Extensions.Logging;
+using Prometheus;
 using Ydb.Sdk;
 using Ydb.Sdk.Services.Topic;
 using Ydb.Sdk.Services.Topic.Reader;
@@ -11,6 +14,7 @@ namespace TopicService;
 
 public class SloTopicContext : ISloContext
 {
+    private const string Job = "TopicService";
     private const string PathTopic = "/Root/testdb/slo-topic";
     private const string ConsumerName = "Consumer";
     private const int PartitionSize = 10;
@@ -45,6 +49,15 @@ public async Task Run(RunConfig config)
     {
         var driver = await Driver.CreateInitialized(
             new DriverConfig(config.Endpoint, config.Db), ISloContext.Factory);
+        var promPgwEndpoint = $"{config.PromPgw}/metrics";
+        using var prometheus = new MetricPusher(promPgwEndpoint, "workload-" + Job,
+            intervalMilliseconds: config.ReportPeriod);
+        prometheus.Start();
+
+        var writeLimiter = new FixedWindowRateLimiter(new FixedWindowRateLimiterOptions
+        {
+            Window = TimeSpan.FromMilliseconds(100), PermitLimit = config.WriteRps / 10, QueueLimit = int.MaxValue
+        });
 
         var cts = new CancellationTokenSource();
         cts.CancelAfter(TimeSpan.FromSeconds(config.Time));
@@ -54,38 +67,88 @@ public async Task Run(RunConfig config)
         var writeTasks = new List<Task>();
         for (var i = 0; i < PartitionSize; i++)
         {
+            const string operationType = "write";
             var producer = "producer-" + (i + 1);
             messageSending[producer] = new ConcurrentQueue<string>();
 
             writeTasks.Add(
                 Task.Run(async () =>
                 {
+                    var metricFactory = Metrics.WithLabels(new Dictionary<string, string>
+                        {
+                            { "operation_type", operationType },
+                            { "sdk", "dotnet" },
+                            { "sdk_version", Environment.Version.ToString() },
+                            { "workload", Job },
+                            { "workload_version", "0.0.0" }
+                        }
+                    );
+
+                    var operationsSuccessTotal = metricFactory.CreateCounter(
+                        "sdk_operations_success_total",
+                        "Total number of successful operations, categorized by type."
+                    );
+
+                    var operationLatencySeconds = metricFactory.CreateHistogram(
+                        "sdk_operation_latency_seconds",
+                        "Latency of operations performed by the SDK in seconds, categorized by type and status.",
+                        ["operation_status"],
+                        new HistogramConfiguration
+                        {
+                            Buckets =
+                            [
+                                0.001, // 1 ms
+                                0.002, // 2 ms
+                                0.003, // 3 ms
+                                0.004, // 4 ms
+                                0.005, // 5 ms
+                                0.0075, // 7.5 ms
+                                0.010, // 10 ms
+                                0.020, // 20 ms
+                                0.050, // 50 ms
+                                0.100, // 100 ms
+                                0.200, // 200 ms
+                                0.500, // 500 ms
+                                1.000 // 1 s
+                            ]
+                        }
+                    );
+
                     using var writer = new WriterBuilder<string>(driver, PathTopic)
                     {
                         ProducerId = producer,
                         BufferMaxSize = 8 * 1024 * 1024
                     }.Build();
 
+                    Logger.LogInformation("Started Writer[ProducerId={ProducerId}]", producer);
+
                     while (!cts.IsCancellationRequested)
                     {
-                        var tasks = new List<Task>();
-                        for (var k = 0; k < 10; k++)
+                        using var lease = await writeLimiter.AcquireAsync(cancellationToken: cts.Token);
+
+                        if (!lease.IsAcquired)
+                        {
+                            continue;
+                        }
+
+                        var textBuilder = new StringBuilder();
+
+                        var size = Random.Shared.Next(1, 100);
+                        for (var j = 0; j < size; j++)
                         {
-                            var textBuilder = new StringBuilder();
-
-                            var size = Random.Shared.Next(1, 100);
-                            for (var j = 0; j < size; j++)
-                            {
-                                textBuilder.Append(Guid.NewGuid());
-                            }
-
-                            var data = textBuilder.ToString();
-                            messageSending[producer].Enqueue(data);
-                            // ReSharper disable once MethodSupportsCancellation
-                            tasks.Add(writer.WriteAsync(data));
+                            textBuilder.Append(Guid.NewGuid());
                         }
 
-                        await Task.WhenAll(tasks);
+                        var data = textBuilder.ToString();
+                        messageSending[producer].Enqueue(data);
+
+                        var sw = Stopwatch.StartNew();
+                        // ReSharper disable once MethodSupportsCancellation
+                        await writer.WriteAsync(data);
+                        sw.Stop();
+
+                        operationsSuccessTotal.Inc();
+                        operationLatencySeconds.WithLabels("success").Observe(sw.Elapsed.TotalSeconds);
                     }
                 }, cts.Token)
             );
@@ -95,9 +158,25 @@ public async Task Run(RunConfig config)
         for (var i = 0; i < PartitionSize; i++)
         {
             var handlerBatch = i % 2 == 0;
-
+            const string operationType = "read";
+            var number = i;
             readTasks.Add(Task.Run(async () =>
             {
+                var metricFactory = Metrics.WithLabels(new Dictionary<string, string>
+                    {
+                        { "operation_type", operationType },
+                        { "sdk", "dotnet" },
+                        { "sdk_version", Environment.Version.ToString() },
+                        { "workload", Job },
+                        { "workload_version", "0.0.0" }
+                    }
+                );
+
+                var operationsSuccessTotal = metricFactory.CreateCounter(
+                    "sdk_operations_success_total",
+                    "Total number of successful operations, categorized by type."
+                );
+
                 using var reader = new ReaderBuilder<string>(driver)
                 {
                     ConsumerName = ConsumerName,
@@ -108,30 +187,44 @@ public async Task Run(RunConfig config)
                     MemoryUsageMaxBytes = 8 * 1024 * 1024,
                 }.Build();
 
+                Logger.LogInformation("Started Reader[{Number}]", number);
+
                 if (handlerBatch)
                 {
-                    await ReadBatchMessages(cts, reader, messageSending);
+                    await ReadBatchMessages(cts, reader, messageSending, operationsSuccessTotal);
                 }
                 else
                 {
-                    await ReadMessage(cts, reader, messageSending);
+                    await ReadMessage(cts, reader, messageSending, operationsSuccessTotal);
                 }
             }, cts.Token));
         }
 
-        await Task.WhenAll(writeTasks);
-        await Task.WhenAll(readTasks);
+        try
+        {
+            await Task.WhenAll(writeTasks);
+            await Task.WhenAll(readTasks);
+        }
+        catch (OperationCanceledException)
+        {
+        }
+
+        await prometheus.StopAsync();
+        
+        Logger.LogInformation("Task finish!");
     }
 
     private static async Task ReadBatchMessages(
         CancellationTokenSource cts,
         IReader<string> reader,
-        ConcurrentDictionary<string, ConcurrentQueue<string>> localStore
+        ConcurrentDictionary<string, ConcurrentQueue<string>> localStore,
+        Counter messageCounter
     )
     {
         while (!cts.IsCancellationRequested)
         {
-            var batchMessages = await reader.ReadBatchAsync();
+            var batchMessages = await reader.ReadBatchAsync(cts.Token);
+            messageCounter.Inc(batchMessages.Batch.Count);
 
             foreach (var message in batchMessages.Batch)
             {
@@ -143,12 +236,14 @@ ConcurrentDictionary<string, ConcurrentQueue<string>> localStore
     private static async Task ReadMessage(
         CancellationTokenSource cts,
         IReader<string> reader,
-        ConcurrentDictionary<string, ConcurrentQueue<string>> localStore
+        ConcurrentDictionary<string, ConcurrentQueue<string>> localStore,
+        Counter messageCounter
     )
     {
         while (!cts.IsCancellationRequested)
         {
-            CheckMessage(localStore, await reader.ReadAsync());
+            CheckMessage(localStore, await reader.ReadAsync(cts.Token));
+            messageCounter.Inc();
         }
     }
 
@@ -159,32 +254,30 @@ private static void CheckMessage(ConcurrentDictionary<string, ConcurrentQueue<st
         {
             if (partition.TryDequeue(out var expectedMessageData))
             {
-                if (expectedMessageData != message.Data)
+                if (expectedMessageData == message.Data)
                 {
-                    Logger.LogCritical(
-                        "Fail assertion messages! expectedData: {ExpectedData}, " +
-                        "actualMessage: [Topic: {Topic}, Data: {Data}, ProducerId: {ProducerId}, CreatedAt: {CreatedAt}]",
-                        expectedMessageData, message.Topic, message.Data, message.ProducerId, message.CreatedAt);
-
-                    throw new Exception("FAILED SLO TEST: ASSERT ERROR!");
+                    return;
                 }
-            }
-            else
-            {
+
                 Logger.LogCritical(
-                    "Unknown message: [Topic: {Topic}, ProducerId: {ProducerId}, CreatedAt: {CreatedAt}]",
-                    message.Topic, message.ProducerId, message.CreatedAt);
+                    "Fail assertion messages! expectedData: {ExpectedData}, " +
+                    "actualMessage: [Topic: {Topic}, Data: {Data}, ProducerId: {ProducerId}, CreatedAt: {CreatedAt}]",
+                    expectedMessageData, message.Topic, message.Data, message.ProducerId, message.CreatedAt);
 
-                throw new Exception("FAILED SLO TEST: UNKNOWN MESSAGE!");
+                throw new Exception("FAILED SLO TEST: ASSERT ERROR!");
             }
-        }
-        else
-        {
+
             Logger.LogCritical(
                 "Unknown message: [Topic: {Topic}, ProducerId: {ProducerId}, CreatedAt: {CreatedAt}]",
                 message.Topic, message.ProducerId, message.CreatedAt);
 
-            throw new Exception("FAILED SLO TEST: NOT FOUND PARTITION FOR PRODUCER_ID!");
+            throw new Exception("FAILED SLO TEST: UNKNOWN MESSAGE!");
         }
+
+        Logger.LogCritical(
+            "Unknown message: [Topic: {Topic}, ProducerId: {ProducerId}, CreatedAt: {CreatedAt}]",
+            message.Topic, message.ProducerId, message.CreatedAt);
+
+        throw new Exception("FAILED SLO TEST: NOT FOUND PARTITION FOR PRODUCER_ID!");
     }
 }