YQ-858 Added QueryUptime metric

Added QueryUptime metric

ref:73d8db54d5c318b572ecb6a39c48fe3a515817d5

6 files changed, 70 insertions, 38 deletions

diff --git a/ydb/core/yq/libs/actors/pending_fetcher.cpp b/ydb/core/yq/libs/actors/pending_fetcher.cpp
index f849e9aef29..5a7f71d9439 100644
--- a/ydb/core/yq/libs/actors/pending_fetcher.cpp
+++ b/ydb/core/yq/libs/actors/pending_fetcher.cpp
@@ -173,7 +173,7 @@ private:
     }
 
     void HandleWakeup(NActors::TEvents::TEvWakeup::TPtr& ev) {
-        switch(ev->Get()->Tag) { 
+        switch(ev->Get()->Tag) {
         case WAKEUP_TAG_FETCH:
             Schedule(PendingFetchPeriod, new NActors::TEvents::TEvWakeup(WAKEUP_TAG_FETCH));
             if (!HasRunningRequest) {
@@ -230,12 +230,11 @@ private:
         }
         auto queryId = itA->second;
         RunActorMap.erase(itA);
-        
+
         auto itC = CountersMap.find(queryId);
         if (itC != CountersMap.end()) {
             auto& info = itC->second;
-            if (info.RunActorId == runActorId) 
-            {
+            if (info.RunActorId == runActorId) {
                 info.RunActorId = TActorId();
                 info.CleanupDeadline = Now() + CLEANUP_PERIOD;
                 Schedule(CLEANUP_PERIOD, new NActors::TEvents::TEvWakeup(WAKEUP_TAG_CLEANUP));
@@ -285,6 +284,31 @@ private:
             serviceAccounts[identity.value()] = identity.signature();
         }
 
+        NDq::SetYqlLogLevels(NActors::NLog::PRI_TRACE);
+
+        const TVector<TString> path = StringSplitter(task.scope()).Split('/').SkipEmpty(); // yandexcloud://{folder_id}
+        const TString folderId = path.size() == 2 && path.front().StartsWith(NYdb::NYq::TScope::YandexCloudScopeSchema)
+                            ? path.back() : TString{};
+        const TString cloudId = task.sensor_labels().at("cloud_id");
+        const TString queryId = task.query_id().value();
+
+        ::NYq::NCommon::TServiceCounters queryCounters(ServiceCounters);
+        auto publicCountersParent = ServiceCounters.PublicCounters;
+
+        if (cloudId && folderId) {
+            publicCountersParent = publicCountersParent->GetSubgroup("cloud_id", cloudId)->GetSubgroup("folder_id", folderId);
+        }
+        queryCounters.PublicCounters = publicCountersParent->GetSubgroup("query_id",
+            task.automatic() ? (task.query_name() ? task.query_name() : "automatic") : queryId);
+
+        auto rootCountersParent = ServiceCounters.RootCounters;
+        queryCounters.RootCounters = rootCountersParent->GetSubgroup("query_id",
+            task.automatic() ? (folderId ? "automatic_" + folderId : "automatic") : queryId);
+        queryCounters.Counters = queryCounters.RootCounters;
+
+        const auto queryUptimeCounter = queryCounters.PublicCounters->GetNamedCounter("name", "query.uptime_seconds", false);
+        const auto createdAt = TInstant::Now();
+
         TRunActorParams params(
             YqSharedResources->YdbDriver, S3Gateway,
             FunctionRegistry, RandomProvider,
@@ -293,7 +317,7 @@ private:
             CommonConfig, CheckpointCoordinatorConfig,
             PrivateApiConfig, GatewaysConfig, PingerConfig,
             task.text(), task.scope(), task.user_token(),
-            DatabaseResolver, task.query_id().value(),
+            DatabaseResolver, queryId,
             task.user_id(), Guid, task.generation(),
             VectorFromProto(task.connection()),
             VectorFromProto(task.binding()),
@@ -305,7 +329,7 @@ private:
             task.state_load_mode(),
             task.disposition(),
             task.status(),
-            task.sensor_labels().at("cloud_id"),
+            cloudId,
             VectorFromProto(task.result_set_meta()),
             VectorFromProto(task.dq_graph()),
             task.dq_graph_index(),
@@ -313,27 +337,9 @@ private:
             task.automatic(),
             task.query_name(),
             NProtoInterop::CastFromProto(task.deadline()),
-            ClientCounters);
-
-        NDq::SetYqlLogLevels(NActors::NLog::PRI_TRACE);
-
-        const TVector<TString> path = StringSplitter(params.Scope.ToString()).Split('/').SkipEmpty(); // yandexcloud://{folder_id}
-        const TString folderId = path.size() == 2 && path.front().StartsWith(NYdb::NYq::TScope::YandexCloudScopeSchema)
-                            ? path.back() : TString{};
-
-        ::NYq::NCommon::TServiceCounters queryCounters(ServiceCounters);
-        auto publicCountersParent = ServiceCounters.PublicCounters;
-
-        if (params.CloudId && folderId) {
-            publicCountersParent = publicCountersParent->GetSubgroup("cloud_id", params.CloudId)->GetSubgroup("folder_id", folderId);
-        }
-        queryCounters.PublicCounters = publicCountersParent->GetSubgroup("query_id",
-            params.Automatic ? (params.QueryName ? params.QueryName : "automatic") : params.QueryId);
-
-        auto rootCountersParent = ServiceCounters.RootCounters;
-        queryCounters.RootCounters = rootCountersParent->GetSubgroup("query_id",
-            params.Automatic ? (folderId ? "automatic_" + folderId : "automatic") : params.QueryId);
-        queryCounters.Counters = queryCounters.RootCounters;
+            ClientCounters,
+            queryUptimeCounter,
+            createdAt);
 
         auto runActorId = Register(CreateRunActor(SelfId(), queryCounters, std::move(params)));
 
diff --git a/ydb/core/yq/libs/actors/pinger.cpp b/ydb/core/yq/libs/actors/pinger.cpp
index 55fc80d6a0b..899eb7423af 100644
--- a/ydb/core/yq/libs/actors/pinger.cpp
+++ b/ydb/core/yq/libs/actors/pinger.cpp
@@ -59,7 +59,7 @@ private:
 class TPingerActor : public NActors::TActorBootstrapped<TPingerActor> {
     class TRetryState {
     public:
-        void Init(const TInstant& now, const TInstant& startLeaseTime, const TDuration& maxRetryTime) {
+        void Init(TInstant now, TInstant startLeaseTime, TDuration maxRetryTime) {
             StartRequestTime = now;
             StartLeaseTime = startLeaseTime;
             Delay = TDuration::Zero();
@@ -149,7 +149,9 @@ public:
         const TPrivateClient& client,
         const TActorId parent,
         const NConfig::TPingerConfig& config,
-        const TInstant& deadline)
+        TInstant deadline,
+        const NMonitoring::TDynamicCounters::TCounterPtr& queryUptime,
+        TInstant createdAt)
         : Config(config)
         , Scope(scope)
         , UserId(userId)
@@ -158,6 +160,8 @@ public:
         , Client(client)
         , Parent(parent)
         , Deadline(deadline)
+        , QueryUptime(queryUptime)
+        , CreatedAt(createdAt)
     {
     }
 
@@ -378,6 +382,7 @@ private:
     }
 
     void Ping(Yq::Private::PingTaskRequest request, ui64 cookie) {
+        QueryUptime->Set((TInstant::Now() - CreatedAt).Seconds());
         // Fill ids
         request.set_scope(Scope.ToString());
         request.set_owner_id(OwnerId);
@@ -426,6 +431,9 @@ private:
     const TActorId Parent;
     const TInstant Deadline;
 
+    const NMonitoring::TDynamicCounters::TCounterPtr QueryUptime;
+    const TInstant CreatedAt;
+
     std::deque<TForwardPingReqInfo> ForwardRequests;
     bool Finishing = false;
     bool FatalError = false; // Nonretryable error from PingTask or all retries finished.
@@ -441,7 +449,9 @@ IActor* CreatePingerActor(
     const TPrivateClient& client,
     const TActorId parent,
     const NConfig::TPingerConfig& config,
-    const TInstant& deadline)
+    TInstant deadline,
+    const NMonitoring::TDynamicCounters::TCounterPtr& queryUptime,
+    TInstant createdAt)
 {
     return new TPingerActor(
         scope,
@@ -451,7 +461,9 @@ IActor* CreatePingerActor(
         client,
         parent,
         config,
-        deadline);
+        deadline,
+        queryUptime,
+        createdAt);
 }
 
 } /* NYq */
diff --git a/ydb/core/yq/libs/actors/proxy.h b/ydb/core/yq/libs/actors/proxy.h
index c535f76ffb4..e58062b5718 100644
--- a/ydb/core/yq/libs/actors/proxy.h
+++ b/ydb/core/yq/libs/actors/proxy.h
@@ -54,8 +54,7 @@ NActors::IActor* CreatePendingFetcher(
 NActors::IActor* CreateRunActor(
     const NActors::TActorId& fetcherId,
     const ::NYq::NCommon::TServiceCounters& serviceCounters,
-    TRunActorParams&& params
-    );
+    TRunActorParams&& params);
 
 struct TResultId {
     TString Id;
@@ -85,7 +84,9 @@ NActors::IActor* CreatePingerActor(
     const NYq::TPrivateClient& client,
     const NActors::TActorId parent,
     const NConfig::TPingerConfig& config,
-    const TInstant& deadline);
+    TInstant deadline,
+    const NMonitoring::TDynamicCounters::TCounterPtr& queryUptime,
+    TInstant createdAt);
 
 TString MakeInternalError(const TString& text);
 
diff --git a/ydb/core/yq/libs/actors/run_actor.cpp b/ydb/core/yq/libs/actors/run_actor.cpp
index eb374c6622c..891b415dd46 100644
--- a/ydb/core/yq/libs/actors/run_actor.cpp
+++ b/ydb/core/yq/libs/actors/run_actor.cpp
@@ -91,17 +91,19 @@ public:
         , TRunActorParams&& params)
         : FetcherId(fetcherId)
         , Params(std::move(params))
-        , CreatedAt(TInstant::Now())
+        , CreatedAt(Params.CreatedAt)
         , QueryCounters(queryCounters)
         , EnableCheckpointCoordinator(Params.QueryType == YandexQuery::QueryContent::STREAMING && Params.CheckpointCoordinatorConfig.GetEnabled())
         , MaxTasksPerOperation(Params.CommonConfig.GetMaxTasksPerOperation() ? Params.CommonConfig.GetMaxTasksPerOperation() : 40)
     {
+        Params.QueryUptimeCounter->Set(0);
     }
 
     static constexpr char ActorName[] = "YQ_RUN_ACTOR";
 
     void Bootstrap() {
         LOG_D("Start run actor. Compute state: " << YandexQuery::QueryMeta::ComputeStatus_Name(Params.Status));
+        Params.QueryUptimeCounter->Set((TInstant::Now() - CreatedAt).Seconds());
         LogReceivedParams();
         Pinger = Register(
             CreatePingerActor(
@@ -119,7 +121,9 @@ public:
                     Params.ClientCounters),
                 SelfId(),
                 Params.PingerConfig,
-                Params.Deadline
+                Params.Deadline,
+                Params.QueryUptimeCounter,
+                CreatedAt
                 ));
         Become(&TRunActor::StateFuncWrapper<&TRunActor::StateFunc>);
         try {
@@ -1287,6 +1291,7 @@ private:
     NActors::TActorId CheckpointCoordinatorId;
     TString SessionId;
     ::NYq::NCommon::TServiceCounters QueryCounters;
+    const NMonitoring::TDynamicCounters::TCounterPtr QueryUptime;
     bool EnableCheckpointCoordinator = false;
     bool RetryNeeded = false;
     Yq::Private::PingTaskRequest QueryStateUpdateRequest;
diff --git a/ydb/core/yq/libs/actors/run_actor_params.cpp b/ydb/core/yq/libs/actors/run_actor_params.cpp
index 1446cae1bbe..deb1279d628 100644
--- a/ydb/core/yq/libs/actors/run_actor_params.cpp
+++ b/ydb/core/yq/libs/actors/run_actor_params.cpp
@@ -44,7 +44,9 @@ TRunActorParams::TRunActorParams(
     bool automatic,
     const TString& queryName,
     const TInstant& deadline,
-    const NMonitoring::TDynamicCounterPtr& clientCounters
+    const NMonitoring::TDynamicCounterPtr& clientCounters,
+    const NMonitoring::TDynamicCounters::TCounterPtr& queryUptimeCounter,
+    TInstant createdAt
     )
     : Driver(driver)
     , S3Gateway(s3Gateway)
@@ -86,6 +88,8 @@ TRunActorParams::TRunActorParams(
     , QueryName(queryName)
     , Deadline(deadline)
     , ClientCounters(clientCounters)
+    , QueryUptimeCounter(queryUptimeCounter)
+    , CreatedAt(createdAt)
     {
     }
 
diff --git a/ydb/core/yq/libs/actors/run_actor_params.h b/ydb/core/yq/libs/actors/run_actor_params.h
index bad5c3c32b4..1ce6ab64f27 100644
--- a/ydb/core/yq/libs/actors/run_actor_params.h
+++ b/ydb/core/yq/libs/actors/run_actor_params.h
@@ -58,7 +58,9 @@ struct TRunActorParams { // TODO2 : Change name
         bool automatic,
         const TString& queryName,
         const TInstant& deadline,
-        const NMonitoring::TDynamicCounterPtr& clientCounters
+        const NMonitoring::TDynamicCounterPtr& clientCounters,
+        const NMonitoring::TDynamicCounters::TCounterPtr& queryUptimeCounter,
+        TInstant createdAt
     );
 
     TRunActorParams(const TRunActorParams& params) = default;
@@ -107,6 +109,8 @@ struct TRunActorParams { // TODO2 : Change name
     TInstant Deadline;
 
     const NMonitoring::TDynamicCounterPtr ClientCounters;
+    const NMonitoring::TDynamicCounters::TCounterPtr QueryUptimeCounter;
+    const TInstant CreatedAt;
 };
 
 } /* NYq */