blob: 12965bbcecbff3599f8d19c9bed09f7ec09192e1 [file] [log] [blame]
Scott Baker45fb7a12013-12-31 00:56:19 -08001import os
2import imp
3import inspect
Sapan Bhatia24836f12013-08-27 10:16:05 -04004import time
5import traceback
6import commands
7import threading
8import json
Sapan Bhatiaab202a62014-09-03 11:30:21 -04009import pdb
Sapan Bhatia24836f12013-08-27 10:16:05 -040010
11from datetime import datetime
12from collections import defaultdict
13from core.models import *
14from django.db.models import F, Q
Scott Baker2e6c9a42014-09-05 14:48:38 -070015from django.db import connection
Tony Mack387a73f2013-09-18 07:59:14 -040016#from openstack.manager import OpenStackManager
17from openstack.driver import OpenStackDriver
Sapan Bhatia24836f12013-08-27 10:16:05 -040018from util.logger import Logger, logging, logger
19#from timeout import timeout
Sapan Bhatia757e0b62013-09-02 16:55:00 -040020from planetstack.config import Config
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040021from observer.steps import *
Scott Baker45fb7a12013-12-31 00:56:19 -080022from syncstep import SyncStep
Sapan Bhatia45cbbc32014-03-11 17:48:30 -040023from toposort import toposort
Sapan Bhatia13d89152014-07-23 10:35:33 -040024from observer.error_mapper import *
Sapan Bhatia24836f12013-08-27 10:16:05 -040025
Sapan Bhatia13c7f112013-09-02 14:19:35 -040026debug_mode = False
Sapan Bhatia24836f12013-08-27 10:16:05 -040027
Andy Bavier04111b72013-10-22 16:47:10 -040028logger = Logger(level=logging.INFO)
Sapan Bhatia24836f12013-08-27 10:16:05 -040029
Sapan Bhatia13c7f112013-09-02 14:19:35 -040030class StepNotReady(Exception):
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040031 pass
Sapan Bhatia24836f12013-08-27 10:16:05 -040032
Scott Baker7771f412014-01-02 16:36:41 -080033class NoOpDriver:
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040034 def __init__(self):
35 self.enabled = True
Sapan Bhatiaab202a62014-09-03 11:30:21 -040036 self.dependency_graph = None
37
38STEP_STATUS_WORKING=1
39STEP_STATUS_OK=2
40STEP_STATUS_KO=3
41
42def 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
Scott Baker7771f412014-01-02 16:36:41 -080051
Sapan Bhatia24836f12013-08-27 10:16:05 -040052class PlanetStackObserver:
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040053 #sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivileges,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
54 sync_steps = []
Sapan Bhatia24836f12013-08-27 10:16:05 -040055
Sapan Bhatiaab202a62014-09-03 11:30:21 -040056
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040057 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()
Scott Baker7771f412014-01-02 16:36:41 -080063
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040064 self.driver_kind = getattr(Config(), "observer_driver", "openstack")
65 if self.driver_kind=="openstack":
66 self.driver = OpenStackDriver()
67 else:
68 self.driver = NoOpDriver()
Sapan Bhatia24836f12013-08-27 10:16:05 -040069
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040070 def wait_for_event(self, timeout):
71 self.event_cond.acquire()
72 self.event_cond.wait(timeout)
73 self.event_cond.release()
Scott Baker45fb7a12013-12-31 00:56:19 -080074
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040075 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()
Sapan Bhatia24836f12013-08-27 10:16:05 -040080
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040081 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"
Scott Baker45fb7a12013-12-31 00:56:19 -080087
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040088 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)
Scott Baker45fb7a12013-12-31 00:56:19 -080094
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040095 # 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)
Scott Baker45fb7a12013-12-31 00:56:19 -080098
Sapan Bhatiaf73664b2014-04-28 13:07:18 -040099 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])
Scott Baker45fb7a12013-12-31 00:56:19 -0800103
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400104 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
Sapan Bhatia24836f12013-08-27 10:16:05 -0400112
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400113 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 = {}
Sapan Bhatia24836f12013-08-27 10:16:05 -0400122
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400123 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__]
Sapan Bhatia04c94ad2013-09-02 18:00:28 -0400131
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400132 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
Sapan Bhatia24836f12013-08-27 10:16:05 -0400174
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400175 self.dependency_graph = step_graph
176 self.deletion_dependency_graph = invert_graph(step_graph)
Sapan Bhatia24836f12013-08-27 10:16:05 -0400177
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400178 self.ordered_steps = toposort(self.dependency_graph, map(lambda s:s.__name__,self.sync_steps))
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400179 print "Order of steps=",self.ordered_steps
180 self.load_run_times()
181
Sapan Bhatia24836f12013-08-27 10:16:05 -0400182
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400183 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
Sapan Bhatia24836f12013-08-27 10:16:05 -0400190
Sapan Bhatia285decb2014-04-30 00:31:44 -0400191 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()
Sapan Bhatia13c7f112013-09-02 14:19:35 -0400196
Sapan Bhatia285decb2014-04-30 00:31:44 -0400197
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)
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400202 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
Sapan Bhatia285decb2014-04-30 00:31:44 -0400217 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
Sapan Bhatia36938ca2013-09-02 14:35:24 -0400225
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400226 def save_run_times(self):
227 run_times = json.dumps(self.last_run_times)
228 open('/tmp/observer_run_times','w').write(run_times)
Sapan Bhatia36938ca2013-09-02 14:35:24 -0400229
Sapan Bhatia285decb2014-04-30 00:31:44 -0400230 deletion_run_times = json.dumps(self.last_deletion_run_times)
231 open('/tmp/observer_deletion_run_times','w').write(deletion_run_times)
232
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400233 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
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400241 def sync(self, S, deletion):
Scott Baker2e6c9a42014-09-05 14:48:38 -0700242 try:
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400243 step = self.step_lookup[S]
244 start_time=time.time()
Scott Baker72aa9302014-09-04 10:36:51 -0700245
246 logger.info("Starting to work on step %s" % step.__name__)
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400247
248 dependency_graph = self.dependency_graph if not deletion else self.deletion_dependency_graph
Sapan Bhatia51f48932014-08-25 04:17:12 -0400249
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400250 # 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
Sapan Bhatia51f48932014-08-25 04:17:12 -0400256
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400257 if (has_deps):
258 for d in deps:
Scott Baker72aa9302014-09-04 10:36:51 -0700259 if d==step.__name__:
260 logger.info(" step %s self-wait skipped" % step.__name__)
261 continue
262
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400263 cond = self.step_conditions[d]
264 cond.acquire()
265 if (self.step_status[d] is STEP_STATUS_WORKING):
Scott Baker72aa9302014-09-04 10:36:51 -0700266 logger.info(" step %s wait on dep %s" % (step.__name__, d))
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400267 cond.wait()
268 cond.release()
269 go = self.step_status[d] == STEP_STATUS_OK
270 else:
271 go = True
272
273 if (not go):
Scott Baker72aa9302014-09-04 10:36:51 -0700274 # SMBAKER: sync_step was not defined here, so I changed
275 # this from 'sync_step' to 'step'. Verify.
276 self.failed_steps.append(step)
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400277 my_status = STEP_STATUS_KO
278 else:
279 sync_step = step(driver=self.driver,error_map=self.error_mapper)
Sapan Bhatia51f48932014-08-25 04:17:12 -0400280 sync_step.__name__ = step.__name__
281 sync_step.dependencies = []
282 try:
283 mlist = sync_step.provides
Scott Baker72aa9302014-09-04 10:36:51 -0700284
Sapan Bhatia51f48932014-08-25 04:17:12 -0400285 for m in mlist:
286 sync_step.dependencies.extend(self.model_dependency_graph[m.__name__])
287 except KeyError:
288 pass
289 sync_step.debug_mode = debug_mode
290
291 should_run = False
292 try:
293 # Various checks that decide whether
294 # this step runs or not
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400295 self.check_class_dependency(sync_step, self.failed_steps) # dont run Slices if Sites failed
Sapan Bhatia51f48932014-08-25 04:17:12 -0400296 self.check_schedule(sync_step, deletion) # dont run sync_network_routes if time since last run < 1 hour
297 should_run = True
298 except StepNotReady:
Scott Baker72aa9302014-09-04 10:36:51 -0700299 logger.info('Step not ready: %s'%sync_step.__name__)
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400300 self.failed_steps.append(sync_step)
301 my_status = STEP_STATUS_KO
Sapan Bhatia51f48932014-08-25 04:17:12 -0400302 except Exception,e:
Scott Baker72aa9302014-09-04 10:36:51 -0700303 logger.error('%r' % e)
Sapan Bhatia51f48932014-08-25 04:17:12 -0400304 logger.log_exc("sync step failed: %r. Deletion: %r"%(sync_step,deletion))
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400305 self.failed_steps.append(sync_step)
306 my_status = STEP_STATUS_KO
Sapan Bhatia51f48932014-08-25 04:17:12 -0400307
308 if (should_run):
309 try:
310 duration=time.time() - start_time
311
312 logger.info('Executing step %s' % sync_step.__name__)
313
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400314 failed_objects = sync_step(failed=list(self.failed_step_objects), deletion=deletion)
Sapan Bhatia51f48932014-08-25 04:17:12 -0400315
316 self.check_duration(sync_step, duration)
Sapan Bhatia51f48932014-08-25 04:17:12 -0400317
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400318 if failed_objects:
319 self.failed_step_objects.update(failed_objects)
320
Scott Baker72aa9302014-09-04 10:36:51 -0700321 logger.info("Step %r succeeded" % step)
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400322 my_status = STEP_STATUS_OK
Sapan Bhatia51f48932014-08-25 04:17:12 -0400323 self.update_run_time(sync_step,deletion)
324 except Exception,e:
Scott Baker72aa9302014-09-04 10:36:51 -0700325 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))
Sapan Bhatia51f48932014-08-25 04:17:12 -0400326 logger.log_exc(e)
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400327 self.failed_steps.append(S)
328 my_status = STEP_STATUS_KO
329 else:
Scott Baker72aa9302014-09-04 10:36:51 -0700330 logger.info("Step %r succeeded due to non-run" % step)
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400331 my_status = STEP_STATUS_OK
Scott Baker72aa9302014-09-04 10:36:51 -0700332
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400333 try:
334 my_cond = self.step_conditions[S]
335 my_cond.acquire()
336 self.step_status[S]=my_status
337 my_cond.notify_all()
338 my_cond.release()
339 except KeyError,e:
Scott Baker72aa9302014-09-04 10:36:51 -0700340 logger.info('Step %r is a leaf' % step)
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400341 pass
Scott Baker2e6c9a42014-09-05 14:48:38 -0700342 finally:
343 connection.close()
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400344
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400345 def run(self):
346 if not self.driver.enabled:
347 return
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400348
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400349 if (self.driver_kind=="openstack") and (not self.driver.has_openstack):
350 return
351
352 while True:
353 try:
Sapan Bhatia31ebe5c2014-04-29 00:24:09 -0400354 error_map_file = getattr(Config(), "error_map_path", "/opt/planetstack/error_map.txt")
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400355 self.error_mapper = ErrorMapper(error_map_file)
356
357 # Set of whole steps that failed
358 self.failed_steps = []
359
360 # Set of individual objects within steps that failed
361 self.failed_step_objects = set()
362
363 # Set up conditions and step status
364 # This is needed for steps to run in parallel
365 # while obeying dependencies.
366
367 providers = set()
368 for v in self.dependency_graph.values():
369 if (v):
370 providers.update(v)
371
372 self.step_conditions = {}
373 self.step_status = {}
374 for p in list(providers):
375 self.step_conditions[p] = threading.Condition()
376 self.step_status[p] = STEP_STATUS_WORKING
377
Sapan Bhatia31ebe5c2014-04-29 00:24:09 -0400378
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400379 logger.info('Waiting for event')
380 tBeforeWait = time.time()
Sapan Bhatia13d89152014-07-23 10:35:33 -0400381 self.wait_for_event(timeout=30)
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400382 logger.info('Observer woke up')
383
Sapan Bhatia285decb2014-04-30 00:31:44 -0400384 # Two passes. One for sync, the other for deletion.
Sapan Bhatia0f727b82014-08-18 02:44:20 -0400385 for deletion in [False,True]:
Sapan Bhatia51f48932014-08-25 04:17:12 -0400386 threads = []
Sapan Bhatiae82f5e52014-07-23 10:02:45 -0400387 logger.info('Deletion=%r...'%deletion)
Sapan Bhatiaab202a62014-09-03 11:30:21 -0400388 schedule = self.ordered_steps if not deletion else reversed(self.ordered_steps)
389
390 for S in schedule:
391 thread = threading.Thread(target=self.sync, args=(S, deletion))
392
393 logger.info('Deletion=%r...'%deletion)
394 threads.append(thread)
Sapan Bhatia285decb2014-04-30 00:31:44 -0400395
Sapan Bhatia51f48932014-08-25 04:17:12 -0400396 # Start threads
397 for t in threads:
398 t.start()
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400399
Sapan Bhatia51f48932014-08-25 04:17:12 -0400400 # Wait for all threads to finish before continuing with the run loop
401 for t in threads:
402 t.join()
Sapan Bhatia285decb2014-04-30 00:31:44 -0400403
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400404 self.save_run_times()
405 except Exception, e:
Scott Baker72aa9302014-09-04 10:36:51 -0700406 logger.error('Core error. This seems like a misconfiguration or bug: %r. This error will not be relayed to the user!' % e)
Sapan Bhatiaf73664b2014-04-28 13:07:18 -0400407 logger.log_exc("Exception in observer run loop")
408 traceback.print_exc()