Added spilling queue counters (#17394)

Author: lll-phill-lll

ydb/library/yql/dq/actors/spilling/spilling_counters.cpp

@@ -11,6 +11,7 @@ TSpillingCounters::TSpillingCounters(const TIntrusivePtr<::NMonitoring::TDynamic
     SpillingNoSpaceErrors = counters->GetCounter("Spilling/NoSpaceErrors", true);
     SpillingIoErrors = counters->GetCounter("Spilling/IoErrors", true);
     SpillingFileDescriptors = counters->GetCounter("Spilling/FileDescriptors", false);
+    SpillingIOQueueSize = counters->GetCounter("Spilling/IOQueueSize", false);
 }
 
 } // namespace NYql::NDq

ydb/library/yql/dq/actors/spilling/spilling_counters.h

@@ -18,6 +18,7 @@ struct TSpillingCounters : public TThrRefBase {
     ::NMonitoring::TDynamicCounters::TCounterPtr SpillingNoSpaceErrors;
     ::NMonitoring::TDynamicCounters::TCounterPtr SpillingIoErrors;
     ::NMonitoring::TDynamicCounters::TCounterPtr SpillingFileDescriptors;
+    ::NMonitoring::TDynamicCounters::TCounterPtr SpillingIOQueueSize;
 };
 
 struct TSpillingTaskCounters : public TThrRefBase {

ydb/library/yql/dq/actors/spilling/spilling_file.cpp

@@ -712,6 +712,7 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
             TAG(TH2) { s << "Active files"; }
             PRE() { s << "Used space: " << TotalSize_ << Endl; }
             PRE() { s << "Used file descriptors: " << Counters_->SpillingFileDescriptors->Val() << Endl; }
+            PRE() { s << "IO queue size: " << Counters_->SpillingIOQueueSize->Val() << Endl; }
 
             for (const auto& tx : byTx) {
                 TAG(TH2) { s << "Transaction " << tx.first; }
@@ -824,6 +825,7 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
 
         fd.HasActiveOp = true;
 
+        Counters_->SpillingIOQueueSize->Set(IoThreadPool_->Size() + 1);
         return IoThreadPool_->AddAndOwn(std::move(op));
     }