r5457 - in branches/tmckay/sage/python/sage: . aviary

tmckay at fedoraproject.org tmckay at fedoraproject.org
Wed Sep 5 20:00:20 UTC 2012


Author: tmckay
Date: 2012-09-05 20:00:19 +0000 (Wed, 05 Sep 2012)
New Revision: 5457

Modified:
   branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
   branches/tmckay/sage/python/sage/aviary/clients.py
   branches/tmckay/sage/python/sage/util.py
Log:
Add interfaces for Aviary getSubmissionID method
Add assert check on type of object returned to ObjectPool
Add a fake Scheduler object for using sage outside of Cumin


Modified: branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- branches/tmckay/sage/python/sage/aviary/aviaryoperations.py	2012-09-05 18:55:12 UTC (rev 5456)
+++ branches/tmckay/sage/python/sage/aviary/aviaryoperations.py	2012-09-05 20:00:19 UTC (rev 5457)
@@ -11,7 +11,7 @@
 from datetime import datetime
 from threading import Lock
 from suds import *
-from sage.util import CallSync, CallThread, ObjectPool, host_list
+from sage.util import CallSync, CallThread, host_list, parse_URL
 from aviarylocator import AviaryLocator
 from clients import ClientPool, TransportFactory
 
@@ -210,6 +210,24 @@
                             "settings in cumin.conf" % (self.nice, machine))
         return scheme, host
 
+class MockScheduler(object):
+    '''
+    Most methods in the _AviaryXXX objects were written to be compatible
+    with QMF methods which take a scheduler object.
+
+    This is a mock scheduler object containing the fields 
+    referenced by the methods: a scheduler name, a pool identifier, 
+    and a machine hostname/ip.
+
+    This object may be used when a scheduler object from Rosemary
+    is not available (ie, sage is being used outside of Cumin).  The
+    only required value is machine
+    '''
+    def __init__(self, machine, name = None, pool = None):
+        self.Name = name
+        self.Pool = pool
+        self.Machine = machine
+
 class _AviaryJobMethods(object):
 
     # Do this here rather than __init__ so we don't have to worry about
@@ -631,7 +649,63 @@
                        query_client, "getSubmissionSummary", subId)
         t.start()
 
+    def _get_submission_ids(self, host, page_size, mode, host_is_endpoint,
+                            name=None, qdate=None):
 
+        def my_process_results(result):
+            # Fix up the exception message if necessary
+            result = self._pretty_result(result, host)
+            if isinstance(result, Exception):
+                status = result
+                data = None
+            else:
+                status = 0
+                if not hasattr(result, "ids"):
+                    result.ids = []
+                data = result
+            return (status, data)
+
+        # Note, only one of name or qdate is expected to be
+        # set here.
+        query_client = self.query_client_pool.get_object()
+        subId = query_client.factory.create("ns0:SubmissionID")
+        subId.name =  name
+        subId.qdate = qdate
+
+        if host_is_endpoint:
+            servers = None
+        else:
+            servers = self.query_servers
+        self._setup_client(query_client, 
+                           servers,           # server lookup object
+                           host,              # host we want
+                           "getSubmissionID")
+
+        res = self._call_sync(my_process_results, 
+                              self.call_client_retry,
+                              query_client, 
+                              "getSubmissionID", 
+                              page_size, mode, subId)
+
+        self.query_client_pool.return_object(query_client)
+        return res;
+
+    def get_submission_ids_by_name(self, host, page_size, name,
+                                   host_is_endpoint = False):
+
+        return self._get_submission_ids(host, page_size, None, 
+                                        host_is_endpoint, name=name)
+
+    def get_submission_ids_by_qdate(self, host, page_size, qdate, mode,
+                                    host_is_endpoint = False):
+
+        if long(qdate) > sys.maxint:
+            raise Exception("Qdate is larger than max int")
+        if not mode in ("BEFORE", "AFTER"):
+            raise Exception("Mode must be 'BEFORE' or 'AFTER'")
+        return self._get_submission_ids(host, page_size, mode, 
+                                        host_is_endpoint, qdate=qdate)
+
 class _AviaryCommon(object):
     def __init__(self, name, locator, 
                  key="", cert="", root_cert="", domain_verify=True):
@@ -675,8 +749,17 @@
         return {"INTEGER": int, "FLOAT": float, "STRING": str, "BOOLEAN": bool}
 
     def _set_client_info(self, client, refresh=False):
-        scheme, host = client.server_list.find_server(client.service_name, 
-                                                      refresh)
+        # If there is no server_list, then client.server_name is expected to
+        # be an endpoint
+        if client.server_list:
+            scheme, host = client.server_list.find_server(client.server_name, 
+                                                          refresh)
+        else:
+            # We need the scheme value separated out for the transport
+            scheme = parse_URL(client.server_name).scheme
+            if scheme is None:
+                scheme = "http"
+            host = client.server_name
 
         # Have to set the URL for the method.  This might go away someday...
         client.set_options(location=host+client.method_name)
@@ -691,7 +774,7 @@
         # Look up the host and construct the URL.
         # Store information in the client so that retry is possible.
         client.server_list = server_list
-        client.service_name = name
+        client.server_name = name
         client.method_name = meth_name
 
         # This is initial setup before a call so we want to try a 
@@ -732,7 +815,7 @@
         return result
 
     def call_client_retry(self, client, meth_name, *meth_args, **meth_kwargs):
-        # If we fail with a urllib2.URLError (or anything similar) then try
+        # If we fail with a urllib2.URLError (or anything similar) then
         # attempt to get a new endpoint and try again.
         meth = getattr(client.service, meth_name)
         try:
@@ -742,7 +825,7 @@
             # (probably due to a restart on the condor side)
             # Let's get new endpoints, reset the client, 
             # and try again.
-            if client.server_list.should_retry:
+            if client.server_list and client.server_list.should_retry:
                 log.debug("AviaryOperations: received %s, retrying %s"\
                               % (str(e), client.options.location))
                 self._set_client_info(client, refresh=True)

Modified: branches/tmckay/sage/python/sage/aviary/clients.py
===================================================================
--- branches/tmckay/sage/python/sage/aviary/clients.py	2012-09-05 18:55:12 UTC (rev 5456)
+++ branches/tmckay/sage/python/sage/aviary/clients.py	2012-09-05 20:00:19 UTC (rev 5457)
@@ -122,7 +122,7 @@
 
 class ClientPool(ObjectPool):
     def __init__(self, wsdl, max_size):
-        super(ClientPool, self).__init__(max_size)
+        super(ClientPool, self).__init__(max_size, OverrideClient)
         self.wsdl = wsdl
 
     def create_object(self):

Modified: branches/tmckay/sage/python/sage/util.py
===================================================================
--- branches/tmckay/sage/python/sage/util.py	2012-09-05 18:55:12 UTC (rev 5456)
+++ branches/tmckay/sage/python/sage/util.py	2012-09-05 20:00:19 UTC (rev 5457)
@@ -178,8 +178,9 @@
     ObjectPool must be derived from, and create_object must be overridden.
     If max_size is set to None, the pool size is unlimited.
     '''
-    def __init__(self, max_size):
+    def __init__(self, max_size, obj_type):
         self.max_size = max_size
+        self.obj_type = obj_type
         self.pool = list()
         self.lock = Lock()
 
@@ -194,6 +195,7 @@
         return obj
 
     def return_object(self, obj):
+        assert isinstance(obj, self.obj_type)
         self.lock.acquire()
         if self.max_size is None or len(self.pool) < self.max_size:
             self.pool.append(obj)



More information about the cumin-developers mailing list