Changeset b4b19c7 for fedd/federation/experiment_control.py

Timestamp:

Apr 21, 2010 5:31:03 AM (14 years ago)

Author:

Ted Faber <faber@…>

Branches:

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

Children:

f7a54c6

Parents:

f54e8e4

Message:

Get topology information into the info operation, as annotations of a topology description. This required adding such information to the start segment replies as well

File:

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

fedd/federation/experiment_control.py

@@ -1007 +1007 @@
             self.log_collector = log_collector
             self.response = None
+            self.node = { }
+
+        def make_map(self, resp):
+            if 'segmentdescription' in resp and \
+                    'topdldescription' in resp['segmentdescription']:
+                top = topdl.Topology(\
+                        **resp['segmentdescription']['topdldescription'])
+                for e in [e for e in top.elements \
+                        if isinstance(e, topdl.Computer)]:
+                    hn = e.get_attribute('hostname')
+                    if hn:
+                        for n in e.name:
+                            self.node[n] = hn
 
         def __call__(self, uri, aid, topo, masters, attrs=None, connInfo=None):
@@ -1050 +1063 @@
                         for line in  lval.splitlines(True):
                             self.log_collector.write(line)
+                    self.make_map(r['StartSegmentResponseBody'])
                     self.response = r
                 else:
@@ -1088 +1102 @@
 
     def allocate_resources(self, allocated, masters, eid, expid, 
-            tbparams, topo, tmpdir, alloc_log=None, log_collector=None, 
+            tbparams, top, topo, tmpdir, alloc_log=None, log_collector=None, 
             attrs=None, connInfo={}):
 
@@ -1101 +1115 @@
         thread_pool = self.thread_pool(self.nthreads)
         threads = [ ]
+        starters = [ ]
 
         for tb in allocated.keys():
@@ -1118 +1133 @@
                         "No alloc id for testbed %s !?" % tb)
 
+            s = 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)
+            starters.append(s)
             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), 
+                    target=s, name=tb,
                     args=(uri, aid, topo[tb], masters, attrs, connInfo[tb]),
-                    name=tb,
                     pdata=thread_pool, trace_file=self.trace_file)
             threads.append(t)
@@ -1190 +1205 @@
                 return
         else:
+            # Walk through the successes and gather the virtual to physical
+            # mapping.
+            node = { }
+            for s in starters:
+                node.update(s.node)
+            # Assigng the mapping as a hostname attribute
+            for e in [ e for e in top.elements \
+                    if isinstance(e, topdl.Computer)]:
+                for n in e.name:
+                    if n in node:
+                        e.set_attribute('hostname', node[n])
             log.info("[start_segment]: Experiment %s active" % eid)
 
@@ -1205 +1231 @@
             log.debug("[start_experiment]: not removing %s" % tmpdir)
 
-        # Insert the experiment into our state and update the disk copy
+        # 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]
+        self.state[eid]['experimentdescription']['topdldescription'] = \
+                top.to_dict()
         if self.state_filename: self.write_state()
         self.state_lock.release()
@@ -1228 +1256 @@
 
 
-    def create_experiment_state(self, fid, req, expid, expcert, 
+    def create_experiment_state(self, fid, req, expid, expcert,
             state='starting'):
         """
@@ -1662 +1690 @@
                 # Add the service to masters
                 for tb in s.get('export', []):
-                    if s.get('name', None) and s.get('import', None):
+                    if s.get('name', None):
                         if tb not in masters:
                             masters[tb] = [ ]
@@ -1676 +1704 @@
                                 params=params))
                     else:
-                        log.error('Testbed service does not have name " + \
+                        self.log.error('Testbed service does not have name " + \
                                 "and importers')
 
@@ -1761 +1789 @@
             self.state[eid]['vtopo'] = vtopo
             self.state[eid]['vis'] = vis
+            self.state[eid]['experimentdescription'] = \
+                    { 'topdldescription': top.clone() }
             self.state[expid]['federant'] = \
                     [ tbparams[tb]['federant'] for tb in tbparams.keys() \
@@ -1833 +1863 @@
         t  = Thread(target=self.allocate_resources,
                 args=(allocated, masters, eid, expid, tbparams, 
-                    topo, tmpdir, alloc_log, alloc_collector, attrs, connInfo),
+                    top, topo, tmpdir, alloc_log, alloc_collector, attrs,
+                    connInfo),
                 name=eid)
         t.start()
@@ -2011 +2042 @@
                 if f.has_key('allocID'): del f['allocID']
                 if f.has_key('uri'): del f['uri']
+
         return rv