From: Jason Dillaman <dillaman@redhat.com>
Date: Fri, 2 Aug 2019 15:57:29 +0000 (-0400)
Subject: pybind/mgr/rbd_support: use image ids to detect duplicate tasks
X-Git-Tag: v14.2.3~13^2~1
X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=3ce69405ccb4e868b05a9911e7f5885978db937e;p=ceph.git

pybind/mgr/rbd_support: use image ids to detect duplicate tasks

This helps to to avoid the case where new tasks were not being scheduled
when an image name was re-used after having a task created under the
same name.

Fixes: https://tracker.ceph.com/issues/41032
Signed-off-by: Jason Dillaman <dillaman@redhat.com>
(cherry picked from commit 7adb17f83106b2cba87cf343c23fd6d2d26ae0b3)
---

diff --git a/qa/workunits/rbd/test_rbd_tasks.sh b/qa/workunits/rbd/test_rbd_tasks.sh
index 77617a33cbd85..b9663e601247b 100755
--- a/qa/workunits/rbd/test_rbd_tasks.sh
+++ b/qa/workunits/rbd/test_rbd_tasks.sh
@@ -205,6 +205,26 @@ test_duplicate_task() {
   ceph rbd task cancel ${TASK_ID_1}
 }
 
+test_duplicate_name() {
+  echo "test_duplicate_name"
+
+  local IMAGE=`uuidgen`
+  rbd create --size 1G --image-shared ${POOL}/${IMAGE}
+  local TASK_ID_1=`ceph rbd task add remove ${POOL}/${IMAGE} | jq --raw-output ".id"`
+
+  wait_for task_dne ${TASK_ID_1}
+
+  rbd create --size 1G --image-shared ${POOL}/${IMAGE}
+  local TASK_ID_2=`ceph rbd task add remove ${POOL}/${IMAGE} | jq --raw-output ".id"`
+
+  [[ "${TASK_ID_1}" != "${TASK_ID_2}" ]]
+  wait_for task_dne ${TASK_ID_2}
+
+  local TASK_ID_3=`ceph rbd task add remove ${POOL}/${IMAGE} | jq --raw-output ".id"`
+
+  [[ "${TASK_ID_2}" == "${TASK_ID_3}" ]]
+}
+
 test_progress() {
   echo "test_progress"
 
@@ -250,6 +270,7 @@ test_migration_abort
 test_list
 test_cancel
 test_duplicate_task
+test_duplicate_name
 test_progress
 
 echo OK
diff --git a/src/pybind/mgr/rbd_support/module.py b/src/pybind/mgr/rbd_support/module.py
index 65b41faabc3e8..85752cf452f8f 100644
--- a/src/pybind/mgr/rbd_support/module.py
+++ b/src/pybind/mgr/rbd_support/module.py
@@ -725,17 +725,26 @@ class TaskHandler:
         self.tasks_by_sequence[task.sequence] = task
         self.tasks_by_id[task.task_id] = task
 
+    def task_refs_match(self, task_refs, refs):
+        if TASK_REF_IMAGE_ID not in refs and TASK_REF_IMAGE_ID in task_refs:
+            task_refs = task_refs.copy()
+            del task_refs[TASK_REF_IMAGE_ID]
+
+        self.log.debug("task_refs_match: ref1={}, ref2={}".format(task_refs, refs))
+        return task_refs == refs
+
     def find_task(self, refs):
         self.log.debug("find_task: refs={}".format(refs))
 
         # search for dups and return the original
-        for task_id, task in self.tasks_by_id.items():
-            if task.refs == refs:
+        for task_id in reversed(sorted(self.tasks_by_id.keys())):
+            task = self.tasks_by_id[task_id]
+            if self.task_refs_match(task.refs, refs):
                 return task
 
         # search for a completed task (message replay)
-        for task in self.completed_tasks:
-            if task.refs == refs:
+        for task in reversed(self.completed_tasks):
+            if self.task_refs_match(task.refs, refs):
                 return task
 
     def add_task(self, ioctx, message, refs):
@@ -782,6 +791,7 @@ class TaskHandler:
                 # keep a record of the last N tasks to help avoid command replay
                 # races
                 if not task.failed and not task.canceled:
+                    self.log.debug("remove_task: moving to completed tasks")
                     self.completed_tasks.append(task)
                     self.completed_tasks = self.completed_tasks[-MAX_COMPLETED_TASKS:]
 
@@ -974,25 +984,35 @@ class TaskHandler:
                 TASK_REF_POOL_NAME: image_spec[0],
                 TASK_REF_POOL_NAMESPACE: image_spec[1],
                 TASK_REF_IMAGE_NAME: image_spec[2]}
