Profile events for task distribution in ObjectStorageCluster requests

Author: ianton-ru

src/Common/ProfileEvents.cpp

@@ -1159,6 +1159,11 @@ The server successfully detected this situation and will download merged part fr
     M(ObjectStorageListObjectsCachePrefixMatchHits, "Number of times object storage list objects operation miss the cache using prefix matching.", ValueType::Number) \
     M(ParquetMetaDataCacheHits, "Number of times the read from filesystem cache hit the cache.", ValueType::Number) \
     M(ParquetMetaDataCacheMisses, "Number of times the read from filesystem cache miss the cache.", ValueType::Number) \
+    \
+    M(ObjectStorageClusterSentToMatchedReplica, "Number of tasks in ObjectStorageCluster request sent on matched replica.", ValueType::Number) \
+    M(ObjectStorageClusterSentToNonMatchedReplica, "Number of tasks in ObjectStorageCluster request sent on non-matched replica.", ValueType::Number) \
+    M(ObjectStorageClusterProcessedTasks, "Number of processed tasks in ObjectStorageCluster request.", ValueType::Number) \
+    M(ObjectStorageClusterWaitingMicroseconds, "Time of waiting for tasks in ObjectStorageCluster request.", ValueType::Microseconds) \
 
 #ifdef APPLY_FOR_EXTERNAL_EVENTS
     #define APPLY_FOR_EVENTS(M) APPLY_FOR_BUILTIN_EVENTS(M) APPLY_FOR_EXTERNAL_EVENTS(M)

src/Storages/ObjectStorage/StorageObjectStorageSource.cpp

@@ -45,6 +45,8 @@ namespace fs = std::filesystem;
 namespace ProfileEvents
 {
     extern const Event EngineFileLikeReadFiles;
+    extern const Event ObjectStorageClusterProcessedTasks;
+    extern const Event ObjectStorageClusterWaitingMicroseconds;
 }
 
 namespace CurrentMetrics
@@ -507,6 +509,7 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade
                 /// TODO: Make asyncronous waiting without sleep in thread
                 /// Now this sleep is on executor node in worker thread
                 /// Does not block query initiator
+                ProfileEvents::increment(ProfileEvents::ObjectStorageClusterWaitingMicroseconds, retry_after_us.value());
                 sleepForMicroseconds(std::min(Poco::Timestamp::TimeDiff(100000ul), retry_after_us.value()));
                 continue;
             }
@@ -519,6 +522,8 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade
     }
     while (not_a_path || (query_settings.skip_empty_files && object_info->metadata->size_bytes == 0));
 
+    ProfileEvents::increment(ProfileEvents::ObjectStorageClusterProcessedTasks);
+
     ObjectStoragePtr storage_to_use = object_info->getObjectStorage();
     if (!storage_to_use)
         storage_to_use = object_storage;

src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp

@@ -3,6 +3,12 @@
 #include <consistent_hashing.h>
 #include <optional>
 
+namespace ProfileEvents
+{
+    extern const Event ObjectStorageClusterSentToMatchedReplica;
+    extern const Event ObjectStorageClusterSentToNonMatchedReplica;
+};
+
 namespace DB
 {
 
@@ -125,6 +131,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getPreQueuedFile(size_t
             number_of_current_replica
         );
 
+        ProfileEvents::increment(ProfileEvents::ObjectStorageClusterSentToMatchedReplica);
         return next_file;
     }
 
@@ -176,6 +183,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getMatchingFileFromIter
                 file_path, number_of_current_replica
             );
 
+            ProfileEvents::increment(ProfileEvents::ObjectStorageClusterSentToMatchedReplica);
             return object_info;
         }
         LOG_TEST(
@@ -229,6 +237,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getAnyUnprocessedFile(s
                     number_of_current_replica
                 );
 
+                ProfileEvents::increment(ProfileEvents::ObjectStorageClusterSentToNonMatchedReplica);
                 return next_file;
             }