Changeset 22a1a77 for fedd/federation/experiment_control.py

Timestamp:

Nov 29, 2011 6:19:24 PM (12 years ago)

Author:

Ted Faber <faber@…>

Branches:

compt_changes, info-ops, master

Children:

b709861

Parents:

57facae

Message:

Checkpoint: untested operations stuff

File:

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

fedd/federation/experiment_control.py

@@ -37 +37 @@
 from thread_pool import thread_pool, pooled_thread
 from experiment_info import experiment_info, allocation_info, federated_service
+from operation_status import operation_status
 
 import topdl
@@ -65 +66 @@
     call_TerminateSegment = service_caller('TerminateSegment')
     call_InfoSegment = service_caller('InfoSegment')
+    call_OperationSegment = service_caller('OperationSegment')
     call_Ns2Topdl = service_caller('Ns2Topdl')
 
@@ -223 +225 @@
                 'Info': soap_handler('Info', self.get_info),
                 'MultiInfo': soap_handler('MultiInfo', self.get_multi_info),
+                'Operation': soap_handler('Operation', self.do_operation),
                 'Terminate': soap_handler('Terminate', 
                     self.terminate_experiment),
@@ -238 +241 @@
                 'Terminate': xmlrpc_handler('Terminate',
                     self.terminate_experiment),
+                'Operation': xmlrpc_handler('Operation', self.do_operation),
                 'GetValue': xmlrpc_handler('GetValue', self.GetValue),
                 'SetValue': xmlrpc_handler('SetValue', self.SetValue),
@@ -933 +937 @@
             except service_error, e:
                 self.log.error("Info segment failed on %s: %s" % \
+                        (self.testbed, e))
+                return False
+
+    class operation_segment:
+        def __init__(self, debug=False, log=None, testbed="", cert_file=None,
+                cert_pwd=None, trusted_certs=None, caller=None,
+                log_collector=None):
+            self.log = log
+            self.debug = debug
+            self.cert_file = cert_file
+            self.cert_pwd = cert_pwd
+            self.trusted_certs = None
+            self.caller = caller
+            self.testbed = testbed
+            self.status = None
+
+        def __call__(self, uri, aid, op, targets, params):
+            req = { 
+                    'allocID': { 'fedid' : aid },
+                    'operation': op,
+                    'target': targets,
+                    }
+            if params: req['parameter'] = params
+
+
+            try:
+                self.log.debug("Calling OperationSegment at %s " % uri)
+                r = self.caller(uri, req, self.cert_file, self.cert_pwd,
+                        self.trusted_certs)
+                if 'OperationSegmentResponseBody' in r:
+                    r = r['OperationSegmentResponseBody']
+                    if 'status' in r:
+                        self.status = r['status']
+                else:
+                    raise service_error(service_error.internal, 
+                            "Bad response!?: %s" %r)
+                return True
+            except service_error, e:
+                self.log.error("Operation segment failed on %s: %s" % \
                         (self.testbed, e))
                 return False
@@ -2314 +2357 @@
             raise service_error(service_error.req, "No such experiment")
 
+    def operate_on_segments(self, op_params, cert, op, testbeds, params, 
+            results):
+        """
+        Call OperateSegment on multiple testbeds and gather the results.
+        op_params contains the parameters needed to contact that testbed, cert
+        is a certificate containing the fedid to use, op is the operation,
+        testbeds is a dict mapping testbed name to targets in that testbed,
+        params are the parameters to include a,d results is a growing list of
+        the results of the calls.
+        """
+        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)
+
+        try:
+            for tb, targets in testbeds.items():
+                if tb in op_params:
+                    uri, aid = op_params[tb]
+                    operate=self.operation_segment(log=self.log, testbed=uri,
+                                cert_file=cert_file, cert_pwd=None,
+                                trusted_certs=self.trusted_certs,
+                                caller=self.call_OperationSegment)
+                    if operate(uri, aid, op, targets, params):
+                        if operate.status is not None:
+                            results.extend(operate.status)
+                            continue
+                # Something went wrong in a weird way.  Add statuses 
+                # that reflect that to results
+                for t in targets:
+                    results.append(operation_status(t, 
+                        operation_status.federant,
+                        'Unexpected error ion %s' % tb))
+        # Clean up the tmpdir no matter what
+        finally:
+            if tmpdir: self.remove_dirs(tmpdir)
+
+    def do_operation(self, req, fid):
+        """
+        Find the testbeds holding each target and ask them to carry out the
+        operation.  Return the statuses.
+        """
+        # Map an element to the testbed containing it
+        def element_to_tb(e):
+            if isinstance(e, topdl.Computer): return e.get_attribute("testbed")
+            elif isinstance(e, topdl.Testbed): return e.name
+            else: return None
+        # If d is an operation_status object, make it a dict
+        def make_dict(d):
+            if isinstance(d, dict): return d
+            elif isinstance(d, operation_status): return d.to_dict()
+            else: return { }
+
+        req = req.get('OperationRequestBody', None)
+        if not req:
+            raise service_error(service_error.req,
+                    "Bad request format (no OperationRequestBody)")
+        exp = req.get('experiment', None)
+        op = req.get('operation', None)
+        target = set(req.get('target', []))
+        params = req.get('parameter', None)
+
+        if exp:
+            if 'fedid' in exp:
+                key = exp['fedid']
+                keytype = "fedid"
+            elif 'localname' in exp:
+                key = exp['localname']
+                keytype = "localname"
+            else:
+                raise service_error(service_error.req, "Unknown lookup type")
+        else:
+            raise service_error(service_error.req, "No request?")
+
+        if op is None or not target
+            raise service_error(service_error.req, "No request?")
+
+        proof = self.check_experiment_access(fid, key)
+        self.state_lock.acquire()
+        if key in self.state:
+            d1, op_params, cert, d2 = \
+                    self.get_segment_info(self.state[key], need_lock=False)
+            top = self.state[key].top
+            if top is not None:
+                top = top.clone()
+        self.state_lock.release()
+
+        if top is None:
+            raise service_error(service_error.partial, "No topology yet", 
+                    proof=proof)
+
+        testbeds = { }
+        results = []
+        for e in top.elements:
+            if e.name in targets:
+                tb = element_to_tb(e)
+                if tb is not None:
+                    if tb in testbeds: testbeds[tb].append(e.name)
+                    else: testbeds[tb] = [ e.name ]
+                else:
+                    results.append(operation_status(e.name, 
+                        code=operation_status.no_target, 
+                        description='Cannot map target to testbed'))
+
+        self.operate_on_segments(op_params, cert, op, testbeds, params,
+                results)
+
+        return { 
+                'experiment': exp, 
+                'status': [make_dict(r) for r in results]
+                'proof': proof.to_dict()
+                }
+
+
     def get_multi_info(self, req, fid):
         """