KIKIMR-19215: additional counters for external blob storages

.mapping.json

@@ -5269,6 +5269,11 @@
   "ydb/core/tx/columnshard/blobs_action/bs/CMakeLists.linux-x86_64.txt":"",
   "ydb/core/tx/columnshard/blobs_action/bs/CMakeLists.txt":"",
   "ydb/core/tx/columnshard/blobs_action/bs/CMakeLists.windows-x86_64.txt":"",
+  "ydb/core/tx/columnshard/blobs_action/counters/CMakeLists.darwin-x86_64.txt":"",
+  "ydb/core/tx/columnshard/blobs_action/counters/CMakeLists.linux-aarch64.txt":"",
+  "ydb/core/tx/columnshard/blobs_action/counters/CMakeLists.linux-x86_64.txt":"",
+  "ydb/core/tx/columnshard/blobs_action/counters/CMakeLists.txt":"",
+  "ydb/core/tx/columnshard/blobs_action/counters/CMakeLists.windows-x86_64.txt":"",
   "ydb/core/tx/columnshard/blobs_action/tier/CMakeLists.darwin-x86_64.txt":"",
   "ydb/core/tx/columnshard/blobs_action/tier/CMakeLists.linux-aarch64.txt":"",
   "ydb/core/tx/columnshard/blobs_action/tier/CMakeLists.linux-x86_64.txt":"",

ydb/core/tx/columnshard/blobs_action/CMakeLists.darwin-x86_64.txt

@@ -8,6 +8,7 @@
 
 add_subdirectory(abstract)
 add_subdirectory(bs)
+add_subdirectory(counters)
 add_subdirectory(tier)
 add_subdirectory(transaction)
 
@@ -19,8 +20,9 @@ target_link_libraries(tx-columnshard-blobs_action PUBLIC
   libs-apache-arrow
   ydb-core-tablet_flat
   core-tx-tiering
-  columnshard-blobs_action-bs
   columnshard-blobs_action-abstract
+  columnshard-blobs_action-bs
+  columnshard-blobs_action-counters
   columnshard-blobs_action-transaction
   columnshard-blobs_action-tier
 )

ydb/core/tx/columnshard/blobs_action/CMakeLists.linux-aarch64.txt

@@ -8,6 +8,7 @@
 
 add_subdirectory(abstract)
 add_subdirectory(bs)
+add_subdirectory(counters)
 add_subdirectory(tier)
 add_subdirectory(transaction)
 
@@ -20,8 +21,9 @@ target_link_libraries(tx-columnshard-blobs_action PUBLIC
   libs-apache-arrow
   ydb-core-tablet_flat
   core-tx-tiering
-  columnshard-blobs_action-bs
   columnshard-blobs_action-abstract
+  columnshard-blobs_action-bs
+  columnshard-blobs_action-counters
   columnshard-blobs_action-transaction
   columnshard-blobs_action-tier
 )

ydb/core/tx/columnshard/blobs_action/CMakeLists.linux-x86_64.txt

@@ -8,6 +8,7 @@
 
 add_subdirectory(abstract)
 add_subdirectory(bs)
+add_subdirectory(counters)
 add_subdirectory(tier)
 add_subdirectory(transaction)
 
@@ -20,8 +21,9 @@ target_link_libraries(tx-columnshard-blobs_action PUBLIC
   libs-apache-arrow
   ydb-core-tablet_flat
   core-tx-tiering
-  columnshard-blobs_action-bs
   columnshard-blobs_action-abstract
+  columnshard-blobs_action-bs
+  columnshard-blobs_action-counters
   columnshard-blobs_action-transaction
   columnshard-blobs_action-tier
 )

ydb/core/tx/columnshard/blobs_action/CMakeLists.windows-x86_64.txt

@@ -8,6 +8,7 @@
 
 add_subdirectory(abstract)
 add_subdirectory(bs)
+add_subdirectory(counters)
 add_subdirectory(transaction)
 
 add_library(tx-columnshard-blobs_action)
