src/nepi/resources/omf/omf_api.py

   1 import datetime
   2 import logging
   3 import ssl
   4 import sys
   5 import time
   6 import hashlib
   7 import neco
   8 import threading
   9
  10 from neco.resources.omf.omf_client import OMFClient
  11 from neco.resources.omf.omf_messages_5_4 import MessageHandler
  12
  13 class OMFAPI(object):
  14     """
  15     .. class:: Class Args :
  16
  17         :param slice: Xmpp Slice
  18         :type slice: Str
  19         :param host: Xmpp Server
  20         :type host: Str
  21         :param port: Xmpp Port
  22         :type port: Str
  23         :param password: Xmpp password
  24         :type password: Str
  25         :param xmpp_root: Root of the Xmpp Topic Architecture
  26         :type xmpp_root: Str
  27
  28     .. note::
  29
  30        This class is the implementation of an OMF 5.4 API. Since the version 5.4.1, the Topic Architecture start with OMF_5.4 instead of OMF used for OMF5.3
  31
  32     """
  33     def __init__(self, slice, host, port, password, xmpp_root = None):
  34         """
  35
  36         :param slice: Xmpp Slice
  37         :type slice: Str
  38         :param host: Xmpp Server
  39         :type host: Str
  40         :param port: Xmpp Port
  41         :type port: Str
  42         :param password: Xmpp password
  43         :type password: Str
  44         :param xmpp_root: Root of the Xmpp Topic Architecture
  45         :type xmpp_root: Str
  46
  47         """
  48         date = datetime.datetime.now().strftime("%Y-%m-%dt%H.%M.%S")
  49         tz = -time.altzone if time.daylight != 0 else -time.timezone
  50         date += "%+06.2f" % (tz / 3600) # timezone difference is in seconds
  51         self._user = "%s-%s" % (slice, date)
  52         self._slice = slice
  53         self._host = host
  54         self._port = port
  55         self._password = password
  56         self._hostnames = []
  57         self._xmpp_root = xmpp_root or "OMF_5.4"
  58
  59         self._logger = logging.getLogger("neco.omf.omfApi    ")
  60         self._logger.setLevel(neco.LOGLEVEL)
  61
  62         # OMF xmpp client
  63         self._client = None
  64         # message handler
  65         self._message = None
  66
  67         if sys.version_info < (3, 0):
  68             reload(sys)
  69             sys.setdefaultencoding('utf8')
  70
  71         # instantiate the xmpp client
  72         self._init_client()
  73
  74         # register xmpp nodes for the experiment
  75         self._enroll_experiment()
  76         self._enroll_newexperiment()
  77
  78         # register xmpp logger for the experiment
  79         self._enroll_logger()
  80
  81     def _init_client(self):
  82         """ Initialize XMPP Client
  83
  84         """
  85         jid = "%s@%s" % (self._user, self._host)
  86         xmpp = OMFClient(jid, self._password)
  87         # PROTOCOL_SSLv3 required for compatibility with OpenFire
  88         xmpp.ssl_version = ssl.PROTOCOL_SSLv3
  89
  90         if xmpp.connect((self._host, self._port)):
  91             xmpp.process(block=False)
  92             while not xmpp.ready:
  93                 time.sleep(1)
  94             self._client = xmpp
  95             self._message = MessageHandler(self._slice, self._user)
  96         else:
  97             msg = "Unable to connect to the XMPP server."
  98             self._logger.error(msg)
  99             raise RuntimeError(msg)
 100
 101     def _enroll_experiment(self):
 102         """ Create and Subscribe to the Session Topic
 103
 104         """
 105         xmpp_node = self._exp_session_id
 106         self._client.create(xmpp_node)
 107         #print "Create experiment sesion id topics !!"
 108         self._client.subscribe(xmpp_node)
 109         #print "Subscribe to experiment sesion id topics !!"
 110
 111
 112     def _enroll_newexperiment(self):
 113         """ Publish New Experiment Message
 114
 115         """
 116         address = "/%s/%s/%s/%s" % (self._host, self._xmpp_root, self._slice, self._user)
 117         #print address
 118         payload = self._message.newexp_function(self._user, address)
 119         slice_sid = "/%s/%s" % (self._xmpp_root, self._slice)
 120         self._client.publish(payload, slice_sid)
 121
 122     def _enroll_logger(self):
 123         """ Create and Subscribe to the Logger Topic
 124
 125         """
 126         xmpp_node = self._logger_session_id
 127         self._client.create(xmpp_node)
 128         self._client.subscribe(xmpp_node)
 129
 130         payload = self._message.log_function("2",
 131                 "nodeHandler::NodeHandler",
 132                 "INFO",
 133                 "OMF Experiment Controller 5.4 (git 529a626)")
 134         self._client.publish(payload, xmpp_node)
 135
 136     def _host_session_id(self, hostname):
 137         """ Return the Topic Name as /xmpp_root/slice/user/hostname
 138
 139         :param hostname: Full hrn of the node
 140         :type hostname: str
 141
 142         """
 143         return "/%s/%s/%s/%s" % (self._xmpp_root, self._slice, self._user, hostname)
 144
 145     def _host_resource_id(self, hostname):
 146         """ Return the Topic Name as /xmpp_root/slice/resources/hostname
 147
 148         :param hostname: Full hrn of the node
 149         :type hostname: str
 150
 151         """
 152         return "/%s/%s/resources/%s" % (self._xmpp_root, self._slice, hostname)
 153
 154     @property
 155     def _exp_session_id(self):
 156         """ Return the Topic Name as /xmpp_root/slice/user
 157
 158         """
 159         return "/%s/%s/%s" % (self._xmpp_root, self._slice, self._user)
 160
 161     @property
 162     def _logger_session_id(self):
 163         """ Return the Topic Name as /xmpp_root/slice/LOGGER
 164
 165         """
 166         return "/%s/%s/%s/LOGGER" % (self._xmpp_root, self._slice, self._user)
 167
 168     def delete(self, hostname):
 169         """ Delete the topic corresponding to the hostname for this session
 170
 171         :param hostname: Full hrn of the node
 172         :type hostname: str
 173
 174         """
 175         if not hostname in self._hostnames:
 176             return
 177
 178         self._hostnames.remove(hostname)
 179
 180         xmpp_node = self._host_session_id(hostname)
 181         self._client.delete(xmpp_node)
 182
 183     def enroll_host(self, hostname):
 184         """ Create and Subscribe to the session topic and the resources corresponding to the hostname
 185
 186         :param hostname: Full hrn of the node
 187         :type hostname: str
 188
 189         """
 190         if hostname in self._hostnames:
 191             return
 192
 193         self._hostnames.append(hostname)
 194
 195         xmpp_node =  self._host_session_id(hostname)
 196         self._client.create(xmpp_node)
 197         self._client.subscribe(xmpp_node)
 198
 199         xmpp_node =  self._host_resource_id(hostname)
 200         self._client.subscribe(xmpp_node)
 201
 202         payload = self._message.enroll_function("1", "*", "1", hostname)
 203         self._client.publish(payload, xmpp_node)
 204
 205     def configure(self, hostname, attribute, value):
 206         """ Configure attribute on the node
 207
 208         :param hostname: Full hrn of the node
 209         :type hostname: str
 210         :param attribute: Attribute that need to be configured (often written as /net/wX/attribute, with X the interface number)
 211         :type attribute: str
 212         :param value: Value of the attribute
 213         :type value: str
 214
 215         """
 216         payload = self._message.configure_function(hostname, value, attribute)
 217         xmpp_node =  self._host_session_id(hostname)
 218         self._client.publish(payload, xmpp_node)
 219
 220     def execute(self, hostname, app_id, arguments, path, env):
 221         """ Execute command on the node
 222
 223         :param hostname: Full hrn of the node
 224         :type hostname: str
 225         :param app_id: Application Id (Any id that represents in a unique way the application)
 226         :type app_id: str
 227         :param arguments: Arguments of the application
 228         :type arguments: str
 229         :param path: Path of the application
 230         :type path: str
 231         :param env: Environnement values for the application
 232         :type env: str
 233
 234         """
 235         payload = self._message.execute_function(hostname, app_id, arguments, path, env)
 236         xmpp_node =  self._host_session_id(hostname)
 237         self._client.publish(payload, xmpp_node)
 238
 239     def exit(self, hostname, app_id):
 240         """ Kill an application started with OMF
 241
 242         :param hostname: Full hrn of the node
 243         :type hostname: str
 244         :param app_id: Application Id of the application you want to stop
 245         :type app_id: str
 246
 247         """
 248         payload = self._message.exit_function(hostname, app_id)
 249         xmpp_node =  self._host_session_id(hostname)
 250         self._client.publish(payload, xmpp_node)
 251
 252     def release(self, hostname):
 253         """ Delete the session and logger topics. Then disconnect
 254
 255         """
 256         if hostname in self._hostnames:
 257             self.delete(hostname)
 258
 259     def disconnect(self) :
 260         """ Delete the session and logger topics. Then disconnect
 261
 262         """
 263         self._client.delete(self._exp_session_id)
 264         self._client.delete(self._logger_session_id)
 265
 266         time.sleep(1)
 267
 268         # Wait the send queue to be empty before disconnect
 269         self._client.disconnect(wait=True)
 270         self._logger.debug(" Disconnected from XMPP Server")
 271
 272
 273 class OMFAPIFactory(object):
 274     """
 275     .. note::
 276
 277         It allows the different RM to use the same xmpp client if they use the same credentials.
 278         For the moment, it is focused on Xmpp.
 279
 280     """
 281     # use lock to avoid concurrent access to the Api list at the same times by 2 different threads
 282     lock = threading.Lock()
 283     _apis = dict()
 284
 285     @classmethod
 286     def get_api(cls, slice, host, port, password):
 287         """ Get an Api
 288
 289         :param slice: Xmpp Slice Name
 290         :type slice: str
 291         :param host: Xmpp Server Adress
 292         :type host: str
 293         :param port: Xmpp Port (Default : 5222)
 294         :type port: str
 295         :param password: Xmpp Password
 296         :type password: str
 297
 298         """
 299         if slice and host and port and password:
 300             key = cls._make_key(slice, host, port, password)
 301             cls.lock.acquire()
 302             if key in cls._apis:
 303                 cls._apis[key]['cnt'] += 1
 304                 cls.lock.release()
 305                 return cls._apis[key]['api']
 306             else :
 307                 omf_api = cls.create_api(slice, host, port, password)
 308                 cls.lock.release()
 309                 return omf_api
 310         return None
 311
 312     @classmethod
 313     def create_api(cls, slice, host, port, password):
 314         """ Create an API if this one doesn't exist yet with this credentials
 315
 316         :param slice: Xmpp Slice Name
 317         :type slice: str
 318         :param host: Xmpp Server Adress
 319         :type host: str
 320         :param port: Xmpp Port (Default : 5222)
 321         :type port: str
 322         :param password: Xmpp Password
 323         :type password: str
 324
 325         """
 326         omf_api = OMFAPI(slice, host, port, password)
 327         key = cls._make_key(slice, host, port, password)
 328         cls._apis[key] = {}
 329         cls._apis[key]['api'] = omf_api
 330         cls._apis[key]['cnt'] = 1
 331         return omf_api
 332
 333     @classmethod
 334     def release_api(cls, slice, host, port, password):
 335         """ Release an API with this credentials
 336
 337         :param slice: Xmpp Slice Name
 338         :type slice: str
 339         :param host: Xmpp Server Adress
 340         :type host: str
 341         :param port: Xmpp Port (Default : 5222)
 342         :type port: str
 343         :param password: Xmpp Password
 344         :type password: str
 345
 346         """
 347         if slice and host and port and password:
 348             key = cls._make_key(slice, host, port, password)
 349             if key in cls._apis:
 350                 cls._apis[key]['cnt'] -= 1
 351                 #print "Api Counter : " + str(cls._apis[key]['cnt'])
 352                 if cls._apis[key]['cnt'] == 0:
 353                     omf_api = cls._apis[key]['api']
 354                     omf_api.disconnect()
 355
 356
 357     @classmethod
 358     def _make_key(cls, *args):
 359         """ Hash the credentials in order to create a key
 360
 361         :param args: list of arguments used to create the hash (user, host, port, ...)
 362         :type args: list of args
 363
 364         """
 365         skey = "".join(map(str, args))
 366         return hashlib.md5(skey).hexdigest()
 367
 368
 369