planetstack/openstack_observer/event_loop.py

   1 import os
   2 import imp
   3 import inspect
   4 import time
   5 import traceback
   6 import commands
   7 import threading
   8 import json
   9 import pdb
  10
  11 from datetime import datetime
  12 from collections import defaultdict
  13 from core.models import *
  14 from django.db.models import F, Q
  15 #from openstack.manager import OpenStackManager
  16 from openstack.driver import OpenStackDriver
  17 from util.logger import Logger, logging, logger
  18 #from timeout import timeout
  19 from planetstack.config import Config
  20 from observer.steps import *
  21 from syncstep import SyncStep
  22 from toposort import toposort
  23 from observer.error_mapper import *
  24
  25 debug_mode = False
  26
  27 logger = Logger(level=logging.INFO)
  28
  29 class StepNotReady(Exception):
  30         pass
  31
  32 class NoOpDriver:
  33         def __init__(self):
  34                  self.enabled = True
  35                  self.dependency_graph = None
  36
  37 STEP_STATUS_WORKING=1
  38 STEP_STATUS_OK=2
  39 STEP_STATUS_KO=3
  40
  41 def invert_graph(g):
  42         ig = {}
  43         for k,v in g.items():
  44                 for v0 in v:
  45                         try:
  46                                 ig[v0].append(k)
  47                         except:
  48                                 ig=[k]
  49         return ig
  50
  51 class PlanetStackObserver:
  52         #sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivileges,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
  53         sync_steps = []
  54
  55
  56         def __init__(self):
  57                 # The Condition object that gets signalled by Feefie events
  58                 self.step_lookup = {}
  59                 self.load_sync_step_modules()
  60                 self.load_sync_steps()
  61                 self.event_cond = threading.Condition()
  62
  63                 self.driver_kind = getattr(Config(), "observer_driver", "openstack")
  64                 if self.driver_kind=="openstack":
  65                         self.driver = OpenStackDriver()
  66                 else:
  67                         self.driver = NoOpDriver()
  68
  69         def wait_for_event(self, timeout):
  70                 self.event_cond.acquire()
  71                 self.event_cond.wait(timeout)
  72                 self.event_cond.release()
  73
  74         def wake_up(self):
  75                 logger.info('Wake up routine called. Event cond %r'%self.event_cond)
  76                 self.event_cond.acquire()
  77                 self.event_cond.notify()
  78                 self.event_cond.release()
  79
  80         def load_sync_step_modules(self, step_dir=None):
  81                 if step_dir is None:
  82                         if hasattr(Config(), "observer_steps_dir"):
  83                                 step_dir = Config().observer_steps_dir
  84                         else:
  85                                 step_dir = "/opt/planetstack/observer/steps"
  86
  87                 for fn in os.listdir(step_dir):
  88                         pathname = os.path.join(step_dir,fn)
  89                         if os.path.isfile(pathname) and fn.endswith(".py") and (fn!="__init__.py"):
  90                                 module = imp.load_source(fn[:-3],pathname)
  91                                 for classname in dir(module):
  92                                         c = getattr(module, classname, None)
  93
  94                                         # make sure 'c' is a descendent of SyncStep and has a
  95                                         # provides field (this eliminates the abstract base classes
  96                                         # since they don't have a provides)
  97
  98                                         if inspect.isclass(c) and issubclass(c, SyncStep) and hasattr(c,"provides") and (c not in self.sync_steps):
  99                                                 self.sync_steps.append(c)
 100                 logger.info('loaded sync steps: %s' % ",".join([x.__name__ for x in self.sync_steps]))
 101                 # print 'loaded sync steps: %s' % ",".join([x.__name__ for x in self.sync_steps])
 102
 103         def load_sync_steps(self):
 104                 dep_path = Config().observer_dependency_graph
 105                 logger.info('Loading model dependency graph from %s' % dep_path)
 106                 try:
 107                         # This contains dependencies between records, not sync steps
 108                         self.model_dependency_graph = json.loads(open(dep_path).read())
 109                 except Exception,e:
 110                         raise e
 111
 112                 try:
 113                         backend_path = Config().observer_pl_dependency_graph
 114                         logger.info('Loading backend dependency graph from %s' % backend_path)
 115                         # This contains dependencies between backend records
 116                         self.backend_dependency_graph = json.loads(open(backend_path).read())
 117                 except Exception,e:
 118                         logger.info('Backend dependency graph not loaded')
 119                         # We can work without a backend graph
 120                         self.backend_dependency_graph = {}
 121
 122                 provides_dict = {}
 123                 for s in self.sync_steps:
 124                         self.step_lookup[s.__name__] = s
 125                         for m in s.provides:
 126                                 try:
 127                                         provides_dict[m.__name__].append(s.__name__)
 128                                 except KeyError:
 129                                         provides_dict[m.__name__]=[s.__name__]
 130
 131                 step_graph = {}
 132                 for k,v in self.model_dependency_graph.iteritems():
 133                         try:
 134                                 for source in provides_dict[k]:
 135                                         for m in v:
 136                                                 try:
 137                                                         for dest in provides_dict[m]:
 138                                                                 # no deps, pass
 139                                                                 try:
 140                                                                         if (dest not in step_graph[source]):
 141                                                                                 step_graph[source].append(dest)
 142                                                                 except:
 143                                                                         step_graph[source]=[dest]
 144                                                 except KeyError:
 145                                                         pass
 146
 147                         except KeyError:
 148                                 pass
 149                                 # no dependencies, pass
 150
 151                 #import pdb
 152                 #pdb.set_trace()
 153                 if (self.backend_dependency_graph):
 154                         backend_dict = {}
 155                         for s in self.sync_steps:
 156                                 for m in s.serves:
 157                                         backend_dict[m]=s.__name__
 158
 159                         for k,v in backend_dependency_graph.iteritems():
 160                                 try:
 161                                         source = backend_dict[k]
 162                                         for m in v:
 163                                                 try:
 164                                                         dest = backend_dict[m]
 165                                                 except KeyError:
 166                                                         # no deps, pass
 167                                                         pass
 168                                                 step_graph[source]=dest
 169
 170                                 except KeyError:
 171                                         pass
 172                                         # no dependencies, pass
 173
 174                 self.dependency_graph = step_graph
 175                 self.deletion_dependency_graph = invert_graph(step_graph)
 176
 177                 self.ordered_steps = toposort(self.dependency_graph, map(lambda s:s.__name__,self.sync_steps))
 178                 print "Order of steps=",self.ordered_steps
 179                 self.load_run_times()
 180
 181
 182         def check_duration(self, step, duration):
 183                 try:
 184                         if (duration > step.deadline):
 185                                 logger.info('Sync step %s missed deadline, took %.2f seconds'%(step.name,duration))
 186                 except AttributeError:
 187                         # S doesn't have a deadline
 188                         pass
 189
 190         def update_run_time(self, step, deletion):
 191                 if (not deletion):
 192                         self.last_run_times[step.__name__]=time.time()
 193                 else:
 194                         self.last_deletion_run_times[step.__name__]=time.time()
 195
 196
 197         def check_schedule(self, step, deletion):
 198                 last_run_times = self.last_run_times if not deletion else self.last_deletion_run_times
 199
 200                 time_since_last_run = time.time() - last_run_times.get(step.__name__, 0)
 201                 try:
 202                         if (time_since_last_run < step.requested_interval):
 203                                 raise StepNotReady
 204                 except AttributeError:
 205                         logger.info('Step %s does not have requested_interval set'%step.__name__)
 206                         raise StepNotReady
 207
 208         def load_run_times(self):
 209                 try:
 210                         jrun_times = open('/tmp/observer_run_times').read()
 211                         self.last_run_times = json.loads(jrun_times)
 212                 except:
 213                         self.last_run_times={}
 214                         for e in self.ordered_steps:
 215                                 self.last_run_times[e]=0
 216                 try:
 217                         jrun_times = open('/tmp/observer_deletion_run_times').read()
 218                         self.last_deletion_run_times = json.loads(jrun_times)
 219                 except:
 220                         self.last_deletion_run_times={}
 221                         for e in self.ordered_steps:
 222                                 self.last_deletion_run_times[e]=0
 223
 224
 225         def save_run_times(self):
 226                 run_times = json.dumps(self.last_run_times)
 227                 open('/tmp/observer_run_times','w').write(run_times)
 228
 229                 deletion_run_times = json.dumps(self.last_deletion_run_times)
 230                 open('/tmp/observer_deletion_run_times','w').write(deletion_run_times)
 231
 232         def check_class_dependency(self, step, failed_steps):
 233                 step.dependenices = []
 234                 for obj in step.provides:
 235                         step.dependenices.extend(self.model_dependency_graph.get(obj.__name__, []))
 236                 for failed_step in failed_steps:
 237                         if (failed_step in step.dependencies):
 238                                 raise StepNotReady
 239
 240         def sync(self, S, deletion):
 241                 step = self.step_lookup[S]
 242                 start_time=time.time()
 243
 244                 dependency_graph = self.dependency_graph if not deletion else self.deletion_dependency_graph
 245
 246                 # Wait for step dependencies to be met
 247                 try:
 248                         deps = self.dependency_graph[S]
 249                         has_deps = True
 250                 except KeyError:
 251                         has_deps = False
 252
 253                 if (has_deps):
 254                         for d in deps:
 255                                 cond = self.step_conditions[d]
 256                                 cond.acquire()
 257                                 if (self.step_status[d] is STEP_STATUS_WORKING):
 258                                         cond.wait()
 259                                 cond.release()
 260                         go = self.step_status[d] == STEP_STATUS_OK
 261                 else:
 262                         go = True
 263
 264                 if (not go):
 265                         self.failed_steps.append(sync_step)
 266                         my_status = STEP_STATUS_KO
 267                 else:
 268                         sync_step = step(driver=self.driver,error_map=self.error_mapper)
 269                         sync_step.__name__ = step.__name__
 270                         sync_step.dependencies = []
 271                         try:
 272                                 mlist = sync_step.provides
 273
 274                                 for m in mlist:
 275                                         sync_step.dependencies.extend(self.model_dependency_graph[m.__name__])
 276                         except KeyError:
 277                                 pass
 278                         sync_step.debug_mode = debug_mode
 279
 280                         should_run = False
 281                         try:
 282                                 # Various checks that decide whether
 283                                 # this step runs or not
 284                                 self.check_class_dependency(sync_step, self.failed_steps) # dont run Slices if Sites failed
 285                                 self.check_schedule(sync_step, deletion) # dont run sync_network_routes if time since last run < 1 hour
 286                                 should_run = True
 287                         except StepNotReady:
 288                                 logging.info('Step not ready: %s'%sync_step.__name__)
 289                                 self.failed_steps.append(sync_step)
 290                                 my_status = STEP_STATUS_KO
 291                         except Exception,e:
 292                                 logging.error('%r',e)
 293                                 logger.log_exc("sync step failed: %r. Deletion: %r"%(sync_step,deletion))
 294                                 self.failed_steps.append(sync_step)
 295                                 my_status = STEP_STATUS_KO
 296
 297                         if (should_run):
 298                                 try:
 299                                         duration=time.time() - start_time
 300
 301                                         logger.info('Executing step %s' % sync_step.__name__)
 302
 303                                         failed_objects = sync_step(failed=list(self.failed_step_objects), deletion=deletion)
 304
 305                                         self.check_duration(sync_step, duration)
 306
 307                                         if failed_objects:
 308                                                 self.failed_step_objects.update(failed_objects)
 309
 310                                         my_status = STEP_STATUS_OK
 311                                         self.update_run_time(sync_step,deletion)
 312                                 except Exception,e:
 313                                         logging.error('Model step failed. This seems like a misconfiguration or bug: %r. This error will not be relayed to the user!',e)
 314                                         logger.log_exc(e)
 315                                         self.failed_steps.append(S)
 316                                         my_status = STEP_STATUS_KO
 317                         else:
 318                                 my_status = STEP_STATUS_OK
 319
 320                 try:
 321                         my_cond = self.step_conditions[S]
 322                         my_cond.acquire()
 323                         self.step_status[S]=my_status
 324                         my_cond.notify_all()
 325                         my_cond.release()
 326                 except KeyError,e:
 327                         logging.info('Step %r is a leaf')
 328                         pass
 329
 330         def run(self):
 331                 if not self.driver.enabled:
 332                         return
 333
 334                 if (self.driver_kind=="openstack") and (not self.driver.has_openstack):
 335                         return
 336
 337                 while True:
 338                         try:
 339                                 error_map_file = getattr(Config(), "error_map_path", "/opt/planetstack/error_map.txt")
 340                                 self.error_mapper = ErrorMapper(error_map_file)
 341
 342                                 # Set of whole steps that failed
 343                                 self.failed_steps = []
 344
 345                                 # Set of individual objects within steps that failed
 346                                 self.failed_step_objects = set()
 347
 348                                 # Set up conditions and step status
 349                                 # This is needed for steps to run in parallel
 350                                 # while obeying dependencies.
 351
 352                                 providers = set()
 353                                 for v in self.dependency_graph.values():
 354                                         if (v):
 355                                                 providers.update(v)
 356
 357                                 self.step_conditions = {}
 358                                 self.step_status = {}
 359                                 for p in list(providers):
 360                                         self.step_conditions[p] = threading.Condition()
 361                                         self.step_status[p] = STEP_STATUS_WORKING
 362
 363
 364                                 logger.info('Waiting for event')
 365                                 tBeforeWait = time.time()
 366                                 self.wait_for_event(timeout=30)
 367                                 logger.info('Observer woke up')
 368
 369                                 # Two passes. One for sync, the other for deletion.
 370                                 for deletion in [False,True]:
 371                                         threads = []
 372                                         logger.info('Deletion=%r...'%deletion)
 373                                         schedule = self.ordered_steps if not deletion else reversed(self.ordered_steps)
 374
 375                                         for S in schedule:
 376                                                 thread = threading.Thread(target=self.sync, args=(S, deletion))
 377
 378                                                 logger.info('Deletion=%r...'%deletion)
 379                                                 threads.append(thread)
 380
 381                                         # Start threads
 382                                         for t in threads:
 383                                                 t.start()
 384
 385                                         # Wait for all threads to finish before continuing with the run loop
 386                                         for t in threads:
 387                                                 t.join()
 388
 389                                 self.save_run_times()
 390                         except Exception, e:
 391                                 logging.error('Core error. This seems like a misconfiguration or bug: %r. This error will not be relayed to the user!',e)
 392                                 logger.log_exc("Exception in observer run loop")
 393                                 traceback.print_exc()