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