CVE-2015-3241

@@ -1,17 +0,0 @@
-#
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
-#    not use this file except in compliance with the License. You may obtain
-#    a copy of the License at
-#
-#         http://www.apache.org/licenses/LICENSE-2.0
-#
-#    Unless required by applicable law or agreed to in writing, software
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-#    License for the specific language governing permissions and limitations
-#    under the License.
-
-import six
-
-
-six.add_move(six.MovedModule('mox', 'mox', 'mox3.mox'))

@@ -18,7 +18,7 @@
 """
 
 import errno
-import logging as stdlib_logging
+import logging
 import multiprocessing
 import os
 import random
@@ -30,7 +30,6 @@
 import six
 
 from nova.openstack.common.gettextutils import _
-from nova.openstack.common import log as logging
 from nova.openstack.common import strutils
 
 
@@ -116,8 +115,18 @@ def execute(*cmd, **kwargs):
                             execute this command. Defaults to false.
     :type shell:            boolean
     :param loglevel:        log level for execute commands.
-    :type loglevel:         int.  (Should be stdlib_logging.DEBUG or
-                            stdlib_logging.INFO)
+    :type loglevel:         int.  (Should be logging.DEBUG or logging.INFO)
+    :param on_execute:      This function will be called upon process creation
+                            with the object as a argument.  The Purpose of this
+                            is to allow the caller of `processutils.execute` to
+                            track process creation asynchronously.
+    :type on_execute:       function(:class:`subprocess.Popen`)
+    :param on_completion:   This function will be called upon process
+                            completion with the object as a argument.  The
+                            Purpose of this is to allow the caller of
+                            `processutils.execute` to track process completion
+                            asynchronously.
+    :type on_completion:    function(:class:`subprocess.Popen`)
     :returns:               (stdout, stderr) from process execution
     :raises:                :class:`UnknownArgumentError` on
                             receiving unknown arguments
@@ -133,7 +142,9 @@ def execute(*cmd, **kwargs):
     run_as_root = kwargs.pop('run_as_root', False)
     root_helper = kwargs.pop('root_helper', '')
     shell = kwargs.pop('shell', False)
-    loglevel = kwargs.pop('loglevel', stdlib_logging.DEBUG)
+    loglevel = kwargs.pop('loglevel', logging.DEBUG)
+    on_execute = kwargs.pop('on_execute', None)
+    on_completion = kwargs.pop('on_completion', None)
 
     if isinstance(check_exit_code, bool):
         ignore_exit_code = not check_exit_code
@@ -142,8 +153,7 @@ def execute(*cmd, **kwargs):
         check_exit_code = [check_exit_code]
 
     if kwargs:
-        raise UnknownArgumentError(_('Got unknown keyword args '
-                                     'to utils.execute: %r') % kwargs)
+        raise UnknownArgumentError(_('Got unknown keyword args: %r') % kwargs)
 
     if run_as_root and hasattr(os, 'geteuid') and os.geteuid() != 0:
         if not root_helper:
@@ -176,23 +186,32 @@ def execute(*cmd, **kwargs):
                                    preexec_fn=preexec_fn,
                                    shell=shell,
                                    env=env_variables)
-            result = None
-            for _i in six.moves.range(20):
-                # NOTE(russellb) 20 is an arbitrary number of retries to
-                # prevent any chance of looping forever here.
-                try:
-                    if process_input is not None:
-                        result = obj.communicate(process_input)
-                    else:
-                        result = obj.communicate()
-                except OSError as e:
-                    if e.errno in (errno.EAGAIN, errno.EINTR):
-                        continue
-                    raise
-                break
-            obj.stdin.close()  # pylint: disable=E1101
-            _returncode = obj.returncode  # pylint: disable=E1101
-            LOG.log(loglevel, 'Result was %s' % _returncode)
+
+            if on_execute:
+                on_execute(obj)
+
+            try:
+                result = None
+                for _i in six.moves.range(20):
+                    # NOTE(russellb) 20 is an arbitrary number of retries to
+                    # prevent any chance of looping forever here.
+                    try:
+                        if process_input is not None:
+                            result = obj.communicate(process_input)
+                        else:
+                            result = obj.communicate()
+                    except OSError as e:
+                        if e.errno in (errno.EAGAIN, errno.EINTR):
+                            continue
+                        raise
+                    break
+                obj.stdin.close()  # pylint: disable=E1101
+                _returncode = obj.returncode  # pylint: disable=E1101
+                LOG.log(loglevel, 'Result was %s' % _returncode)
+            finally:
+                if on_completion:
+                    on_completion(obj)
+
             if not ignore_exit_code and _returncode not in check_exit_code:
                 (stdout, stderr) = result
                 sanitized_stdout = strutils.mask_password(stdout)

@@ -23,6 +23,7 @@
 import random
 import re
 import shutil
+import signal
 import threading
 import time
 import uuid
@@ -9951,6 +9952,15 @@ def test_shared_storage_detection_easy(self):
         self.mox.ReplayAll()
         self.assertTrue(drvr._is_storage_shared_with('foo', '/path'))
 
+    def test_store_pid_remove_pid(self):
+        instance = objects.Instance(**self.test_instance)
+        drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
+        popen = mock.Mock(pid=3)
+        drvr.job_tracker.add_job(instance, popen.pid)
+        self.assertIn(3, drvr.job_tracker.jobs[instance.uuid])
+        drvr.job_tracker.remove_job(instance, popen.pid)
+        self.assertNotIn(instance.uuid, drvr.job_tracker.jobs)
+
     @mock.patch('nova.virt.libvirt.host.Host.get_domain')
     def test_get_domain_info_with_more_return(self, mock_get_domain):
         instance = objects.Instance(**self.test_instance)
@@ -11533,12 +11543,18 @@ def fake_get_host_ip_addr():
         def fake_execute(*args, **kwargs):
             pass
 
+        def fake_copy_image(src, dest, host=None, receive=False,
+                            on_execute=None, on_completion=None):
+            self.assertIsNotNone(on_execute)
+            self.assertIsNotNone(on_completion)
+
         self.stubs.Set(self.drvr, 'get_instance_disk_info',
                        fake_get_instance_disk_info)
         self.stubs.Set(self.drvr, '_destroy', fake_destroy)
         self.stubs.Set(self.drvr, 'get_host_ip_addr',
                        fake_get_host_ip_addr)
         self.stubs.Set(utils, 'execute', fake_execute)
+        self.stubs.Set(libvirt_utils, 'copy_image', fake_copy_image)
 
         ins_ref = self._create_instance(params=params_for_instance)
 
@@ -12646,6 +12662,28 @@ def test_delete_instance_files(self, get_instance_path, exists, exe,
         shutil.assert_called_with('/path_del')
         self.assertTrue(result)
 
+    @mock.patch('shutil.rmtree')
+    @mock.patch('nova.utils.execute')
+    @mock.patch('os.path.exists')
+    @mock.patch('os.kill')
+    @mock.patch('nova.virt.libvirt.utils.get_instance_path')
+    def test_delete_instance_files_kill_running(
+            self, get_instance_path, kill, exists, exe, shutil):
+        get_instance_path.return_value = '/path'
+        instance = objects.Instance(uuid='fake-uuid', id=1)
+        self.drvr.job_tracker.jobs[instance.uuid] = [3, 4]
+
+        exists.side_effect = [False, False, True, False]
+
+        result = self.drvr.delete_instance_files(instance)
+        get_instance_path.assert_called_with(instance)
+        exe.assert_called_with('mv', '/path', '/path_del')
+        kill.assert_has_calls([mock.call(3, signal.SIGKILL), mock.call(3, 0),
+                               mock.call(4, signal.SIGKILL), mock.call(4, 0)])
+        shutil.assert_called_with('/path_del')
+        self.assertTrue(result)
+        self.assertNotIn(instance.uuid, self.drvr.job_tracker.jobs)
+
     @mock.patch('shutil.rmtree')
     @mock.patch('nova.utils.execute')
     @mock.patch('os.path.exists')

@@ -62,7 +62,8 @@ def test_copy_image_local_cp(self, mock_execute):
         mock_execute.assert_called_once_with('cp', 'src', 'dest')
 
     _rsync_call = functools.partial(mock.call,
-                                    'rsync', '--sparse', '--compress')
+                                    'rsync', '--sparse', '--compress',
+                                    on_execute=None, on_completion=None)
 
     @mock.patch('nova.utils.execute')
     def test_copy_image_rsync(self, mock_execute):
@@ -85,7 +86,8 @@ def test_copy_image_scp(self, mock_execute):
 
         mock_execute.assert_has_calls([
             self._rsync_call('--dry-run', 'src', 'host:dest'),
-            mock.call('scp', 'src', 'host:dest'),
+            mock.call('scp', 'src', 'host:dest',
+                      on_execute=None, on_completion=None),
         ])
         self.assertEqual(2, mock_execute.call_count)
 
@@ -110,7 +112,8 @@ def test_copy_image_scp_ipv6(self, mock_execute):
 
         mock_execute.assert_has_calls([
             self._rsync_call('--dry-run', 'src', '[2600::]:dest'),
-            mock.call('scp', 'src', '[2600::]:dest'),
+            mock.call('scp', 'src', '[2600::]:dest',
+                      on_execute=None, on_completion=None),
         ])
         self.assertEqual(2, mock_execute.call_count)
 

@@ -97,6 +97,7 @@
 from nova.virt.libvirt import host
 from nova.virt.libvirt import imagebackend
 from nova.virt.libvirt import imagecache
+from nova.virt.libvirt import instancejobtracker
 from nova.virt.libvirt.storage import dmcrypt
 from nova.virt.libvirt.storage import lvm
 from nova.virt.libvirt.storage import rbd_utils
@@ -468,6 +469,8 @@ def __init__(self, virtapi, read_only=False):
                    'expect': ', '.join("'%s'" % k for k in
                                        sysinfo_serial_funcs.keys())})
 
+        self.job_tracker = instancejobtracker.InstanceJobTracker()
+
     def _get_volume_drivers(self):
         return libvirt_volume_drivers
 
@@ -6290,6 +6293,11 @@ def migrate_disk_and_power_off(self, context, instance, dest,
                     # finish_migration/_create_image to re-create it for us.
                     continue
 
+                on_execute = lambda process: self.job_tracker.add_job(
+                    instance, process.pid)
+                on_completion = lambda process: self.job_tracker.remove_job(
+                    instance, process.pid)
+
                 if info['type'] == 'qcow2' and info['backing_file']:
                     tmp_path = from_path + "_rbase"
                     # merge backing file
@@ -6299,11 +6307,15 @@ def migrate_disk_and_power_off(self, context, instance, dest,
                     if shared_storage:
                         utils.execute('mv', tmp_path, img_path)
                     else:
-                        libvirt_utils.copy_image(tmp_path, img_path, host=dest)
+                        libvirt_utils.copy_image(tmp_path, img_path, host=dest,
+                                                 on_execute=on_execute,
+                                                 on_completion=on_completion)
                         utils.execute('rm', '-f', tmp_path)
 
                 else:  # raw or qcow2 with no backing file
-                    libvirt_utils.copy_image(from_path, img_path, host=dest)
+                    libvirt_utils.copy_image(from_path, img_path, host=dest,
+                                             on_execute=on_execute,
+                                             on_completion=on_completion)
         except Exception:
             with excutils.save_and_reraise_exception():
                 self._cleanup_remote_migration(dest, inst_base,
@@ -6690,6 +6702,8 @@ def delete_instance_files(self, instance):
         # invocation failed due to the absence of both target and
         # target_resize.
         if not remaining_path and os.path.exists(target_del):
+            self.job_tracker.terminate_jobs(instance)
+
             LOG.info(_LI('Deleting instance files %s'), target_del,
                      instance=instance)
             remaining_path = target_del

@@ -0,0 +1,98 @@
+# Copyright 2015 NTT corp.
+# All Rights Reserved.
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+import collections
+import errno
+import os
+import signal
+
+from oslo_log import log as logging
+
+from nova.i18n import _LE
+from nova.i18n import _LW
+
+
+LOG = logging.getLogger(__name__)
+
+
+class InstanceJobTracker(object):
+    def __init__(self):
+        self.jobs = collections.defaultdict(list)
+
+    def add_job(self, instance, pid):
+        """Appends process_id of instance to cache.
+
+        This method will store the pid of a process in cache as
+        a key: value pair which will be used to kill the process if it
+        is running while deleting the instance. Instance uuid is used as
+        a key in the cache and pid will be the value.
+
+        :param instance: Object of instance
+        :param pid: Id of the process
+        """
+        self.jobs[instance.uuid].append(pid)
+
+    def remove_job(self, instance, pid):
+        """Removes pid of process from cache.
+
+        This method will remove the pid of a process from the cache.
+
+        :param instance: Object of instance
+        :param pid: Id of the process
+        """
+        uuid = instance.uuid
+        if uuid in self.jobs and pid in self.jobs[uuid]:
+            self.jobs[uuid].remove(pid)
+
+        # remove instance.uuid if no pid's remaining
+        if not self.jobs[uuid]:
+            self.jobs.pop(uuid, None)
+
+    def terminate_jobs(self, instance):
+        """Kills the running processes for given instance.
+
+        This method is used to kill all running processes of the instance if
+        it is deleted in between.
+
+        :param instance: Object of instance
+        """
+        pids_to_remove = list(self.jobs.get(instance.uuid, []))
+        for pid in pids_to_remove:
+            try:
+                # Try to kill the process
+                os.kill(pid, signal.SIGKILL)
+            except OSError as exc:
+                if exc.errno != errno.ESRCH:
+                    LOG.error(_LE('Failed to kill process %(pid)s '
+                                  'due to %(reason)s, while deleting the '
+                                  'instance.'), {'pid': pid, 'reason': exc},
+                              instance=instance)
+
+            try:
+                # Check if the process is still alive.
+                os.kill(pid, 0)
+            except OSError as exc:
+                if exc.errno != errno.ESRCH:
+                    LOG.error(_LE('Unexpected error while checking process '
+                                  '%(pid)s.'), {'pid': pid},
+                              instance=instance)
+            else:
+                # The process is still around
+                LOG.warn(_LW("Failed to kill a long running process "
+                             "%(pid)s related to the instance when "
+                             "deleting it."), {'pid': pid},
+                         instance=instance)
+
+            self.remove_job(instance, pid)

@@ -183,13 +183,16 @@ def get_disk_backing_file(path, basename=True):
     return backing_file
 
 
-def copy_image(src, dest, host=None, receive=False):
+def copy_image(src, dest, host=None, receive=False,
+               on_execute=None, on_completion=None):
     """Copy a disk image to an existing directory
 
     :param src: Source image
     :param dest: Destination path
     :param host: Remote host
     :param receive: Reverse the rsync direction
