[C++ SDK] Supported metric loading to OTLP in SLO workload

Author: Gazizonoki

contrib/libs/opentelemetry-cpp/api/ya.make

@@ -19,7 +19,7 @@ NO_COMPILER_WARNINGS()
 NO_UTIL()
 
 CFLAGS(
-    -DOPENTELEMETRY_ABI_VERSION_NO=1
+    GLOBAL -DOPENTELEMETRY_ABI_VERSION_NO=2
     GLOBAL -DOPENTELEMETRY_STL_VERSION=2023
 )
 

contrib/libs/opentelemetry-cpp/ya.make

@@ -33,7 +33,7 @@ NO_COMPILER_WARNINGS()
 NO_UTIL()
 
 CFLAGS(
-    -DOPENTELEMETRY_ABI_VERSION_NO=1
+    GLOBAL -DOPENTELEMETRY_ABI_VERSION_NO=2
     GLOBAL -DOPENTELEMETRY_STL_VERSION=2023
 )
 

ydb/public/sdk/cpp/tests/slo_workloads/key_value/generate.cpp

@@ -8,7 +8,7 @@ using namespace NYdb::NTable;
 
 
 TGenerateInitialContentJob::TGenerateInitialContentJob(const TCreateOptions& createOpts, std::uint32_t maxId)
-    : TThreadJob(createOpts.CommonOptions)
+    : TThreadJob(createOpts.CommonOptions, "generate")
     , Executor(createOpts.CommonOptions, Stats, TExecutor::ModeBlocking)
     , PackGenerator(
         createOpts.CommonOptions
@@ -81,5 +81,4 @@ UPSERT INTO `%s` SELECT * FROM AS_TABLE($items);
 void TGenerateInitialContentJob::OnFinish() {
     Executor.Finish();
     Executor.Wait();
-    Stats.Flush();
 }

ydb/public/sdk/cpp/tests/slo_workloads/key_value/key_value.h

@@ -48,7 +48,6 @@ class TReadJob : public TThreadJob {
 private:
     std::unique_ptr<TExecutor> Executor;
     std::uint32_t ObjectIdRange;
-    bool SaveResult;
 };
 
 int CreateTable(TDatabaseOptions& dbOptions);

ydb/public/sdk/cpp/tests/slo_workloads/key_value/run.cpp

@@ -7,7 +7,7 @@ using namespace NYdb;
 using namespace NYdb::NTable;
 
 TWriteJob::TWriteJob(const TCommonOptions& opts, std::uint32_t maxId)
-    : TThreadJob(opts)
+    : TThreadJob(opts, "write")
     , Executor(opts, Stats, TExecutor::ModeNonBlocking)
     , Generator(opts, maxId)
 {}
@@ -77,15 +77,13 @@ UPSERT INTO `%s` SELECT * FROM AS_TABLE($items);
 void TWriteJob::OnFinish() {
     Executor.Finish();
     Executor.Wait();
-    Stats.Flush();
 }
 
 // Implementation of TReadJob
 TReadJob::TReadJob(const TCommonOptions& opts, std::uint32_t maxId)
-    : TThreadJob(opts)
-    , Executor(opts.RetryMode ? new TExecutorWithRetry(opts, Stats) : new TExecutor(opts, Stats))
+    : TThreadJob(opts, "read")
+    , Executor(std::make_unique<TExecutor>(opts, Stats))
     , ObjectIdRange(static_cast<std::uint32_t>(maxId * 1.25)) // 20% of requests with no result
-    , SaveResult(opts.SaveResult)
 {}
 
 void TReadJob::ShowProgress(TStringBuilder& report) {
@@ -154,8 +152,4 @@ void TReadJob::OnFinish() {
     if (infly) {
         Cerr << "Warning: thread A finished while having " << infly << " infly requests." << Endl;
     }
-    Stats.Flush();
-    if (SaveResult) {
-        Stats.SaveResult();
-    }
 }

ydb/public/sdk/cpp/tests/slo_workloads/utils/executor.cpp

@@ -274,13 +274,17 @@ bool TExecutor::Execute(const NYdb::NTable::TTableClient::TOperationFunc& func)
             }
         }
 
-        TStatUnit stat = Stats.CreateStatUnit();
+        auto stat = Stats.StartRequest();
+
+        auto future = InsistentClient.ExecuteWithRetry([func, stat](NYdb::NTable::TSession session) {
+            auto result = func(session);
+            return result;
+        });
 
