avoid reconnecting live nbd sockets; resolves #3138

cloud/blockstore/libs/endpoint_proxy/server/bootstrap.cpp

@@ -31,6 +31,7 @@ void TBootstrap::Init()
         Options.NbdRequestTimeout,
         Options.NbdReconnectDelay,
         Options.WithoutLibnl,
+        Options.RestartEvents,
     },
     CreateWallClockTimer(),
     Scheduler,

cloud/blockstore/libs/endpoint_proxy/server/options.cpp

@@ -33,20 +33,26 @@ TOptions::TOptions()
         .StoreResult(&StoredEndpointsPath);
 
     Opts.AddLongOption("nbd-request-timeout")
-        .OptionalArgument("NUM")
+        .RequiredArgument("NUM")
         .Handler1T<TString>([this] (const auto& s) {
             NbdRequestTimeout = TDuration::Parse(s);
         });
 
     Opts.AddLongOption("nbd-reconnect-delay")
-        .OptionalArgument("NUM")
+        .RequiredArgument("NUM")
         .Handler1T<TString>([this] (const auto& s) {
             NbdReconnectDelay = TDuration::Parse(s);
         });
 
     Opts.AddLongOption("without-libnl")
         .NoArgument()
         .SetFlag(&WithoutLibnl);
+
+    Opts.AddLongOption(
+            "restart-events",
+            "debug option. issue multiple restart events for each io error")
+        .RequiredArgument("NUM")
+        .StoreResult(&RestartEvents);
 }
 
 void TOptions::Parse(int argc, char** argv)

cloud/blockstore/libs/endpoint_proxy/server/options.h

@@ -21,6 +21,7 @@ struct TOptions final: TOptionsBase
     TDuration NbdRequestTimeout = TDuration::Minutes(10);
     TDuration NbdReconnectDelay = TDuration::MilliSeconds(100);
     bool WithoutLibnl = false;
+    ui32 RestartEvents = 1;
 
     TOptions();
 

cloud/blockstore/libs/endpoint_proxy/server/server.cpp

@@ -79,12 +79,11 @@ struct TEndpoint: std::enable_shared_from_this<TEndpoint>
     NBD::IDevicePtr NbdDevice;
     std::unique_ptr<TNetworkAddress> ListenAddress;
     IProxyRequestStatsPtr RequestStats;
-
     TString UnixSocketPath;
     TString InternalUnixSocketPath;
     TString NbdDevicePath;
-
     NBD::TStorageOptions NbdOptions;
+    std::atomic<ui64> Generation = 0;
 };
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -191,16 +190,19 @@ struct TRestartAlarmContext: TRequestContextBase
 {
     std::weak_ptr<TEndpoint> Endpoint;
     grpc::ServerCompletionQueue& CQ;
+    ui64 Generation;
     TBackoffDelayProvider Backoff;
     grpc::Alarm Alarm;
 
     TRestartAlarmContext(
             std::weak_ptr<TEndpoint> endpoint,
             grpc::ServerCompletionQueue& cq,
+            ui64 generation,
             TDuration minReconnectDelay,
             TDuration maxReconnectDelay)
         : Endpoint(std::move(endpoint))
         , CQ(cq)
+        , Generation(generation)
         , Backoff{minReconnectDelay, maxReconnectDelay}
     {
         SetAlarm();
@@ -275,27 +277,36 @@ struct TServer: IEndpointProxyServer
     {
         std::weak_ptr<TEndpoint> Endpoint;
         grpc::ServerCompletionQueue& CQ;
+        ui64 Generation;
         TDuration ReconnectDelay;
+        ui32 RestartEvents;
 
         TErrorHandler(
                 std::weak_ptr<TEndpoint> endpoint,
                 grpc::ServerCompletionQueue& cq,
-                TDuration reconnectDelay)
+                ui64 generation,
+                TDuration reconnectDelay,
+                ui32 restartEvents)
             : Endpoint(std::move(endpoint))
             , CQ(cq)
+            , Generation(generation)
             , ReconnectDelay(reconnectDelay)
+            , RestartEvents(restartEvents)
         {}
 
         void ProcessException(std::exception_ptr e) override
         {
             Y_UNUSED(e);
 
-            // context will be held by grpc until CQ->Next returns it
-            new TRestartAlarmContext(
-                Endpoint,
-                CQ,
-                ReconnectDelay,
-                MAX_RECONNECT_DELAY);
+            for (ui32 i = 0; i < RestartEvents; i++) {
+                // context will be held by grpc until CQ->Next returns it
+                new TRestartAlarmContext(
+                    Endpoint,
+                    CQ,
+                    Generation,
+                    ReconnectDelay,
+                    MAX_RECONNECT_DELAY);
+            }
         }
     };
 
@@ -945,7 +956,9 @@ struct TServer: IEndpointProxyServer
             std::make_shared<TErrorHandler>(
                 ep.shared_from_this(),
                 *CQ,
-                Config.NbdReconnectDelay),
+                ep.Generation,
+                Config.NbdReconnectDelay,
+                Config.RestartEvents),
             ep.NbdOptions);
 
         // TODO fix StartEndpoint signature - it's actually synchronous