-        task = self.find_task(refs)
-        if task:
-            return 0, task.to_json(), ''
 
         with self.open_ioctx(image_spec) as ioctx:
-            with rbd.Image(ioctx, image_spec[2]) as image:
-                try:
-                    image.parent_id()
-                except rbd.ImageNotFound:
-                    raise rbd.ImageNotFound("Image {} does not have a parent".format(
-                        self.format_image_spec(image_spec)), errno=errno.ENOENT)
+            try:
+                with rbd.Image(ioctx, image_spec[2]) as image:
+                    refs[TASK_REF_IMAGE_ID] = image.id()
+
+                    try:
+                        parent_image_id = image.parent_id()
+                    except rbd.ImageNotFound:
+                        parent_image_id = None
+
+            except rbd.ImageNotFound:
+                pass
+
+            task = self.find_task(refs)
+            if task:
+                return 0, task.to_json(), ''
+
+            if TASK_REF_IMAGE_ID not in refs:
+                raise rbd.ImageNotFound("Image {} does not exist".format(
+                    self.format_image_spec(image_spec)), errno=errno.ENOENT)
+            if not parent_image_id:
+                raise rbd.ImageNotFound("Image {} does not have a parent".format(
+                    self.format_image_spec(image_spec)), errno=errno.ENOENT)
 
             return 0, self.add_task(ioctx,
                                     "Flattening image {}".format(
                                         self.format_image_spec(image_spec)),
-                                    {TASK_REF_ACTION: TASK_REF_ACTION_FLATTEN,
-                                     TASK_REF_POOL_NAME: image_spec[0],
-                                     TASK_REF_POOL_NAMESPACE: image_spec[1],
-                                     TASK_REF_IMAGE_NAME: image_spec[2]}), ""
+                                    refs), ""
 
     def queue_remove(self, image_spec):
         image_spec = self.extract_image_spec(image_spec)
@@ -1002,15 +1022,26 @@ class TaskHandler:
                 TASK_REF_POOL_NAME: image_spec[0],
                 TASK_REF_POOL_NAMESPACE: image_spec[1],
                 TASK_REF_IMAGE_NAME: image_spec[2]}
-        task = self.find_task(refs)
-        if task:
-            return 0, task.to_json(), ''
 
         with self.open_ioctx(image_spec) as ioctx:
-            with rbd.Image(ioctx, image_spec[2]) as image:
-                if list(image.list_snaps()):
-                    raise rbd.ImageBusy("Image {} has snapshots".format(
-                        self.format_image_spec(image_spec)), errno=errno.EBUSY)
+            try:
+                with rbd.Image(ioctx, image_spec[2]) as image:
+                    refs[TASK_REF_IMAGE_ID] = image.id()
+                    snaps = list(image.list_snaps())
+
+            except rbd.ImageNotFound:
+                pass
+
+            task = self.find_task(refs)
+            if task:
+                return 0, task.to_json(), ''
+
+            if TASK_REF_IMAGE_ID not in refs:
+                raise rbd.ImageNotFound("Image {} does not exist".format(
+                    self.format_image_spec(image_spec)), errno=errno.ENOENT)
+            if snaps:
+                raise rbd.ImageBusy("Image {} has snapshots".format(
+                    self.format_image_spec(image_spec)), errno=errno.EBUSY)
 
             return 0, self.add_task(ioctx,
                                     "Removing image {}".format(
@@ -1038,12 +1069,16 @@ class TaskHandler:
                                         self.format_image_spec(image_id_spec)),
                                     refs), ''
 
