r5670 - in branches/tmckay: mint/python/mint mint/python/mint/aviary sage/python/sage/aviary sage/rpc-defs/aviary
by tmckay@fedoraproject.org
Author: tmckay
Date: 2013-02-04 18:05:45 +0000 (Mon, 04 Feb 2013)
New Revision: 5670
Modified:
branches/tmckay/mint/python/mint/aviary/endpoints.py
branches/tmckay/mint/python/mint/aviary/locator.py
branches/tmckay/mint/python/mint/aviary/session.py
branches/tmckay/mint/python/mint/aviary/submissions.py
branches/tmckay/mint/python/mint/aviary/work.py
branches/tmckay/mint/python/mint/main.py
branches/tmckay/mint/python/mint/update.py
branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
branches/tmckay/sage/rpc-defs/aviary/aviary-collector.xsd
Log:
Intermediate update. Includes collector and negotiator.
Modified: branches/tmckay/mint/python/mint/aviary/endpoints.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/endpoints.py 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/mint/python/mint/aviary/endpoints.py 2013-02-04 18:05:45 UTC (rev 5670)
@@ -1,6 +1,7 @@
from util import now_minus_duration
from datetime import datetime
from submissions import QuerySubmissionsBefore, QuerySubmissionsAfter
+from collector import QueryCollector, QueryNegotiator
class Endpoint(object):
def __init__(self, url, id):
@@ -17,9 +18,13 @@
class SubmissionEndpoint(Endpoint):
def __init__(self, url, id, factory):
super(SubmissionEndpoint, self).__init__(url, id)
- self.size = 4 # do we want to control size per endpoint record?
+ self.size = 8 # do we want to control size per endpoint record?
self.app = factory.app
+ my_classes = ("Submission")
+ self.classes = [c for c in factory.abbrev_classes \
+ if c in my_classes]
+
# Set up time spans for polling/history and initialize search indices
self.init_query_indices(factory.poll_delta,
factory.history_delta)
@@ -53,21 +58,75 @@
self.before_cutoff = now_minus_duration(self.history_delta, now)
def get_work_list(self):
+ if "Submission" in self.classes:
+ return (QuerySubmissionsAfter(self),
+ QuerySubmissionsBefore(self),)
+ return ()
+
+class CollectorEndpoint(Endpoint):
+ def __init__(self, url, id, factory):
+ super(CollectorEndpoint, self).__init__(url, id)
- return (QuerySubmissionsAfter(self),
- QuerySubmissionsBefore(self),)
-
+ self.app = factory.app
+
+ my_classes = ("Collector","Negotiator")
+ self.classes = [c for c in factory.abbrev_classes \
+ if c in my_classes]
+
+ def get_work_list(self):
+ work = []
+ if "Collector" in self.classes:
+ work.append(QueryCollector(self))
+
+ if "Negotiator" in self.classes:
+ work.append(QueryNegotiator(self))
+ return work
+
class EndpointFactory(object):
- def __init__(self, app, poll_delta, history_delta):
-
+ def __init__(self, app, classes, poll_delta, history_delta):
+ '''
+ classes is a list of (package, class) tuples containing
+ the names of packages and classes that should be polled.
+ The package and class names come from the Rosemary data
+ model and determine which endpoint types we are interested in
+ and which classes provided by that endpoint we are interested in.
+ '''
self.app = app
+ self.classes = classes
+ # For now we only support com.redhat.grid so
+ # make a class list without package names
+ self.abbrev_classes = [c[1] for c in self.classes \
+ if c[0] == "com.redhat.grid"]
+
self.poll_delta = poll_delta
self.history_delta = history_delta
self.type_map = \
- {("CUSTOM","QUERY_SERVER"): self._make_submission_endpoint}
+ {("CUSTOM","QUERY_SERVER"): self._make_submission_endpoint,
+ ("COLLECTOR", ""): self._make_collector_endpoint}
+
+ def get_endpoint_types(self):
+ # This routine maps Rosemary class names to Aviary
+ # endpoint types (resource, sub_type) so a locator class can know
+ # what types of endpoints to report
+ classes_to_aviary = \
+ {"Collector": ("COLLECTOR", ""),
+ "Negotiator": ("COLLECTOR", ""),
+ "Submission": ("CUSTOM", "QUERY_SERVER")}
+
+ endpoints = set()
+ for c in self.abbrev_classes:
+ try:
+ endpoints.add(classes_to_aviary[c])
+ except:
+ pass
+ return list(endpoints)
+
+ def _make_collector_endpoint(self, url, id):
+ return CollectorEndpoint(url, id, self)
+
def _make_submission_endpoint(self, url, id):
return SubmissionEndpoint(url, id, self)
Modified: branches/tmckay/mint/python/mint/aviary/locator.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/locator.py 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/mint/python/mint/aviary/locator.py 2013-02-04 18:05:45 UTC (rev 5670)
@@ -7,8 +7,7 @@
log = logging.getLogger("mint.aviary")
class LocatorThread(MintBlockingDaemonThread):
- def __init__(self, aviary_ops, foreman, endpoint_factory,
- resource="ANY", subtype=""):
+ def __init__(self, aviary_ops, foreman, endpoint_factory):
# We don't care about app for this thread,
# at least not currently
@@ -17,16 +16,19 @@
self.aviary_ops = aviary_ops
self.foreman = foreman
self.ep_fact = endpoint_factory
- self.resource = resource
- self.subtype = subtype
self.daemon = True
self.stop_requested = False
- self.interval_secs = 60
+ self.interval_secs = 30
self.start()
def run(self):
+ # Find out from the endpoint factory what endpoint
+ # types we should query for. This is a list of
+ # (resource, subtype) tuples.
+ endpoint_types = self.ep_fact.get_endpoint_types()
+
def block():
self._condition.acquire()
if not self.stop_requested:
@@ -36,8 +38,7 @@
endpoints = dict()
while not self.stop_requested:
try:
- ends = self.aviary_ops.get_endpoints(self.resource,
- self.subtype)
+ ends = self.aviary_ops.get_endpoints_list(endpoint_types)
except Exception:
import traceback
message = traceback.format_exc()
Modified: branches/tmckay/mint/python/mint/aviary/session.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/session.py 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/mint/python/mint/aviary/session.py 2013-02-04 18:05:45 UTC (rev 5670)
@@ -34,20 +34,24 @@
log.info("Starting %s", self)
# Handle bind by class
+ classes = []
for cls in self.classes:
pname = cls._package._name
cname = cls._name
-
+ classes.append((pname, cname))
+
log.info("Querying for Aviary class %s.%s" % (pname, cname))
self.foreman = Foreman()
endpoint_factory = EndpointFactory(self.app,
+ classes,
poll_delta=timedelta(days=64),
history_delta=timedelta(weeks=52))
self.locatorThread = LocatorThread(self.app.aviary_ops,
- self.foreman, endpoint_factory)
+ self.foreman,
+ endpoint_factory)
def stop(self):
log.info("Stopping %s", self)
Modified: branches/tmckay/mint/python/mint/aviary/submissions.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/submissions.py 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/mint/python/mint/aviary/submissions.py 2013-02-04 18:05:45 UTC (rev 5670)
@@ -1,7 +1,6 @@
from work import Work
from util import now_minus_duration
from datetime import datetime
-
from mint.update import AviarySubmissionUpdate
def max(stuff):
@@ -30,7 +29,7 @@
self.ep.after_index,
"AFTER")
- #print ("querying after", self.ep.after_index, datetime.fromtimestamp(self.ep.after_index))
+ print ("querying after", self.ep.after_index, datetime.fromtimestamp(self.ep.after_index))
#print (ids.status, ids.got_data)
# if data.ids is empty (remaining == 0)
@@ -39,12 +38,15 @@
# yet. This is how we will accomplish polling,
# just looping over the data continuously.
if ids.status == 0 and ids.got_data:
+ print len(ids.data.ids)
#print ids.data.ids
# ids.data.ids is a list of SubmissionIDs
# Each of these ids needs to be polled to
# get info out.
if ids.data.ids:
+ print len(ids.data.ids)
+ print ids.data.ids
work = PollSubmissionIDs(ids.data.ids, self.ep)
foreman.add_work((work,))
@@ -86,7 +88,7 @@
self.ep.before_index,
"BEFORE")
- #print ("querying before", self.ep.before_index, datetime.fromtimestamp(self.ep.before_index))
+ print ("querying before", self.ep.before_index, datetime.fromtimestamp(self.ep.before_index))
#print (ids.status, ids.got_data)
# if data.ids is empty (remaining == 0)
@@ -95,6 +97,7 @@
# yet. This is how we will accomplish polling
# We need a date to qdate function.
if ids.status == 0 and ids.got_data:
+ print len(ids.data.ids)
# ids.data.ids is a list of SubmissionIDs
# each of these ids needs to be polled to
# get info out.
@@ -116,12 +119,10 @@
# datetime.fromtimestamp(self.ep.before_index),
# datetime.fromtimestamp(self.ep.before_cutoff))
self.ep.before_queries_completed = True
-# self.ep.before_index = now_minus_duration(self.ep.poll_delta)
else:
- #print "query before reset"
+ print "query before exiting"
self.ep.before_queries_completed = True
-# self.ep.before_index = now_minus_duration(self.ep.poll_delta)
else:
print "query before failed"
@@ -153,9 +154,6 @@
# summaries can include job summaries, but we don't need those right now.
# We need to adapt these summaries to a QMF style submission
-# <property name="jobserverRef" type="objId" parentRef="y" index="y" references="com.redhat.grid:JobServer"/>
-# <property name="schedulerRef" type="objId" parentRef="y" index="y" references="com.redhat.grid:Scheduler"/>
-
if summaries.status == 0 and summaries.got_data:
for summary in summaries.data:
summary.Name = str(summary.id.name)
@@ -173,20 +171,8 @@
summary.endpoint_id = self.ep.id
#print summary.Name
- a = AviarySubmissionUpdate(self.ep.app.model, summary)
- self.ep.app.update_thread.enqueue(a)
+ if self.ep.app.model:
+ a = AviarySubmissionUpdate(self.ep.app.model, summary)
+ self.ep.app.update_thread.enqueue(a)
-# summary.statistics = {
-
-# properties = Name, Owner, Qdate, jobserverRef, schedulerRef
-# statistics = Idle, Running, Removed, Completed, Held, TransferringOutput, Suspended
-
-# (SubmissionSummary){
-# id =
-# (SubmissionID){
-# _scheduler = "RHEL6Virt-tmckay"
-# _pool = "RHEL6Virt-tmckay"
-# name = "RHEL6Virt-tmckay#24"
-# owner = "tmckay"
-
self.complete = True
Modified: branches/tmckay/mint/python/mint/aviary/work.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/work.py 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/mint/python/mint/aviary/work.py 2013-02-04 18:05:45 UTC (rev 5670)
@@ -1,11 +1,20 @@
-from threading import Thread, Lock
+import bisect
+import logging
+
+from threading import Thread, Lock, Condition
from Queue import Queue
from time import sleep
+from datetime import timedelta, datetime
+log = logging.getLogger("mint.aviary")
+
class Work(object):
def __init__(self):
- pass
+ # Set this to None for no delay on reruning
+ # the task when it is returned to the queue
+ self.recycle_interval = timedelta(seconds=10)
+
def do_the_job(self, foreman):
pass
@@ -22,11 +31,14 @@
for _ in range(num_workers):
self.workers.append(Worker(self))
+ self.waiting = WaitList(self.work_to_do)
+
def end_work(self):
# Add an empty work object on the queue so that any worker
# waiting will be woken up with a "get"
if not self.stop_requested:
self.stop_requested = True
+ self.waiting.stop()
for _ in range(len(self.workers)):
self.work_to_do.put(Work())
for w in self.workers:
@@ -45,10 +57,13 @@
# continued by this thread or requeued
if current is not None and \
not current.completed():
- # If there is other work to do, requeue
- # the current task and get new work.
- # Otherwise, just continue.
- if self.work_to_do.empty():
+
+ # If there is a recycle interval specified,
+ # put the task on the wait list
+ if current.recycle_interval is not None:
+ self.waiting.add_waiter(current,
+ datetime.now() + current.recycle_interval)
+ elif self.work_to_do.empty():
work = current
else:
self.work_to_do.put(current)
@@ -77,3 +92,67 @@
break # quittin' time
work.do_the_job(self.foreman)
sleep(1)
+
+
+class WaitList(Thread):
+ def __init__(self, ready_queue):
+ super(WaitList, self).__init__()
+
+ self.waiting = []
+ self.ready_queue = ready_queue
+
+ self._lock = Lock()
+ self._condition = Condition(self._lock)
+
+ self.stop_requested = False
+
+ self.start()
+
+ def stop(self, timeout=5):
+ self._condition.acquire()
+ self.stop_requested = True
+ self._condition.notify()
+ self._condition.release()
+
+ if self.isAlive():
+ self.join(timeout)
+ log.info("%s stopped" % self.getName())
+
+ def add_waiter(self, work, alarm):
+ self._condition.acquire()
+
+ # Ask where this will be inserted, and then insert it
+ idx = bisect.bisect_right(self.waiting, (alarm, work))
+ bisect.insort_right(self.waiting, (alarm, work))
+
+ # If we are inserting the first item on the list we
+ # want to wake up the thread so it can reset its sleep time
+ if idx == 0:
+ self._condition.notify()
+
+ self._condition.release()
+
+ def run(self):
+ self._condition.acquire()
+ while True:
+ # Sleep forever or until the alarm time
+ # for the first item on the list
+ if self.waiting:
+ d = self.waiting[0][0] - datetime.now()
+ s = d.days * 86400 + d.seconds
+ else:
+ s = None
+ if s is None or s > 0:
+ self._condition.wait(s)
+
+ # Might be time to shut down...
+ if self.stop_requested:
+ break
+
+ # See if something finished...
+ if self.waiting and \
+ datetime.now() > self.waiting[0][0]:
+ work = self.waiting.pop(0)[1]
+ self.ready_queue.put(work)
+ work = None
+ self._condition.release()
Modified: branches/tmckay/mint/python/mint/main.py
===================================================================
--- branches/tmckay/mint/python/mint/main.py 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/mint/python/mint/main.py 2013-02-04 18:05:45 UTC (rev 5670)
@@ -87,7 +87,10 @@
if not (self.aviary_query_servers or self.aviary_locator):
return
- supported_classes = [self.model.com_redhat_grid.Submission]
+ supported_classes = [self.model.com_redhat_grid.Submission,
+ self.model.com_redhat_grid.Collector,
+ self.model.com_redhat_grid.Negotiator]
+
for c in supported_classes:
if c in self._qmf_classes:
self._qmf_classes.remove(c)
Modified: branches/tmckay/mint/python/mint/update.py
===================================================================
--- branches/tmckay/mint/python/mint/update.py 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/mint/python/mint/update.py 2013-02-04 18:05:45 UTC (rev 5670)
@@ -698,12 +698,13 @@
self.delete_agent_objects(cursor, stats, agent)
-class AviarySubmissionUpdate(Update):
- def __init__(self, model, submission):
- super(AviarySubmissionUpdate, self).__init__(model)
+class AviaryUpdate(Update):
+ def __init__(self, model, aviary_obj, cls):
+ super(AviaryUpdate, self).__init__(model)
- self.submission = submission
- self.cls = model.com_redhat_grid.Submission
+ self.aviary_obj = aviary_obj
+ self.cls = cls
+ self.delete = False
def do_process(self, cursor, stats):
agent_id = self.get_agent_id()
@@ -711,29 +712,42 @@
# For aviary stuff, the agent is inferred
# There really are no separate updates, so
- # we just base the agent id on info in the submission
+ # we just base the agent id on info in the object
try:
agent = self.model.agents_by_id[agent_id]
except KeyError:
+ if self.delete:
+ raise UpdateDropped()
agent = MintAgent(self.model, agent_id)
stats.agents_created += 1
try:
obj = agent.get_object(cursor, self.cls, object_id)
except RosemaryNotFound:
+ if self.delete:
+ raise UpdateDropped()
obj = self.create_object(cursor, stats, self.cls)
return
- self.update_object(cursor, stats, obj)
+ if self.delete:
+ self.delete_object(cursor, stats, obj)
+ del agent.objects_by_id[obj._qmf_object_id]
+ else:
+ self.update_object(cursor, stats, obj)
def get_agent_id(self):
- return self.submission.endpoint_id.resource + ":" + \
- self.submission.endpoint_id.sub_type + ":" + \
- self.submission.endpoint_id.name
+ return self.aviary_obj.endpoint_id.resource + ":" + \
+ self.aviary_obj.endpoint_id.sub_type + ":" + \
+ self.aviary_obj.endpoint_id.name
def get_object_id(self):
- return self.submission.id.name
+ return self.aviary_obj.id.name
+ def delete_object(self, cursor, stats, obj):
+ obj.delete(cursor)
+ self.model.print_event(3, "Deleted %s", obj)
+ stats.objects_deleted += 1
+
def create_object(self, cursor, stats, cls):
update_time = datetime.now()
create_time = update_time
@@ -834,39 +848,15 @@
# are in 10s of microseconds, so we need to
# handle our own conversion here
if prop.type == "absTime":
- return datetime.fromtimestamp(value)
+ return datetime.fromtimestamp(float(value))
t = qmf_type_code_by_string[prop.type]
return transformers[t](value)
def process_properties(self, obj, columns, cursor):
cls = obj._class
- # Fix up submissions to hold a scheduler ref
- # remember that we saw this
- ref = cls._references_by_name["schedulerRef"]
- class FakeProp(object):
- def __init__(self, name):
- self.name = name
-
- class FakeOID(object):
- def __init__(self, agent, objname):
- self.agentName = agent
- self.objectName = objname
-
- val = FakeOID("com.redhat.grid:scheduler:" + \
- self.submission.id._scheduler,
- self.submission.id._scheduler)
- prop = FakeProp("schedulerRef")
- try:
- col, nvalue = self.process_reference(obj, prop, val, cursor)
- if nvalue != getattr(obj, col.name):
- setattr(obj, col.name, nvalue)
- columns.append(col)
- except MappingException, e:
- log.debug(e)
-
for name, prop in cls._properties_by_name.iteritems():
- value = getattr(self.submission, name)
+ value = getattr(self.aviary_obj, name)
if value is not None:
value = self.transform_value(prop, value)
if value == getattr(obj, prop.sql_column.name):
@@ -883,11 +873,7 @@
if not ref.sql_column:
raise MappingException("Reference %s has no column" % ref.name)
- # Okay, for submissions coming from Aviary we may actually see
- # a submission before we see its scheduler. So, let's go ahead
- # and assume the agent and create a deferred link.
- # If the agent never shows up, no harm done. We should probably
- # add a timeout to the agent and delete it if it does not show up.
+ # Just create the agent if we haven't seen it.
value = None
try:
agent = self.model.agents_by_id[oid.agentName]
@@ -916,7 +902,7 @@
# loop through the list of properties, look for
# the value in the object
for name, stat in self.cls._statistics_by_name.iteritems():
- value = getattr(self.submission, name)
+ value = getattr(self.aviary_obj, name)
if value is not None:
value = self.transform_value(stat, value)
@@ -966,6 +952,54 @@
return "%s(%s,%s,%s)" % (name, agent_id, self.cls, obj_id)
+class AviarySubmissionUpdate(AviaryUpdate):
+ def __init__(self, model, submission):
+ super(AviarySubmissionUpdate, self).__init__(model,
+ submission,
+ model.com_redhat_grid.Submission)
+
+ def process_properties(self, obj, columns, cursor):
+ cls = obj._class
+
+ # Fix up submissions to hold a scheduler ref
+ # remember that we saw this
+ ref = cls._references_by_name["schedulerRef"]
+
+ class FakeProp(object):
+ def __init__(self, name):
+ self.name = name
+
+ class FakeOID(object):
+ def __init__(self, agent, objname):
+ self.agentName = agent
+ self.objectName = objname
+
+ val = FakeOID("com.redhat.grid:scheduler:" + \
+ self.aviary_obj.id._scheduler,
+ self.aviary_obj.id._scheduler)
+ prop = FakeProp("schedulerRef")
+ try:
+ col, nvalue = self.process_reference(obj, prop, val, cursor)
+ if nvalue != getattr(obj, col.name):
+ setattr(obj, col.name, nvalue)
+ columns.append(col)
+ except MappingException, e:
+ log.debug(e)
+
+ super(AviarySubmissionUpdate, self).process_properties(obj, columns, cursor)
+
+
+class AviaryCollectorUpdate(AviaryUpdate):
+ def __init__(self, model, aviary_obj):
+
+ pkg = model._packages_by_name[aviary_obj.pkg_name]
+ cls = pkg._classes_by_lowercase_name[aviary_obj.class_name.lower()]
+
+
+ super(AviaryCollectorUpdate, self).__init__(model,
+ aviary_obj,
+ cls)
+
class UpdateDropped(Exception):
pass
Modified: branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- branches/tmckay/sage/python/sage/aviary/aviaryoperations.py 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/sage/python/sage/aviary/aviaryoperations.py 2013-02-04 18:05:45 UTC (rev 5670)
@@ -244,12 +244,15 @@
self._lock.release()
return res
-class Fake_Resource_ID(object):
- def __init__(self, resource, sub_type, name="", pool=""):
+class ResourceID(object):
+ def __init__(self, resource="COLLECTOR", pool="", name="",
+ address="", sub_type="", birthdate=None):
+ self.resource = resource
+ self.pool = pool
self.name = name
- self.pool = pool
- self.resource = resource
+ self.address = address
self.sub_type = sub_type
+ self.birthdate = birthdate
class Fake_Endpoint(object):
def __init__(self, resource_id, url_list):
@@ -301,7 +304,7 @@
def get_endpoints(self):
endpoints = []
for host, urls in self.servers.iteritems():
- res_id = Fake_Resource_ID(self.resource, self.subtype, host)
+ res_id = ResourceID(self.resource, sub_type=self.subtype, name=host)
url_strs = [str(u) for u in urls]
ep = Fake_Endpoint(res_id, url_strs)
endpoints.append(ep)
@@ -379,17 +382,26 @@
subtype)
def get_hosts(self, resource, subtype):
- if resource == "ANY" or \
- (resource == self.resource and \
- (not subtype or self.subtype == subtype)):
+ if self.servers and \
+ (resource == "ANY" or \
+ (resource == self.resource and \
+ (not subtype or self.subtype == subtype))):
return self.servers.get_hosts()
return set()
- def get_endpoints(self, resource, subtype):
- if resource == "ANY" or \
- (resource == self.resource and \
- (not subtype or self.subtype == subtype)):
- return self.servers.get_endpoints()
+ def get_endpoints(self, types):
+ # Types is a list of (resource, sub_type) tuples
+
+ if not self.servers:
+ return []
+
+ # Lets see if our resource/subtype is in
+ # any of those specified by the type list
+ for t in types:
+ if t[0] == "ANY" or \
+ (t[0] == self.resource and \
+ (not t[1] or self.subtype == t[1])):
+ return self.servers.get_endpoints()
return []
def _set_client_info(self, client, refresh=False):
@@ -1214,31 +1226,19 @@
self.client_pool.return_object(client)
return res;
-# This class mimics the ResourceID structure that Aviary expects
-class ResourceID(object):
- def __init__(self,
- resource="COLLECTOR",
- pool="", name="", address="", sub_type="", birthdate=None):
- self.resource = resource
- self.pool = pool
- self.name = name
- self.address = address
- self.sub_type = sub_type
- self.birthdate = birthdate
+def set_res_id(obj, resID):
+ resID.resource = obj.resource
+ if hasattr(obj, "pool") and obj.pool:
+ resID.pool = obj.pool
+ if hasattr(obj, "name") and obj.name:
+ resID.name = obj.name
+ if hasattr(obj, "address") and obj.address:
+ resID.address = obj.address
+ if hasattr(obj, "sub_type") and obj.sub_type:
+ resID.sub_type = self.sub_type
+ if hasattr(obj, "birthdate") and obj.birthdate is not None:
+ resID.birthdate = obj.birthdate
- def set(self, resID):
- resID.resource = self.resource
- if self.pool:
- resID.poool = self.pool
- if self.name:
- resID.name = self.name
- if self.address:
- resID.address = self.address
- if self.sub_type:
- resID.subtype = self.sub_type
- if not self.birthdate is None:
- resID.birthdate = self.birthdate
-
class AttributeRequest(object):
def __init__(self, res_id, name_list):
assert isinstance(res_id, ResourceID)
@@ -1264,23 +1264,23 @@
# Equivalence? We don't want no stinking QMF structural equivalence!
self.use_MethodResult_for_sync_calls = False
- def get_collector(self, host, ids):
- return self._get_object(host, ids, "getCollector")
+ def get_collector(self, host, ids=[], no_summary=False):
+ return self._get_object(host, ids, "getCollector", no_summary)
- def get_negotiator(self, host, ids):
- return self._get_object(host, ids, "getNegotiator")
+ def get_negotiator(self, host, ids=[], no_summary=False):
+ return self._get_object(host, ids, "getNegotiator", no_summary)
- def get_scheduler(self, host, ids):
- return self._get_object(host, ids, "getScheduler")
+ def get_scheduler(self, host, ids=[], no_summary=False):
+ return self._get_object(host, ids, "getScheduler", no_summary)
- def get_submitter(self, host, ids):
- return self._get_object(host, ids, "getSubmitter")
+ def get_submitter(self, host, ids=[], no_summary=False):
+ return self._get_object(host, ids, "getSubmitter", no_summary)
- def get_master(self, host, ids):
- return self._get_object(host, ids, "getMaster")
+ def get_master(self, host, ids=[], no_summary=False):
+ return self._get_object(host, ids, "getMaster", no_summary)
- def get_slot(self, host, ids):
- return self._get_object(host, ids, "getSlot")
+ def get_slot(self, host, ids=[], no_summary=False):
+ return self._get_object(host, ids, "getSlot", no_summary)
def get_master_id(self, host, page_size, mode=None, offset=None):
return self._get_obj_by_id(host, page_size, mode, offset, "getMasterID")
@@ -1288,9 +1288,9 @@
def get_slot_id(self, host, page_size, mode=None, offset=None):
return self._get_obj_by_id(host, page_size, mode, offset, "getSlotID")
- def get_attributes(self, host, request):
- if not type(request) in (list, tuple):
- request = [request]
+ def get_attributes(self, host, ids, names=[]):
+ if not type(ids) in (list, tuple):
+ ids = [ids]
def my_process_results(result):
data = None
@@ -1298,9 +1298,8 @@
if isinstance(result, Exception):
status = result
else:
- status = _AviaryCommon._get_status(result.status)
- if status == "OK" and hasattr(result, "results"):
- data = result.results
+ status = 0
+ data = result
return (status, data)
client = self.client_pool.get_object()
@@ -1309,20 +1308,23 @@
host,
"getAttributes")
+ args = []
+ for id in ids:
+ args.append({"id":id,
+ "names": names})
+
+
res = self._call_sync(self.call_client_retry,
my_process_results,
client,
"getAttributes",
- request)
+ args)
self.client_pool.return_object(client)
return res;
def _get_obj_by_id(self, host, page_size, mode, offset, meth_name):
- if offset:
- assert isinstance(offset, ResourceID)
-
assert mode is None or mode in ("BEFORE", "AFTER")
def my_process_results(result):
@@ -1340,8 +1342,7 @@
client = self.client_pool.get_object()
if offset:
- resID = client.factory.create("ns1:ResourceID")
- offset.set(resID)
+ resID = offset
else:
resID = None
@@ -1359,8 +1360,7 @@
self.client_pool.return_object(client)
return res;
-
- def _get_object(self, host, ids, meth_name, callback=None):
+ def _get_object(self, host, ids, meth_name, no_summary, callback=None):
if callback:
assert callable(callback)
@@ -1373,6 +1373,10 @@
host,
meth_name)
+ if no_summary:
+ client.set_enable_attributes(True)
+ client.set_attributes({"includeSummaries": "false"})
+
def result_tuple(result, host):
data = None
result = self._pretty_result(result, host)
@@ -1380,12 +1384,13 @@
status = result
else:
# no status for the call itself?
- status = "OK"
+ status = 0
data = result
return (status, data)
if callback:
def my_callback(result):
+ client.set_enable_attributes(False)
self.client_pool.return_object(client)
callback(*result_tuple(result, host))
@@ -1398,6 +1403,7 @@
res = self._call_sync(self.call_client_retry, my_process_results,
client, meth_name, ids)
+ client.set_enable_attributes(False)
self.client_pool.return_object(client)
return res;
@@ -1454,6 +1460,8 @@
return list(hosts)
def get_endpoints(self, resource, subtype=""):
+ # Works for a single resource/subtype
+
# This should return a list of objects with id/location
# id location
# [ ( (name,pool,resource,sub_type), [full urls] ) ]
@@ -1468,8 +1476,29 @@
# Make something that looks like locator responses.
endpoints = []
for m in self.mechs:
- endpoints.extend(m.get_endpoints(resource, subtype))
+ endpoints.extend(m.get_endpoints([(resource, subtype)]))
return endpoints
-
+ def get_endpoints_list(self, types):
+ # types is a list of (resource, subtype) tuples
+ # This should return a list of objects with id/location
+ # id location
+ # [ ( (name,pool,resource,sub_type), [full urls] ) ]
+ if len(types) == 0:
+ return []
+
+ if len(types) == 1:
+ return self.get_endpoints(types[0][0],
+ types[0][1])
+
+ if self.locator:
+ endpoints = []
+ for t in types:
+ endpoints.extend(self.get_endpoints(t[0], t[1]))
+
+ else:
+ endpoints = []
+ for m in self.mechs:
+ endpoints.extend(m.get_endpoints(types))
+ return endpoints
Modified: branches/tmckay/sage/rpc-defs/aviary/aviary-collector.xsd
===================================================================
--- branches/tmckay/sage/rpc-defs/aviary/aviary-collector.xsd 2013-02-01 21:13:38 UTC (rev 5669)
+++ branches/tmckay/sage/rpc-defs/aviary/aviary-collector.xsd 2013-02-04 18:05:45 UTC (rev 5670)
@@ -164,8 +164,8 @@
<xs:complexType name="AttributeResponse">
<xs:sequence>
<xs:element name="id" type="mrg:ResourceID"/>
- <xs:element name="attrs" type="mrg:Attribute" minOccurs="0" maxOccurs="unbounded"/>
<xs:element name="status" type="mrg:Status"/>
+ <xs:element name="ad" type="mrg:Attributes" minOccurs="0"/>
</xs:sequence>
</xs:complexType>
<xs:element name="GetAttributes">
@@ -180,7 +180,6 @@
<xs:complexType>
<xs:sequence>
<xs:element name="results" type="coll:AttributeResponse" minOccurs="0" maxOccurs="unbounded"/>
- <xs:element name="status" type="mrg:Status"/>
</xs:sequence>
</xs:complexType>
</xs:element>
11 years, 2 months
r5669 - branches/elephant/cumin/python/cumin/gridhadoop
by croberts@fedoraproject.org
Author: croberts
Date: 2013-02-01 21:13:38 +0000 (Fri, 01 Feb 2013)
New Revision: 5669
Added:
branches/elephant/cumin/python/cumin/gridhadoop/scheduler.py
Modified:
branches/elephant/cumin/python/cumin/gridhadoop/datanode.py
branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py
branches/elephant/cumin/python/cumin/gridhadoop/main.py
branches/elephant/cumin/python/cumin/gridhadoop/namenode.py
branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py
Log:
first pass at new UI page "schedulers" for grid-hadoop. In addition to being necessary to let us work with a chosen scheduler in a multi schedd environment, it will be more congruent with the regular grid interface and will make other top-level tabs possible.
Modified: branches/elephant/cumin/python/cumin/gridhadoop/datanode.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/datanode.py 2013-01-30 17:56:34 UTC (rev 5668)
+++ branches/elephant/cumin/python/cumin/gridhadoop/datanode.py 2013-02-01 21:13:38 UTC (rev 5669)
@@ -49,7 +49,7 @@
def __init__(self, app, name, cls):
super(DataNodeTable, self).__init__(app, name, cls)
- self.id_col = self.DataNodeColumn(app, "idcol", cls.Id, cls.Id, ".datanodeframe")
+ self.id_col = self.DataNodeColumn(app, "idcol", cls.Id, cls.Id, "datanodeframe")
self.sub_col = SubmittedColumn(app, "sub", cls.Submitted)
self.state_col = ObjectTableColumn(app, "statecol", cls.State)
self.uptime_col = UptimeColumn(app, "uptimecol", cls.Uptime)
@@ -66,9 +66,9 @@
id = unescape_entity(record[self.id_field.index])
#need to make this work for user/grid-hadoop path too somehow
# this is way less than ideal
- frame = self.page.page_widgets_by_path[self.path.rsplit(".", 11)[0] + self.frame_path]
- hadoop_schedulers = self.app.remote.get_hosts("SCHEDULER", "HADOOP")
- return frame.get_href(session, id, hadoop_schedulers[0])
+ frame = self.frame.children_by_name[self.frame_path]
+
+ return frame.get_href(session, id, frame.parent.host.get(session))
class DataNodeDelete(ObjectSelectorTask):
def __init__(self, app, selector, name):
@@ -144,7 +144,7 @@
self.form.id.set(form_session, obj._id)
## TODO find a better way to do this
- self.form.nn_id.set(form_session, self.app.gridhadoop.frame.children_by_name["gridhadoop.namenodeframe"].id.get(session))
+ self.form.nn_id.set(form_session, self.app.gridhadoop.frame.children_by_name["gridhadoop.ghschedframe"].children_by_name["gridhadoop.ghschedframe.namenodeframe"].id.get(session))
self.form.return_url.set(form_session, session.marshal())
self.form.show(form_session)
Modified: branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py 2013-01-30 17:56:34 UTC (rev 5668)
+++ branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py 2013-02-01 21:13:38 UTC (rev 5669)
@@ -69,12 +69,8 @@
class JobTrackerColumn(ObjectLinkColumn):
def render_cell_href(self, session, record):
id = unescape_entity(record[self.id_field.index])
- #need to make this work for user/grid-hadoop path too somehow
- frame = self.page.page_widgets_by_path[self.path.rsplit(".", 7)[0] + self.frame_path]
-
- #if isinstance(frame, TagObjectFrame):
- hadoop_schedulers = self.app.remote.get_hosts("SCHEDULER", "HADOOP")
- return frame.get_href(session, id, hadoop_schedulers[0])
+ frame = self.frame.children_by_name[self.frame.name + self.frame_path]
+ return frame.get_href(session, id, frame.parent.host.get(session))
class JobTrackerDelete(ObjectSelectorTask):
def __init__(self, app, selector, name):
Modified: branches/elephant/cumin/python/cumin/gridhadoop/main.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/main.py 2013-01-30 17:56:34 UTC (rev 5668)
+++ branches/elephant/cumin/python/cumin/gridhadoop/main.py 2013-02-01 21:13:38 UTC (rev 5669)
@@ -14,6 +14,7 @@
from datanode import DataNodeSelector
from jobtracker import JobTrackerSelector
from tasktracker import TaskTrackerSelector
+from scheduler import HadoopSchedulerSelector, HadoopSchedulerFrame
strings = StringCatalog(__file__)
@@ -37,13 +38,16 @@
self.view = GridHadoopView(app, name + ".view", check_owner)
self.add_mode(self.view)
+
+ self.scheduler_frame = HadoopSchedulerFrame(app, name + ".ghschedframe", check_owner)
+ self.add_mode(self.scheduler_frame)
- self.name_node_frame = NameNodeFrame(app, name + ".namenodeframe")
- self.add_mode(self.name_node_frame)
+# self.name_node_frame = NameNodeFrame(app, name + ".namenodeframe")
+# self.add_mode(self.name_node_frame)
+#
+# self.job_tracker_frame = JobTrackerFrame(app, name + ".jobtrackerframe")
+# self.add_mode(self.job_tracker_frame)
- self.job_tracker_frame = JobTrackerFrame(app, name + ".jobtrackerframe")
- self.add_mode(self.job_tracker_frame)
-
def render_title(self, session):
return "Grid-hadoop"
@@ -57,5 +61,6 @@
self.tabs = TabbedModeSet(app, "childtabs")
self.add_child(self.tabs)
- self.tabs.add_tab(NameNodeSelector(app, name + ".namenodes", check_owner))
- self.tabs.add_tab(JobTrackerSelector(app, name + ".jobtrackers", check_owner))
+ #self.tabs.add_tab(NameNodeSelector(app, name + ".namenodes", check_owner))
+ #self.tabs.add_tab(JobTrackerSelector(app, name + ".jobtrackers", check_owner))
+ self.tabs.add_tab(HadoopSchedulerSelector(app, name + ".hscheds"))
Modified: branches/elephant/cumin/python/cumin/gridhadoop/namenode.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/namenode.py 2013-01-30 17:56:34 UTC (rev 5668)
+++ branches/elephant/cumin/python/cumin/gridhadoop/namenode.py 2013-02-01 21:13:38 UTC (rev 5669)
@@ -68,12 +68,8 @@
class NameNodeColumn(ObjectLinkColumn):
def render_cell_href(self, session, record):
id = unescape_entity(record[self.id_field.index])
- #need to make this work for user/grid-hadoop path too somehow
- # this is way less than ideal
- frame = self.page.page_widgets_by_path[self.path.rsplit(".", 7)[0] + self.frame_path]
-
- hadoop_schedulers = self.app.remote.get_hosts("SCHEDULER", "HADOOP")
- return frame.get_href(session, id, hadoop_schedulers[0])
+ frame = self.frame.children_by_name[self.frame.name + self.frame_path]
+ return frame.get_href(session, id, frame.parent.host.get(session))
class NameNodeCreate(Task):
def __init__(self, app, name):
Added: branches/elephant/cumin/python/cumin/gridhadoop/scheduler.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/scheduler.py (rev 0)
+++ branches/elephant/cumin/python/cumin/gridhadoop/scheduler.py 2013-02-01 21:13:38 UTC (rev 5669)
@@ -0,0 +1,88 @@
+from cumin.objectframe import ObjectFrame
+
+from hadoop import *
+from hadoopobjectframe import *
+
+from jobtracker import JobTrackerFrame, JobTrackerSelector
+from namenode import NameNodeFrame, NameNodeSelector
+
+class HadoopSchedulerAdapter(HadoopAdapter):
+ def get_sage_results(self, values):
+ owner_filter = self.get_owner_filter(values['session'])
+ results = self.app.remote.get_hosts("SCHEDULER", "HADOOP")
+ return results
+
+ def do_get_data(self, values):
+ data = self.get_sage_results(values)
+ if data is None or len(data) == 0:
+ return {}
+ return data
+
+ def process_record(self, key, record):
+ field_data = list()
+ for column in self.columns:
+ val = record
+ field_data.append(val)
+ return field_data
+
+class HadoopSchedulerSelector(ObjectSelector):
+ def __init__(self, app, name, check_owner=False):
+ cls = app.model.com_redhat_cumin_grid_hadoop.Scheduler
+
+ super(HadoopSchedulerSelector, self).__init__(app, name, cls)
+
+ self.add_search_filter(self.table.id_col)
+ self.table.adapter = HadoopSchedulerAdapter(app, cls)
+
+ self.enable_csv_export()
+
+ def create_table(self, app, name, cls):
+ return HadoopSchedulerTable(app, name, cls)
+
+ def render_title(self, session):
+ return "Schedulers"
+
+class HadoopSchedulerTable(ObjectQmfSelectorTable):
+ def __init__(self, app, name, cls):
+ super(HadoopSchedulerTable, self).__init__(app, name, cls)
+
+ self.id_col = self.SchedNameColumn(app, "namecol", cls.Name, cls.Name, ".ghschedframe")
+ self.add_column(self.id_col)
+
+
+ class SchedNameColumn(ObjectLinkColumn):
+ def render_cell_href(self, session, record):
+ id = unescape_entity(record[self.id_field.index])
+ #need to make this work for user/grid-hadoop path too somehow
+ # this is way less than ideal
+ frame = self.page.page_widgets_by_path[self.path.rsplit(".", 7)[0] + self.frame_path]
+ return frame.get_href(session, id, id)
+
+class HadoopSchedulerFrame(HadoopObjectFrame):
+ def __init__(self, app, name, check_owner):
+ super(HadoopSchedulerFrame, self).__init__(app, name)
+
+ self.name_node_frame = NameNodeFrame(app, name + ".namenodeframe")
+ self.add_mode(self.name_node_frame)
+
+ self.job_tracker_frame = JobTrackerFrame(app, name + ".jobtrackerframe")
+ self.add_mode(self.job_tracker_frame)
+
+ self.view.add_tab(NameNodeSelector(app, name + ".namenodes", check_owner))
+ self.view.add_tab(JobTrackerSelector(app, name + ".jobtrackers", check_owner))
+
+ def get_object(self, session, id):
+ class Object(object):
+ pass
+ fake = Object()
+ fake.name = "fake scheduler"
+ fake.id = id
+ return fake
+
+ def get_title(self, session):
+ retval = ""
+ try:
+ retval = "Hadoop scheduler:'%s' (%s)" % (xml_escape(self.object.get(session).name), xml_escape(self.object.get(session).id))
+ except Exception, e:
+ pass
+ return retval
Property changes on: branches/elephant/cumin/python/cumin/gridhadoop/scheduler.py
___________________________________________________________________
Added: svn:mime-type
+ text/plain
Modified: branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py 2013-01-30 17:56:34 UTC (rev 5668)
+++ branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py 2013-02-01 21:13:38 UTC (rev 5669)
@@ -49,7 +49,7 @@
def __init__(self, app, name, cls):
super(TaskTrackerTable, self).__init__(app, name, cls)
- self.id_col = self.TaskTrackerColumn(app, "idcol", cls.Id, cls.Id, ".tasktrackerframe")
+ self.id_col = self.TaskTrackerColumn(app, "idcol", cls.Id, cls.Id, "tasktrackerframe")
self.sub_col = SubmittedColumn(app, "sub", cls.Submitted)
self.state_col = ObjectTableColumn(app, "statecol", cls.State)
self.uptime_col = UptimeColumn(app, "uptimecol", cls.Uptime)
@@ -64,11 +64,8 @@
class TaskTrackerColumn(ObjectLinkColumn):
def render_cell_href(self, session, record):
id = unescape_entity(record[self.id_field.index])
- #need to make this work for user/grid-hadoop path too somehow
- # this is way less than ideal
- frame = self.page.page_widgets_by_path[self.path.rsplit(".", 11)[0] + self.frame_path]
- hadoop_schedulers = self.app.remote.get_hosts("SCHEDULER", "HADOOP")
- return frame.get_href(session, id, hadoop_schedulers[0])
+ frame = self.frame.children_by_name[self.frame_path]
+ return frame.get_href(session, id, frame.parent.host.get(session))
class TaskTrackerDelete(ObjectSelectorTask):
def __init__(self, app, selector, name):
@@ -138,7 +135,7 @@
self.form.id.set(form_session, obj._id)
## TODO find a better way to do this
- self.form.jt_id.set(form_session, self.app.gridhadoop.frame.children_by_name["gridhadoop.jobtrackerframe"].id.get(session))
+ self.form.jt_id.set(form_session, self.app.gridhadoop.frame.children_by_name["gridhadoop.ghschedframe"].children_by_name["gridhadoop.ghschedframe.jobtrackerframe"].id.get(session))
self.form.return_url.set(form_session, session.marshal())
self.form.show(form_session)
11 years, 3 months