src/nepi/resources/omf/omf5_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 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19 #         Julien Tribino <julien.tribino@inria.fr>
  20
  21 import ssl
  22 import sys
  23 import time
  24
  25 from nepi.util.logger import Logger
  26
  27 from nepi.resources.omf.omf_client import OMFClient
  28 from nepi.resources.omf.messages_5_4 import MessageHandler
  29
  30 class OMF5API(Logger):
  31     """
  32     .. class:: Class Args :
  33
  34         :param host: Xmpp Server
  35         :type host: str
  36         :param slice: Xmpp Slice
  37         :type slice: str
  38         :param port: Xmpp Port
  39         :type port: str
  40         :param password: Xmpp password
  41         :type password: str
  42         :param xmpp_root: Root of the Xmpp Topic Architecture
  43         :type xmpp_root: str
  44
  45     .. note::
  46
  47        This class is the implementation of an OMF 5.4 API.
  48        Since the version 5.4.1, the Topic Architecture start with OMF_5.4
  49        instead of OMF used for OMF5.3
  50
  51     """
  52     def __init__(self, host, slice, port, password, xmpp_root = None,
  53             exp_id = None):
  54         """
  55         :param host: Xmpp Server
  56         :type host: str
  57         :param slice: Xmpp Slice
  58         :type slice: 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(OMF5API, self).__init__("OMF5API")
  68         self._exp_id = exp_id
  69         self._user = "%s-%s" % (slice, self._exp_id)
  70         self._slice = slice
  71         self._host = host
  72         self._port = port
  73         self._password = password
  74         self._hostnames = []
  75         self._xmpp_root = xmpp_root or "OMF_5.4"
  76
  77         # OMF xmpp client
  78         self._client = None
  79
  80         # message handler
  81         self._message = None
  82
  83         if sys.version_info < (3, 0):
  84             reload(sys)
  85             sys.setdefaultencoding('utf8')
  86
  87         # instantiate the xmpp client
  88         self._init_client()
  89
  90         # register xmpp nodes for the experiment
  91         self._enroll_experiment()
  92         self._enroll_newexperiment()
  93
  94         # register xmpp logger for the experiment
  95         self._enroll_logger()
  96
  97     def _init_client(self):
  98         """ Initialize XMPP Client
  99
 100         """
 101         jid = "%s@%s" % (self._user, self._host)
 102         xmpp = OMFClient(jid, self._password)
 103         # PROTOCOL_SSLv3 required for compatibility with OpenFire
 104         xmpp.ssl_version = ssl.PROTOCOL_SSLv3
 105
 106         if xmpp.connect((self._host, self._port)):
 107             xmpp.process(block=False)
 108             while not xmpp.ready:
 109                 time.sleep(1)
 110             self._client = xmpp
 111             self._message = MessageHandler(self._slice, self._user)
 112         else:
 113             msg = "Unable to connect to the XMPP server."
 114             self.error(msg)
 115             raise RuntimeError(msg)
 116
 117     def _enroll_experiment(self):
 118         """ Create and Subscribe to the Session Topic
 119
 120         """
 121         xmpp_node = self._exp_session_id
 122         self._client.create(xmpp_node)
 123         #print "Create experiment sesion id topics !!"
 124         self._client.subscribe(xmpp_node)
 125         #print "Subscribe to experiment sesion id topics !!"
 126
 127
 128     def _enroll_newexperiment(self):
 129         """ Publish New Experiment Message
 130
 131         """
 132         address = "/%s/%s/%s/%s" % (self._host, self._xmpp_root, self._slice,
 133                 self._user)
 134         #print address
 135         payload = self._message.newexp_function(self._user, address)
 136         slice_sid = "/%s/%s" % (self._xmpp_root, self._slice)
 137         self._client.publish(payload, slice_sid)
 138
 139     def _enroll_logger(self):
 140         """ Create and Subscribe to the Logger Topic
 141
 142         """
 143         xmpp_node = self._logger_session_id
 144         self._client.create(xmpp_node)
 145         self._client.subscribe(xmpp_node)
 146
 147         payload = self._message.log_function("2",
 148                 "nodeHandler::NodeHandler",
 149                 "INFO",
 150                 "OMF Experiment Controller 5.4 (git 529a626)")
 151         self._client.publish(payload, xmpp_node)
 152
 153     def _host_session_id(self, hostname):
 154         """ Return the Topic Name as /xmpp_root/slice/user/hostname
 155
 156         :param hostname: Full hrn of the node
 157         :type hostname: str
 158
 159         """
 160         return "/%s/%s/%s/%s" % (self._xmpp_root, self._slice, self._user,
 161                 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
 203             corresponding to the hostname
 204
 205         :param hostname: Full hrn of the node
 206         :type hostname: str
 207
 208         """
 209         if hostname in self._hostnames:
 210             return
 211
 212         self._hostnames.append(hostname)
 213
 214         xmpp_node =  self._host_session_id(hostname)
 215         self._client.create(xmpp_node)
 216         self._client.subscribe(xmpp_node)
 217
 218         xmpp_node =  self._host_resource_id(hostname)
 219         self._client.subscribe(xmpp_node)
 220
 221         payload = self._message.enroll_function("1", "*", "1", hostname)
 222         self._client.publish(payload, xmpp_node)
 223
 224     def configure(self, hostname, attribute, value):
 225         """ Configure attribute on the node
 226
 227         :param hostname: Full hrn of the node
 228         :type hostname: str
 229         :param attribute: Attribute that need to be configured (
 230             often written as /net/wX/attribute, with X the interface number)
 231         :type attribute: str
 232         :param value: Value of the attribute
 233         :type value: str
 234
 235         """
 236         payload = self._message.configure_function(hostname, value, attribute)
 237         xmpp_node =  self._host_session_id(hostname)
 238         self._client.publish(payload, xmpp_node)
 239
 240
 241     def send_stdin(self, hostname, value, app_id):
 242         """ Send to the stdin of the application the value
 243
 244         :param hostname: Full hrn of the node
 245         :type hostname: str
 246         :param appid: Application Id (Any id that represents in a unique
 247             way the application)
 248         :type appid: str
 249         :param value: parameter to execute in the stdin of the application
 250         :type value: str
 251
 252         """
 253         payload = self._message.stdin_function(hostname, value, app_id)
 254         xmpp_node =  self._host_session_id(hostname)
 255         self._client.publish(payload, xmpp_node)
 256
 257
 258     def execute(self, hostname, app_id, arguments, path, env):
 259         """ Execute command on the node
 260
 261         :param hostname: Full hrn of the node
 262         :type hostname: str
 263         :param app_id: Application Id (Any id that represents in a unique
 264             way the application)
 265         :type app_id: str
 266         :param arguments: Arguments of the application
 267         :type arguments: str
 268         :param path: Path of the application
 269         :type path: str
 270         :param env: Environnement values for the application
 271         :type env: str
 272
 273         """
 274         payload = self._message.execute_function(hostname, app_id, arguments,
 275                 path, env)
 276         xmpp_node =  self._host_session_id(hostname)
 277         self._client.publish(payload, xmpp_node)
 278
 279     def exit(self, hostname, app_id):
 280         """ Kill an application started with OMF
 281
 282         :param hostname: Full hrn of the node
 283         :type hostname: str
 284         :param app_id: Application Id of the application you want to stop
 285         :type app_id: str
 286
 287         """
 288         payload = self._message.exit_function(hostname, app_id)
 289         xmpp_node =  self._host_session_id(hostname)
 290         self._client.publish(payload, xmpp_node)
 291
 292     def release(self, hostname):
 293         """ Delete the session and logger topics. Then disconnect
 294
 295         """
 296         if hostname in self._hostnames:
 297             self.delete(hostname)
 298
 299     def disconnect(self) :
 300         """ Delete the session and logger topics. Then disconnect
 301
 302         """
 303         self._client.delete(self._exp_session_id)
 304         self._client.delete(self._logger_session_id)
 305
 306         time.sleep(1)
 307
 308         # Wait the send queue to be empty before disconnect
 309         self._client.disconnect(wait=True)
 310         msg = " Disconnected from XMPP Server"
 311         self.debug(msg)
 312