Changeset 79b6596 for fedd

Timestamp:

Jul 22, 2009 1:29:23 AM (15 years ago)

Author:

Ted Faber <faber@…>

Branches:

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

Children:

c2dbca8

Parents:

012dba5

Message:

Initial swing at timeouts.

Also, the filesystem config script needs to take the ship_configs commands into
account.

File:

• fedd/federation/experiment_control.py (modified) (13 diffs)

fedd/federation/experiment_control.py

@@ -11 +11 @@
 import pickle
 import logging
+import signal
+import time
 
 import traceback
@@ -38 +40 @@
     Thred safe.
     """
+
+    class ssh_cmd_timeout(RuntimeError): pass
     
     class thread_pool:
@@ -487 +491 @@
         return rv == 0
 
-    def ssh_cmd(self, user, host, cmd, wname=None):
+    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
@@ -509 +513 @@
             else:
                 sub = Popen(sh_str, shell=True)
-            return sub.wait() == 0
+            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:
             return True
@@ -533 +552 @@
         return True
 
-    def start_segment(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
+    def get_state(self, user, host, ssh_key, tb, pid, eid):
         # command to test experiment state
         expinfo_exec = "/usr/testbed/bin/expinfo"  
-        # 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)
         # 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
@@ -562 +564 @@
         cmd = [self.ssh_exec, '-o', 'IdentitiesOnly yes', '-o', 
                 'StrictHostKeyChecking yes', '-i', 
-                self.ssh_privkey_file, "%s@%s" % (user, host), 
+                ssh_key, "%s@%s" % (user, host), 
                 expinfo_exec, pid, eid]
 
-        # Get status
-        self.log.debug("[start_segment]: %s"% " ".join(cmd))
         dev_null = None
         try:
@@ -582 +582 @@
                 if m: state = m.group(1)
                 else:
-                    m = no_exp_re.match(line)
-                    if m: state = "none"
+                    for reg, st in ((no_exp_re, "none"),
+                            (swapping_re, "swapping")):
+                        m = reg.match(line)
+                        if m: state = st
             rv = status.wait()
 
@@ -592 +594 @@
             raise service_error(service_error.internal,
                     "Cannot get status of segment %s:%s/%s" % (tb, pid, eid))
-
-        if state not in ('active', 'swapped', 'none'):
-            self.log.debug("[start_segment]:unknown state %s" % state)
-            return False
+        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 start_segment(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, self.ssh_privkey_file, tb, pid, eid)
 
         self.log.debug("[start_segment]: %s: %s" % (tb, state))
@@ -610 +635 @@
                 return False
             self.log.info("[start_segment]: Creating %s on %s" % (eid, tb))
-            if not self.ssh_cmd(user, host,
-                    "/usr/testbed/bin/startexp -i -f -w -p %s -e %s null.tcl" \
-                            % (pid, eid), "startexp"):
-                return False
+            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, self.ssh_privkey_file, 
+                        tb, eid, pid)
+                if state != "swapped":
+                    return False
+
         
         # Open up a temporary file to contain a script for setting up the
@@ -624 +661 @@
             return False
 
+        scriptbase = os.path.basename(scriptname)
+
         # Script the filesystem changes
         print >>scriptfile, "/bin/rm -rf %s" % proj_dir
@@ -631 +670 @@
             print >>scriptfile, "mkdir -p %s" % d
         print >>scriptfile, 'mkdir -p %s' % proj_dir
-        print >>scriptfile, "rm %s" % scriptname
+        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, scriptname):
+        if self.scp_file(scriptname, user, host, scriptbase):
             os.remove(scriptname)
         else:
@@ -642 +681 @@
 
         # Execute the script (and the script's last line deletes it)
-        if not self.ssh_cmd(user, host, "sh -x %s" % scriptname):
+        if not self.ssh_cmd(user, host, "sh -x %s" % scriptbase):
             return False
 
@@ -663 +702 @@
         # now)
         self.log.info("[start_segment]: Modifying %s on %s" % (eid, tb))
-        if not self.ssh_cmd(user, host,
-                "/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
-                        (pid, eid, tclfile),
-                "modexp"):
-            return False
+        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:
+            print "modexp timeout"
+            # 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))
-            if not self.ssh_cmd(user, host,
-                    "/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
-                    "swapexp"):
-                return False
+            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:
+                state = self.get_state(user, host, self.ssh_privkey_file,
+                        tb, eid, pid)
+                self.log.debug("[start_segment]: swapin timed out (state)")
+                return state == 'active'
+        # Everything has gone OK.
         return True