[Blockstorage] Allow injecting errors in DSProxy (#46723)

This patch adds an ability to inject random failures into the DSProxies. It is implemented with a proxy actor, receiving all events for a target DSProxy and randomly (with a configured probability) rejecting such requests.

This request originates from NBS (ydb-platform/nbs#6467), however we find that it may be useful for other teams (ydb included), and it is much more straightforward to implement in node_wardern.

Considering gaps, currently all actors are running with their own RNG initialized with a random seed, but there is only one RandomSeed parameter in configuration. Hence, it may be the case that it won't allow to reproduce some very specific failure scenarios. A better approach would be deriving seeds for every DSProxy from root seed.

5 files changed, 136 insertions, 2 deletions

diff --git a/ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.cpp b/ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.cpp
index 601645f3ef9..534ec86a629 100644
--- a/ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.cpp
+++ b/ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.cpp
@@ -1,8 +1,10 @@
 #include "dsproxy_mock.h"
 #include "model.h"
 #include <ydb/core/base/blobstorage.h>
+#include <ydb/core/blobstorage/dsproxy/dsproxy.h>
 #include <ydb/core/blobstorage/vdisk/common/vdisk_events.h>
 #include <ydb/core/util/stlog.h>
+#include <util/random/fast.h>
 
 #define YDB_LOG_THIS_FILE_COMPONENT BS_PROXY
 
@@ -148,6 +150,105 @@ namespace NKikimr {
                 , Model(MakeIntrusive<NFake::TProxyDS>(groupId))
             {}
         };
+
+        ////////////////////////////////////////////////////////////////////////////////
+
+        class TBlobStorageFailureInjectingActor final
+            : public TActor<TBlobStorageFailureInjectingActor>
+        {
+        private:
+            const TActorId RealProxy;
+            const TGroupId GroupId;
+            const double FailureProbability;
+            const ui64 RandomFailureSeed;
+            const NKikimrProto::EReplyStatus ErrorReplyStatus;
+            TFastRng64 Rng;
+            const TString FailureErrorReason;
+
+        public:
+            TBlobStorageFailureInjectingActor(
+                    TActorId realProxy,
+                    TGroupId groupId,
+                    TBSFailureInjectionConfig config)
+                : TActor(&TThis::StateWork)
+                , RealProxy(realProxy)
+                , GroupId(groupId)
+                , FailureProbability(std::clamp(config.GetFailureProbability(), 0.0, 1.0))
+                , RandomFailureSeed(config.HasRandomSeed() ? config.GetRandomSeed() : RandomNumber<ui64>())
+                , ErrorReplyStatus(config.GetErrorReplyStatus())
+                , Rng(RandomFailureSeed)
+                , FailureErrorReason(TStringBuilder()
+                    << "injected by BSProxyInterceptor"
+                    << " group " << GroupId
+                    << " seed " << RandomFailureSeed)
+            {
+            }
+
+        private:
+            bool ShouldInjectFailure()
+            {
+                return Rng.GenRandReal4() < FailureProbability;
+            }
+
+            template <typename TRequest>
+            bool MaybeInjectFailure(
+                TAutoPtr<IEventHandle>& ev,
+                TRequest& request,
+                const TString& eventName)
+            {
+                if (!ShouldInjectFailure()) {
+                    return false;
+                }
+
+                auto response = request.MakeErrorResponse(ErrorReplyStatus, FailureErrorReason, GroupId);
+                response->ExecutionRelay = std::move(request.ExecutionRelay);
+
+                LOG_WARN_S(*TlsActivationContext, NKikimrServices::BS_PROXY,
+                    "[BSProxyInterceptor] group " << GroupId
+                    << " injecting " << eventName
+                    << " failure; not forwarding to " << RealProxy.ToString()
+                    << " sender " << ev->Sender.ToString()
+                    << " cookie " << ev->Cookie
+                    << " probability " << FailureProbability
+                    << " seed " << RandomFailureSeed);
+
+                Send(ev->Sender, response.release(), 0, ev->Cookie);
+                return true;
+            }
+
+            STFUNC(StateWork)
+            {
+                LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_PROXY,
+                    "[BSProxyInterceptor] group " << GroupId << " " << ev->GetTypeName() << " " << ev->ToString());
+
+                #define HANDLE_EVENT(evType)                                        \
+                    case evType::EventType: {                                       \
+                        auto* msg = ev->Get<evType>();                              \
+                        if (MaybeInjectFailure(ev, *msg, ev->GetTypeName())) {      \
+                            return;                                                 \
+                        }                                                           \
+                        break;                                                      \
+                    }
+
+                switch (ev->GetTypeRewrite()) {
+                    DSPROXY_ENUM_EVENTS(HANDLE_EVENT)
+                    case TEvents::TEvPoison::EventType: {
+                        TActor::PassAway();
+                        [[fallthrough]];
+                    }
+                    default: {
+                        LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_PROXY,
+                            "[BSProxyInterceptor] group " << GroupId
+                            << " skipping event type " << ev->GetTypeName());
+                        break;
+                    }
+                }
+
+                TActivationContext::Forward(ev, RealProxy);
+                #undef HANDLE_EVENT
+            }
+        };
+
     } // anon
 
     IActor *CreateBlobStorageGroupProxyMockActor(TIntrusivePtr<NFake::TProxyDS> model) {
@@ -158,4 +259,8 @@ namespace NKikimr {
         return new TBlobStorageGroupProxyMockActor(groupId);
     }
 
+    IActor *CreateBlobStorageGroupFailureInjectingActor(TActorId actorId, TGroupId groupId, const TBSFailureInjectionConfig& config) {
+        return new TBlobStorageFailureInjectingActor(actorId, groupId, config);
+    }
+
 } // NKikimr
