Allow setting max subsystems limit in SPDK.

Fixes #1103

Signed-off-by: Gil Bregman <gbregman@il.ibm.com>

.github/workflows/build-container.yml

@@ -575,7 +575,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test: ["sanity", "no_huge", "ns_lb_change", "no_subsystems", "auto_load_balance", "state_transitions", "state_transitions_both_gws", "state_transitions_loop", "state_transitions_rand_loop", "late_registration", "late_registration_loop", "4gws", "4gws_loop", "4gws_create_delete", "4gws_create_delete_loop", "namespaces", "namespaces_loop", "mtls", "notify", "ceph_status", "blocklist", "main_exit", "cluster_pool", "flat_bdev_per_cluster", "set_qos", "set_qos_2ms"]
+        test: ["sanity", "no_huge", "ns_lb_change", "no_subsystems", "auto_load_balance", "state_transitions", "state_transitions_both_gws", "state_transitions_loop", "state_transitions_rand_loop", "late_registration", "late_registration_loop", "4gws", "4gws_loop", "4gws_create_delete", "4gws_create_delete_loop", "namespaces", "namespaces_loop", "mtls", "notify", "ceph_status", "blocklist", "main_exit", "cluster_pool", "flat_bdev_per_cluster", "set_qos", "set_qos_2ms", "max_subsystems"]
     runs-on: ubuntu-latest
     env:
       HUGEPAGES: 1024  # 4 spdk instances

ceph-nvmeof.conf

