Change in vdsm[master]: v2v: Convert VM from external source to Data Domain

nsoffer at redhat.com nsoffer at redhat.com
Mon Mar 16 00:04:03 UTC 2015


Nir Soffer has posted comments on this change.

Change subject: v2v: Convert VM from external source to Data Domain
......................................................................


Patch Set 10:

(32 comments)

https://gerrit.ovirt.org/#/c/37509/10/vdsm/v2v.py
File vdsm/v2v.py:

Line 159: 
Line 160: def delete_job(jobId):
Line 161:     try:
Line 162:         job = _get_job(jobId)
Line 163:         _validate_job_done(job)
This will prevent removal of aborted or failed jobs.

This check should be _validate_job_finished:

    def _validate_job_finished(job):
        if job.status not in (_STATUS_DONE, _STATUS_ERROR, _STATUS_ABORT):
            raise ...
Line 164:         _remove_job(jobId)
Line 165:     except V2VJobError as e:
Line 166:         return errCode[e.code]
Line 167:     return {'status': doneCode}


Line 231:                           file_name, e.errno, e.message)
Line 232:             raise
Line 233: 
Line 234: 
Line 235: class ImportVm():
This is old style class - use ImportVm(object)
Line 236:     def __init__(self, uri, username, password, vmProperties, jobId, cif):
Line 237:         self._uri = uri
Line 238:         self._username = username
Line 239:         self._password = password


Line 273:             portion ie if we have 2 disks the first will take
Line 274:             0-50 and the second 50-100
Line 275:         '''
Line 276:         return (self._disk_progress / self._disk_count +
Line 277:                 int(100 / self._disk_count) * (self._current_disk - 1))
This is little too complected, and may never return 100 for some values. For example, when finishing copy of disk 3 of 3:

    (100 / 3) + (100 / 3 * 2) -> 99

We can simplify this to:

    completed = (self._disk_count - 1) * 100
    return (completed + self._disk_progress) / self._disk_count

Which gives:

    (200 + 100) / 3 -> 100
Line 278: 
Line 279:     @traceback(msg="Error importing vm")
Line 280:     def _run(self):
Line 281:         self._create_passwd_file()


Line 282:         try:
Line 283:             self._import()
Line 284:         except Exception as ex:
Line 285:             self._status = STATUS.ERROR
Line 286:             self._status_msg = ex.message
You must abort here, otherwise you are leaving stale virt-v2v processes.

Better have a public abort() and private _abort(). abort() will set the status to ABORT and invoke _abort. Here you will invoke _abort.
Line 287:             raise
Line 288:         finally:
Line 289:             self._delete_passwd_file()
Line 290:             self._teardownVolumes()


Line 286:             self._status_msg = ex.message
Line 287:             raise
Line 288:         finally:
Line 289:             self._delete_passwd_file()
Line 290:             self._teardownVolumes()
Please use pep8_style_function_names - you already use this for most functions.
Line 291: 
Line 292:     def _import(self):
Line 293:         self._status = STATUS.STARTING
Line 294:         # TODO: use the process handling http://gerrit.ovirt.org/#/c/33909/


Line 314:         self._watch_process_output()
Line 315: 
Line 316:         if self._proc.returncode != 0:
Line 317:             self._status = STATUS.ERROR
Line 318:             self._handle_process_errors()
Instead of this, get process error and raise it:

    error = self._get_process_error()
    raise V2VProcessError(error)
Line 319:         else:
Line 320:             self._status = STATUS.DONE
Line 321: 
Line 322:     def disk_copy_stage(self, status_msg, current_disk, disk_count):


Line 315: 
Line 316:         if self._proc.returncode != 0:
Line 317:             self._status = STATUS.ERROR
Line 318:             self._handle_process_errors()
Line 319:         else:
If you raise, you don't need the else.
Line 320:             self._status = STATUS.DONE
Line 321: 
Line 322:     def disk_copy_stage(self, status_msg, current_disk, disk_count):
Line 323:         self._status = STATUS.DISK_COPY


Line 322:     def disk_copy_stage(self, status_msg, current_disk, disk_count):
Line 323:         self._status = STATUS.DISK_COPY
Line 324:         self._status_msg = status_msg
Line 325:         self._current_disk = int(current_disk)
Line 326:         self._disk_count = int(disk_count)
I understand from this that we don't know the number of disks or the size of the disks before we start the import?
Line 327:         self._disk_progress = 0
Line 328:         self._watch_copy_prpgress()
Line 329: 
Line 330:     def progress_stage(self, progress):


Line 333:             if self._disk_progress != 100:
Line 334:                 self._watch_copy_prpgress()
Line 335:         except ValueError:
Line 336:             self._status = STATUS.ERROR
Line 337:             self._abort_msg('error converting disk progress')
Instead of this, abort the process and raise:

    self.abort()
    raise JobProgressError("Error parsing progress: %r" % progress).

This will set self._status to STATUS_ERROR in _run.
Line 338:             raise
Line 339: 
Line 340:     def _watch_process_output(self):
Line 341:         while not self._aborted and self._proc.returncode is None:


Line 340:     def _watch_process_output(self):
Line 341:         while not self._aborted and self._proc.returncode is None:
Line 342:             line = self._proc.stdout.readline()
Line 343:             try:
Line 344:                 self._parser.parse_line(line)
I this would return a value, it would be much simpler - the logic in disk_copy_stage can move here - or you can call disk_copy_stage() from here:

    try:
        msg, current, count = self._parser.parse_line(line)
    except ...
        raise ...
    else:
        starts copying a new disk ...
Line 345:             except JobOutputParsing:
Line 346:                 self._status = STATUS.ERROR
Line 347:                 self._abort_msg("error parsing 'Copying disk' section: %s"
Line 348:                                 % line)


Line 344:                 self._parser.parse_line(line)
Line 345:             except JobOutputParsing:
Line 346:                 self._status = STATUS.ERROR
Line 347:                 self._abort_msg("error parsing 'Copying disk' section: %s"
Line 348:                                 % line)
Do not handle it here, let it bubble up to _run
Line 349:                 raise
Line 350: 
Line 351:     def _watch_copy_prpgress(self):
Line 352:         while not self._aborted and self._proc.returncode is None:


Line 353:             buf = ""
Line 354:             while '\r' not in buf:
Line 355:                 buf += self._proc.stdout.read(1)
Line 356:             try:
Line 357:                 self._parser.parse_progress(buf)
Same here - progress return from parse_progress, and handled here.
Line 358:             except JobOutputParsing:
Line 359:                 self._status = STATUS.ERROR
Line 360:                 self._abort_msg('error parsing disk progress')
Line 361:                 raise


Line 357:                 self._parser.parse_progress(buf)
Line 358:             except JobOutputParsing:
Line 359:                 self._status = STATUS.ERROR
Line 360:                 self._abort_msg('error parsing disk progress')
Line 361:                 raise
Don't handle it here - just let it bubble up to _run.
Line 362: 
Line 363:     def _handle_process_errors(self):
Line 364:         error = self._proc.stderr.readlines()
Line 365:         if len(error) > 0:


Line 364:         error = self._proc.stderr.readlines()
Line 365:         if len(error) > 0:
Line 366:             msg = 'error process ended with errors: %s' % error
Line 367:             self._status_msg = msg
Line 368:             logging.error(msg)
Rename to _get_process_error and return an error message.
Line 369: 
Line 370:     def _create_passwd_file(self):
Line 371:         try:
Line 372:             os.mkdir(_V2V_DIR, 0o700)


Line 366:             msg = 'error process ended with errors: %s' % error
Line 367:             self._status_msg = msg
Line 368:             logging.error(msg)
Line 369: 
Line 370:     def _create_passwd_file(self):
The first part should be extracted to _create_v2v_dir()
Line 371:         try:
Line 372:             os.mkdir(_V2V_DIR, 0o700)
Line 373:         except OSError as e:
Line 374:             if e.errno == os.errno.EEXIST:


Line 370:     def _create_passwd_file(self):
Line 371:         try:
Line 372:             os.mkdir(_V2V_DIR, 0o700)
Line 373:         except OSError as e:
Line 374:             if e.errno == os.errno.EEXIST:
os.errno is not a public interface of module "os". Use errno module.
Line 375:                 os.chmod(_V2V_DIR, 0o700)
Line 376:             else:
Line 377:                 raise
Line 378: 


Line 373:         except OSError as e:
Line 374:             if e.errno == os.errno.EEXIST:
Line 375:                 os.chmod(_V2V_DIR, 0o700)
Line 376:             else:
Line 377:                 raise
I would reverse the order:

    if e.errno != errno.EEXIST:
        raise
    os.chmod(_V2V_DIR, 0o700)

Why? we first handle errors and run away - we don't have anything to do here if we failed.

Then we continue with the normal flow. This is most of the time simpler and more clear.
Line 378: 
Line 379:         self._pass_file = os.path.join(_V2V_DIR, "%s.tmp" % self.id)
Line 380:         f = os.fdopen(os.open(self._pass_file,
Line 381:                               os.O_WRONLY | os.O_CREAT, 0600), 'w')


Line 375:                 os.chmod(_V2V_DIR, 0o700)
Line 376:             else:
Line 377:                 raise
Line 378: 
Line 379:         self._pass_file = os.path.join(_V2V_DIR, "%s.tmp" % self.id)
You should not add variables outside of __init__. Either initialize this in __init__, or use a method returning this value.

    def _password_file(self):
        return os.path.join(_V2V_DIR, "%s.tmp" % self.id)
Line 380:         f = os.fdopen(os.open(self._pass_file,
Line 381:                               os.O_WRONLY | os.O_CREAT, 0600), 'w')
Line 382:         with closing(f):
Line 383:             f.write(self._password)


Line 379:         self._pass_file = os.path.join(_V2V_DIR, "%s.tmp" % self.id)
Line 380:         f = os.fdopen(os.open(self._pass_file,
Line 381:                               os.O_WRONLY | os.O_CREAT, 0600), 'w')
Line 382:         with closing(f):
Line 383:             f.write(self._password)
Why do you need fdopen? You are writing tiny file, and do not need any buffering. Just use the fd:

    fd = os.open(self._pass_file, os.O_WRONLY | os.O_CREAT, 0600)
    try:
        os.write(fd, self._password)
    finally:
        os.close(fd)
Line 384: 
Line 385:     def _delete_passwd_file(self):
Line 386:         try:
Line 387:             os.remove(self._pass_file)


Line 394:         self._status = STATUS.ABORT
Line 395:         logging.info('aborting job id: %r', self.id)
Line 396:         if self._proc.returncode is None:
Line 397:             ret, out, err = execCmd([EXT_KILL, '-9', self._proc.pid],
Line 398:                                     sudo=True)
virt-v2v runs as root? If we are the parent process, it runs as vdsm, and we do not need to kill it with kill command running as root.

You can simply use Popen api:

    self._proc.kill()

If virt-v2v is in D state, this may fail to stop it, and in this case we would like to log an error.

We can do something like this:

    for i in range(self.ABORT_RETRIES):
        if self._proc.returncode is not None:
            self.log.debug("virt-v2v terminated with exit code: %d", 
                            self._proc.returncode)
            return
        self.log.debug("Trying to kill virt-v2v (attempt %d/%d)", 
                       i + 1, self.ABORT_RETRIES)
        self._proc.kill()
        time.sleep(self.ABORT_DELAY)

    log.error("Error killing virt-v2v (pid: %d)", self._proc.pid)

https://docs.python.org/2/library/subprocess.html#subprocess.Popen.kill
Line 399:             if len(err) > 0:
Line 400:                 self._status_msg = 'Error kill %s: %s' % (self._proc.pid, err)
Line 401:                 logging.error(self._status_msg)
Line 402: 


Line 402: 
Line 403:     def _abort_msg(self, msg):
Line 404:         self._status_msg = msg
Line 405:         logging.error(msg)
Line 406:         self._aborted = True
Not needed if you let errors bubble up to _run and handle them there.
Line 407: 
Line 408:     def _generate_disk_parameters(self):
Line 409:         images = []
Line 410:         for disk in self._vmProperties['disks']:


Line 430:             if res['status']['code']:
Line 431:                 self._status = STATUS.ERROR
Line 432:                 self._status_msg = 'Bad volume specification: %s' % drive
Line 433:                 raise VolumeError(drive)
Line 434:             else:
Why else? if prepareImage failed, we just raised, and we never reach this point.
Line 435:                 self._preparedImages.extend([drive])
Line 436:         return self._extract_storage_domain_path(res['path'])
Line 437: 
Line 438:     def _extract_storage_domain_path(self, path):


Line 431:                 self._status = STATUS.ERROR
Line 432:                 self._status_msg = 'Bad volume specification: %s' % drive
Line 433:                 raise VolumeError(drive)
Line 434:             else:
Line 435:                 self._preparedImages.extend([drive])
Use self._preparedImages.append(drive)

Also, you call this _prepareVolumes and _teardownVolumes - so why not _preparedVolumes?
Line 436:         return self._extract_storage_domain_path(res['path'])
Line 437: 
Line 438:     def _extract_storage_domain_path(self, path):
Line 439:         ''' prepareImage returns /prefix/sdUUID/images/imgUUID/volUUID


Line 432:                 self._status_msg = 'Bad volume specification: %s' % drive
Line 433:                 raise VolumeError(drive)
Line 434:             else:
Line 435:                 self._preparedImages.extend([drive])
Line 436:         return self._extract_storage_domain_path(res['path'])
Don't use res, use self._preparedImages[0]['path']
Line 437: 
Line 438:     def _extract_storage_domain_path(self, path):
Line 439:         ''' prepareImage returns /prefix/sdUUID/images/imgUUID/volUUID
Line 440:             we need storage domain absolute path so we go up 3 levels '''


Line 437: 
Line 438:     def _extract_storage_domain_path(self, path):
Line 439:         ''' prepareImage returns /prefix/sdUUID/images/imgUUID/volUUID
Line 440:             we need storage domain absolute path so we go up 3 levels '''
Line 441:         path = os.path.split(path)
split returns a tuple:

    >>> os.path.split('/prefix/a/b/c')
    ('/prefix/a/b', 'c')

You can use os.path.dirname():

    >>> os.path.dirname(os.path.dirname(os.path.dirname('/prefix/a/b/c')))
    '/prefix'

But why do we need the storage domain path? I don't think we should be interested in this.
Line 442:         path = os.path.split(path)
Line 443:         return os.path.split(path)
Line 444: 
Line 445:     def _teardownVolumes(self):


Line 451:             except Exception as e:
Line 452:                 logging.error('Error teardownVolumePath: %s', e)
Line 453: 
Line 454: 
Line 455: class OutputParser():
This is old style class - use OutputParser(object)
Line 456:     def __init__(self, importVm):
Line 457:         self._importVm = importVm
Line 458:         self._copy_disk_re = re.compile(r'.*(Copying disk (\d+)/(\d+)).*')
Line 459:         self._disk_progress_re = re.compile(r'\s+\((\d+).*')


Line 453: 
Line 454: 
Line 455: class OutputParser():
Line 456:     def __init__(self, importVm):
Line 457:         self._importVm = importVm
Lets remove this - keeping an object and calling it is not needed in this context.
Line 458:         self._copy_disk_re = re.compile(r'.*(Copying disk (\d+)/(\d+)).*')
Line 459:         self._disk_progress_re = re.compile(r'\s+\((\d+).*')
Line 460: 
Line 461:     def parse_line(self, line):


Line 455: class OutputParser():
Line 456:     def __init__(self, importVm):
Line 457:         self._importVm = importVm
Line 458:         self._copy_disk_re = re.compile(r'.*(Copying disk (\d+)/(\d+)).*')
Line 459:         self._disk_progress_re = re.compile(r'\s+\((\d+).*')
These should be class variables (like static member in java) - why compile the pattern again for each instance?

    class OutputParser(object):
        COPY_DISK_RE = re.compile(r'.*(Copying disk (\d+)/(\d+)).*')
        ...
Line 460: 
Line 461:     def parse_line(self, line):
Line 462:         if 'Copying disk' in line:
Line 463:             m = self._copy_disk_re.match(line)


Line 462:         if 'Copying disk' in line:
Line 463:             m = self._copy_disk_re.match(line)
Line 464:             if m is None:
Line 465:                 raise JobOutputParsing()
Line 466:             else:
else is unneeded
Line 467:                 self._importVm.disk_copy_stage(m.groups()[0],
Line 468:                                                m.groups()[1],
Line 469:                                                m.groups()[2])
Line 470: 


Line 465:                 raise JobOutputParsing()
Line 466:             else:
Line 467:                 self._importVm.disk_copy_stage(m.groups()[0],
Line 468:                                                m.groups()[1],
Line 469:                                                m.groups()[2])
Instead of this, return the values:

    return m.groups()
Line 470: 
Line 471:     def parse_progress(self, buf):
Line 472:         m = self._disk_progress_re.match(buf)
Line 473:         if m is None:


Line 471:     def parse_progress(self, buf):
Line 472:         m = self._disk_progress_re.match(buf)
Line 473:         if m is None:
Line 474:             raise JobOutputParsing()
Line 475:         else:
else is unneeded
Line 476:             self._importVm.progress_stage(m.groups()[0])
Line 477: 
Line 478: 
Line 479: def _mem_to_mib(size, unit):


Line 472:         m = self._disk_progress_re.match(buf)
Line 473:         if m is None:
Line 474:             raise JobOutputParsing()
Line 475:         else:
Line 476:             self._importVm.progress_stage(m.groups()[0])
Return the progress instead:

   return m.group(0)

Note: you should use group(index) instead of creating a list and indexing it.
Line 477: 
Line 478: 
Line 479: def _mem_to_mib(size, unit):
Line 480:     lunit = unit.lower()


-- 
To view, visit https://gerrit.ovirt.org/37509
To unsubscribe, visit https://gerrit.ovirt.org/settings

Gerrit-MessageType: comment
Gerrit-Change-Id: I34bd86d5a87ea8c42113c4a732f87ddd4ceab9ea
Gerrit-PatchSet: 10
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Shahar Havivi <shavivi at redhat.com>
Gerrit-Reviewer: Dan Kenigsberg <danken at redhat.com>
Gerrit-Reviewer: Federico Simoncelli <fsimonce at redhat.com>
Gerrit-Reviewer: Francesco Romani <fromani at redhat.com>
Gerrit-Reviewer: Michal Skrivanek <michal.skrivanek at redhat.com>
Gerrit-Reviewer: Nir Soffer <nsoffer at redhat.com>
Gerrit-Reviewer: Piotr Kliczewski <piotr.kliczewski at gmail.com>
Gerrit-Reviewer: Saggi Mizrahi <smizrahi at redhat.com>
Gerrit-Reviewer: Shahar Havivi <shavivi at redhat.com>
Gerrit-Reviewer: Yaniv Bronhaim <ybronhei at redhat.com>
Gerrit-Reviewer: automation at ovirt.org
Gerrit-Reviewer: oVirt Jenkins CI Server
Gerrit-HasComments: Yes


More information about the vdsm-patches mailing list