Merge pull request #247 from mhkarimi1383/metadata-dir-seperation

♻️ Separating metadata dir and data dir

Author: mergify[bot]

deploy/helm/rawfile-localpv/README.md

@@ -74,6 +74,7 @@ Please follow the [install guide](https://github.com/openebs/rawfile-localpv/tre
 | node.internalGRPC.port | int | `4500` | Port Number used for internal communication gRPC server |
 | node.internalGRPC.workers | int | `10` | gRPC worker count used for internal communication |
 | node.kubeletPath | string | `"/var/lib/kubelet"` | Kubelet path (Set to `/var/lib/k0s/kubelet` for k0s) |
+| node.metadataDirPath | string | `"/var/local/openebs/rawfile/meta"` | Metadata dir path for rawfile volumes metadata and tasks store file |
 | node.metrics.enabled | bool | `false` |  |
 | node.priorityClassName | string | `"system-node-critical"` | priorityClassName for node component since this part is critical for node `system-node-critical` is default |
 | node.resources | object | `{}` | Sets compute resources for node component |

deploy/helm/rawfile-localpv/templates/controller/statefulset.yaml

@@ -32,8 +32,6 @@ spec:
               value: "{{ .Values.provisionerName }}"
             - name: CSI_DRIVER__ENDPOINT
               value: unix:///csi/csi.sock
-            - name: NODE_DATADIR
-              value: "{{ .Values.node.dataDirPath }}"
             - name: CSI_DRIVER__NODE_ID
               valueFrom:
                 fieldRef:

deploy/helm/rawfile-localpv/templates/node-plugin/daemonset.yaml

@@ -35,6 +35,10 @@ spec:
           hostPath:
             path: {{ .Values.node.dataDirPath }}
             type: DirectoryOrCreate
+        - name: metadata-dir
+          hostPath:
+            path: {{ .Values.node.metadataDirPath }}
+            type: DirectoryOrCreate
         - name: device
           hostPath:
             path: /dev
@@ -48,8 +52,6 @@ spec:
           env:
             - name: PROVISIONER_NAME
               value: "{{ .Values.provisionerName }}"
-            - name: NODE_DATADIR
-              value: "{{ .Values.node.dataDirPath }}"
             - name: CSI_DRIVER__ENDPOINT
               value: unix:///csi/csi.sock
             - name: CSI_DRIVER__NODE_ID
@@ -61,6 +63,8 @@ spec:
               value: {{ .Values.metrics.enabled | toString | quote }}
             - name: CSI_DRIVER__METRICS_PORT
               value: {{ .Values.metrics.port | toString | quote }}
+            - name: CSI_DRIVER__METADATA_DIR
+              value: {{ .Values.node.metadataDirPath }}
             - name: NAMESPACE
               value: {{ .Release.Namespace }}
             - name: LOG_LEVEL
@@ -120,6 +124,8 @@ spec:
               mountPropagation: "Bidirectional"
             - name: data-dir
               mountPath: /data
+            - name: metadata-dir
+              mountPath: {{ .Values.node.metadataDirPath }}
             - name: device
               mountPath: /dev
           resources:

deploy/helm/rawfile-localpv/values.yaml

@@ -129,6 +129,9 @@ node:
   # -- Data dir path for provisioner to be used by provisioner
   dataDirPath: /var/csi/rawfile
 
+  # -- Metadata dir path for rawfile volumes metadata and tasks store file
+  metadataDirPath: /var/local/openebs/rawfile/meta
+
   # -- Kubelet path (Set to `/var/lib/k0s/kubelet` for k0s)
   kubeletPath: /var/lib/kubelet
 

rawfile/config/model.py

@@ -12,6 +12,7 @@
     AnyUrl,
     BaseModel,
     ByteSize,
+    DirectoryPath,
     StringConstraints,
     Field,
     model_validator,
@@ -36,6 +37,10 @@ class CSIDriverCmd(BaseModel):
         description="Listen ip for gRPC server (used for internal communication only)",
         default=None,
     )
+    metadata_dir: DirectoryPath | None = Field(
+        description="Directory to store Metadata files, required and should point to an existing path when running node plugin",
+        default=None,
+    )
     internal_port: int | None = Field(
         description="Listen port for gRPC server (used for internal communication only)",
         default=None,
@@ -69,13 +74,16 @@ class CSIDriverCmd(BaseModel):
     )
 
     @model_validator(mode="after")
