Changeset b4b19c7 for fedd/federation

Timestamp:

Apr 21, 2010 5:31:03 AM (15 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

Location:

fedd/federation

Files:

• emulab_access.py (modified) (2 diffs)
• experiment_control.py (modified) (13 diffs)
• local_emulab_segment.py (modified) (3 diffs)
• protogeni_access.py (modified) (2 diffs)
• proxy_emulab_segment.py (modified) (2 diffs)
• proxy_protogeni_segment.py (modified) (2 diffs)

fedd/federation/emulab_access.py

@@ -1552 +1552 @@
                     debug=self.create_debug, log=alloc_log)
             rv = starter(self, ename, proj, user, expfile, tmpdir)
+            # Copy the assigned names into the return topology
+            rvtopo = topo.clone()
+            for e in [ e for e in rvtopo.elements \
+                   if isinstance(e, topdl.Computer)]:
+                for n in e.name:
+                    if n in starter.node:
+                        e.set_attribute('hostname', starter.node[n])
         except service_error, e:
             err = e
@@ -1581 +1588 @@
                     'allocationLog': logv,
                     }
-            retval = self.allocation[aid]['started']
+            retval = copy.copy(self.allocation[aid]['started'])
             self.write_state()
             self.state_lock.release()
-
+            retval['segmentdescription'] =  \
+                    { 'topdldescription': rvtopo.to_dict() }
             return retval
         elif err:

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
 

fedd/federation/local_emulab_segment.py

@@ -90 +90 @@
 $ns run 
 """
+        self.node = { }
 
     def get_state(self, pid, eid):
@@ -138 +139 @@
             self.log.debug("State is %s" % state)
             return state
+
+    def get_mapping(self, pid, eid):
+        # command to test experiment state
+        expinfo_exec = "/usr/testbed/bin/expinfo"  
+        # The expinfo command. 
+        cmd = [ expinfo_exec, '-m', 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:
+            rv = 0
+        else:
+            self.log.debug("Getting mapping for %s %s" % (pid, eid))
+            phys_start = re.compile('^Physical\s+Node\s+Mapping')
+            phys_line = re.compile('(\S+)\s+\S+\s+\S+\s+(.*)')
+            phys_end = re.compile('^$')
+            status = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+                    stderr=dev_null, close_fds=True)
+
+            # Parse the info output.  Format:
+            #
+            # stuff
+            # Physical Node Mapping:
+            # ID              Type         OS              Physical    
+            # --------------- ------------ --------------- ------------
+            # virtual         dummy        dummy           physical
+            #
+            foundit = False
+            skip = 0
+            for line in status.stdout:
+                if phys_start.match(line):
+                    skip = 2
+                    foundit = True
+                elif not foundit:
+                    continue
+                elif skip > 0:
+                    skip -= 1
+                elif phys_end.match(line):
+                    break
+                else:
+                    m = phys_line.match(line.strip())
+                    if m: self.node[m.group(1)] = m.group(2)
+                    else: self.log.warn(
+                            "Matching failed while parsing node mapping")
+            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 :
+            raise service_error(service_error.internal,
+                    "Cannot get node mapping of segment:%s/%s" % (pid, eid))
+        else:
+            return True
+
 
 
@@ -242 +302 @@
                 state = self.get_state(pid, eid)
                 self.log.debug("[start_segment]: state is %s" % state)
-                return state == 'active'
+                if state != 'active':
+                    return False
         # Everything has gone OK.
+        self.get_mapping(pid,eid)
         return True
 

fedd/federation/protogeni_access.py

@@ -1097 +1097 @@
                     "%s/%s" % (self.staging_dir, ename), tmpdir, cf, cpw,
                     certfile, topo, connInfo, services)
+            # Copy the assigned names into the return topology
+            rvtopo = topo.clone()
+            for e in [ e for e in rvtopo.elements \
+                   if isinstance(e, topdl.Computer)]:
+                for n in e.name:
+                    if n in starter.node:
+                        e.set_attribute('hostname', starter.node[n])
         except service_error, e:
             err = e
@@ -1126 +1133 @@
                     'allocationLog': logv,
                     }
-            retval = self.allocation[aid]['started']
+            retval = copy.copy(self.allocation[aid]['started'])
             self.write_state()
+            retval['segmentdescription'] =  \
+                    { 'topdldescription': rvtopo.to_dict() }
             self.state_lock.release()
 

fedd/federation/proxy_emulab_segment.py

@@ -25 +25 @@
 $ns run 
 """
+        self.node = { }
 
     def get_state(self, user, host, pid, eid):
@@ -78 +79 @@
 
 
+    def get_mapping(self, pid, eid):
+        # command to test experiment state
+        expinfo_exec = "/usr/testbed/bin/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 no', '-i', 
+                self.ssh_privkey_file, "%s@%s" % (user, host), 
+                expinfo_exec, '-m', 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:
+            rv = 0
+        else:
+            self.log.debug("Getting mapping for %s %s" % (pid, eid))
+            phys_start = re.compile('^Physical\s+Node\s+Mapping')
+            phys_line = re.compile('(\S+)\s+\S+\s+\S+\s+(.*)')
+            phys_end = re.compile('^$')
+            status = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+                    stderr=dev_null, close_fds=True)
+
+            # Parse the info output.  Format:
+            #
+            # stuff
+            # Physical Node Mapping:
+            # ID              Type         OS              Physical    
+            # --------------- ------------ --------------- ------------
+            # virtual         dummy        dummy           physical
+            #
+            foundit = False
+            skip = 0
+            for line in status.stdout:
+                if phys_start.match(line):
+                    skip = 2
+                    foundit = True
+                elif not foundit:
+                    continue
+                elif skip > 0:
+                    skip -= 1
+                elif phys_end.match(line):
+                    break
+                else:
+                    m = phys_line.match(line.strip())
+                    if m: self.node[m.group(1)] = m.group(2)
+                    else: self.log.warn(
+                            "Matching failed while parsing node mapping")
+            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 :
+            raise service_error(service_error.internal,
+                    "Cannot get node mapping of segment:%s/%s" % (pid, eid))
+        else:
+            return True
+
+
     def __call__(self, parent, eid, pid, user, tclfile, tmpdir, timeout=0):
         """

fedd/federation/proxy_protogeni_segment.py

@@ -86 +86 @@
         segment_base.__init__(self, log=log, keyfile=keyfile, debug=debug,
                 ch_url=cm_url, sa_url=sa_url, cm_url=cm_url)
+        self.node = { }
 
 
@@ -131 +132 @@
 
         p.Parse(manifest)
+        # Make the node dict that the callers expect
+        for k in mp.d:
+            self.node[k] = mp.d.get('hostname', '')
         return mp.d