Package flumotion :: Package common :: Module planet
[hide private]

Source Code for Module flumotion.common.planet

  1  # -*- Mode: Python; -*- 
  2  # vi:si:et:sw=4:sts=4:ts=4 
  3   
  4  # Flumotion - a streaming media server 
  5  # Copyright (C) 2004,2005,2006,2007,2008,2009 Fluendo, S.L. 
  6  # Copyright (C) 2010,2011 Flumotion Services, S.A. 
  7  # All rights reserved. 
  8  # 
  9  # This file may be distributed and/or modified under the terms of 
 10  # the GNU Lesser General Public License version 2.1 as published by 
 11  # the Free Software Foundation. 
 12  # This file is distributed without any warranty; without even the implied 
 13  # warranty of merchantability or fitness for a particular purpose. 
 14  # See "LICENSE.LGPL" in the source distribution for more information. 
 15  # 
 16  # Headers in this file shall remain intact. 
 17   
 18  """serializable objects from worker through manager to admin. 
 19  Used by planet, flow, job and component. 
 20  """ 
 21   
 22  from twisted.spread import pb 
 23  from twisted.internet import defer 
 24  from zope.interface import implements 
 25   
 26  from flumotion.twisted import flavors 
 27  from flumotion.common import enum, log 
 28   
 29  __version__ = "$Rev$" 
 30   
 31   
