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.load_run_times()
 140
 141
 142         def check_duration(self):
 143                 try:
 144                         if (duration > S.deadline):
 145                                 logger.info('Sync step %s missed deadline, took %.2f seconds'%(S.name,duration))
 146                 except AttributeError:
 147                         # S doesn't have a deadline
 148                         pass
 149
 150         def update_run_time(self, step):
 151                 self.last_run_times[step.name]=time.time()
 152
 153         def check_schedule(self, step):
 154                 time_since_last_run = time.time() - self.last_run_times[step.name]
 155                 try:
 156                         if (time_since_last_run < step.requested_interval):
 157                                 raise StepNotReady
 158                 except AttributeError:
 159                         logger.info('Step %s does not have requested_interval set'%step.name)
 160                         raise StepNotReady
 161
 162         def load_run_times(self):
 163                 try:
 164                         jrun_times = open('/tmp/observer_run_times').read()
 165                         self.last_run_times = json.loads(jrun_times)
 166                 except:
 167                         self.last_run_times={}
 168                         for e in self.ordered_steps:
 169                                 self.last_run_times[e.name]=0
 170
 171
 172
 173         def save_run_times(self):
 174                 run_times = json.dumps(self.last_run_times)
 175                 open('/tmp/observer_run_times','w').write(run_times)
 176
 177         def check_class_dependency(self, step, failed_steps):
 178                 for failed_step in failed_steps:
 179                         if (failed_step in self.dependency_graph[step.name]):
 180                                 raise StepNotReady
 181
 182         def run(self):
 183                 if not self.manager.enabled or not self.manager.has_openstack:
 184                         return
 185
 186                 while True:
 187                         try:
 188                                 logger.info('Waiting for event')
 189                                 tBeforeWait = time.time()
 190                                 self.wait_for_event(timeout=300)
 191                                 logger.info('Observer woke up')
 192
 193                                 # Set of whole steps that failed
 194                                 failed_steps = []
 195
 196                                 # Set of individual objects within steps that failed
 197                                 failed_step_objects = []
 198
 199                                 for S in self.ordered_steps:
 200                                         start_time=time.time()
 201
 202                                         sync_step = S()
 203                                         sync_step.dependencies = self.dependencies[sync_step.name]
 204                                         sync_step.debug_mode = debug_mode
 205
 206                                         should_run = False
 207                                         try:
 208                                                 # Various checks that decide whether
 209                                                 # this step runs or not
 210                                                 self.check_class_dependency(sync_step, failed_steps) # dont run Slices if Sites failed
 211                                                 self.check_schedule(sync_step) # dont run sync_network_routes if time since last run < 1 hour
 212                                                 should_run = True
 213                                         except StepNotReady:
 214                                                 logging.info('Step not ready: %s'%sync_step.name)
 215                                                 failed_steps.add(sync_step)
 216                                         except:
 217                                                 failed_steps.add(sync_step)
 218
 219                                         if (should_run):
 220                                                 try:
 221                                                         duration=time.time() - start_time
 222
 223                                                         # ********* This is the actual sync step
 224                                                         failed_objects = sync_step(failed=failed_step_objects)
 225
 226
 227                                                         check_deadline(sync_step, duration)
 228                                                         failed_step_objects.extend(failed_objects)
 229                                                         self.update_run_time(sync_step)
 230                                                 except:
 231                                                         failed_steps.add(S)
 232                                 self.save_run_times()
 233                         except:
 234                                 logger.log_exc("Exception in observer run loop")
 235                                 traceback.print_exc()