src/nepi/resources/omf/omf_api.py

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