-    def validate_image_migrating(self, ioctx, image_spec):
-        with rbd.Image(ioctx, image_spec[2]) as image:
-            features = image.features()
-            if features & rbd.RBD_FEATURE_MIGRATING == 0:
-                raise rbd.InvalidArgument("Image {} is not migrating".format(
-                    self.format_image_spec(image_spec)), errno=errno.EINVAL)
+    def get_migration_status(self, ioctx, image_spec):
+        try:
+            return rbd.RBD().migration_status(ioctx, image_spec[2])
+        except (rbd.InvalidArgument, rbd.ImageNotFound):
+            return None
+
+    def validate_image_migrating(self, migration_status):
+        if not migration_status:
+            raise rbd.InvalidArgument("Image {} is not migrating".format(
+                self.format_image_spec(image_spec)), errno=errno.EINVAL)
 
     def resolve_pool_name(self, pool_id):
         osd_map = self.module.get('osd_map')
@@ -1060,14 +1095,17 @@ class TaskHandler:
                 TASK_REF_POOL_NAME: image_spec[0],
                 TASK_REF_POOL_NAMESPACE: image_spec[1],
                 TASK_REF_IMAGE_NAME: image_spec[2]}
-        task = self.find_task(refs)
-        if task:
-            return 0, task.to_json(), ''
 
         with self.open_ioctx(image_spec) as ioctx:
-            self.validate_image_migrating(ioctx, image_spec)
+            status = self.get_migration_status(ioctx, image_spec)
+            if status:
+                refs[TASK_REF_IMAGE_ID] = status['dest_image_id']
+
+            task = self.find_task(refs)
+            if task:
+                return 0, task.to_json(), ''
 
-            status = rbd.RBD().migration_status(ioctx, image_spec[2])
+            self.validate_image_migrating(status)
             if status['state'] not in [rbd.RBD_IMAGE_MIGRATION_STATE_PREPARED,
                                        rbd.RBD_IMAGE_MIGRATION_STATE_EXECUTING]:
                 raise rbd.InvalidArgument("Image {} is not in ready state".format(
@@ -1093,14 +1131,17 @@ class TaskHandler:
                 TASK_REF_POOL_NAME: image_spec[0],
                 TASK_REF_POOL_NAMESPACE: image_spec[1],
                 TASK_REF_IMAGE_NAME: image_spec[2]}
-        task = self.find_task(refs)
-        if task:
-            return 0, task.to_json(), ''
 
         with self.open_ioctx(image_spec) as ioctx:
-            self.validate_image_migrating(ioctx, image_spec)
+            status = self.get_migration_status(ioctx, image_spec)
+            if status:
+                refs[TASK_REF_IMAGE_ID] = status['dest_image_id']
+
+            task = self.find_task(refs)
+            if task:
+                return 0, task.to_json(), ''
 
-            status = rbd.RBD().migration_status(ioctx, image_spec[2])
+            self.validate_image_migrating(status)
             if status['state'] != rbd.RBD_IMAGE_MIGRATION_STATE_EXECUTED:
                 raise rbd.InvalidArgument("Image {} has not completed migration".format(
                     self.format_image_spec(image_spec)), errno=errno.EINVAL)
@@ -1118,13 +1159,17 @@ class TaskHandler:
                 TASK_REF_POOL_NAME: image_spec[0],
                 TASK_REF_POOL_NAMESPACE: image_spec[1],
                 TASK_REF_IMAGE_NAME: image_spec[2]}
-        task = self.find_task(refs)
-        if task:
-            return 0, task.to_json(), ''
 
         with self.open_ioctx(image_spec) as ioctx:
-            self.validate_image_migrating(ioctx, image_spec)
+            status = self.get_migration_status(ioctx, image_spec)
+            if status:
+                refs[TASK_REF_IMAGE_ID] = status['dest_image_id']
+
+            task = self.find_task(refs)
+            if task:
+                return 0, task.to_json(), ''
 
+            self.validate_image_migrating(status)
             return 0, self.add_task(ioctx,
                                     "Aborting image migration for {}".format(
                                         self.format_image_spec(image_spec)),