From: Julien Tribino Date: Thu, 5 Jun 2014 08:16:26 +0000 (+0200) Subject: Merge the OMF 6 branch X-Git-Tag: nepi-3.1.0~30 X-Git-Url: http://git.onelab.eu/?a=commitdiff_plain;h=2e80f0fafa0c2ef6a5f536efd4c868c91468f962;hp=-c;p=nepi.git Merge the OMF 6 branch --- 2e80f0fafa0c2ef6a5f536efd4c868c91468f962 diff --combined src/nepi/resources/omf/application.py index 8ef2b509,a7f131b8..b6d8d13a --- a/src/nepi/resources/omf/application.py +++ b/src/nepi/resources/omf/application.py @@@ -18,12 -18,14 +18,14 @@@ # Author: Alina Quereilhac # Julien Tribino + 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 @@@ -46,20 -48,21 +48,21 @@@ class OMFApplication(OMFResource) """ 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) @@@ -78,17 -81,30 +81,30 @@@ """ 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 @@@ -101,14 -117,14 +117,14 @@@ 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"] @@@ -150,79 -166,194 +166,194 @@@ 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() diff --combined src/nepi/resources/omf/omf_client.py index 05af4f1e,113bd1e3..42f1d516 --- a/src/nepi/resources/omf/omf_client.py +++ b/src/nepi/resources/omf/omf_client.py @@@ -19,23 -19,20 +19,23 @@@ # Julien Tribino 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 @@@ -43,7 -40,7 +43,7 @@@ import tracebac import xml.etree.ElementTree as ET # inherit from BaseXmpp and XMLstream classes -class OMFClient(sleekxmpp.ClientXMPP, Logger): +class OMFClient(BaseOMFClient, Logger): """ .. class:: Class Args : @@@ -74,6 -71,7 +74,7 @@@ 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') @@@ -83,7 -81,16 +84,16 @@@ 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 @@@ -193,8 -200,9 +203,9 @@@ 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): @@@ -212,8 -220,9 +223,9 @@@ 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): @@@ -324,44 -333,17 +336,17 @@@ % (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 @@@ -370,9 -352,5 +355,5 @@@ :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)