Author: tmckay
Date: 2012-04-17 21:22:17 +0000 (Tue, 17 Apr 2012)
New Revision: 5310
Added:
trunk/sage/python/sage/aviary/aviarylocator.py
trunk/sage/python/sage/aviary/clients.py
trunk/sage/rpc-defs/aviary/aviary-locator.wsdl
trunk/sage/rpc-defs/aviary/aviary-locator.xsd
Modified:
trunk/cumin/bin/cumin-web
trunk/cumin/python/cumin/config.py
trunk/cumin/python/cumin/main.py
trunk/sage/python/sage/aviary/aviaryoperations.py
Log:
Initial check in for aviary locator integration, some more clean up to follow.
BZ733516
Modified: trunk/cumin/bin/cumin-web
===================================================================
--- trunk/cumin/bin/cumin-web 2012-04-17 20:40:12 UTC (rev 5309)
+++ trunk/cumin/bin/cumin-web 2012-04-17 21:22:17 UTC (rev 5310)
@@ -18,8 +18,9 @@
sys.stdout = sys.__stdout__
def set_aviary_configs(cumin, values):
- cumin.aviary_job_servers = values.aviary_job_servers
- cumin.aviary_query_servers = values.aviary_query_servers
+# cumin.aviary_job_servers = values.aviary_job_servers+
+# cumin.aviary_query_servers = values.aviary_query_servers
+ cumin.aviary_locator = values.aviary_locator
cumin.aviary_key = values.aviary_key
cumin.aviary_cert = values.aviary_cert
cumin.aviary_root_cert = values.aviary_root_cert
Modified: trunk/cumin/python/cumin/config.py
===================================================================
--- trunk/cumin/python/cumin/config.py 2012-04-17 20:40:12 UTC (rev 5309)
+++ trunk/cumin/python/cumin/config.py 2012-04-17 21:22:17 UTC (rev 5310)
@@ -193,6 +193,9 @@
param = ConfigParameter(self, "aviary-query-servers", str)
param.default = ""
+ param = ConfigParameter(self, "aviary-locator", str)
+ param.default = "localhost"
+
param = ConfigParameter(self, "aviary-key", str)
param.default = ""
Modified: trunk/cumin/python/cumin/main.py
===================================================================
--- trunk/cumin/python/cumin/main.py 2012-04-17 20:40:12 UTC (rev 5309)
+++ trunk/cumin/python/cumin/main.py 2012-04-17 21:22:17 UTC (rev 5310)
@@ -92,12 +92,13 @@
# Aviary interface. If server values are "",
# Aviary operations for that server type will not be used.
- self.aviary_job_servers = ""
- self.aviary_query_servers = ""
+ #self.aviary_job_servers = ""
+ #self.aviary_query_servers = ""
self.aviary_key = ""
self.aviary_cert = ""
self.aviary_root_cert = ""
self.aviary_domain_verify=True
+ self.aviary_locator = ""
# For development use only
self.aviary_suds_logs = False
@@ -187,7 +188,7 @@
ops = [QmfOperations("qmf", self.session)]
imports_ok = True
- if self.aviary_job_servers or self.aviary_query_servers:
+ if self.aviary_locator:
try:
from sage.aviary.aviaryoperations import \
SudsLogging, AviaryOperationsFactory
@@ -198,21 +199,23 @@
aviary_dir = os.path.join(self.home, "rpc-defs/aviary")
# The factory will choose an impl that gives us jobs, queries, or both
- aviary_itf = AviaryOperationsFactory("aviary", aviary_dir,
- self.aviary_job_servers,
- self.aviary_query_servers,
- self.aviary_key, self.aviary_cert,
- self.aviary_root_cert,
- self.aviary_domain_verify)
+ # For the moment, we have True/True selecting both hardwired
+ aviary_itf = AviaryOperationsFactory("aviary", aviary_dir,
+ self.aviary_locator,
+ key=self.aviary_key,
+ cert=self.aviary_cert,
+ root_cert=self.aviary_root_cert,
+ domain_verify=self.aviary_domain_verify)
ops.insert(0, aviary_itf)
else:
log.info("Imports failed for Aviary interface, disabling")
+
log.info("%s Aviary interface for job submission and control." % \
- ((self.aviary_job_servers and imports_ok) and "Enabled" or
"Disabled"))
+ ((self.aviary_locator and imports_ok) and "Enabled" or
"Disabled"))
log.info("%s Aviary interface for query operations." % \
- ((self.aviary_query_servers and imports_ok) and "Enabled" or
"Disabled"))
+ ((self.aviary_locator and imports_ok) and "Enabled" or
"Disabled"))
self.remote.add_mechanisms(ops)
Added: trunk/sage/python/sage/aviary/aviarylocator.py
===================================================================
--- trunk/sage/python/sage/aviary/aviarylocator.py (rev 0)
+++ trunk/sage/python/sage/aviary/aviarylocator.py 2012-04-17 21:22:17 UTC (rev 5310)
@@ -0,0 +1,47 @@
+import logging
+import os
+
+from sage.util import parse_URL
+from clients import OverrideClient
+
+log = logging.getLogger("sage.aviary.locator")
+
+class AviaryLocator(object):
+ def __init__(self, datadir, locator_uri):
+ '''
+ Constructor.
+
+ datadir -- the directory containing a wsdl file for the locator
+
+ locator_uri -- the URI used to the Aviary locator.
+ There are sane defaults for the URI so all that must be supplied
+ is the hostname, although scheme, port, and path may also be supplied.
+ '''
+ self.locator_uri = self._get_uri(locator_uri)
+ self.datadir = datadir
+ self.wsdl = "file:" + os.path.join(self.datadir,
"aviary-locator.wsdl")
+
+ def _get_uri(self, locator):
+ uri = parse_URL(locator)
+ # Fill in scheme, port, path if they are blank
+ if uri.scheme is None:
+ uri.scheme = "http"
+ if uri.port is None:
+ uri.port = "9000"
+ if uri.path is None:
+ uri.path = "services/locator/locate"
+ return str(uri)
+
+ def get_endpoints(self, resource, sub_type):
+ '''
+ Return endpoints from Aviary for the resource and sub_type.
+ See documentation on the Aviary locator for information on
+ legal values for resource and sub_type.
+ '''
+ client = OverrideClient(self.wsdl, cache=None)
+ client.set_options(location=self.locator_uri)
+ res_id = client.factory.create("ns0:ResourceID")
+ res_id.resource = resource
+ res_id.sub_type = sub_type
+ res_id.name = None
+ return client.service.locate(res_id)
Modified: trunk/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- trunk/sage/python/sage/aviary/aviaryoperations.py 2012-04-17 20:40:12 UTC (rev 5309)
+++ trunk/sage/python/sage/aviary/aviaryoperations.py 2012-04-17 21:22:17 UTC (rev 5310)
@@ -9,11 +9,13 @@
import sage
from datetime import datetime
+from threading import Lock
from suds import *
-from suds.client import Client
from suds.transport.https import HttpAuthenticated
from sage.util import CallSync, CallThread, ObjectPool, host_list
from sage.https import *
+from aviarylocator import AviaryLocator
+from clients import ClientPool
try:
from sage.https_full import HTTPSFullCertTransport
has_full_cert = True
@@ -67,25 +69,101 @@
#the clients that we pool so that we can call set_options on the
#transport.
+class ServerList(object):
+ def __init__(self, locator, resource, subtype):
+
+ self._lock = Lock()
+ self.servers = None
+ self.locator = locator
+ self.resource = resource
+ self.subtype = subtype
+
+ # Nice, friendly strings for error messages on lookup
+ nice = {"JOB": "job service",
+ "QUERY_SERVER": "query service"}
+ try:
+ self.nice = nice[subtype]
+ except:
+ self.nice = subtype
+
+ def _get_host(self, name, servers):
+ scheme = ""
+ host = ""
+ if name in servers:
+ urls = servers[name]
+ if len(urls) > 0:
+ url = random.sample(urls, 1)[0]
+ scheme = url.scheme
+ host = str(url)
+ # A particular method name is going to be appended to path,
+ # so ensure the final "/" here.
+ if not host.endswith("/"):
+ host += "/"
+ return scheme, host
+
+ def _find_server(self, machine, refresh=False):
+ # If we already have a list of values then return that list unless
+ # refresh is True
+ self._lock.acquire()
+ try:
+ if self.servers is None or refresh:
+ try:
+ result = self.locator.get_endpoints(self.resource,
+ self.subtype)
+ except Exception, e:
+ result = _AviaryCommon._pretty_result(e,
+ self.locator.locator_uri)
+ raise result
+
+ urls = []
+ # If it's not okay, we just produce an empty sever list
+ if result.status.code == "OK":
+ for r in result.resources:
+ urls.extend(r.location)
+ urls = ",".join(urls)
+ if urls == "":
+ log.info("AviaryOperations: locator returned "\
+ "no endpoints for %s %s" % (self.resource,
+ self.subtype))
+
+ # Parse the urls and produce a dictionary of
+ # sage_URLs by machine
+ self.servers = host_list(urls)
+
+ scheme, host = self._get_host(machine, self.servers)
+ finally:
+ self._lock.release()
+ return scheme, host
+
+ def find_server(self, machine, refresh=False):
+ # Update the list if necessary and return host info for machine.
+ # If refresh is True, the server list will be updated. This is
+ # used when a previously cached host fails during a remote method
+ # invocation and we want to do a retry before giving up.
+ scheme, host = self._find_server(machine, refresh is True)
+ if host == "" and refresh == "on_no_host":
+ # Okay, if we didn't find a host and on_no_host has been set,
+ # try one more time with a forced refresh before we error.
+ # The case here is that the service on the particular host has
+ # not yet been discovered since Cumin startup or the last refresh.
+ # But, the service may appear at any point so it's possible that
+ # it's there now. Think "on demand polling".
+ scheme, host = self._find_server(machine, True)
+ if host == "":
+ raise Exception("Cannot locate %s on %s" % (self.nice, machine))
+ return scheme, host
+
class _AviaryJobMethods(object):
# Do this here rather than __init__ so we don't have to worry about
# matching parameter lists in multiple inheritance cases with super
- def init(self, job_servers, datadir):
+ def init(self, datadir):
- # Replace any occurrence of locahost with output of gethostname()
- # before parsing to match Machine fields of QMF objects later on.
- host = socket.gethostname()
- job_servers = string.replace(job_servers, "localhost", host)
-
- self.job_servers = host_list(job_servers,
- default_scheme = "http",
- default_port="9090",
- default_path="/services/job/")
-
+ self.job_servers = ServerList(self.locator, "SCHEDULER",
"JOB")
job_wsdl = "file:" + os.path.join(datadir,
"aviary-job.wsdl")
- self.job_client_pool = JobClientPool(job_wsdl, None)
+ self.job_client_pool = ClientPool(job_wsdl, None)
+
def set_job_attribute(self, scheduler, job_id, name, value, callback, submission):
assert callback
@@ -95,11 +173,12 @@
# massage results for use by standard callback
cb_args = self._cb_args_dataless(result)
callback(*cb_args)
-
- job_client = self._setup_client(self.job_client_pool,
- scheduler.Machine,
- self.job_servers,
- "setJobAttribute")
+
+ job_client = self.job_client_pool.get_object()
+ self._setup_client(job_client,
+ self.job_servers, # server lookup object
+ scheduler.Machine, # host we want
+ "setJobAttribute")
# Make a job id parameter (see job wsdl)
jobId = job_client.factory.create('ns0:JobID')
@@ -115,7 +194,8 @@
aviary_attr.type = "STRING"
aviary_attr.value = value
- t = CallThread(job_client.service.setJobAttribute, my_callback, jobId,
aviary_attr)
+ t = CallThread(self.call_client_retry, my_callback,
+ job_client, "setJobAttribute", jobId, aviary_attr)
t.start()
def submit_job(self, scheduler, ad, callback):
@@ -140,10 +220,11 @@
id = None
callback(status, id)
- job_client = self._setup_client(self.job_client_pool,
- scheduler.Machine,
- self.job_servers,
- "submitJob")
+ job_client = self.job_client_pool.get_object()
+ self._setup_client(job_client,
+ self.job_servers, # server lookup object
+ scheduler.Machine, # host we want
+ "submitJob")
# Set basic attributes in the order defined by aviary-job.wsdl.
args = list()
@@ -186,7 +267,8 @@
# suds will consider only a single item
args.append(extras)
- t = CallThread(job_client.service.submitJob, my_callback, *args)
+ t = CallThread(self.call_client_retry, my_callback,
+ job_client, "submitJob", *args)
t.start()
def control_job(self, cmd, scheduler, job_id, reason, submission, *args, **kwargs):
@@ -213,10 +295,11 @@
default = "default" in kwargs and kwargs["default"] or None
timeout = "timeout" in kwargs and kwargs["timeout"] or 5
- client = self._setup_client(self.job_client_pool,
- scheduler.Machine,
- self.job_servers,
- meth_name)
+ client = self.job_client_pool.get_object()
+ self._setup_client(client,
+ self.job_servers, # server lookup object
+ scheduler.Machine, # host we want
+ meth_name)
meth = getattr(client.service, meth_name)
@@ -236,7 +319,8 @@
cb_args = self._cb_args_dataless(result)
callback(*cb_args)
- t = CallThread(meth, my_callback, jobId, reason)
+ t = CallThread(self.call_client_retry, my_callback,
+ client, meth_name, jobId, reason)
t.start()
else:
def my_process_results(result):
@@ -244,7 +328,8 @@
result = self._pretty_result(result, scheduler.Machine)
return self._cb_args_dataless(result)
- res = self._call_sync(my_process_results, meth, jobId, reason)
+ res = self._call_sync(my_process_results, self.call_client_retry,
+ client, meth_name, jobId, reason)
self.job_client_pool.return_object(client)
return res;
@@ -252,20 +337,11 @@
# Do this here rather than __init__ so we don't have to worry about
# matching parameter lists in multiple inheritance cases with super
- def init(self, query_servers, datadir):
+ def init(self, datadir):
- # Replace any occurrence of locahost with output of gethostname()
- # before parsing to match Machine fields of QMF objects later on.
- host = socket.gethostname()
- query_servers = string.replace(query_servers, "localhost", host)
-
- self.query_servers = host_list(query_servers,
- default_scheme = "http",
- default_port="9091",
- default_path="/services/query/")
-
+ self.query_servers = ServerList(self.locator, "CUSTOM",
"QUERY_SERVER")
query_wsdl = "file:" + os.path.join(datadir,
"aviary-query.wsdl")
- self.query_client_pool = QueryClientPool(query_wsdl, None)
+ self.query_client_pool = ClientPool(query_wsdl, None)
def fetch_job_data(self, job_server, job_id, ftype, file, start, end,
scheduler_name, submission, *args, **kwargs):
@@ -298,10 +374,11 @@
data = None
return (status, data)
- client = self._setup_client(self.query_client_pool,
- job_server.Machine,
- self.query_servers,
- "getJobData")
+ client = self.query_client_pool.get_object()
+ self._setup_client(client,
+ self.query_servers, # server lookup object
+ job_server.Machine, # host we want
+ "getJobData")
# Make a job data parameter (see query wsdl)
jobData = client.factory.create('ns0:JobData')
@@ -326,8 +403,8 @@
from_end = start < 0
max_bytes = abs(end - start)
- res = self._call_sync(my_process_results, client.service.getJobData,
- jobData, max_bytes, from_end )
+ res = self._call_sync(my_process_results, self.call_client_retry,
+ client, "getJobData", jobData, max_bytes,
from_end)
self.query_client_pool.return_object(client)
return res;
@@ -376,10 +453,11 @@
data = None
return (status, data)
- client = self._setup_client(self.query_client_pool,
- job_server.Machine,
- self.query_servers,
- "getJobDetails")
+ client = self.query_client_pool.get_object()
+ self._setup_client(client,
+ self.query_servers, # server lookup object
+ job_server.Machine, # host we want
+ "getJobDetails")
# Make a job id parameter (see job wsdl)
jobId = client.factory.create('ns0:JobID')
@@ -389,8 +467,8 @@
jobId.submission.name = submission.Name
jobId.submission.owner = submission.Owner
- res = self._call_sync(my_process_results,
- client.service.getJobDetails, jobId)
+ res = self._call_sync(my_process_results, self.call_client_retry,
+ client, "getJobDetails", jobId)
self.query_client_pool.return_object(client)
return res;
@@ -452,10 +530,11 @@
data = {"Jobs": None}
callback(status, data)
- query_client = self._setup_client(self.query_client_pool,
- machine_name,
- self.query_servers,
- "getSubmissionSummary")
+ query_client = self.query_client_pool.get_object()
+ self._setup_client(query_client,
+ self.query_servers, # server lookup object
+ machine_name, # host we want
+ "getSubmissionSummary")
# What we really want here is the job summaries from the
# submission summary response. To get those, we have to
@@ -468,14 +547,22 @@
subId.name = submission.Name
subId.owner = submission.Owner
- t = CallThread(query_client.service.getSubmissionSummary,
- my_callback, subId)
+ t = CallThread(self.call_client_retry, my_callback,
+ query_client, "getSubmissionSummary", subId)
+
+ #t = CallThread(query_client.service.getSubmissionSummary,
+ # my_callback, subId)
t.start()
class _AviaryCommon(object):
- def __init__(self, name, key="", cert="", root_cert="",
domain_verify=True):
+ def __init__(self, name, locator,
+ key="", cert="", root_cert="",
domain_verify=True):
self.name = name
+
+ # Put this here to be referenced by AviaryOperations types
+ self.locator = locator
+
self.type_to_aviary = self._type_to_aviary()
self.aviary_to_type = self._aviary_to_type()
@@ -509,38 +596,13 @@
def _aviary_to_type(cls):
return {"INTEGER": int, "FLOAT": float, "STRING":
str, "BOOLEAN": bool}
- @classmethod
- def _get_host(cls, name, servers):
- scheme = ""
- host = ""
- if name in servers:
- urls = servers[name]
- if len(urls) > 0:
- url = random.sample(urls, 1)[0]
- scheme = url.scheme
- host = str(url)
- # A particular method name is going to be appended to path,
- # so ensure the file "/" here. Since we supply default path
- # values when the host list is parsed, we know the last portion
- # of the host string has to be a path.
- if not host.endswith("/"):
- host += "/"
- return scheme, host
+ def _set_client_info(self, client, refresh=False):
+ scheme, host = client.server_list.find_server(client.service_name,
+ refresh)
- @classmethod
- def _raise_no_host(cls, name):
- raise Exception("No aviary job servers specified for %s, check config
files" % name)
+ # Have to set the URL for the method. This might go away someday...
+ client.set_options(location=host+client.method_name)
- def _setup_client(self, client_pool, host_name, host_list, method):
- # Get a client object from the client_pool. Use host_name as
- # a lookup in the host_list to find an initial URL and append the
- # method name. Set up the client to point at that URL and return.
- client = client_pool.get_object()
- scheme, host = self._get_host(host_name, host_list)
- if host == "":
- self._raise_no_host(host_name)
- # Have to set the URL for the method. This might go away someday...
- client.set_options(location=host+method)
# Since we pool the clients and reuse them for different requests
# and since its possible to be using servers with different schemes,
# we have to always reset the transport here.
@@ -565,6 +627,17 @@
# this is the default transport when none is specified
the_transport = HttpAuthenticated()
client.set_options(transport=the_transport)
+
+ def _setup_client(self, client, server_list, name, meth_name):
+ # 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.method_name = meth_name
+
+ # This is initial setup before a call so we want to try a
+ # refresh on the server list if our service comes up missing
+ self._set_client_info(client, refresh="on_no_host")
return client
@classmethod
@@ -593,6 +666,24 @@
return Exception("Trouble reaching host %s, %s" % (host,
result.reason))
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
+ # attempt to get a new endpoint and try again.
+ meth = getattr(client.service, meth_name)
+ try:
+ result = meth(*meth_args, **meth_kwargs)
+ except Exception, e:
+ result = e
+
+ # If we get a URL error, our endpoint may have moved
+ # (probably due to a restart on the condor side)
+ # Let's try to get new endpoints, reset the client,
+ # and try again.
+ if isinstance(result, urllib2.URLError):
+ self._set_client_info(client, refresh=True)
+ result = meth(*meth_args, **meth_kwargs)
+ return result
+
def _call_sync(self, process_results, meth, *meth_args, **meth_kwargs):
# Common interface with QMF operations requires that a MethodResult
# object (or something just like it) be returned for synchronous calls.
@@ -600,7 +691,7 @@
# handles status semantics.
# meth here is a suds.client.Method which has a name attr....
- sync = CallSync(log, name=meth.method.name)
+ sync = CallSync(log, name=meth.__name__)
try:
result = meth(*meth_args, **meth_kwargs)
except Exception, e:
@@ -610,120 +701,49 @@
return sync
class AviaryOperations(_AviaryCommon, _AviaryJobMethods, _AviaryQueryMethods):
- def __init__(self, name, datadir, job_servers, query_servers,
+ def __init__(self, name, datadir, locator,
key="", cert="", root_cert="",
domain_verify=True):
- super(AviaryOperations, self).__init__(name, key, cert, root_cert,
+ super(AviaryOperations, self).__init__(name, locator,
+ key, cert, root_cert,
domain_verify)
- _AviaryJobMethods.init(self, job_servers, datadir)
- _AviaryQueryMethods.init(self, query_servers, datadir)
+ _AviaryJobMethods.init(self, datadir)
+ _AviaryQueryMethods.init(self, datadir)
class AviaryJobOperations(_AviaryCommon, _AviaryJobMethods):
- def __init__(self, name, datadir, job_servers,
+ def __init__(self, name, datadir, locator,
key="", cert="", root_cert="",
domain_verify=True):
- super(AviaryJobOperations, self).__init__(name, key, cert, root_cert,
+ super(AviaryJobOperations, self).__init__(name, locator,
+ key, cert, root_cert,
domain_verify)
- _AviaryJobMethods.init(self, job_servers, datadir)
+ _AviaryJobMethods.init(self, datadir)
class AviaryQueryOperations(_AviaryCommon, _AviaryQueryMethods):
- def __init__(self, name, datadir, query_servers,
+ def __init__(self, name, datadir, locator,
key="", cert="", root_cert="",
domain_verify=True):
- super(AviaryQueryOperations, self).__init__(name, key, cert, root_cert,
+ super(AviaryQueryOperations, self).__init__(name, locator,
+ key, cert, root_cert,
domain_verify)
- _AviaryQueryMethods.init(self, query_servers, datadir)
+ _AviaryQueryMethods.init(self, datadir)
-def AviaryOperationsFactory(name, datadir, job_servers=None, query_servers=None,
+def AviaryOperationsFactory(name, datadir, locator_uri,
+ job_server=True, query_server=True,
key="", cert="", root_cert="",
domain_verify=True):
+ locator = AviaryLocator(datadir, locator_uri)
+ if job_server and query_server:
+ res = AviaryOperations(name, datadir, locator,
+ key, cert, root_cert, domain_verify)
+ elif job_server:
+ res = AviaryJobOperations(name, datadir, locator,
+ key, cert, root_cert, domain_verify)
+ elif query_server:
+ res = AviaryQueryOperations(name, datadir, locator,
+ key, cert, root_cert, domain_verify)
+ return res
- if job_servers and query_servers:
- return AviaryOperations(name, datadir, job_servers, query_servers,
- key, cert, root_cert, domain_verify)
- elif job_servers:
- return AviaryJobOperations(name, datadir, job_servers,
- key, cert, root_cert, domain_verify)
- elif query_servers:
- return AviaryQueryOperations(name, datadir, query_servers,
- key, cert, root_cert, domain_verify)
- return None
-
-try:
- # Some of this stuff does not exist pre suds 0.4.1
- # Make it work anyway for testing on such hosts by
- # declaring simple OverridesPlugin in the exception case.
- from suds.plugin import MessagePlugin
- from suds.sax.attribute import Attribute
- class OverridesPlugin(MessagePlugin):
- '''
- Plugin which allows optional addition of attributes
- to the suds message after marshalling.
- '''
- def __init__(self):
- self.set_attributes = False
- self.attributes = dict()
-
- def marshalled(self, context):
- if self.set_attributes:
- sj_body = context.envelope.getChild('Body')[0]
- for k,v in self.attributes.iteritems():
- sj_body.attributes.append(Attribute(k, v))
-
-except:
- class OverridesPlugin(object):
- def __init__(self):
- self.set_attributes = False
-
-class OverrideClient(Client):
- '''
- Instantiate a Client object with an OverridesPlugin and
- retain a pointer to the plugin so that it can be controlled.
- '''
- def __init__(self, *args, **kwargs):
- self.override = OverridesPlugin()
- try:
- super(OverrideClient, self).__init__(plugins=[self.override], *args,
**kwargs)
- except:
- super(OverrideClient, self).__init__(*args, **kwargs)
-
- def set_enable_attributes(self, truth):
- '''
- Set the flag that controls whether extra attributes are marshalled.
-
- The attribute set is specified through the set_attributes() method.
- If 'truth' is True, then service calls made through this client
- will have the extra attributes appended to the attribute set of service
- request.
- '''
- self.override.set_attributes = truth
-
- def set_attributes(self, attrs):
- '''
- Specify extra attributes to append to the attribute set
- of service requests made through this client.
-
- The extra attributes will be added if set_enable_attributes()
- has been called with a value of True sometime prior to the
- time of a request.
- '''
- self.override.attributes = attrs
-
-class JobClientPool(ObjectPool):
- def __init__(self, job_wsdl, max_size):
- super(JobClientPool, self).__init__(max_size)
- self.job_wsdl = job_wsdl
-
- def create_object(self):
- return OverrideClient(self.job_wsdl, cache=None)
-
-class QueryClientPool(ObjectPool):
- def __init__(self, query_wsdl, max_size):
- super(QueryClientPool, self).__init__(max_size)
- self.query_wsdl = query_wsdl
-
- def create_object(self):
- return OverrideClient(self.query_wsdl, cache=None)
Added: trunk/sage/python/sage/aviary/clients.py
===================================================================
--- trunk/sage/python/sage/aviary/clients.py (rev 0)
+++ trunk/sage/python/sage/aviary/clients.py 2012-04-17 21:22:17 UTC (rev 5310)
@@ -0,0 +1,70 @@
+from suds.client import Client
+from sage.util import ObjectPool
+
+try:
+ # Some of this stuff does not exist pre suds 0.4.1
+ # Make it work anyway for testing on such hosts by
+ # declaring simple OverridesPlugin in the exception case.
+ from suds.plugin import MessagePlugin
+ from suds.sax.attribute import Attribute
+ class OverridesPlugin(MessagePlugin):
+ '''
+ Plugin which allows optional addition of attributes
+ to the suds message after marshalling.
+ '''
+ def __init__(self):
+ self.set_attributes = False
+ self.attributes = dict()
+
+ def marshalled(self, context):
+ if self.set_attributes:
+ sj_body = context.envelope.getChild('Body')[0]
+ for k,v in self.attributes.iteritems():
+ sj_body.attributes.append(Attribute(k, v))
+
+except:
+ class OverridesPlugin(object):
+ def __init__(self):
+ self.set_attributes = False
+
+class OverrideClient(Client):
+ '''
+ Instantiate a Client object with an OverridesPlugin and
+ retain a pointer to the plugin so that it can be controlled.
+ '''
+ def __init__(self, *args, **kwargs):
+ self.override = OverridesPlugin()
+ try:
+ super(OverrideClient, self).__init__(plugins=[self.override], *args,
**kwargs)
+ except:
+ super(OverrideClient, self).__init__(*args, **kwargs)
+
+ def set_enable_attributes(self, truth):
+ '''
+ Set the flag that controls whether extra attributes are marshalled.
+
+ The attribute set is specified through the set_attributes() method.
+ If 'truth' is True, then service calls made through this client
+ will have the extra attributes appended to the attribute set of service
+ request.
+ '''
+ self.override.set_attributes = truth
+
+ def set_attributes(self, attrs):
+ '''
+ Specify extra attributes to append to the attribute set
+ of service requests made through this client.
+
+ The extra attributes will be added if set_enable_attributes()
+ has been called with a value of True sometime prior to the
+ time of a request.
+ '''
+ self.override.attributes = attrs
+
+class ClientPool(ObjectPool):
+ def __init__(self, wsdl, max_size):
+ super(ClientPool, self).__init__(max_size)
+ self.wsdl = wsdl
+
+ def create_object(self):
+ return OverrideClient(self.wsdl, cache=None)
Added: trunk/sage/rpc-defs/aviary/aviary-locator.wsdl
===================================================================
--- trunk/sage/rpc-defs/aviary/aviary-locator.wsdl (rev 0)
+++ trunk/sage/rpc-defs/aviary/aviary-locator.wsdl 2012-04-17 21:22:17 UTC (rev 5310)
@@ -0,0 +1,54 @@
+<?xml version="1.0" encoding="UTF-8"
standalone="no"?>
+<!--
+/*
+ * Copyright 2000-2011 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *
http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+-->
+<wsdl:definitions
xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
xmlns:tns="http://grid.redhat.com/aviary-locator/"
xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:loc="http://locator.aviary.grid.redhat.com"
xmlns:ns="http://common.aviary.grid.redhat.com"
targetNamespace="http://grid.redhat.com/aviary-locator/"
name="aviary-locator">
+ <wsdl:types>
+ <xsd:schema
targetNamespace="http://grid.redhat.com/aviary-locator/">
+ <xsd:import schemaLocation="aviary-locator.xsd"
namespace="http://locator.aviary.grid.redhat.com"/>
+ </xsd:schema>
+ </wsdl:types>
+ <wsdl:message name="LocateRequest">
+ <wsdl:part name="parameters" element="loc:Locate"/>
+ </wsdl:message>
+ <wsdl:message name="LocateResponse">
+ <wsdl:part name="parameters" element="loc:LocateResponse"/>
+ </wsdl:message>
+ <wsdl:portType name="AviaryLocator">
+ <wsdl:operation name="locate">
+ <wsdl:input message="tns:LocateRequest"/>
+ <wsdl:output message="tns:LocateResponse"/>
+ </wsdl:operation>
+ </wsdl:portType>
+ <wsdl:binding name="AviaryLocatorSOAPBinding"
type="tns:AviaryLocator">
+ <soap:binding style="document"
transport="http://schemas.xmlsoap.org/soap/http"/>
+ <wsdl:operation name="locate">
+ <soap:operation
soapAction="http://grid.redhat.com/aviary-locator/locate"/>
+ <wsdl:input>
+ <soap:body use="literal"/>
+ </wsdl:input>
+ <wsdl:output>
+ <soap:body use="literal"/>
+ </wsdl:output>
+ </wsdl:operation>
+ </wsdl:binding>
+ <wsdl:service name="AviaryLocatorService">
+ <wsdl:port name="AviaryLocatorSOAPPort"
binding="tns:AviaryLocatorSOAPBinding">
+ <soap:address location="http://localhost"/>
+ </wsdl:port>
+ </wsdl:service>
+</wsdl:definitions>
Added: trunk/sage/rpc-defs/aviary/aviary-locator.xsd
===================================================================
--- trunk/sage/rpc-defs/aviary/aviary-locator.xsd (rev 0)
+++ trunk/sage/rpc-defs/aviary/aviary-locator.xsd 2012-04-17 21:22:17 UTC (rev 5310)
@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+/*
+ * Copyright 2000-2011 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *
http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+-->
+<xs:schema
xmlns:xs="http://www.w3.org/2001/XMLSchema"
xmlns:loc="http://locator.aviary.grid.redhat.com"
xmlns:mrg="http://common.aviary.grid.redhat.com"
targetNamespace="http://locator.aviary.grid.redhat.com">
+ <xs:import namespace="http://common.aviary.grid.redhat.com"
schemaLocation="aviary-common.xsd"/>
+ <!-- declare message-level elements using anonymous complex types for simpler
generated types -->
+ <!-- use of refs, inheritance into anonymous types, etc. eschewed for simplicity and
interop -->
+ <xs:element name="Locate">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element name="id" type="mrg:ResourceID"/>
+ </xs:sequence>
+ <xs:attribute name="partialMatches" type="xs:boolean"
default="true"/>
+ </xs:complexType>
+ </xs:element>
+ <xs:element name="LocateResponse">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element name="resources" type="mrg:ResourceLocation"
minOccurs="0" maxOccurs="unbounded"/>
+ <xs:element name="status" type="mrg:Status"/>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+</xs:schema>