Merge 9c4457f into 752521d

Author: uzhastik

ydb/core/fq/libs/actors/pending_fetcher.cpp

@@ -53,6 +53,7 @@
 #include <ydb/core/fq/libs/control_plane_storage/control_plane_storage.h>
 #include <ydb/core/fq/libs/control_plane_storage/events/events.h>
 #include <ydb/core/fq/libs/events/events.h>
+#include <ydb/core/fq/libs/metrics/sanitize_label.h>
 #include <ydb/core/fq/libs/private_client/internal_service.h>
 
 #include <ydb/library/actors/core/log.h>
@@ -358,25 +359,50 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
         const TString folderId = NYdb::NFq::TScope(task.scope()).ParseFolder();
         const TString cloudId = task.sensor_labels().at("cloud_id");
         const TString queryId = task.query_id().value();
+        const bool isStreaming = task.query_type() == FederatedQuery::QueryContent::STREAMING;
+        TString queryIdLabel;
+        TString queryNameLabel = SanitizeLabel(task.query_name());
+        if (task.automatic()) {
+            queryIdLabel = isStreaming ? "streaming" : "analytics";
+        } else if (isStreaming) {
+            queryIdLabel = queryId;
+        } else {
+            queryIdLabel = "manual";
+        }
 
         ::NYql::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);
+
+        ::NMonitoring::TDynamicCounterPtr queryPublicCounters = publicCountersParent;
+        // use original query id here
+        queryPublicCounters = queryPublicCounters->GetSubgroup("query_id", queryId);
+
+        if (queryNameLabel) {
+            queryPublicCounters = queryPublicCounters->GetSubgroup("query_name", queryNameLabel);
+        }
+        queryCounters.PublicCounters = queryPublicCounters;
 
         auto rootCountersParent = ServiceCounters.RootCounters;
         std::set<std::pair<TString, TString>> sensorLabels(task.sensor_labels().begin(), task.sensor_labels().end());
         for (const auto& [label, item]: sensorLabels) {
             rootCountersParent = rootCountersParent->GetSubgroup(label, item);
         }
 
-        queryCounters.RootCounters = rootCountersParent->GetSubgroup("query_id",
-            task.automatic() ? (folderId ? "automatic_" + folderId : "automatic") : queryId);
-        queryCounters.Counters = queryCounters.RootCounters;
+        ::NMonitoring::TDynamicCounterPtr queryRootCounters = rootCountersParent;
+        if (queryIdLabel) {
+            queryRootCounters = queryRootCounters->GetSubgroup("query_id", queryIdLabel);
+        }
+
+        if (!task.automatic() && isStreaming && queryNameLabel) {
+            queryRootCounters = queryRootCounters->GetSubgroup("query_name", queryNameLabel);
+        }
+
+        queryCounters.RootCounters = queryRootCounters;
+        queryCounters.Counters = queryRootCounters;
 
         queryCounters.InitUptimeCounter();
         const auto createdAt = TInstant::Now();

ydb/core/fq/libs/actors/ya.make

@@ -44,6 +44,7 @@ PEERDIR(
     ydb/core/fq/libs/db_schema
     ydb/core/fq/libs/events
     ydb/core/fq/libs/grpc
+    ydb/core/fq/libs/metrics
     ydb/core/fq/libs/private_client
     ydb/core/fq/libs/rate_limiter/utils
     ydb/core/fq/libs/result_formatter

ydb/core/fq/libs/row_dispatcher/common.cpp renamed to ydb/core/fq/libs/metrics/sanitize_label.cpp

@@ -1,12 +1,13 @@
-#include "common.h"
+#include "sanitize_label.h"
 
 #include <util/generic/string.h>
 
 namespace NFq {
 
-TString CleanupCounterValueString(const TString& value) {
-    TString clean;
-    constexpr auto valueLenghtLimit = 200;
+TString SanitizeLabel(const TString& value) {
+    TString result;
+    result.reserve(value.size());
+    constexpr auto labelLengthLimit = 200;
 
     for (auto c : value) {
         switch (c) {
@@ -19,13 +20,13 @@ TString CleanupCounterValueString(const TString& value) {
         case '\\':
             continue;
         default:
-            clean.push_back(c);
-            if (clean.size() == valueLenghtLimit) {
-                break;
+            result.push_back(c);
+            if (result.size() == labelLengthLimit) {
+                return result;
             }
         }
     }
-    return clean;
+    return result;
 }
 
 } // namespace NFq

ydb/core/fq/libs/metrics/sanitize_label.h

@@ -0,0 +1,9 @@
+#pragma once
+
+#include <util/generic/fwd.h>
+
+namespace NFq {
+
+TString SanitizeLabel(const TString& value);
+
+} // namespace NFq

ydb/core/fq/libs/metrics/ut/sanitize_label_ut.cpp

@@ -0,0 +1,29 @@
+#include <ydb/core/fq/libs/metrics/sanitize_label.h>
+
+#include <library/cpp/testing/unittest/registar.h>
+
+Y_UNIT_TEST_SUITE(SanitizeLable) {
+    Y_UNIT_TEST(Empty) {
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel(""), "");
+    }
+
+    Y_UNIT_TEST(SkipSingleBadSymbol) {
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel("|"), "");
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel("*"), "");
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel("?"), "");
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel("\""), "");
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel("'"), "");
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel("`"), "");
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel("\\"), "");
+    }
+
+    Y_UNIT_TEST(SkipBadSymbols) {
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel("a|b*c?d\"e'f`g\\h"), "abcdefgh");
+    }
+
+    Y_UNIT_TEST(Truncate200) {
+        TString s1(400, 'a');
+        TString s2(200, 'a');
+        UNIT_ASSERT_VALUES_EQUAL(NFq::SanitizeLabel(s1), s2);
+    }
+}

