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