Changeset cc8d8e9

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

Location:

fedd/federation

Files:

• access.py (modified) (3 diffs)
• deter_impl.py (modified) (2 diffs)
• experiment_control.py (modified) (10 diffs)
• topdl.py (modified) (7 diffs)
• util.py (modified) (1 diff)

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'] }

fedd/federation/deter_impl.py

@@ -41 +41 @@
                     raise RuntimeError(
                             "Error reading accessDB %s: %s" % (access_db, e))
-                except ValueError:
+                except ValueError, e:
                     raise RuntimeError("%s" % e)
 
@@ -62 +62 @@
                 self.get_handler = self.experiment.get_handler
             else:
-                self.get_handler = self.experiment.get_handler
+                self.get_handler = None
 
             if config.has_section("splitter"):

fedd/federation/experiment_control.py

@@ -196 +196 @@
     call_RequestAccess = service_caller('RequestAccess')
     call_ReleaseAccess = service_caller('ReleaseAccess')
+    call_StartSegment = service_caller('StartSegment')
     call_Ns2Split = service_caller('Ns2Split')
 
@@ -230 +231 @@
         self.trusted_certs = config.get("experiment_control", "trusted_certs") \
                 or config.get("globals", "trusted_certs")
+
+        # XXX:
+        self.repodir = '/usr/local/etc/fedd/repo'
 
         self.exp_stem = "fed-stem"
