YQ-3766 Shared reading: add unread stats (#10635)

Author: kardymonds

ydb/core/fq/libs/row_dispatcher/actors_factory.cpp

@@ -10,6 +10,8 @@ struct TActorFactory : public IActorFactory {
 
     NActors::TActorId RegisterTopicSession(
         const TString& topicPath,
+        const TString& endpoint,
+        const TString& database,
         const NConfig::TRowDispatcherConfig& config,
         NActors::TActorId rowDispatcherActorId,
         ui32 partitionId,
@@ -20,6 +22,8 @@ struct TActorFactory : public IActorFactory {
 
         auto actorPtr = NFq::NewTopicSession(
             topicPath,
+            endpoint,
+            database,
             config,
             rowDispatcherActorId,
             partitionId,

ydb/core/fq/libs/row_dispatcher/actors_factory.h

@@ -13,6 +13,8 @@ struct IActorFactory : public TThrRefBase {
 
     virtual NActors::TActorId RegisterTopicSession(
         const TString& topicPath,
+        const TString& endpoint,
+        const TString& database,
         const NConfig::TRowDispatcherConfig& config,
         NActors::TActorId rowDispatcherActorId,
         ui32 partitionId,

ydb/core/fq/libs/row_dispatcher/events/data_plane.h

@@ -6,6 +6,7 @@
 
 #include <ydb/core/fq/libs/row_dispatcher/protos/events.pb.h>
 #include <ydb/library/yql/providers/pq/proto/dq_io.pb.h>
+#include <ydb/core/fq/libs/row_dispatcher/events/topic_session_stats.h>
 
 namespace NFq {
 
@@ -26,6 +27,7 @@ struct TEvRowDispatcher {
         EvCoordinatorChangesSubscribe,
         EvCoordinatorRequest,
         EvCoordinatorResult,
+        EvSessionStatistic,
         EvEnd,
     };
 
@@ -120,6 +122,12 @@ struct TEvRowDispatcher {
         TEvSessionError() = default;
         NActors::TActorId ReadActorId;
     };
+
+    struct TEvSessionStatistic : public NActors::TEventLocal<TEvSessionStatistic, EEv::EvSessionStatistic> {
+        TEvSessionStatistic(const TopicSessionStatistic& stat)
+        : Stat(stat) {}
+        TopicSessionStatistic Stat;
+    };
 };
 
 } // namespace NFq

ydb/core/fq/libs/row_dispatcher/events/topic_session_stats.h

@@ -0,0 +1,33 @@
+#pragma once
+
+#include <ydb/library/actors/core/actorid.h>
+#include <util/generic/vector.h>
+
+namespace NFq {
+
+struct TopicSessionClientStatistic {
+    NActors::TActorId ReadActorId;
+    ui32 PartitionId = 0;
+    i64 UnreadRows = 0;
+    i64 UnreadBytes = 0;
+    ui64 Offset = 0;
+};
+
+struct TopicSessionCommonStatistic {
+    ui64 UnreadBytes = 0;
+};
+
+struct TopicSessionParams {
+    TString Endpoint;
+    TString Database;
+    TString TopicPath;
+    ui64 PartitionId = 0;
+};
+
+struct TopicSessionStatistic {
+    TopicSessionParams SessionKey; 
+    TVector<TopicSessionClientStatistic> Clients;
+    TopicSessionCommonStatistic Common;
+};
+
+} // namespace NFq

ydb/core/fq/libs/row_dispatcher/row_dispatcher.cpp

@@ -6,6 +6,7 @@
 #include <ydb/library/actors/core/hfunc.h>
 #include <ydb/library/actors/core/interconnect.h>
 #include <ydb/library/yql/dq/actors/common/retry_queue.h>
+#include <ydb/library/yql/providers/dq/counters/counters.h>
 
 #include <ydb/core/fq/libs/actors/logging/log.h>
 #include <ydb/core/fq/libs/events/events.h>
@@ -48,16 +49,22 @@ struct TEvPrivate {
     enum EEv : ui32 {
         EvBegin = EventSpaceBegin(NActors::TEvents::ES_PRIVATE),
         EvCoordinatorPing = EvBegin + 20,
-        EvPrintState,
+        EvUpdateMetrics,
         EvEnd
     };
 
     static_assert(EvEnd < EventSpaceEnd(NActors::TEvents::ES_PRIVATE), "expect EvEnd < EventSpaceEnd(NActors::TEvents::ES_PRIVATE)");
     struct TEvCoordinatorPing : NActors::TEventLocal<TEvCoordinatorPing, EvCoordinatorPing> {};
-    struct TEvPrintState : public NActors::TEventLocal<TEvPrintState, EvPrintState> {};
+    struct TEvUpdateMetrics : public NActors::TEventLocal<TEvUpdateMetrics, EvUpdateMetrics> {};
 };
 
-ui64 PrintStatePeriodSec = 60;
+struct TQueryStat {
+    const TString QueryId;
+    NYql::TCounters::TEntry UnreadRows;
+    NYql::TCounters::TEntry UnreadBytes;
+};
+
+ui64 UpdateMetricsPeriodSec = 60;
 
 class TRowDispatcher : public TActorBootstrapped<TRowDispatcher> {
 
@@ -84,19 +91,19 @@ class TRowDispatcher : public TActorBootstrapped<TRowDispatcher> {
     struct TopicSessionKey {
         TString Endpoint;
         TString Database;
-        TString TopicName;
+        TString TopicPath;
         ui64 PartitionId;
 
         size_t Hash() const noexcept {
             ui64 hash = std::hash<TString>()(Endpoint);
             hash = CombineHashes<ui64>(hash, std::hash<TString>()(Database));
-            hash = CombineHashes<ui64>(hash, std::hash<TString>()(TopicName));
+            hash = CombineHashes<ui64>(hash, std::hash<TString>()(TopicPath));
             hash = CombineHashes<ui64>(hash, std::hash<ui64>()(PartitionId));
             return hash;
         }
         bool operator==(const TopicSessionKey& other) const {
             return Endpoint == other.Endpoint && Database == other.Database
-                && TopicName == other.TopicName && PartitionId == other.PartitionId;
+                && TopicPath == other.TopicPath && PartitionId == other.PartitionId;
         }
     };
 
@@ -154,10 +161,12 @@ class TRowDispatcher : public TActorBootstrapped<TRowDispatcher> {
         TActorId TopicSessionId;
         const TString QueryId;
         ConsumerCounters Counters;
+        TopicSessionClientStatistic Stat;
     };
 
     struct SessionInfo {
         TMap<TActorId, TAtomicSharedPtr<ConsumerInfo>> Consumers;     // key - ReadActor actor id
+        TopicSessionCommonStatistic Stat;
     };
 
     struct TopicSessionInfo {
@@ -198,15 +207,16 @@ class TRowDispatcher : public TActorBootstrapped<TRowDispatcher> {
     void Handle(NFq::TEvRowDispatcher::TEvMessageBatch::TPtr& ev);
     void Handle(NFq::TEvRowDispatcher::TEvSessionError::TPtr& ev);
     void Handle(NFq::TEvRowDispatcher::TEvStatus::TPtr& ev);
+    void Handle(NFq::TEvRowDispatcher::TEvSessionStatistic::TPtr& ev);
 
     void Handle(NActors::TEvents::TEvPing::TPtr& ev);
     void Handle(const NYql::NDq::TEvRetryQueuePrivate::TEvRetry::TPtr&);
     void Handle(const NYql::NDq::TEvRetryQueuePrivate::TEvPing::TPtr&);
     void Handle(const NYql::NDq::TEvRetryQueuePrivate::TEvSessionClosed::TPtr&);
-    void Handle(NFq::TEvPrivate::TEvPrintState::TPtr&);
+    void Handle(NFq::TEvPrivate::TEvUpdateMetrics::TPtr&);
 
     void DeleteConsumer(const ConsumerSessionKey& key);
-    void PrintInternalState();
+    void UpdateMetrics();
 
     STRICT_STFUNC(
         StateFunc, {
@@ -223,12 +233,13 @@ class TRowDispatcher : public TActorBootstrapped<TRowDispatcher> {
         hFunc(NFq::TEvRowDispatcher::TEvStopSession, Handle);
         hFunc(NFq::TEvRowDispatcher::TEvSessionError, Handle);
         hFunc(NFq::TEvRowDispatcher::TEvStatus, Handle);
+        hFunc(NFq::TEvRowDispatcher::TEvSessionStatistic, Handle);
         hFunc(NYql::NDq::TEvRetryQueuePrivate::TEvRetry, Handle);
         hFunc(NYql::NDq::TEvRetryQueuePrivate::TEvPing, Handle);
         hFunc(NYql::NDq::TEvRetryQueuePrivate::TEvSessionClosed, Handle);
         hFunc(NActors::TEvents::TEvPing, Handle);
         hFunc(NFq::TEvRowDispatcher::TEvNewDataArrived, Handle);
-        hFunc(NFq::TEvPrivate::TEvPrintState, Handle);
+        hFunc(NFq::TEvPrivate::TEvUpdateMetrics, Handle);
     })
 };
 
@@ -261,7 +272,7 @@ void TRowDispatcher::Bootstrap() {
     auto coordinatorId = Register(NewCoordinator(SelfId(), config, YqSharedResources, Tenant, Counters).release());
     Register(NewLeaderElection(SelfId(), coordinatorId, config, CredentialsProviderFactory, YqSharedResources, Tenant, Counters).release());
     Schedule(TDuration::Seconds(CoordinatorPingPeriodSec), new TEvPrivate::TEvCoordinatorPing());
-    Schedule(TDuration::Seconds(PrintStatePeriodSec), new NFq::TEvPrivate::TEvPrintState());
+    Schedule(TDuration::Seconds(UpdateMetricsPeriodSec), new NFq::TEvPrivate::TEvUpdateMetrics());
 }
 
 void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvCoordinatorChanged::TPtr& ev) {
@@ -320,31 +331,42 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvCoordinatorChangesSubscrib
     Send(ev->Sender, new NFq::TEvRowDispatcher::TEvCoordinatorChanged(*CoordinatorActorId), IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession);
 }
 
-void TRowDispatcher::PrintInternalState() {
+void TRowDispatcher::UpdateMetrics() {
     if (Consumers.empty()) {
         return;
     }
+    TMap<TString, TQueryStat> queryStats;
     TStringStream str;
-    str << "Consumers:\n";
-    for (auto& [key, consumerInfo] : Consumers) {
-        str << "    query id " << consumerInfo->QueryId << ", partId: " << key.PartitionId << ", read actor id: " << key.ReadActorId
-            << ", queueId " << consumerInfo->EventQueueId << ", get next " << consumerInfo->Counters.GetNextBatch
-            << ", data arrived " << consumerInfo->Counters.NewDataArrived << ", message batch " << consumerInfo->Counters.MessageBatch << "\n";
-        str << "        ";
-        consumerInfo->EventsQueue.PrintInternalState(str);
-    }
 
-    str << "\nSessions:\n";
-    for (auto& [key, sessionInfo1] : TopicSessions) {
-        str << "  " << key.Endpoint << " / " << key.Database << " / " << key.TopicName << ", id: " << key.PartitionId << "\n";
-        for (auto& [actorId, sessionInfo2] : sessionInfo1.Sessions) {
-            str << "    session id: " << actorId << "\n";
-            for (auto& [actorId2, consumer] : sessionInfo2.Consumers) {
-                str << "      read actor id: " << actorId2  << "\n";
+    str << "Statistics:\n";
+    for (auto& [key, sessionsInfo] : TopicSessions) {
+        str << "  " << key.Endpoint << " / " << key.Database << " / " << key.TopicPath << " / " << key.PartitionId;
+        for (auto& [actorId, sessionInfo] : sessionsInfo.Sessions) {
+            str << " / " << actorId << "\n";
+            str << "    unread bytes " << sessionInfo.Stat.UnreadBytes << "\n";
+            for (auto& [readActorId, consumer] : sessionInfo.Consumers) {
+                auto& stat = queryStats[consumer->QueryId];
+                stat.UnreadRows.Add(NYql::TCounters::TEntry(consumer->Stat.UnreadRows));
+                stat.UnreadBytes.Add(NYql::TCounters::TEntry(consumer->Stat.UnreadBytes));
+                str << "    " << consumer->QueryId << " " << readActorId << " unread rows "
+                    << consumer->Stat.UnreadRows << " unread bytes " << consumer->Stat.UnreadBytes << " offset " << consumer->Stat.Offset
+                    << " get " << consumer->Counters.GetNextBatch
+                    << " arrived " << consumer->Counters.NewDataArrived << " batch " << consumer->Counters.MessageBatch << " ";
+                str << " retry queue: ";
+                consumer->EventsQueue.PrintInternalState(str);
             }
         }
     }
     LOG_ROW_DISPATCHER_DEBUG(str.Str());
+
+    for (const auto& [queryId, stat] : queryStats) {
+        LOG_ROW_DISPATCHER_DEBUG("UnreadBytes " <<  queryId << " " <<  stat.UnreadBytes.Max);
+        auto queryGroup = Metrics.Counters->GetSubgroup("queryId", queryId);
+        queryGroup->GetCounter("MaxUnreadRows")->Set(stat.UnreadRows.Max);
+        queryGroup->GetCounter("AvgUnreadRows")->Set(stat.UnreadRows.Avg);
+        queryGroup->GetCounter("MaxUnreadBytes")->Set(stat.UnreadBytes.Max);
+        queryGroup->GetCounter("AvgUnreadBytes")->Set(stat.UnreadBytes.Avg);
+    }
 }
 
 void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
@@ -383,6 +405,8 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
         LOG_ROW_DISPATCHER_DEBUG("Create new session " << readOffset);
         sessionActorId = ActorFactory->RegisterTopicSession(
             source.GetTopicPath(),
+            source.GetEndpoint(),
+            source.GetDatabase(),
             Config,
             SelfId(),
             ev->Get()->Record.GetPartitionId(),
@@ -407,7 +431,7 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
 
     Forward(ev, sessionActorId);
     Metrics.ClientsCount->Set(Consumers.size());
-    PrintInternalState();
+    UpdateMetrics();
 }
 
 void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvGetNextBatch::TPtr& ev) {
@@ -490,7 +514,7 @@ void TRowDispatcher::DeleteConsumer(const ConsumerSessionKey& key) {
     ConsumersByEventQueueId.erase(consumerIt->second->EventQueueId);
     Consumers.erase(consumerIt);
     Metrics.ClientsCount->Set(Consumers.size());
-    PrintInternalState();
+    UpdateMetrics();
 }
 
 void TRowDispatcher::Handle(const NYql::NDq::TEvRetryQueuePrivate::TEvSessionClosed::TPtr& ev) {
@@ -577,9 +601,37 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvStatus::TPtr& ev) {
     it->second->EventsQueue.Send(ev.Release()->Release().Release());
 }
 
-void TRowDispatcher::Handle(NFq::TEvPrivate::TEvPrintState::TPtr&) {
-    Schedule(TDuration::Seconds(PrintStatePeriodSec), new NFq::TEvPrivate::TEvPrintState());
-    PrintInternalState();
+void TRowDispatcher::Handle(NFq::TEvPrivate::TEvUpdateMetrics::TPtr&) {
+    Schedule(TDuration::Seconds(UpdateMetricsPeriodSec), new NFq::TEvPrivate::TEvUpdateMetrics());
+    UpdateMetrics();
+}
+
+void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvSessionStatistic::TPtr& ev) {
+    LOG_ROW_DISPATCHER_TRACE("TEvSessionStatistic from " << ev->Sender);
+    const auto& key = ev->Get()->Stat.SessionKey;
+    TopicSessionKey sessionKey{key.Endpoint, key.Database, key.TopicPath, key.PartitionId};
+
+    auto sessionsIt = TopicSessions.find(sessionKey);
+    if (sessionsIt == TopicSessions.end()) {
+        return;
+    }
+    auto& sessionsInfo = sessionsIt->second;
+    auto sessionIt = sessionsInfo.Sessions.find(ev->Sender);
+    if (sessionIt == sessionsInfo.Sessions.end()) {
+        return;
+    }
+
+    auto& sessionInfo = sessionIt->second;
+    sessionInfo.Stat = ev->Get()->Stat.Common;
+
+    for (const auto& clientStat : ev->Get()->Stat.Clients) {
+        auto it = sessionInfo.Consumers.find(clientStat.ReadActorId);
+        if (it == sessionInfo.Consumers.end()) {
+            continue;
+        }
+        auto consumerInfoPtr = it->second; 
+        consumerInfoPtr->Stat = clientStat;
+    }
 }
 
 } // namespace