-    def validate_internal_endpoint(
+    def validate_node_plugin(
         self,
     ):
-        if self.plugin_type == "node" and not self.internal_ip:
-            raise ValueError(
-                "Internal Communication IP/PORT is required on node plugin"
-            )
+        if self.plugin_type == "node":
+            if not self.internal_ip:
+                raise ValueError(
+                    "Internal Communication IP/PORT is required on node plugin"
+                )
+            if not self.metadata_dir:
+                raise ValueError("Metadata Dir is required when running node plugin")
         return self
 
 
@@ -123,15 +131,6 @@ def settings_customise_sources(
     namespace: str = Field(
         description="K8s Namespace of the driver",
     )
-    node_datadir: str = Field(
-        description="""
-                    Data Directory path of the driver,
-                    where raw files, their matadata and their lock files are getting stored
-
-                    NOTE that this path is used only for task pods, For CSI driver itself `/data` is used
-                    meaning that this path has to be mounted to `/data` in the CSI driver pod
-                    """,
-    )
     log_level: Annotated[
         Literal["TRACE", "DEBUG", "INFO", "SUCCESS", "WARNING", "ERROR", "CRITICAL"],
         StringConstraints(strip_whitespace=True, to_upper=True),

rawfile/internal_svc.py

@@ -9,10 +9,10 @@
     snapshots_dir,
     patch_metadata,
 )
-from utils.remote import is_attached
 from utils.lock import VolLock
 from typing import Final
 from utils.storage_pool import get_capacity
+from utils.volume_manager import manager as volume_manager
 import grpc
 
 SIGNATURE_METADATA: Final[str] = "x-signature"
@@ -54,13 +54,13 @@ def ExpandRawFile(
             size_inc = request.new_size - metadata(request.volume_id)["size"]
             if size_inc <= 0:
                 return internal_pb2.ExpandRawFileResponse(
-                    is_attached=is_attached(request.volume_id),
+                    is_attached=volume_manager.is_attached(request.volume_id),
                     status=internal_pb2.ExpandRawFileStatus.OK,
                 )
 
             if get_capacity() < size_inc:
                 return internal_pb2.ExpandRawFileResponse(
-                    is_attached=is_attached(request.volume_id),
+                    is_attached=volume_manager.is_attached(request.volume_id),
                     status=internal_pb2.ExpandRawFileStatus.RESOURCE_EXHAUSTED,
                 )
             if metadata(request.volume_id).get("thin_provision", False):
@@ -72,7 +72,7 @@ def ExpandRawFile(
                 {"size": request.new_size},
             )
             return internal_pb2.ExpandRawFileResponse(
-                is_attached=is_attached(request.volume_id),
+                is_attached=volume_manager.is_attached(request.volume_id),
                 status=internal_pb2.ExpandRawFileStatus.OK,
             )
 

rawfile/rawfile.py

@@ -22,17 +22,32 @@
 import os
 
 
-def node_driver_preflight_checks():
+def __create_and_check_directory(dir: Path):
+    if not dir.exists():
+        logger.info("Creating directory", path=str(dir))
+        dir.mkdir(parents=True, exist_ok=True)
+    if not dir.is_dir():
+        raise RuntimeError(f"{dir} is not a directory")
+    if not os.access(dir, os.W_OK | os.R_OK | os.X_OK):
+        raise RuntimeError(f"{dir} is not accessible")
+    dir.chmod(consts.D_PERMS)
+
+
+def node_driver_preflight_checks(task_manager: task_manager.TaskManager):
+    if not config.csi_driver:
+        raise RuntimeError("CSI Driver configuration is missing")
+    if not config.csi_driver.metadata_dir:
+        raise RuntimeError("Metadata directory is not set for node plugin")
     data_dir = Path(consts.DATA_DIR)
-    if not data_dir.exists():
-        logger.info("Creating data directory", path=str(data_dir))
-        data_dir.mkdir(parents=True, exist_ok=True)
-    if not data_dir.is_dir():
-        raise RuntimeError(f"{data_dir} is not a directory")
-    if not os.access(data_dir, os.W_OK | os.R_OK | os.X_OK):
-        raise RuntimeError(f"{data_dir} is not accessible")
-    data_dir.chmod(consts.D_PERMS)
+    dirs = (
+        config.csi_driver.metadata_dir,
+        data_dir,
+    )
+    for dir in dirs:
+        __create_and_check_directory(dir)
+    volume_manager.migrate_metadata_dir()
     volume_manager.migrate_all_volume_schemas()
+    task_manager.migrate_tasks_file_path()
     consts.COW_SUPPORTED = is_cow_supported(data_dir)
 
 
@@ -54,7 +69,10 @@ def csi_driver(driver_config: CSIDriverCmd):
         futures.ThreadPoolExecutor(max_workers=int(driver_config.grpc_workers))
     )
     bg_task_executor = ThreadPoolExecutor(max_workers=5)
-    _task_manager = task_manager.TaskManager(bg_task_executor)
+    _task_manager = task_manager.TaskManager(
+        bg_task_executor,
+        tasks_store_path=Path(f"{driver_config.metadata_dir}/tasks.json"),
+    )
     internal_server = None
     csi_pb2_grpc.add_IdentityServicer_to_server(
         bd2fs.Bd2FsIdentityServicer(
@@ -63,7 +81,7 @@ def csi_driver(driver_config: CSIDriverCmd):
         server,
     )
     if driver_config.plugin_type == "node":
-        node_driver_preflight_checks()
+        node_driver_preflight_checks(task_manager=_task_manager)
         csi_pb2_grpc.add_NodeServicer_to_server(
             bd2fs.Bd2FsNodeServicer(
                 rawfile_servicer.RawFileNodeServicer(node_name=driver_config.nodeid),

rawfile/utils/rawfile.py

@@ -4,8 +4,8 @@
 from typing import Any
 
 from utils.logs import logger
-
-from consts import D_PERMS, DATA_DIR, F_PERMS, OWNER_UMASK
+from config import config
+from consts import D_PERMS, F_PERMS, OWNER_UMASK
 import os
 from enum import Enum
 from utils.commands import run
@@ -17,20 +17,20 @@ class AccessType(Enum):
     block = 2
 
 
-def img_dir(volume_id):
-    return Path(f"{DATA_DIR}/{volume_id}")
+def meta_dir(volume_id):
+    return Path(f"{config.csi_driver.metadata_dir}/{volume_id}")
 
 
 def meta_file(volume_id):
-    return Path(f"{img_dir(volume_id)}/disk.meta")
+    return Path(f"{meta_dir(volume_id)}/disk.meta")
 
 
 def meta_file_tmp(volume_id):
-    return Path(f"{img_dir(volume_id)}/disk.meta.tmp")
+    return Path(f"{meta_dir(volume_id)}/disk.meta.tmp")
 
 
 def lock_file(volume_id):
-    return Path(f"{img_dir(volume_id)}/disk.lock")
+    return Path(f"{meta_dir(volume_id)}/disk.lock")
 
 
 def metadata(volume_id) -> dict[str, Any]:
@@ -49,9 +49,11 @@ def img_file(volume_id):
 
 
 def snapshots_dir(volume_id: str, temporary: bool = False):
+    from utils.volume_manager import manager as volume_manager
+
     if temporary:
-        return Path(f"{img_dir(volume_id)}/snapshots/temp")
-    return Path(f"{img_dir(volume_id)}/snapshots")
+        return Path(f"{volume_manager._get_volume_path(volume_id)}/snapshots/temp")
+    return Path(f"{volume_manager._get_volume_path(volume_id)}/snapshots")
 
 
 def img_size(volume_id) -> int:
@@ -89,11 +91,18 @@ def update_metadata(volume_id: str, obj: dict) -> dict:
 
 
 def update_permissions(volume_id: str) -> None:
-    _img_dir = img_dir(volume_id)
+    from utils.volume_manager import manager as volume_manager
+    from itertools import chain
+
+    _img_dir = volume_manager._get_volume_path(volume_id)
     if not _img_dir.exists():
         return
     _img_dir.chmod(D_PERMS)
-    for each in _img_dir.glob("**/*"):
+    ## set permissions recursively on all files and dirs under the volume path
+    # we go 3 levels deep to cover most cases (img file, snapshots, temp snapshots, etc.)
+    for each in chain(
+        _img_dir.glob("**/*"), _img_dir.glob("**/**/*"), _img_dir.glob("**/**/**/*")
+    ):
         each.chmod(F_PERMS)
 
 

rawfile/utils/remote.py

@@ -5,13 +5,3 @@
 def get_capacity():
     cap = utils.storage_pool.get_capacity()
     return max(0, cap)
-
-
-def is_attached(volume_id):
-    img_dir = utils.rawfile.img_dir(volume_id)
-    if not img_dir.exists():
-        return False
-
-    img_file = utils.rawfile.img_file(volume_id)
-    loops = utils.rawfile.attached_loops(img_file.as_posix())
-    return len(loops) > 0

rawfile/utils/task_manager.py

@@ -1,5 +1,6 @@
 from concurrent.futures import Executor, Future
 from enum import StrEnum
+import shutil
 from typing import Callable, TypedDict
 import time
 import consts
@@ -45,13 +46,15 @@ class TaskInfo(TypedDict):
 
 
 class TaskManager:
-    def __init__(self, executor: Executor, retry_interval=5, max_retry=5):
+    def __init__(
+        self, executor: Executor, tasks_store_path: Path, retry_interval=5, max_retry=5
+    ):
         self._executor = executor
         self._tasks: dict = {}
         self._shutting_down = False
         self._retry_interval = retry_interval
         self._max_retry = max_retry
-        self._tasks_store_path = Path(f"{consts.DATA_DIR}/tasks.json")
+        self._tasks_store_path = tasks_store_path
         self._tasks_store_path.parent.mkdir(exist_ok=True, parents=True)
         self._tasks_store_path.touch(exist_ok=True)
         self._lock = Lock()
@@ -94,7 +97,7 @@ def get_tasks(
         self, state: TaskState | None = None, retriable: bool | None = None
     ) -> dict[str, TaskInfo]:
         with self._lock:
-            with open(f"{consts.DATA_DIR}/tasks.json") as tasks_file:
+            with open(self._tasks_store_path) as tasks_file:
                 data = json.loads(tasks_file.read() or "{}")
                 if retriable is not None:
                     if retriable:
@@ -198,3 +201,9 @@ def shutdown(self, timeout: int):
                 info["state"] = TaskState.FAILED
                 self.save_task(task_id, info)
         self._executor.shutdown(wait=False, cancel_futures=True)
+
+    def migrate_tasks_file_path(self):
+        src = Path(f"{consts.DATA_DIR}/tasks.json")
+        dst = self._tasks_store_path
+        if src.exists():
+            shutil.move(src, dst)