import commands
import threading
import json
+import pdb
from datetime import datetime
from collections import defaultdict
self.enabled = True
self.dependency_graph = None
+STEP_STATUS_WORKING=1
+STEP_STATUS_OK=2
+STEP_STATUS_KO=3
+
+def invert_graph(g):
+ ig = {}
+ for k,v in g.items():
+ for v0 in v:
+ try:
+ ig[v0].append(k)
+ except:
+ ig=[k]
+ return ig
+
class PlanetStackObserver:
- #sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivileges,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
+ #sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivilege,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
sync_steps = []
- STEP_STATUS_WORKING=1
- STEP_STATUS_OK=2
- STEP_STATUS_KO=3
-
+
def __init__(self):
# The Condition object that gets signalled by Feefie events
self.step_lookup = {}
else:
self.driver = NoOpDriver()
- def wait_for_event(self, timeout, cond=self.event_cond):
- cond.acquire()
- cond.wait(timeout)
- cond.release()
+ def wait_for_event(self, timeout):
+ self.event_cond.acquire()
+ self.event_cond.wait(timeout)
+ self.event_cond.release()
- def wake_up(self, cond=self.event_cond):
+ def wake_up(self):
logger.info('Wake up routine called. Event cond %r'%self.event_cond)
- cond.acquire()
- cond.notify()
- cond.release()
+ self.event_cond.acquire()
+ self.event_cond.notify()
+ self.event_cond.release()
def load_sync_step_modules(self, step_dir=None):
if step_dir is None:
# no dependencies, pass
self.dependency_graph = step_graph
+ self.deletion_dependency_graph = invert_graph(step_graph)
self.ordered_steps = toposort(self.dependency_graph, map(lambda s:s.__name__,self.sync_steps))
print "Order of steps=",self.ordered_steps
self.last_deletion_run_times[e]=0
-
def save_run_times(self):
run_times = json.dumps(self.last_run_times)
open('/tmp/observer_run_times','w').write(run_times)
step = self.step_lookup[S]
start_time=time.time()
+ dependency_graph = self.dependency_graph if not deletion else self.deletion_dependency_graph
+
# Wait for step dependencies to be met
- deps = self.dependency_graph[S]
- for d in deps:
- cond = self.step_conditions[d]
- acquire(cond)
- if (self.step_status is STEP_STATUS_WORKING):
- cond.wait()
- cond.release()
-
- sync_step = step(driver=self.driver,error_map=error_mapper)
- sync_step.__name__ = step.__name__
- sync_step.dependencies = []
try:
- mlist = sync_step.provides
-
- for m in mlist:
- sync_step.dependencies.extend(self.model_dependency_graph[m.__name__])
+ deps = self.dependency_graph[S]
+ has_deps = True
except KeyError:
- pass
- sync_step.debug_mode = debug_mode
+ has_deps = False
+
+ if (has_deps):
+ for d in deps:
+ cond = self.step_conditions[d]
+ cond.acquire()
+ if (self.step_status[d] is STEP_STATUS_WORKING):
+ cond.wait()
+ cond.release()
+ go = self.step_status[d] == STEP_STATUS_OK
+ else:
+ go = True
- should_run = False
- try:
- # Various checks that decide whether
- # this step runs or not
- self.check_class_dependency(sync_step, self.failed_steps) # dont run Slices if Sites failed
- self.check_schedule(sync_step, deletion) # dont run sync_network_routes if time since last run < 1 hour
- should_run = True
- except StepNotReady:
- logging.info('Step not ready: %s'%sync_step.__name__)
- self.failed_steps.append(sync_step)
- except Exception,e:
- logging.error('%r',e)
- logger.log_exc("sync step failed: %r. Deletion: %r"%(sync_step,deletion))
+ if (not go):
self.failed_steps.append(sync_step)
+ my_status = STEP_STATUS_KO
+ else:
+ sync_step = step(driver=self.driver,error_map=self.error_mapper)
+ sync_step.__name__ = step.__name__
+ sync_step.dependencies = []
+ try:
+ mlist = sync_step.provides
+
+ for m in mlist:
+ sync_step.dependencies.extend(self.model_dependency_graph[m.__name__])
+ except KeyError:
+ pass
+ sync_step.debug_mode = debug_mode
- if (should_run):
+ should_run = False
try:
- duration=time.time() - start_time
+ # Various checks that decide whether
+ # this step runs or not
+ self.check_class_dependency(sync_step, self.failed_steps) # dont run Slices if Sites failed
+ self.check_schedule(sync_step, deletion) # dont run sync_network_routes if time since last run < 1 hour
+ should_run = True
+ except StepNotReady:
+ logging.info('Step not ready: %s'%sync_step.__name__)
+ self.failed_steps.append(sync_step)
+ my_status = STEP_STATUS_KO
+ except Exception,e:
+ logging.error('%r',e)
+ logger.log_exc("sync step failed: %r. Deletion: %r"%(sync_step,deletion))
+ self.failed_steps.append(sync_step)
+ my_status = STEP_STATUS_KO
- logger.info('Executing step %s' % sync_step.__name__)
+ if (should_run):
+ try:
+ duration=time.time() - start_time
+
+ logger.info('Executing step %s' % sync_step.__name__)
- failed_objects = sync_step(failed=list(self.failed_step_objects), deletion=deletion)
+ failed_objects = sync_step(failed=list(self.failed_step_objects), deletion=deletion)
- self.check_duration(sync_step, duration)
+ self.check_duration(sync_step, duration)
- if failed_objects:
- self.failed_step_objects.update(failed_objects)
+ if failed_objects:
+ self.failed_step_objects.update(failed_objects)
+ my_status = STEP_STATUS_OK
+ self.update_run_time(sync_step,deletion)
+ except Exception,e:
+ logging.error('Model step failed. This seems like a misconfiguration or bug: %r. This error will not be relayed to the user!',e)
+ logger.log_exc(e)
+ self.failed_steps.append(S)
+ my_status = STEP_STATUS_KO
+ else:
my_status = STEP_STATUS_OK
- self.update_run_time(sync_step,deletion)
- except Exception,e:
- logging.error('Model step failed. This seems like a misconfiguration or bug: %r. This error will not be relayed to the user!',e)
- logger.log_exc(e)
- self.failed_steps.append(S)
- my_status = STEP_STATUS_KO
- else:
- my_status = STEP_STATUS_OK
try:
my_cond = self.step_conditions[S]
self.step_status[S]=my_status
my_cond.notify_all()
my_cond.release()
- except:
+ except KeyError,e:
+ logging.info('Step %r is a leaf')
pass
- if (self.step_conditions.has_key(S)):
-
-
def run(self):
if not self.driver.enabled:
for v in self.dependency_graph.values():
if (v):
providers.update(v)
+
self.step_conditions = {}
self.step_status = {}
for p in list(providers):
self.step_conditions[p] = threading.Condition()
- self.step_status[p] = STEP_STATUS_IDLE
+ self.step_status[p] = STEP_STATUS_WORKING
logger.info('Waiting for event')
for deletion in [False,True]:
threads = []
logger.info('Deletion=%r...'%deletion)
- schedule = self.sync_schedule if not deletion else self.delete_schedule
+ schedule = self.ordered_steps if not deletion else reversed(self.ordered_steps)
- thread = threading.Thread(target=self.sync, args=(schedule.start_conditions, schedule.ordered_steps,deletion, schedule.signal_sem))
+ for S in schedule:
+ thread = threading.Thread(target=self.sync, args=(S, deletion))
- logger.info('Deletion=%r...'%deletion)
- threads.append(thread)
+ logger.info('Deletion=%r...'%deletion)
+ threads.append(thread)
# Start threads
for t in threads:
git://git.onelab.eu / plstackapi.git / blobdiff