Merge 2c68427 into 4621fcc

Author: kardymonds

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

@@ -17,6 +17,8 @@ namespace {
 class TTopicFormatHandler : public NActors::TActor<TTopicFormatHandler>, public ITopicFormatHandler, public TTypeParser {
     using TBase = NActors::TActor<TTopicFormatHandler>;
 
+    static constexpr char ActorName[] = "FQ_ROW_DISPATCHER_FORMAT_HANDLER";
+
     struct TCounters {
         TCountersDesc Desc;
 

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

@@ -61,11 +61,15 @@ struct TUserPoolMetrics {
         auto microsecGroup = execpoolGroup->GetSubgroup("sensor", "ElapsedMicrosecByActivity");
         Session = microsecGroup->GetNamedCounter("activity", "FQ_ROW_DISPATCHER_SESSION", true);
         RowDispatcher = microsecGroup->GetNamedCounter("activity", "FQ_ROW_DISPATCHER", true);
-        Compiler = microsecGroup->GetNamedCounter("activity", "FQ_ROW_DISPATCHER_COMPILER", true);
+        CompilerActor = microsecGroup->GetNamedCounter("activity", "FQ_ROW_DISPATCHER_COMPILE_ACTOR", true);
+        CompilerService = microsecGroup->GetNamedCounter("activity", "FQ_ROW_DISPATCHER_COMPILE_SERVICE", true);
+        FormatHandler = microsecGroup->GetNamedCounter("activity", "FQ_ROW_DISPATCHER_FORMAT_HANDLER", true);
     }
     ::NMonitoring::TDynamicCounters::TCounterPtr Session;
     ::NMonitoring::TDynamicCounters::TCounterPtr RowDispatcher;
-    ::NMonitoring::TDynamicCounters::TCounterPtr Compiler;
+    ::NMonitoring::TDynamicCounters::TCounterPtr CompilerActor;
+    ::NMonitoring::TDynamicCounters::TCounterPtr CompilerService;
+    ::NMonitoring::TDynamicCounters::TCounterPtr FormatHandler;
 };
 
 struct TEvPrivate {
@@ -1145,7 +1149,9 @@ void TRowDispatcher::UpdateCpuTime() {
     }
     auto currentCpuTime = UserPoolMetrics.Session->Val()
         + UserPoolMetrics.RowDispatcher->Val()
-        + UserPoolMetrics.Compiler->Val();
+        + UserPoolMetrics.CompilerActor->Val()
+        + UserPoolMetrics.CompilerService->Val()
+        + UserPoolMetrics.FormatHandler->Val();
     auto diff = (currentCpuTime - LastCpuTime) / Consumers.size();
     for (auto& [actorId, consumer] : Consumers) {
         consumer->CpuMicrosec += diff;

ydb/tests/fq/yds/test_row_dispatcher.py

@@ -21,12 +21,12 @@
 import ydb.public.api.protos.draft.fq_pb2 as fq
 
 YDS_CONNECTION = "yds"
-
+COMPUTE_NODE_COUNT = 3
 
 @pytest.fixture
 def kikimr(request):
     kikimr_conf = StreamingOverKikimrConfig(
-        cloud_mode=True, node_count={"/cp": TenantConfig(1), "/compute": TenantConfig(3)}
+        cloud_mode=True, node_count={"/cp": TenantConfig(1), "/compute": TenantConfig(COMPUTE_NODE_COUNT)}
     )
     kikimr = StreamingOverKikimr(kikimr_conf)
     kikimr.compute_plane.fq_config['row_dispatcher']['enabled'] = True
@@ -957,6 +957,7 @@ def test_sensors(self, kikimr, client):
 
         wait_actor_count(kikimr, "DQ_PQ_READ_ACTOR", 1)
         wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_SESSION", 1)
+        wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_COMPILE_SERVICE", COMPUTE_NODE_COUNT)
         wait_row_dispatcher_sensor_value(kikimr, "ClientsCount", 1)
         wait_row_dispatcher_sensor_value(kikimr, "RowsSent", 1, exact_match=False)
         wait_row_dispatcher_sensor_value(kikimr, "IncomingRequests", 1, exact_match=False)