src/neco/resources/omf/omf_api.py

   1 import datetime
   2 import logging
   3 import ssl
   4 import sys
   5 import time
   6
   7 from neco.resources.omf.omf_client import OMFClient
   8 from neco.resources.omf.omf_messages_5_4 import MessageHandler
   9
  10 class OMFAPI(object):
  11     def __init__(self, slice, host, port, password, xmpp_root = None):
  12         date = datetime.datetime.now().strftime("%Y-%m-%dt%H.%M.%S")
  13         tz = -time.altzone if time.daylight != 0 else -time.timezone
  14         date += "%+06.2f" % (tz / 3600) # timezone difference is in seconds
  15         self._user = "%s-%s" % (slice, date)
  16         self._slice = slice
  17         self._host = host
  18         self._port = port
  19         self._password = password
  20         self._hostnames = []
  21         self._xmpp_root = xmpp_root or "OMF_5.4"
  22
  23         self._logger = logging.getLogger("neco.resources.omf")
  24
  25         # OMF xmpp client
  26         self._client = None
  27         # message handler
  28         self._message = None
  29
  30         if sys.version_info < (3, 0):
  31             reload(sys)
  32             sys.setdefaultencoding('utf8')
  33
  34         # instantiate the xmpp client
  35         self._init_client()
  36
  37         # register xmpp nodes for the experiment
  38         self._enroll_experiment()
  39         self._enroll_newexperiment()
  40
  41         # register xmpp logger for the experiment
  42         self._enroll_logger()
  43
  44     def _init_client(self):
  45         jid = "%s@%s" % (self._user, self._host)
  46         xmpp = OMFClient(jid, self._password)
  47         # PROTOCOL_SSLv3 required for compatibility with OpenFire
  48         xmpp.ssl_version = ssl.PROTOCOL_SSLv3
  49
  50         if xmpp.connect((self._host, self._port)):
  51             xmpp.process(threaded=True)
  52             while not xmpp.ready:
  53                 time.sleep(1)
  54             self._client = xmpp
  55             self._message = MessageHandler(self._slice, self._user)
  56         else:
  57             msg = "Unable to connect to the XMPP server."
  58             self._logger.error(msg)
  59             raise RuntimeError(msg)
  60
  61     def _enroll_experiment(self):
  62         xmpp_node = self._exp_session_id
  63         self._client.create(xmpp_node)
  64         #print "Create experiment sesion id topics !!"
  65         self._client.subscribe(xmpp_node)
  66         #print "Subscribe to experiment sesion id topics !!"
  67
  68
  69     def _enroll_newexperiment(self):
  70         address = "/%s/%s/%s/%s" % (self._host, self._xmpp_root, self._slice, self._user)
  71         print address
  72         payload = self._message.newexpfunction(self._user, address)
  73         slice_sid = "/%s/%s" % (self._xmpp_root, self._slice)
  74         self._client.publish(payload, slice_sid)
  75
  76     def _enroll_logger(self):
  77         xmpp_node = self._logger_session_id
  78         self._client.create(xmpp_node)
  79         self._client.subscribe(xmpp_node)
  80
  81         payload = self._message.logfunction("2",
  82                 "nodeHandler::NodeHandler",
  83                 "INFO",
  84                 "OMF Experiment Controller 5.4 (git 529a626)")
  85         self._client.publish(payload, xmpp_node)
  86
  87     def _host_session_id(self, hostname):
  88         return "/%s/%s/%s/%s" % (self._xmpp_root, self._slice, self._user, hostname)
  89
  90     def _host_resource_id(self, hostname):
  91         return "/%s/%s/resources/%s" % (self._xmpp_root, self._slice, hostname)
  92
  93     @property
  94     def _exp_session_id(self):
  95         return "/%s/%s/%s" % (self._xmpp_root, self._slice, self._user)
  96
  97     @property
  98     def _logger_session_id(self):
  99         return "/%s/%s/%s/LOGGER" % (self._xmpp_root, self._slice, self._user)
 100
 101     def delete(self, hostname):
 102         if not hostname in self._hostnames:
 103             return
 104
 105         self._hostnames.remove(hostname)
 106
 107         xmpp_node = self._host_session_id(hostname)
 108         self._client.delete(xmpp_node)
 109
 110     def enroll_host(self, hostname):
 111         if hostname in self._hostnames:
 112             return
 113
 114         self._hostnames.append(hostname)
 115
 116         xmpp_node =  self._host_session_id(hostname)
 117         self._client.create(xmpp_node)
 118         self._client.subscribe(xmpp_node)
 119
 120         xmpp_node =  self._host_resource_id(hostname)
 121         self._client.subscribe(xmpp_node)
 122
 123         payload = self._message.enrollfunction("1", "*", "1", hostname)
 124         self._client.publish(payload, xmpp_node)
 125
 126     def configure(self, hostname, attribute, value):
 127         payload = self._message.configurefunction(hostname, value, attribute)
 128         xmpp_node =  self._host_session_id(hostname)
 129         self._client.publish(payload, xmpp_node)
 130
 131     def execute(self, hostname, app_id, arguments, path, env):
 132         payload = self._message.executefunction(hostname, app_id, arguments, path, env)
 133         xmpp_node =  self._host_session_id(hostname)
 134         self._client.publish(payload, xmpp_node)
 135
 136     def exit(self, hostname, app_id):
 137         payload = self._message.exitfunction(hostname, app_id)
 138         xmpp_node =  self._host_session_id(hostname)
 139         self._client.publish(payload, xmpp_node)
 140
 141     def disconnect(self):
 142         self._client.delete(self._exp_session_id)
 143         self._client.delete(self._logger_session_id)
 144
 145         for hostname in self._hostnames[:]:
 146             self.delete(hostname)
 147
 148         time.sleep(1)
 149         self._client.disconnect()
 150
 151
 152 class OMFAPIFactory(object):
 153     _Api = dict()
 154
 155     @classmethod
 156     def get_api(cls, slice, host, port, password):
 157         if slice and host and port and password:
 158             key = cls._hash_api(slice, host, port)
 159             if key in cls._Api:
 160                 return cls._Api[key]
 161             else :
 162                 return cls.create_api(slice, host, port, password)
 163         return None
 164
 165     @classmethod
 166     def create_api(cls, slice, host, port, password):
 167         OmfApi = OMFAPI(slice, host, port, password)
 168         key = cls._hash_api(slice, host, port)
 169         cls._Api[key] = OmfApi
 170         return OmfApi
 171
 172     @classmethod
 173     def _hash_api(cls, slice, host, port):
 174         res = slice + "_" + host + "_" + port
 175         return res
 176
 177
 178
 179
 180