# Author: Alina Quereilhac <alina.quereilhac@inria.fr>
# Julien Tribino <julien.tribino@inria.fr>
+ import os
+
from nepi.execution.resource import ResourceManager, clsinit_copy, \
ResourceState, reschedule_delay
from nepi.execution.attribute import Attribute, Flags
from nepi.resources.omf.omf_resource import ResourceGateway, OMFResource
- from nepi.resources.omf.node import OMFNode
- from nepi.resources.omf.omf_api import OMFAPIFactory
+ from nepi.resources.omf.node import OMFNode, confirmation_counter, reschedule_check
+ from nepi.resources.omf.omf_api_factory import OMFAPIFactory
from nepi.util import sshfuncs
""" Register the attributes of an OMF application
"""
- appid = Attribute("appid", "Name of the application")
- path = Attribute("path", "Path of the application")
- args = Attribute("args", "Argument of the application")
+ command = Attribute("command", "Command to execute")
env = Attribute("env", "Environnement variable of the application")
+
+ # For OMF 5:
+ appid = Attribute("appid", "Name of the application")
stdin = Attribute("stdin", "Input of the application", default = "")
sources = Attribute("sources", "Sources of the application",
- flags = Flags.ExecReadOnly)
+ flags = Flags.Design)
sshuser = Attribute("sshUser", "user to connect with ssh",
- flags = Flags.ExecReadOnly)
+ flags = Flags.Design)
sshkey = Attribute("sshKey", "key to use for ssh",
- flags = Flags.ExecReadOnly)
+ flags = Flags.Design)
+
cls._register_attribute(appid)
- cls._register_attribute(path)
- cls._register_attribute(args)
+ cls._register_attribute(command)
cls._register_attribute(env)
cls._register_attribute(stdin)
cls._register_attribute(sources)
"""
super(OMFApplication, self).__init__(ec, guid)
+ self.set('command', "")
self.set('appid', "")
- self.set('path', "")
- self.set('args', "")
+ self._path= ""
+ self._args = ""
self.set('env', "")
self._node = None
self._omf_api = None
+ self._topic_app = None
+ self.create_id = None
+ self._create_cnt = 0
+ self._start_cnt = 0
+ self.release_id = None
+ self._release_cnt = 0
self.add_set_hook()
+ def _init_command(self):
+ comm = self.get('command').split(' ')
+ self._path= comm[0]
+ if len(comm)>1:
+ self._args = ' '.join(comm[1:])
+
@property
def exp_id(self):
return self.ec.exp_id
def stdin_hook(self, old_value, new_value):
""" Set a hook to the stdin attribute in order to send a message at each time
- the value of this parameter is changed
+ the value of this parameter is changed. Used ofr OMF 5.4 only
"""
self._omf_api.send_stdin(self.node.get('hostname'), new_value, self.get('appid'))
return new_value
def add_set_hook(self):
- """ Initialize the hooks
+ """ Initialize the hooks for OMF 5.4 only
"""
attr = self._attrs["stdin"]
def do_deploy(self):
""" Deploy the RM. It means nothing special for an application
for now (later it will be upload sources, ...)
- It becomes DEPLOYED after getting the xmpp client.
+ It becomes DEPLOYED after the topic for the application has been created
"""
+ if not self.node or self.node.state < ResourceState.READY:
+ self.debug("---- RESCHEDULING DEPLOY ---- node state %s "
+ % self.node.state )
+ self.ec.schedule(reschedule_delay, self.deploy)
+ return
- self.set('xmppSlice',self.node.get('xmppSlice'))
- self.set('xmppHost',self.node.get('xmppHost'))
+ self._init_command()
+
+ self.set('xmppUser',self.node.get('xmppUser'))
+ self.set('xmppServer',self.node.get('xmppServer'))
self.set('xmppPort',self.node.get('xmppPort'))
self.set('xmppPassword',self.node.get('xmppPassword'))
+ self.set('version',self.node.get('version'))
- if not (self.get('xmppSlice') and self.get('xmppHost')
- and self.get('xmppPort') and self.get('xmppPassword')):
- msg = "Credentials are not initialzed. XMPP Connections impossible"
+ if not self.get('xmppServer'):
+ msg = "XmppServer is not initialzed. XMPP Connections impossible"
self.error(msg)
raise RuntimeError, msg
- if not self._omf_api :
- self._omf_api = OMFAPIFactory.get_api(self.get('xmppSlice'),
- self.get('xmppHost'), self.get('xmppPort'),
- self.get('xmppPassword'), exp_id = self.exp_id)
+ if not (self.get('xmppUser') or self.get('xmppPort')
+ or self.get('xmppPassword')):
+ msg = "Credentials are not all initialzed. Default values will be used"
+ self.warn(msg)
- if self.get('sources'):
- gateway = ResourceGateway.AMtoGateway[self.get('xmppHost')]
- user = self.get('sshUser') or self.get('xmppSlice')
- dst = user + "@"+ gateway + ":"
- (out, err), proc = sshfuncs.rcopy(self.get('sources'), dst)
+ if not self.get('command') :
+ msg = "Application's Command is not initialized"
+ self.error(msg)
+ raise RuntimeError, msg
+
+ if not self._omf_api :
+ self._omf_api = OMFAPIFactory.get_api(self.get('version'),
+ self.get('xmppServer'), self.get('xmppUser'), self.get('xmppPort'),
+ self.get('xmppPassword'), exp_id = self.exp_id)
+
+ if self.get('version') == "5":
+ if self.get('sources'):
+ gateway = ResourceGateway.AMtoGateway[self.get('xmppServer')]
+ user = self.get('sshUser') or self.get('xmppUser')
+ dst = user + "@"+ gateway + ":"
+ (out, err), proc = sshfuncs.rcopy(self.get('sources'), dst)
+ else :
+ # For OMF 6 :
+ if not self.create_id:
+ props = {}
+ if self.get('command'):
+ props['application:binary_path'] = self.get('command')
+ props['application:hrn'] = self.get('command')
+ props['application:membership'] = self._topic_app
+ props['application:type'] = "application"
+
+ self.create_id = os.urandom(16).encode('hex')
+ self._omf_api.frcp_create( self.create_id, self.node.get('hostname'), "application", props = props)
+
+ if self._create_cnt > confirmation_counter:
+ msg = "Couldn't retrieve the confirmation of the creation"
+ self.error(msg)
+ raise RuntimeError, msg
+
+ uid = self.check_deploy(self.create_id)
+ if not uid:
+ self._create_cnt +=1
+ self.ec.schedule(reschedule_check, self.deploy)
+ return
+
+ self._topic_app = uid
+ self._omf_api.enroll_topic(self._topic_app)
super(OMFApplication, self).do_deploy()
+ def check_deploy(self, cid):
+ """ Check, through the mail box in the parser,
+ if the confirmation of the creation has been received
+
+ :param cid: the id of the original message
+ :type guid: string
+
+ """
+ uid = self._omf_api.check_mailbox("create", cid)
+ if uid :
+ return uid
+ return False
+
def do_start(self):
""" Start the RM. It means : Send Xmpp Message Using OMF protocol
to execute the application.
- It becomes STARTED before the messages are sent (for coordination)
"""
- if not (self.get('appid') and self.get('path')) :
- msg = "Application's information are not initialized"
- self.error(msg)
- raise RuntimeError, msg
- if not self.get('args'):
- self.set('args', " ")
if not self.get('env'):
self.set('env', " ")
- # Some information to check the information in parameter
- msg = " " + self.get_rtype() + " ( Guid : " + str(self._guid) +") : " + \
- self.get('appid') + " : " + self.get('path') + " : " + \
- self.get('args') + " : " + self.get('env')
- self.info(msg)
+ if self.get('version') == "5":
+ # Some information to check the command for OMF5
+ msg = " " + self.get_rtype() + " ( Guid : " + str(self._guid) +") : " + \
+ self.get('appid') + " : " + self._path + " : " + \
+ self._args + " : " + self.get('env')
+ self.debug(msg)
- self._omf_api.execute(self.node.get('hostname'),self.get('appid'), \
- self.get('args'), self.get('path'), self.get('env'))
+ self._omf_api.execute(self.node.get('hostname'),self.get('appid'), \
+ self._args, self._path, self.get('env'))
+ else:
+ #For OMF 6
+ if self._start_cnt == 0:
+ props = {}
+ props['state'] = "running"
+
+ guards = {}
+ guards['type'] = "application"
+ guards['name'] = self.get('command')
+
+ self._omf_api.frcp_configure(self._topic_app, props = props, guards = guards )
+
+ if self._start_cnt > confirmation_counter:
+ msg = "Couldn't retrieve the confirmation that the application started"
+ self.error(msg)
+ raise RuntimeError, msg
+
+ res = self.check_start(self._topic_app)
+ if not res:
+ self._start_cnt +=1
+ self.ec.schedule(reschedule_check, self.start)
+ return
super(OMFApplication, self).do_start()
+ def check_start(self, uid):
+ """ Check, through the mail box in the parser,
+ if the confirmation of the start has been received
+
+ :param uid: the id of the original message
+ :type guid: string
+
+ """
+ res = self._omf_api.check_mailbox("started", uid)
+ if res :
+ return True
+ return False
+
def do_stop(self):
""" Stop the RM. It means : Send Xmpp Message Using OMF protocol to
kill the application.
State is set to STOPPED after the message is sent.
"""
-
- self._omf_api.exit(self.node.get('hostname'),self.get('appid'))
+ if self.get('version') == 5:
+ self._omf_api.exit(self.node.get('hostname'),self.get('appid'))
super(OMFApplication, self).do_stop()
+ def check_release(self, cid):
+ """ Check, through the mail box in the parser,
+ if the confirmation of the release has been received
+
+ :param cid: the id of the original message
+ :type guid: string
+
+ """
+ res = self._omf_api.check_mailbox("release", cid)
+ if res :
+ return res
+ return False
+
def do_release(self):
""" Clean the RM at the end of the experiment and release the API.
"""
if self._omf_api:
- OMFAPIFactory.release_api(self.get('xmppSlice'),
- self.get('xmppHost'), self.get('xmppPort'),
- self.get('xmppPassword'), exp_id = self.exp_id)
+ if self.get('version') == "6":
+ if not self.release_id:
+ self.release_id = os.urandom(16).encode('hex')
+ self._omf_api.frcp_release( self.release_id, self.node.get('hostname'),self._topic_app, res_id=self._topic_app)
+
+ if self._release_cnt < confirmation_counter:
+ cid = self.check_release(self.release_id)
+ if not cid:
+ self._release_cnt +=1
+ self.ec.schedule(reschedule_check, self.release)
+ return
+ else:
+ msg = "Couldn't retrieve the confirmation of the release"
+ self.error(msg)
+
+
+ OMFAPIFactory.release_api(self.get('version'),
+ self.get('xmppServer'), self.get('xmppUser'), self.get('xmppPort'),
+ self.get('xmppPassword'), exp_id = self.exp_id)
super(OMFApplication, self).do_release()
# Julien Tribino <julien.tribino@inria.fr>
from nepi.util.logger import Logger
-
+ from nepi.resources.omf.omf6_parser import OMF6Parser
-
try:
import sleekxmpp
from sleekxmpp.exceptions import IqError, IqTimeout
class BaseOMFClient(sleekxmpp.ClientXMPP):
pass
except ImportError:
- print "SleekXMPP is not installed. Without this library, \n" + \
- " You will be not able to use OMF Resources \n"+ \
- " If you want to install SleekXmpp : \n"+ \
- " git clone -b develop git://github.com/fritzy/SleekXMPP.git \n"+ \
- " cd SleekXMPP \n"+ \
- "sudo python setup.py install\n"
+ msg = ("SleekXMPP is not installed. Without this library "
+ "you will be not able to use OMF Resources "
+ "if you want to install SleekXmpp: \n"
+ " git clone -b develop git://github.com/fritzy/SleekXMPP.git \n"
+ " cd SleekXMPP \n"
+ " sudo python setup.py install\n")
+
+ logger = Logger("BaseOMFClient")
+ logger.debug(msg)
+
class BaseOMFClient(object):
pass
import xml.etree.ElementTree as ET
# inherit from BaseXmpp and XMLstream classes
-class OMFClient(sleekxmpp.ClientXMPP, Logger):
+class OMFClient(BaseOMFClient, Logger):
"""
.. class:: Class Args :
self._ready = False
self._registered = False
self._server = None
+ self._parser = None
self.register_plugin('xep_0077') # In-band registration
self.register_plugin('xep_0030')
self.add_event_handler("session_start", self.start)
self.add_event_handler("register", self.register)
self.add_event_handler("pubsub_publish", self.handle_omf_message)
+
+ #Init the parser
+ self._init_parser()
+ def _init_parser(self):
+ """ Init the parser depending on the OMF Version
+
+ """
+ self._parser = OMF6Parser()
+
@property
def ready(self):
""" Check if the client is ready
try:
self['xep_0060'].create_node(self._server, node, config = config)
except:
- error = traceback.format_exc()
- msg = ' Could not create topic: %s\ntraceback:\n%s' % (node, error)
+ #error = traceback.format_exc()
+ #msg = ' Could not create topic: %s\ntraceback:\n%s' % (node, error)
+ msg = 'Could not create the topic : '+node+' . Maybe the topic already exists'
self.error(msg)
def delete(self, node):
msg = ' Deleted node: %s' % node
self.info(msg)
except:
- error = traceback.format_exc()
- msg = ' Could not delete topic: %s\ntraceback:\n%s' % (node, error)
+ #error = traceback.format_exc()
+ #msg = ' Could not delete topic: %s\ntraceback:\n%s' % (node, error)
+ msg = 'Could not delete the topic : '+node+' . Maybe It is not the owner of the topic'
self.error(msg)
def publish(self, data, node):
% (self.boundjid.bare, node, error)
self.error(msg)
- def _check_for_tag(self, root, namespaces, tag):
- """ Check if an element markup is in the ElementTree
+ def check_mailbox(self, itype, attr):
+ """ Check the mail box
- :param root: Root of the tree
- :type root: ElementTree Element
- :param namespaces: Namespaces of the element
- :type namespaces: str
- :param tag: Tag that will search in the tree
- :type tag: str
+ :param itype: type of mail
+ :type itype: str
+ :param attr: value wanted
+ :type attr: str
"""
- for element in root.iter(namespaces+tag):
- if element.text:
- return element
- else :
- return None
-
- def _check_output(self, root, namespaces):
- """ Check the significative element in the answer and display it
-
- :param root: Root of the tree
- :type root: ElementTree Element
- :param namespaces: Namespaces of the tree
- :type namespaces: str
+ return self._parser.check_mailbox(itype, attr)
- """
- fields = ["TARGET", "REASON", "PATH", "APPID", "VALUE"]
- response = ""
- for elt in fields:
- msg = self._check_for_tag(root, namespaces, elt)
- if msg is not None:
- response = response + " " + msg.text + " :"
- deb = self._check_for_tag(root, namespaces, "MESSAGE")
- if deb is not None:
- msg = response + " " + deb.text
- self.debug(msg)
- else :
- self.info(response)
def handle_omf_message(self, iq):
""" Handle published/received message
:type iq: Iq Stanza
"""
- namespaces = "{http://jabber.org/protocol/pubsub}"
- for i in iq['pubsub_event']['items']:
- root = ET.fromstring(str(i))
- self._check_output(root, namespaces)
-
+ self._parser.handle(iq)
git://git.onelab.eu / nepi.git / commitdiff