planetstack/openstack_observer/event_loop.py

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