25-1-4: Import counters (#27315)

Author: CyberROFL

ydb/core/tx/datashard/export_scan.cpp

@@ -32,7 +32,7 @@ class TExportScan: private NActors::IActorCallback, public NTable::IScan {
         TStats()
             : IBuffer::TStats()
         {
-            auto counters = GetServiceCounters(AppData()->Counters, "tablets")->GetSubgroup("subsystem", "store_to_yt");
+            auto counters = GetServiceCounters(AppData()->Counters, "tablets")->GetSubgroup("subsystem", "export");
 
             MonRows = counters->GetCounter("Rows", true);
             MonBytesRead = counters->GetCounter("BytesRead", true);

ydb/core/tx/datashard/import_s3.cpp

@@ -8,6 +8,7 @@
 
 #include <ydb/core/backup/common/checksum.h>
 #include <ydb/core/base/appdata.h>
+#include <ydb/core/base/counters.h>
 #include <ydb/core/protos/datashard_config.pb.h>
 #include <ydb/core/protos/flat_scheme_op.pb.h>
 #include <ydb/library/services/services.pb.h>
@@ -296,25 +297,80 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
             for (ui32 id : tableInfo.GetValueColumnIds(columnNames)) {
                 rowScheme.AddValueColumnIds(id);
             }
+
+            CellBytes = 0;
         }
 
         void AddRow(const TVector<TCell>& keys, const TVector<TCell>& values) {
             Y_ABORT_UNLESS(Record);
             auto& row = *Record->AddRows();
             row.SetKeyColumns(TSerializedCellVec::Serialize(keys));
             row.SetValueColumns(TSerializedCellVec::Serialize(values));
+
+            for (const auto& x : keys) {
+                CellBytes += x.Size();
+            }
+            for (const auto& x : values) {
+                CellBytes += x.Size();
+            }
         }
 
         const std::shared_ptr<NKikimrTxDataShard::TEvUploadRowsRequest>& GetRecord() {
             Y_ABORT_UNLESS(Record);
             return Record;
         }
 
+        ui64 GetCellBytes() const {
+            return CellBytes;
+        }
+
     private:
         std::shared_ptr<NKikimrTxDataShard::TEvUploadRowsRequest> Record;
+        ui64 CellBytes;
 
     }; // TUploadRowsRequestBuilder
 
+    struct TCounters {
+        struct TLatency {
+            TInstant Begin;
+            ::NMonitoring::THistogramPtr Counter;
+
+            explicit TLatency(::NMonitoring::THistogramPtr counter)
+                : Counter(counter)
+            {
+            }
+
+            void Start(TInstant begin) {
+                Begin = begin;
+            }
+
+            void Finish(TInstant end) {
+                Counter->Collect((end - Begin).MilliSeconds());
+                Begin = TInstant::Zero();
+            }
+        };
+
+        ::NMonitoring::TDynamicCounters::TCounterPtr BytesReceived;
+        ::NMonitoring::TDynamicCounters::TCounterPtr BytesWritten;
+        TLatency LatencyRead;
+        TLatency LatencyProcess;
+        TLatency LatencyWrite;
+
+        explicit TCounters(::NMonitoring::TDynamicCounterPtr counters)
+            : BytesReceived(counters->GetCounter("BytesReceived", true))
+            , BytesWritten(counters->GetCounter("BytesWritten", true))
+            , LatencyRead(counters->GetHistogram("LatencyReadMs", ::NMonitoring::ExponentialHistogram(10, 4, 1)))
+            , LatencyProcess(counters->GetHistogram("LatencyProcessMs", ::NMonitoring::ExponentialHistogram(10, 4, 1)))
+            , LatencyWrite(counters->GetHistogram("LatencyWriteMs", ::NMonitoring::ExponentialHistogram(10, 4, 1)))
+        {
+        }
+
+    }; // TCounters
+
+    static TInstant Now() {
+        return TlsActivationContext->Now();
+    }
+
     void AllocateResource() {
         IMPORT_LOG_D("AllocateResource");
 
@@ -347,7 +403,6 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
             Send(client, new TEvents::TEvPoisonPill());
         }
 
-
         Client = RegisterWithSameMailbox(CreateS3Wrapper(ExternalStorageConfig->ConstructStorageOperator()));
 
         HeadObject(Settings.GetDataKey(DataFormat, CompressionCodec));
@@ -490,6 +545,9 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
             << ", content-length# " << ContentLength
             << ", body-size# " << msg.Body.size());
 
+        *Counters.BytesReceived += msg.Body.size();
+        Counters.LatencyRead.Finish(Now());
+
         Reader->Feed(std::move(msg.Body));
         Process();
     }
@@ -534,6 +592,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
 
         case TReadController::NOT_ENOUGH_DATA:
             if (SumWithSaturation(ProcessedBytes, Reader->PendingBytes()) < ContentLength) {
+                Counters.LatencyRead.Start(Now());
                 return GetObject(Settings.GetDataKey(DataFormat, CompressionCodec),
                     Reader->NextRange(ContentLength, ProcessedBytes));
             } else {
@@ -546,6 +605,8 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
                 << ": " << error);
         }
 
+        Counters.LatencyProcess.Start(Now());
+
         if (Checksum) {
             Checksum->AddData(data);
         }
@@ -615,6 +676,9 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
             << ": count# " << record->RowsSize()
             << ", size# " << record->ByteSizeLong());
 
+        Counters.LatencyProcess.Finish(Now());
+        Counters.LatencyWrite.Start(Now());
+
         Send(DataShard, new TEvDataShard::TEvS3UploadRowsRequest(TxId, record, {
             ETag, ProcessedBytes, WrittenBytes, WrittenRows, GetChecksumState()
         }));
@@ -623,6 +687,9 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
     void Handle(TEvDataShard::TEvS3UploadRowsResponse::TPtr& ev) {
         IMPORT_LOG_D("Handle " << ev->Get()->ToString());
 
+        *Counters.BytesWritten += RequestBuilder.GetCellBytes();
+        Counters.LatencyWrite.Finish(Now());
+
         const auto& record = ev->Get()->Record;
         switch (record.GetStatus()) {
         case NKikimrTxDataShard::TError::OK:
@@ -816,6 +883,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
         , ReadBatchSize(task.GetS3Settings().GetLimits().GetReadBatchSize())
         , ReadBufferSizeLimit(AppData()->DataShardConfig.GetRestoreReadBufferSizeLimit())
         , Checksum(task.GetValidateChecksums() ? CreateChecksum() : nullptr)
+        , Counters(GetServiceCounters(AppData()->Counters, "tablets")->GetSubgroup("subsystem", "import"))
     {
     }
 
@@ -895,6 +963,9 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
 
     NBackup::IChecksum::TPtr Checksum;
     TString ExpectedChecksum;
+
+    TCounters Counters;
+
 }; // TS3Downloader
 
 IActor* CreateS3Downloader(const TActorId& dataShard, ui64 txId, const NKikimrSchemeOp::TRestoreTask& task, const TTableInfo& info) {