Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 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 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 15 | debug_mode = False |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 16 | |
| 17 | logger = Logger(logfile='observer.log', level=logging.INFO) |
| 18 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 19 | class StepNotReady(Exception): |
| 20 | pass |
| 21 | |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 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 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 61 | def __init__(self): |
| 62 | self.manager = OpenStackManager() |
| 63 | # The Condition object that gets signalled by Feefie events |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 64 | self.load_sync_steps() |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 65 | self.event_cond = threading.Condition() |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 66 | self.load_enacted() |
| 67 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 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() |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 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: |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 107 | pass |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 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 |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 129 | pass |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 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) |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 139 | self.last_run_times={} |
| 140 | for e in self.ordered_steps: |
| 141 | self.last_run_times[e.name]=0 |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 142 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 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 |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 150 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 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 = [] |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 184 | |
| 185 | for S in self.ordered_steps: |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 186 | start_time=time.time() |
| 187 | |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 188 | sync_step = S() |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 189 | sync_step.dependencies = self.dependencies[sync_step.name] |
| 190 | sync_step.debug_mode = debug_mode |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 191 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 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) |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 204 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 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() |