Author: tmckay
Date: 2013-02-11 21:53:18 +0000 (Mon, 11 Feb 2013)
New Revision: 5683
Modified:
branches/tmckay/cumin/bin/cumin-web
branches/tmckay/cumin/python/cumin/config.py
branches/tmckay/cumin/python/cumin/grid/tags.py
branches/tmckay/cumin/python/cumin/main.py
branches/tmckay/mint/python/mint/main.py
branches/tmckay/sage/python/sage/wallaby/wallabyoperations.py
Log:
Intermediate commit
Modified: branches/tmckay/cumin/bin/cumin-web
===================================================================
--- branches/tmckay/cumin/bin/cumin-web 2013-02-11 21:10:39 UTC (rev 5682)
+++ branches/tmckay/cumin/bin/cumin-web 2013-02-11 21:53:18 UTC (rev 5683)
@@ -34,12 +34,11 @@
cumin.aviary_prefer_condor = values.aviary_prefer_condor
def set_wallaby_configs(cumin, values, brokers):
- if values.wallaby_broker == "":
- cumin.wallaby_broker = brokers[0]
- else:
- if values.wallaby_broker == "None":
- values.wallaby_broker = None
- cumin.wallaby_broker = values.wallaby_broker
+ if values.wallaby_broker == "None":
+ values.wallaby_broker = None
+
+ cumin.wallaby_broker = values.wallaby_broker
+
# Let 0 indicate "no timeout", since the timeout
# value is an int in the config and None can't be
# specified
Modified: branches/tmckay/cumin/python/cumin/config.py
===================================================================
--- branches/tmckay/cumin/python/cumin/config.py 2013-02-11 21:10:39 UTC (rev 5682)
+++ branches/tmckay/cumin/python/cumin/config.py 2013-02-11 21:53:18 UTC (rev 5683)
@@ -231,7 +231,7 @@
# param.default = False
param = ConfigParameter(self, "wallaby-broker", str)
- param.default = ""
+ param.default = "http://localhost"
param = ConfigParameter(self, "wallaby-refresh", int)
param.default = 60
Modified: branches/tmckay/cumin/python/cumin/grid/tags.py
===================================================================
--- branches/tmckay/cumin/python/cumin/grid/tags.py 2013-02-11 21:10:39 UTC (rev 5682)
+++ branches/tmckay/cumin/python/cumin/grid/tags.py 2013-02-11 21:53:18 UTC (rev 5683)
@@ -461,6 +461,8 @@
self.page.redirect.set(session, url)
def validate(self, session):
+ return True
+
results = self.app.wallaby.validate_configuration()
#handle the set of explanations that could come back if things didn't go
well
@@ -1141,7 +1143,7 @@
return "Activate wallaby configuration"
def callback(self, results):
- if len(results[0]) != 0 or len(results[1]) != 0:
+ if results is not None:
self.invoc.status = self.invoc.FAILED
self.invoc.end()
Modified: branches/tmckay/cumin/python/cumin/main.py
===================================================================
--- branches/tmckay/cumin/python/cumin/main.py 2013-02-11 21:10:39 UTC (rev 5682)
+++ branches/tmckay/cumin/python/cumin/main.py 2013-02-11 21:53:18 UTC (rev 5683)
@@ -251,13 +251,8 @@
self.remote.add_mechanisms(ops)
# Create RPC interface for Wallaby
- # The Wallaby client API needs a broker connection. The broker host
- # for Wallaby is not necessarily the same as the broker used by
- # cumin-web for grid data, and flags set on the Session for cumin
- # may not be appropriate for the Wallaby API. So, it gets its own.
self.wallaby = WallabyOperations(self.wallaby_broker,
- self.wallaby_refresh,
- self.sasl_mech_list)
+ self.wallaby_refresh)
self.model.init()
self.session.init()
Modified: branches/tmckay/mint/python/mint/main.py
===================================================================
--- branches/tmckay/mint/python/mint/main.py 2013-02-11 21:10:39 UTC (rev 5682)
+++ branches/tmckay/mint/python/mint/main.py 2013-02-11 21:53:18 UTC (rev 5683)
@@ -94,6 +94,9 @@
self.model.com_redhat_grid.Scheduler,
self.model.com_redhat_grid.Slot]
+# supported_classes = [self.model.com_redhat_grid.Submitter,
+# self.model.com_redhat_grid.Scheduler]
+
for c in supported_classes:
if c in self._qmf_classes:
self._qmf_classes.remove(c)
Modified: branches/tmckay/sage/python/sage/wallaby/wallabyoperations.py
===================================================================
--- branches/tmckay/sage/python/sage/wallaby/wallabyoperations.py 2013-02-11 21:10:39 UTC
(rev 5682)
+++ branches/tmckay/sage/python/sage/wallaby/wallabyoperations.py 2013-02-11 21:53:18 UTC
(rev 5683)
@@ -1,18 +1,14 @@
import logging
import time
-from qmf.console import Session
-from sage.util import CallThread, get_sasl_mechanisms
+from sage.util import CallThread, parse_URL
from threading import Lock, Condition
#import cProfile
imports_ok = True
try:
- import wallaby
- import wallaby.tagging
- import wallaby.collections
- setup = wallaby.tagging.setup
+ import wallaroo
except:
imports_ok = False
@@ -33,19 +29,20 @@
'''
Wrapper around the Wallaby client library.
'''
- def __init__(self, broker_uri, refresh_interval=None, sasl_mech_list=None):
+ def __init__(self, wallaroo_uri, refresh_interval=None):
'''
Constructor.
- broker_uri -- the URI used to connect to a QMF message broker
- where a Wallaby agent is connected. The simplest URI is just a
- hostname but a full URI can specify scheme://user/password@host:port
- or a subset of those components as long as the host is included.
+ wallaroo_uri -- the URI used to connect to a wallaroo server.
+ The simplest URI is just a hostname but a full URI can specify
+ scheme://user/password@host:port or a subset of those components
+ as long as the host is included.
+
Examples:
localhost
localhost:5672
- amqp://fred/flintsone@quarry.bedrock.com:1234
+
http://fred/flintsone@quarry.bedrock.com:1234
refresh_interval -- default refresh interval in seconds for all items
maintained by WallabyOperations' internal caching thread. A value of
@@ -53,23 +50,12 @@
item after a successful call unless the refresh() method is used.
The refresh interval may be set for items individually with the
set_interval() method.
-
- sasl_mech_list -- restricts the list of sasl mechanisms
- that will be allowed when connecting to a QMF message broker.
- If the broker URL contains no credentials, default is ANONYMOUS.
- If the broker URL does contain credentials, default is
- 'PLAIN DIGEST-MD5'
'''
- self.broker_uri = broker_uri
- self.sasl_mech_list = get_sasl_mechanisms(broker_uri,
- sasl_mech_list)
+ self.wallaroo_uri = wallaroo_uri
# A wallaby Store object
self._store = None
- # A QMF broker
- self._broker = None
-
# The cache maintenance thread
self._maintain_cache = None
@@ -80,12 +66,13 @@
# an attribute on the Wallaby Store object, with the exception of
# WBTypes.TAGS. The TAGS data is a subset of the GROUPS produced
# in this module.
- self._cache = {WBTypes.NODES: self.CacheData(refresh_interval),
- WBTypes.GROUPS: self.CacheData(refresh_interval),
- WBTypes.FEATURES: self.CacheData(refresh_interval),
- WBTypes.TAGS: self.CacheData(refresh_interval,
-
synthetic=self._generate_tag_data)}
+ self._cache = {WBTypes.NODES: self.NodeData(refresh_interval),
+ WBTypes.GROUPS: self.GroupData(refresh_interval),
+ WBTypes.FEATURES: self.FeatureData(refresh_interval),
+ WBTypes.TAGS: self.TagData(refresh_interval,
+ synthetic=True)}
+
# Cache a list of nodes that are members of a tag
self._nodes_by_tag = dict()
@@ -103,9 +90,9 @@
'''
Start the caching thread.
- This thread will attempt to connect to the broker and retrieve
- a Store object from the Wallaby agent. If successful, it will
- periodically retrieve and cache data from Wallaby.
+ This thread will attempt to connect to wallaroo and retrieve
+ a Store object. If successful, it will periodically retrieve
+ and cache data from wallaroo.
Only one caching thread may run at a time. The thread may
be restarted if it has previously been stopped.
@@ -114,35 +101,35 @@
should only be called from a single thread.
retry_secs -- how often the caching thread will retry failed
- operations. This includes attempts to connect to the broker
- and retrieve a Store object as well as calls to Wallaby that
+ operations. This includes attempts to connect to wallaroo
+ and retrieve a Store object as well as calls to wallaroo that
return no data.
'''
- # The connection to the broker can actually take a long
- # time to complete. We don't want to hang a calling function,
- # so we handle the connection and retrieval of the
- # initial Store object from Wallaby in a thread.
- # (There may need to be more work here if the broker or wallaby
- # going away and coming back causes a problem, but with
- # manageConnections=True and well-known agent/object ids for
- # Wallaby it appears to recover on its own...)
- # Similarly, getting node lists etc may take a long time
- # especially over a slow network. So we use the same thread
- # to retrieve things like node lists at defined intervals.
-
# 'self' here is really a term of art since this is a local
# function, but it refers to the WallabyOperations object
# so the code reads nicely
def maintain_cache(self):
+
+ # Turn the URL into a dictionary of args that the wallaroo
+ # client.connect() method wants to see
+ url = parse_URL(self.wallaroo_uri)
+ arg_map = (("host",),
+ ("port",),
+ ("scheme",),
+ ("user", "username"),
+ ("password", "pw"))
+ connect_args = dict()
+ for arg in arg_map:
+ v = getattr(url, arg[0])
+ if v is not None:
+ connect_args[arg[-1]] = v
+
# Get initinal connection and Store obect
- self.session = Session(manageConnections=True)
- self.broker = self.session.addBroker(self.broker_uri,
mechanisms=self.sasl_mech_list)
while not self._stop:
- self._store = self._get_store()
+ self._store = wallaroo.client.connect(**connect_args)
if self._store is not None:
- setup(self._store)
self._partition_group = self._store.getPartitionGroup().name
log.debug("WallabyOperations: found wallaby store
object")
break
@@ -163,21 +150,15 @@
while not self._stop:
start_processing = time.time()
for attr, val in self._cache.iteritems():
- if self._stop:
- break
+ if self._stop:
+ break
- # val.remaining is the number of seconds left before
- # the next update of this data item. None is
"forever".
- # Synthetic items are not retreived from the store.
- if not val.synthetic and \
- val.remaining is not None and val.remaining <= 0:
- d = get_values(attr, getattr, self._store, attr, [])
- # If the data is empty, _set_cache will leave the
- # remaining field set to 0 for the attribute so we
- # will try to get it again on our next retry.
- # Otherwise, remaining will be reset to the full
- # interval for this attribute.
- self._set_cache(attr, d)
+ # val.remaining is the number of seconds left before
+ # the next update of this data item. None is
"forever".
+ # Synthetic items are not retreived from the store.
+ if not val.synthetic and \
+ val.remaining is not None and val.remaining <= 0:
+ val.get_values(attr, self)
# Now handle the synthetics. val.synthetic generates
# and stores it's own results.
@@ -187,7 +168,7 @@
if val.synthetic and \
val.remaining is not None and val.remaining <= 0:
- get_values(attr, val.synthetic, *val.args)
+ val.get_values(attr, self)
log.debug("WallabyOperations: total refresh processing time
%s" \
% (time.time() - start_processing))
@@ -225,12 +206,6 @@
self._set_cache(attr, [])
self._store = None
- # Have to clean up the broker
- try:
- self.session.delBroker(self.broker)
- except:
- pass
-
#end maintain_cache
def get_values(attr, call, *args):
@@ -252,8 +227,9 @@
try:
maintain_cache(self)
log.debug("WallabyOperations: cache maintenance thread
exited")
- except:
- pass
+ except Exception:
+ import traceback
+ traceback.print_exc()
if self._maintain_cache is not None and \
self._maintain_cache.isAlive():
@@ -262,7 +238,7 @@
self._stop = False
- if self.broker_uri is not None:
+ if self.wallaroo_uri is not None:
# self._maintain_cache =
CallThread(cProfile.runctx('maintain_cache(self)', globals(), locals(),
filename='sage.stats'), None)
self._maintain_cache = CallThread(wrap_maintain_cache, None)
self._maintain_cache.daemon = True
@@ -443,7 +419,7 @@
self._lock.acquire()
try:
if n in self._cache[WBTypes.NODES].data:
- names = self._cache[WBTypes.NODES].data[n].getTags()
+ names = self._cache[WBTypes.NODES].data[n].getLabels()
finally:
self._lock.release()
return names
@@ -461,7 +437,7 @@
self._lock.acquire()
try:
for name in names:
- self._store.addTag(name)
+ self._store.addLabel(name)
except Exception, e:
log.debug("WallabyOperations: create_tag, exception suppressed,
%s" % str(e))
return False
@@ -518,7 +494,7 @@
# cast to list in case tags is a tuple, list required
start = time.time()
self._lock.acquire()
- n.modifyTags("REPLACE", list(tags),
create_missing_tags=True)
+ n.modifyLabels("REPLACE", list(tags),
create_missing_tags=True)
n.update()
status = True
@@ -555,7 +531,7 @@
else:
try:
start = time.time()
- g.modifyFeatures("REPLACE", features)
+ g.modifyFeatures("REPLACE", features, {})
g.update()
status = True
delta = time.time() - start
@@ -601,7 +577,7 @@
returns result of store.validateConfiguration.
'''
if self._store is not None:
- return self._store.validateConfiguration()
+ return self._store.checkBranchValidity()
else:
log.debug("WallabyOperations: validate_configuration," \
" store object not yet created")
@@ -628,23 +604,6 @@
if t is not None and t < min:
t = min
return t
-
- def _get_store(self):
- # Ideally there should only be a single Store object from a single agent.
- # We constrain the search a bit, hopefully more efficient
- store = None
- try:
- agents = [agent for agent in self.session.getAgents() \
- if "com.redhat.grid.config:Store" in agent.label]
- for agent in agents:
- s = self.session.getObjects(_agent = agent, _class =
"Store")
- if len(s) > 0:
- # And finally wrap the QMF object in a wallaby.Store wrapper
- store = wallaby.Store(s[0], self.session)
- break
- except:
- pass
- return store
def _generate_tag_data(self):
# figure out the tag list and nodes per tag
@@ -655,8 +614,8 @@
if not g.name.startswith("+++") and \
self._store.isTag(g):
tags.append(g)
- nodes = g.membership()
- nodes_by_tag[g.name] = nodes
+ #nodes = g.membership()
+ nodes_by_tag[g.name] = []
self._lock.acquire()
try:
@@ -670,7 +629,7 @@
def _set_cache(self, attr, data):
self._lock.acquire()
try:
- self._cache[attr].data = self._to_dict(data)
+ self._cache[attr].data = dict(data)
self._cache[attr].reset_remaining(len(data) == 0)
finally:
self._lock.release()
@@ -710,6 +669,44 @@
else:
self.remaining = self.interval
+ class NodeData(CacheData):
+ def get_values(self, attr, ops):
+ ops._set_cache(attr,
+ [(name, ops._store.getNode(name)) for name in ops._store.nodes])
+
+ class GroupData(CacheData):
+ def get_values(self, attr, ops):
+ ops._set_cache(attr,
+ [(name, ops._store.getGroupByName(name)) for name in
ops._store.groups])
+
+ class FeatureData(CacheData):
+ def get_values(self, attr, ops):
+ ops._set_cache(attr,
+ [(name, ops._store.getFeature(name)) for name in ops._store.features])
+
+ class TagData(CacheData):
+ def get_values(self, attr, ops):
+ # figure out the tag list and nodes per tag
+ groups = ops.get_data(WBTypes.GROUPS)
+ tags = []
+ nodes_by_tag = dict()
+ for g in groups:
+ if not g.name.startswith("+++") and \
+ ops._store.isLabel(g):
+ tags.append(g)
+ nodes = [] #g.membership()
+ nodes_by_tag[g.name] = nodes
+
+ ops._lock.acquire()
+ try:
+ ops._cache[WBTypes.TAGS].data = ops._to_dict(tags)
+ ops._cache[WBTypes.TAGS].reset_remaining(len(tags) == 0)
+ ops._nodes_by_tag = nodes_by_tag
+ finally:
+ ops._lock.release()
+ log.debug("WallabyOperations: %s list updated (%s items)" \
+ % (WBTypes.TAGS, len(tags)))
+
else:
class WallabyOperations(object):
'''