Enable multiarch

To prevent cluttering the namespace with lots of failing task pods.

.drone.yml

@@ -0,0 +1,2 @@
+kind: template
+load: docker-multiarch.yaml

Dockerfile

@@ -1,11 +1,18 @@
-FROM python:3.9-slim-buster
+FROM python:3.10-slim-buster as builder
+RUN apt-get update && apt-get install -y build-essential libbtrfsutil-dev
+RUN pip wheel -w /wheels "https://github.com/kdave/btrfs-progs/archive/refs/tags/v5.16.1.tar.gz#egg=btrfsutil&subdirectory=libbtrfsutil/python"
+
+FROM python:3.10-slim-buster
 
 WORKDIR /app/
 
 RUN apt-get update && \
-    apt-get install -y e2fsprogs btrfs-progs xfsprogs && \
+    apt-get install -y e2fsprogs btrfs-progs libbtrfsutil1 xfsprogs && \
     rm -rf /var/lib/apt/lists/*
 
+COPY --from=builder /wheels/ /wheels/
+RUN pip install /wheels/*
+
 ENV PIP_NO_CACHE_DIR 1
 ADD ./requirements.txt ./
 RUN pip install -r ./requirements.txt

README.md

@@ -48,7 +48,7 @@ Features
 - [ ] Online shrinking: If fs supports it (e.g. btrfs)
 - [ ] Offline expansion/shrinking
 - [ ] Ephemeral inline volume
-- [ ] Snapshots: If the fs supports it (e.g. btrfs)
+- [x] Filesystem-level snapshots: `btrfs` supported
 
 Motivation
 ---

bd2fs.py

@@ -1,7 +1,6 @@
 from pathlib import Path
 
 import grpc
-
 from csi import csi_pb2, csi_pb2_grpc
 from csi.csi_pb2 import (
     NodeStageVolumeRequest,
@@ -11,6 +10,8 @@ from csi.csi_pb2 import (
     NodeExpandVolumeRequest,
     CreateVolumeRequest,
 )
+from google.protobuf.timestamp_pb2 import Timestamp
+
 from declarative import (
     be_mounted,
     be_unmounted,
@@ -19,6 +20,8 @@ from declarative import (
     be_fs_expanded,
 )
 from fs_util import path_stats, mountpoint_to_dev
+from orchestrator.k8s import volume_to_node, run_on_node
+from remote import btrfs_create_snapshot, btrfs_delete_snapshot
 from util import log_grpc_request
 
 
@@ -30,7 +33,7 @@ def get_fs(request):
 
 
 class Bd2FsIdentityServicer(csi_pb2_grpc.IdentityServicer):
-    def __init__(self, bds):
+    def __init__(self, bds: csi_pb2_grpc.IdentityServicer):
         self.bds = bds
 
     @log_grpc_request
@@ -47,7 +50,7 @@ class Bd2FsIdentityServicer(csi_pb2_grpc.IdentityServicer):
 
 
 class Bd2FsNodeServicer(csi_pb2_grpc.NodeServicer):
-    def __init__(self, bds):
+    def __init__(self, bds: csi_pb2_grpc.NodeServicer):
         self.bds = bds
 
     # @log_grpc_request
@@ -167,7 +170,7 @@ class Bd2FsNodeServicer(csi_pb2_grpc.NodeServicer):
 
 
 class Bd2FsControllerServicer(csi_pb2_grpc.ControllerServicer):
-    def __init__(self, bds):
+    def __init__(self, bds: csi_pb2_grpc.ControllerServicer):
         self.bds = bds
 
     @log_grpc_request
@@ -219,3 +222,32 @@ class Bd2FsControllerServicer(csi_pb2_grpc.ControllerServicer):
         response = self.bds.ControllerExpandVolume(request, context)
         assert response.node_expansion_required
         return response
+
+    @log_grpc_request
+    def CreateSnapshot(self, request: csi_pb2.CreateSnapshotRequest, context):
+        volume_id = request.source_volume_id
+        name = request.name
+
+        snapshot_id, creation_time_ns = btrfs_create_snapshot(
+            volume_id=volume_id, name=name
+        )
+
+        nano = 10**9
+        return csi_pb2.CreateSnapshotResponse(
+            snapshot=csi_pb2.Snapshot(
+                size_bytes=0,
+                snapshot_id=snapshot_id,
+                source_volume_id=volume_id,
+                creation_time=Timestamp(
+                    seconds=creation_time_ns // nano, nanos=creation_time_ns % nano
+                ),
+                ready_to_use=True,
+            )
+        )
+
+    @log_grpc_request
+    def DeleteSnapshot(self, request: csi_pb2.DeleteSnapshotRequest, context):
+        snapshot_id = request.snapshot_id
+        volume_id, name = snapshot_id.rsplit("/", 1)
+        btrfs_delete_snapshot(volume_id=volume_id, name=name)
+        return csi_pb2.DeleteSnapshotResponse()

consts.py

@@ -1,8 +1,7 @@
 import os
 
 PROVISIONER_NAME = os.getenv("PROVISIONER_NAME", "rawfile.csi.openebs.io")
-PROVISIONER_VERSION = "0.7.0"
+PROVISIONER_VERSION = "0.8.0"
 DATA_DIR = "/data"
 CONFIG = {}
-RESOURCE_EXHAUSTED_EXIT_CODE = 101
 VOLUME_IN_USE_EXIT_CODE = 102

declarative.py

@@ -109,6 +109,6 @@ def be_fs_expanded(dev, path):
     elif fs == "btrfs":
         run(f"btrfs filesystem resize max {path}")
     elif fs == "xfs":
-        run(f"xfs_growfs -d {dev}")
+        run(f"xfs_growfs -d {path}")
     else:
         raise Exception(f"Unsupported fsType: {fs}")

deploy/charts/rawfile-csi/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 name: rawfile-csi
 description: RawFile Driver Container Storage Interface
 type: application
-version: 0.7.0
-appVersion: 0.7.0
+version: 0.8.0
+appVersion: 0.8.0

deploy/charts/rawfile-csi/templates/00-rbac.yaml

@@ -124,3 +124,34 @@ roleRef:
   kind: ClusterRole
   name: {{ include "rawfile-csi.fullname" . }}-resizer
   apiGroup: rbac.authorization.k8s.io
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: {{ include "rawfile-csi.fullname" . }}-snapshotter
+rules:
+  - apiGroups: [""]
+    resources: ["events"]
+    verbs: ["list", "watch", "create", "update", "patch"]
+  - apiGroups: ["snapshot.storage.k8s.io"]
+    resources: ["volumesnapshotclasses"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: ["snapshot.storage.k8s.io"]
+    resources: ["volumesnapshotcontents"]
+    verbs: ["create", "get", "list", "watch", "update", "delete", "patch"]
+  - apiGroups: ["snapshot.storage.k8s.io"]
+    resources: ["volumesnapshotcontents/status"]
+    verbs: ["update", "patch"]
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: {{ include "rawfile-csi.fullname" . }}-snapshotter
+subjects:
+  - kind: ServiceAccount
+    name: {{ include "rawfile-csi.fullname" . }}-driver
+    namespace: {{ .Release.Namespace }}
+roleRef:
+  kind: ClusterRole
+  name: {{ include "rawfile-csi.fullname" . }}-snapshotter
+  apiGroup: rbac.authorization.k8s.io

deploy/charts/rawfile-csi/templates/01-node-plugin.yaml

@@ -153,3 +153,20 @@ spec:
           volumeMounts:
             - name: socket-dir
               mountPath: /csi
+        - name: external-snapshotter
+          image: gcr.io/k8s-staging-sig-storage/csi-snapshotter:v5.0.1
+          imagePullPolicy: IfNotPresent
+          args:
+            - "--csi-address=$(ADDRESS)"
+            - "--node-deployment=true"
+            - "--extra-create-metadata=true"
+          env:
+            - name: ADDRESS
+              value: /csi/csi.sock
+            - name: NODE_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
+          volumeMounts:
+            - name: socket-dir
+              mountPath: /csi

deploy/charts/rawfile-csi/values.yaml

@@ -3,7 +3,7 @@ provisionerName: "rawfile.csi.openebs.io"
 defaults: &defaults
   image:
     repository: docker.io/openebs/rawfile-localpv
-    tag: 0.7.0
+    tag: 0.8.0
     pullPolicy: Always
   resources:
     limits:

orchestrator/k8s.py

@@ -66,9 +66,9 @@ def run_on_node(fn, node):
         return False
 
     wait_for(is_finished, "task to finish")
+    task_pod.delete()
     if task_pod.obj["status"]["phase"] != "Succeeded":
         exit_code = task_pod.obj["status"]["containerStatuses"][0]["state"][
             "terminated"
         ]["exitCode"]
         raise CalledProcessError(returncode=exit_code, cmd=f"Task: {name}")
-    task_pod.delete()

rawfile_servicer.py

@@ -8,7 +8,6 @@ import rawfile_util
 from consts import (
     PROVISIONER_VERSION,
     PROVISIONER_NAME,
-    RESOURCE_EXHAUSTED_EXIT_CODE,
     VOLUME_IN_USE_EXIT_CODE,
 )
 from csi import csi_pb2, csi_pb2_grpc
@@ -141,6 +140,7 @@ class RawFileControllerServicer(csi_pb2_grpc.ControllerServicer):
                 Cap(rpc=Cap.RPC(type=Cap.RPC.CREATE_DELETE_VOLUME)),
                 Cap(rpc=Cap.RPC(type=Cap.RPC.GET_CAPACITY)),
                 Cap(rpc=Cap.RPC(type=Cap.RPC.EXPAND_VOLUME)),
+                Cap(rpc=Cap.RPC(type=Cap.RPC.CREATE_DELETE_SNAPSHOT)),
             ]
         )
 
@@ -191,15 +191,7 @@ class RawFileControllerServicer(csi_pb2_grpc.ControllerServicer):
                 grpc.StatusCode.INVALID_ARGUMENT, "Topology key not found... why?"
             )
 
-        try:
-            init_rawfile(volume_id=request.name, size=size),
-        except CalledProcessError as exc:
-            if exc.returncode == RESOURCE_EXHAUSTED_EXIT_CODE:
-                context.abort(
-                    grpc.StatusCode.RESOURCE_EXHAUSTED, "Not enough disk space"
-                )
-            else:
-                raise exc
+        init_rawfile(volume_id=request.name, size=size),
 
         return csi_pb2.CreateVolumeResponse(
             volume=csi_pb2.Volume(
@@ -233,17 +225,9 @@ class RawFileControllerServicer(csi_pb2_grpc.ControllerServicer):
         node_name = volume_to_node(volume_id)
         size = request.capacity_range.required_bytes
 
-        try:
-            run_on_node(
-                expand_rawfile.as_cmd(volume_id=volume_id, size=size), node=node_name
-            )
-        except CalledProcessError as exc:
-            if exc.returncode == RESOURCE_EXHAUSTED_EXIT_CODE:
-                context.abort(
-                    grpc.StatusCode.RESOURCE_EXHAUSTED, "Not enough disk space"
-                )
-            else:
-                raise exc
+        run_on_node(
+            expand_rawfile.as_cmd(volume_id=volume_id, size=size), node=node_name
+        )
 
         return csi_pb2.ControllerExpandVolumeResponse(
             capacity_bytes=size,

rawfile_util.py

@@ -129,8 +129,4 @@ def get_volumes_stats() -> [dict]:
 
 
 def get_capacity():
-    disk_free_size = path_stats(DATA_DIR)["fs_avail"]
-    capacity = disk_free_size
-    for volume_stat in get_volumes_stats().values():
-        capacity -= volume_stat["total"] - volume_stat["used"]
-    return capacity
+    return path_stats(DATA_DIR)["fs_avail"]

remote.py

@@ -1,3 +1,5 @@
+from contextlib import contextmanager
+
 from util import remote_fn
 
 
@@ -32,10 +34,6 @@ def init_rawfile(volume_id, size):
     import rawfile_util
     from volume_schema import LATEST_SCHEMA_VERSION
     from util import run
-    from consts import RESOURCE_EXHAUSTED_EXIT_CODE
-
-    if rawfile_util.get_capacity() < size:
-        raise CalledProcessError(returncode=RESOURCE_EXHAUSTED_EXIT_CODE, cmd="")
 
     img_dir = rawfile_util.img_dir(volume_id)
     img_dir.mkdir(exist_ok=True)
@@ -67,17 +65,67 @@ def expand_rawfile(volume_id, size):
     import rawfile_util
 
     from util import run
-    from consts import RESOURCE_EXHAUSTED_EXIT_CODE
 
     img_file = rawfile_util.img_file(volume_id)
     size_inc = size - rawfile_util.metadata(volume_id)["size"]
     if size_inc <= 0:
         return
-    if rawfile_util.get_capacity() < size_inc:
-        exit(RESOURCE_EXHAUSTED_EXIT_CODE)
 
     rawfile_util.patch_metadata(
         volume_id,
         {"size": size},
     )
     run(f"truncate -s {size} {img_file}")
+
+
+@contextmanager
+def mount_root_subvol(volume_id):
+    import tempfile
+    import pathlib
+
+    import rawfile_util
+    from util import run
+
+    root_subvol = tempfile.mkdtemp(prefix="rawfile-")
+
+    img_file = rawfile_util.img_file(volume_id)
+    loop_dev = rawfile_util.attach_loop(img_file)
+
+    run(f"mount -t btrfs -o subvolid=0 {loop_dev} {root_subvol}")
+    try:
+        yield root_subvol
+    finally:
+        run(f"umount {root_subvol}")
+        pathlib.Path(root_subvol).rmdir()
+
+
+def btrfs_delete_snapshot(volume_id, name):
+    import btrfsutil
+
+    with mount_root_subvol(volume_id) as root_subvol:
+        snapshots_dir = f"{root_subvol}/.snapshots"
+        snapshot_path = f"{snapshots_dir}/{name}"
+        btrfsutil.delete_subvolume(snapshot_path)
+
+
+def btrfs_create_snapshot(volume_id, name):
+    import btrfsutil
+    import time
+    import pathlib
+
+    # TODO: check fstype
+
+    with mount_root_subvol(volume_id) as root_subvol:
+        default_subvol_id = btrfsutil.get_default_subvolume(root_subvol)
+        default_subvol = btrfsutil.subvolume_path(root_subvol, default_subvol_id)
+        default_subvol = f"{root_subvol}/{default_subvol}"
+
+        snapshots_dir = f"{root_subvol}/.snapshots"
+        pathlib.Path(snapshots_dir).mkdir(parents=True, exist_ok=True)
+
+        snapshot_subvol = f"{snapshots_dir}/{name}"
+        btrfsutil.create_snapshot(default_subvol, snapshot_subvol, read_only=True)
+
+    snapshot_id = f"{volume_id}/{name}"
+    creation_time_ns = time.time_ns()
+    return snapshot_id, creation_time_ns

requirements.in

@@ -5,3 +5,4 @@ pyyaml
 pykube-ng
 munch
 prometheus_client
+#https://github.com/kdave/btrfs-progs/archive/refs/tags/v5.16.1.tar.gz#egg=btrfsutil&subdirectory=libbtrfsutil/python

requirements.txt

@@ -1,43 +1,42 @@
 #
-# This file is autogenerated by pip-compile with python 3.9
+# This file is autogenerated by pip-compile with python 3.10
 # To update, run:
 #
 #    pip-compile
 #
-certifi==2021.5.30
+certifi==2021.10.8
     # via requests
-chardet==4.0.0
+charset-normalizer==2.0.12
     # via requests
-click==8.0.1
+click==8.0.3
     # via -r requirements.in
-grpcio==1.38.1
+grpcio==1.43.0
     # via
     #   -r requirements.in
     #   grpcio-tools
-grpcio-tools==1.38.1
+grpcio-tools==1.43.0
     # via -r requirements.in
-idna==2.10
+idna==3.3
     # via requests
 munch==2.5.0
     # via -r requirements.in
-prometheus-client==0.11.0
+prometheus-client==0.13.1
     # via -r requirements.in
-protobuf==3.17.3
+protobuf==3.19.4
     # via grpcio-tools
-pykube-ng==21.3.0
+pykube-ng==22.1.0
     # via -r requirements.in
-pyyaml==5.4.1
+pyyaml==6.0
     # via
     #   -r requirements.in
     #   pykube-ng
-requests==2.25.1
+requests==2.27.1
     # via pykube-ng
 six==1.16.0
     # via
     #   grpcio
     #   munch
-    #   protobuf
-urllib3==1.26.6
+urllib3==1.26.8
     # via requests
 
 # The following packages are considered to be unsafe in a requirements file: