Adds percentage progress update

coriolis/db/api.py

@@ -302,12 +302,22 @@ def add_task_event(context, task_id, level, message):
     context.session.add(task_event)
 
 
+def _get_progress_update(context, task_id, current_step):
+    q = _soft_delete_aware_query(context, models.TaskProgressUpdate)
+    return q.filter(
+        models.TaskProgressUpdate.task_id == task_id,
+        models.TaskProgressUpdate.current_step == current_step).first()
+
+
 @enginefacade.writer
 def add_task_progress_update(context, task_id, current_step, total_steps,
                              message):
-    task_progress_update = models.TaskProgressUpdate()
+    task_progress_update = _get_progress_update(context, task_id, current_step)
+    if not task_progress_update:
+        task_progress_update = models.TaskProgressUpdate()
+        context.session.add(task_progress_update)
+
     task_progress_update.task_id = task_id
     task_progress_update.current_step = current_step
     task_progress_update.total_steps = total_steps
     task_progress_update.message = message
-    context.session.add(task_progress_update)

coriolis/events.py

@@ -2,6 +2,11 @@
 # All Rights Reserved.
 
 import abc
+import collections
+
+
+_PercStepData = collections.namedtuple(
+    "_PercStepData", "last_value max_value perc_threshold message_format")
 
 
 class EventManager(object):
@@ -11,10 +16,33 @@ class EventManager(object):
         self._event_handler = event_handler
         self._current_step = 0
         self._total_steps = None
+        self._percentage_steps = {}
 
     def set_total_progress_steps(self, total_steps):
         self._total_steps = total_steps
 
+    def add_percentage_step(self, max_value, perc_threshold=1,
+                            message_format="{:.0f}%"):
+        self._current_step += 1
+        self._percentage_steps[self._current_step] = _PercStepData(
+            0, max_value, perc_threshold, message_format)
+        return self._current_step
+
+    def set_percentage_step(self, step, value):
+        step_data = self._percentage_steps[step]
+
+        old_perc = (step_data.last_value * 100 / step_data.max_value //
+                    step_data.perc_threshold * step_data.perc_threshold)
+        perc = (value * 100 / step_data.max_value //
+                step_data.perc_threshold * step_data.perc_threshold)
+
+        if perc > old_perc and self._event_handler:
+            self._event_handler.progress_update(
+                step, self._total_steps, step_data.message_format.format(perc))
+            self._percentage_steps[step] = _PercStepData(
+                value, step_data.max_value, step_data.perc_threshold,
+                step_data.message_format)
+
     def progress_update(self, message):
         self._current_step += 1
         if self._event_handler:

coriolis/providers/vmware_vsphere/__init__.py

@@ -15,6 +15,7 @@ import uuid
 import eventlet
 from oslo_config import cfg
 from oslo_log import log as logging
+from oslo_utils import units
 import paramiko
 from pyVim import connect
 from pyVmomi import vim
@@ -574,14 +575,26 @@ class ExportProvider(base.BaseReplicaExportProvider):
 
                 backup_disk_path = disk.backing.fileName
 
+                disk_size = changed_disk_areas.length
+                changed_area_size = sum(
+                    [x.length for x in changed_disk_areas.changedArea])
+
                 if change_id == '*':
                     self._event_manager.progress_update(
-                        "Performing full CBT replica for disk: %s" %
-                        backup_disk_path)
+                        "Performing full CBT replica for disk: {path}. "
+                        "Disk size: {disk_size:,}, written area size: "
+                        "{changed_area_size:,}".format(
+                            path=backup_disk_path,
+                            disk_size=disk_size,
+                            changed_area_size=changed_area_size))
                 else:
                     self._event_manager.progress_update(
-                        "Performing incremental CBT replica for disk: %s" %
-                        backup_disk_path)
+                        "Performing incremental CBT replica for disk: {path}."
+                        "Disk size: {disk_size:,}, changed area size: "
+                        "{changed_area_size:,}".format(
+                            path=backup_disk_path,
+                            disk_size=disk_size,
+                            changed_area_size=changed_area_size))
 
                 with vixdisklib.open(
                         conn, backup_disk_path) as disk_handle:
@@ -590,6 +603,12 @@ class ExportProvider(base.BaseReplicaExportProvider):
                         # Create a sparse file
                         f.truncate(disk.capacityInBytes)
 
+                        total_written_bytes = 0
+                        perc_step = self._event_manager.add_percentage_step(
+                            changed_area_size,
+                            message_format="Disk %s replica progress: "
+                            "{:.0f}%%" % backup_disk_path)
+
                         for area in changed_disk_areas.changedArea:
                             start_sector = area.start // sector_size
                             num_sectors = area.length // sector_size
@@ -601,8 +620,8 @@ class ExportProvider(base.BaseReplicaExportProvider):
                                 curr_num_sectors = min(
                                     num_sectors - i, max_sectors_per_read)
 
-                                buf = vixdisklib.get_buffer(
-                                    curr_num_sectors * sector_size)
+                                buf_size = curr_num_sectors * sector_size
+                                buf = vixdisklib.get_buffer(buf_size)
 
                                 LOG.debug(
                                     "Read start sector: %s, num sectors: %s" %
@@ -615,6 +634,10 @@ class ExportProvider(base.BaseReplicaExportProvider):
 
                                 f.write(buf.raw)
 
+                                total_written_bytes += buf_size
+                                self._event_manager.set_percentage_step(
+                                    perc_step, total_written_bytes)
+
                 disk_path['change_id'] = disk.backing.changeId
 
     def _backup_disks(self, vm, export_path, connection_info, context):