planetstack/ec2_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
  10 from datetime import datetime
  11 from collections import defaultdict
  12 from core.models import *
  13 from django.db.models import F, Q
  14 #from openstack.manager import OpenStackManager
  15 from openstack.driver import OpenStackDriver
  16 from util.logger import Logger, logging, logger
  17 #from timeout import timeout
  18 from planetstack.config import Config
  19 from observer.steps import *
  20 from syncstep import SyncStep
  21 from toposort import toposort
  22 from observer.error_mapper import *
  23
  24 debug_mode = False
  25
  26 logger = Logger(level=logging.INFO)
  27
  28 class StepNotReady(Exception):
  29         pass
  30
  31 class NoOpDriver:
  32         def __init__(self):
  33                  self.enabled = True
  34                  self.dependency_graph = None
  35
  36 STEP_STATUS_WORKING=1
  37 STEP_STATUS_OK=2
  38 STEP_STATUS_KO=3
  39
  40 def invert_graph(g):
  41         ig = {}
  42         for k,v in g.items():
  43                 for v0 in v:
  44                         try:
  45                                 ig[v0].append(k)
  46                         except:
  47                                 ig=[k]
  48         return ig
  49
  50 class PlanetStackObserver:
  51         #sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivileges,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
  52         sync_steps = []
  53
  54
  55         def __init__(self):
  56                 # The Condition object that gets signalled by Feefie events
  57                 self.step_lookup = {}
  58                 self.load_sync_step_modules()
  59                 self.load_sync_steps()
  60                 self.event_cond = threading.Condition()
  61
  62                 self.driver_kind = getattr(Config(), "observer_driver", "openstack")
  63                 if self.driver_kind=="openstack":
  64                         self.driver = OpenStackDriver()
  65                 else:
  66                         self.driver = NoOpDriver()
  67
  68         def wait_for_event(self, timeout):
  69                 self.event_cond.acquire()
  70                 self.event_cond.wait(timeout)
  71                 self.event_cond.release()
  72
  73         def wake_up(self):
  74                 logger.info('Wake up routine called. Event cond %r'%self.event_cond)
  75                 self.event_cond.acquire()
  76                 self.event_cond.notify()
  77                 self.event_cond.release()
  78
  79         def load_sync_step_modules(self, step_dir=None):
  80                 if step_dir is None:
  81                         if hasattr(Config(), "observer_steps_dir"):
  82                                 step_dir = Config().observer_steps_dir
  83                         else:
  84                                 step_dir = "/opt/planetstack/observer/steps"
  85
  86                 for fn in os.listdir(step_dir):
  87                         pathname = os.path.join(step_dir,fn)
  88                         if os.path.isfile(pathname) and fn.endswith(".py") and (fn!="__init__.py"):
  89                                 module = imp.load_source(fn[:-3],pathname)
  90                                 for classname in dir(module):
  91                                         c = getattr(module, classname, None)
  92
  93                                         # make sure 'c' is a descendent of SyncStep and has a
  94                                         # provides field (this eliminates the abstract base classes
  95                                         # since they don't have a provides)
  96
  97                                         if inspect.isclass(c) and issubclass(c, SyncStep) and hasattr(c,"provides") and (c not in self.sync_steps):
  98                                                 self.sync_steps.append(c)
  99                 logger.info('loaded sync steps: %s' % ",".join([x.__name__ for x in self.sync_steps]))
 100                 # print 'loaded sync steps: %s' % ",".join([x.__name__ for x in self.sync_steps])
 101
 102         def load_sync_steps(self):
 103                 dep_path = Config().observer_dependency_graph
 104                 logger.info('Loading model dependency graph from %s' % dep_path)
 105                 try:
 106                         # This contains dependencies between records, not sync steps
 107                         self.model_dependency_graph = json.loads(open(dep_path).read())
 108                 except Exception,e:
 109                         raise e
 110
 111                 try:
 112                         backend_path = Config().observer_pl_dependency_graph
 113                         logger.info('Loading backend dependency graph from %s' % backend_path)
 114                         # This contains dependencies between backend records
 115                         self.backend_dependency_graph = json.loads(open(backend_path).read())
 116                 except Exception,e:
 117                         logger.info('Backend dependency graph not loaded')
 118                         # We can work without a backend graph
 119                         self.backend_dependency_graph = {}
 120
 121                 provides_dict = {}
 122                 for s in self.sync_steps:
 123                         self.step_lookup[s.__name__] = s
 124                         for m in s.provides:
 125                                 try:
 126                                         provides_dict[m.__name__].append(s.__name__)
 127                                 except KeyError:
 128                                         provides_dict[m.__name__]=[s.__name__]
 129
 130                 step_graph = {}
 131                 for k,v in self.model_dependency_graph.iteritems():
 132                         try:
 133                                 for source in provides_dict[k]:
 134                                         for m in v:
 135                                                 try:
 136                                                         for dest in provides_dict[m]:
 137                                                                 # no deps, pass
 138                                                                 try:
 139                                                                         if (dest not in step_graph[source]):
 140                                                                                 step_graph[source].append(dest)
 141                                                                 except:
 142                                                                         step_graph[source]=[dest]
 143                                                 except KeyError:
 144                                                         pass
 145
 146                         except KeyError:
 147                                 pass
 148                                 # no dependencies, pass
 149
 150                 #import pdb
 151                 #pdb.set_trace()
 152                 if (self.backend_dependency_graph):
 153                         backend_dict = {}
 154                         for s in self.sync_steps:
 155                                 for m in s.serves:
 156                                         backend_dict[m]=s.__name__
 157
 158                         for k,v in backend_dependency_graph.iteritems():
 159                                 try:
 160                                         source = backend_dict[k]
 161                                         for m in v:
 162                                                 try:
 163                                                         dest = backend_dict[m]
 164                                                 except KeyError:
 165                                                         # no deps, pass
 166                                                         pass
 167                                                 step_graph[source]=dest
 168
 169                                 except KeyError:
 170                                         pass
 171                                         # no dependencies, pass
 172
 173                 self.dependency_graph = step_graph
 174                 self.deletion_dependency_graph = invert_graph(step_graph)
 175
 176                 self.ordered_steps = toposort(self.dependency_graph, map(lambda s:s.__name__,self.sync_steps))
 177                 print "Order of steps=",self.ordered_steps
 178                 self.load_run_times()
 179
 180
 181         def check_duration(self, step, duration):
 182                 try:
 183                         if (duration > step.deadline):
 184                                 logger.info('Sync step %s missed deadline, took %.2f seconds'%(step.name,duration))
 185                 except AttributeError:
 186                         # S doesn't have a deadline
 187                         pass
 188
 189         def update_run_time(self, step, deletion):
 190                 if (not deletion):
 191                         self.last_run_times[step.__name__]=time.time()
 192                 else:
 193                         self.last_deletion_run_times[step.__name__]=time.time()
 194
 195
 196         def check_schedule(self, step, deletion):
 197                 last_run_times = self.last_run_times if not deletion else self.last_deletion_run_times
 198
 199                 time_since_last_run = time.time() - last_run_times.get(step.__name__, 0)
 200                 try:
 201                         if (time_since_last_run < step.requested_interval):
 202                                 raise StepNotReady
 203                 except AttributeError:
 204                         logger.info('Step %s does not have requested_interval set'%step.__name__)
 205                         raise StepNotReady
 206
 207         def load_run_times(self):
 208                 try:
 209                         jrun_times = open('/tmp/observer_run_times').read()
 210                         self.last_run_times = json.loads(jrun_times)
 211                 except:
 212                         self.last_run_times={}
 213                         for e in self.ordered_steps:
 214                                 self.last_run_times[e]=0
 215                 try:
 216                         jrun_times = open('/tmp/observer_deletion_run_times').read()
 217                         self.last_deletion_run_times = json.loads(jrun_times)
 218                 except:
 219                         self.last_deletion_run_times={}
 220                         for e in self.ordered_steps:
 221                                 self.last_deletion_run_times[e]=0
 222
 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_IDLE
 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()