@@ -1007,6 +1020,11 @@ struct TServer: IEndpointProxyServer
     bool ProcessAlarm(TRestartAlarmContext* context)
     {
         if (auto ep = context->Endpoint.lock()) {
+            if (context->Generation != ep->Generation) {
+                return false;
+            }
+            ep->Generation++;
+
             const auto tag = TStringBuilder()
                 << "UnixSocketPath: " << ep->UnixSocketPath.Quote() << " - ";
 

cloud/blockstore/libs/endpoint_proxy/server/server.h

@@ -30,6 +30,7 @@ struct TEndpointProxyServerConfig
     TDuration NbdRequestTimeout;
     TDuration NbdReconnectDelay;
     bool WithoutLibnl;
+    ui32 RestartEvents;
 
     TEndpointProxyServerConfig(
             ui16 port,
@@ -42,7 +43,8 @@ struct TEndpointProxyServerConfig
             TString storedEndpointsPath,
             TDuration nbdRequestTimeout,
             TDuration nbdReconnectDelay,
-            bool withoutLibnl)
+            bool withoutLibnl,
+            ui32 restartEvents)
         : Port(port)
         , SecurePort(securePort)
         , RootCertsFile(std::move(rootCertsFile))
@@ -54,6 +56,7 @@ struct TEndpointProxyServerConfig
         , NbdRequestTimeout(nbdRequestTimeout)
         , NbdReconnectDelay(nbdReconnectDelay)
         , WithoutLibnl(withoutLibnl)
+        , RestartEvents(restartEvents)
     {
     }
 };

cloud/blockstore/tests/e2e-tests/test.py

@@ -36,6 +36,8 @@ def init(
         stored_endpoints_path=None,
         nbd_request_timeout=None,
         nbd_reconnect_delay=None,
+        restart_interval=None,
+        proxy_restart_events=None,
 ):
     server_config_patch = TServerConfig()
     server_config_patch.NbdEnabled = True
@@ -71,11 +73,13 @@ def init(
         server_app_config=server,
         storage_config_patches=None,
         use_in_memory_pdisks=True,
+        restart_interval=restart_interval,
         with_endpoint_proxy=with_endpoint_proxy,
         with_netlink=with_netlink,
         stored_endpoints_path=stored_endpoints_path,
         nbd_request_timeout=nbd_request_timeout,
-        nbd_reconnect_delay=nbd_reconnect_delay)
+        nbd_reconnect_delay=nbd_reconnect_delay,
+        proxy_restart_events=proxy_restart_events)
 
     client_config_path = Path(yatest_common.output_path()) / "client-config.txt"
     client_config = TClientAppConfig()
@@ -127,6 +131,88 @@ def log_called_process_error(exc):
     )
 
 
+def test_multiple_errors():
+    env, run = init(
+        restart_interval=5,
+        with_netlink=True,
+        with_endpoint_proxy=True,
+        nbd_request_timeout="1s",
+        proxy_restart_events=2)
+
+    volume_name = "test-disk"
+    block_size = 4096
+    blocks_count = 1024
+    nbd_device = "/dev/nbd0"
+    socket_path = "/tmp/nbd.sock"
+    runtime = 10
+    numjobs = blocks_count
+
+    try:
+        result = run(
+            "createvolume",
+            "--disk-id",
+            volume_name,
+            "--blocks-count",
+            str(blocks_count),
+            "--block-size",
+            str(block_size),
+        )
+        assert result.returncode == 0
+
+        result = run(
+            "startendpoint",
+            "--disk-id",
+            volume_name,
+            "--socket",
+            socket_path,
+            "--ipc-type",
+            "nbd",
+            "--persistent",
+            "--nbd-device",
+            nbd_device
+        )
+        assert result.returncode == 0
+
+        proc = subprocess.Popen(
+            [
+                "fio",
+                "--name=fio",
+                "--ioengine=sync",
+                "--direct=1",
+                "--time_based=1",
+                "--rw=randrw",
+                "--rwmixread=50",
+                "--filename=" + nbd_device,
+                "--runtime=" + str(runtime),
+                "--blocksize=" + str(block_size),
+                "--numjobs=" + str(numjobs),
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT)
+        proc.communicate(timeout=60)
+        assert proc.returncode == 0
+
+    except subprocess.CalledProcessError as e:
+        log_called_process_error(e)
+        raise
+
+    finally:
+        run(
+            "stopendpoint",
+            "--socket",
+            socket_path,
+        )
+
+        result = run(
+            "destroyvolume",
+            "--disk-id",
+            volume_name,
+            input=volume_name,
+        )
+
+        cleanup_after_test(env)
+
+
 def test_stop_start():
     env, run = init(
         with_netlink=True,

cloud/blockstore/tests/python/lib/endpoint_proxy.py

@@ -11,7 +11,8 @@ def __init__(
             with_netlink,
             stored_endpoints_path,
             nbd_request_timeout,
-            nbd_reconnect_delay
+            nbd_reconnect_delay,
+            restart_events,
     ):
         command = [yatest_common.binary_path(
             "cloud/blockstore/apps/endpoint_proxy/blockstore-endpoint-proxy")]
@@ -30,6 +31,9 @@ def __init__(
         if nbd_reconnect_delay:
             command += ["--nbd-reconnect-delay", nbd_reconnect_delay]
 
+        if restart_events:
+            command += ["--restart-events", str(restart_events)]
+
         super(EndpointProxy, self).__init__(
             commands=[command],
             cwd=working_dir,

cloud/blockstore/tests/python/lib/loadtest_env.py

@@ -58,6 +58,7 @@ def __init__(
             stored_endpoints_path=None,
             nbd_request_timeout=None,
             nbd_reconnect_delay=None,
+            proxy_restart_events=None,
     ):
 
         self.__endpoint = endpoint
@@ -131,7 +132,8 @@ def __init__(
                 with_netlink=with_netlink,
                 stored_endpoints_path=stored_endpoints_path,
                 nbd_request_timeout=nbd_request_timeout,
-                nbd_reconnect_delay=nbd_reconnect_delay)
+                nbd_reconnect_delay=nbd_reconnect_delay,
+                restart_events=proxy_restart_events)
 
         if run_kikimr:
             self.nbs.setup_cms(self.kikimr_cluster.client)