dev: slo topic job

.github/workflows/slo-topic.yaml

@@ -0,0 +1,35 @@
+name: SLO YDB Topics
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+  schedule:
+    - cron: "45 * * * *"
+jobs:
+  ydb-slo-topic-action:
+    name: SLO Topic 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: Set up Docker Compose
+        uses: hoverkraft-tech/[email protected]
+        with:
+          compose-file: "./slo/playground/configs/compose.yaml"
+      - 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 \
+            --write-rps 50 \
+            --time 600

CHANGELOG.md

@@ -1,8 +1,13 @@
+* Reader client for YDB topics
+* Fixed: send PartitionIds in InitRequest.
+* Do a committed offset on StopPartitionSessionRequest event anyway.
+* Added log info on StopPartitionSessionRequest event.
+* PartitioningSettings were changed to change the PartitionCountLimit to MaxActivePartitions.
 * Dev: updated System.IdentityModel.Tokens.Jwt from version 0.7.0 to version 8.5.0.
 * PartitionSession.Stop uses committedOffset to complete commit tasks.
 * Changed batch type: IReadOnlyCollection<Message<TValue>> -> IReadOnlyList<Message<TValue>>.
 * Invoking TryReadRequestBytes before deserializing message.
-* Updated Ydb.Protos 1.0.6 -> 1.1.0: Updated version of the Grpc.Net.Client library to 2.67.0.
+* Updated Ydb.Protos 1.0.6 -> 1.1.1: Updated version of the Grpc.Net.Client library to 2.67.0 and proto messages.
 * Fixed: YdbDataReader.GetDataTypeName for optional values.
 * Added support for "Columns" collectionName in YdbConnection.GetSchema(Async).
 

slo/playground/configs/chaos.sh

@@ -20,21 +20,31 @@ get_random_container() {
 
 sleep 60
 
-get_random_container
-
-sh -c "docker stop ${nodeForChaos} -t 30"
-sh -c "docker start ${nodeForChaos}"
-
-sleep 60
-
-get_random_container
-
-sh -c "docker restart ${nodeForChaos} -t 0"
-
-sleep 60
+echo "Start CHAOS YDB cluster!"
+
+for i in $(seq 1 5)
+do
+  echo "docker stop/start iteration $i"
+
+  get_random_container
+
+  sh -c "docker stop ${nodeForChaos} -t 10"
+  sh -c "docker start ${nodeForChaos}"
+
+  sleep 40
+done
+
+for i in $(seq 1 5)
+do
+  echo "docker restart iteration $i"
+
+  get_random_container
+
+  sh -c "docker restart ${nodeForChaos} -t 0"
+
+  sleep 40
+done
 
 get_random_container
 
 sh -c "docker kill -s SIGKILL ${nodeForChaos}"
-
-sleep 60

slo/playground/configs/compose.yaml

@@ -1,13 +1,11 @@
-# Code generated by Github Action; DO NOT EDIT.
-
 x-runtime: &runtime
   hostname: localhost
   platform: linux/amd64
   privileged: true
   network_mode: host
 
 x-ydb-node: &ydb-node
-  image: cr.yandex/crptqonuodf51kdj7a7d/ydb:24.2.7
+  image: cr.yandex/crptqonuodf51kdj7a7d/ydb:24.3.11.13
   restart: always
   <<: *runtime
   volumes:
@@ -270,42 +268,16 @@ services:
       tenant-init:
         condition: service_completed_successfully
 
-  prometheus:
-    image: prom/prometheus
-    restart: unless-stopped
+  chaos:
+    image: docker:latest
+    restart: on-failure
+    container_name: ydb-chaos
     <<: *runtime
-    ports:
-      - "9090:9090"
+    entrypoint: ["/bin/sh", "-c", "chmod +x /opt/ydb/chaos.sh && ls -la /opt/ydb && /opt/ydb/chaos.sh"]
     volumes:
-      - ./prometheus.yml:/etc/prometheus/prometheus.yml
-    deploy: &monitoring-deploy
-      resources:
-        limits:
-          cpus: "0.1"
-          memory: 1000M
-        reservations:
-          cpus: "0.001"
-          memory: 50M
-
-  prometheus-pushgateway:
-    image: prom/pushgateway
-    restart: unless-stopped
-    <<: *runtime
-    ports:
-      - "9091:9091"
-    deploy:
-      <<: *monitoring-deploy
-
-#  chaos:
-#    image: docker:latest
-#    restart: on-failure
-#    container_name: ydb-chaos
-#    <<: *runtime
-#    entrypoint: ["/bin/sh", "-c", "ls -la /opt/ydb && /opt/ydb/chaos.sh"]
-#    volumes:
-#      - ./chaos.sh:/opt/ydb/chaos.sh
-#      - ./ydb.yaml:/opt/ydb/cfg/config.yaml
-#      - /var/run/docker.sock:/var/run/docker.sock
-#    depends_on:
-#      static-0:
-#        condition: service_healthy
+      - ./chaos.sh:/opt/ydb/chaos.sh
+      - ./ydb.yaml:/opt/ydb/cfg/config.yaml
+      - /var/run/docker.sock:/var/run/docker.sock
+    depends_on:
+      static-0:
+        condition: service_healthy

slo/playground/configs/ydb.yaml

@@ -1,3 +1,6 @@
+pqconfig:
+  require_credentials_in_new_protocol: false
+
 actor_system_config:
   cpu_count: 1
   node_type: STORAGE
@@ -40,7 +43,7 @@ domains_config:
             vdisk_kind: Default
   state_storage:
     - ring:
-        node: [1]
+        node: [ 1 ]
         nto_select: 1
       ssid: 1
 host_configs:

slo/src/AdoNet/Program.cs

@@ -3,4 +3,4 @@
 using AdoNet;
 using Internal;
 
-await Cli.Run(new SloContext(), args);
+await Cli.Run(new SloTableContext(), args);

slo/src/AdoNet/SloContext.cs → slo/src/AdoNet/SloTableContext.cs

@@ -8,7 +8,7 @@
 
 namespace AdoNet;
 
-public class SloContext : SloContext<YdbDataSource>
+public class SloTableContext : SloTableContext<YdbDataSource>
 {
     private readonly AsyncPolicy _policy = Policy.Handle<YdbException>(exception => exception.IsTransient)
         .WaitAndRetryAsync(10, attempt => TimeSpan.FromMilliseconds(attempt * 10),
@@ -103,6 +103,9 @@ protected 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, LoggerFactory = Factory }));
+        {
+            UseTls = useTls, Host = host, Port = int.Parse(port), Database = config.Db,
+            LoggerFactory = ISloContext.Factory
+        }));
     }
 }