@@ -21,8 +22,9 @@ target_link_libraries(tx-columnshard-blobs_action PUBLIC
   libs-apache-arrow
   ydb-core-tablet_flat
   core-tx-tiering
-  columnshard-blobs_action-bs
   columnshard-blobs_action-abstract
+  columnshard-blobs_action-bs
+  columnshard-blobs_action-counters
   columnshard-blobs_action-transaction
 )
 target_sources(tx-columnshard-blobs_action PRIVATE

ydb/core/tx/columnshard/blobs_action/abstract/action.cpp

@@ -4,15 +4,15 @@
 namespace NKikimr::NOlap {
 
 std::shared_ptr<NKikimr::NOlap::IBlobsWritingAction> TBlobsAction::GetWriting(const TPortionInfo& portionInfo) {
-    return GetStorageAction(portionInfo.GetBlobsStorage()->GetStorageId()).GetWriting();
+    return GetStorageAction(portionInfo.GetBlobsStorage()->GetStorageId()).GetWriting(ConsumerId);
 }
 
 std::shared_ptr<NKikimr::NOlap::IBlobsReadingAction> TBlobsAction::GetReading(const TPortionInfo& portionInfo) {
-    return GetStorageAction(portionInfo.GetBlobsStorage()->GetStorageId()).GetReading();
+    return GetStorageAction(portionInfo.GetBlobsStorage()->GetStorageId()).GetReading(ConsumerId);
 }
 
 std::shared_ptr<NKikimr::NOlap::IBlobsDeclareRemovingAction> TBlobsAction::GetRemoving(const TPortionInfo& portionInfo) {
-    return GetStorageAction(portionInfo.GetBlobsStorage()->GetStorageId()).GetRemoving();
+    return GetStorageAction(portionInfo.GetBlobsStorage()->GetStorageId()).GetRemoving(ConsumerId);
 }
 
 }

ydb/core/tx/columnshard/blobs_action/abstract/action.h

@@ -22,24 +22,24 @@ public:
 
     }
 
