summaryrefslogtreecommitdiffstats
path: root/src/pybind/mgr/progress/module.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/pybind/mgr/progress/module.py')
-rw-r--r--src/pybind/mgr/progress/module.py882
1 files changed, 882 insertions, 0 deletions
diff --git a/src/pybind/mgr/progress/module.py b/src/pybind/mgr/progress/module.py
new file mode 100644
index 000000000..7c98200fa
--- /dev/null
+++ b/src/pybind/mgr/progress/module.py
@@ -0,0 +1,882 @@
+try:
+ from typing import List, Dict, Union, Any, Optional
+ from typing import TYPE_CHECKING
+except ImportError:
+ TYPE_CHECKING = False
+
+from mgr_module import MgrModule, OSDMap, Option
+from mgr_util import to_pretty_timedelta
+from datetime import timedelta
+import os
+import threading
+import datetime
+import uuid
+import time
+import logging
+import json
+
+
+ENCODING_VERSION = 2
+
+# keep a global reference to the module so we can use it from Event methods
+_module = None # type: Optional["Module"]
+
+
+class Event(object):
+ """
+ A generic "event" that has a start time, completion percentage,
+ and a list of "refs" that are (type, id) tuples describing which
+ objects (osds, pools) this relates to.
+ """
+ def __init__(self, id: str,
+ message: str,
+ refs: List[str],
+ add_to_ceph_s: bool,
+ started_at: Optional[float] = None):
+ self._message = message
+ self._refs = refs
+ self.started_at = started_at if started_at else time.time()
+ self.id = id
+ self._add_to_ceph_s = add_to_ceph_s
+
+ def _refresh(self):
+ global _module
+ assert _module
+ _module.log.debug('refreshing mgr for %s (%s) at %f' % (self.id, self._message,
+ self.progress))
+ _module.update_progress_event(
+ self.id, self.twoline_progress(6), self.progress, self._add_to_ceph_s)
+
+ @property
+ def message(self):
+ # type: () -> str
+ return self._message
+
+ @property
+ def refs(self):
+ # type: () -> List[str]
+ return self._refs
+
+ @property
+ def add_to_ceph_s(self):
+ # type: () -> bool
+ return self._add_to_ceph_s
+
+ @property
+ def progress(self):
+ # type: () -> float
+ raise NotImplementedError()
+
+ @property
+ def duration_str(self):
+ duration = time.time() - self.started_at
+ return "(%s)" % (
+ to_pretty_timedelta(timedelta(seconds=duration)))
+
+ @property
+ def failed(self):
+ return False
+
+ @property
+ def failure_message(self):
+ return None
+
+ def summary(self):
+ # type: () -> str
+ return "{0} {1} {2}".format(self.progress, self.message,
+ self.duration_str)
+
+ def _progress_str(self, width):
+ inner_width = width - 2
+ out = "["
+ done_chars = int(self.progress * inner_width)
+ out += done_chars * '='
+ out += (inner_width - done_chars) * '.'
+ out += "]"
+
+ return out
+
+ def twoline_progress(self, indent=4):
+ """
+ e.g.
+
+ - Eating my delicious strudel (since: 30s)
+ [===============..............] (remaining: 04m)
+
+ """
+ time_remaining = self.estimated_time_remaining()
+ if time_remaining:
+ remaining = "(remaining: %s)" % (
+ to_pretty_timedelta(timedelta(seconds=time_remaining)))
+ else:
+ remaining = ''
+ return "{0} {1}\n{2}{3} {4}".format(self._message,
+ self.duration_str,
+ " " * indent,
+ self._progress_str(30),
+ remaining)
+
+ def to_json(self):
+ # type: () -> Dict[str, Any]
+ return {
+ "id": self.id,
+ "message": self.message,
+ "duration": self.duration_str,
+ "refs": self._refs,
+ "progress": self.progress,
+ "started_at": self.started_at,
+ "time_remaining": self.estimated_time_remaining()
+ }
+
+ def estimated_time_remaining(self):
+ elapsed = time.time() - self.started_at
+ progress = self.progress
+ if progress == 0.0:
+ return None
+ return int(elapsed * (1 - progress) / progress)
+
+
+class GhostEvent(Event):
+ """
+ The ghost of a completed event: these are the fields that we persist
+ after the event is complete.
+ """
+
+ def __init__(self, my_id, message, refs, add_to_ceph_s, started_at, finished_at=None,
+ failed=False, failure_message=None):
+ super().__init__(my_id, message, refs, add_to_ceph_s, started_at)
+ self.finished_at = finished_at if finished_at else time.time()
+
+ if failed:
+ self._failed = True
+ self._failure_message = failure_message
+ else:
+ self._failed = False
+
+ @property
+ def progress(self):
+ return 1.0
+
+ @property
+ def failed(self):
+ return self._failed
+
+ @property
+ def failure_message(self):
+ return self._failure_message if self._failed else None
+
+ def to_json(self):
+ d = {
+ "id": self.id,
+ "message": self.message,
+ "refs": self._refs,
+ "started_at": self.started_at,
+ "finished_at": self.finished_at,
+ "add_to_ceph_s:": self.add_to_ceph_s
+ }
+ if self._failed:
+ d["failed"] = True
+ d["failure_message"] = self._failure_message
+ return d
+
+
+class GlobalRecoveryEvent(Event):
+ """
+ An event whoese completion is determined by active+clean/total_pg_num
+ """
+
+ def __init__(self, message, refs, add_to_ceph_s, start_epoch, active_clean_num):
+ # type: (str, List[Any], bool, int, int) -> None
+ super().__init__(str(uuid.uuid4()), message, refs, add_to_ceph_s)
+ self._add_to_ceph_s = add_to_ceph_s
+ self._progress = 0.0
+ self._start_epoch = start_epoch
+ self._active_clean_num = active_clean_num
+ self._refresh()
+
+ def global_event_update_progress(self, log):
+ # type: (logging.Logger) -> None
+ "Update progress of Global Recovery Event"
+ global _module
+ assert _module
+ skipped_pgs = 0
+ active_clean_pgs = _module.get("active_clean_pgs")
+ total_pg_num = active_clean_pgs["total_num_pgs"]
+ new_active_clean_pgs = active_clean_pgs["pg_stats"]
+ new_active_clean_num = len(new_active_clean_pgs)
+ for pg in new_active_clean_pgs:
+ # Disregard PGs that are not being reported
+ # if the states are active+clean. Since it is
+ # possible that some pgs might not have any movement
+ # even before the start of the event.
+ if pg['reported_epoch'] < self._start_epoch:
+ log.debug("Skipping pg {0} since reported_epoch {1} < start_epoch {2}"
+ .format(pg['pgid'], pg['reported_epoch'], self._start_epoch))
+ skipped_pgs += 1
+ continue
+
+ if self._active_clean_num != new_active_clean_num:
+ # Have this case to know when need to update
+ # the progress
+ try:
+ # Might be that total_pg_num is 0
+ self._progress = float(new_active_clean_num) / (total_pg_num - skipped_pgs)
+ except ZeroDivisionError:
+ self._progress = 0.0
+ else:
+ # No need to update since there is no change
+ return
+
+ log.debug("Updated progress to %s", self.summary())
+ self._refresh()
+
+ @property
+ def progress(self):
+ return self._progress
+
+
+class RemoteEvent(Event):
+ """
+ An event that was published by another module: we know nothing about
+ this, rely on the other module to continuously update us with
+ progress information as it emerges.
+ """
+
+ def __init__(self, my_id, message, refs, add_to_ceph_s):
+ # type: (str, str, List[str], bool) -> None
+ super().__init__(my_id, message, refs, add_to_ceph_s)
+ self._progress = 0.0
+ self._failed = False
+ self._refresh()
+
+ def set_progress(self, progress):
+ # type: (float) -> None
+ self._progress = progress
+ self._refresh()
+
+ def set_failed(self, message):
+ self._progress = 1.0
+ self._failed = True
+ self._failure_message = message
+ self._refresh()
+
+ def set_message(self, message):
+ self._message = message
+ self._refresh()
+
+ @property
+ def progress(self):
+ return self._progress
+
+ @property
+ def failed(self):
+ return self._failed
+
+ @property
+ def failure_message(self):
+ return self._failure_message if self._failed else None
+
+
+class PgRecoveryEvent(Event):
+ """
+ An event whose completion is determined by the recovery of a set of
+ PGs to a healthy state.
+
+ Always call update() immediately after construction.
+ """
+
+ def __init__(self, message, refs, which_pgs, which_osds, start_epoch, add_to_ceph_s):
+ # type: (str, List[Any], List[PgId], List[str], int, bool) -> None
+ super().__init__(str(uuid.uuid4()), message, refs, add_to_ceph_s)
+ self._pgs = which_pgs
+ self._which_osds = which_osds
+ self._original_pg_count = len(self._pgs)
+ self._original_bytes_recovered = None # type: Optional[Dict[PgId, float]]
+ self._progress = 0.0
+
+ self._start_epoch = start_epoch
+ self._refresh()
+
+ @property
+ def which_osds(self):
+ return self. _which_osds
+
+ def pg_update(self, pg_progress: Dict, log: Any) -> None:
+ # FIXME: O(pg_num) in python
+ # Sanity check to see if there are any missing PGs and to assign
+ # empty array and dictionary if there hasn't been any recovery
+ pg_to_state: Dict[str, Any] = pg_progress["pgs"]
+ pg_ready: bool = pg_progress["pg_ready"]
+
+ if self._original_bytes_recovered is None:
+ self._original_bytes_recovered = {}
+ missing_pgs = []
+ for pg in self._pgs:
+ pg_str = str(pg)
+ if pg_str in pg_to_state:
+ self._original_bytes_recovered[pg] = \
+ pg_to_state[pg_str]['num_bytes_recovered']
+ else:
+ missing_pgs.append(pg)
+ if pg_ready:
+ for pg in missing_pgs:
+ self._pgs.remove(pg)
+
+ complete_accumulate = 0.0
+
+ # Calculating progress as the number of PGs recovered divided by the
+ # original where partially completed PGs count for something
+ # between 0.0-1.0. This is perhaps less faithful than looking at the
+ # total number of bytes recovered, but it does a better job of
+ # representing the work still to do if there are a number of very
+ # few-bytes PGs that still need the housekeeping of their recovery
+ # to be done. This is subjective...
+
+ complete = set()
+ for pg in self._pgs:
+ pg_str = str(pg)
+ try:
+ info = pg_to_state[pg_str]
+ except KeyError:
+ # The PG is gone! Probably a pool was deleted. Drop it.
+ complete.add(pg)
+ continue
+ # Only checks the state of each PGs when it's epoch >= the OSDMap's epoch
+ if info['reported_epoch'] < self._start_epoch:
+ continue
+
+ state = info['state']
+
+ states = state.split("+")
+
+ if "active" in states and "clean" in states:
+ complete.add(pg)
+ else:
+ if info['num_bytes'] == 0:
+ # Empty PGs are considered 0% done until they are
+ # in the correct state.
+ pass
+ else:
+ recovered = info['num_bytes_recovered']
+ total_bytes = info['num_bytes']
+ if total_bytes > 0:
+ ratio = float(recovered -
+ self._original_bytes_recovered[pg]) / \
+ total_bytes
+ # Since the recovered bytes (over time) could perhaps
+ # exceed the contents of the PG (moment in time), we
+ # must clamp this
+ ratio = min(ratio, 1.0)
+ ratio = max(ratio, 0.0)
+
+ else:
+ # Dataless PGs (e.g. containing only OMAPs) count
+ # as half done.
+ ratio = 0.5
+ complete_accumulate += ratio
+
+ self._pgs = list(set(self._pgs) ^ complete)
+ completed_pgs = self._original_pg_count - len(self._pgs)
+ completed_pgs = max(completed_pgs, 0)
+ try:
+ prog = (completed_pgs + complete_accumulate)\
+ / self._original_pg_count
+ except ZeroDivisionError:
+ prog = 0.0
+
+ self._progress = min(max(prog, 0.0), 1.0)
+
+ self._refresh()
+ log.info("Updated progress to %s", self.summary())
+
+ @property
+ def progress(self):
+ # type: () -> float
+ return self._progress
+
+
+class PgId(object):
+ def __init__(self, pool_id, ps):
+ # type: (str, int) -> None
+ self.pool_id = pool_id
+ self.ps = ps
+
+ def __cmp__(self, other):
+ return (self.pool_id, self.ps) == (other.pool_id, other.ps)
+
+ def __lt__(self, other):
+ return (self.pool_id, self.ps) < (other.pool_id, other.ps)
+
+ def __str__(self):
+ return "{0}.{1:x}".format(self.pool_id, self.ps)
+
+
+class Module(MgrModule):
+ COMMANDS = [
+ {"cmd": "progress",
+ "desc": "Show progress of recovery operations",
+ "perm": "r"},
+ {"cmd": "progress json",
+ "desc": "Show machine readable progress information",
+ "perm": "r"},
+ {"cmd": "progress clear",
+ "desc": "Reset progress tracking",
+ "perm": "rw"},
+ {"cmd": "progress on",
+ "desc": "Enable progress tracking",
+ "perm": "rw"},
+ {"cmd": "progress off",
+ "desc": "Disable progress tracking",
+ "perm": "rw"}
+
+ ]
+
+ MODULE_OPTIONS = [
+ Option(
+ 'max_completed_events',
+ default=50,
+ type='int',
+ desc='number of past completed events to remember',
+ runtime=True
+ ),
+ Option(
+ 'sleep_interval',
+ default=5,
+ type='secs',
+ desc='how long the module is going to sleep',
+ runtime=True
+ ),
+ Option(
+ 'enabled',
+ default=True,
+ type='bool',
+ ),
+ Option(
+ 'allow_pg_recovery_event',
+ default=False,
+ type='bool',
+ desc='allow the module to show pg recovery progress',
+ runtime=True
+ )
+ ]
+
+ def __init__(self, *args, **kwargs):
+ super(Module, self).__init__(*args, **kwargs)
+
+ self._events = {} # type: Dict[str, Union[RemoteEvent, PgRecoveryEvent, GlobalRecoveryEvent]]
+ self._completed_events = [] # type: List[GhostEvent]
+
+ self._old_osd_map = None # type: Optional[OSDMap]
+
+ self._ready = threading.Event()
+ self._shutdown = threading.Event()
+
+ self._latest_osdmap = None # type: Optional[OSDMap]
+
+ self._dirty = False
+
+ global _module
+ _module = self
+
+ # only for mypy
+ if TYPE_CHECKING:
+ self.max_completed_events = 0
+ self.sleep_interval = 0
+ self.enabled = True
+ self.allow_pg_recovery_event = False
+
+ def config_notify(self):
+ for opt in self.MODULE_OPTIONS:
+ setattr(self,
+ opt['name'],
+ self.get_module_option(opt['name']))
+ self.log.debug(' %s = %s', opt['name'], getattr(self, opt['name']))
+
+ def _osd_in_out(self, old_map, old_dump, new_map, osd_id, marked):
+ # type: (OSDMap, Dict, OSDMap, str, str) -> None
+ # A function that will create or complete an event when an
+ # OSD is marked in or out according to the affected PGs
+ affected_pgs = []
+ for pool in old_dump['pools']:
+ pool_id = pool['pool'] # type: str
+ for ps in range(0, pool['pg_num']):
+
+ # Was this OSD affected by the OSD coming in/out?
+ # Compare old and new osds using
+ # data from the json dump
+ old_up_acting = old_map.pg_to_up_acting_osds(pool['pool'], ps)
+ old_osds = set(old_up_acting['acting'])
+ new_up_acting = new_map.pg_to_up_acting_osds(pool['pool'], ps)
+ new_osds = set(new_up_acting['acting'])
+
+ # Check the osd_id being in the acting set for both old
+ # and new maps to cover both out and in cases
+ was_on_out_or_in_osd = osd_id in old_osds or osd_id in new_osds
+ if not was_on_out_or_in_osd:
+ continue
+
+ self.log.debug("pool_id, ps = {0}, {1}".format(
+ pool_id, ps
+ ))
+
+ self.log.debug(
+ "old_up_acting: {0}".format(json.dumps(old_up_acting, indent=4, sort_keys=True)))
+
+ # Has this OSD been assigned a new location?
+ # (it might not be if there is no suitable place to move
+ # after an OSD is marked in/out)
+
+ is_relocated = old_osds != new_osds
+
+ self.log.debug(
+ "new_up_acting: {0}".format(json.dumps(new_up_acting,
+ indent=4,
+ sort_keys=True)))
+
+ if was_on_out_or_in_osd and is_relocated:
+ # This PG is now in motion, track its progress
+ affected_pgs.append(PgId(pool_id, ps))
+
+ # In the case that we ignored some PGs, log the reason why (we may
+ # not end up creating a progress event)
+
+ self.log.warning("{0} PGs affected by osd.{1} being marked {2}".format(
+ len(affected_pgs), osd_id, marked))
+
+ # In the case of the osd coming back in, we might need to cancel
+ # previous recovery event for that osd
+ if marked == "in":
+ for ev_id in list(self._events):
+ try:
+ ev = self._events[ev_id]
+ if isinstance(ev, PgRecoveryEvent) and osd_id in ev.which_osds:
+ self.log.info("osd.{0} came back in, cancelling event".format(
+ osd_id
+ ))
+ self._complete(ev)
+ except KeyError:
+ self.log.warning("_osd_in_out: ev {0} does not exist".format(ev_id))
+
+ if len(affected_pgs) > 0:
+ r_ev = PgRecoveryEvent(
+ "Rebalancing after osd.{0} marked {1}".format(osd_id, marked),
+ refs=[("osd", osd_id)],
+ which_pgs=affected_pgs,
+ which_osds=[osd_id],
+ start_epoch=self.get_osdmap().get_epoch(),
+ add_to_ceph_s=False
+ )
+ r_ev.pg_update(self.get("pg_progress"), self.log)
+ self._events[r_ev.id] = r_ev
+
+ def _osdmap_changed(self, old_osdmap, new_osdmap):
+ # type: (OSDMap, OSDMap) -> None
+ old_dump = old_osdmap.dump()
+ new_dump = new_osdmap.dump()
+
+ old_osds = dict([(o['osd'], o) for o in old_dump['osds']])
+
+ for osd in new_dump['osds']:
+ osd_id = osd['osd']
+ new_weight = osd['in']
+ if osd_id in old_osds:
+ old_weight = old_osds[osd_id]['in']
+
+ if new_weight == 0.0 and old_weight > new_weight:
+ self.log.warning("osd.{0} marked out".format(osd_id))
+ self._osd_in_out(old_osdmap, old_dump, new_osdmap, osd_id, "out")
+ elif new_weight >= 1.0 and old_weight == 0.0:
+ # Only consider weight>=1.0 as "in" to avoid spawning
+ # individual recovery events on every adjustment
+ # in a gradual weight-in
+ self.log.warning("osd.{0} marked in".format(osd_id))
+ self._osd_in_out(old_osdmap, old_dump, new_osdmap, osd_id, "in")
+
+ def _pg_state_changed(self):
+
+ # This function both constructs and updates
+ # the global recovery event if one of the
+ # PGs is not at active+clean state
+ active_clean_pgs = self.get("active_clean_pgs")
+ total_pg_num = active_clean_pgs["total_num_pgs"]
+ active_clean_num = len(active_clean_pgs["pg_stats"])
+ try:
+ # There might be a case where there is no pg_num
+ progress = float(active_clean_num) / total_pg_num
+ except ZeroDivisionError:
+ return
+ if progress < 1.0:
+ self.log.warning(("Starting Global Recovery Event,"
+ "%d pgs not in active + clean state"),
+ total_pg_num - active_clean_num)
+ ev = GlobalRecoveryEvent("Global Recovery Event",
+ refs=[("global", "")],
+ add_to_ceph_s=True,
+ start_epoch=self.get_osdmap().get_epoch(),
+ active_clean_num=active_clean_num)
+ ev.global_event_update_progress(self.log)
+ self._events[ev.id] = ev
+
+ def _process_osdmap(self):
+ old_osdmap = self._latest_osdmap
+ self._latest_osdmap = self.get_osdmap()
+ assert old_osdmap
+ assert self._latest_osdmap
+ self.log.info(("Processing OSDMap change %d..%d"),
+ old_osdmap.get_epoch(), self._latest_osdmap.get_epoch())
+
+ self._osdmap_changed(old_osdmap, self._latest_osdmap)
+
+ def _process_pg_summary(self):
+ # if there are no events we will skip this here to avoid
+ # expensive get calls
+ if len(self._events) == 0:
+ return
+
+ global_event = False
+ data = self.get("pg_progress")
+ for ev_id in list(self._events):
+ try:
+ ev = self._events[ev_id]
+ # Check for types of events
+ # we have to update
+ if isinstance(ev, PgRecoveryEvent):
+ ev.pg_update(data, self.log)
+ self.maybe_complete(ev)
+ elif isinstance(ev, GlobalRecoveryEvent):
+ global_event = True
+ ev.global_event_update_progress(self.log)
+ self.maybe_complete(ev)
+ except KeyError:
+ self.log.warning("_process_pg_summary: ev {0} does not exist".format(ev_id))
+ continue
+
+ if not global_event:
+ # If there is no global event
+ # we create one
+ self._pg_state_changed()
+
+ def maybe_complete(self, event):
+ # type: (Event) -> None
+ if event.progress >= 1.0:
+ self._complete(event)
+
+ def _save(self):
+ self.log.info("Writing back {0} completed events".format(
+ len(self._completed_events)
+ ))
+ # TODO: bound the number we store.
+ encoded = json.dumps({
+ "events": [ev.to_json() for ev in self._completed_events],
+ "version": ENCODING_VERSION,
+ "compat_version": ENCODING_VERSION
+ })
+ self.set_store("completed", encoded)
+
+ def _load(self):
+ stored = self.get_store("completed")
+
+ if stored is None:
+ self.log.info("No stored events to load")
+ return
+
+ decoded = json.loads(stored)
+ if decoded['compat_version'] > ENCODING_VERSION:
+ raise RuntimeError("Cannot decode version {0}".format(
+ decoded['compat_version']))
+
+ if decoded['compat_version'] < ENCODING_VERSION:
+ # we need to add the "started_at" and "finished_at" attributes to the events
+ for ev in decoded['events']:
+ ev['started_at'] = None
+ ev['finished_at'] = None
+
+ for ev in decoded['events']:
+ self._completed_events.append(GhostEvent(ev['id'], ev['message'],
+ ev['refs'], ev['started_at'],
+ ev['finished_at'],
+ ev.get('failed', False),
+ ev.get('failure_message')))
+
+ self._prune_completed_events()
+
+ def _prune_completed_events(self):
+ length = len(self._completed_events)
+ if length > self.max_completed_events:
+ self._completed_events = self._completed_events[length - self.max_completed_events : length]
+ self._dirty = True
+
+ def serve(self):
+ self.config_notify()
+ self.clear_all_progress_events()
+ self.log.info("Loading...")
+
+ self._load()
+ self.log.info("Loaded {0} historic events".format(self._completed_events))
+
+ self._latest_osdmap = self.get_osdmap()
+ self.log.info("Loaded OSDMap, ready.")
+
+ self._ready.set()
+
+ while not self._shutdown.is_set():
+ # Lazy periodic write back of completed events
+ if self._dirty:
+ self._save()
+ self._dirty = False
+
+ if self.enabled:
+ if self.allow_pg_recovery_event:
+ self._process_osdmap()
+ self._process_pg_summary()
+
+ self._shutdown.wait(timeout=self.sleep_interval)
+
+ self._shutdown.wait()
+
+ def shutdown(self):
+ self._shutdown.set()
+ self.clear_all_progress_events()
+
+ def update(self, ev_id, ev_msg, ev_progress, refs=None, add_to_ceph_s=False):
+ # type: (str, str, float, Optional[list], bool) -> None
+ """
+ For calling from other mgr modules
+ """
+ if not self.enabled:
+ return
+
+ if refs is None:
+ refs = []
+ try:
+ ev = self._events[ev_id]
+ assert isinstance(ev, RemoteEvent)
+ except KeyError:
+ # if key doesn't exist we create an event
+ ev = RemoteEvent(ev_id, ev_msg, refs, add_to_ceph_s)
+ self._events[ev_id] = ev
+ self.log.info("update: starting ev {0} ({1})".format(
+ ev_id, ev_msg))
+ else:
+ self.log.debug("update: {0} on {1}".format(
+ ev_progress, ev_msg))
+
+ ev.set_progress(ev_progress)
+ ev.set_message(ev_msg)
+
+ def _complete(self, ev):
+ # type: (Event) -> None
+ duration = (time.time() - ev.started_at)
+ self.log.info("Completed event {0} ({1}) in {2} seconds".format(
+ ev.id, ev.message, int(round(duration))
+ ))
+ self.complete_progress_event(ev.id)
+
+ self._completed_events.append(
+ GhostEvent(ev.id, ev.message, ev.refs, ev.add_to_ceph_s, ev.started_at,
+ failed=ev.failed, failure_message=ev.failure_message))
+ assert ev.id
+ del self._events[ev.id]
+ self._prune_completed_events()
+ self._dirty = True
+
+ def complete(self, ev_id):
+ """
+ For calling from other mgr modules
+ """
+ if not self.enabled:
+ return
+ try:
+ ev = self._events[ev_id]
+ assert isinstance(ev, RemoteEvent)
+ ev.set_progress(1.0)
+ self.log.info("complete: finished ev {0} ({1})".format(ev_id,
+ ev.message))
+ self._complete(ev)
+ except KeyError:
+ self.log.warning("complete: ev {0} does not exist".format(ev_id))
+ pass
+
+ def fail(self, ev_id, message):
+ """
+ For calling from other mgr modules to mark an event as failed (and
+ complete)
+ """
+ try:
+ ev = self._events[ev_id]
+ assert isinstance(ev, RemoteEvent)
+ ev.set_failed(message)
+ self.log.info("fail: finished ev {0} ({1}): {2}".format(ev_id,
+ ev.message,
+ message))
+ self._complete(ev)
+ except KeyError:
+ self.log.warning("fail: ev {0} does not exist".format(ev_id))
+
+ def on(self):
+ self.set_module_option('enabled', "true")
+
+ def off(self):
+ self.set_module_option('enabled', "false")
+
+ def _handle_ls(self):
+ if len(self._events) or len(self._completed_events):
+ out = ""
+ chrono_order = sorted(self._events.values(),
+ key=lambda x: x.started_at, reverse=True)
+ for ev in chrono_order:
+ out += ev.twoline_progress()
+ out += "\n"
+
+ if len(self._completed_events):
+ # TODO: limit number of completed events to show
+ out += "\n"
+ for ghost_ev in self._completed_events:
+ out += "[{0}]: {1}\n".format("Complete" if not ghost_ev.failed else "Failed",
+ ghost_ev.twoline_progress())
+
+ return 0, out, ""
+ else:
+ return 0, "", "Nothing in progress"
+
+ def _json(self):
+ return {
+ 'events': [ev.to_json() for ev in self._events.values()],
+ 'completed': [ev.to_json() for ev in self._completed_events]
+ }
+
+ def clear(self):
+ self._events = {}
+ self._completed_events = []
+ self._dirty = True
+ self._save()
+ self.clear_all_progress_events()
+
+ def _handle_clear(self):
+ self.clear()
+ return 0, "", ""
+
+ def handle_command(self, _, cmd):
+ if cmd['prefix'] == "progress":
+ return self._handle_ls()
+ elif cmd['prefix'] == "progress clear":
+ # The clear command isn't usually needed - it's to enable
+ # the admin to "kick" this module if it seems to have done
+ # something wrong (e.g. we have a bug causing a progress event
+ # that never finishes)
+ return self._handle_clear()
+ elif cmd['prefix'] == "progress json":
+ return 0, json.dumps(self._json(), indent=4, sort_keys=True), ""
+ elif cmd['prefix'] == "progress on":
+ if self.enabled:
+ return 0, "", "progress already enabled!"
+ self.on()
+ return 0, "", "progress enabled"
+ elif cmd['prefix'] == "progress off":
+ if not self.enabled:
+ return 0, "", "progress already disabled!"
+ self.off()
+ self.clear()
+ return 0, "", "progress disabled"
+ else:
+ raise NotImplementedError(cmd['prefix'])