Changeset cc8d8e9 for fedd/federation/access.py

Timestamp:

Aug 28, 2009 6:07:42 PM (15 years ago)

Author:

Ted Faber <faber@…>

Branches:

axis_example, compt_changes, info-ops, master, version-2.00, version-3.01, version-3.02

Children:

6c57fe9

Parents:

4c8a0b7

Message:

checkpoint

File:

• fedd/federation/access.py (modified) (3 diffs)

fedd/federation/access.py

@@ -105 +105 @@
             'RequestAccess': soap_handler("RequestAccess", self.RequestAccess),
             'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
+            'StartSegment': soap_handler("StartSegment", self.StartSegment),
             }
         self.xmlrpc_services =  {\
@@ -111 +112 @@
             'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
                 self.ReleaseAccess),
+            'StartSegment': xmlrpc_handler('StartSegment',
+                self.StartSegment),
             }
 
@@ -772 +775 @@
 
 
+
+    class emulab_segment:
+        class ssh_cmd_timeout(RuntimeError): pass
+
+        def __init__(self, log=None, keyfile=None, debug=False):
+            self.log = log or logging.getLogger(\
+                    'fedd.experiment_control.emulab_segment')
+            self.ssh_privkey_file = keyfile
+            self.debug = debug
+            self.ssh_exec="/usr/bin/ssh"
+            self.scp_exec = "/usr/bin/scp"
+            self.ssh_cmd_timeout = emulab_segment.ssh_cmd_timeout
+
+        def scp_file(self, file, user, host, dest=""):
+            """
+            scp a file to the remote host.  If debug is set the action is only
+            logged.
+            """
+
+            scp_cmd = [self.scp_exec, '-o', 'IdentitiesOnly yes', 
+                    '-o', 'StrictHostKeyChecking yes', '-i', 
+                    self.ssh_privkey_file, file, 
+                    "%s@%s:%s" % (user, host, dest)]
+            rv = 0
+
+            try:
+                dnull = open("/dev/null", "w")
+            except IOError:
+                self.log.debug("[ssh_file]: failed to open " + \
+                        "/dev/null for redirect")
+                dnull = Null
+
+            self.log.debug("[scp_file]: %s" % " ".join(scp_cmd))
+            if not self.debug:
+                rv = call(scp_cmd, stdout=dnull, stderr=dnull, close_fds=True,
+                        close_fds=True)
+
+            return rv == 0
+
+        def ssh_cmd(self, user, host, cmd, wname=None, timeout=None):
+            """
+            Run a remote command on host as user.  If debug is set, the action
+            is only logged.  Commands are run without stdin, to avoid stray
+            SIGTTINs.
+            """
+            sh_str = ("%s -n -o 'IdentitiesOnly yes' -o " + \
+                    "'StrictHostKeyChecking yes' -i %s %s@%s %s") % \
+                    (self.ssh_exec, self.ssh_privkey_file, 
+                            user, host, cmd)
+
+            try:
+                dnull = open("/dev/null", "w")
+            except IOError:
+                self.log.debug("[ssh_cmd]: failed to open /dev/null " + \
+                        "for redirect")
+                dnull = Null
+
+            self.log.debug("[ssh_cmd]: %s" % sh_str)
+            if not self.debug:
+                if dnull:
+                    sub = Popen(sh_str, shell=True, stdout=dnull, stderr=dnull,
+                            close_fds=True)
+                else:
+                    sub = Popen(sh_str, shell=True,
+                            close_fds=True)
+                if timeout:
+                    i = 0
+                    rv = sub.poll()
+                    while i < timeout:
+                        if rv is not None: break
+                        else:
+                            time.sleep(1)
+                            rv = sub.poll()
+                            i += 1
+                    else:
+                        self.log.debug("Process exceeded runtime: %s" % sh_str)
+                        os.kill(sub.pid, signal.SIGKILL)
+                        raise self.ssh_cmd_timeout();
+                    return rv == 0
+                else:
+                    return sub.wait() == 0
+            else:
+                if timeout == 0:
+                    self.log.debug("debug timeout raised on %s " % sh_str)
+                    raise self.ssh_cmd_timeout()
+                else:
+                    return True
+
+    class start_segment(emulab_segment):
+        def __init__(self, log=None, keyfile=None, debug=False):
+            experiment_control_local.emulab_segment.__init__(self,
+                    log=log, keyfile=keyfile, debug=debug)
+
+        def create_config_tree(self, src_dir, dest_dir, script):
+            """
+            Append commands to script that will create the directory hierarchy
+            on the remote federant.
+            """
+
+            if os.path.isdir(src_dir):
+                print >>script, "mkdir -p %s" % dest_dir
+                print >>script, "chmod 770 %s" % dest_dir
+
+                for f in os.listdir(src_dir):
+                    if os.path.isdir(f):
+                        self.create_config_tree("%s/%s" % (src_dir, f), 
+                                "%s/%s" % (dest_dir, f), script)
+            else:
+                self.log.debug("[create_config_tree]: Not a directory: %s" \
+                        % src_dir)
+
+        def ship_configs(self, host, user, src_dir, dest_dir):
+            """
+            Copy federant-specific configuration files to the federant.
+            """
+            for f in os.listdir(src_dir):
+                if os.path.isdir(f):
+                    if not self.ship_configs(host, user, "%s/%s" % (src_dir, f), 
+                            "%s/%s" % (dest_dir, f)):
+                        return False
+                else:
+                    if not self.scp_file("%s/%s" % (src_dir, f), 
+                            user, host, dest_dir):
+                        return False
+            return True
+
+        def get_state(self, user, host, tb, pid, eid):
+            # command to test experiment state
+            expinfo_exec = "/usr/testbed/bin/expinfo"  
+            # Regular expressions to parse the expinfo response
+            state_re = re.compile("State:\s+(\w+)")
+            no_exp_re = re.compile("^No\s+such\s+experiment")
+            swapping_re = re.compile("^No\s+information\s+available.")
+            state = None    # Experiment state parsed from expinfo
+            # The expinfo ssh command.  Note the identity restriction to use
+            # only the identity provided in the pubkey given.
+            cmd = [self.ssh_exec, '-o', 'IdentitiesOnly yes', '-o', 
+                    'StrictHostKeyChecking yes', '-i', 
+                    self.ssh_privkey_file, "%s@%s" % (user, host), 
+                    expinfo_exec, pid, eid]
+
+            dev_null = None
+            try:
+                dev_null = open("/dev/null", "a")
+            except IOError, e:
+                self.log.error("[get_state]: can't open /dev/null: %s" %e)
+
+            if self.debug:
+                state = 'swapped'
+                rv = 0
+            else:
+                status = Popen(cmd, stdout=PIPE, stderr=dev_null,
+                        close_fds=True)
+                for line in status.stdout:
+                    m = state_re.match(line)
+                    if m: state = m.group(1)
+                    else:
+                        for reg, st in ((no_exp_re, "none"),
+                                (swapping_re, "swapping")):
+                            m = reg.match(line)
+                            if m: state = st
+                rv = status.wait()
+
+            # If the experiment is not present the subcommand returns a
+            # non-zero return value.  If we successfully parsed a "none"
+            # outcome, ignore the return code.
+            if rv != 0 and state != 'none':
+                raise service_error(service_error.internal,
+                        "Cannot get status of segment %s:%s/%s" % \
+                                (tb, pid, eid))
+            elif state not in ('active', 'swapped', 'swapping', 'none'):
+                raise service_error(service_error.internal,
+                        "Cannot get status of segment %s:%s/%s" % \
+                                (tb, pid, eid))
+            else: return state
+
+
+        def __call__(self, tb, eid, tbparams, tmpdir, timeout=0):
+            """
+            Start a sub-experiment on a federant.
+
+            Get the current state, modify or create as appropriate, ship data
+            and configs and start the experiment.  There are small ordering
+            differences based on the initial state of the sub-experiment.
+            """
+            # ops node in the federant
+            host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
+            user = tbparams[tb]['user']     # federant user
+            pid = tbparams[tb]['project']   # federant project
+            # XXX
+            base_confs = ( "hosts",)
+            tclfile = "%s.%s.tcl" % (eid, tb)   # sub-experiment description
+            # Configuration directories on the remote machine
+            proj_dir = "/proj/%s/exp/%s/tmp" % (pid, eid)
+            tarfiles_dir = "/proj/%s/tarfiles/%s" % (pid, eid)
+            rpms_dir = "/proj/%s/rpms/%s" % (pid, eid)
+
+            state = self.get_state(user, host, tb, pid, eid)
+
+            self.log.debug("[start_segment]: %s: %s" % (tb, state))
+            self.log.info("[start_segment]:transferring experiment to %s" % tb)
+
+            if not self.scp_file("%s/%s/%s" % \
+                    (tmpdir, tb, tclfile), user, host):
+                return False
+            
+            if state == 'none':
+                # Create a null copy of the experiment so that we capture any
+                # logs there if the modify fails.  Emulab software discards the
+                # logs from a failed startexp
+                if not self.scp_file("%s/null.tcl" % tmpdir, user, host):
+                    return False
+                self.log.info("[start_segment]: Creating %s on %s" % (eid, tb))
+                timedout = False
+                try:
+                    if not self.ssh_cmd(user, host,
+                            ("/usr/testbed/bin/startexp -i -f -w -p %s " + 
+                            "-e %s null.tcl") % (pid, eid), "startexp",
+                            timeout=60 * 10):
+                        return False
+                except self.ssh_cmd_timeout:
+                    timedout = True
+
+                if timedout:
+                    state = self.get_state(user, host, tb, pid, eid)
+                    if state != "swapped":
+                        return False
+
+            
+            # Open up a temporary file to contain a script for setting up the
+            # filespace for the new experiment.
+            self.log.info("[start_segment]: creating script file")
+            try:
+                sf, scriptname = tempfile.mkstemp()
+                scriptfile = os.fdopen(sf, 'w')
+            except IOError:
+                return False
+
+            scriptbase = os.path.basename(scriptname)
+
+            # Script the filesystem changes
+            print >>scriptfile, "/bin/rm -rf %s" % proj_dir
+            # Clear and create the tarfiles and rpm directories
+            for d in (tarfiles_dir, rpms_dir):
+                print >>scriptfile, "/bin/rm -rf %s/*" % d
+                print >>scriptfile, "mkdir -p %s" % d
+            print >>scriptfile, 'mkdir -p %s' % proj_dir
+            self.create_config_tree("%s/%s" % (tmpdir, tb),
+                    proj_dir, scriptfile)
+            if os.path.isdir("%s/tarfiles" % tmpdir):
+                self.create_config_tree("%s/tarfiles" % tmpdir, tarfiles_dir,
+                        scriptfile)
+            if os.path.isdir("%s/rpms" % tmpdir):
+                self.create_config_tree("%s/rpms" % tmpdir, rpms_dir, 
+                        scriptfile)
+            print >>scriptfile, "rm -f %s" % scriptbase
+            scriptfile.close()
+
+            # Move the script to the remote machine
+            # XXX: could collide tempfile names on the remote host
+            if self.scp_file(scriptname, user, host, scriptbase):
+                os.remove(scriptname)
+            else:
+                return False
+
+            # Execute the script (and the script's last line deletes it)
+            if not self.ssh_cmd(user, host, "sh -x %s" % scriptbase):
+                return False
+
+            for f in base_confs:
+                if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
+                        "%s/%s" % (proj_dir, f)):
+                    return False
+            if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
+                    proj_dir):
+                return False
+            if os.path.isdir("%s/tarfiles" % tmpdir):
+                if not self.ship_configs(host, user,
+                        "%s/tarfiles" % tmpdir, tarfiles_dir):
+                    return False
+            if os.path.isdir("%s/rpms" % tmpdir):
+                if not self.ship_configs(host, user,
+                        "%s/rpms" % tmpdir, tarfiles_dir):
+                    return False
+            # Stage the new configuration (active experiments will stay swapped
+            # in now)
+            self.log.info("[start_segment]: Modifying %s on %s" % (eid, tb))
+            try:
+                if not self.ssh_cmd(user, host,
+                        "/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
+                                (pid, eid, tclfile),
+                        "modexp", timeout= 60 * 10):
+                    return False
+            except self.ssh_cmd_timeout:
+                self.log.error("Modify command failed to complete in time")
+                # There's really no way to see if this succeeded or failed, so
+                # if it hangs, assume the worst.
+                return False
+            # Active experiments are still swapped, this swaps the others in.
+            if state != 'active':
+                self.log.info("[start_segment]: Swapping %s in on %s" % \
+                        (eid, tb))
+                timedout = False
+                try:
+                    if not self.ssh_cmd(user, host,
+                            "/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
+                            "swapexp", timeout=10*60):
+                        return False
+                except self.ssh_cmd_timeout:
+                    timedout = True
+               
+                # If the command was terminated, but completed successfully,
+                # report success.
+                if timedout:
+                    self.log.debug("[start_segment]: swapin timed out " +\
+                            "checking state")
+                    state = self.get_state(user, host, tb, pid, eid)
+                    self.log.debug("[start_segment]: state is %s" % state)
+                    return state == 'active'
+            # Everything has gone OK.
+            return True
+
+    class stop_segment(emulab_segment):
+        def __init__(self, log=None, keyfile=None, debug=False):
+            experiment_control_local.emulab_segment.__init__(self,
+                    log=log, keyfile=keyfile, debug=debug)
+
+        def __call__(self, tb, eid, tbparams):
+            """
+            Stop a sub experiment by calling swapexp on the federant
+            """
+            user = tbparams[tb]['user']
+            host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
+            pid = tbparams[tb]['project']
+
+            self.log.info("[stop_segment]: Stopping %s on %s" % (eid, tb))
+            rv = False
+            try:
+                # Clean out tar files: we've gone over quota in the past
+                self.ssh_cmd(user, host, "rm -rf /proj/%s/rpms/%s" % (pid, eid))
+                self.ssh_cmd(user, host, "rm -rf /proj/%s/tarfiles/%s" % \
+                        (pid, eid))
+                rv = self.ssh_cmd(user, host,
+                        "/usr/testbed/bin/swapexp -w %s %s out" % (pid, eid))
+            except self.ssh_cmd_timeout:
+                rv = False
+            return rv
+
+    def StartSegment(self, req, fid):
+        try:
+            req = req['StartSegmentRequestBody']
+        except KeyError:
+            raise service_error(server_error.req, "Badly formed request")
+        auth_attr = req['allocID']['fedid']
+        if self.auth.check_attribute(fid, auth_attr):
+            print "OK"
+        else:
+            print "Fail"
+        return { 'allocID': req['allocID'] }