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