src/nepi/testbeds/omf/omf_api.py

   1 import datetime
   2 import logging
   3 import ssl
   4 import sys
   5 import time
   6
   7 from nepi.testbeds.omf.omf_client import OMFClient
   8 from nepi.testbeds.omf.omf_messages import MessageHandler
   9
  10 class OmfAPI(object):
  11     def __init__(self, slice, host, port, password):
  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
  22         self._logger = logging.getLogger("nepi.testbeds.omf")
  23
  24         # OMF xmpp client
  25         self._client = None
  26         # message handler
  27         self._message = None
  28
  29         if sys.version_info < (3, 0):
  30             reload(sys)
  31             sys.setdefaultencoding('utf8')
  32
  33         # instantiate the xmpp client
  34         self._init_client()
  35
  36         # register xmpp nodes for the experiment
  37         self._enroll_experiment()
  38
  39         # register xmpp logger for the experiment
  40         self._enroll_logger()
  41
  42     def _init_client(self):
  43         jid = "%s@%s" % (self._user, self._host)
  44         xmpp = OMFClient(jid, self._password)
  45         # PROTOCOL_SSLv3 required for compatibility with OpenFire
  46         xmpp.ssl_version = ssl.PROTOCOL_SSLv3
  47
  48         if xmpp.connect((self._host, self._port)):
  49             xmpp.process(threaded=True)
  50             while not xmpp.ready:
  51                 time.sleep(1)
  52             self._client = xmpp
  53             self._message = MessageHandler(self._slice, self._user)
  54         else:
  55             msg = "Unable to connect to the XMPP server."
  56             self._logger.error(msg)
  57             raise RuntimeError(msg)
  58
  59     def _enroll_experiment(self):
  60         xmpp_node = self._exp_session_id
  61         self._client.create(xmpp_node)
  62         self._client.subscribe(xmpp_node)
  63
  64         address = "/%s/OMF/%s/%s" % (self._host, self._slice, self._user)
  65         payload = self._message.newexpfunction(self._user, address)
  66         slice_sid = "/OMF/%s" % (self._slice)
  67         self._client.publish(payload, slice_sid)
  68
  69     def _enroll_logger(self):
  70         xmpp_node = self._logger_session_id
  71         self._client.create(xmpp_node)
  72         self._client.subscribe(xmpp_node)
  73
  74         payload = self._message.logfunction("2",
  75                 "nodeHandler::NodeHandler",
  76                 "INFO",
  77                 "OMF Experiment Controller 5.4 (git 529a626)")
  78         self._client.publish(payload, xmpp_node)
  79
  80     def _host_session_id(self, hostname):
  81         return "/OMF/%s/%s/%s" % (self._slice, self._user, hostname)
  82
  83     def _host_resource_id(self, hostname):
  84         return "/OMF/%s/resources/%s" % (self._slice, hostname)
  85
  86     @property
  87     def _exp_session_id(self):
  88         return "/OMF/%s/%s" % (self._slice, self._user)
  89
  90     @property
  91     def _logger_session_id(self):
  92         return "/OMF/%s/%s/LOGGER" % (self._slice, self._user)
  93
  94     def delete(self, hostname):
  95         if not hostname in self._hostnames:
  96             return
  97
  98         self._hostnames.remove(hostname)
  99
 100         xmpp_node = self._host_session_id(hostname)
 101         self._client.delete(xmpp_node)
 102
 103     def enroll_host(self, hostname):
 104         if hostname in self._hostnames:
 105             return
 106
 107         self._hostnames.append(hostname)
 108
 109         xmpp_node =  self._host_session_id(hostname)
 110         self._client.create(xmpp_node)
 111         self._client.subscribe(xmpp_node)
 112
 113         xmpp_node =  self._host_resource_id(hostname)
 114         self._client.subscribe(xmpp_node)
 115
 116         payload = self._message.enrollfunction("1", "*", "1", hostname)
 117         self._client.publish(payload, xmpp_node)
 118
 119     def configure(self, hostname, attribute, value):
 120         payload = self._message.configurefunction(hostname, value, attribute)
 121         xmpp_node =  self._host_session_id(hostname)
 122         self._client.publish(payload, xmpp_node)
 123
 124     def execute(self, hostname, app_id, arguments, path, env):
 125         payload = self._message.executefunction(hostname, app_id, arguments, path, env)
 126         xmpp_node =  self._host_session_id(hostname)
 127         self._client.publish(payload, xmpp_node)
 128
 129     def exit(self, hostname, app_id):
 130         payload = self._message.exitfunction(hostname, app_id)
 131         xmpp_node =  self._host_session_id(hostname)
 132         self._client.publish(payload, xmpp_node)
 133
 134     def disconnect(self):
 135         self._client.delete(self._exp_session_id)
 136         self._client.delete(self._logger_session_id)
 137
 138         for hostname in self._hostnames[:]:
 139             self.delete(hostname)
 140
 141         time.sleep(5)
 142         self._client.disconnect()
 143