ydb/core/fq/libs/metrics/ut/ya.make

@@ -1,13 +1,12 @@
 UNITTEST_FOR(ydb/core/fq/libs/metrics)
 
-FORK_SUBTESTS()
-
 IF (SANITIZER_TYPE OR WITH_VALGRIND)
     SIZE(MEDIUM)
 ENDIF()
 
 SRCS(
     metrics_ut.cpp
+    sanitize_label_ut.cpp
 )
 
 YQL_LAST_ABI_VERSION()

ydb/core/fq/libs/metrics/ya.make

@@ -1,13 +1,14 @@
 LIBRARY()
 
 SRCS(
+    sanitize_label.cpp
     status_code_counters.cpp
 )
 
 PEERDIR(
     library/cpp/monlib/dynamic_counters
-    ydb/library/yql/dq/actors/protos
     yql/essentials/public/issue
+    ydb/library/yql/dq/actors/protos
 )
 
 YQL_LAST_ABI_VERSION()

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

@@ -1,7 +1,6 @@
 #include "row_dispatcher.h"
 
 #include "actors_factory.h"
-#include "common.h"
 #include "coordinator.h"
 #include "leader_election.h"
 
@@ -15,6 +14,7 @@
 
 #include <ydb/core/fq/libs/actors/logging/log.h>
 #include <ydb/core/fq/libs/events/events.h>
+#include <ydb/core/fq/libs/metrics/sanitize_label.h>
 #include <ydb/core/mon/mon.h>
 
 #include <ydb/core/fq/libs/row_dispatcher/events/data_plane.h>
@@ -598,7 +598,7 @@ void TRowDispatcher::UpdateMetrics() {
 
 void TRowDispatcher::SetQueryMetrics(const TQueryStatKey& queryKey, ui64 unreadBytesMax, ui64 unreadBytesAvg, i64 readLagMessagesMax) {
     auto queryGroup = Metrics.Counters->GetSubgroup("query_id", queryKey.QueryId);
-    auto topicGroup = queryGroup->GetSubgroup("read_group", CleanupCounterValueString(queryKey.ReadGroup));
+    auto topicGroup = queryGroup->GetSubgroup("read_group", SanitizeLabel(queryKey.ReadGroup));
     topicGroup->GetCounter("MaxUnreadBytes")->Set(unreadBytesMax);
     topicGroup->GetCounter("AvgUnreadBytes")->Set(unreadBytesAvg);
     topicGroup->GetCounter("MaxReadLag")->Set(readLagMessagesMax);
@@ -740,7 +740,7 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
     LOG_ROW_DISPATCHER_DEBUG("Received TEvStartSession from " << ev->Sender << ", read group " << ev->Get()->Record.GetSource().GetReadGroup() << ", topicPath " << ev->Get()->Record.GetSource().GetTopicPath() <<
         " part id " << ev->Get()->Record.GetPartitionId() << " query id " << ev->Get()->Record.GetQueryId() << " cookie " << ev->Cookie);
     auto queryGroup = Metrics.Counters->GetSubgroup("query_id", ev->Get()->Record.GetQueryId());
-    auto topicGroup = queryGroup->GetSubgroup("read_group", CleanupCounterValueString(ev->Get()->Record.GetSource().GetReadGroup()));
+    auto topicGroup = queryGroup->GetSubgroup("read_group", SanitizeLabel(ev->Get()->Record.GetSource().GetReadGroup()));
     topicGroup->GetCounter("StartSession", true)->Inc();
 
     NodesTracker.AddNode(ev->Sender.NodeId());

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

@@ -1,8 +1,7 @@
 #include "topic_session.h"
 
-#include "common.h"
-
 #include <ydb/core/fq/libs/actors/logging/log.h>
+#include <ydb/core/fq/libs/metrics/sanitize_label.h>
 #include <ydb/core/fq/libs/row_dispatcher/events/data_plane.h>
 #include <ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.h>
 
@@ -25,7 +24,7 @@ namespace {
 
 struct TTopicSessionMetrics {
     void Init(const ::NMonitoring::TDynamicCounterPtr& counters, const TString& topicPath, ui32 partitionId) {
-        TopicGroup = counters->GetSubgroup("topic", CleanupCounterValueString(topicPath));
+        TopicGroup = counters->GetSubgroup("topic", SanitizeLabel(topicPath));
         AllSessionsDataRate = counters->GetCounter("AllSessionsDataRate", true);
 
         PartitionGroup = TopicGroup->GetSubgroup("partition", ToString(partitionId));
@@ -107,7 +106,7 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
             }
             Y_UNUSED(TDuration::TryParse(Settings.GetSource().GetReconnectPeriod(), ReconnectPeriod));
             auto queryGroup = Counters->GetSubgroup("query_id", ev->Get()->Record.GetQueryId());
-            auto topicGroup = queryGroup->GetSubgroup("read_group", CleanupCounterValueString(readGroup));
+            auto topicGroup = queryGroup->GetSubgroup("read_group", SanitizeLabel(readGroup));
             FilteredDataRate = topicGroup->GetCounter("FilteredDataRate", true);
             RestartSessionByOffsetsByQuery = counters->GetCounter("RestartSessionByOffsetsByQuery", true);
         }