r5310 - in trunk: cumin/bin cumin/python/cumin sage/python/sage/aviary sage/rpc-defs/aviary
by tmckay@fedoraproject.org
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>
12 years
r5309 - branches/play_plumage/mint/python/mint/plumage
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-17 20:40:12 +0000 (Tue, 17 Apr 2012)
New Revision: 5309
Modified:
branches/play_plumage/mint/python/mint/plumage/session.py
Log:
Changing the current PlumageSession thread to wake up every 5 min instead of every minute.
Modified: branches/play_plumage/mint/python/mint/plumage/session.py
===================================================================
--- branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 18:59:24 UTC (rev 5308)
+++ branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 20:40:12 UTC (rev 5309)
@@ -120,8 +120,8 @@
while True:
if self.stop_requested:
break
- sleep(60)
- itemSet = self.collection.find({"ts": {'$gt': datetime.now() - timedelta(minutes=1) + UTC_DIFF}}).sort("ts", -1)
+ sleep(300)
+ itemSet = self.collection.find({"ts": {'$gt': datetime.now() - timedelta(seconds=300) + UTC_DIFF}}).sort("ts", -1)
for item in itemSet:
record = OSUtil()
@@ -133,4 +133,4 @@
obj = ObjectUpdate(self.app.model, record, self.cls)
self.app.update_thread.enqueue(obj)
- log.info("PlumageSessionThread--current: pass completed")
+ log.info("PlumageSessionThread--current: pass completed %s records added" % itemSet.count())
12 years
r5308 - branches/play_plumage/mint/python/mint/plumage
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-17 18:59:24 +0000 (Tue, 17 Apr 2012)
New Revision: 5308
Modified:
branches/play_plumage/mint/python/mint/plumage/session.py
Log:
First pass at changes to have cumin-report contain 2 threads. 1 thread starts at "now" and moves back through time. The other thread will wake up every minute and load all records that are new in the past minute.
Modified: branches/play_plumage/mint/python/mint/plumage/session.py
===================================================================
--- branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 17:44:16 UTC (rev 5307)
+++ branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 18:59:24 UTC (rev 5308)
@@ -13,8 +13,11 @@
log = logging.getLogger("mint.plumage.session")
-UTC_DIFF = datetime.utcnow() - datetime.now()
+UTC_DIFF = datetime.utcnow() - datetime.now() # needed because everything in mongo is UTC
+class OSUtil(object):
+ pass
+
class PlumageSession(object):
def __init__(self, app, server_host, server_port):
self.app = app
@@ -42,6 +45,11 @@
self.server_host,
self.server_port,
cls))
+ self.threads.append(CurrentPlumageSessionThread(
+ self.app,
+ self.server_host,
+ self.server_port,
+ cls))
# Update our list, minus the black_list
self.app.classes.difference_update(black_list)
@@ -58,6 +66,11 @@
self.server_host,
self.server_port,
cls))
+ self.threads.append(CurrentPlumageSessionThread(
+ self.app,
+ self.server_host,
+ self.server_port,
+ cls))
def start(self):
log.info("Starting %s", self)
for t in self.threads:
@@ -86,9 +99,6 @@
# probably specified to us from a config option (with a corresponding
# query specification in the xml)
- class OSUtil(object):
- pass
-
itemSet = self.collection.find().sort("ts", -1)
for item in itemSet:
@@ -98,92 +108,29 @@
#required since MongoDB uses UTC for everything, here we get our sanity back
setattr(record, name, value - UTC_DIFF)
else:
- setattr(record, name, item[name])
+ setattr(record, name, value)
obj = ObjectUpdate(self.app.model, record, self.cls)
self.app.update_thread.enqueue(obj)
-
-# If you'd rather fake the data, use the following instead
-# while True:
-# if self.stop_requested:
-# break
-# # We need to get some data, and we need to turn it into some kind
-# # of update object.
-# os = OSUtil()
-#
-# os.ts = time.time() * 1000000000
-#
-# machines = ["croberts.redhat.com", "tmckay.redhat.com", "shadowman.redhat.com"]
-# os.m = machines[randrange(0,3)]
-# os.n = "%d@%s" % (randrange(1, 16), os.m)
-#
-# architectures = ["X86_64", "X86", "Itanium"]
-# os.ar = architectures[randrange(0,3)]
-#
-# oses = ["LINUX", "WINDOWS", "MAC"]
-# os.os = oses[randrange(0,3)]
-#
-# os.ki = randrange(0, 100)
-#
-# os.la = randrange(0, 100)
-#
-# states = ["Claimed", "Unclaimed", "Owner"]
-# os.st = states[randrange(0,3)]
-#
-# os.cpu = randrange(0,65)
-#
-# os.mem = randrange(0, 100000)
-#
-# obj = ObjectUpdate(self.app.model, os, self.cls)
-# self.app.update_thread.enqueue(obj)
-
-'''
-class MintConsole(Console):
- def __init__(self, model):
- self.model = model
-
- def brokerConnected(self, qmf_broker):
- message = "Broker %s:%i is connected"
- self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
-
- def brokerInfo(self, qmf_broker):
- message = "Broker info from %s:%i"
- self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
-
- def brokerDisconnected(self, qmf_broker):
- message = "Broker %s:%i is disconnected"
- self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
-
- def newAgent(self, qmf_agent):
- self.model.print_event(3, "Creating %s", qmf_agent)
-
- up = AgentUpdate(self.model, qmf_agent)
- self.model.app.update_thread.enqueue(up)
-
- def delAgent(self, qmf_agent):
- self.model.print_event(3, "Deleting %s", qmf_agent)
-
- up = AgentDelete(self.model, qmf_agent)
- self.model.app.update_thread.enqueue(up)
-
- def heartbeat(self, qmf_agent, timestamp):
- message = "Heartbeat from %s at %s"
- self.model.print_event(5, message, qmf_agent, timestamp)
-
- up = AgentUpdate(self.model, qmf_agent)
- self.model.app.update_thread.enqueue(up)
-
- def newPackage(self, name):
- self.model.print_event(2, "New package %s", name)
-
- def newClass(self, kind, classKey):
- self.model.print_event(2, "New class %s", classKey)
-
- def objectProps(self, broker, qmf_object):
- up = ObjectUpdate(self.model, qmf_object)
- self.model.app.update_thread.enqueue(up)
-
- def objectStats(self, broker, qmf_object):
- up = ObjectUpdate(self.model, qmf_object)
- self.model.app.update_thread.enqueue(up)
-'''
+
+ log.info("PlumageSessionThread--history: run completed")
+
+class CurrentPlumageSessionThread(PlumageSessionThread):
+ def run(self):
+ while True:
+ if self.stop_requested:
+ break
+ sleep(60)
+ itemSet = self.collection.find({"ts": {'$gt': datetime.now() - timedelta(minutes=1) + UTC_DIFF}}).sort("ts", -1)
+
+ for item in itemSet:
+ record = OSUtil()
+ for name, value in item.iteritems():
+ if isinstance(value, datetime):
+ setattr(record, name, value - UTC_DIFF)
+ else:
+ setattr(record, name, value)
+
+ obj = ObjectUpdate(self.app.model, record, self.cls)
+ self.app.update_thread.enqueue(obj)
+ log.info("PlumageSessionThread--current: pass completed")
12 years
r5307 - in branches/play_plumage: cumin/bin mint/python/mint/plumage
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-17 17:44:16 +0000 (Tue, 17 Apr 2012)
New Revision: 5307
Modified:
branches/play_plumage/cumin/bin/cumin-report
branches/play_plumage/mint/python/mint/plumage/main.py
branches/play_plumage/mint/python/mint/plumage/session.py
Log:
Small refactoring of cumin-report gathering of the plumage server information.
Modified: branches/play_plumage/cumin/bin/cumin-report
===================================================================
--- branches/play_plumage/cumin/bin/cumin-report 2012-04-17 16:14:47 UTC (rev 5306)
+++ branches/play_plumage/cumin/bin/cumin-report 2012-04-17 17:44:16 UTC (rev 5307)
@@ -142,11 +142,8 @@
model_dir = [os.path.join(config.home, x) for x in ("model/admin", "model/plumage")]
- broker_uris = []
-# broker_uris = [x.strip() for x in opts.brokers.split(",")]
+ mint = Plumage(model_dir, values.plumage_host, values.plumage_port, opts.database)
- mint = Plumage(model_dir, broker_uris, opts.database)
-
mint.print_event_level = opts.print_events
mint.expire_enabled = opts.expire_enabled
Modified: branches/play_plumage/mint/python/mint/plumage/main.py
===================================================================
--- branches/play_plumage/mint/python/mint/plumage/main.py 2012-04-17 16:14:47 UTC (rev 5306)
+++ branches/play_plumage/mint/python/mint/plumage/main.py 2012-04-17 17:44:16 UTC (rev 5307)
@@ -12,7 +12,7 @@
log = logging.getLogger("mint.plumage")
class Plumage(object):
- def __init__(self, model_dir, server_url, database_dsn):
+ def __init__(self, model_dir, server_host, server_port, database_dsn):
# Can we just use a different model dir here?
# That would control visible packages, vacuuming, etc.
@@ -22,7 +22,7 @@
self.database = MintDatabase(self, database_dsn)
self.admin = CuminAdmin(self)
self.update_thread = UpdateThread(self)
- self.session = PlumageSession(self, server_url)
+ self.session = PlumageSession(self, server_host, server_port)
self.expire_enabled = True
self.expire_thread = ExpireThread(self)
Modified: branches/play_plumage/mint/python/mint/plumage/session.py
===================================================================
--- branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 16:14:47 UTC (rev 5306)
+++ branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 17:44:16 UTC (rev 5307)
@@ -16,9 +16,10 @@
UTC_DIFF = datetime.utcnow() - datetime.now()
class PlumageSession(object):
- def __init__(self, app, server_url):
+ def __init__(self, app, server_host, server_port):
self.app = app
- self.server_url = server_url
+ self.server_host = server_host
+ self.server_port = server_port
self.threads = []
def check(self):
@@ -38,7 +39,8 @@
# We're going to have a thread for this...
self.threads.append(PlumageSessionThread(
self.app,
- self.server_url,
+ self.server_host,
+ self.server_port,
cls))
# Update our list, minus the black_list
@@ -53,7 +55,8 @@
for cls in pkg._classes:
self.threads.append(PlumageSessionThread(
self.app,
- self.server_url,
+ self.server_host,
+ self.server_port,
cls))
def start(self):
log.info("Starting %s", self)
@@ -66,15 +69,14 @@
t.stop()
def __repr__(self):
- return "%s(%s)" % (self.__class__.__name__, self.server_url)
+ return "%s(%s:%s)" % (self.__class__.__name__, self.server_host, self.server_port)
class PlumageSessionThread(MintDaemonThread):
- def __init__(self, app, server_url, cls):
+ def __init__(self, app, server_host, server_port, cls):
super(PlumageSessionThread, self).__init__(app)
- self.server_url = server_url
self.cls = cls
- self.connection = Connection(app.plumage_host, app.plumage_port) #Needs to be from config
+ self.connection = Connection(server_host, server_port) #Needs to be from config
self.database = Database(self.connection, self.cls.database)
self.collection = Collection(self.database, self.cls.collection)
12 years
r5306 - branches/play_plumage/cumin/bin
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-17 16:14:47 +0000 (Tue, 17 Apr 2012)
New Revision: 5306
Modified:
branches/play_plumage/cumin/bin/cumin-report
Log:
Update cumin-report to expect the new config values.
Modified: branches/play_plumage/cumin/bin/cumin-report
===================================================================
--- branches/play_plumage/cumin/bin/cumin-report 2012-04-17 16:04:46 UTC (rev 5305)
+++ branches/play_plumage/cumin/bin/cumin-report 2012-04-17 16:14:47 UTC (rev 5306)
@@ -155,6 +155,9 @@
mint.vacuum_enabled = opts.vacuum_enabled
mint.vacuum_thread.interval = values.vacuum_interval
+
+ mint.plumage_host = values.plumage_host
+ mint.plumage_port = values.plumage_port
# Data model init has been moved here in the
# bootstrap process so we can process the binding options next...
12 years
r5305 - in branches/play_plumage: cumin/etc cumin/model/plumage cumin/python/cumin mint/python/mint/plumage rosemary/python/rosemary
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-17 16:04:46 +0000 (Tue, 17 Apr 2012)
New Revision: 5305
Modified:
branches/play_plumage/cumin/etc/cumin.conf
branches/play_plumage/cumin/model/plumage/plumage.xml
branches/play_plumage/cumin/model/plumage/rosemary.xml
branches/play_plumage/cumin/python/cumin/config.py
branches/play_plumage/cumin/python/cumin/model.py
branches/play_plumage/mint/python/mint/plumage/main.py
branches/play_plumage/mint/python/mint/plumage/session.py
branches/play_plumage/rosemary/python/rosemary/model.py
Log:
Updating to pull actual data from ODS defined by config entry plumage_host and plumage_port (default is localhost:27017). The database name and collection are defined in plumage/rosemary.xml. For this update, plumage.xml has also been changed to use the same field names as the plumage database.
Modified: branches/play_plumage/cumin/etc/cumin.conf
===================================================================
--- branches/play_plumage/cumin/etc/cumin.conf 2012-04-16 19:47:45 UTC (rev 5304)
+++ branches/play_plumage/cumin/etc/cumin.conf 2012-04-17 16:04:46 UTC (rev 5305)
@@ -189,6 +189,8 @@
expire-threshold: 90d
# vacuum-enabled: True
vacuum-interval: 1h
+# plumage_host: localhost
+# plumage_port: 27017
# *************** Some parameter explanations by section ***************
Modified: branches/play_plumage/cumin/model/plumage/plumage.xml
===================================================================
--- branches/play_plumage/cumin/model/plumage/plumage.xml 2012-04-16 19:47:45 UTC (rev 5304)
+++ branches/play_plumage/cumin/model/plumage/plumage.xml 2012-04-17 16:04:46 UTC (rev 5305)
@@ -1,24 +1,15 @@
<?xml version="1.0" encoding="utf-8"?>
<schema package="com.redhat.grid.plumage" package_type="generic">
<class name="OSUtil">
- <property name="uuid" unique="y"
- type="uuid" access="RC" desc="UUID of System Image"/>
-
- <property name="osName"
- type="sstr" access="RO" desc="Operating System Name" />
-
- <statistic name="update_time" type="absTime" unit="uint32" />
- <statistic name="machine" type="sstr" optional="y" default=""/>
- <statistic name="slot" type="sstr" optional="y" default="" />
- <statistic name="architecture" type="sstr" optional="y" default=""/>
+ <statistic name="ts" type="absTime" unit="uint32" />
+ <statistic name="m" type="sstr" optional="y" default=""/>
+ <statistic name="n" type="sstr" optional="y" default="" />
+ <statistic name="ar" type="sstr" optional="y" default=""/>
<statistic name="os" type="sstr" optional="y" default=""/>
- <statistic name="kbidle" type="uint32" uint="uint32" optional="y" default="0"/>
- <statistic name="lastactive" type="uint32" uint="uint32" optional="y" default="0"/>
- <statistic name="state" type="sstr" optional="y" default=""/>
+ <statistic name="ki" type="uint32" uint="uint32" optional="y" default="0"/>
+ <statistic name="la" type="uint32" uint="uint32" optional="y" default="0"/>
+ <statistic name="st" type="sstr" optional="y" default=""/>
<statistic name="cpu" type="uint32" uint="uint32" optional="y" default="0" />
<statistic name="mem" type="uint32" uint="uint32" optional="y" default="0"/>
- <statistic name="avail" type="uint32" unit="uint32" optional="y" default="0"/>
- <statistic name="used" type="uint32" unit="uint32" optional="y" default="0"/>
-
</class>
</schema>
Modified: branches/play_plumage/cumin/model/plumage/rosemary.xml
===================================================================
--- branches/play_plumage/cumin/model/plumage/rosemary.xml 2012-04-16 19:47:45 UTC (rev 5304)
+++ branches/play_plumage/cumin/model/plumage/rosemary.xml 2012-04-17 16:04:46 UTC (rev 5305)
@@ -1,28 +1,23 @@
<model>
<package name="com.redhat.grid.plumage" persistent="time">
<class name="OSUtil">
- <source>some stuff goes here about a query</source>
+ <source>
+ <database>condor_stats</database>
+ <collection>samples.machine</collection>
+ </source>
- <statistic name="update_time" timestamp="y">
+ <statistic name="ts" timestamp="y">
</statistic>
- <statistic name="used">
- <title>Used</title>
- </statistic>
-
- <statistic name="avail">
- <title>Available</title>
- </statistic>
-
- <statistic name="machine">
+ <statistic name="m">
<title>Machine</title>
</statistic>
- <statistic name="slot">
+ <statistic name="n">
<title>Slot</title>
</statistic>
- <statistic name="architecture">
+ <statistic name="ar">
<title>Architecture</title>
</statistic>
@@ -30,15 +25,15 @@
<title>OS</title>
</statistic>
- <statistic name="kbidle">
+ <statistic name="ki">
<title>KB idle</title>
</statistic>
- <statistic name="lastactive">
+ <statistic name="la">
<title>Last active</title>
</statistic>
- <statistic name="state">
+ <statistic name="st">
<title>State</title>
</statistic>
Modified: branches/play_plumage/cumin/python/cumin/config.py
===================================================================
--- branches/play_plumage/cumin/python/cumin/config.py 2012-04-16 19:47:45 UTC (rev 5304)
+++ branches/play_plumage/cumin/python/cumin/config.py 2012-04-17 16:04:46 UTC (rev 5305)
@@ -83,7 +83,7 @@
def __init__(self, section_name="report", strict_section=False):
super(CuminReportConfig, self).__init__()
- data = CuminDataConfigSection(self, section_name, strict_section)
+ data = CuminReportConfigSection(self, section_name, strict_section)
_common_data_section(self, data, section_name)
class CuminMasterConfig(CuminConfig):
@@ -224,8 +224,11 @@
super(CuminReportConfigSection, self).__init__(config, name,
strict_section)
- param = ConfigParameter(self, "plumage_server", str)
+ param = ConfigParameter(self, "plumage_host", str)
param.default = "localhost"
+
+ param = ConfigParameter(self, "plumage_port", int)
+ param.default = 27017
class CuminOptionParser(OptionParser,object):
def __init__(self):
Modified: branches/play_plumage/cumin/python/cumin/model.py
===================================================================
--- branches/play_plumage/cumin/python/cumin/model.py 2012-04-16 19:47:45 UTC (rev 5304)
+++ branches/play_plumage/cumin/python/cumin/model.py 2012-04-17 16:04:46 UTC (rev 5305)
@@ -423,10 +423,10 @@
class AggregateSamplesSqlAdapter(SamplesSqlAdapter):
def __init__(self, app, cls, sig):
super(AggregateSamplesSqlAdapter, self).__init__(app, cls, sig)
- self.querymap = {"used":"sum(case when state = 'Claimed' then 1 else 0 end) as value",
- "avail":"sum(case when state = 'Claimed' or state = 'Available' then 1 else 0 end) as value",
- "total":"sum(case when state = 'Claimed' or state = 'Available' or state = 'Unavailable' then 1 else 0 end) as value",
- "unavail":"sum(case when state = 'Unavailable' then 1 else 0 end) as value"}
+ self.querymap = {"used":"sum(case when st = 'Claimed' then 1 else 0 end) as value",
+ "avail":"sum(case when st = 'Claimed' or st = 'Unclaimed' then 1 else 0 end) as value",
+ "total":"sum(case when st = 'Claimed' or st = 'Unclaimed' or st = 'Owner' then 1 else 0 end) as value",
+ "unavail":"sum(case when st = 'Owner' then 1 else 0 end) as value"}
self.max_samples = 60
Modified: branches/play_plumage/mint/python/mint/plumage/main.py
===================================================================
--- branches/play_plumage/mint/python/mint/plumage/main.py 2012-04-16 19:47:45 UTC (rev 5304)
+++ branches/play_plumage/mint/python/mint/plumage/main.py 2012-04-17 16:04:46 UTC (rev 5305)
@@ -29,6 +29,9 @@
self.vacuum_enabled = True
self.vacuum_thread = VacuumThread(self)
+
+ self.plumage_host = "localhost"
+ self.plumage_port = 27017
self.print_event_level = 0
Modified: branches/play_plumage/mint/python/mint/plumage/session.py
===================================================================
--- branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-16 19:47:45 UTC (rev 5304)
+++ branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 16:04:46 UTC (rev 5305)
@@ -5,9 +5,16 @@
from time import sleep
from update import ObjectUpdate
from random import randrange
+from datetime import timedelta, datetime
+from pymongo import Connection
+from pymongo.collection import Collection
+from pymongo.database import Database
+
log = logging.getLogger("mint.plumage.session")
+UTC_DIFF = datetime.utcnow() - datetime.now()
+
class PlumageSession(object):
def __init__(self, app, server_url):
self.app = app
@@ -66,66 +73,68 @@
super(PlumageSessionThread, self).__init__(app)
self.server_url = server_url
self.cls = cls
-
+
+ self.connection = Connection(app.plumage_host, app.plumage_port) #Needs to be from config
+ self.database = Database(self.connection, self.cls.database)
+ self.collection = Collection(self.database, self.cls.collection)
+
def run(self):
- uuid = 1
- while True:
- if self.stop_requested:
- break
- # We create objects here. Tag them with the right class,
- # probably specified to us from a config option (with a corresponding
- # query specification in the xml)
- #cls = self.app.model.com_redhat_grid_plumage.OSUtil
+ # We create objects here. Tag them with the right class,
+ # probably specified to us from a config option (with a corresponding
+ # query specification in the xml)
- class OSUtil(object):
- pass
-
- # We need to get some data, and we need to turn it into some kind
- # of update object.
- os = OSUtil()
-
- os.uuid = str(uuid)
- os.osName = "Linux"
- os.update_time = time.time() * 1000000000
-
- # random, 16 slots
- os.avail = randrange(0, 17)
-
- # some number of available used
- os.used = randrange(0, os.avail+1)
-
- machines = ["croberts.redhat.com", "tmckay.redhat.com", "shadowman.redhat.com"]
- os.machine = machines[randrange(0,3)]
- os.slot = "%d@%s" % (randrange(1, 16), os.machine)
-
- architectures = ["X86_64", "X86", "Itanium"]
- os.architecture = architectures[randrange(0,3)]
-
- oses = ["LINUX", "WINDOWS", "MAC"]
- os.os = oses[randrange(0,3)]
-
- os.kbidle = randrange(0, 100)
-
- os.lastactive = randrange(0, 100)
-
- states = ["Claimed", "Available", "Unavailable"]
- os.state = states[randrange(0,3)]
+ class OSUtil(object):
+ pass
+
+ itemSet = self.collection.find().sort("ts", -1)
+
+ for item in itemSet:
+ record = OSUtil()
+ for name, value in item.iteritems():
+ if isinstance(value, datetime):
+ #required since MongoDB uses UTC for everything, here we get our sanity back
+ setattr(record, name, value - UTC_DIFF)
+ else:
+ setattr(record, name, item[name])
- os.cpu = randrange(0,65)
-
- os.mem = randrange(0, 100000)
-
- obj = ObjectUpdate(self.app.model, os, self.cls)
+ obj = ObjectUpdate(self.app.model, record, self.cls)
self.app.update_thread.enqueue(obj)
- sleep(30)
-
- # five machines
- uuid = (uuid + 1) % 6
- if uuid == 0:
- uuid = 1
-
+# If you'd rather fake the data, use the following instead
+# while True:
+# if self.stop_requested:
+# break
+# # We need to get some data, and we need to turn it into some kind
+# # of update object.
+# os = OSUtil()
+#
+# os.ts = time.time() * 1000000000
+#
+# machines = ["croberts.redhat.com", "tmckay.redhat.com", "shadowman.redhat.com"]
+# os.m = machines[randrange(0,3)]
+# os.n = "%d@%s" % (randrange(1, 16), os.m)
+#
+# architectures = ["X86_64", "X86", "Itanium"]
+# os.ar = architectures[randrange(0,3)]
+#
+# oses = ["LINUX", "WINDOWS", "MAC"]
+# os.os = oses[randrange(0,3)]
+#
+# os.ki = randrange(0, 100)
+#
+# os.la = randrange(0, 100)
+#
+# states = ["Claimed", "Unclaimed", "Owner"]
+# os.st = states[randrange(0,3)]
+#
+# os.cpu = randrange(0,65)
+#
+# os.mem = randrange(0, 100000)
+#
+# obj = ObjectUpdate(self.app.model, os, self.cls)
+# self.app.update_thread.enqueue(obj)
+
'''
class MintConsole(Console):
def __init__(self, model):
Modified: branches/play_plumage/rosemary/python/rosemary/model.py
===================================================================
--- branches/play_plumage/rosemary/python/rosemary/model.py 2012-04-16 19:47:45 UTC (rev 5304)
+++ branches/play_plumage/rosemary/python/rosemary/model.py 2012-04-17 16:04:46 UTC (rev 5305)
@@ -757,6 +757,8 @@
# sample table
self.add_unique_to_samples = True
self.source = ""
+ self.database = ""
+ self.collection = ""
def init(self):
super(Generic_class, self).init()
@@ -768,7 +770,10 @@
# Todo: this could be optimized if we use the shortest
# set of unique column groups but probably there is only one
- self.signature_cols = self._unique_col_groups[self.col_group_list[0]]
+ if self._unique_col_groups is not None and len(self._unique_col_groups) > 0:
+ self.signature_cols = self._unique_col_groups[self.col_group_list[0]]
+ else:
+ self.signature_cols = []
# Set up a query that we can use on a unique
# column set
@@ -780,6 +785,8 @@
super(Generic_class, self).extend(elem)
self.source = elem.findtext("source")
+ self.database = elem.findtext("source/database")
+ self.collection = elem.findtext("source/collection")
def load_class_components(self, elem):
for child in elem.findall("property"):
@@ -841,6 +848,7 @@
self.optional = None
self.unique = None
self.formatter = None
+ self.storage = None
self.title = None
self.description = None
12 years
r5304 - branches/play_plumage/mint/python/mint/plumage
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-16 19:47:45 +0000 (Mon, 16 Apr 2012)
New Revision: 5304
Modified:
branches/play_plumage/mint/python/mint/plumage/update.py
Log:
transform_timestamp now checks to see if it was given a datetime object before blindly converting.
Modified: branches/play_plumage/mint/python/mint/plumage/update.py
===================================================================
--- branches/play_plumage/mint/python/mint/plumage/update.py 2012-04-13 20:06:19 UTC (rev 5303)
+++ branches/play_plumage/mint/python/mint/plumage/update.py 2012-04-16 19:47:45 UTC (rev 5304)
@@ -334,7 +334,10 @@
def transform_timestamp(value):
if value != 0:
- return datetime.fromtimestamp(value / 1000000000)
+ if isinstance(value,datetime):
+ return value
+ else:
+ return datetime.fromtimestamp(value / 1000000000)
def transform_pickle(value):
return pickle.dumps(x)
12 years
r5303 - branches/play_plumage/cumin/resources
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-13 20:06:19 +0000 (Fri, 13 Apr 2012)
New Revision: 5303
Modified:
branches/play_plumage/cumin/resources/app.js
Log:
Tweaking the colors used in rendering the charts.
Modified: branches/play_plumage/cumin/resources/app.js
===================================================================
--- branches/play_plumage/cumin/resources/app.js 2012-04-13 17:22:53 UTC (rev 5302)
+++ branches/play_plumage/cumin/resources/app.js 2012-04-13 20:06:19 UTC (rev 5303)
@@ -629,7 +629,7 @@
}
},
series: getSeries(dataContainer),
- seriesColors: ['#00cc00','#cc0000','#0000cc','#eeee00','912cee','#ff1493'],
+ seriesColors: ['#009926','#992600','#0036d6','#ffc414','730099','#ff00cc'],
seriesDefaults: {
lineWidth:1,
fill:true,
12 years
r5302 - branches/play_plumage/cumin/python/cumin
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-13 17:22:53 +0000 (Fri, 13 Apr 2012)
New Revision: 5302
Modified:
branches/play_plumage/cumin/python/cumin/model.py
Log:
Now limiting the number of samples that will be crammed into any single reporting chart.
Modified: branches/play_plumage/cumin/python/cumin/model.py
===================================================================
--- branches/play_plumage/cumin/python/cumin/model.py 2012-04-13 15:18:31 UTC (rev 5301)
+++ branches/play_plumage/cumin/python/cumin/model.py 2012-04-13 17:22:53 UTC (rev 5302)
@@ -428,6 +428,7 @@
"total":"sum(case when state = 'Claimed' or state = 'Available' or state = 'Unavailable' then 1 else 0 end) as value",
"unavail":"sum(case when state = 'Unavailable' then 1 else 0 end) as value"}
+ self.max_samples = 60
def avg_samples(self, stat, secs, interval, secs2):
updated_col = self.table._columns_by_name[self.update_col]
@@ -464,6 +465,13 @@
del self.query.filters[:]
self.query.filters = org_filters
+ reduced_samples = None
+ if len(samples) > self.max_samples:
+ reduced_interval = int(len(samples) / self.max_samples)
+ # show every N samples + we always want the first and last
+ reduced_samples = [sample for i, sample in enumerate(samples) if i % reduced_interval == 0 or i == 0 or i == len(samples) -1]
+ samples = reduced_samples
+
return samples
12 years
r5301 - in branches/play_plumage/cumin: python/cumin/grid resources
by croberts@fedoraproject.org
Author: croberts
Date: 2012-04-13 15:18:31 +0000 (Fri, 13 Apr 2012)
New Revision: 5301
Modified:
branches/play_plumage/cumin/python/cumin/grid/dashboard.py
branches/play_plumage/cumin/resources/app.js
Log:
Adding more data to the utilization chart for demo purposes.
Modified: branches/play_plumage/cumin/python/cumin/grid/dashboard.py
===================================================================
--- branches/play_plumage/cumin/python/cumin/grid/dashboard.py 2012-04-12 22:33:53 UTC (rev 5300)
+++ branches/play_plumage/cumin/python/cumin/grid/dashboard.py 2012-04-13 15:18:31 UTC (rev 5301)
@@ -95,10 +95,10 @@
self.add_child(os_data)
chart = self.UtilChart(app, "util", app.model.com_redhat_grid_plumage.OSUtil)
- chart.stats = ["total", "avail", "used"]
+ chart.stats = ["total", "avail", "unavail", "used"]
# these titles are a temporary requirement. Eventually, we will have an "aggregate" class available that will
# let us store queries or other info in the rosemary.xml model but NOT have an actual column in the database
- chart.titles = ["Total", "Available", "Used"]
+ chart.titles = ["Total", "Available", "Unavailable", "Used", ]
chart.duration.param.default = "3600"
self.add_child(chart)
Modified: branches/play_plumage/cumin/resources/app.js
===================================================================
--- branches/play_plumage/cumin/resources/app.js 2012-04-12 22:33:53 UTC (rev 5300)
+++ branches/play_plumage/cumin/resources/app.js 2012-04-13 15:18:31 UTC (rev 5301)
@@ -629,7 +629,7 @@
}
},
series: getSeries(dataContainer),
- seriesColors: ['#00cc00','#cc0000','#0000cc'],
+ seriesColors: ['#00cc00','#cc0000','#0000cc','#eeee00','912cee','#ff1493'],
seriesDefaults: {
lineWidth:1,
fill:true,
12 years