@@ -403 +407 @@
         State format is a simple pickling of the state dictionary.
         """
+        
+        def get_experiment_id(state):
+            """
+            Pull the fedid experimentID out of the saved state.  This is kind
+            of a gross walk through the dict.
+            """
+
+            if state.has_key('experimentID'):
+                for e in state['experimentID']:
+                    if e.has_key('fedid'):
+                        return e['fedid']
+                else:
+                    return None
+            else:
+                return None
+
+        def get_alloc_ids(state):
+            """
+            Pull the fedids of the identifiers of each allocation from the
+            state.  Again, a dict dive that's best isolated.
+            """
+
+            return [ f['allocID']['fedid'] 
+                    for f in state.get('federant',[]) \
+                        if f.has_key('allocID') and \
+                            f['allocID'].has_key('fedid')]
+
+
         try:
             f = open(self.state_filename, "r")
@@ -415 +447 @@
                     "Unpickling failed: %s") % e)
         
-        for k in self.state.keys():
+        for s in self.state.values():
             try:
-                # This list should only have one element in it, but phrasing it
-                # as a for loop doesn't cost much, really.  We have to find the
-                # fedid elements anyway.
-                for eid in [ f['fedid'] \
-                        for f in self.state[k]['experimentID']\
-                            if f.has_key('fedid') ]:
-                    self.auth.set_attribute(self.state[k]['owner'], eid)
+
+                eid = get_experiment_id(s)
+                if eid : 
+                    # Give the owner rights to the experiment
+                    self.auth.set_attribute(s['owner'], eid)
+                    # And holders of the eid as well
+                    self.auth.set_attribute(eid, eid)
                     # allow overrides to control experiments as well
                     for o in self.overrides:
                         self.auth.set_attribute(o, eid)
+                    # Set permissions to allow reading of the software repo, if
+                    # any, as well.
+                    for a in get_alloc_ids(s):
+                        self.auth.set_attribute(a, 'repo/%s' % eid)
+                else:
+                    raise KeyError("No experiment id")
             except KeyError, e:
                 self.log.warning("[read_state]: State ownership or identity " +\
@@ -970 +1008 @@
             nodes = [ n['vname'] for n in topo['node'] ]
             topo_lans = topo['lan']
-        except KeyError:
-            raise service_error(service_error.internal, "Bad topology")
+        except KeyError, e:
+            raise service_error(service_error.internal, "Bad topology: %s" %e)
 
         lans = { }
@@ -2103 +2141 @@
 
         return rv
+
+    class new_start_segment:
+        def __init__(self, debug=False, log=None, cert_file=None,
+                cert_pwd=None, trusted_certs=None, caller=None):
+            self.log = log
+            self.debug = debug
+            self.cert_file = cert_file
+            self.cert_pwd = cert_pwd
+            self.trusted_certs = None
+            self.caller = caller
+
+        def __call__(self, uri, aid, topo):
+            req = {
+                    'allocID': { 'fedid' : aid }, 
+                    'segmentdescription': { 
+                        'topdldescription': topo.to_dict(),
+                    },
+                }
+
+            print "calling %s"  % uri
+            r = self.caller(uri, req, self.cert_file, self.cert_pwd,
+                    self.trusted_certs)
+            print r
+            return True
+
+
     
+
+    def new_allocate_resources(self, allocated, master, eid, expid, expcert, 
+            tbparams, topo, tmpdir, alloc_log=None):
+        started = { }           # Testbeds where a sub-experiment started 
+                                # successfully
+
+        # XXX
+        fail_soft = False
+
+        log = alloc_log or self.log
+
+        thread_pool = self.thread_pool(self.nthreads)
+        threads = [ ]
+
+        for tb in [ k for k in allocated.keys() if k != master]:
+            # Create and start a thread to start the segment, and save it to
+            # get the return value later
+            thread_pool.wait_for_slot()
+            uri = self.tbmap.get(tb, None)
+            if not uri:
+                raise service_error(service_error.internal, 
+                        "Unknown testbed %s !?" % tb)
+
+            if tbparams[tb].has_key('allocID') and \
+                    tbparams[tb]['allocID'].has_key('fedid'):
+                aid = tbparams[tb]['allocID']['fedid']
+            else:
+                raise service_error(service_error.internal, 
+                        "No alloc id for testbed %s !?" % tb)
+
+            t  = self.pooled_thread(\
+                    target=self.new_start_segment(log=log, debug=self.debug,
+                        cert_file=self.cert_file, cert_pwd=self.cert_pwd,
+                        trusted_certs=self.trusted_certs,
+                        caller=self.call_StartSegment), 
+                    args=(uri, aid, topo[tb]), name=tb,
+                    pdata=thread_pool, trace_file=self.trace_file)
+            threads.append(t)
+            t.start()
+
+        # Wait until all finish 
+        thread_pool.wait_for_all_done()
+
+        # If none failed, start the master
+        failed = [ t.getName() for t in threads if not t.rv ]
+
+        if len(failed) == 0:
+            uri = self.tbmap.get(master, None)
+            if not uri:
+                raise service_error(service_error.internal, 
+                        "Unknown testbed %s !?" % master)
+
+            if tbparams[master].has_key('allocID') and \
+                    tbparams[master]['allocID'].has_key('fedid'):
+                aid = tbparams[master]['allocID']['fedid']
+            else:
+                raise service_error(service_error.internal, 
+                    "No alloc id for testbed %s !?" % master)
+            starter = self.new_start_segment(log=log, debug=self.debug,
+                    cert_file=self.cert_file, cert_pwd=self.cert_pwd,
+                    trusted_certs=self.trusted_certs,
+                    caller=self.call_StartSegment)
+            if not starter(uri, aid, topo[master]):
+                failed.append(master)
+
+        succeeded = [tb for tb in allocated.keys() if tb not in failed]
+        # If one failed clean up, unless fail_soft is set
+        if failed and False:
+            if not fail_soft:
+                thread_pool.clear()
+                for tb in succeeded:
+                    # Create and start a thread to stop the segment
+                    thread_pool.wait_for_slot()
+                    t  = self.pooled_thread(\
+                            target=self.stop_segment(log=log,
+                                keyfile=self.ssh_privkey_file,
+                                debug=self.debug), 
+                            args=(tb, eid, tbparams), name=tb,
+                            pdata=thread_pool, trace_file=self.trace_file)
+                    t.start()
+                # Wait until all finish
+                thread_pool.wait_for_all_done()
+
+                # release the allocations
+                for tb in tbparams.keys():
+                    self.release_access(tb, tbparams[tb]['allocID'])
+                # Remove the placeholder
+                self.state_lock.acquire()
+                self.state[eid]['experimentStatus'] = 'failed'
+                if self.state_filename: self.write_state()
+                self.state_lock.release()
+
+                log.error("Swap in failed on %s" % ",".join(failed))
+                return
+        else:
+            log.info("[start_segment]: Experiment %s active" % eid)
+
+        log.debug("[start_experiment]: removing %s" % tmpdir)
+
+        # Walk up tmpdir, deleting as we go
+        for path, dirs, files in os.walk(tmpdir, topdown=False):
+            for f in files:
+                os.remove(os.path.join(path, f))
+            for d in dirs:
+                os.rmdir(os.path.join(path, d))
+        os.rmdir(tmpdir)
+
+        # Insert the experiment into our state and update the disk copy
+        self.state_lock.acquire()
+        self.state[expid]['experimentStatus'] = 'active'
+        self.state[eid] = self.state[expid]
+        if self.state_filename: self.write_state()
+        self.state_lock.release()
+        return
+
 
     def new_create_experiment(self, req, fid):
@@ -2274 +2453 @@
 
             allocated = { }         # Testbeds we can access
-# XXX here's where we're working
-            def out_topo(filename, t):
-                try:
-                    f = open("/tmp/%s" % filename, "w")
-                    print >> f, "%s" % \
-                            topdl.topology_to_xml(t, top="experiment")
-                    f.close()
-                except IOError, e:
-                    raise service_error(service_error.internal, "Can't open file")
-
-            try:
-
-                top = topdl.topology_from_xml(file=split_data, top="experiment")
-                subs = sorted(top.substrates, 
-                        cmp=lambda x,y: cmp(len(x.interfaces), len(y.interfaces)),
-                        reverse=True)
-                ips = ip_allocator(int(ip_addr("10.0.0.0")), 2 **24)
-                for s in subs:
-                    a = ips.allocate(len(s.interfaces)+2)
-                    if a :
-                        base, num = a
-                        if num < len(s.interfaces) +2 : 
-                            raise service_error(service_error.internal,
-                                    "Allocator returned wrong number of IPs??")
-                    else:
-                        raise service_error(service_error.req, 
-                                "Cannot allocate IP addresses")
-
-                    base += 1
-                    for i in s.interfaces:
-                        i.attribute.append(
-                                topdl.Attribute('ip4_address', 
-                                    "%s" % ip_addr(base)))
-                        base += 1
-
-                testbeds = set([ a.value for e in top.elements \
-                        for a in e.attribute \
-                            if a.attribute == 'testbed'] )
-                topo ={ }
-                for tb in testbeds:
-                    self.get_access(tb, None, user, tbparams, master,
-                            export_project, access_user)
-                    topo[tb] = top.clone()
-                    to_delete = [ ]
-                    for e in topo[tb].elements:
-                        etb = e.get_attribute('testbed')
-                        if etb and etb != tb:
-                            for i in e.interface:
-                                for s in i.subs:
-                                    try:
-                                        s.interfaces.remove(i)
-                                    except ValueError:
-                                        raise service_error(service_error.internal,
-                                                "Can't remove interface??")
-                            to_delete.append(e)
-                    for e in to_delete:
-                        topo[tb].elements.remove(e)
-                    topo[tb].make_indices()
-
-
-
-                for s in top.substrates:
-                    tests = { }
-                    for i in s.interfaces:
-                        e = i.element
-                        tb = e.get_attribute('testbed')
-                        if tb and not tests.has_key(tb):
-                            for i in e.interface:
-                                if s in i.subs:
-                                    tests[tb]= \
-                                            i.get_attribute('ip4_address')
-                    if len(tests) < 2:
-                        continue
-
-                    # More than one testbed is on this substrate.  Insert
-                    # some gateways into the subtopologies.
-
-                    for st in tests.keys():
-                        for dt in [ t for t in tests.keys() if t != st]:
-                            myname =  "%stunnel" % dt
-                            desthost  =  "%stunnel" % st
-                            sproject = tbparams[st].get('project', 'project')
-                            dproject = tbparams[dt].get('project', 'project')
-                            sdomain = ".%s.%s%s" % (eid, sproject,
-                                    tbparams[st].get('domain', ".example.com"))
-                            ddomain = ".%s.%s%s" % (eid, dproject,
-                                    tbparams[dt].get('domain', ".example.com"))
-                            boss = tbparams[master].get('boss', "boss")
-                            fs = tbparams[master].get('fs', "fs")
-                            event_server = "%s%s" % \
-                                    (tbparams[st].get('eventserver', "event_server"),
-                                            tbparams[dt].get('domain', "example.com"))
-                            remote_event_server = "%s%s" % \
-                                    (tbparams[dt].get('eventserver', "event_server"),
-                                            tbparams[dt].get('domain', "example.com"))
-                            seer_control = "%s%s" % \
-                                    (tbparams[st].get('control', "control"), sdomain)
-                            local_key_dir = "/proj/%s/exp/%s/tmp" % ( sproject, eid)
-                            remote_conf_dir = "/proj/%s/exp/%s/tmp" % ( dproject, eid)
-                            conf_file = "%s%s.gw.conf" % (myname, sdomain)
-                            remote_conf_file = "%s%s.gw.conf" % (desthost, ddomain)
-                            # translate to lower case so the `hostname` hack for specifying
-                            # configuration files works.
-                            conf_file = conf_file.lower();
-                            remote_conf_file = remote_conf_file.lower();
-                            active = ("%s" % (st == master))
-                            portal = topdl.Computer(**{
-                                    'name': "%stunnel" % dt,
-                                    'attribute' : [{
-                                        'attribute': n,
-                                        'value': v,
-                                        } for n, v in (\
-                                                ('gateway', 'true'),
-                                                ('boss', boss),
-                                                ('fs', fs),
-                                                ('event_server', event_server),
-                                                ('remote_event_server', remote_event_server),
-                                                ('seer_control', seer_control),
-                                                ('local_key_dir', local_key_dir),
-                                                ('remote_conf_dir', remote_conf_dir),
-                                                ('conf_file', conf_file),
-                                                ('remote_conf_file', remote_conf_file),
-                                                ('remote_script_dir', "/usr/local/federation/bin"),
-                                                ('local_script_dir', "/usr/local/federation/bin"),
-                                                )],
-                                    'interface': [{
-                                        'substrate': s.name,
-                                        'attribute': [ { 
-                                            'attribute': 'ip4_addreess', 
-                                            'value': tests[dt],
-                                            }, ],
-                                        }, ],
-                                    })
-                            topo[st].elements.append(portal)
-                # Connect the gateway nodes into the topologies and clear out
-                # substrates that are not in the topologies
-                for tb in testbeds:
-                    topo[tb].incorporate_elements()
-                    topo[tb].substrates = \
-                            [s for s in topo[tb].substrates \
-                                if len(s.interfaces) >0]
-
-                softdir ="%s/software" % tmpdir
-                softmap = { }
-                os.mkdir(softdir)
-                pkgs = set([fedkit, gatewaykit])
-                pkgs.update([x.location for e in top.elements \
-                        for x in e.software])
-                for pkg in pkgs:
-                    loc = pkg
-
-                    scheme, host, path = urlparse(loc)[0:3]
-                    dest = os.path.basename(path)
-                    if not scheme:
-                        if not loc.startswith('/'):
-                            loc = "/%s" % loc
-                        loc = "file://%s" %loc
-                    try:
-                        u = urlopen(loc)
-                    except Exception, e:
-                        raise service_error(service_error.req, 
-                                "Cannot open %s: %s" % (loc, e))
-                    try:
-                        f = open("%s/%s" % (softdir, dest) , "w")
-                        data = u.read(4096)
-                        while data:
-                            f.write(data)
-                            data = u.read(4096)
-                        f.close()
-                        u.close()
-                    except Exception, e:
-                        raise service_error(service_error.internal,
-                                "Could not copy %s: %s" % (loc, e))
-                    path = re.sub("/tmp", "", softdir)
-                    # XXX
-                    softmap[pkg] = \
-                            "https://users.isi.deterlab.net:23232/%s/%s" %\
-                            ( path, dest)
-
-                # Convert the software locations in the segments into the local
-                # copies on this host
-                for soft in [ s for tb in topo.values() \
-                        for e in tb.elements \
-                            for s in e.software ]:
-                    if softmap.has_key(soft.location):
-                        soft.location = softmap[soft.location]
-                for tb in testbeds:
-                    out_topo("%s.xml" %tb, topo[tb])
-
-                vtopo = topdl.topology_to_vtopo(top)
-                vis = self.genviz(vtopo)
-
-            except Exception, e:
-                traceback.print_exc()
-                raise service_error(service_error.internal, "%s"  % e)
-
-
-
-            # Build the testbed topologies:
-
-
-            if True:
-                raise service_error(service_error.internal, "Developing")
-
-# XXX old code
-            # Objects to parse the splitter output (defined above)
-            parse_current_testbed = self.current_testbed(eid, tmpdir,
-                    self.fedkit, self.gatewaykit)
-            parse_allbeds = self.allbeds(self.get_access)
-            parse_gateways = self.gateways(eid, master, tmpdir,
-                    gw_pubkey_base, gw_secretkey_base, self.copy_file,
-                    self.fedkit)
-            parse_vtopo = self.shunt_to_string("^#\s+Begin\s+Vtopo",
-                        "^#\s+End\s+Vtopo")
-            parse_hostnames = self.shunt_to_file("^#\s+Begin\s+hostnames",
-                        "^#\s+End\s+hostnames", tmpdir + "/hosts")
-            parse_tarfiles = self.shunt_to_list("^#\s+Begin\s+tarfiles",
-                    "^#\s+End\s+tarfiles")
-            parse_rpms = self.shunt_to_list("^#\s+Begin\s+rpms",
-                    "^#\s+End\s+rpms")
-
-            # Working on the split data
-            for line in split_data:
-                line = line.rstrip()
-                if parse_current_testbed(line, master, allocated, tbparams):
-                    continue
-                elif parse_allbeds(line, user, tbparams, master, export_project,
-                        access_user):
-                    continue
-                elif parse_gateways(line, allocated, tbparams):
-                    continue
-                elif parse_vtopo(line):
-                    continue
-                elif parse_hostnames(line):
-                    continue
-                elif parse_tarfiles(line):
-                    continue
-                elif parse_rpms(line):
-                    continue
-                else:
-                    raise service_error(service_error.internal, 
-                            "Bad tcl parse? %s" % line)
-            # Virtual topology and visualization
-            vtopo = self.gentopo(parse_vtopo.str)
-            if not vtopo:
-                raise service_error(service_error.internal, 
-                        "Failed to generate virtual topology")
-
-            vis = self.genviz(vtopo)
-            if not vis:
-                raise service_error(service_error.internal, 
-                        "Failed to generate visualization")
-
-            
+            # Allocate IP addresses: The allocator is a buddy system memory
+            # allocator.  Allocate from the largest substrate to the
+            # smallest to make the packing more likely to work - i.e.
+            # avoiding internal fragmentation.
+            top = topdl.topology_from_xml(file=split_data, top="experiment")
+            subs = sorted(top.substrates, 
+                    cmp=lambda x,y: cmp(len(x.interfaces), 
+                        len(y.interfaces)),
+                    reverse=True)
+            ips = ip_allocator(int(ip_addr("10.0.0.0")), 2 **24)
+            for s in subs:
+                a = ips.allocate(len(s.interfaces)+2)
+                if a :
+                    base, num = a
+                    if num < len(s.interfaces) +2 : 
+                        raise service_error(service_error.internal,
+                                "Allocator returned wrong number of IPs??")
+                else:
+                    raise service_error(service_error.req, 
+                            "Cannot allocate IP addresses")
+
+                base += 1
+                for i in s.interfaces:
+                    i.attribute.append(
+                            topdl.Attribute('ip4_address', 
+                                "%s" % ip_addr(base)))
+                    base += 1
+
+            # Find the testbeds to look up
+            testbeds = set([ a.value for e in top.elements \
+                    for a in e.attribute \
+                        if a.attribute == 'testbed'] )
+
+            # Make per testbed topologies.  Copy the main topo and remove
+            # interfaces and nodes that don't live in the testbed.
+            topo ={ }
+            for tb in testbeds:
+                self.get_access(tb, None, user, tbparams, master,
+                        export_project, access_user)
+                allocated[tb] = 1
+                topo[tb] = top.clone()
+                to_delete = [ ]
+                for e in topo[tb].elements:
+                    etb = e.get_attribute('testbed')
+                    if etb and etb != tb:
+                        for i in e.interface:
+                            for s in i.subs:
+                                try:
+                                    s.interfaces.remove(i)
+                                except ValueError:
+                                    raise service_error(service_error.internal,
+                                            "Can't remove interface??")
+                        to_delete.append(e)
+                for e in to_delete:
+                    topo[tb].elements.remove(e)
+                topo[tb].make_indices()
+
+
+
+            # Now, for each substrate in the main topology, find those that
+            # have nodes on more than one testbed.  Insert portal nodes
+            # into the copies of those substrates on the sub topologies.
+            for s in top.substrates:
+                tests = { }
+                for i in s.interfaces:
+                    e = i.element
+                    tb = e.get_attribute('testbed')
+                    if tb and not tests.has_key(tb):
+                        for i in e.interface:
+                            if s in i.subs:
+                                tests[tb]= \
+                                        i.get_attribute('ip4_address')
+                if len(tests) < 2:
+                    continue
+
+                # More than one testbed is on this substrate.  Insert
+                # some portals into the subtopologies.
+
+                for st in tests.keys():
+                    for dt in [ t for t in tests.keys() if t != st]:
+                        myname =  "%stunnel" % dt
+                        desthost  =  "%stunnel" % st
+                        sproject = tbparams[st].get('project', 'project')
+                        dproject = tbparams[dt].get('project', 'project')
+                        sdomain = ".%s.%s%s" % (eid, sproject,
+                                tbparams[st].get('domain', ".example.com"))
+                        ddomain = ".%s.%s%s" % (eid, dproject,
+                                tbparams[dt].get('domain', ".example.com"))
+                        boss = tbparams[master].get('boss', "boss")
+                        fs = tbparams[master].get('fs', "fs")
+                        event_server = "%s%s" % \
+                                (tbparams[st].get('eventserver', "event_server"),
+                                        tbparams[dt].get('domain', "example.com"))
+                        remote_event_server = "%s%s" % \
+                                (tbparams[dt].get('eventserver', "event_server"),
+                                        tbparams[dt].get('domain', "example.com"))
+                        seer_control = "%s%s" % \
+                                (tbparams[st].get('control', "control"), sdomain)
+                        local_key_dir = "/proj/%s/exp/%s/tmp" % ( sproject, eid)
+                        remote_conf_dir = "/proj/%s/exp/%s/tmp" % ( dproject, eid)
+                        conf_file = "%s%s.gw.conf" % (myname, sdomain)
+                        remote_conf_file = "%s%s.gw.conf" % (desthost, ddomain)
+                        # translate to lower case so the `hostname` hack for specifying
+                        # configuration files works.
+                        conf_file = conf_file.lower();
+                        remote_conf_file = remote_conf_file.lower();
+                        active = ("%s" % (st == master))
+                        portal = topdl.Computer(**{
+                                'name': "%stunnel" % dt,
+                                'attribute' : [{
+                                    'attribute': n,
+                                    'value': v,
+                                    } for n, v in (\
+                                            ('gateway', 'true'),
+                                            ('boss', boss),
+                                            ('fs', fs),
+                                            ('event_server', event_server),
+                                            ('remote_event_server', remote_event_server),
+                                            ('seer_control', seer_control),
+                                            ('local_key_dir', local_key_dir),
+                                            ('remote_conf_dir', remote_conf_dir),
+                                            ('conf_file', conf_file),
+                                            ('remote_conf_file', remote_conf_file),
+                                            ('remote_script_dir', "/usr/local/federation/bin"),
+                                            ('local_script_dir', "/usr/local/federation/bin"),
+                                            )],
+                                'interface': [{
+                                    'substrate': s.name,
+                                    'attribute': [ { 
+                                        'attribute': 'ip4_addreess', 
+                                        'value': tests[dt],
+                                        }, ],
+                                    }, ],
+                                })
+                        topo[st].elements.append(portal)
+            # Connect the gateway nodes into the topologies and clear out
+            # substrates that are not in the topologies
+            for tb in testbeds:
+                topo[tb].incorporate_elements()
+                topo[tb].substrates = \
+                        [s for s in topo[tb].substrates \
+                            if len(s.interfaces) >0]
+
+            # Copy the rpms and tarfiles to a distribution directory from
+            # which the federants can retrieve them
+            linkpath = "%s/software" %  expid
+            softdir ="%s/%s" % ( self.repodir, linkpath)
+            softmap = { }
+            pkgs = set([ t for l in [self.fedkit, self.gatewaykit] \
+                    for p, t in l ])
+            pkgs.update([x.location for e in top.elements \
+                    for x in e.software])
+            try:
+                os.makedirs(softdir)
+            except IOError, e:
+                raise service_error(
+                        "Cannot create software directory: %s" % e)
+            for pkg in pkgs:
+                loc = pkg
+
+                scheme, host, path = urlparse(loc)[0:3]
+                dest = os.path.basename(path)
+                if not scheme:
+                    if not loc.startswith('/'):
+                        loc = "/%s" % loc
+                    loc = "file://%s" %loc
+                try:
+                    u = urlopen(loc)
+                except Exception, e:
+                    raise service_error(service_error.req, 
+                            "Cannot open %s: %s" % (loc, e))
+                try:
+                    f = open("%s/%s" % (softdir, dest) , "w")
+                    data = u.read(4096)
+                    while data:
+                        f.write(data)
+                        data = u.read(4096)
+                    f.close()
+                    u.close()
+                except Exception, e:
+                    raise service_error(service_error.internal,
+                            "Could not copy %s: %s" % (loc, e))
+                path = re.sub("/tmp", "", linkpath)
+                # XXX
+                softmap[pkg] = \
+                        "https://users.isi.deterlab.net:23232/%s/%s" %\
+                        ( path, dest)
+
+                # Allow the individual testbeds to access the software.
+                for tb in tbparams.keys():
+                    self.auth.set_attribute(tbparams[tb]['allocID']['fedid'], 
+                            "%s/%s" % ( path, dest))
+
+            # Convert the software locations in the segments into the local
+            # copies on this host
+            for soft in [ s for tb in topo.values() \
+                    for e in tb.elements \
+                        for s in e.software ]:
+                if softmap.has_key(soft.location):
+                    soft.location = softmap[soft.location]
+
+            vtopo = topdl.topology_to_vtopo(top)
+            vis = self.genviz(vtopo)
+
             # save federant information
             for k in allocated.keys():
@@ -2543 +2672 @@
                     [ tbparams[tb]['federant'] for tb in tbparams.keys() \
                         if tbparams[tb].has_key('federant') ]
-            if self.state_filename: self.write_state()
+            if self.state_filename: 
+                self.write_state()
             self.state_lock.release()
-
-            # Copy tarfiles and rpms needed at remote sites into a staging area
-            try:
-                if self.fedkit:
-                    for t in self.fedkit:
-                        parse_tarfiles.list.append(t[1])
-                if self.gatewaykit:
-                    for t in self.gatewaykit:
-                        parse_tarfiles.list.append(t[1])
-                for t in parse_tarfiles.list:
-                    if not os.path.exists("%s/tarfiles" % tmpdir):
-                        os.mkdir("%s/tarfiles" % tmpdir)
-                    self.copy_file(t, "%s/tarfiles/%s" % \
-                            (tmpdir, os.path.basename(t)))
-                for r in parse_rpms.list:
-                    if not os.path.exists("%s/rpms" % tmpdir):
-                        os.mkdir("%s/rpms" % tmpdir)
-                    self.copy_file(r, "%s/rpms/%s" % \
-                            (tmpdir, os.path.basename(r)))
-                # A null experiment file in case we need to create a remote
-                # experiment from scratch
-                f = open("%s/null.tcl" % tmpdir, "w")
-                print >>f, """
-set ns [new Simulator]
-source tb_compat.tcl
-
-set a [$ns node]
-
-$ns rtproto Session
-$ns run 
-"""
-                f.close()
-
-            except IOError, e:
-                raise service_error(service_error.internal, 
-                        "Cannot stage tarfile/rpm: %s" % e.strerror)
-
         except service_error, e:
             # If something goes wrong in the parse (usually an access error)
@@ -2587 +2680 @@
             # exceptions.  Failing at this point returns a fault to the remote
             # caller. 
+
             self.state_lock.acquire()
             del self.state[eid]
@@ -2616 +2710 @@
         
         # Start a thread to do the resource allocation
-        t  = Thread(target=self.allocate_resources,
+        t  = Thread(target=self.new_allocate_resources,
                 args=(allocated, master, eid, expid, expcert, tbparams, 
-                    tmpdir, alloc_log),
+                    topo, tmpdir, alloc_log),
                 name=eid)
         t.start()

fedd/federation/topdl.py

@@ -48 +48 @@
 class Capacity(base):
     def __init__(self, rate, kind):
-        self.rate = rate
+        self.rate = float(rate)
         self.kind = kind
 
@@ -55 +55 @@
 
     def to_dict(self):
-        return { 'rate': self.rate, 'kind': self.kind }
+        return { 'rate': float(self.rate), 'kind': self.kind }
 
 class Latency(base):
     def __init__(self, time, kind):
-        self.time = time
+        self.time = float(time)
         self.kind = kind
 
@@ -66 +66 @@
 
     def to_dict(self):
-        return { 'time': self.time, 'kind': self.kind }
+        return { 'time': float(self.time), 'kind': self.kind }
 
 class Substrate(base):
@@ -117 +117 @@
 class Storage(base):
     def __init__(self, amount, persistence, attribute=[]):
-        if isinstance(amount, basestring):
-            self.amount = float(amount)
-        else:
-            self.amount = amount
+        self.amount = float(amount)
         self.presistence = persistence
         self.attribute = [ self.init_class(Attribute, a) \
@@ -411 +408 @@
         f = open(filename, "r")
         xp.ParseFile(f)
+        f.close()
     elif file:
         xp.ParseFile(file)
@@ -455 +453 @@
 
     for eidx, e in enumerate(t.elements):
-        n = { }
         if e.name: name = e.name[0]
         else: name = "unnamed_node%d" % eidx
         
         ips = [ ]
-        for i in e.interfaces:
+        for idx, i in enumerate(e.interface):
             ip = i.get_attribute('ip4_address')
             ips.append(ip)
@@ -490 +487 @@
             })
 
-        nodes.append(n)
-    return { 'vtopo': { 'node': node, 'lan': lan } }
+    return { 'node': nodes, 'lan': lans }
+
+def topology_to_ns2(t):
+    out = """
+set ns [new Simulator]
+source tb_compat.tcl
+
+"""
+    for e in t.elements:
+        rpms = ""
+        tarfiles = ""
+        if isinstance(e, Computer):
+            name = e.name[0]
+            out += "set %s [$ns node]\n" % name
+            if e.os and len(e.os) == 1:
+                osid = e.os[0].get_attribute('osid')
+                if osid:
+                    out += "tb-set-node-os $%s %s\n" % (name, osid)
+            for s in e.software:
+                if s.install:
+                    tarfiles += "%s %s " % (s.install, s.location)
+                else:
+                    rpms += "%s " % s.location
+            if rpms:
+                out += "tb-set-node-rpms $%s %s\n" % (name, rpms)
+            if tarfiles:
+                out += "tb-set-node-tarfiles $%s %s\n" % (name, tarfiles)
+            startcmd = e.get_attribute('startup')
+            if startcmd:
+                out+= 'tb-set-node-startcmd $%s "%s"\n' % (name, startcmd)
+            out+= "\n"
+    
+    for idx, s in enumerate(t.substrates):
+        loss = s.get_attribute('loss')
+        if s.latency:
+            delay = s.latency.time
+        else:
+            delay = 0
+        name = s.name or "sub%d" % idx
+
+        if len(s.interfaces) > 2:
+            # Lan
+            members = [ i.element.name[0] for i in s.interfaces]
+            out += 'set %s [$ns make-lan "%s" %f %fms ]\n' % \
+                    (name, " ".join(members), s.capacity.rate, delay)
+            if loss:
+                "tb-set-lan-loss $%s %f\n" % (name, float(loss))
+
+            for i in s.interfaces:
+                e = i.element
+                ip = e.get_attribute("ip4_address")
+                if ip:
+                    out += "tb-set-ip-lan $%s $%s %s\n" % (e.name, name, ip)
+                if i.capacity and i.capacity.rate != s.capacity.rate:
+                    out += "tb-set-node-lan-bandwidth $%s $%s %f\n" % \
+                            (e.name[0], name, i.capacity.rate)
+                if i.latency and i.latency.time != delay:
+                    out += "tb-set-node-lan-delay $%s $%s %fms\n" % \
+                            (e.name[0], name, i.latency.time)
+                iloss = i.get_attribute('loss')
+                if loss and iloss != loss :
+                    out += "tb-set-node-lan-loss $%s $%s %f" % \
+                            (e.name[0], name, float(loss))
+            out+= "\n"
+        elif len(s.interfaces) == 2:
+            f = s.interfaces[0]
+            t = s.interfaces[1]
+
+            out += "set %s [$ns duplex-link $%s $%s %f %fms DropTail]\n" %\
+                    (name, f.element.name[0], t.element.name[0],
+                            s.capacity.rate, delay)
+            if loss:
+                out += "tb-set-link-loss $%s %f\n" % (name, float(loss))
+
+            for i in s.interfaces:
+                lloss = i.get_attribute("loss")
+                cap_override = i.capacity and \
+                        i.capacity.rate != s.capacity.rate
+                delay_override = i.latency and \
+                        i.latency.time != delay
+                loss_override = lloss and lloss != loss
+                if cap_override or delay_override or loss_override:
+                    if i.capacity: cap = i.capacity.rate
+                    else: cap = s.capacity.rate
+
+                    if i.latency: delay = i.latency.time
+
+                    if lloss: loss = lloss
+                    else: loss = loss or 0.0
+
+                    out += "tb-set-link-simplex-params $%s $%s %fms %f %f\n" % \
+                            (name, i.element.name[0], delay, cap, loss)
+            out+= "\n"
+    out+="""
+$ns run
+"""
+    return out

fedd/federation/util.py

@@ -115 +115 @@
         else:
             raise ValueError("Badly formatted line in accessdb: %s line %d" %\
-                    (nf, lineno))
+                    (fn, lineno))
     f.close()
     return rv