diff --git a/ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.h b/ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.h
index eb0aa2e0e51..ea57c8dd1cc 100644
--- a/ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.h
+++ b/ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.h
@@ -2,6 +2,8 @@
 
 #include "defs.h"
 #include <ydb/core/base/blobstorage_common.h>
+#include <ydb/core/protos/blobstorage.pb.h>
+
 namespace NKikimr {
 
     namespace NFake {
@@ -11,4 +13,7 @@ namespace NKikimr {
     IActor *CreateBlobStorageGroupProxyMockActor(TIntrusivePtr<NFake::TProxyDS> model);
     IActor *CreateBlobStorageGroupProxyMockActor(TGroupId groupId);
 
+    using TBSFailureInjectionConfig = NKikimrBlobStorage::TNodeWardenServiceSet::TFailureInjectionConfig;
+    IActor *CreateBlobStorageGroupFailureInjectingActor(TActorId actorId, TGroupId groupId, const TBSFailureInjectionConfig& config);
+
 } // NKikimr
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_impl.h b/ydb/core/blobstorage/nodewarden/node_warden_impl.h
index 1416551804a..7c5c13b3541 100644
--- a/ydb/core/blobstorage/nodewarden/node_warden_impl.h
+++ b/ydb/core/blobstorage/nodewarden/node_warden_impl.h
@@ -6,6 +6,7 @@
 #include <ydb/core/base/tablet_pipe.h>
 #include <ydb/core/blobstorage/dsproxy/group_sessions.h>
 #include <ydb/core/blobstorage/dsproxy/dsproxy_nodemon.h>
+#include <ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.h>
 #include <ydb/core/blobstorage/incrhuge/incrhuge.h>
 #include <ydb/core/cms/console/configs_dispatcher.h>
 #include <ydb/core/cms/console/console.h>
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_proxy.cpp b/ydb/core/blobstorage/nodewarden/node_warden_proxy.cpp
index 21056c02f9a..688b5824fb0 100644
--- a/ydb/core/blobstorage/nodewarden/node_warden_proxy.cpp
+++ b/ydb/core/blobstorage/nodewarden/node_warden_proxy.cpp
@@ -126,8 +126,23 @@ void TNodeWarden::StartLocalProxy(ui32 groupId) {
     // subscribe for group information changes through distconf cache
     Send(SelfId(), new TEvNodeWardenQueryCache(Sprintf("G%08" PRIx32, groupId), true));
 
-    group.ProxyId = as->Register(proxy.release(), TMailboxType::ReadAsFilled, AppData()->SystemPoolId);
-    as->RegisterLocalService(MakeBlobStorageProxyID(groupId), group.ProxyId);
+    auto id = as->Register(proxy.release(), TMailboxType::ReadAsFilled, AppData()->SystemPoolId);
+
+    // determine if we want to inject BS errors
+    if (Cfg->BlobStorageConfig.GetServiceSet().HasFailureInjectionConfig()) {
+        auto const& fiConfig = Cfg->BlobStorageConfig.GetServiceSet().GetFailureInjectionConfig();
+        if (fiConfig.GetFailureProbability() > 0) {
+            const auto gid = TGroupId::FromValue(groupId);
+            if (IsDynamicGroup(gid) || fiConfig.GetIncludeStaticGroups()) {
+                id = as->Register(
+                    CreateBlobStorageGroupFailureInjectingActor(id, gid, fiConfig),
+                    TMailboxType::ReadAsFilled, AppData()->SystemPoolId);
+            }
+        }
+    }
+
+    group.ProxyId = id;
+    as->RegisterLocalService(MakeBlobStorageProxyID(groupId), id);
 }
 
 void TNodeWarden::StartVirtualGroupAgent(ui32 groupId) {
diff --git a/ydb/core/protos/blobstorage.proto b/ydb/core/protos/blobstorage.proto
index a83698da532..d3e00489294 100644
--- a/ydb/core/protos/blobstorage.proto
+++ b/ydb/core/protos/blobstorage.proto
@@ -1059,12 +1059,20 @@ message TNodeWardenServiceSet {
         optional uint64 TotalResponseBytesPerSecond = 6; // quota for generated responses at network layer
     }
 
+    message TFailureInjectionConfig {
+        optional double FailureProbability = 1; // positive values enable failure injection
+        optional bool IncludeStaticGroups = 2; // also inject failures in request to static groups
+        optional uint64 RandomSeed = 3;
+        optional NKikimrProto.EReplyStatus ErrorReplyStatus = 4 [default = ERROR];
+    }
+
     repeated TPDisk PDisks = 1;
     repeated TVDisk VDisks = 2;
     repeated NKikimrBlobStorage.TGroupInfo Groups = 3;
     repeated uint64 AvailabilityDomains = 4;
     optional TReplBrokerConfig ReplBrokerConfig = 5;
     optional bool EnableProxyMock = 7;
+    optional TFailureInjectionConfig FailureInjectionConfig = 8;
 }
 
 message TNodeWardenCache {