Added a timeout waiting for a response from the tablet (#8260)

Author: andrewstalin

ydb/core/statistics/service/service.cpp

@@ -28,6 +28,7 @@ namespace NStat {
 static constexpr TDuration DefaultAggregateKeepAlivePeriod = TDuration::MilliSeconds(500);
 static constexpr TDuration DefaultAggregateKeepAliveTimeout = TDuration::Seconds(3);
 static constexpr TDuration DefaultAggregateKeepAliveAckTimeout = TDuration::Seconds(3);
+static constexpr TDuration DefaultStatisticsRequestTimeout = TDuration::Seconds(5);
 static constexpr size_t DefaultMaxInFlightTabletRequests = 5;
 static constexpr size_t DefaultFanOutFactor = 5;
 
@@ -37,6 +38,7 @@ TStatServiceSettings::TStatServiceSettings()
     : AggregateKeepAlivePeriod(DefaultAggregateKeepAlivePeriod)
     , AggregateKeepAliveTimeout(DefaultAggregateKeepAliveTimeout)
     , AggregateKeepAliveAckTimeout(DefaultAggregateKeepAliveAckTimeout)
+    , StatisticsRequestTimeout(DefaultStatisticsRequestTimeout)
     , MaxInFlightTabletRequests(DefaultMaxInFlightTabletRequests)
     , FanOutFactor(DefaultFanOutFactor)
 {}

ydb/core/statistics/service/service.h

@@ -9,6 +9,7 @@ struct TStatServiceSettings {
     TDuration AggregateKeepAlivePeriod;
     TDuration AggregateKeepAliveTimeout;
     TDuration AggregateKeepAliveAckTimeout;
+    TDuration StatisticsRequestTimeout;
     size_t MaxInFlightTabletRequests;
     size_t FanOutFactor;
 
@@ -29,6 +30,11 @@ struct TStatServiceSettings {
         return *this;
     }
 
+    TStatServiceSettings& SetStatisticsRequestTimeout(const TDuration& val) {
+        StatisticsRequestTimeout = val;
+        return *this;
+    }
+
     TStatServiceSettings& SetMaxInFlightTabletRequests(size_t val) {
         MaxInFlightTabletRequests = val;
         return *this;

ydb/core/statistics/service/service_impl.cpp

@@ -25,7 +25,6 @@
 namespace NKikimr {
 namespace NStat {
 
-
 struct TAggregationStatistics {
     using TColumnsStatistics = ::google::protobuf::RepeatedPtrField<::NKikimrStat::TColumnStatistics>;
 
@@ -128,6 +127,7 @@ class TStatService : public TActorBootstrapped<TStatService> {
             EvDispatchKeepAlive,
             EvKeepAliveTimeout,
             EvKeepAliveAckTimeout,
+            EvStatisticsRequestTimeout,
 
             EvEnd
         };
@@ -155,6 +155,13 @@ class TStatService : public TActorBootstrapped<TStatService> {
             ui64 Round;
             ui32 NodeId;
         };
+
+        struct TEvStatisticsRequestTimeout: public NActors::TEventLocal<TEvStatisticsRequestTimeout, EvStatisticsRequestTimeout> {
+            TEvStatisticsRequestTimeout(ui64 round, ui64 tabletId): Round(round), TabletId(tabletId) {}
+
+            ui64 Round;
+            ui64 TabletId;
+        };
     };
 
     void Bootstrap() {
@@ -195,6 +202,7 @@ class TStatService : public TActorBootstrapped<TStatService> {
             hFunc(TEvStatistics::TEvAggregateKeepAlive, Handle);
             hFunc(TEvPrivate::TEvDispatchKeepAlive, Handle);
             hFunc(TEvPrivate::TEvKeepAliveTimeout, Handle);
+            hFunc(TEvPrivate::TEvStatisticsRequestTimeout, Handle);
             hFunc(TEvStatistics::TEvStatisticsResponse, Handle);
             hFunc(TEvStatistics::TEvAggregateStatisticsResponse, Handle);
 
@@ -902,7 +910,31 @@ class TStatService : public TActorBootstrapped<TStatService> {
         }
     }
 
-    void SendStatisticsRequest(const TActorId& clientId) {
+    void Handle(TEvPrivate::TEvStatisticsRequestTimeout::TPtr& ev) {
+        const auto round = ev->Get()->Round;
+        if (IsNotCurrentRound(round)) {
+            LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
+                "Skip TEvStatisticsRequestTimeout");
+            return;
+        }
+
+        const auto tabletId = ev->Get()->TabletId;
+        auto tabletPipe = AggregationStatistics.LocalTablets.TabletsPipes.find(tabletId);
+        if (tabletPipe == AggregationStatistics.LocalTablets.TabletsPipes.end()) {
+            LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
+                "Tablet " << tabletId << " has already been processed");
+            return;
+        }
+
+        LOG_ERROR_S(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
+                "No result was received from the tablet " << tabletId);
+
+        auto clientId = tabletPipe->second;
+        OnTabletError(tabletId);
+        NTabletPipe::CloseClient(SelfId(), clientId);
+    }
+
+    void SendStatisticsRequest(const TActorId& clientId, ui64 tabletId) {
         auto request = std::make_unique<TEvStatistics::TEvStatisticsRequest>();
         auto& record = request->Record;
         record.MutableTypes()->Add(NKikimrStat::TYPE_COUNT_MIN_SKETCH);
@@ -916,7 +948,9 @@ class TStatService : public TActorBootstrapped<TStatService> {
             columnTags->Add(tag);
         }
 
-        NTabletPipe::SendData(SelfId(), clientId, request.release(), AggregationStatistics.Round);
+        const auto round = AggregationStatistics.Round;
+        NTabletPipe::SendData(SelfId(), clientId, request.release(), round);
+        Schedule(Settings.StatisticsRequestTimeout, new TEvPrivate::TEvStatisticsRequestTimeout(round, tabletId));
 
         LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
             "TEvStatisticsRequest send"
@@ -928,7 +962,7 @@ class TStatService : public TActorBootstrapped<TStatService> {
         LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
             "Tablet " << tabletId << " is not local.");
 
-        constexpr auto error = NKikimrStat::TEvAggregateStatisticsResponse::TYPE_NON_LOCAL_TABLET;
+        const auto error = NKikimrStat::TEvAggregateStatisticsResponse::TYPE_NON_LOCAL_TABLET;
         AggregationStatistics.FailedTablets.emplace_back(tabletId, 0, error);
 
         AggregationStatistics.LocalTablets.TabletsPipes.erase(tabletId);
@@ -966,7 +1000,7 @@ class TStatService : public TActorBootstrapped<TStatService> {
 
         if (tabletPipe != tabletsPipes.end() && clientId == tabletPipe->second) {
             if (ev->Get()->Status == NKikimrProto::OK) {
-                SendStatisticsRequest(clientId);
+                SendStatisticsRequest(clientId, tabletId);
             } else {
                 OnTabletError(tabletId);
             }

ydb/core/statistics/service/ut/ut_service.cpp

@@ -532,6 +532,67 @@ Y_UNIT_TEST_SUITE(StatisticsService) {
             UNIT_ASSERT_VALUES_EQUAL(expectedError, actualError);
         }
     }
+
+    Y_UNIT_TEST(ShouldBeCcorrectProcessingTabletTimeout) {
+        size_t nodeCount = 1;
+        auto runtime = TTestActorRuntime(nodeCount, 1, false);
+        auto settings = GetDefaultSettings()
+            .SetStatisticsRequestTimeout(TDuration::MilliSeconds(10));
+        auto indexToActorMap = InitializeRuntime(runtime, nodeCount, settings);
+        auto nodeIdToIndexMap = GetNodeIdToIndexMap(indexToActorMap);
+        std::vector<ui64> localTabletsIds = {1, 2, 3, 4, 5, 6, 7};
+        std::vector<TAggregateStatisticsRequest::TTablets> nodesTablets = {{.NodeId = indexToActorMap[0].NodeId(), .Ids{localTabletsIds}}};
+
+        std::unordered_map<TActorId, ui64> pipeToTablet;
+        std::vector<TTestActorRuntimeBase::TEventObserverHolder> observers;
+        observers.emplace_back(runtime.AddObserver<TEvTabletResolver::TEvForward>([&](TEvTabletResolver::TEvForward::TPtr& ev) {
+            auto tabletId = ev->Get()->TabletID;
+            auto recipient = indexToActorMap[nodeIdToIndexMap[ev->Sender.NodeId()]];
+            pipeToTablet[ev->Sender] = tabletId;
+
+            runtime.Send(new IEventHandle(recipient, ev->Sender,
+                new TEvTabletPipe::TEvClientConnected(tabletId, NKikimrProto::OK, ev->Sender, ev->Sender,
+                true, false, 0), 0, ev->Cookie), nodeIdToIndexMap[ev->Sender.NodeId()], true);
+            ev.Reset();
+        }));
+        observers.emplace_back(runtime.AddObserver([&](TAutoPtr<IEventHandle>& ev) {
+            switch (ev->GetTypeRewrite()) {
+                case TEvTabletPipe::EvSend:
+                    auto msg = ev->Get<TEvStatistics::TEvStatisticsRequest>();
+                    if (msg != nullptr) {
+                        auto tabletId = pipeToTablet[ev->Recipient];
+                        if (tabletId % 2 != 0) {
+                            auto senderNodeIndex = nodeIdToIndexMap[ev->Sender.NodeId()];
+                            runtime.Send(new IEventHandle(ev->Sender, ev->Sender,
+                                CreateStatisticsResponse(TStatisticsResponse{
+                                    .TabletId = tabletId,
+                                    .Status = NKikimrStat::TEvStatisticsResponse::STATUS_SUCCESS
+                                }).release(), 0, ev->Cookie), senderNodeIndex, true);
+                        }
+                    }
+                    break;
+            }
+        }));
+
+        auto sender = runtime.AllocateEdgeActor();
+        runtime.Send(indexToActorMap[0], sender, CreateStatisticsRequest(TAggregateStatisticsRequest{
+            .Round = 1,
+            .PathId{3, 3},
+            .Nodes{ nodesTablets },
+            .ColumnTags{1}
+        }).release());
+
+        auto res = runtime.GrabEdgeEvent<TEvStatistics::TEvAggregateStatisticsResponse>(sender);
+        const auto& record = res->Get()->Record;
+        size_t expectedFailedTabletsCount = 3;
+        UNIT_ASSERT_VALUES_EQUAL(expectedFailedTabletsCount, record.GetFailedTablets().size());
+
+        ui32 expectedError = NKikimrStat::TEvAggregateStatisticsResponse::TYPE_NON_LOCAL_TABLET;
+        for (const auto& fail : record.GetFailedTablets()) {
+            ui32 actualError = fail.GetError();
+            UNIT_ASSERT_VALUES_EQUAL(expectedError, actualError);
+        }
+    }
 }
 
 } // NSysView