Merge "Track known repair scripts and their routing keys"
This commit is contained in:
commit
99a921150b
|
@ -66,6 +66,9 @@ class Engine(object):
|
||||||
self.run_queue = collections.deque()
|
self.run_queue = collections.deque()
|
||||||
# Private variables
|
# Private variables
|
||||||
self._watchdog_event_fn = {self.repair_cfg: self.repair_modified}
|
self._watchdog_event_fn = {self.repair_cfg: self.repair_modified}
|
||||||
|
# Private variables to keep track of repair scripts.
|
||||||
|
self._repairs = []
|
||||||
|
self._known_routing_keys = set()
|
||||||
LOG.info('Created engine obj %s', self.name)
|
LOG.info('Created engine obj %s', self.name)
|
||||||
|
|
||||||
# TODO(praneshp): Move to utils?
|
# TODO(praneshp): Move to utils?
|
||||||
|
@ -183,6 +186,7 @@ class Engine(object):
|
||||||
new_additions.sort(key=operator.itemgetter('time'))
|
new_additions.sort(key=operator.itemgetter('time'))
|
||||||
self.run_queue.extend(new_additions)
|
self.run_queue.extend(new_additions)
|
||||||
LOG.info("Run queue till %s is %s", next_iteration, self.run_queue)
|
LOG.info("Run queue till %s is %s", next_iteration, self.run_queue)
|
||||||
|
LOG.info("Repair scripts at %s: %s", next_iteration, self._repairs)
|
||||||
except Exception:
|
except Exception:
|
||||||
LOG.exception("Could not run serializer for %s at %s",
|
LOG.exception("Could not run serializer for %s at %s",
|
||||||
self.name, current_time)
|
self.name, current_time)
|
||||||
|
@ -244,6 +248,7 @@ class Engine(object):
|
||||||
data['routing_key'])
|
data['routing_key'])
|
||||||
if message_queue not in self.known_queues:
|
if message_queue not in self.known_queues:
|
||||||
self.known_queues.append(message_queue)
|
self.known_queues.append(message_queue)
|
||||||
|
self._known_routing_keys.add(data['routing_key'])
|
||||||
kwargs = data
|
kwargs = data
|
||||||
kwargs['conf'] = script_args['cfg']
|
kwargs['conf'] = script_args['cfg']
|
||||||
kwargs['exchange'] = self.entropy_exchange
|
kwargs['exchange'] = self.entropy_exchange
|
||||||
|
@ -252,6 +257,7 @@ class Engine(object):
|
||||||
self.running_repairs.append(script)
|
self.running_repairs.append(script)
|
||||||
imported_module = imp.load_module(react_script, *available_modules)
|
imported_module = imp.load_module(react_script, *available_modules)
|
||||||
future = self.executor.submit(imported_module.main, **kwargs)
|
future = self.executor.submit(imported_module.main, **kwargs)
|
||||||
|
self._repairs.append(future)
|
||||||
return future
|
return future
|
||||||
except Exception:
|
except Exception:
|
||||||
LOG.exception("Could not setup %s", script)
|
LOG.exception("Could not setup %s", script)
|
||||||
|
|
Loading…
Reference in New Issue