slo/src/Internal/Cli.cs

@@ -14,7 +14,7 @@ public static class Cli
 
     private static readonly Option<string> PromPgwOption = new(
         "--prom-pgw",
-        "prometheus push gateway") { IsRequired = true };
+        "prometheus push gateway");
 
     private static readonly Option<string> ResourceYdbPath = new(
         new[] { "-t", "--resource-ydb-path" },
@@ -23,7 +23,7 @@ public static class Cli
 
     private static readonly Option<int> WriteTimeoutOption = new(
         "--write-timeout",
-        () => 1000,
+        () => 100,
         "write timeout seconds");
 
     private static readonly Option<int> ReportPeriodOption = new(
@@ -101,7 +101,7 @@ public static class Cli
         CreateCommand, RunCommand
     };
 
-    public static async Task<int> Run<T>(SloContext<T> sloContext, string[] args) where T : IDisposable
+    public static async Task<int> Run(ISloContext sloContext, string[] args)
     {
         CreateCommand.SetHandler(async createConfig => { await sloContext.Create(createConfig); },
             new CreateConfigBinder(EndpointArgument, DbArgument, ResourceYdbPath, MinPartitionsCountOption,

slo/src/Internal/SloContext.cs → slo/src/Internal/SloTableContext.cs

@@ -7,19 +7,26 @@
 
 namespace Internal;
 
-public abstract class SloContext<T> where T : IDisposable
+public interface ISloContext
 {
     // ReSharper disable once StaticMemberInGenericType
-    protected static readonly ILoggerFactory Factory =
+    public static readonly ILoggerFactory Factory =
         LoggerFactory.Create(builder =>
         {
             builder.AddConsole().SetMinimumLevel(LogLevel.Information);
             builder.AddFilter("Ydb.Sdk.Ado", LogLevel.Debug);
             builder.AddFilter("Ydb.Sdk.Services.Query", LogLevel.Debug);
-            builder.AddFilter("Ydb.Sdk.Services.Topic", LogLevel.Debug);
         });
 
-    protected static readonly ILogger Logger = Factory.CreateLogger<SloContext<T>>();
+
+    public Task Create(CreateConfig config);
+
+    public Task Run(RunConfig runConfig);
+}
+
+public abstract class SloTableContext<T> : ISloContext where T : IDisposable
+{
+    protected static readonly ILogger Logger = ISloContext.Factory.CreateLogger<SloTableContext<T>>();
 
     private volatile int _maxId;
 

slo/src/TableService/Program.cs

@@ -3,4 +3,4 @@
 using Internal;
 using TableService;
 
-return await Cli.Run(new SloContext(), args);
+return await Cli.Run(new SloTableContext(), args);

slo/src/TableService/SloContext.cs → slo/src/TableService/SloTableContext.cs

@@ -7,7 +7,7 @@
 
 namespace TableService;
 
-public class SloContext : SloContext<TableClient>
+public class SloTableContext : SloTableContext<TableClient>
 {
     private readonly TxControl _txControl = TxControl.BeginSerializableRW().Commit();
     protected override string Job => "TableService";
@@ -79,6 +79,7 @@ protected override async Task Create(TableClient client, string createTableSql,
 
     protected override async Task<TableClient> CreateClient(Config config)
     {
-        return new TableClient(await Driver.CreateInitialized(new DriverConfig(config.Endpoint, config.Db), Factory));
+        return new TableClient(await Driver.CreateInitialized(new DriverConfig(config.Endpoint, config.Db),
+            ISloContext.Factory));
     }
 }

slo/src/TopicService/Program.cs

@@ -0,0 +1,4 @@
+using Internal;
+using TopicService;
+
+return await Cli.Run(new SloTopicContext(), args);