/usr/lib/python2.7/dist-packages/LTSPAgent/plugin.py is in ltsp-cluster-agent 0.8-1ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | import logging, time, threading, types, os, sys
from configobj import ConfigObj
#long var, less name clashing :)
DELAY_ATTR="__automatic_threaded_update_delay"
class Plugin(object):
def __init__(self):
"""Code execute before parent sets properties"""
self.LOGGER = logging.getLogger(self.__class__.__name__)
self._update_threads = {}
self.__init_config()
def __init_config(self):
if hasattr(self,'config'):
return
# Find and load a configuration file
if os.path.exists(os.getcwd()+'/config/agent.d/%s.conf' % self.__class__.__name__):
self.config = ConfigObj(os.getcwd()+'/config/agent.d/%s.conf' % self.__class__.__name__)
elif os.path.exists('/etc/ltsp/agent.d/%s.conf' % self.__class__.__name__):
self.config = ConfigObj('/etc/ltsp/agent.d/%s.conf' % self.__class__.__name__)
elif os.path.exists('%s\\ltsp\\config\\agent.d\\%s.conf' % (sys.prefix,self.__class__.__name__)):
self.config = ConfigObj('%s\\ltsp\\config\\agent.d\\%s.conf' % (sys.prefix,self.__class__.__name__))
else:
self.config = {}
def __init_threads(self):
#Look into our functions to see if they were decorated with @auto_update
for key in dir(self):
val = getattr(self, key)
if isinstance(val, (types.MethodType,types.FunctionType,)) and \
hasattr(val, DELAY_ATTR):
self.LOGGER.info("Plugin %s registered timed function: %s", self.__class__.__name__, key)
self._update_threads[key] = updateThread(val, getattr(val, DELAY_ATTR))()
def init_plugin(self):
"""Code executed once all properties have been set from parent"""
self.__init_threads()
def rpc_functions(self):
return ()
def start_threads(self):
for name, thread in self._update_threads.iteritems():
thread.start()
def stop_threads(self):
for name, thread in self._update_threads.iteritems():
thread.stop = True
def auto_update(seconds):
def __deco(f):
setattr(f, DELAY_ATTR, seconds)
return f
return __deco
class PluginUpdateThread(object):
stop = False
class updateThread(object):
def __new__(cls, update_f, delay_secs):
def run(self):
interval = delay_secs
while not self.stop:
start = time.time()
update_f()
duration=time.time()-start
if (interval-duration) > 0:
time.sleep(interval-duration)
return type("updateThread%d" % id(update_f), \
(threading.Thread, PluginUpdateThread),
dict(run=run),
)
|