Changeset 109a32a

Timestamp:

Mar 1, 2010 4:38:17 AM (14 years ago)

Author:

Ted Faber <faber@…>

Branches:

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

Children:

8353ac6

Parents:

2761484

Message:

More dragon cleanup and finally removing the last "special" testbed. We
self-organize now.

Tested w/o full actually allocating.

File:

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

fedd/federation/experiment_control.py

@@ -1031 +1031 @@
             tbparams, topo, tmpdir, alloc_log=None, log_collector=None, 
             attrs=None, connInfo={}, services=[]):
-        def get_vlan(r):
-            if r.has_key('StartSegmentResponseBody'):
-                srb = r['StartSegmentResponseBody']
-                if srb.has_key('fedAttr'):
-                    for k, v in [ (a['attribute'], a['value']) \
-                            for a in srb['fedAttr']]:
-                        if k == 'vlan': return v
-            return None
 
         started = { }           # Testbeds where a sub-experiment started 
@@ -1046 +1038 @@
         fail_soft = False
 
-        non_transit = [ k for k in allocated.keys() \
-                if not topo[k].get_attribute('transit')]
-        transit = [ k for k in allocated.keys() \
-                if topo[k].get_attribute('transit')]
-
         log = alloc_log or self.log
 
@@ -1056 +1043 @@
         threads = [ ]
 
-        for tb in transit:
-            uri = tbparams[tb]['uri']
+        for tb in allocated.keys():
+            # Create and start a thread to start the segment, and save it
+            # to get the return value later
+            thread_pool.wait_for_slot()
+            uri = tbparams[tb].get('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'):
@@ -1065 +1059 @@
                         "No alloc id for testbed %s !?" % tb)
 
-            m = re.search('(\d+)', tb)
-            if m:
-                to_repl = "unassigned%s" % m.group(1)
-            else:
-                raise service_error(service_error.internal, 
-                        "Bad dynamic allocation name")
-                break
-
-            ss = self.start_segment(log=log, debug=self.debug, 
-                testbed=tb, cert_file=self.cert_file, 
-                cert_pwd=self.cert_pwd, 
-                trusted_certs=self.trusted_certs,
-                caller=self.call_StartSegment,
-                log_collector=log_collector)
-            t = self.pooled_thread(
-                    target=ss,
-                    args =(uri, aid, topo[tb], False, attrs, connInfo[tb],
-                        services),
-                    name=tb, pdata=thread_pool, trace_file=self.trace_file)
+            t  = self.pooled_thread(\
+                    target=self.start_segment(log=log, debug=self.debug,
+                        testbed=tb, cert_file=self.cert_file,
+                        cert_pwd=self.cert_pwd,
+                        trusted_certs=self.trusted_certs,
+                        caller=self.call_StartSegment,
+                        log_collector=log_collector), 
+                    args=(uri, aid, topo[tb], tb == master, 
+                        attrs, connInfo[tb], services),
+                    name=tb,
+                    pdata=thread_pool, trace_file=self.trace_file)
             threads.append(t)
             t.start()
-            # Wait until the this transit node finishes (keep pinging the log,
-            # though)
-
-            mins = 0
-            while not thread_pool.wait_for_all_done(60.0):
-                mins += 1
-                alloc_log.info("Waiting for transit (it has been %d mins)" \
-                        % mins)
-
-            if t.rv:
-                vlan = get_vlan(ss.response)
-                if vlan is not None:
-                    for v in connInfo.values():
-                        for i in v:
-                            for a in i.get('fedAttr', []):
-                                if a.get('attribute', "") == 'vlan_id' and \
-                                        a.get('value', "") == to_repl:
-                                    a['value'] = vlan
-            else:
-                break
-            thread_pool.clear()
-
-
-        failed = [ t.getName() for t in threads if not t.rv ]
-
-        if len(failed) == 0:
-            for tb in non_transit:
-                # 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.start_segment(log=log, debug=self.debug,
-                            testbed=tb, cert_file=self.cert_file,
-                            cert_pwd=self.cert_pwd,
-                            trusted_certs=self.trusted_certs,
-                            caller=self.call_StartSegment,
-                            log_collector=log_collector), 
-                        args=(uri, aid, topo[tb], tb == master, 
-                            attrs, connInfo[tb], services),
-                        name=tb,
-                        pdata=thread_pool, trace_file=self.trace_file)
-                threads.append(t)
-                t.start()
-
-            # Wait until all finish (keep pinging the log, though)
-            mins = 0
-            while not thread_pool.wait_for_all_done(60.0):
-                mins += 1
-                alloc_log.info("Waiting for sub threads (it has been %d mins)" \
-                        % mins)
-
-            thread_pool.clear()
+
+        # Wait until all finish (keep pinging the log, though)
+        mins = 0
+        while not thread_pool.wait_for_all_done(60.0):
+            mins += 1
+            alloc_log.info("Waiting for sub threads (it has been %d mins)" \
+                    % mins)
+
+        thread_pool.clear()
 
         failed = [ t.getName() for t in threads if not t.rv ]
