r5670 - in branches/tmckay: mint/python/mint mint/python/mint/aviary sage/python/sage/aviary sage/rpc-defs/aviary

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>