32 -class ManagerPlanetState(flavors.StateCacheable):
33 """ 34 I represent the state of a planet in the manager. 35 36 I have the following keys: 37 38 - name 39 - manager 40 - atmosphere: L{ManagerAtmosphereState} 41 - flows (list): list of L{ManagerFlowState} 42 """ 43 # FIXME: why is there a 'parent' key ? 44
45 - def __init__(self):
46 flavors.StateCacheable.__init__(self) 47 self.addKey('name') 48 self.addKey('version') 49 self.addKey('parent') 50 self.addKey('manager') 51 self.addKey('atmosphere') 52 self.addListKey('flows') 53 self.addDictKey('messages') 54 55 # we always have at least one atmosphere 56 self.set('atmosphere', ManagerAtmosphereState()) 57 self.get('atmosphere').set('parent', self)
58
59 - def getComponents(self):
60 """ 61 Return a list of all component states in this planet 62 (from atmosphere and all flows). 63 64 @rtype: list of L{ManagerComponentState} 65 """ 66 ret = [] 67 68 a = self.get('atmosphere') 69 if a: 70 ret.extend(a.get('components')) 71 72 flows = self.get('flows') 73 if flows: 74 for flow in flows: 75 ret.extend(flow.get('components')) 76 77 return ret
78 79
80 -class AdminPlanetState(flavors.StateRemoteCache):
81 """ 82 I represent the state of a planet in an admin client. 83 See L{ManagerPlanetState}. 84 """ 85
86 - def invalidate(self):
87 for flow in self.get('flows'): 88 flow.invalidate() 89 90 self.get('atmosphere').invalidate() 91 92 flavors.StateRemoteCache.invalidate(self)
93 94 pb.setUnjellyableForClass(ManagerPlanetState, AdminPlanetState) 95 96
97 -class ManagerAtmosphereState(flavors.StateCacheable):
98 """ 99 I represent the state of an atmosphere in the manager. 100 The atmosphere contains components that do not participate in a flow, 101 but provide services to flow components. 102 103 I have the following keys: 104 105 - name: string, "atmosphere" 106 - parent: L{ManagerPlanetState} 107 - components (list): list of L{ManagerComponentState} 108 """ 109
110 - def __init__(self):
111 flavors.StateCacheable.__init__(self) 112 self.addKey('parent') 113 self.addListKey('components') 114 self.addKey('name') 115 self.set('name', 'atmosphere')
116
117 - def empty(self):
118 """ 119 Clear out all component entries. 120 121 @returns: a DeferredList that will fire when all notifications 122 are done. 123 """ 124 # make a copy, so we can iterate safely while modifying 125 components = self.get('components')[:] 126 127 dList = [self.remove('components', c) for c in components] 128 return defer.DeferredList(dList)
129 130
131 -class AdminAtmosphereState(flavors.StateRemoteCache):
132 """ 133 I represent the state of an atmosphere in an admin client. 134 See L{ManagerAtmosphereState}. 135 """ 136
137 - def invalidate(self):
138 for component in self.get('components'): 139 component.invalidate() 140 141 flavors.StateRemoteCache.invalidate(self)
142 143 pb.setUnjellyableForClass(ManagerAtmosphereState, AdminAtmosphereState) 144 145
146 -class ManagerFlowState(flavors.StateCacheable):
147 """ 148 I represent the state of a flow in the manager. 149 150 I have the following keys: 151 152 - name: string, name of the flow 153 - parent: L{ManagerPlanetState} 154 - components (list): list of L{ManagerComponentState} 155 """ 156
157 - def __init__(self, **kwargs):
158 """ 159 ManagerFlowState constructor. Any keyword arguments are 160 intepreted as initial key-value pairs to set on the new 161 ManagerFlowState. 162 """ 163 flavors.StateCacheable.__init__(self) 164 self.addKey('name') 165 self.addKey('parent') 166 self.addListKey('components') 167 for k, v in kwargs.items(): 168 self.set(k, v)
169
170 - def empty(self):
171 """ 172 Clear out all component entries 173 """ 174 # take a copy of the list because we're modifying while running 175 components = self.get('components')[:] 176 177 dList = [self.remove('components', c) for c in components] 178 return defer.DeferredList(dList)
179 180
181 -class AdminFlowState(flavors.StateRemoteCache):
182 """ 183 I represent the state of a flow in an admin client. 184 See L{ManagerFlowState}. 185 """ 186
187 - def invalidate(self):
188 for component in self.get('components'): 189 component.invalidate() 190 191 flavors.StateRemoteCache.invalidate(self)
192 193 pb.setUnjellyableForClass(ManagerFlowState, AdminFlowState) 194 195 # moods 196 # FIXME. make epydoc like this 197 """ 198 @cvar moods: an enum representing the mood a component can be in. 199 """ 200 moods = enum.EnumClass( 201 'Moods', 202 ('happy', 'hungry', 'waking', 'sleeping', 'lost', 'sad')) 203 moods.can_stop = staticmethod(lambda m: m != moods.sleeping) 204 moods.can_start = staticmethod(lambda m: m == moods.sleeping) 205 206 _jobStateKeys = ['mood', 'manager-ip', 'pid', 'workerName'] 207 _jobStateListKeys = ['messages', ] 208 209 # FIXME: maybe make Atmosphere and Flow subclass from a ComponentGroup class ? 210 211
212 -class ManagerComponentState(flavors.StateCacheable):
213 """ 214 I represent the state of a component in the manager. 215 I have my own state, and also proxy state from the L{ManagerJobState} 216 when the component is actually created in a worker. 217 218 I have the following keys of my own: 219 220 - name: str, name of the component, unique in the parent 221 - parent: L{ManagerFlowState} or L{ManagerAtmosphereState} 222 - type: str, type of the component 223 - moodPending: int, the mood value the component is being set to 224 - workerRequested: str, name of the worker this component is 225 requested to be started on. 226 - config: dict, the configuration dict for this component 227 228 It also has a special key, 'mood'. This acts as a proxy for the mood 229 in the L{WorkerJobState}, when there is a job attached (the job's copy 230 is authoritative when it connects), and is controlled independently at 231 other times. 232 233 I proxy the following keys from the serialized L{WorkerJobState}: 234 - mood, manager-ip, pid, workerName 235 - messages (list) 236 """ 237
238 - def __init__(self):
239 flavors.StateCacheable.__init__(self) 240 # our additional keys 241 self.addKey('name') 242 self.addKey('type') 243 self.addKey('parent') 244 self.addKey('moodPending') 245 self.addKey('workerRequested') 246 self.addKey('config') # dictionary 247 self.addKey('lastKnownPid') 248 249 # proxied from job state or combined with our state (mood) 250 for k in _jobStateKeys: 251 self.addKey(k) 252 for k in _jobStateListKeys: 253 self.addListKey(k) 254 self._jobState = None
255
256 - def __repr__(self):
257 return "<%s.%s name=%r>" % (self.__module__, 258 self.__class__.__name__, 259 self._dict['name'])
260
261 - def setJobState(self, jobState):
262 """ 263 Set the job state I proxy from. 264 265 @type jobState: L{ManagerJobState} 266 """ 267 self._jobState = jobState 268 for key in _jobStateKeys: 269 # only set non-None values 270 if key == 'mood': 271 continue 272 v = jobState.get(key) 273 if v != None: 274 self.set(key, v) 275 for key in _jobStateListKeys: 276 valueList = jobState.get(key) 277 if valueList != None: 278 for v in valueList: 279 self.append(key, v) 280 # set mood last; see #552 281 self.set('mood', jobState.get('mood')) 282 283 # only proxy keys we want proxied; eaterNames and feederNames 284 # are ignored for example 285 proxiedKeys = _jobStateKeys + _jobStateListKeys 286 287 def proxy(attr): 288 289 def event(state, key, value): 290 if key in proxiedKeys: 291 getattr(self, attr)(key, value)
292 return event
293 294 jobState.addListener(self, set_=proxy('set'), append=proxy('append'), 295 remove=proxy('remove')) 296
297 - def set(self, key, value):
298 # extend set so we can log mood changes 299 if key == 'mood': 300 log.info('componentstate', 'mood of %s changed to %s', 301 self.get('name'), moods.get(value).name) 302 flavors.StateCacheable.set(self, key, value) 303 if key == 'mood' and value == self.get('moodPending'): 304 # we have reached our pending mood 305 self.set('moodPending', None)
306
307 - def setMood(self, moodValue):
308 if self._jobState and moodValue != moods.sad.value: 309 log.warning('componentstate', 'cannot set component mood to ' 310 'something other than sad when we have a ' 311 'jobState -- fix your code!') 312 elif moodValue == self.get('mood'): 313 log.log('componentstate', '%s already in mood %d', 314 self.get('name'), moodValue) 315 else: 316 log.debug('componentstate', 317 'manager sets mood of %s from %s to %d', 318 self.get('name'), self.get('mood'), moodValue) 319 self.set('mood', moodValue)
320
321 - def clearJobState(self, shutdownRequested):
322 """ 323 Remove the job state. 324 """ 325 # Clear messages proxied from job 326 for m in self._jobState.get('messages'): 327 self.remove('messages', m) 328 329 self.set('lastKnownPid', self._jobState.get('pid')) 330 331 self._jobState.removeListener(self) 332 self._jobState = None 333 334 # Clearing a job state means that a component logged out. If the 335 # component logs out due to an explicit manager request, go to 336 # sleeping. Otherwise if the component is sad, leave the mood as 337 # it is, or otherwise go to lost, because it got disconnected 338 # for an unknown reason (probably network related). 339 if shutdownRequested: 340 log.debug('componentstate', "Shutdown was requested, %s" 341 " now sleeping", self.get('name')) 342 self.setMood(moods.sleeping.value) 343 elif self.get('mood') != moods.sad.value: 344 log.debug('componentstate', "Shutdown was NOT requested," 345 " %s now lost, last know pid is: %r", 346 self.get('name'), self.get('lastKnownPid')) 347 self.setMood(moods.lost.value)
348 349
350 -class AdminComponentState(flavors.StateRemoteCache):
351 """ 352 I represent the state of a component in the admin client. 353 See L{ManagerComponentState}. 354 """ 355
356 - def __repr__(self):
357 return "<%s.%s name=%r>" % (self.__module__, 358 self.__class__.__name__, 359 self._dict['name'])
360 361 pb.setUnjellyableForClass(ManagerComponentState, AdminComponentState) 362 363 # state of an existing component running in a job process 364 # exchanged between worker and manager 365 366
367 -class WorkerJobState(flavors.StateCacheable):
368 """ 369 I represent the state of a job in the worker, running a component. 370 371 I have the following keys: 372 373 - mood: int, value of the mood this component is in 374 - ip: string, IP address of the worker 375 - pid: int, PID of the job process 376 - workerName: string, name of the worker I'm running on 377 - messages: list of L{flumotion.common.messages.Message} 378 379 In addition, if I am the state of a FeedComponent, then I also 380 have the following keys: 381 382 - eaterNames: list of feedId being eaten by the eaters 383 - feederNames: list of feedId being fed by the feeders 384 385 @todo: change eaterNames and feederNames to eaterFeedIds and ... 386 """ 387
388 - def __init__(self):
389 flavors.StateCacheable.__init__(self) 390 for k in _jobStateKeys: 391 self.addKey(k) 392 for k in _jobStateListKeys: 393 self.addListKey(k)
394 395
396 -class ManagerJobState(flavors.StateRemoteCache):
397 """ 398 I represent the state of a job in the manager. 399 See L{WorkerJobState}. 400 """ 401 pass
402 403 pb.setUnjellyableForClass(WorkerJobState, ManagerJobState) 404