@@ -1419 +1350 @@
                 aid = tbparams[st]['allocID']['fedid']
             except:
-                self.log.debug("Can't get allocation id?")
+                self.log.debug("[new_portal_node] Can't get alloc id for %s?" \
+                        % st)
                 aid = None
             info = {
@@ -1511 +1443 @@
                             ('vlan', 'unassigned%d' % idx),)])
         name = "dragon%d" % idx
+        store_key = 'fedid:%s/vlan%d' % (expid, idx)
         for tb in tbs.keys():
             seg = topdl.Segment(
@@ -1529 +1462 @@
             segs.append(seg)
 
-
-        try:
-            aid = tbparams[tb]['allocID']['fedid']
-        except:
-            self.log.debug("Can't get allocation id?")
-            aid = None
+            # Give this allocation the rights to access the key of the
+            # vlan_id
+            try:
+                aid = tbparams[tb]['allocID']['fedid']
+                self.auth.set_attribute(aid, store_key)
+            except:
+                self.log.debug("[new_dragon_topo] Can't get alloc id for %s?"\
+                        % tb)
+
         connInfo[name] = [ { 
+            'type': 'transit',
             'parameter': [ { 
                 'name': 'vlan_id',
-                'key': "fedid:%s/vlan%d" % (expid, idx),
+                'key': store_key,
                 'store': self.store_url,
                 'type': 'output'
                 } ]
             } ]
-
-        # Give this allocation the rights to access the key of the
-        # vlan_id
-        if aid:
-            self.auth.set_attribute(aid, 'fedid:%s/vlan%d' % (expid, idx))
 
         topo[name] = \
@@ -1554 +1486 @@
                             topdl.Attribute(attribute="transit", value='true'),
                             topdl.Attribute(attribute="dynamic", value='true'),
-                            topdl.Attribute(attribute="testbed", value='dragon'),
+                            topdl.Attribute(attribute="testbed", 
+                                value='dragon'),
+                            topdl.Attribute(attribute="store_keys", 
+                                value=store_key),
                             ]
                         )
 
     def create_dragon_substrate(self, sub, topo, tbs, tbparams, master, eid,
-            connInfo, peer={ }, expid=None):
+            connInfo, expid=None):
         """
         Add attribiutes to the various elements indicating that they are to be
@@ -1570 +1505 @@
                 if s.name == name: return s
             else: return None
+
 
         mdomain = tbparams[master].get('domain', '.example.com')
@@ -1597 +1533 @@
                     aid = tbparams[tb]['allocID']['fedid']
                 except:
-                    self.log.debug("Can't get allocation id?")
+                    self.log.debug("[creat_dragon_substrate] " + 
+                            "Can't get alloc id for %s?" %tb)
                     aid = None
 
@@ -1612 +1549 @@
                                     if isinstance(i.element, topdl.Computer) ],
                         'fedAttr': [ 
-                            { 'attribute': 'vlan_id', 
-                                'value': 'unassigned%d' % dn },
                             { 'attribute': 'masterdomain', 'value': mdomain},
                             { 'attribute': 'masterexperiment', 'value': 
@@ -1626 +1561 @@
                             } ]
                         }
-                if peer.has_key(tb):
-                    info['peer'] = peer[tb]
+                if tbs.has_key(tb):
+                    info['peer'] = tbs[tb]
                 connInfo[tb].append(info)
 
@@ -1639 +1574 @@
                         "No substrate %s in testbed %s" % (sub.name, tb))
 
-        self.new_dragon_topo(dn, sub, topo, tbs, tbparams)
+        self.new_dragon_topo(dn, sub, topo, tbs, tbparams, connInfo, expid)
 
     def insert_internet_portals(self, sub, topo, tbs, tbparams, master, eid,
@@ -1892 +1827 @@
 
                     self.create_dragon_substrate(csub, topo, 
-                            {tb: 1, master:1}, tbparams, master, eid, connInfo,
-                            {tb: ip_addr(mip), master: ip_addr(dip)}, expid)
+                            {tb: ip_addr(mip), master: ip_addr(dip)}, 
+                            tbparams, master, eid, connInfo,
+                            expid)
                 else:
                     self.add_control_portal(master, tb, master, eid, topo, 
@@ -2239 +2175 @@
                     del tbparams[tb]
                     allocated[k] = 1
+                    store_keys = t.get_attribute('store_keys')
+                    # Give the testbed access to keys it exports or imports
+                    if store_keys:
+                        for sk in store_keys.split(" "):
+                            self.auth.set_attribute(\
+                                    tbparams[k]['allocID']['fedid'], sk)
                 else:
                     raise service_error(service_error.internal, 
@@ -2748 +2690 @@
             if v is not None:
                 rv['value'] = v
+            self.log.debug("[GetValue] got %s from %s" % (v, name))
             return rv
         else:
@@ -2769 +2712 @@
                 self.synch_store.set_value(name, v)
                 self.write_store()
+                self.log.debug("[SetValue] set %s to %s" % (name, v))
             except synch_store.CollisionError:
                 # Translate into a service_error