Source Code for Module Ganga.Runtime.spyware

 1  from Ganga.Utility.Config import getConfig 
 2  config = getConfig('Configuration') 
 3  config.addOption('UsageMonitoringURL', "http://gangamon.cern.ch:8888/apmon/ganga.conf",'MonALISA configuration file used to setup the destination of usage messages') 
 4  config.addOption('UsageMonitoringMSG',True,"enable usage monitoring through MSG server defined in MSGMS configuration") 
 5   
 6  monitor = None 
 7   
 8  import time, os.path 
 9   
10 -def ganga_started(session_type,**extended_attributes): 
11      host = getConfig('System')['GANGA_HOSTNAME'] 
12      version = getConfig('System')['GANGA_VERSION'] 
13      user = getConfig('Configuration')['user'] 
14      runtime_packages = ':'.join(map(os.path.basename,filter(lambda x:x, config['RUNTIME_PATH'].split(':'))))     
15      start = long(time.time()*1000) 
16   
17      usage_message = {'user':user,'host':host,'start':start,'session':session_type,'runtime_packages':runtime_packages,'version':version} 
18   
19      usage_message.update(extended_attributes) 
20   
21      if config['UsageMonitoringURL']: 
22          import ApMon.apmon 
23          global monitor 
24          # the ApMon constructor may start background threads to refresh the configuration from URL 
25          # NOTE: the configuration (including defaultLogLevel) is overriden from the config file specified in URL 
26          monitor = ApMon.apmon.ApMon(config['UsageMonitoringURL'], defaultLogLevel=ApMon.apmon.Logger.FATAL) 
27          monitor.sendParameters('GangaUsage','%s@%s_%s'%(user,host,start),usage_message) 
28          # stop any background threads started by the ApMon constructor 
29          monitor.free() 
30   
31      if config['UsageMonitoringMSG']: 
32          from Ganga.Lib.MonitoringServices.MSGMS import MSGUtil 
33          msg_config = getConfig('MSGMS') 
34          p = MSGUtil.createPublisher( 
35              msg_config['server'], 
36              msg_config['port'], 
37              msg_config['username'], 
38              msg_config['password']) 
39          # start publisher thread and enqueue usage message for sending 
40          p.start() 
41          p.send(msg_config['usage_message_destination'],repr(usage_message),{'persistent':'true'}) 
42          # ask publisher thread to stop. it will send queued message anyway. 
43          p.stop() 
44   
45 -def ganga_job_submitted(application_name, backend_name, plain_job, master_job, sub_jobs): 
46      host = getConfig('System')['GANGA_HOSTNAME'] 
47      user = getConfig('Configuration')['user'] 
48      runtime_packages = ':'.join(map(os.path.basename,filter(lambda x:x, config['RUNTIME_PATH'].split(':'))))     
49      start = long(time.time()*1000) 
50   
51      job_submitted_message = {'application':application_name, 'backend':backend_name, 'user':user, 'host':host, 'start':start, 'plain_job':plain_job, 'master_job':master_job, 'sub_jobs':sub_jobs, 'runtime_packages':runtime_packages} 
52   
53      if config['UsageMonitoringMSG']: 
54          from Ganga.Lib.MonitoringServices.MSGMS import MSGUtil 
55          msg_config = getConfig('MSGMS') 
56          p = MSGUtil.createPublisher( 
57              msg_config['server'], 
58              msg_config['port'], 
59              msg_config['username'], 
60              msg_config['password']) 
61   
62          # start publisher thread and enqueue usage message for sending 
63          p.start() 
64          p.send(msg_config['job_submission_message_destination'],repr(job_submitted_message),{'persistent':'true'}) 
65          #p.send('/queue/test.ganga.jobsubmission',repr(job_submitted_message),{'persistent':'true'}) 
66          # ask publisher thread to stop. it will send queued message anyway. 
67          p.stop() 
68