Merge "Changes behaviour when an image fails uploading"

glance/async/flows/base_import.py

@@ -71,13 +71,16 @@ class _CreateImage(task.Task):
         return image.image_id
 
     def revert(self, *args, **kwargs):
-        # TODO(flaper87): Define the revert rules for images on failures.
-        # Deleting the image may not be what we want since users could upload
-        # the image data in a separate step. However, it really depends on
-        # when the failure happened. I guess we should check if data has been
-        # written, although at that point failures are (should be) unexpected,
-        # at least image-workflow wise.
-        pass
+        # TODO(NiallBunting): Deleting the image like this could be considered
+        # a brute force way of reverting images. It may be worth checking if
+        # data has been written.
+        result = kwargs.get('result', None)
+        if result is not None:
+            if kwargs.get('flow_failures', None) is not None:
+                image = self.image_repo.get(result)
+                LOG.debug("Deleting image whilst reverting.")
+                image.delete()
+                self.image_repo.remove(image)
 
 
 class _ImportToFS(task.Task):
@@ -306,9 +309,14 @@ class _ImportToStore(task.Task):
         #     os.rename(file_path, image_path)
         #
         # image_import.set_image_data(image, image_path, None)
-
-        image_import.set_image_data(image, file_path or self.uri, self.task_id)
-
+        try:
+            image_import.set_image_data(image,
+                                        file_path or self.uri, self.task_id)
+        except IOError as e:
+            msg = (_('Uploading the image failed due to: %(exc)s') %
+                   {'exc': encodeutils.exception_to_unicode(e)})
+            LOG.error(msg)
+            raise exception.UploadException(message=msg)
         # NOTE(flaper87): We need to save the image again after the locations
         # have been set in the image.
         self.image_repo.save(image)

glance/async/taskflow_executor.py

@@ -170,6 +170,9 @@ class TaskExecutor(glance.async.TaskExecutor):
                 max_workers=CONF.taskflow_executor.max_workers)
             with llistener.DynamicLoggingListener(engine, log=LOG):
                 engine.run()
+        except exception.UploadException as exc:
+            task.fail(encodeutils.exception_to_unicode(exc))
+            self.task_repo.save(task)
         except Exception as exc:
             with excutils.save_and_reraise_exception():
                 LOG.error(_LE('Failed to execute task %(task_id)s: %(exc)s') %

glance/tests/unit/async/test_taskflow_executor.py

@@ -20,10 +20,10 @@ from oslo_config import cfg
 from taskflow import engines
 
 from glance.async import taskflow_executor
+from glance.common.scripts.image_import import main as image_import
 from glance import domain
 import glance.tests.utils as test_utils
 
-
 CONF = cfg.CONF
 TENANT1 = '6838eb7b-6ded-434a-882c-b344c77fe8df'
 
@@ -89,3 +89,14 @@ class TestTaskExecutor(test_utils.BaseTestCase):
                               self.task.task_id)
         self.assertEqual('failure', self.task.status)
         self.task_repo.save.assert_called_with(self.task)
+
+    def test_task_fail_upload(self):
+        with mock.patch.object(image_import, 'set_image_data') as import_mock:
+            import_mock.side_effect = IOError
+
+            self.task_repo.get.return_value = self.task
+            self.executor.begin_processing(self.task.task_id)
+
+        self.assertEqual('failure', self.task.status)
+        self.task_repo.save.assert_called_with(self.task)
+        self.assertEqual(1, import_mock.call_count)