Change in vdsm[master]: v2v: use new jobs infrastructure
shavivi at redhat.com
shavivi at redhat.com
Thu Jan 28 13:20:10 UTC 2016
Shahar Havivi has uploaded a new change for review.
Change subject: v2v: use new jobs infrastructure
......................................................................
v2v: use new jobs infrastructure
Change-Id: Iac71559e02502580de9c2e537733ec7286682050
Signed-off-by: Shahar Havivi <shaharh at redhat.com>
---
M tests/v2vTests.py
M vdsm/v2v.py
2 files changed, 25 insertions(+), 129 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/58/52858/1
diff --git a/tests/v2vTests.py b/tests/v2vTests.py
index b3988c9..677ba25 100644
--- a/tests/v2vTests.py
+++ b/tests/v2vTests.py
@@ -29,7 +29,7 @@
from testlib import namedTemporaryDir, permutations, expandPermutations
import v2v
-from vdsm import libvirtconnection
+from vdsm import jobs, libvirtconnection
from vdsm.password import ProtectedPassword
from vdsm.commands import execCmd
from vdsm.utils import CommandPath
@@ -290,7 +290,7 @@
self._vms = [MockVirDomain(*spec) for spec in VM_SPECS]
def tearDown(self):
- v2v._jobs.clear()
+ jobs._clear()
def testGetExternalVMs(self):
if not v2v.supported():
@@ -462,7 +462,7 @@
def testSuccessfulImportOVA(self):
with temporary_ovf_dir() as ovapath:
v2v.convert_ova(ovapath, self.vminfo, self.job_id, FakeIRS())
- job = v2v._jobs[self.job_id]
+ job = jobs.get(self.job_id)
job.wait()
self.assertEqual(job.status, v2v.STATUS.DONE)
@@ -501,7 +501,7 @@
self.vminfo,
self.job_id,
FakeIRS())
- job = v2v._jobs[self.job_id]
+ job = jobs.get(self.job_id)
job.wait()
self.assertEqual(job.status, v2v.STATUS.DONE)
diff --git a/vdsm/v2v.py b/vdsm/v2v.py
index 0574fd3..23b5cf3 100644
--- a/vdsm/v2v.py
+++ b/vdsm/v2v.py
@@ -32,7 +32,6 @@
import re
import signal
import tarfile
-import threading
import xml.etree.ElementTree as ET
import zipfile
@@ -41,15 +40,14 @@
from vdsm.commands import execCmd
from vdsm.constants import P_VDSM_RUN
from vdsm.define import errCode, doneCode
-from vdsm import libvirtconnection, response, concurrent
+from vdsm import jobs, libvirtconnection, response, concurrent
from vdsm.infra import zombiereaper
from vdsm.utils import traceback, CommandPath, NICENESS, IOCLASS
import caps
-_lock = threading.Lock()
-_jobs = {}
+_V2V_JOB_TYPE = 'v2v'
_V2V_DIR = os.path.join(P_VDSM_RUN, 'v2v')
_VIRT_V2V = CommandPath('virt-v2v', '/usr/bin/virt-v2v')
@@ -72,27 +70,17 @@
DiskProgress = namedtuple('DiskProgress', ['progress'])
-class STATUS:
+class STATUS(jobs.STATUS):
'''
STARTING: request granted and starting the import process
COPYING_DISK: copying disk in progress
- ABORTED: user initiated aborted
- FAILED: error during import process
- DONE: convert process successfully finished
'''
STARTING = 'starting'
COPYING_DISK = 'copying_disk'
- ABORTED = 'aborted'
- FAILED = 'error'
- DONE = 'done'
class V2VError(Exception):
''' Base class for v2v errors '''
-
-
-class ClientError(Exception):
- ''' Base class for client error '''
class InvalidVMConfiguration(ValueError):
@@ -103,23 +91,8 @@
''' Error while parsing virt-v2v output '''
-class JobExistsError(ClientError):
- ''' Job already exists in _jobs collection '''
- err_name = 'JobExistsError'
-
-
-class VolumeError(ClientError):
+class VolumeError(jobs.ClientError):
''' Error preparing volume '''
-
-
-class NoSuchJob(ClientError):
- ''' Job not exists in _jobs collection '''
- err_name = 'NoSuchJob'
-
-
-class JobNotDone(ClientError):
- ''' Import process still in progress '''
- err_name = 'JobNotDone'
class NoSuchOvf(V2VError):
@@ -131,7 +104,7 @@
''' virt-v2v process had error in execution '''
-class InvalidInputError(ClientError):
+class InvalidInputError(jobs.ClientError):
''' Invalid input received '''
@@ -167,7 +140,7 @@
command = LibvirtCommand(uri, username, password, vminfo, job_id, irs)
job = ImportVm(job_id, command)
job.start()
- _add_job(job_id, job)
+ jobs.add(job)
return {'status': doneCode}
@@ -175,7 +148,7 @@
command = OvaCommand(ova_path, vminfo, job_id, irs)
job = ImportVm(job_id, command)
job.start()
- _add_job(job_id, job)
+ jobs.add(job)
return response.success()
@@ -197,12 +170,12 @@
def get_converted_vm(job_id):
try:
- job = _get_job(job_id)
- _validate_job_done(job)
+ job = jobs.get(job_id)
+ job.validate_done()
ovf = _read_ovf(job_id)
- except ClientError as e:
+ except jobs.ClientError as e:
logging.info('Converted VM error %s', e)
- return errCode[e.err_name]
+ return errCode[e.name]
except V2VError as e:
logging.error('Converted VM error %s', e)
return errCode[e.err_name]
@@ -210,68 +183,7 @@
def delete_job(job_id):
- try:
- job = _get_job(job_id)
- _validate_job_finished(job)
- _remove_job(job_id)
- except ClientError as e:
- logging.info('Cannot delete job, error: %s', e)
- return errCode[e.err_name]
- return {'status': doneCode}
-
-
-def abort_job(job_id):
- try:
- job = _get_job(job_id)
- job.abort()
- except ClientError as e:
- logging.info('Cannot abort job, error: %s', e)
- return errCode[e.err_name]
- return {'status': doneCode}
-
-
-def get_jobs_status():
- ret = {}
- with _lock:
- items = tuple(_jobs.items())
- for job_id, job in items:
- ret[job_id] = {
- 'status': job.status,
- 'description': job.description,
- 'progress': job.progress
- }
- return ret
-
-
-def _add_job(job_id, job):
- with _lock:
- if job_id in _jobs:
- raise JobExistsError("Job %r exists" % job_id)
- _jobs[job_id] = job
-
-
-def _get_job(job_id):
- with _lock:
- if job_id not in _jobs:
- raise NoSuchJob("No such job %r" % job_id)
- return _jobs[job_id]
-
-
-def _remove_job(job_id):
- with _lock:
- if job_id not in _jobs:
- raise NoSuchJob("No such job %r" % job_id)
- del _jobs[job_id]
-
-
-def _validate_job_done(job):
- if job.status != STATUS.DONE:
- raise JobNotDone("Job %r is %s" % (job.id, job.status))
-
-
-def _validate_job_finished(job):
- if job.status not in (STATUS.DONE, STATUS.FAILED, STATUS.ABORTED):
- raise JobNotDone("Job %r is %s" % (job.id, job.status))
+ return jobs.delete(job_id)
def _read_ovf(job_id):
@@ -555,17 +467,17 @@
return env
-class ImportVm(object):
+class ImportVm(jobs.Job):
+ _JOB_TYPE = _V2V_JOB_TYPE
TERM_DELAY = 30
PROC_WAIT_TIMEOUT = 30
def __init__(self, job_id, command):
- self._id = job_id
+ super(ImportVm, self).__init__(job_id)
self._command = command
self._thread = None
self._status = STATUS.STARTING
- self._description = ''
self._disk_progress = 0
self._disk_count = 1
self._current_disk = 1
@@ -579,18 +491,6 @@
def wait(self):
if self._thread is not None and self._thread.is_alive():
self._thread.join()
-
- @property
- def id(self):
- return self._id
-
- @property
- def status(self):
- return self._status
-
- @property
- def description(self):
- return self._description
@property
def progress(self):
@@ -667,11 +567,6 @@
else:
raise RuntimeError("Job %r got unexpected parser event: %s" %
(self._id, event))
-
- def abort(self):
- self._status = STATUS.ABORTED
- logging.info('Job %r aborting...', self._id)
- self._abort()
def _abort(self):
self._aborted = True
@@ -910,8 +805,8 @@
return _read_ovf_from_zip_ova(ova_path)
elif tarfile.is_tarfile(ova_path):
return _read_ovf_from_tar_ova(ova_path)
- raise ClientError('Unknown ova format, supported formats:'
- ' tar, zip or a directory')
+ raise jobs.ClientError('Unknown ova format, supported formats:'
+ ' tar, zip or a directory')
def _find_ovf(entries):
@@ -927,7 +822,8 @@
if name is not None:
with open(os.path.join(ova_path, name), 'r') as ovf_file:
return ovf_file.read()
- raise ClientError('OVA directory %s does not contain ovf file' % ova_path)
+ raise jobs.ClientError('OVA directory %s does not contain ovf file' %
+ ova_path)
def _read_ovf_from_zip_ova(ova_path):
@@ -936,7 +832,7 @@
name = _find_ovf(zf.namelist())
if name is not None:
return zf.read(name)
- raise ClientError('OVA does not contains file with .ovf suffix')
+ raise jobs.ClientError('OVA does not contains file with .ovf suffix')
def _read_ovf_from_tar_ova(ova_path):
@@ -945,7 +841,7 @@
if member.name.endswith('.ovf'):
with closing(tar.extractfile(member)) as ovf:
return ovf.read()
- raise ClientError('OVA does not contains file with .ovf suffix')
+ raise jobs.ClientError('OVA does not contains file with .ovf suffix')
def _add_general_ovf_info(vm, node, ns):
--
To view, visit https://gerrit.ovirt.org/52858
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Iac71559e02502580de9c2e537733ec7286682050
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Shahar Havivi <shavivi at redhat.com>
More information about the vdsm-patches
mailing list