@@ -98,7 +98,8 @@ transport_tcp_options = {"in_capsule_data_size" : 8192, "max_io_qpairs_per_ctrlr
 # Example value: {"small_pool_count" : 8192, "large_pool_count" : 1024, "small_bufsize" : 8192, "large_bufsize" : 135168}
 # iobuf_options =
 
-qos_timeslice_in_usecs = 0
+#qos_timeslice_in_usecs = 0
+#max_subsystems = 0
 
 [monitor]
 #timeout = 1.0

control/grpc.py

@@ -600,6 +600,12 @@ def __init__(self, config: GatewayConfig, gateway_state: GatewayStateHandler,
             "gateway",
             "max_subsystems",
             GatewayService.MAX_SUBSYSTEMS_DEFAULT)
+        spdk_max_subsystems = self.config.getint_with_default("spdk", "max_subsystems", 0)
+        if spdk_max_subsystems > 0 and self.max_subsystems > spdk_max_subsystems:
+            self.logger.warning(f"Maximal number of subsystems can't be bigger than the SPDK "
+                                f"limit, will decrease it from {self.max_subsystems} to "
+                                f"{spdk_max_subsystems}")
+            self.max_subsystems = spdk_max_subsystems
         self.max_namespaces = self.config.getint_with_default(
             "gateway",
             "max_namespaces",

control/server.py

@@ -501,6 +501,7 @@ def _start_spdk(self, omap_state):
 
         # Start target
         self.logger.debug(f"Configuring server {self.name}")
+        waiting_for_rpc = False
         spdk_tgt_path = self.config.get("spdk", "tgt_path")
         self.logger.info(f"SPDK Target Path: {spdk_tgt_path}")
         sockdir = self.config.get_with_default("spdk", "rpc_socket_dir", "/var/tmp/")
@@ -525,7 +526,9 @@ def _start_spdk(self, omap_state):
         cmd = [spdk_tgt_path, "-u", "-r", self.spdk_rpc_socket_path]
 
         iobuf_options = self.config.get_with_default("spdk", "iobuf_options", "")
-        if iobuf_options:
+        max_subsystems = self.config.getint_with_default("spdk", "max_subsystems", 0)
+        if iobuf_options or max_subsystems > 0:
+            waiting_for_rpc = True
             cmd += ["--wait-for-rpc"]
 
         # Add extra args from the conf file
@@ -598,8 +601,16 @@ def _start_spdk(self, omap_state):
                 conn_retries=conn_retries,
             )
 
+            # Set max subsystems
+            if max_subsystems > 0:
+                self._set_max_subsystems(max_subsystems)
+
             # Initialize pool and buffer sizes
-            self._initialize_iobuf_options(iobuf_options)
+            if iobuf_options:
+                self._initialize_iobuf_options(iobuf_options)
+
+            if waiting_for_rpc:
+                self._initialize_framework()
 
             self.spdk_rpc_ping_client = rpc_client.JSONRPCClient(
                 self.spdk_rpc_socket_path,
@@ -713,12 +724,18 @@ def _stop_discovery(self):
 
         self.discovery_pid = None
 
+    def _set_max_subsystems(self, max_subsystems):
+        """Sets SPDK's max subsystems attribute."""
+
+        try:
+            rpc_nvmf.nvmf_set_max_subsystems(self.spdk_rpc_client, max_subsystems=max_subsystems)
+        except Exception:
+            self.logger.exception(f"Failure setting max subsystems {max_subsystems}")
+            pass
+
     def _initialize_iobuf_options(self, options):
         """Initialize pool and buffer sizes."""
 
-        if not options:
-            return
-
         args = {}
         self.logger.debug(f"initialize_iobuf_options: options: {options}")
 
@@ -734,6 +751,9 @@ def _initialize_iobuf_options(self, options):
             self.logger.exception("IObuf set options returned with error")
             pass
 
+    def _initialize_framework(self):
+        """In case we started SPDK with the "wait for rpc" flag, we need to call this"""
+
         try:
             spdk.rpc.framework_start_init(self.spdk_rpc_client)
         except Exception:

tests/ceph-nvmeof.cluster_pool.conf

@@ -92,7 +92,8 @@ transport_tcp_options = {"in_capsule_data_size" : 8192, "max_io_qpairs_per_ctrlr
 # Example value: {"small_pool_count" : 8192, "large_pool_count" : 1024, "small_bufsize" : 8192, "large_bufsize" : 135168}
 # iobuf_options =
 
-# qos_timeslice_in_usecs = 0
+#qos_timeslice_in_usecs = 0
+#max_subsystems = 0
 
 [monitor]
 #timeout = 1.0

tests/ceph-nvmeof.flat_bdevs_per_cluster.conf

@@ -92,7 +92,8 @@ transport_tcp_options = {"in_capsule_data_size" : 8192, "max_io_qpairs_per_ctrlr
 # Example value: {"small_pool_count" : 8192, "large_pool_count" : 1024, "small_bufsize" : 8192, "large_bufsize" : 135168}
 # iobuf_options =
 
-# qos_timeslice_in_usecs = 0
+#qos_timeslice_in_usecs = 0
+#max_subsystems = 0
 
 [monitor]
 #timeout = 1.0

tests/ceph-nvmeof.no-huge.conf

@@ -93,7 +93,8 @@ transport_tcp_options = {"in_capsule_data_size" : 8192, "max_io_qpairs_per_ctrlr
 # Example value: {"small_pool_count" : 8192, "large_pool_count" : 1024, "small_bufsize" : 8192, "large_bufsize" : 135168}
 # iobuf_options =
 
-# qos_timeslice_in_usecs = 0
+#qos_timeslice_in_usecs = 0
+#max_subsystems = 0
 
 [monitor]
 #timeout = 1.0

tests/ceph-nvmeof.tls.conf

@@ -91,7 +91,8 @@ transport_tcp_options = {"in_capsule_data_size" : 8192, "max_io_qpairs_per_ctrlr
 # Example value: {"small_pool_count" : 8192, "large_pool_count" : 1024, "small_bufsize" : 8192, "large_bufsize" : 135168}
 # iobuf_options =
 
-# qos_timeslice_in_usecs = 0
+#qos_timeslice_in_usecs = 0
+#max_subsystems = 0
 
 [monitor]
 #timeout = 1.0

tests/test_max_subsystems.py

@@ -0,0 +1,92 @@
+import pytest
+from control.server import GatewayServer
+from control.cli import main as cli
+from control.cephutils import CephUtils
+import grpc
+
+config = "ceph-nvmeof.conf"
+pool = "rbd"
+group_name = "group1"
+subsystem_prefix = "nqn.2016-06.io.spdk:cnode"
+max_subsystems = 4
+spdk_max_subsystems = 3
+
+
+@pytest.fixture(scope="module")
+def gateway(config, request):
+    """Sets up and tears down Gateway"""
+
+    addr = config.get("gateway", "addr")
+    port = config.getint("gateway", "port")
+    config.config["gateway"]["group"] = group_name
+    config.config["gateway"]["max_subsystems"] = str(max_subsystems)
+    config.config["gateway-logs"]["log_level"] = "debug"
+    config.config["spdk"]["max_subsystems"] = str(spdk_max_subsystems)
+    ceph_utils = CephUtils(config)
+
+    with GatewayServer(config) as gateway:
+
+        # Start gateway
+        gateway.gw_logger_object.set_log_level("debug")
+        ceph_utils.execute_ceph_monitor_command(
+            "{" + f'"prefix":"nvme-gw create", "id": "{gateway.name}", "pool": "{pool}", '
+            f'"group": "{group_name}"' + "}"
+        )
+        gateway.serve()
+
+        # Bind the client and Gateway
+        grpc.insecure_channel(f"{addr}:{port}")
+        yield gateway
+
+        # Stop gateway
+        gateway.server.stop(grace=1)
+        gateway.gateway_rpc.gateway_state.delete_state()
+
+
+class TestMaxSubsystems:
+    def test_max_subsystems(self, caplog, gateway):
+        gw = gateway
+        found = False
+        lookfor = f"Maximal number of subsystems can't be bigger than the SPDK limit, will " \
+                  f"decrease it from {max_subsystems} to {spdk_max_subsystems}"
+        for oneline in caplog.get_records("setup"):
+            if oneline.message == lookfor:
+                found = True
+                break
+        assert found
+
+        for i in range(spdk_max_subsystems):
+            subsys = f"{subsystem_prefix}{i}"
+            caplog.clear()
+            cli(["subsystem", "add", "--subsystem", f"{subsys}", "--no-group-append"])
+            assert f"Adding subsystem {subsys}: Successful" in caplog.text
+
+        caplog.clear()
+        subsys = f"{subsystem_prefix}XXX"
+        cli(["subsystem", "add", "--subsystem", f"{subsys}", "--no-group-append"])
+        assert f"Failure creating subsystem {subsys}: Maximal number of subsystems " \
+               f"({spdk_max_subsystems}) has already been reached" in caplog.text
+
+        for i in range(spdk_max_subsystems):
+            subsys = f"{subsystem_prefix}{i}"
+            caplog.clear()
+            cli(["subsystem", "del", "--subsystem", f"{subsys}"])
+            assert f"Deleting subsystem {subsys}: Successful" in caplog.text
+
+        caplog.clear()
+        cli(["subsystem", "list"])
+        assert "No subsystems" in caplog.text
+
+        gw.gateway_rpc.max_subsystems = 100
+        for i in range(spdk_max_subsystems):
+            subsys = f"{subsystem_prefix}{i}"
+            caplog.clear()
+            cli(["subsystem", "add", "--subsystem", f"{subsys}", "--no-group-append"])
+            assert f"Adding subsystem {subsys}: Successful" in caplog.text
+
+        subsys = f"{subsystem_prefix}XXX"
+        caplog.clear()
+        cli(["subsystem", "add", "--subsystem", f"{subsys}", "--no-group-append"])
+        assert f"Failure creating subsystem {subsys}: Maximal number of subsystems " \
+               f"({spdk_max_subsystems}) has already been reached" not in caplog.text
+        assert f'"message": "Unable to create subsystem {subsys}"' in caplog.text