+    :param on_execute: Callback method to store pid of process in cache
+    :param on_completion: Callback method to remove pid of process from cache
     """
 
     if not host:
@@ -211,11 +214,14 @@ def copy_image(src, dest, host=None, receive=False):
             # Do a relatively light weight test first, so that we
             # can fall back to scp, without having run out of space
             # on the destination for example.
-            execute('rsync', '--sparse', '--compress', '--dry-run', src, dest)
+            execute('rsync', '--sparse', '--compress', '--dry-run', src, dest,
+                    on_execute=on_execute, on_completion=on_completion)
         except processutils.ProcessExecutionError:
-            execute('scp', src, dest)
+            execute('scp', src, dest, on_execute=on_execute,
+                    on_completion=on_completion)
         else:
-            execute('rsync', '--sparse', '--compress', src, dest)
+            execute('rsync', '--sparse', '--compress', src, dest,
+                    on_execute=on_execute, on_completion=on_completion)
 
 
 def write_to_file(path, contents, umask=None):

@@ -23,6 +23,7 @@
 import random
 import re
 import shutil
+import signal
 import threading
 import time
 import uuid
@@ -9817,6 +9818,15 @@ def test_shared_storage_detection_easy(self):
         self.mox.ReplayAll()
         self.assertTrue(drvr._is_storage_shared_with('foo', '/path'))
 
+    def test_store_pid_remove_pid(self):
+        instance = objects.Instance(**self.test_instance)
+        drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
+        popen = mock.Mock(pid=3)
+        drvr.job_tracker.add_job(instance, popen.pid)
+        self.assertIn(3, drvr.job_tracker.jobs[instance.uuid])
+        drvr.job_tracker.remove_job(instance, popen.pid)
+        self.assertNotIn(instance.uuid, drvr.job_tracker.jobs)
+
     @mock.patch('nova.virt.libvirt.host.Host.get_domain')
     def test_get_domain_info_with_more_return(self, mock_get_domain):
         instance = objects.Instance(**self.test_instance)
@@ -11316,12 +11326,18 @@ def fake_get_host_ip_addr():
         def fake_execute(*args, **kwargs):
             pass
 
+        def fake_copy_image(src, dest, host=None, receive=False,
+                            on_execute=None, on_completion=None):
+            self.assertIsNotNone(on_execute)
+            self.assertIsNotNone(on_completion)
+
         self.stubs.Set(self.drvr, 'get_instance_disk_info',
                        fake_get_instance_disk_info)
         self.stubs.Set(self.drvr, '_destroy', fake_destroy)
         self.stubs.Set(self.drvr, 'get_host_ip_addr',
                        fake_get_host_ip_addr)
         self.stubs.Set(utils, 'execute', fake_execute)
+        self.stubs.Set(libvirt_utils, 'copy_image', fake_copy_image)
 
         ins_ref = self._create_instance(params=params_for_instance)
 
@@ -12425,6 +12441,28 @@ def test_delete_instance_files(self, get_instance_path, exists, exe,
         shutil.assert_called_with('/path_del')
         self.assertTrue(result)
 
+    @mock.patch('shutil.rmtree')
+    @mock.patch('nova.utils.execute')
+    @mock.patch('os.path.exists')
+    @mock.patch('os.kill')
+    @mock.patch('nova.virt.libvirt.utils.get_instance_path')
+    def test_delete_instance_files_kill_running(
+            self, get_instance_path, kill, exists, exe, shutil):
+        get_instance_path.return_value = '/path'
+        instance = objects.Instance(uuid='fake-uuid', id=1)
+        self.drvr.job_tracker.jobs[instance.uuid] = [3, 4]
+
+        exists.side_effect = [False, False, True, False]
+
+        result = self.drvr.delete_instance_files(instance)
+        get_instance_path.assert_called_with(instance)
+        exe.assert_called_with('mv', '/path', '/path_del')
+        kill.assert_has_calls([mock.call(3, signal.SIGKILL), mock.call(3, 0),
+                               mock.call(4, signal.SIGKILL), mock.call(4, 0)])
+        shutil.assert_called_with('/path_del')
+        self.assertTrue(result)
+        self.assertNotIn(instance.uuid, self.drvr.job_tracker.jobs)
+
     @mock.patch('shutil.rmtree')
     @mock.patch('nova.utils.execute')
     @mock.patch('os.path.exists')

@@ -62,7 +62,8 @@ def test_copy_image_local_cp(self, mock_execute):
         mock_execute.assert_called_once_with('cp', 'src', 'dest')
 
     _rsync_call = functools.partial(mock.call,
-                                    'rsync', '--sparse', '--compress')
+                                    'rsync', '--sparse', '--compress',
+                                    on_execute=None, on_completion=None)
 
     @mock.patch('nova.utils.execute')
     def test_copy_image_rsync(self, mock_execute):
@@ -85,7 +86,8 @@ def test_copy_image_scp(self, mock_execute):
 
         mock_execute.assert_has_calls([
             self._rsync_call('--dry-run', 'src', 'host:dest'),
-            mock.call('scp', 'src', 'host:dest'),
+            mock.call('scp', 'src', 'host:dest',
+                      on_execute=None, on_completion=None),
         ])
         self.assertEqual(2, mock_execute.call_count)
 
@@ -110,7 +112,8 @@ def test_copy_image_scp_ipv6(self, mock_execute):
 
         mock_execute.assert_has_calls([
             self._rsync_call('--dry-run', 'src', '[2600::]:dest'),
-            mock.call('scp', 'src', '[2600::]:dest'),
+            mock.call('scp', 'src', '[2600::]:dest',
+                      on_execute=None, on_completion=None),
         ])
         self.assertEqual(2, mock_execute.call_count)
 

@@ -95,6 +95,7 @@
 from nova.virt.libvirt import host
 from nova.virt.libvirt import imagebackend
 from nova.virt.libvirt import imagecache
+from nova.virt.libvirt import instancejobtracker
 from nova.virt.libvirt import lvm
 from nova.virt.libvirt import rbd_utils
 from nova.virt.libvirt import utils as libvirt_utils
@@ -465,6 +466,8 @@ def __init__(self, virtapi, read_only=False):
                    'expect': ', '.join("'%s'" % k for k in
                                        sysinfo_serial_funcs.keys())})
 
+        self.job_tracker = instancejobtracker.InstanceJobTracker()
+
     def _get_volume_drivers(self):
         return libvirt_volume_drivers
 
@@ -6301,6 +6304,11 @@ def migrate_disk_and_power_off(self, context, instance, dest,
                     # finish_migration/_create_image to re-create it for us.
                     continue
 
+                on_execute = lambda process: self.job_tracker.add_job(
+                    instance, process.pid)
+                on_completion = lambda process: self.job_tracker.remove_job(
+                    instance, process.pid)
+
                 if info['type'] == 'qcow2' and info['backing_file']:
                     tmp_path = from_path + "_rbase"
                     # merge backing file
@@ -6310,11 +6318,15 @@ def migrate_disk_and_power_off(self, context, instance, dest,
                     if shared_storage:
                         utils.execute('mv', tmp_path, img_path)
                     else:
-                        libvirt_utils.copy_image(tmp_path, img_path, host=dest)
+                        libvirt_utils.copy_image(tmp_path, img_path, host=dest,
+                                                 on_execute=on_execute,
+                                                 on_completion=on_completion)
                         utils.execute('rm', '-f', tmp_path)
 
                 else:  # raw or qcow2 with no backing file
-                    libvirt_utils.copy_image(from_path, img_path, host=dest)
+                    libvirt_utils.copy_image(from_path, img_path, host=dest,
+                                             on_execute=on_execute,
+                                             on_completion=on_completion)
         except Exception:
             with excutils.save_and_reraise_exception():
                 self._cleanup_remote_migration(dest, inst_base,
@@ -6683,6 +6695,8 @@ def delete_instance_files(self, instance):
         # invocation failed due to the absence of both target and
         # target_resize.
         if not remaining_path and os.path.exists(target_del):
+            self.job_tracker.terminate_jobs(instance)
+
             LOG.info(_LI('Deleting instance files %s'), target_del,
                      instance=instance)
             remaining_path = target_del

@@ -0,0 +1,98 @@
+# Copyright 2015 NTT corp.
+# All Rights Reserved.
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+import collections
+import errno
+import os
+import signal
+
+from oslo_log import log as logging
+
+from nova.i18n import _LE
+from nova.i18n import _LW
+
+
+LOG = logging.getLogger(__name__)
+
+
+class InstanceJobTracker(object):
+    def __init__(self):
+        self.jobs = collections.defaultdict(list)
+
+    def add_job(self, instance, pid):
+        """Appends process_id of instance to cache.
+
+        This method will store the pid of a process in cache as
+        a key: value pair which will be used to kill the process if it
+        is running while deleting the instance. Instance uuid is used as
+        a key in the cache and pid will be the value.
+
+        :param instance: Object of instance
+        :param pid: Id of the process
+        """
+        self.jobs[instance.uuid].append(pid)
+
+    def remove_job(self, instance, pid):
+        """Removes pid of process from cache.
+
+        This method will remove the pid of a process from the cache.
+
+        :param instance: Object of instance
+        :param pid: Id of the process
+        """
+        uuid = instance.uuid
+        if uuid in self.jobs and pid in self.jobs[uuid]:
+            self.jobs[uuid].remove(pid)
+
+        # remove instance.uuid if no pid's remaining
+        if not self.jobs[uuid]:
+            self.jobs.pop(uuid, None)
+
+    def terminate_jobs(self, instance):
+        """Kills the running processes for given instance.
+
+        This method is used to kill all running processes of the instance if
+        it is deleted in between.
+
+        :param instance: Object of instance
+        """
+        pids_to_remove = list(self.jobs.get(instance.uuid, []))
+        for pid in pids_to_remove:
+            try:
+                # Try to kill the process
+                os.kill(pid, signal.SIGKILL)
+            except OSError as exc:
+                if exc.errno != errno.ESRCH:
+                    LOG.error(_LE('Failed to kill process %(pid)s '
+                                  'due to %(reason)s, while deleting the '
+                                  'instance.'), {'pid': pid, 'reason': exc},
+                              instance=instance)
+
+            try:
+                # Check if the process is still alive.
+                os.kill(pid, 0)
+            except OSError as exc:
+                if exc.errno != errno.ESRCH:
+                    LOG.error(_LE('Unexpected error while checking process '
+                                  '%(pid)s.'), {'pid': pid},
+                              instance=instance)
+            else:
+                # The process is still around
+                LOG.warn(_LW("Failed to kill a long running process "
+                             "%(pid)s related to the instance when "
+                             "deleting it."), {'pid': pid},
+                         instance=instance)
+
+            self.remove_job(instance, pid)

@@ -294,13 +294,16 @@ def get_disk_backing_file(path, basename=True):
     return backing_file
 
 
-def copy_image(src, dest, host=None, receive=False):
+def copy_image(src, dest, host=None, receive=False,
+               on_execute=None, on_completion=None):
     """Copy a disk image to an existing directory
 
     :param src: Source image
     :param dest: Destination path
     :param host: Remote host
     :param receive: Reverse the rsync direction
+    :param on_execute: Callback method to store pid of process in cache
+    :param on_completion: Callback method to remove pid of process from cache
     """
 
     if not host:
@@ -322,11 +325,14 @@ def copy_image(src, dest, host=None, receive=False):
             # Do a relatively light weight test first, so that we
             # can fall back to scp, without having run out of space
             # on the destination for example.
-            execute('rsync', '--sparse', '--compress', '--dry-run', src, dest)
+            execute('rsync', '--sparse', '--compress', '--dry-run', src, dest,
+                    on_execute=on_execute, on_completion=on_completion)
         except processutils.ProcessExecutionError:
-            execute('scp', src, dest)
+            execute('scp', src, dest, on_execute=on_execute,
+                    on_completion=on_completion)
         else:
-            execute('rsync', '--sparse', '--compress', src, dest)
+            execute('rsync', '--sparse', '--compress', src, dest,
+                    on_execute=on_execute, on_completion=on_completion)
 
 
 def write_to_file(path, contents, umask=None):