Moved commit "Replication stats (total & per item): lag, initial scan progress" from ydb repo

Author: CyberROFL

include/ydb-cpp-sdk/client/draft/ydb_replication.h

@@ -10,6 +10,7 @@
 namespace Ydb::Replication {
     class ConnectionParams;
     class DescribeReplicationResult;
+    class DescribeReplicationResult_Stats;
 }
 
 namespace NYdb {
@@ -24,7 +25,11 @@ namespace NYdb::NReplication {
 
 class TDescribeReplicationResult;
 using TAsyncDescribeReplicationResult = NThreading::TFuture<TDescribeReplicationResult>;
-struct TDescribeReplicationSettings: public TOperationRequestSettings<TDescribeReplicationSettings> {};
+
+struct TDescribeReplicationSettings: public TOperationRequestSettings<TDescribeReplicationSettings> {
+    using TSelf = TDescribeReplicationSettings;
+    FLUENT_SETTING_DEFAULT(bool, IncludeStats, false);
+};
 
 struct TStaticCredentials {
     std::string User;
@@ -58,15 +63,28 @@ class TConnectionParams: private TCommonClientSettings {
     > Credentials_;
 };
 
-struct TRunningState {
+class TStats {
 public:
-    TRunningState() = default;
-    explicit TRunningState(const std::optional<TDuration>& lag);
+    TStats() = default;
+    TStats(const Ydb::Replication::DescribeReplicationResult_Stats& stats);
 
     const std::optional<TDuration>& GetLag() const;
+    const std::optional<float>& GetInitialScanProgress() const;
 
 private:
     std::optional<TDuration> Lag_;
+    std::optional<float> InitialScanProgress_;
+};
+
+class TRunningState {
+public:
+    TRunningState() = default;
+    explicit TRunningState(const TStats& stats);
+
+    const TStats& GetStats() const;
+
+private:
+    TStats Stats_;
 };
 
 struct TDoneState {};
@@ -89,6 +107,7 @@ class TReplicationDescription {
         uint64_t Id;
         std::string SrcPath;
         std::string DstPath;
+        TStats Stats;
         std::optional<std::string> SrcChangefeedName;
     };
 

include/ydb-cpp-sdk/client/table/table.h

@@ -271,8 +271,11 @@ class TChangefeedDescription {
 public:
     class TInitialScanProgress {
     public:
+        TInitialScanProgress();
         explicit TInitialScanProgress(uint32_t total, uint32_t completed);
 
+        TInitialScanProgress& operator+=(const TInitialScanProgress& other);
+
         uint32_t GetPartsTotal() const;
         uint32_t GetPartsCompleted() const;
         float GetProgress() const; // percentage

src/client/draft/ydb_replication.cpp

@@ -59,15 +59,33 @@ const TOAuthCredentials& TConnectionParams::GetOAuthCredentials() const {
     return std::get<TOAuthCredentials>(Credentials_);
 }
 
-TRunningState::TRunningState(const std::optional<TDuration>& lag)
-    : Lag_(lag)
+static TDuration DurationToDuration(const google::protobuf::Duration& value) {
+    return TDuration::MilliSeconds(google::protobuf::util::TimeUtil::DurationToMilliseconds(value));
+}
+
+TStats::TStats(const Ydb::Replication::DescribeReplicationResult_Stats& stats)
+    : Lag_(stats.has_lag() ? std::make_optional(DurationToDuration(stats.lag())) : std::nullopt)
+    , InitialScanProgress_(stats.has_initial_scan_progress() ? std::make_optional(stats.initial_scan_progress()) : std::nullopt)
 {
 }
 
-const std::optional<TDuration>& TRunningState::GetLag() const {
+const std::optional<TDuration>& TStats::GetLag() const {
     return Lag_;
 }
 
+const std::optional<float>& TStats::GetInitialScanProgress() const {
+    return InitialScanProgress_;
+}
+
+TRunningState::TRunningState(const TStats& stats)
+    : Stats_(stats)
+{
+}
+
+const TStats& TRunningState::GetStats() const {
+    return Stats_;
+}
+
 class TErrorState::TImpl {
 public:
     NYql::TIssues Issues;
@@ -87,10 +105,6 @@ const NYql::TIssues& TErrorState::GetIssues() const {
     return Impl_->Issues;
 }
 
-TDuration DurationToDuration(const google::protobuf::Duration& value) {
-    return TDuration::MilliSeconds(google::protobuf::util::TimeUtil::DurationToMilliseconds(value));
-}
-
 template <typename T>
 NYql::TIssues IssuesFromMessage(const ::google::protobuf::RepeatedPtrField<T>& message) {
     NYql::TIssues issues;
@@ -107,15 +121,15 @@ TReplicationDescription::TReplicationDescription(const Ydb::Replication::Describ
             .Id = item.id(),
             .SrcPath = item.source_path(),
             .DstPath = item.destination_path(),
+            .Stats = TStats(item.stats()),
             .SrcChangefeedName = item.has_source_changefeed_name()
                 ? std::make_optional(item.source_changefeed_name()) : std::nullopt,
         });
     }
 
     switch (desc.state_case()) {
     case Ydb::Replication::DescribeReplicationResult::kRunning:
-        State_ = TRunningState(desc.running().has_lag()
-            ? std::make_optional(DurationToDuration(desc.running().lag())) : std::nullopt);
+        State_ = TRunningState(desc.running().stats());
         break;
 
     case Ydb::Replication::DescribeReplicationResult::kError:
@@ -183,6 +197,7 @@ class TReplicationClient::TImpl: public TClientImplCommon<TReplicationClient::TI
 
         auto request = MakeOperationRequest<DescribeReplicationRequest>(settings);
         request.set_path(TStringType{path});
+        request.set_include_stats(settings.IncludeStats_);
 
         auto promise = NThreading::NewPromise<TDescribeReplicationResult>();
 

src/client/table/table.cpp

@@ -2398,11 +2398,22 @@ TChangefeedDescription::TChangefeedDescription(const Ydb::Table::ChangefeedDescr
     : TChangefeedDescription(FromProto(proto))
 {}
 
+TChangefeedDescription::TInitialScanProgress::TInitialScanProgress()
+    : PartsTotal(0)
+    , PartsCompleted(0)
+{}
+
 TChangefeedDescription::TInitialScanProgress::TInitialScanProgress(uint32_t total, uint32_t completed)
     : PartsTotal(total)
     , PartsCompleted(completed)
 {}
 
+TChangefeedDescription::TInitialScanProgress& TChangefeedDescription::TInitialScanProgress::operator+=(const TInitialScanProgress& other) {
+    PartsTotal += other.PartsTotal;
+    PartsCompleted += other.PartsCompleted;
+    return *this;
+}
+
 uint32_t TChangefeedDescription::TInitialScanProgress::GetPartsTotal() const {
     return PartsTotal;
 }