Lock OMAP file before changing it to avoid corruption in multi-gatewa…

…y environemnt.

Fixes #56

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

.env

@@ -8,6 +8,8 @@ QUAY_CEPH="${CONTAINER_REGISTRY}/vstart-cluster"
 QUAY_NVMEOF="${CONTAINER_REGISTRY}/nvmeof"
 QUAY_NVMEOFCLI="${CONTAINER_REGISTRY}/nvmeof-cli"
 MAINTAINER="Ceph Developers <[email protected]>"
+COMPOSE_PROJECT_NAME="ceph-nvmeof"
+NVMEOF_CONTAINER_NAME="${COMPOSE_PROJECT_NAME}-nvmeof-1"
 
 # Performance
 NVMEOF_NOFILE=20480 # Max number of open files (depends on number of hosts connected)

.github/workflows/build-container.yml

@@ -197,7 +197,7 @@ jobs:
 
       - name: Test
         run: |
-          make demo OPTS=-T
+          make demo OPTS=-T NVMEOF_CONTAINER_NAME="ceph-nvmeof_nvmeof_1"
 
       - name: Get subsystems
         run: |

ceph-nvmeof.conf

@@ -18,6 +18,10 @@ state_update_interval_sec = 5
 #min_controller_id = 1
 #max_controller_id = 65519
 enable_discovery_controller = false
+#omap_file_lock_duration = 60
+#omap_file_lock_retries = 15
+#omap_file_lock_retry_sleep_interval = 5
+#omap_file_update_retries = 10
 
 [discovery]
 addr = 0.0.0.0

control/server.py

@@ -100,7 +100,7 @@ def serve(self):
         local_state = LocalGatewayState()
         gateway_state = GatewayStateHandler(self.config, local_state,
                                             omap_state, self.gateway_rpc_caller)
-        self.gateway_rpc = GatewayService(self.config, gateway_state,
+        self.gateway_rpc = GatewayService(self.config, gateway_state, omap_state,
                                           self.spdk_rpc_client)
         self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
         pb2_grpc.add_GatewayServicer_to_server(self.gateway_rpc, self.server)

control/state.py

@@ -11,6 +11,7 @@
 import threading
 import rados
 import logging
+import errno
 from typing import Dict
 from collections import defaultdict
 from abc import ABC, abstractmethod
@@ -162,6 +163,8 @@ class OmapGatewayState(GatewayState):
     """
 
     OMAP_VERSION_KEY = "omap_version"
+    OMAP_FILE_LOCK_NAME = "omap_file_lock"
+    OMAP_FILE_LOCK_COOKIE = "omap_file_cookie"
 
     def __init__(self, config):
         self.config = config
@@ -173,6 +176,8 @@ def __init__(self, config):
         ceph_pool = self.config.get("ceph", "pool")
         ceph_conf = self.config.get("ceph", "config_file")
         rados_id = self.config.get_with_default("ceph", "id", "")
+        self.omap_file_lock_retries = self.config.getint_with_default("gateway", "omap_file_lock_retries", 15)
+        self.omap_file_lock_retry_sleep_interval = self.config.getint_with_default("gateway", "omap_file_lock_retry_sleep_interval", 5)
 
         try:
             conn = rados.Rados(conffile=ceph_conf, rados_id=rados_id)
@@ -222,6 +227,49 @@ def get_omap_version(self) -> int:
                 f" invalid number of values ({value_list}).")
             raise
 
+    def lock_omap(self, duration):
+        got_lock = False
+
+        for i in range(1, self.omap_file_lock_retries):
+            try:
+                with rados.WriteOpCtx() as write_op:
+                    self.ioctx.lock_exclusive(self.omap_name, self.OMAP_FILE_LOCK_NAME, self.OMAP_FILE_LOCK_COOKIE, "OMAP file changes lock", duration, 0)
+                    got_lock = True
+                    break
+            except rados.ObjectExists as ex:
+                self.logger.info(f"We already locked the OMAP file")
+                got_lock = True
+                break
+            except rados.ObjectBusy as ex:
+                self.logger.warning(f"Someone else locked the OMAP file, will try again in {self.omap_file_lock_retry_sleep_interval} seconds")
+                time.sleep(self.omap_file_lock_retry_sleep_interval)
+            except Exception as ex:
+                self.logger.error(f"Unable to lock OMAP file: {ex}. Exiting!")
+                raise
+
+        if not got_lock:
+            self.logger.error(f"Unable to lock OMAP file after {self.omap_file_lock_retries} retries. Exiting!")
+            raise Exception("Unable to lock OMAP file")
+
+        omap_version = self.get_omap_version()
+
+        if omap_version > self.version:
+            self.logger.warning(f"Local version {self.version} differs from OMAP file version {omap_version}, need to read the OMAP file")
+            self.unlock_omap()
+            raise OSError(errno.EAGAIN, "Unable to lock OMAP file, file not current", self.omap_name)
+
+        return True
+
+    def unlock_omap(self):
+        try:
+            with rados.WriteOpCtx() as write_op:
+                self.ioctx.unlock(self.omap_name, self.OMAP_FILE_LOCK_NAME, self.OMAP_FILE_LOCK_COOKIE)
+        except rados.ObjectNotFound as ex:
+            self.logger.warning(f"No such lock, the lock duration might have passed")
+        except Exception as ex:
+            self.logger.error(f"Unable to unlock OMAP file: {ex}.")
+            pass
+
     def get_state(self) -> Dict[str, str]:
         """Returns dict of all OMAP keys and values."""
         with rados.ReadOpCtx() as read_op:

mk/demo.mk

@@ -7,7 +7,7 @@ rbd: CMD = bash -c "rbd -p $(RBD_POOL) info $(RBD_IMAGE_NAME) || rbd -p $(RBD_PO
 
 # demo
 # the fist gateway in docker enviroment, hostname defaults to container id
-demo: export NVMEOF_HOSTNAME != docker ps -q -f name=ceph-nvmeof_nvmeof_1
+demo: export NVMEOF_HOSTNAME != docker ps -q -f name=$(NVMEOF_CONTAINER_NAME)
 demo: rbd ## Expose RBD_IMAGE_NAME as NVMe-oF target
 	$(NVMEOF_CLI) create_bdev --pool $(RBD_POOL) --image $(RBD_IMAGE_NAME) --bdev $(BDEV_NAME)
 	$(NVMEOF_CLI) create_subsystem --subnqn $(NQN)