-        auto future = InsistentClient.ExecuteWithRetry(func);
         future.Subscribe([this, stat, SemaphoreWrapper](const TAsyncFinalStatus& future) mutable {
             Y_ABORT_UNLESS(future.HasValue());
             TFinalStatus resultStatus = future.GetValue();
-            Stats.Report(stat, resultStatus);
+            Stats.FinishRequest(stat, resultStatus);
             if (resultStatus) {
                 CheckForError(*resultStatus);
             }
@@ -351,7 +355,6 @@ bool TExecutor::IsStopped() {
 }
 
 void TExecutor::Finish() {
-    // Stats.UpdateSessionStats(InsistentClient.GetSessionStats());
     with_lock(Lock) {
         if (!AllJobsLaunched) {
             AllJobsLaunched = true;
@@ -361,9 +364,6 @@ void TExecutor::Finish() {
 }
 
 void TExecutor::UpdateStats() {
-    if (Infly > MaxSecInfly) {
-        MaxSecInfly = Infly;
-    }
     std::uint64_t activeSessions = InsistentClient.GetActiveSessions();
     if (activeSessions > MaxSecSessions) {
         MaxSecSessions = activeSessions;
@@ -382,10 +382,10 @@ void TExecutor::UpdateStats() {
 }
 
 void TExecutor::ReportStats() {
+    Stats.ReportStats(MaxSecSessions, MaxSecReadPromises, MaxSecExecutorPromises);
     TInstant now = TInstant::Now();
     if (now.Seconds() > LastReportSec) {
-        Stats.ReportStats(MaxSecInfly, MaxSecSessions, MaxSecReadPromises, MaxSecExecutorPromises);
-        MaxSecInfly = 0;
+        Stats.ReportStats(MaxSecSessions, MaxSecReadPromises, MaxSecExecutorPromises);
         MaxSecSessions = 0;
         MaxSecReadPromises = 0;
         MaxSecExecutorPromises = 0;
@@ -442,93 +442,3 @@ void TExecutor::Report(TStringBuilder& out) const {
         }
     }
 }
-
-
-TExecutorWithRetry::TExecutorWithRetry(const TCommonOptions& opts, TStat& stats)
-    : TExecutor(opts, stats)
-{}
-
-bool TExecutorWithRetry::Execute(const NYdb::NTable::TTableClient::TOperationFunc& func) {
-    auto threadFunc = [this, func]() {
-        if (IsStopped()) {
-            DecrementWaiting();
-            return;
-        }
-
-        with_lock(Lock) {
-            --Waiting;
-            if (Infly < Opts.MaxInfly) {
-                ++Infly;
-                if (Infly > MaxInfly) {
-                    MaxInfly = Infly;
-                }
-                UpdateStats();
-            } else {
-                Stats.ReportMaxInfly();
-                UpdateStats();
-                return;
-            }
-        }
-
-        std::shared_ptr<TRetryContext> context = std::make_shared<TRetryContext>(Stats);
-
-        auto executeOperation = [this, func]() {
-            return InsistentClient.ExecuteWithRetry(func);
-        };
-
-        context->HandleStatusFunc = std::make_unique<std::function<void(const TAsyncFinalStatus& resultFuture)>>(
-            [this, executeOperation, context](const TAsyncFinalStatus& future) mutable {
-            Y_ABORT_UNLESS(future.HasValue());
-            TFinalStatus resultStatus = future.GetValue();
-            if (resultStatus) {
-                // Reply received
-                CheckForError(*resultStatus);
-                if (resultStatus->IsSuccess()) {
-                    //Ok received
-                    Stats.Report(context->LifeTimeStat, resultStatus->GetStatus()); 
-                    DecrementInfly();
-                    context->HandleStatusFunc.reset();
-                    return;
-                }
-            }
-            if (IsStopped() || TInstant::Now() - context->LifeTimeStat.Start > GlobalTimeout) {
-                // Application stopped working or global timeout reached. Ok reply hasn't received yet
-                Stats.Report(context->LifeTimeStat, TInnerStatus::StatusNotFinished);
-                DecrementInfly();
-                context->HandleStatusFunc.reset();
-                return;
-            }
-            Stats.Report(context->PerRequestStat, resultStatus);
-            context->PerRequestStat = Stats.CreateStatUnit();
-            // Retrying:
-            executeOperation().Subscribe(*context->HandleStatusFunc);
-        });
-
-        context->Retries.fetch_add(1);
-        Y_ABORT_UNLESS(context->Retries.load() < 500, "Too much retries");
-
-        executeOperation().Subscribe(*context->HandleStatusFunc);
-    };
-
-    if (IsStopped()) {
-        return false;
-    }
-
-    bool CanLaunchJob = false;
-
-    with_lock(Lock) {
-        if (!AllJobsLaunched) {
-            CanLaunchJob = true;
-            ++Waiting;
-        }
-    }
-
-    if (CanLaunchJob) {
-        if (!InputQueue->AddFunc(threadFunc)) {
-            DecrementWaiting();
-        }
-    }
-    ++InProgressCount;
-    InProgressSum += InputQueue->Size();
-    return true;
-}

ydb/public/sdk/cpp/tests/slo_workloads/utils/executor.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "statistics.h"
 #include "utils.h"
 
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/table/table.h>
@@ -56,7 +57,7 @@ class TInsistentClient {
         bool Valid = true;
     };
 
-    struct TOperationContext : public TThrRefBase {
+    struct TOperationContext {
         bool Finished = false;
         TAdaptiveLock Lock;
         TCheckedIterator RetryIter;
@@ -131,7 +132,6 @@ class TExecutor {
     std::uint32_t Wait(TDuration waitTimeout);
     bool IsStopped();
     void Finish();
-    std::uint32_t GetTotal() const;
     void Report(TStringBuilder& out) const;
 
 protected:
@@ -167,8 +167,6 @@ class TExecutor {
     TInstant Deadline;
     // Last second we reported Infly
     std::uint64_t LastReportSec = 0;
-    // Max infly for current second
-    std::uint64_t MaxSecInfly = 0;
     // Max Active sessions for current second
     std::uint64_t MaxSecSessions = 0;
 
@@ -178,21 +176,3 @@ class TExecutor {
     std::uint64_t MaxSecReadPromises = 0;
     std::uint64_t MaxSecExecutorPromises = 0;
 };
-
-class TExecutorWithRetry : public TExecutor {
-public:
-    struct TRetryContext {
-        TRetryContext(TStat& stat)
-            : LifeTimeStat(stat.CreateStatUnit())
-            , PerRequestStat(stat.CreateStatUnit())
-        {}
-
-        TStatUnit LifeTimeStat;
-        TStatUnit PerRequestStat;
-        std::unique_ptr<std::function<void(const TAsyncFinalStatus& resultFuture)>> HandleStatusFunc;
-        std::atomic<std::uint64_t> Retries = 0;
-    };
-
-    TExecutorWithRetry(const TCommonOptions& opts, TStat& stats);
-    bool Execute(const NYdb::NTable::TTableClient::TOperationFunc& func) override;
-};

ydb/public/sdk/cpp/tests/slo_workloads/utils/job.cpp

@@ -5,18 +5,12 @@
 const std::string LogFileName = "benchmark.log";
 
 
-TThreadJob::TThreadJob(const TCommonOptions& opts)
+TThreadJob::TThreadJob(const TCommonOptions& opts, const std::string& operationType)
     : RpsProvider(opts.Rps)
     , Prefix(opts.DatabaseOptions.Prefix)
-    , Stats(
-        opts.ReactionTime,
-        opts.ResultFileName,
-        !opts.DontPushMetrics,
-        opts.RetryMode
-    )
+    , Stats(opts.DontPushMetrics ? std::nullopt : std::make_optional(opts.MetricsPushUrl), operationType)
     , StopOnError(opts.StopOnError)
     , MaxDelay(opts.ReactionTime)
-    , UseFollowers(opts.UseFollowers)
 {
 }
 
@@ -28,7 +22,7 @@ void TThreadJob::Start(TInstant deadline) {
 
 void TThreadJob::StartThread() {
     auto threadFunc = [this]() {
-        Stats.Reset();
+        Stats.Start();
         RpsProvider.Reset();
         DoJob();
         Stats.Finish();

ydb/public/sdk/cpp/tests/slo_workloads/utils/job.h

@@ -1,13 +1,14 @@
 #pragma once
 
+#include "statistics.h"
 #include "utils.h"
 
 #include <util/string/builder.h>
 #include <util/thread/pool.h>
 
 class TThreadJob {
 public:
-    TThreadJob(const TCommonOptions& opts);
+    TThreadJob(const TCommonOptions& opts, const std::string& operationType);
     virtual ~TThreadJob() = default;
 
     virtual void Start(TInstant deadline);
@@ -31,10 +32,9 @@ class TThreadJob {
     TStat Stats;
     bool StopOnError;
     TDuration MaxDelay;
-    bool UseFollowers;
 };
 
-class TJobContainer : public TThrRefBase {
+class TJobContainer {
 public:
     void Add(TThreadJob* job);
     void Start(TInstant deadline = TInstant());