Make sure to use the correct NSID when creating namespaces in update().

Fixes #435

Signed-off-by: Gil Bregman <[email protected]>

.github/workflows/build-container.yml

@@ -88,7 +88,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test: ["cli", "state", "multi_gateway", "server", "grpc", "omap_lock", "old_omap", "log_files"]
+        test: ["cli", "state", "multi_gateway", "server", "grpc", "omap_lock", "old_omap", "log_files", "nsid"]
     runs-on: ubuntu-latest
     env:
       HUGEPAGES: 512  # for multi gateway test, approx 256 per gateway instance

control/grpc.py

@@ -139,8 +139,9 @@ def __init__(self, config: GatewayConfig, gateway_state: GatewayStateHandler, om
                         self.logger.warning(f"The actual huge page count {hugepages_val} differs from the requested value of {requested_hugepages_val}")
                 else:
                     self.logger.warning(f"Can't read actual huge pages count value from {hugepages_file}")
-            except Exception as ex:
+            except Exception:
                 self.logger.exception(f"Can't read actual huge pages count value from {hugepages_file}")
+                pass
         else:
             self.logger.warning(f"Can't find huge pages file {hugepages_file}")
         self.config = config
@@ -176,7 +177,7 @@ def parse_json_exeption(self, ex):
             return None
 
         json_error_text = "Got JSON-RPC error response"
-        rsp = None
+        resp = None
         try:
             resp_index = ex.message.find(json_error_text)
             if resp_index >= 0:
@@ -191,6 +192,14 @@ def parse_json_exeption(self, ex):
         if resp:
             if resp["code"] < 0:
                 resp["code"] = -resp["code"]
+        else:
+            resp={}
+            if "timeout" in ex.message.lower():
+                resp["code"] = errno.ETIMEDOUT
+            else:
+                resp["code"] = errno.EINVAL
+            resp["message"] = ex.message
+
         return resp
 
     def _init_cluster_context(self) -> None:
@@ -258,6 +267,10 @@ def create_bdev(self, name, uuid, rbd_pool_name, rbd_image_name, block_size, cre
                          f" {rbd_pool_name}/{rbd_image_name} (size {rbd_image_size} MiB)"
                          f" with block size {block_size}, {cr_img_msg}")
 
+        if block_size == 0:
+            return BdevStatus(status=errno.EINVAL,
+                                   error_message=f"Failure creating bdev {name}: block size can not be zero")
+
         if create_image:
             rc = self.ceph_utils.pool_exists(rbd_pool_name)
             if not rc:
@@ -272,6 +285,7 @@ def create_bdev(self, name, uuid, rbd_pool_name, rbd_image_name, block_size, cre
                     self.logger.info(f"Image {rbd_image_name} already exists")
             except Exception:
                 self.logger.exception(f"Can't create RBD image {rbd_image_name}")
+                pass
                 return BdevStatus(status=errno.ENODEV, error_message=f"Failure creating bdev {name}: Can't create RBD image {rbd_image_name}")
 
         try:
@@ -621,6 +635,7 @@ def check_if_image_used(self, pool_name, image_name):
                     break
             except Exception:
                 self.logger.exception(f"Got exception while parsing {val}")
+                pass
                 continue
         return errmsg, nqn
 
@@ -740,7 +755,8 @@ def set_ana_state_safe(self, ana_info: pb2.ana_info, context=None):
                         if not ret:
                             raise Exception(f"nvmf_subsystem_listener_set_ana_state({nqn=}, {listener=}, {ana_state=}, {grp_id=}) error")
                     except Exception as ex:
-                        self.logger.exception("nvmf_subsystem_listener_set_ana_state: ")
+                        self.logger.exception("nvmf_subsystem_listener_set_ana_state:")
+                        pass
                         if context:
                             context.set_code(grpc.StatusCode.INTERNAL)
                             context.set_details(f"{ex}")
@@ -758,19 +774,23 @@ def namespace_add_safe(self, request, context):
             request.uuid = str(uuid.uuid4())
 
         with self.omap_lock(context=context):
-            errmsg, ns_nqn = self.check_if_image_used(request.rbd_pool_name, request.rbd_image_name)
-            if errmsg and ns_nqn:
-                if request.force:
-                    self.logger.warning(f"{errmsg}, will continue as the \"force\" argument was used")
-                else:
-                    errmsg = f"{errmsg}, either delete the namespace or use the \"force\" argument,\nyou can find the offending namespace by using the \"namespace list --subsystem {ns_nqn}\" CLI command"
-                    self.logger.error(errmsg)
-                    return pb2.nsid_status(status=errno.EEXIST, error_message=errmsg)
+            if context:
+                errmsg, ns_nqn = self.check_if_image_used(request.rbd_pool_name, request.rbd_image_name)
+                if errmsg and ns_nqn:
+                    if request.force:
+                        self.logger.warning(f"{errmsg}, will continue as the \"force\" argument was used")
+                    else:
+                        errmsg = f"{errmsg}, either delete the namespace or use the \"force\" argument,\nyou can find the offending namespace by using the \"namespace list --subsystem {ns_nqn}\" CLI command"
+                        self.logger.error(errmsg)
+                        return pb2.nsid_status(status=errno.EEXIST, error_message=errmsg)
 
             bdev_name = self.find_unique_bdev_name(request.uuid)
 
+            create_image = request.create_image
+            if not context:
+                create_image = False
             ret_bdev = self.create_bdev(bdev_name, request.uuid, request.rbd_pool_name,
-                                        request.rbd_image_name, request.block_size, request.create_image, request.size)
+                                        request.rbd_image_name, request.block_size, create_image, request.size)
             if ret_bdev.status != 0:
                 errmsg = f"Failure adding namespace {nsid_msg}to {request.subsystem_nqn}: {ret_bdev.error_message}"
                 self.logger.error(errmsg)
@@ -788,6 +808,13 @@ def namespace_add_safe(self, request, context):
                 self.logger.warning(f"Returned bdev name {ret_bdev.bdev_name} differs from requested one {bdev_name}")
 
             ret_ns = self.create_namespace(request.subsystem_nqn, bdev_name, request.nsid, request.anagrpid, request.uuid, context)
+
+            if ret_ns.status == 0 and request.nsid and ret_ns.nsid != request.nsid:
+                errmsg = f"Returned NSID {ret_ns.nsid} differs from requested one {request.nsid}"
+                self.logger.error(errmsg)
+                ret_ns.status = errno.EINVAL
+                ret_ns.error_message = errmsg
+
             if ret_ns.status != 0:
                 try:
                     ret_del = self.delete_bdev(bdev_name)
@@ -799,11 +826,9 @@ def namespace_add_safe(self, request, context):
                 self.logger.error(errmsg)
                 return pb2.nsid_status(status=ret_ns.status, error_message=errmsg)
 
-            if request.nsid and ret_ns.nsid != request.nsid:
-                self.logger.warning(f"Return NSID {ret_ns.nsid} differs from requested one {request.nsid}")
-
             if context:
                 # Update gateway state
+                request.nsid = ret_ns.nsid
                 try:
                     json_req = json_format.MessageToJson(
                         request, preserving_proto_field_name=True, including_default_value_fields=True)
@@ -1007,6 +1032,7 @@ def get_bdev_info(self, bdev_name, need_to_lock):
                 ret_bdev = bdevs[0]
             except Exception:
                 self.logger.exception(f"Got exception while getting bdev {bdev_name} info")
+                pass
 
         return ret_bdev
 
@@ -1178,8 +1204,8 @@ def namespace_get_io_stats(self, request, context=None):
             return io_stats
         except Exception as ex:
             self.logger.exception(f"{s=} parse error: ")
-            exmsg = str(ex)
             pass
+            exmsg = str(ex)
 
         return pb2.namespace_io_stats_info(status=errno.EINVAL,
                                error_message=f"Failure getting IO stats for namespace {nsid_msg}on {request.subsystem_nqn}: Error parsing returned stats:\n{exmsg}") 
@@ -2092,7 +2118,7 @@ def get_subsystems_safe(self, request, context):
                 json_format.Parse(json.dumps(s), subsystem)
                 subsystems.append(subsystem)
             except Exception:
-                self.logger.exception(f"{s=} parse error: ")
+                self.logger.exception(f"{s=} parse error:")
                 raise
 
         return pb2.subsystems_info(subsystems=subsystems)

tests/test_cli.py

@@ -216,6 +216,16 @@ def test_create_subsystem_with_discovery_nqn(self, caplog, gateway):
         assert "Can't add a discovery subsystem" in caplog.text
         assert rc == 2
 
+    def test_add_namespace_wrong_block_size(self, caplog, gateway):
+        gw, stub = gateway
+        caplog.clear()
+        add_namespace_req = pb2.namespace_add_req(subsystem_nqn=subsystem, rbd_pool_name=pool, rbd_image_name="junkimage",
+                                                  create_image=True, size=16*1024*1024, force=True)
+        ret = stub.namespace_add(add_namespace_req)
+        assert ret.status != 0
+        assert f"Failure adding namespace" in caplog.text
+        assert f"block size can not be zero" in caplog.text
+
     def test_add_namespace(self, caplog, gateway):
         caplog.clear()
         cli(["namespace", "add", "--subsystem", subsystem, "--rbd-pool", "junk", "--rbd-image", image2, "--uuid", uuid, "--size", "16MiB", "--rbd-create-image"])

tests/test_nsid.py

@@ -0,0 +1,179 @@
+import pytest
+import copy
+import grpc
+import json
+import time
+from google.protobuf import json_format
+from control.server import GatewayServer
+from control.proto import gateway_pb2 as pb2
+from control.proto import gateway_pb2_grpc as pb2_grpc
+import spdk.rpc.bdev as rpc_bdev
+
+image = "mytestdevimage"
+pool = "rbd"
+subsystem_prefix = "nqn.2016-06.io.spdk:cnode"
+host_nqn_prefix = "nqn.2014-08.org.nvmexpress:uuid:22207d09-d8af-4ed2-84ec-a6d80b"
+
+def setup_config(config, gw1_name, gw2_name, gw_group, update_notify, update_interval_sec, disable_unlock, lock_duration,
+                 sock1_name, sock2_name):
+    """Sets up the config objects for gateways A and B """
+
+    configA = copy.deepcopy(config)
+    configA.config["gateway"]["name"] = gw1_name
+    configA.config["gateway"]["group"] = gw_group
+    configA.config["gateway"]["state_update_notify"] = str(update_notify)
+    configA.config["gateway"]["state_update_interval_sec"] = str(update_interval_sec)
+    configA.config["gateway"]["omap_file_disable_unlock"] = str(disable_unlock)
+    configA.config["gateway"]["omap_file_lock_duration"] = str(lock_duration)
+    configA.config["gateway"]["enable_spdk_discovery_controller"] = "True"
+    configA.config["spdk"]["rpc_socket_name"] = sock1_name
+    configB = copy.deepcopy(configA)
+    portA = configA.getint("gateway", "port")
+    configA.config["gateway"]["port"] = str(portA)
+    portB = portA + 1
+    configB.config["gateway"]["name"] = gw2_name
+    configB.config["gateway"]["port"] = str(portB)
+    configB.config["spdk"]["rpc_socket_name"] = sock2_name
+    configB.config["spdk"]["tgt_cmd_extra_args"] = "-m 0x02"
+
+    return configA, configB
+
+def start_servers(gatewayA, gatewayB, addr, portA, portB):
+    gatewayA.serve()
+    # Delete existing OMAP state
+    gatewayA.gateway_rpc.gateway_state.delete_state()
+    # Create new
+    gatewayB.serve()
+    gatewayB.gateway_rpc.gateway_state.delete_state()
+
+    # Bind the client and Gateways A & B
+    channelA = grpc.insecure_channel(f"{addr}:{portA}")
+    stubA = pb2_grpc.GatewayStub(channelA)
+    channelB = grpc.insecure_channel(f"{addr}:{portB}")
+    stubB = pb2_grpc.GatewayStub(channelB)
+
+    return stubA, stubB
+
+def test_multi_gateway_namespace_ids(config, image, caplog):
+    """Tests NSID are OK after a gateway restart
+    """
+    configA, configB = setup_config(config, "GatewayAAA", "GatewayBBB", "Group1", True, 5, False, 60,
+                                    "spdk_GatewayAAA.sock", "spdk_GatewayBBB.sock")
+
+    addr = configA.get("gateway", "addr")
+    portA = configA.getint("gateway", "port")
+    portB = configB.getint("gateway", "port")
+    # Start servers
+    with (
+       GatewayServer(configA) as gatewayA,
+       GatewayServer(configB) as gatewayB,
+    ):
+        stubA, stubB = start_servers(gatewayA, gatewayB, addr, portA, portB)
+
+        # Send requests to create a subsystem on GatewayA
+        caplog.clear()
+        subsystem = f"{subsystem_prefix}WWW"
+        subsystem_add_req = pb2.create_subsystem_req(subsystem_nqn=subsystem)
+        ret_subsystem = stubA.create_subsystem(subsystem_add_req)
+        assert ret_subsystem.status == 0
+        assert f"create_subsystem {subsystem}: True" in caplog.text
+        assert f"Failure creating subsystem {subsystem}" not in caplog.text
+        time.sleep(10)
+        caplog.clear()
+        # Send requests to create a namespace on GatewayA
+        namespace_req = pb2.namespace_add_req(subsystem_nqn=subsystem,
+                                              rbd_pool_name=pool, rbd_image_name=f"{image}WWW", block_size=4096,
+                                              create_image=True, size=16*1024*1024, force=True)
+        ret_ns = stubA.namespace_add(namespace_req)
+        assert ret_ns.status == 0
+        time.sleep(10)
+        namespace_req2 = pb2.namespace_add_req(subsystem_nqn=subsystem,
+                                              rbd_pool_name=pool, rbd_image_name=f"{image}EEE", block_size=4096,
+                                              create_image=True, size=16*1024*1024, force=True)
+        ret_ns = stubA.namespace_add(namespace_req2)
+        assert ret_ns.status == 0
+        time.sleep(10)
+
+        namespace_list_req = pb2.list_namespaces_req(subsystem=subsystem)
+        listA = json.loads(json_format.MessageToJson(
+            stubA.list_namespaces(namespace_list_req),
+            preserving_proto_field_name=True, including_default_value_fields=True))
+        assert listA["status"] == 0
+        assert len(listA["namespaces"]) == 2
+        nsidA1 = listA["namespaces"][0]["nsid"]
+        nsidA2 = listA["namespaces"][1]["nsid"]
+        bdevA1 = listA["namespaces"][0]["bdev_name"]
+        bdevA2 = listA["namespaces"][1]["bdev_name"]
+        uuidA1 = listA["namespaces"][0]["uuid"]
+        uuidA2 = listA["namespaces"][1]["uuid"]
+        imgA1 = listA["namespaces"][0]["rbd_image_name"]
+        imgA2 = listA["namespaces"][1]["rbd_image_name"]
+        time.sleep(10)
+        listB = json.loads(json_format.MessageToJson(
+            stubB.list_namespaces(namespace_list_req),
+            preserving_proto_field_name=True, including_default_value_fields=True))
+        assert listB["status"] == 0
+        assert len(listB["namespaces"]) == 2
+        nsidB1 = listB["namespaces"][0]["nsid"]
+        nsidB2 = listB["namespaces"][1]["nsid"]
+        bdevB1 = listB["namespaces"][0]["bdev_name"]
+        bdevB2 = listB["namespaces"][1]["bdev_name"]
+        uuidB1 = listB["namespaces"][0]["uuid"]
+        uuidB2 = listB["namespaces"][1]["uuid"]
+        imgB1 = listB["namespaces"][0]["rbd_image_name"]
+        imgB2 = listB["namespaces"][1]["rbd_image_name"]
+        if nsidA1 == nsidB1:
+            assert nsidA2 == nsidB2
+            assert bdevA1 == bdevB1
+            assert bdevA2 == bdevB2
+            assert uuidA1 == uuidB1
+            assert uuidA2 == uuidB2
+            assert imgA1 == imgB1
+            assert imgA2 == imgB2
+        elif nsidA1 == nsidB2:
+            assert nsidA2 == nsidB1
+            assert bdevA1 == bdevB2
+            assert bdevA2 == bdevB1
+            assert uuidA1 == uuidB2
+            assert uuidA2 == uuidB1
+            assert imgA1 == imgB2
+            assert imgA2 == imgB1
+        else:
+            assert False
+        gatewayB.__exit__(None, None, None)
+        gatewayB = GatewayServer(configB)
+        gatewayB.serve()
+        channelB = grpc.insecure_channel(f"{addr}:{portB}")
+        stubB = pb2_grpc.GatewayStub(channelB)
+        time.sleep(10)
+        listB = json.loads(json_format.MessageToJson(
+            stubB.list_namespaces(namespace_list_req),
+            preserving_proto_field_name=True, including_default_value_fields=True))
+        assert listB["status"] == 0
+        assert len(listB["namespaces"]) == 2
+        nsidB1 = listB["namespaces"][0]["nsid"]
+        nsidB2 = listB["namespaces"][1]["nsid"]
+        bdevB1 = listB["namespaces"][0]["bdev_name"]
+        bdevB2 = listB["namespaces"][1]["bdev_name"]
+        uuidB1 = listB["namespaces"][0]["uuid"]
+        uuidB2 = listB["namespaces"][1]["uuid"]
+        imgB1 = listB["namespaces"][0]["rbd_image_name"]
+        imgB2 = listB["namespaces"][1]["rbd_image_name"]
+        if nsidA1 == nsidB1:
+            assert nsidA2 == nsidB2
+            assert bdevA1 == bdevB1
+            assert bdevA2 == bdevB2
+            assert uuidA1 == uuidB1
+            assert uuidA2 == uuidB2
+            assert imgA1 == imgB1
+            assert imgA2 == imgB2
+        elif nsidA1 == nsidB2:
+            assert nsidA2 == nsidB1
+            assert bdevA1 == bdevB2
+            assert bdevA2 == bdevB1
+            assert uuidA1 == uuidB2
+            assert uuidA2 == uuidB1
+            assert imgA1 == imgB2
+            assert imgA2 == imgB1
+        else:
+            assert False