Changeset 6e33086 for fedd/federation/experiment_control.py

Timestamp:

Nov 28, 2011 7:34:11 PM (12 years ago)

Author:

Ted Faber <faber@…>

Branches:

compt_changes, info-ops, master

Children:

ec3aa4d

Parents:

45e880d

Message:

InfoSegment? to emulab access controllers

File:

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

fedd/federation/experiment_control.py

@@ -90 +90 @@
     call_StartSegment = service_caller('StartSegment')
     call_TerminateSegment = service_caller('TerminateSegment')
+    call_InfoSegment = service_caller('InfoSegment')
     call_Ns2Topdl = service_caller('Ns2Topdl')
 
@@ -158 +159 @@
                 or 'legacy'
         self.auth_dir = config.get('experiment_control', 'auth_dir')
+        # XXX: document this!
+        self.info_cache_limit = \
+                config.getint('experiment_control', 'info_cache', 600)
         if dt: self.direct_transit = [ tb.strip() for tb in dt.split(",")]
         else: self.direct_transit = [ ]
@@ -930 +934 @@
                 return False
 
+    class info_segment(start_segment):
+        def __init__(self, debug=False, log=None, testbed="", cert_file=None,
+                cert_pwd=None, trusted_certs=None, caller=None,
+                log_collector=None):
+            experiment_control_local.start_segment.__init__(self, debug, 
+                    log, testbed, cert_file, cert_pwd, trusted_certs, 
+                    caller, log_collector)
+
+        def __call__(self, uri, aid):
+            req = { 'allocID': { 'fedid' : aid } }
+
+            try:
+                self.log.debug("Calling InfoSegment at %s " % uri)
+                r = self.caller(uri, req, self.cert_file, self.cert_pwd,
+                        self.trusted_certs)
+                if r.has_key('InfoSegmentResponseBody'):
+                    self.make_map(r['InfoSegmentResponseBody'])
+                    if 'proof' in r: self.proof = r['proof']
+                    self.response = r
+                else:
+                    raise service_error(service_error.internal, 
+                            "Bad response!?: %s" %r)
+                return True
+            except service_error, e:
+                self.log.error("Info segment failed on %s: %s" % \
+                        (self.testbed, e))
+                return False
+
+    def annotate_topology(self, top, data):
+        def add_new(ann, attr):
+            for a in ann:
+                if a not in attr: attr.append(a)
+
+        # Annotate the topology with embedding info
+        for e in top.elements:
+            if isinstance(e, topdl.Computer):
+                for s in data:
+                    ann = s.node.get(e.name, None)
+                    if ann is not None:
+                        add_new(ann[0], e.localname)
+                        e.status = ann[1]
+                        add_new(ann[2], e.service)
+                        add_new(ann[3], e.operation)
+                        break
+
+
+
     def allocate_resources(self, allocated, masters, eid, expid, 
             tbparams, top, topo, tmpdir, alloc_log=None, log_collector=None, 
@@ -1046 +1097 @@
                 self.state_lock.acquire()
                 self.state[eid].status = 'failed'
+                self.state[eid].updated()
                 if self.state_filename: self.write_state()
                 self.state_lock.release()
@@ -1065 +1117 @@
                 if s.proof:
                     proofs[s.testbed] = s.proof
-
-            # Annotate the topology with embedding info
-            for e in top.elements:
-                if isinstance(e, topdl.Computer):
-                    for s in starters:
-                        ann = s.node.get(e.name, None)
-                        if ann is not None:
-                            e.localname.extend(ann[0])
-                            e.status = ann[1]
-                            e.service.extend(ann[2])
-                            e.operation.extend(ann[3])
-                            break
-
+            self.annotate_topology(top, starters)
             log.info("[start_segment]: Experiment %s active" % eid)
 
@@ -1092 +1132 @@
         self.state[eid] = self.state[expid]
         self.state[eid].top = top
+        self.state[eid].updated()
         # Append startup proofs
         for f in self.state[eid].get_all_allocations():
@@ -1645 +1686 @@
             exp = self.state[key]
             exp.status = "starting"
+            exp.updated()
             expid = exp.fedid
             eid = exp.localname
@@ -1980 +2022 @@
         proofs = [copy.deepcopy(p) for k in tbparams.keys()\
                     for p in tbparams[k].proof]
+        exp.updated()
         if self.state_filename: 
             self.write_state()
@@ -2192 +2235 @@
             return (None, None)
 
+    def update_info(self, key, force=False):
+        top = None
+        self.state_lock.acquire()
+        if key in self.state:
+            if force or self.state[key].older_than(self.info_cache_limit):
+                top = self.state[key].top 
+                if top is not None: top = top.clone()
+                d1, info_params, cert, d2 = \
+                        self.get_segment_info(self.state[key], need_lock=False)
+        self.state_lock.release()
+
+        if top is None: return
+
+        try:
+            tmpdir = tempfile.mkdtemp(prefix="info-")
+        except EnvironmentError:
+            raise service_error(service_error.internal, 
+                    "Cannot create tmp dir")
+        cert_file = self.make_temp_certfile(cert, tmpdir)
+
+        data = []
+        try:
+            for k, (uri, aid) in info_params.items():
+                info=self.info_segment(log=self.log, testbed=uri,
+                            cert_file=cert_file, cert_pwd=None,
+                            trusted_certs=self.trusted_certs,
+                            caller=self.call_InfoSegment)
+                info(uri, aid)
+                data.append(info)
+        # Clean up the tmpdir no matter what
+        finally:
+            if tmpdir: self.remove_dirs(tmpdir)
+
+        self.annotate_topology(top, data)
+        self.state_lock.acquire()
+        if key in self.state:
+            self.state[key].top = top
+            self.state[key].updated()
+            if self.state_filename: self.write_state()
+        self.state_lock.release()
+
     
     def get_info(self, req, fid):
@@ -2205 +2289 @@
         exp = req.get('experiment', None)
         legacy = req.get('legacy', False)
+        fresh = req.get('fresh', False)
         if exp:
             if exp.has_key('fedid'):
@@ -2219 +2304 @@
         proof = self.check_experiment_access(fid, key)
 
+        self.update_info(key, fresh)
+
         self.state_lock.acquire()
         if self.state.has_key(key):
             rv = self.state[key].get_info()
-            # Copy the topo if we need to generate legacy representations
+            # Copy the topo if we need legacy annotations
             if legacy:
                 top = self.state[key].top 
@@ -2229 +2316 @@
 
         # If the legacy visualization and topology representations are
-        # requested, calculate them and add them to teh return.
+        # requested, calculate them and add them to the return.
         if legacy and rv is not None:
             if top is not None:
@@ -2300 +2387 @@
                     "Experiment has no status!?")
 
-
-    def get_termination_info(self, fed_exp):
+    def get_segment_info(self, fed_exp, need_lock=True):
         ids = []
         term_params = { }
-        self.state_lock.acquire()
+        if need_lock: self.state_lock.acquire()
         ids = [ x for x in (fed_exp.localname, fed_exp.fedid) if x is not None ]
         expcert = fed_exp.identity
@@ -2315 +2401 @@
             aid = fed.allocID
             term_params[aid] = (uri, aid)
+        if need_lock: self.state_lock.release()
+        return ids, term_params, expcert, repo
+
+
+    def get_termination_info(self, fed_exp):
+        self.state_lock.acquire()
+        ids, term_params, expcert, repo = self.get_segment_info(fed_exp, False)
         # Change the experiment state
         fed_exp.status = 'terminating'
+        fed_exp.updated()
         if self.state_filename: self.write_state()
         self.state_lock.release()
@@ -2414 +2508 @@
         if fed_exp:
             status = self.check_termination_status(fed_exp, force)
+            # get_termination_info updates the experiment state
             ids, term_params, expcert, repo = self.get_termination_info(fed_exp)
             self.deallocate_resources(term_params, expcert, status, force,