blob: 27475ca57f316802bded5a7408a1561c8a8902e0 [file] [log] [blame]
Sapan Bhatia24836f12013-08-27 10:16:05 -04001import time
2import traceback
3import commands
4import threading
5import json
6
7from datetime import datetime
8from collections import defaultdict
9from core.models import *
10from django.db.models import F, Q
Tony Mack387a73f2013-09-18 07:59:14 -040011#from openstack.manager import OpenStackManager
12from openstack.driver import OpenStackDriver
Sapan Bhatia24836f12013-08-27 10:16:05 -040013from util.logger import Logger, logging, logger
14#from timeout import timeout
Sapan Bhatia757e0b62013-09-02 16:55:00 -040015from planetstack.config import Config
Sapan Bhatia04c94ad2013-09-02 18:00:28 -040016from observer.steps import *
Sapan Bhatia24836f12013-08-27 10:16:05 -040017
Sapan Bhatia13c7f112013-09-02 14:19:35 -040018debug_mode = False
Sapan Bhatia24836f12013-08-27 10:16:05 -040019
Andy Bavier04111b72013-10-22 16:47:10 -040020logger = Logger(level=logging.INFO)
Sapan Bhatia24836f12013-08-27 10:16:05 -040021
Sapan Bhatia13c7f112013-09-02 14:19:35 -040022class StepNotReady(Exception):
Tony Macke1509e82013-10-09 12:38:04 -040023 pass
Sapan Bhatia13c7f112013-09-02 14:19:35 -040024
Sapan Bhatia467b7ce2013-10-02 09:25:46 -040025def toposort(g, steps=None):
Tony Macke1509e82013-10-09 12:38:04 -040026 if (not steps):
27 keys = set(g.keys())
28 values = set({})
29 for v in g.values():
30 values=values | set(v)
31
32 steps=list(keys|values)
33 reverse = {}
Sapan Bhatia24836f12013-08-27 10:16:05 -040034
Tony Macke1509e82013-10-09 12:38:04 -040035 for k,v in g.items():
36 for rk in v:
37 try:
38 reverse[rk].append(k)
39 except:
40 reverse[rk]=k
Sapan Bhatia24836f12013-08-27 10:16:05 -040041
Tony Macke1509e82013-10-09 12:38:04 -040042 sources = []
43 for k,v in g.items():
44 if not reverse.has_key(k):
45 sources.append(k)
Sapan Bhatia24836f12013-08-27 10:16:05 -040046
47
Tony Macke1509e82013-10-09 12:38:04 -040048 for k,v in reverse.iteritems():
49 if (not v):
50 sources.append(k)
Sapan Bhatia24836f12013-08-27 10:16:05 -040051
Tony Macke1509e82013-10-09 12:38:04 -040052 order = []
53 marked = []
Sapan Bhatia04c94ad2013-09-02 18:00:28 -040054
Tony Macke1509e82013-10-09 12:38:04 -040055 while sources:
56 n = sources.pop()
57 try:
58 for m in g[n]:
59 if m not in marked:
60 sources.append(m)
61 marked.append(m)
62 except KeyError:
63 pass
64 if (n in steps):
65 order.append(n)
Sapan Bhatia972a2e82013-10-02 00:03:02 -040066
Tony Macke1509e82013-10-09 12:38:04 -040067 order.reverse()
68 order.extend(set(steps)-set(order))
69 return order
Sapan Bhatia24836f12013-08-27 10:16:05 -040070
71class PlanetStackObserver:
Tony Mack46c2d502013-10-09 13:04:28 -040072 sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivileges,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
Sapan Bhatia24836f12013-08-27 10:16:05 -040073
Tony Macke1509e82013-10-09 12:38:04 -040074 def __init__(self):
75 # The Condition object that gets signalled by Feefie events
76 self.step_lookup = {}
77 self.load_sync_steps()
78 self.event_cond = threading.Condition()
79 self.driver = OpenStackDriver()
Sapan Bhatia24836f12013-08-27 10:16:05 -040080
Tony Macke1509e82013-10-09 12:38:04 -040081 def wait_for_event(self, timeout):
82 self.event_cond.acquire()
83 self.event_cond.wait(timeout)
84 self.event_cond.release()
85
86 def wake_up(self):
87 logger.info('Wake up routine called. Event cond %r'%self.event_cond)
88 self.event_cond.acquire()
89 self.event_cond.notify()
90 self.event_cond.release()
Sapan Bhatia24836f12013-08-27 10:16:05 -040091
Tony Macke1509e82013-10-09 12:38:04 -040092 def load_sync_steps(self):
Tony Mack3587dc72013-10-14 22:01:43 -040093 dep_path = Config().observer_dependency_graph
Tony Macke1509e82013-10-09 12:38:04 -040094 try:
95 # This contains dependencies between records, not sync steps
96 self.model_dependency_graph = json.loads(open(dep_path).read())
97 except Exception,e:
98 raise e
Sapan Bhatia24836f12013-08-27 10:16:05 -040099
Tony Macke1509e82013-10-09 12:38:04 -0400100 try:
101 backend_path = Config().observer_pl_dependency_graph
102 # This contains dependencies between backend records
103 self.backend_dependency_graph = json.loads(open(backend_path).read())
104 except Exception,e:
105 # We can work without a backend graph
106 self.backend_dependency_graph = {}
Sapan Bhatia24836f12013-08-27 10:16:05 -0400107
Tony Macke1509e82013-10-09 12:38:04 -0400108 provides_dict = {}
109 for s in self.sync_steps:
110 self.step_lookup[s.__name__] = s
111 for m in s.provides:
112 try:
113 provides_dict[m.__name__].append(s.__name__)
114 except KeyError:
115 provides_dict[m.__name__]=[s.__name__]
Sapan Bhatia04c94ad2013-09-02 18:00:28 -0400116
Tony Macke1509e82013-10-09 12:38:04 -0400117
118 step_graph = {}
119 for k,v in self.model_dependency_graph.iteritems():
120 try:
121 for source in provides_dict[k]:
122 for m in v:
123 try:
124 for dest in provides_dict[m]:
125 # no deps, pass
126 try:
127 step_graph[source].append(dest)
128 except:
129 step_graph[source]=[dest]
130 except KeyError:
131 pass
132
133 except KeyError:
134 pass
135 # no dependencies, pass
136
137 #import pdb
138 #pdb.set_trace()
139 if (self.backend_dependency_graph):
140 backend_dict = {}
141 for s in self.sync_steps:
142 for m in s.serves:
143 backend_dict[m]=s.__name__
144
145 for k,v in backend_dependency_graph.iteritems():
146 try:
147 source = backend_dict[k]
148 for m in v:
149 try:
150 dest = backend_dict[m]
151 except KeyError:
152 # no deps, pass
153 pass
154 step_graph[source]=dest
155
156 except KeyError:
157 pass
158 # no dependencies, pass
Sapan Bhatia24836f12013-08-27 10:16:05 -0400159
Tony Macke1509e82013-10-09 12:38:04 -0400160 dependency_graph = step_graph
Sapan Bhatia24836f12013-08-27 10:16:05 -0400161
Tony Macke1509e82013-10-09 12:38:04 -0400162 self.ordered_steps = toposort(dependency_graph, map(lambda s:s.__name__,self.sync_steps))
163 print "Order of steps=",self.ordered_steps
164 self.load_run_times()
165
Sapan Bhatia24836f12013-08-27 10:16:05 -0400166
Tony Macke1509e82013-10-09 12:38:04 -0400167 def check_duration(self, step, duration):
168 try:
169 if (duration > step.deadline):
170 logger.info('Sync step %s missed deadline, took %.2f seconds'%(step.name,duration))
171 except AttributeError:
172 # S doesn't have a deadline
173 pass
Sapan Bhatia24836f12013-08-27 10:16:05 -0400174
Tony Macke1509e82013-10-09 12:38:04 -0400175 def update_run_time(self, step):
176 self.last_run_times[step.__name__]=time.time()
Sapan Bhatia13c7f112013-09-02 14:19:35 -0400177
Tony Macke1509e82013-10-09 12:38:04 -0400178 def check_schedule(self, step):
179 time_since_last_run = time.time() - self.last_run_times.get(step.__name__, 0)
180 try:
181 if (time_since_last_run < step.requested_interval):
182 raise StepNotReady
183 except AttributeError:
184 logger.info('Step %s does not have requested_interval set'%step.__name__)
185 raise StepNotReady
186
187 def load_run_times(self):
188 try:
189 jrun_times = open('/tmp/observer_run_times').read()
190 self.last_run_times = json.loads(jrun_times)
191 except:
192 self.last_run_times={}
193 for e in self.ordered_steps:
194 self.last_run_times[e]=0
Sapan Bhatia36938ca2013-09-02 14:35:24 -0400195
196
Tony Macke1509e82013-10-09 12:38:04 -0400197 def save_run_times(self):
198 run_times = json.dumps(self.last_run_times)
199 open('/tmp/observer_run_times','w').write(run_times)
Sapan Bhatia36938ca2013-09-02 14:35:24 -0400200
Tony Macke1509e82013-10-09 12:38:04 -0400201 def check_class_dependency(self, step, failed_steps):
Tony Mack2fd8d302013-10-05 09:59:38 -0400202 step.dependenices = []
203 for obj in step.provides:
204 step.dependenices.extend(self.model_dependency_graph.get(obj.__name__, []))
205 for failed_step in failed_steps:
206 if (failed_step in step.dependencies):
207 raise StepNotReady
Sapan Bhatia13c7f112013-09-02 14:19:35 -0400208
Tony Macke1509e82013-10-09 12:38:04 -0400209 def run(self):
210 if not self.driver.enabled or not self.driver.has_openstack:
211 return
212 while True:
213 try:
214 logger.info('Waiting for event')
215 tBeforeWait = time.time()
216 self.wait_for_event(timeout=30)
217 logger.info('Observer woke up')
Sapan Bhatia13c7f112013-09-02 14:19:35 -0400218
Tony Macke1509e82013-10-09 12:38:04 -0400219 # Set of whole steps that failed
220 failed_steps = []
Sapan Bhatia13c7f112013-09-02 14:19:35 -0400221
Tony Macke1509e82013-10-09 12:38:04 -0400222 # Set of individual objects within steps that failed
Tony Mack808b2392013-10-16 11:34:32 -0400223 failed_step_objects = set()
Sapan Bhatia24836f12013-08-27 10:16:05 -0400224
Tony Macke1509e82013-10-09 12:38:04 -0400225 for S in self.ordered_steps:
226 step = self.step_lookup[S]
227 start_time=time.time()
228
229 sync_step = step(driver=self.driver)
230 sync_step.__name__ = step.__name__
231 sync_step.dependencies = []
232 try:
233 mlist = sync_step.provides
234
235 for m in mlist:
236 sync_step.dependencies.extend(self.model_dependency_graph[m.__name__])
237 except KeyError:
238 pass
239 sync_step.debug_mode = debug_mode
Sapan Bhatia24836f12013-08-27 10:16:05 -0400240
Tony Macke1509e82013-10-09 12:38:04 -0400241 should_run = False
242 try:
243 # Various checks that decide whether
244 # this step runs or not
245 self.check_class_dependency(sync_step, failed_steps) # dont run Slices if Sites failed
246 self.check_schedule(sync_step) # dont run sync_network_routes if time since last run < 1 hour
247 should_run = True
248 except StepNotReady:
249 logging.info('Step not ready: %s'%sync_step.__name__)
250 failed_steps.append(sync_step)
251 except:
252 failed_steps.append(sync_step)
Sapan Bhatia24836f12013-08-27 10:16:05 -0400253
Tony Macke1509e82013-10-09 12:38:04 -0400254 if (should_run):
255 try:
256 duration=time.time() - start_time
Sapan Bhatia13c7f112013-09-02 14:19:35 -0400257
Tony Macke1509e82013-10-09 12:38:04 -0400258 # ********* This is the actual sync step
259 #import pdb
260 #pdb.set_trace()
Tony Mack808b2392013-10-16 11:34:32 -0400261 failed_objects = sync_step(failed=list(failed_step_objects))
Sapan Bhatia13c7f112013-09-02 14:19:35 -0400262
263
Tony Macke1509e82013-10-09 12:38:04 -0400264 self.check_duration(sync_step, duration)
265 if failed_objects:
Tony Mack808b2392013-10-16 11:34:32 -0400266 failed_step_objects.update(failed_objects)
Tony Macke1509e82013-10-09 12:38:04 -0400267 self.update_run_time(sync_step)
268 except:
Tony Macke1509e82013-10-09 12:38:04 -0400269 failed_steps.append(S)
270 self.save_run_times()
271 except:
272 logger.log_exc("Exception in observer run loop")
273 traceback.print_exc()