-    const std::shared_ptr<IBlobsDeclareRemovingAction>& GetRemoving() {
+    const std::shared_ptr<IBlobsDeclareRemovingAction>& GetRemoving(const TString& consumerId) {
         if (!Removing) {
-            Removing = Storage->StartDeclareRemovingAction();
+            Removing = Storage->StartDeclareRemovingAction(consumerId);
         }
         return Removing;
     }
-    const std::shared_ptr<IBlobsWritingAction>& GetWriting() {
+    const std::shared_ptr<IBlobsWritingAction>& GetWriting(const TString& consumerId) {
         if (!Writing) {
-            Writing = Storage->StartWritingAction();
+            Writing = Storage->StartWritingAction(consumerId);
         }
         return Writing;
     }
     const std::shared_ptr<IBlobsWritingAction>& GetWritingOptional() const {
         return Writing;
     }
-    const std::shared_ptr<IBlobsReadingAction>& GetReading() {
+    const std::shared_ptr<IBlobsReadingAction>& GetReading(const TString& consumerId) {
         if (!Reading) {
-            Reading = Storage->StartReadingAction();
+            Reading = Storage->StartReadingAction(consumerId);
         }
         return Reading;
     }
@@ -78,6 +78,7 @@ class TBlobsAction {
 private:
     std::shared_ptr<IStoragesManager> Storages;
     THashMap<TString, TStorageAction> StorageActions;
+    const TString ConsumerId;
 
     TStorageAction& GetStorageAction(const TString& storageId) {
         auto it = StorageActions.find(storageId);
@@ -87,8 +88,9 @@ private:
         return it->second;
     }
 public:
-    TBlobsAction(std::shared_ptr<IStoragesManager> storages)
+    explicit TBlobsAction(std::shared_ptr<IStoragesManager> storages, const TString& consumerId)
         : Storages(storages)
+        , ConsumerId(consumerId)
     {
 
     }
@@ -155,19 +157,19 @@ public:
     }
 
     std::shared_ptr<IBlobsDeclareRemovingAction> GetRemoving(const TString& storageId) {
-        return GetStorageAction(storageId).GetRemoving();
+        return GetStorageAction(storageId).GetRemoving(ConsumerId);
     }
 
     std::shared_ptr<IBlobsDeclareRemovingAction> GetRemoving(const TPortionInfo& portionInfo);
 
     std::shared_ptr<IBlobsWritingAction> GetWriting(const TString& storageId) {
-        return GetStorageAction(storageId).GetWriting();
+        return GetStorageAction(storageId).GetWriting(ConsumerId);
     }
 
     std::shared_ptr<IBlobsWritingAction> GetWriting(const TPortionInfo& portionInfo);
 
     std::shared_ptr<IBlobsReadingAction> GetReading(const TString& storageId) {
-        return GetStorageAction(storageId).GetReading();
+        return GetStorageAction(storageId).GetReading(ConsumerId);
     }
 
     std::shared_ptr<IBlobsReadingAction> GetReading(const TPortionInfo& portionInfo);

ydb/core/tx/columnshard/blobs_action/abstract/read.cpp

@@ -5,10 +5,83 @@ namespace NKikimr::NOlap {
 
 void IBlobsReadingAction::StartReading(THashMap<TUnifiedBlobId, THashSet<TBlobRange>>&& ranges) {
     AFL_VERIFY(ranges.size());
+    AFL_VERIFY(Counters);
     for (auto&& i : ranges) {
         AFL_VERIFY(i.second.size());
+        for (auto&& br : i.second) {
+            Counters->OnRequest(br.Size);
+        }
     }
     return DoStartReading(ranges);
 }
 
+void IBlobsReadingAction::ExtractBlobsDataTo(THashMap<TBlobRange, TString>& result) {
+    AFL_VERIFY(Started);
+    if (result.empty()) {
+        std::swap(result, Replies);
+    } else {
+        for (auto&& i : Replies) {
+            AFL_VERIFY(result.emplace(i.first, std::move(i.second)).second);
+        }
+        Replies.clear();
+    }
+    RangesForResult.clear();
+}
+
+void IBlobsReadingAction::Start(const THashSet<TBlobRange>& rangesInProgress) {
+    Y_VERIFY(!Started);
+    Y_VERIFY(RangesForRead.size() + RangesForResult.size());
+    for (auto&& i : RangesForRead) {
+        for (auto&& r : i.second) {
+            WaitingRanges.emplace(r, TMonotonic::Now());
+        }
+    }
+    THashMap<TUnifiedBlobId, THashSet<TBlobRange>> rangesFiltered;
+    if (rangesInProgress.empty()) {
+        rangesFiltered = RangesForRead;
+    } else {
+        for (auto&& i : RangesForRead) {
+            for (auto&& r : i.second) {
+                if (!rangesInProgress.contains(r)) {
+                    rangesFiltered[r.BlobId].emplace(r);
+                }
+            }
+        }
+    }
+    if (rangesFiltered.size()) {
+        StartReading(std::move(rangesFiltered));
+    }
+    Started = true;
+    for (auto&& i : RangesForResult) {
+        AFL_VERIFY(Replies.emplace(i.first, i.second).second);
+    }
+}
+
+void IBlobsReadingAction::OnReadResult(const TBlobRange& range, const TString& data) {
+    AFL_VERIFY(Counters);
+    auto it = WaitingRanges.find(range);
+    Y_VERIFY(it != WaitingRanges.end());
+    Counters->OnReply(range.Size, TMonotonic::Now() - it->second);
+    WaitingRanges.erase(it);
+    Replies.emplace(range, data);
+}
+
+void IBlobsReadingAction::OnReadError(const TBlobRange& range, const TErrorStatus& replyStatus) {
+    AFL_VERIFY(Counters);
+    auto it = WaitingRanges.find(range);
+    Y_VERIFY(it != WaitingRanges.end());
+    Counters->OnFail(range.Size, TMonotonic::Now() - it->second);
+    WaitingRanges.erase(it);
+    Fails.emplace(range, replyStatus);
+}
+
+void IBlobsReadingAction::AddRange(const TBlobRange& range, const TString& result /*= Default<TString>()*/) {
+    Y_VERIFY(!Started);
+    if (!result) {
+        Y_VERIFY(RangesForRead[range.BlobId].emplace(range).second);
+    } else {
+        Y_VERIFY(RangesForResult.emplace(range, result).second);
+    }
+}
+
 }

ydb/core/tx/columnshard/blobs_action/abstract/read.h

@@ -1,29 +1,43 @@
 #pragma once
 #include "common.h"
+#include <ydb/core/tx/columnshard/blobs_action/counters/read.h>
 #include <ydb/core/tx/columnshard/blob.h>
 #include <ydb/core/protos/base.pb.h>
+#include <ydb/library/conclusion/status.h>
 #include <util/generic/hash_set.h>
 
 namespace NKikimr::NOlap {
 
 class IBlobsReadingAction: public ICommonBlobsAction {
+public:
+    using TErrorStatus = TConclusionSpecialStatus<NKikimrProto::EReplyStatus, NKikimrProto::EReplyStatus::OK, NKikimrProto::EReplyStatus::ERROR>;
 private:
     using TBase = ICommonBlobsAction;
+
     THashMap<TUnifiedBlobId, THashSet<TBlobRange>> RangesForRead;
-    THashSet<TBlobRange> WaitingRanges;
+    THashMap<TBlobRange, TString> RangesForResult;
+    THashMap<TBlobRange, TMonotonic> WaitingRanges;
     THashMap<TBlobRange, TString> Replies;
-    THashMap<TBlobRange, NKikimrProto::EReplyStatus> Fails;
+    THashMap<TBlobRange, TErrorStatus> Fails;
+    std::shared_ptr<NBlobOperations::TReadCounters> Counters;
     bool Started = false;
 protected:
     virtual void DoStartReading(const THashMap<TUnifiedBlobId, THashSet<TBlobRange>>& range) = 0;
     void StartReading(THashMap<TUnifiedBlobId, THashSet<TBlobRange>>&& ranges);
 public:
+
+    void SetCounters(std::shared_ptr<NBlobOperations::TReadCounters> counters) {
+        Counters = counters;
+    }
+
     IBlobsReadingAction(const TString& storageId)
         : TBase(storageId)
     {
 
     }
 
+    void ExtractBlobsDataTo(THashMap<TBlobRange, TString>& result);
+
     ui64 GetExpectedBlobsSize() const {
         ui64 result = 0;
         for (auto&& i : RangesForRead) {
@@ -31,6 +45,9 @@ public:
                 result += b.Size;
             }
         }
+        for (auto&& i : RangesForResult) {
+            result += i.first.Size;
+        }
         return result;
     }
 
@@ -39,7 +56,7 @@ public:
         for (auto&& i : RangesForRead) {
             result += i.second.size();
         }
-        return result;
+        return result + RangesForResult.size();
     }
 
     void FillExpectedRanges(THashSet<TBlobRange>& ranges) const {
@@ -48,52 +65,20 @@ public:
                 Y_VERIFY(ranges.emplace(b).second);
             }
         }
+        for (auto&& i : RangesForResult) {
+            Y_VERIFY(ranges.emplace(i.first).second);
+        }
     }
 
     const THashMap<TUnifiedBlobId, THashSet<TBlobRange>>& GetRangesForRead() const {
         return RangesForRead;
     }
 
-    void AddRange(const TBlobRange& range) {
-        Y_VERIFY(!Started);
-        Y_VERIFY(RangesForRead[range.BlobId].emplace(range).second);
-    }
+    void AddRange(const TBlobRange& range, const TString& result = Default<TString>());
 
-    void Start(const THashSet<TBlobRange>& rangesInProgress) {
-        Y_VERIFY(!Started);
-        Y_VERIFY(RangesForRead.size());
-        for (auto&& i : RangesForRead) {
-            for (auto&& r : i.second) {
-                WaitingRanges.emplace(r);
-            }
-        }
-        THashMap<TUnifiedBlobId, THashSet<TBlobRange>> rangesFiltered;
-        if (rangesInProgress.empty()) {
-            rangesFiltered = RangesForRead;
-        } else {
-            for (auto&& i : RangesForRead) {
-                for (auto&& r : i.second) {
-                    if (!rangesInProgress.contains(r)) {
-                        rangesFiltered[r.BlobId].emplace(r);
-                    }
-                }
-            }
-        }
-        if (rangesFiltered.size()) {
-            StartReading(std::move(rangesFiltered));
-        }
-        Started = true;
-    }
-
-    void OnReadResult(const TBlobRange& range, const TString& data) {
-        Y_VERIFY(WaitingRanges.erase(range));
-        Replies.emplace(range, data);
-    }
-
-    void OnReadError(const TBlobRange& range, const NKikimrProto::EReplyStatus replyStatus) {
-        Y_VERIFY(WaitingRanges.erase(range));
-        Fails.emplace(range, replyStatus);
-    }
+    void Start(const THashSet<TBlobRange>& rangesInProgress);
+    void OnReadResult(const TBlobRange& range, const TString& data);
+    void OnReadError(const TBlobRange& range, const TErrorStatus& replyStatus);
 
     bool HasFails() const {
         return Fails.size();

ydb/core/tx/columnshard/blobs_action/abstract/storage.h

@@ -4,6 +4,7 @@
 #include "read.h"
 
 #include <ydb/core/tx/columnshard/blobs_action/blob_manager_db.h>
+#include <ydb/core/tx/columnshard/blobs_action/counters/storage.h>
 #include <ydb/library/accessor/accessor.h>
 
 namespace NKikimr::NColumnShard {
@@ -34,6 +35,7 @@ private:
         Y_VERIFY(GCActivity);
         GCActivity = false;
     }
+    std::shared_ptr<NBlobOperations::TStorageCounters> Counters;
 protected:
     virtual std::shared_ptr<IBlobsDeclareRemovingAction> DoStartDeclareRemovingAction() = 0;
     virtual std::shared_ptr<IBlobsWritingAction> DoStartWritingAction() = 0;
@@ -47,8 +49,9 @@ protected:
     }
 public:
     IBlobsStorageOperator(const TString& storageId)
-        : StorageId(storageId) {
-
+        : StorageId(storageId)
+    {
+        Counters = std::make_shared<NBlobOperations::TStorageCounters>(storageId);
     }
 
     virtual std::shared_ptr<IBlobInUseTracker> GetBlobsTracker() const = 0;
@@ -65,14 +68,19 @@ public:
     void OnTieringModified(const std::shared_ptr<NColumnShard::TTiersManager>& tiers) {
         return DoOnTieringModified(tiers);
     }
-    std::shared_ptr<IBlobsDeclareRemovingAction> StartDeclareRemovingAction() {
+
+    std::shared_ptr<IBlobsDeclareRemovingAction> StartDeclareRemovingAction(const TString& /*consumerId*/) {
         return DoStartDeclareRemovingAction();
     }
-    std::shared_ptr<IBlobsWritingAction> StartWritingAction() {
-        return DoStartWritingAction();
+    std::shared_ptr<IBlobsWritingAction> StartWritingAction(const TString& consumerId) {
+        auto result = DoStartWritingAction();
+        result->SetCounters(Counters->GetConsumerCounter(consumerId)->GetWriteCounters());
+        return result;
     }
-    std::shared_ptr<IBlobsReadingAction> StartReadingAction() {
-        return DoStartReadingAction();
+    std::shared_ptr<IBlobsReadingAction> StartReadingAction(const TString& consumerId) {
+        auto result = DoStartReadingAction();
+        result->SetCounters(Counters->GetConsumerCounter(consumerId)->GetReadCounters());
+        return result;
     }
     bool StartGC() {
         if (!GCActivity) {