YQ-855: add public sensors query.source_input_messages, query.sink_output_messages

ref:fa5ba662be924fbeb067449c60547b2946e377f1

1 files changed, 9 insertions, 2 deletions

diff --git a/ydb/library/yql/providers/dq/actors/task_controller.cpp b/ydb/library/yql/providers/dq/actors/task_controller.cpp
index e206500791..2b210fd28d 100644
--- a/ydb/library/yql/providers/dq/actors/task_controller.cpp
+++ b/ydb/library/yql/providers/dq/actors/task_controller.cpp
@@ -212,6 +212,8 @@ private:
         YQL_LOG(DEBUG) << " " << SelfId() << " ExportStats " << (taskId ? ToString(taskId) : "Summary");
         TString name;
         std::map<TString, TString> labels;
+        static const TString SourceLabel = "Source";
+        static const TString SinkLabel = "Sink";
         for (const auto& [k, v] : stat.Get()) {
             labels.clear();
             if (auto group = GroupForExport(stat, k, taskId, name, labels)) {
@@ -225,10 +227,15 @@ private:
                         publicCounterName = "query.cpu_usage_us";
                         isDeriv = true;
                     } else if (name == "Bytes") {
-                        if (labels.find("Source") != labels.end()) publicCounterName = "query.input_bytes";
-                        else if (labels.find("Sink") != labels.end()) publicCounterName = "query.output_bytes";
+                        if (labels.count(SourceLabel)) publicCounterName = "query.input_bytes";
+                        else if (labels.count(SinkLabel)) publicCounterName = "query.output_bytes";
+                        isDeriv = true;
+                    } else if (name == "RowsIn") {
+                        if (labels.count(SourceLabel)) publicCounterName = "query.source_input_messages";
+                        else if (labels.count(SinkLabel)) publicCounterName = "query.sink_output_messages"; // RowsIn == RowsOut for Sinks
                         isDeriv = true;
                     }
+
                     if (publicCounterName) {
                         *ServiceCounters.PublicCounters->GetNamedCounter("name", publicCounterName, isDeriv) = v.Count;
                     }