planetstack/observer/event_loop.py

   1 import time
   2 import traceback
   3 import commands
   4 import threading
   5 import json
   6
   7 from datetime import datetime
   8 from collections import defaultdict
   9 from core.models import *
  10 from django.db.models import F, Q
  11 from openstack.manager import OpenStackManager
  12 from util.logger import Logger, logging, logger
  13 #from timeout import timeout
  14
  15 debug_mode = False
  16
  17 logger = Logger(logfile='observer.log', level=logging.INFO)
  18
  19 class StepNotReady(Exception):
  20         pass
  21
  22 def toposort(g, steps):
  23         reverse = {}
  24
  25         for k,v in g.items():
  26                 for rk in v:
  27                         try:
  28                                 reverse[rk].append(k)
  29                         except:
  30                                 reverse[rk]=k
  31
  32         sources = []
  33         for k,v in g.items():
  34                 if not reverse.has_key(k):
  35                         sources.append(k)
  36
  37
  38         for k,v in reverse.iteritems():
  39                 if (not v):
  40                         sources.append(k)
  41
  42         order = []
  43         marked = []
  44         while sources:
  45                 n = sources.pop()
  46                 try:
  47                         for m in g[n]:
  48                                 if m not in marked:
  49                                         sources.append(m)
  50                                         marked.append(m)
  51                 except KeyError:
  52                         pass
  53                 if (n in steps):
  54                         order.append(n)
  55
  56         return order
  57
  58 class PlanetStackObserver:
  59         sync_steps = ['SyncNetworks','SyncNetworkSlivers','SyncSites','SyncSitePrivileges','SyncSlices','SyncSliceMemberships','SyncSlivers','SyncSliverIps']
  60
  61         def __init__(self):
  62                 self.manager = OpenStackManager()
  63                 # The Condition object that gets signalled by Feefie events
  64                 self.load_sync_steps()
  65                 self.event_cond = threading.Condition()
  66                 self.load_enacted()
  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_steps(self):
  80                 dep_path = Config().pl_dependency_path
  81                 try:
  82                         # This contains dependencies between records, not sync steps
  83                         self.model_dependency_graph = json.loads(open(dep_path).read())
  84                 except Exception,e:
  85                         raise e
  86
  87                 backend_path = Config().backend_dependency_path
  88                 try:
  89                         # This contains dependencies between backend records
  90                         self.backend_dependency_graph = json.loads(open(backend_path).read())
  91                 except Exception,e:
  92                         raise e
  93
  94                 provides_dict = {}
  95                 for s in sync_steps:
  96                         for m in s.provides:
  97                                 provides_dict[m]=s.__name__
  98
  99                 step_graph = {}
 100                 for k,v in model_dependency_graph.iteritems():
 101                         try:
 102                                 source = provides_dict[k]
 103                                 for m in v:
 104                                         try:
 105                                                 dest = provides_dict[m]
 106                                         except KeyError:
 107                                                 pass
 108                                                 # no deps, pass
 109                                         step_graph[source]=dest
 110
 111                         except KeyError:
 112                                 pass
 113                                 # no dependencies, pass
 114
 115                 if (backend_dependency_graph):
 116                         backend_dict = {}
 117                         for s in sync_steps:
 118                                 for m in s.serves:
 119                                         backend_dict[m]=s.__name__
 120
 121                         for k,v in backend_dependency_graph.iteritems():
 122                                 try:
 123                                         source = backend_dict[k]
 124                                         for m in v:
 125                                                 try:
 126                                                         dest = backend_dict[m]
 127                                                 except KeyError:
 128                                                         # no deps, pass
 129                                                         pass
 130                                                 step_graph[source]=dest
 131
 132                                 except KeyError:
 133                                         pass
 134                                         # no dependencies, pass
 135
 136                 dependency_graph = step_graph
 137
 138                 self.ordered_steps = toposort(dependency_graph, steps)
 139                 self.last_run_times={}
 140                 for e in self.ordered_steps:
 141                         self.last_run_times[e.name]=0
 142
 143         def check_duration(self):
 144                 try:
 145                         if (duration > S.deadline):
 146                                 logger.info('Sync step %s missed deadline, took %.2f seconds'%(S.name,duration))
 147                 except AttributeError:
 148                         # S doesn't have a deadline
 149                         pass
 150
 151         def update_run_time(self, step):
 152                 self.last_run_times[step.name]=time.time()
 153
 154         def check_schedule(self, step):
 155                 time_since_last_run = time.time() - self.last_run_times[step.name]
 156                 try:
 157                         if (time_since_last_run < step.requested_interval):
 158                                 raise StepNotReady
 159                 except AttributeError:
 160                         logger.info('Step %s does not have requested_interval set'%step.name)
 161                         raise StepNotReady
 162
 163         def check_class_dependency(self, step, failed_steps):
 164                 for failed_step in failed_steps:
 165                         if (failed_step in self.dependency_graph[step.name]):
 166                                 raise StepNotReady
 167
 168         def run(self):
 169                 if not self.manager.enabled or not self.manager.has_openstack:
 170                         return
 171
 172                 while True:
 173                         try:
 174                                 logger.info('Waiting for event')
 175                                 tBeforeWait = time.time()
 176                                 self.wait_for_event(timeout=300)
 177                                 logger.info('Observer woke up')
 178
 179                                 # Set of whole steps that failed
 180                                 failed_steps = []
 181
 182                                 # Set of individual objects within steps that failed
 183                                 failed_step_objects = []
 184
 185                                 for S in self.ordered_steps:
 186                                         start_time=time.time()
 187
 188                                         sync_step = S()
 189                                         sync_step.dependencies = self.dependencies[sync_step.name]
 190                                         sync_step.debug_mode = debug_mode
 191
 192                                         should_run = False
 193                                         try:
 194                                                 # Various checks that decide whether
 195                                                 # this step runs or not
 196                                                 self.check_class_dependency(sync_step, failed_steps) # dont run Slices if Sites failed
 197                                                 self.check_schedule(sync_step) # dont run sync_network_routes if time since last run < 1 hour
 198                                                 should_run = True
 199                                         except StepNotReady:
 200                                                 logging.info('Step not ready: %s'%sync_step.name)
 201                                                 failed_steps.add(sync_step)
 202                                         except:
 203                                                 failed_steps.add(sync_step)
 204
 205                                         if (should_run):
 206                                                 try:
 207                                                         duration=time.time() - start_time
 208
 209                                                         # ********* This is the actual sync step
 210                                                         failed_objects = sync_step(failed=failed_step_objects)
 211
 212
 213                                                         check_deadline(sync_step, duration)
 214                                                         failed_step_objects.extend(failed_objects)
 215                                                         self.update_run_time(sync_step)
 216                                                 except:
 217                                                         failed_steps.add(S)
 218                         except:
 219                                 logger.log_exc("Exception in observer run loop")
 220                                 traceback.print_exc()