X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=src%2Fnepi%2Fresources%2Fomf%2Fapplication.py;h=03db08892ae220a49b3666d4fa771cd436b053ee;hb=6285ca51026efb69642eea9dfc7c480e722d84a9;hp=1ec831fd3a28056c65189cf0f56fd6fbcd618ef1;hpb=dac2434143d573f76c14eac647645d478cfb4f80;p=nepi.git diff --git a/src/nepi/resources/omf/application.py b/src/nepi/resources/omf/application.py index 1ec831fd..03db0889 100644 --- a/src/nepi/resources/omf/application.py +++ b/src/nepi/resources/omf/application.py @@ -3,9 +3,8 @@ # Copyright (C) 2013 INRIA # # This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation; # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of @@ -18,13 +17,17 @@ # Author: Alina Quereilhac # Julien Tribino +from __future__ import print_function + import os +from nepi.util.timefuncs import tnow from nepi.execution.resource import ResourceManager, clsinit_copy, \ - ResourceState, reschedule_delay + ResourceState +from nepi.execution.trace import Trace, TraceAttr 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.node import OMFNode, confirmation_counter, reschedule_check from nepi.resources.omf.omf_api_factory import OMFAPIFactory from nepi.util import sshfuncs @@ -40,8 +43,8 @@ class OMFApplication(OMFResource): :type guid: int """ - _rtype = "OMFApplication" - _authorized_connections = ["OMFNode"] + _rtype = "omf::Application" + _authorized_connections = ["omf::Node", "wilabt::sfa::Node"] @classmethod def _register_attributes(cls): @@ -55,11 +58,11 @@ class OMFApplication(OMFResource): 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(command) @@ -83,8 +86,8 @@ class OMFApplication(OMFResource): self.set('command', "") self.set('appid', "") - self.path= "" - self.args = "" + self._path= "" + self._args = "" self.set('env', "") self._node = None @@ -92,15 +95,26 @@ class OMFApplication(OMFResource): 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 + + # For performance tests + self.begin_deploy_time = None + self.begin_start_time = None + self.begin_release_time = None + self.dperf = True + self.sperf = True + self.rperf = True self.add_set_hook() def _init_command(self): comm = self.get('command').split(' ') - self.path= comm[0] + self._path= comm[0] if len(comm)>1: - self.args = ' '.join(comm[1:]) + self._args = ' '.join(comm[1:]) @property def exp_id(self): @@ -114,14 +128,14 @@ class OMFApplication(OMFResource): 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"] @@ -163,15 +177,20 @@ class OMFApplication(OMFResource): 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) + self.ec.schedule(self.reschedule_delay, self.deploy) return + ## For performance test + if self.dperf: + self.begin_deploy_time = tnow() + self.dperf = False + self._init_command() self.set('xmppUser',self.node.get('xmppUser')) @@ -183,19 +202,27 @@ class OMFApplication(OMFResource): if not self.get('xmppServer'): msg = "XmppServer is not initialzed. XMPP Connections impossible" self.error(msg) - raise RuntimeError, msg + raise RuntimeError(msg) 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 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": + + self.begin_deploy_time = tnow() + if self.get('sources'): gateway = ResourceGateway.AMtoGateway[self.get('xmppServer')] user = self.get('sshUser') or self.get('xmppUser') @@ -213,71 +240,140 @@ class OMFApplication(OMFResource): 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.ec.schedule(reschedule_delay, self.deploy) + 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 trace(self, name, attr = TraceAttr.ALL, block = 512, offset = 0): + self.info("Retrieving '%s' trace %s " % (name, attr)) + if name == 'stdout' : + suffix = '.out' + elif name == 'stderr' : + suffix = '.err' + else : + suffix = '.misc' + + trace_path = '/tmp/'+ self._topic_app + suffix + + if attr == TraceAttr.PATH: + return trace_path + + if attr == TraceAttr.ALL: + try: + with open(trace_path ,'r') as f: + return f.read() + except IOError: + print("File with traces has not been found") + 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('command') : - msg = "Application's Command is not initialized" - self.error(msg) - raise RuntimeError, msg + ## For performance test + if self.sperf: + self.begin_start_time = tnow() + self.sperf = False if not self.get('env'): self.set('env', " ") if self.get('version') == "5": + self.begin_start_time = tnow() # 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.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._args, self._path, self.get('env')) else: #For OMF 6 - props = {} - props['state'] = "running" + if self._start_cnt == 0: + props = {} + props['state'] = "running" - guards = {} - guards['type'] = "application" - guards['name'] = self.get('command') + guards = {} + guards['type'] = "application" + guards['name'] = self.get('command') - self._omf_api.frcp_configure(self._topic_app, props = props, guards = guards ) + 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. """ + + 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 @@ -287,17 +383,34 @@ class OMFApplication(OMFResource): """ Clean the RM at the end of the experiment and release the API. """ - 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) - - cid = self.check_release(self.release_id) - if not cid: - self.ec.schedule(reschedule_delay, self.release) - return + ## For performance test + if self.rperf: + self.begin_release_time = tnow() + self.rperf = False if self._omf_api: + if self.get('version') == "6" and self._topic_app: + 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) + + # Remove the stdout and stderr of the application + try: + os.remove('/tmp/'+self._topic_app +'.out') + os.remove('/tmp/'+self._topic_app +'.err') + except OSError: + pass + OMFAPIFactory.release_api(self.get('version'), self.get('xmppServer'), self.get('xmppUser'), self.get('xmppPort'), self.get('xmppPassword'), exp_id = self.exp_id)