First version of api generator

[plstackapi.git] / planetstack / observer / event_loop.py

diff --git a/planetstack/observer/event_loop.py b/planetstack/observer/event_loop.py
index 22a4e0b..bdf5ce3 100644 (file)
--- a/planetstack/observer/event_loop.py
+++ b/planetstack/observer/event_loop.py
@@ -1,3 +1,6 @@
+import os
+import imp
+import inspect
 import time
 import traceback
 import commands
@@ -14,10 +17,11 @@ from util.logger import Logger, logging, logger
 #from timeout import timeout
 from planetstack.config import Config
 from observer.steps import *
+from syncstep import SyncStep
 
 debug_mode = False
 
-logger = Logger(logfile='observer.log', level=logging.INFO)
+logger = Logger(level=logging.INFO)
 
 class StepNotReady(Exception):
     pass
@@ -68,29 +72,65 @@ def toposort(g, steps=None):
     order.extend(set(steps)-set(order))
     return order
 
+class NoOpDriver:
+    def __init__(self):
+         self.enabled = True
+
 class PlanetStackObserver:
-    sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivileges,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
+    #sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivileges,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
+    sync_steps = []
 
     def __init__(self):
         # The Condition object that gets signalled by Feefie events
         self.step_lookup = {}
+        self.load_sync_step_modules()
         self.load_sync_steps()
         self.event_cond = threading.Condition()
-        self.driver = OpenStackDriver()
+
+
+        self.driver_kind = getattr(Config(), "observer_driver", "openstack")
+        if self.driver_kind=="openstack":
+            self.driver = OpenStackDriver()
+        else:
+            self.driver = NoOpDriver()
 
     def wait_for_event(self, timeout):
         self.event_cond.acquire()
         self.event_cond.wait(timeout)
         self.event_cond.release()
-        
+
     def wake_up(self):
         logger.info('Wake up routine called. Event cond %r'%self.event_cond)
         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:
+            if hasattr(Config(), "observer_steps_dir"):
+                step_dir = Config().observer_steps_dir
+            else:
+                step_dir = "/opt/planetstack/observer/steps"
+
+        for fn in os.listdir(step_dir):
+            pathname = os.path.join(step_dir,fn)
+            if os.path.isfile(pathname) and fn.endswith(".py") and (fn!="__init__.py"):
+                module = imp.load_source(fn[:-3],pathname)
+                for classname in dir(module):
+                    c = getattr(module, classname, None)
+
+                    # make sure 'c' is a descendent of SyncStep and has a
+                    # provides field (this eliminates the abstract base classes
+                    # since they don't have a provides)
+
+                    if inspect.isclass(c) and issubclass(c, SyncStep) and hasattr(c,"provides") and (c not in self.sync_steps):
+                        self.sync_steps.append(c)
+        logger.info('loaded sync steps: %s' % ",".join([x.__name__ for x in self.sync_steps]))
+        # print 'loaded sync steps: %s' % ",".join([x.__name__ for x in self.sync_steps])
+
     def load_sync_steps(self):
-        dep_path = Config().observer_backend_dependency_graph
+        dep_path = Config().observer_dependency_graph
+        logger.info('Loading model dependency graph from %s' % dep_path)
         try:
             # This contains dependencies between records, not sync steps
             self.model_dependency_graph = json.loads(open(dep_path).read())
@@ -99,9 +139,11 @@ class PlanetStackObserver:
 
         try:
             backend_path = Config().observer_pl_dependency_graph
+            logger.info('Loading backend dependency graph from %s' % backend_path)
             # This contains dependencies between backend records
             self.backend_dependency_graph = json.loads(open(backend_path).read())
         except Exception,e:
+            logger.info('Backend dependency graph not loaded')
             # We can work without a backend graph
             self.backend_dependency_graph = {}
 
@@ -207,8 +249,11 @@ class PlanetStackObserver:
                 raise StepNotReady
 
     def run(self):
-        if not self.driver.enabled or not self.driver.has_openstack:
+        if not self.driver.enabled:
+            return
+        if (self.driver_kind=="openstack") and (not self.driver.has_openstack):
             return
+
         while True:
             try:
                 logger.info('Waiting for event')
@@ -220,7 +265,7 @@ class PlanetStackObserver:
                 failed_steps = []
 
                 # Set of individual objects within steps that failed
-                failed_step_objects = []
+                failed_step_objects = set()
 
                 for S in self.ordered_steps:
                     step = self.step_lookup[S]
@@ -255,18 +300,19 @@ class PlanetStackObserver:
                         try:
                             duration=time.time() - start_time
 
+                            logger.info('Executing step %s' % sync_step.__name__)
+
                             # ********* This is the actual sync step
                             #import pdb
                             #pdb.set_trace()
-                            failed_objects = sync_step(failed=failed_step_objects)
+                            failed_objects = sync_step(failed=list(failed_step_objects))
 
 
                             self.check_duration(sync_step, duration)
                             if failed_objects:
-                                failed_step_objects.extend(failed_objects)
+                                failed_step_objects.update(failed_objects)
                             self.update_run_time(sync_step)
                         except:
-                            raise
                             failed_steps.append(S)
                 self.save_run_times()
             except: