Changeset 22a1a77

Timestamp:

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

Author:

Ted Faber <faber@…>

Branches:

compt_changes, info-ops, master

Children:

b709861

Parents:

57facae

Message:

Checkpoint: untested operations stuff

Files:

• fedd/federation/experiment_control.py (modified) (6 diffs)
• fedd/federation/operation_status.py (added)
• wsdl/fedd.wsdl (modified) (7 diffs)
• wsdl/fedd_types.xsd (modified) (3 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):
         """

wsdl/fedd.wsdl

@@ -74 +74 @@
   </message>
 
+  <message name="OperationRequestMessage">
+    <part name="OperationRequestBody" type="xsd1:operationRequestType"/>
+  </message>
+
+  <message name="OperationResponseMessage">
+    <part name="OperationResponseBody" type="xsd1:operationResponseType"/>
+  </message>
+
   <message name="MultiInfoRequestMessage">
     <part name="MultiInfoRequestBody" type="xsd1:multiInfoRequestType"/>
@@ -116 +124 @@
     <part name="InfoSegmentResponseBody" 
       type="xsd1:infoSegmentResponseType"/>
+  </message>
+
+  <message name="OperationSegmentRequestMessage">
+    <part name="OperationSegmentRequestBody" 
+      type="xsd1:operationSegmentRequestType"/>
+  </message>
+
+  <message name="OperationSegmentResponseMessage">
+    <part name="OperationSegmentResponseBody" 
+      type="xsd1:operationSegmentResponseType"/>
   </message>
 
@@ -214 +232 @@
     </operation>
 
+
+    <operation name="Operation">
+      <documentation>
+        A one-stop request for meta-data on the experiment.  Includes all the
+        info from a Vtopo and a Vis request.
+      </documentation>
+      <input message="tns:OperationRequestMessage"/>
+      <output message="tns:OperationResponseMessage"/>
+      <fault name="OperationFeddFault" message="tns:FaultMessage"/>
+    </operation>
+
     <operation name="MultiInfo">
       <documentation>
@@ -251 +280 @@
     </operation>
 
-
     <operation name="InfoSegment">
       <documentation>
@@ -261 +289 @@
     </operation>
 
+    <operation name="OperationSegment">
+      <documentation>
+        Get info about a segment
+      </documentation>
+      <input message="tns:OperationSegmentRequestMessage"/>
+      <output message="tns:OperationSegmentResponseMessage"/>
+      <fault name="OperationSegmentFeddFault" message="tns:FaultMessage"/>
+    </operation>
+
     <operation name="SetValue">
       <documentation>
@@ -436 +473 @@
         </fault>
       </operation>
+      <operation name="Operation">
+        <documentation>
+          The bindings of this operation are straightforward SOAP RPC 1.1.
+        </documentation>
+        <soap:operation soapAction="Operation"/> 
+        <input>
+          <soap:body use="literal" parts="tns:OperationRequestBody"
+            namespace="http://www.isi.edu/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </input>
+        <output>
+          <soap:body use="literal" parts="tns:OperationResponseBody"
+            namespace="http://www.isi.edu/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </output>
+        <fault name="OperationFeddFault">
+          <soap:fault use="literal"  name="tns:FeddFault"
+            namespace="http://www.isi.edu/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </fault>
+      </operation>
       <operation name="MultiInfo">
         <documentation>
@@ -536 +594 @@
         </output>
         <fault name="InfoSegmentFeddFault">
+          <soap:fault use="literal"  name="tns:FeddFault"
+            namespace="http://www.isi.edu/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </fault>
+      </operation>
+      <operation name="OperationSegment">
+        <documentation>
+          The bindings of this operation are straightforward SOAP RPC 1.1.
+        </documentation>
+        <soap:operation soapAction="OperationSegment"/> 
+        <input>
+          <soap:body use="literal" parts="tns:OperationSegmentRequestBody"
+            namespace="http://www.isi.edu/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </input>
+        <output>
+          <soap:body use="literal" parts="tns:OperationSegmentResponseBody"
+            namespace="http://www.isi.edu/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </output>
+        <fault name="OperationSegmentFeddFault">
           <soap:fault use="literal"  name="tns:FeddFault"
             namespace="http://www.isi.edu/fedd.wsdl"

wsdl/fedd_types.xsd

@@ -455 +455 @@
   </xsd:complexType>
 
+  <xsd:complexType name="operationStatusType">
+    <xsd:annotation>
+      <xsd:documentation>
+        Result of an operation.  The target, success or failure code and
+        descriptive text
+      </xsd:documentation>
+    </xsd:annotation>
+    <xsd:sequence>
+      <xsd:element name="target" type="xsd:string"/>
+      <xsd:element name="code" type="xsd:int">
+        <xsd:restriction>
+          <xsd:enumeration value="0"/>  <!-- success -->
+          <xsd:enumeration value="1"/>  <!-- access denied -->
+          <xsd:enumeration value="2"/>  <!-- busy, retry -->
+          <xsd:enumeration value="3"/>  <!-- not supported -->
+          <xsd:enumeration value="4"/>  <!-- bad parameter -->
+          <xsd:enumeration value="5"/>  <!-- internal error -->
+          <xsd:enumeration value="6"/>  <!-- partial success -->
+          <xsd:enumeration value="7"/>  <!-- no such target -->
+          <xsd:enumeration value="8"/>  <!-- federant error -->
+        </xsd:restriction>
+      </xsd:element>
+      <xsd:element name="description" type="xsd:string" 
+        minOccurs="0" maxOccurs="1"/>
+    </xsd:sequence>
+  </xsd:complexType>
+
+
   <xsd:complexType name="newRequestType">
     <xsd:annotation>
@@ -733 +761 @@
   </xsd:complexType>
 
+  <xsd:complexType name="operationRequestType">
+    <xsd:annotation>
+      <xsd:documentation>
+        A request to operate on one or more elements of this experiment
+      </xsd:documentation>
+    </xsd:annotation>
+    <xsd:sequence>
+      <xsd:element name="experiment" type="tns:IDType"/>
+      <xsd:element name="operation" type="xsd:string"/>
+      <xsd:element name="target" type="xsd:string"
+        minOccurs="1" maxOccurs="unbounded"/>
+      <xsd:element name="parameter" type="tns:fedAttrType"
+        minOccurs="0" maxOccurs="unbounded"/>
+    </xsd:sequence>
+  </xsd:complexType>
+
+  <xsd:complexType name="operationResponseType">
+    <xsd:annotation>
+      <xsd:documentation>
+        A status of requested operations.
+      </xsd:documentation>
+    </xsd:annotation>
+    <xsd:sequence>
+      <xsd:element name="experiment" type="tns:IDType"/>
+      <xsd:element name="status" type="tns:operationStatusType"
+        minOccurs="1" maxOccurs="unbounded"/>
+      <xsd:element name="proof" type="tns:proofType" />
+    </xsd:sequence>
+  </xsd:complexType>
+
   <xsd:complexType name="terminateRequestType">
     <xsd:annotation>
@@ -846 +904 @@
       <xsd:element name="fedAttr" type="tns:fedAttrType" minOccurs="0"
         maxOccurs="unbounded"/>
+      <xsd:element name="proof" type="tns:proofType" />
+    </xsd:sequence>
+  </xsd:complexType>
+
+  <xsd:complexType name="operationSegmentRequestType">
+    <xsd:annotation>
+      <xsd:documentation>
+        A request to operate on one or more elements of this segment (or the
+        whole segment)
+      </xsd:documentation>
+    </xsd:annotation>
+    <xsd:sequence>
+      <xsd:element name="allocID" type="tns:IDType"/>
+      <xsd:element name="operation" type="xsd:string"/>
+      <xsd:element name="target" type="xsd:string"
+        minOccurs="1" maxOccurs="unbounded"/>
+      <xsd:element name="parameter" type="tns:fedAttrType"
+        minOccurs="0" maxOccurs="unbounded"/>
+    </xsd:sequence>
+  </xsd:complexType>
+
+  <xsd:complexType name="operationSegmentResponseType">
+    <xsd:annotation>
+      <xsd:documentation>
+        A status of requested operations.
+      </xsd:documentation>
+    </xsd:annotation>
+    <xsd:sequence>
+      <xsd:element name="allocID" type="tns:IDType"/>
+      <xsd:element name="status" type="tns:operationStatusType"
+        minOccurs="1" maxOccurs="unbounded"/>
       <xsd:element name="proof" type="tns:proofType" />
     </xsd:sequence>