Author: tmckay
Date: 2013-01-22 19:56:02 +0000 (Tue, 22 Jan 2013)
New Revision: 5663
Modified:
branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
branches/tmckay/sage/rpc-defs/aviary/aviary-common.xsd
Log:
Add routines for collector types (but do they work?)
Modified: branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- branches/tmckay/sage/python/sage/aviary/aviaryoperations.py 2013-01-18 23:19:32 UTC
(rev 5662)
+++ branches/tmckay/sage/python/sage/aviary/aviaryoperations.py 2013-01-22 19:56:02 UTC
(rev 5663)
@@ -74,9 +74,10 @@
class BaseServerList(object):
# Nice, friendly strings for error messages on lookup
- _nice = {"JOB": "job service",
- "QUERY_SERVER": "query service",
- "HADOOP": "hadoop service"}
+ _nice = {("SCHEDULER","JOB"): "job service",
+ ("CUSTOM","QUERY_SERVER"): "query service",
+ ("SCHEDULER","HADOOP"): "hadoop service",
+ ("COLLECTOR",""): "collector service"}
def __init__(self, resource, subtype):
self.servers = None
@@ -84,7 +85,7 @@
self.resource = resource
self.subtype = subtype
try:
- self.nice = BaseServerList._nice[subtype]
+ self.nice = BaseServerList._nice[(resource,subtype)]
except:
self.nice = subtype
@@ -1213,6 +1214,38 @@
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(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)
+
+ self.res_id = res_id
+ self.name_list = name_list
+
class _AviaryCollectorMethods(_AviaryCommon):
def __init__(self, locator, transports, datadir, collector_servers):
@@ -1228,10 +1261,147 @@
"COLLECTOR",
"")
-
# 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_negotiator(self, host, ids):
+ return self._get_object(host, ids, "getNegotiator")
+
+ def get_scheduler(self, host, ids):
+ return self._get_object(host, ids, "getScheduler")
+
+ def get_submitter(self, host, ids):
+ return self._get_object(host, ids, "getSubmitter")
+
+ def get_master(self, host, ids):
+ return self._get_object(host, ids, "getMaster")
+
+ def get_slot(self, host, ids):
+ return self._get_object(host, ids, "getSlot")
+
+ def get_master_id(self, host, page_size, mode=None, offset=None):
+ return self._get_obj_by_id(host, page_size, mode, offset,
"getMasterID")
+
+ 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 my_process_results(result):
+ data = None
+ result = self._pretty_result(result, host)
+ if isinstance(result, Exception):
+ status = result
+ else:
+ status = _AviaryCommon._get_status(result.status)
+ if status == "OK" and hasattr(result, "results"):
+ data = result.results
+ return (status, data)
+
+ client = self.client_pool.get_object()
+ self._setup_client(client,
+ self.servers,
+ host,
+ "getAttributes")
+
+ res = self._call_sync(self.call_client_retry,
+ my_process_results,
+ client,
+ "getAttributes",
+ request)
+
+ 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):
+ # Fix up the exception message if necessary
+ result = self._pretty_result(result, host)
+ if isinstance(result, Exception):
+ status = result
+ data = None
+ else:
+ status = 0
+ if not hasattr(result, "results"):
+ result.results = []
+ data = result
+ return (status, data)
+
+ client = self.client_pool.get_object()
+ if offset:
+ resID = client.factory.create("ns1:ResourceID")
+ offset.set(resID)
+ else:
+ resID = None
+
+ self._setup_client(client,
+ self.servers,
+ host,
+ "getMasterID")
+
+ res = self._call_sync(self.call_client_retry,
+ my_process_results,
+ client,
+ "getMasterID",
+ page_size, mode, resID)
+
+ self.client_pool.return_object(query_client)
+ return res;
+
+
+ def _get_object(self, host, ids, meth_name, callback=None):
+ if callback:
+ assert callable(callback)
+
+ if type(ids) not in (list, tuple):
+ ids = (ids)
+
+ client = self.client_pool.get_object()
+ self._setup_client(client,
+ self.servers,
+ host,
+ meth_name)
+
+ def result_tuple(result, host):
+ data = None
+ result = self._pretty_result(result, host)
+ if isinstance(result, Exception):
+ status = result
+ else:
+ # no status for the call itself?
+ status = "OK"
+ if hasattr(result, "results"):
+ data = result.results
+ return (status, data)
+
+ if callback:
+ def my_callback(result):
+ self.client_pool.return_object(client)
+ callback(*result_tuple(result, host))
+
+ t = CallThread(self.call_client_retry, my_callback,
+ client, meth_name, ids)
+ t.start()
+ else:
+ def my_process_results(result):
+ return result_tuple(result, host)
+
+ res = self._call_sync(self.call_client_retry, my_process_results,
+ client, meth_name, ids)
+ self.client_pool.return_object(client)
+ return res;
+
class AviaryOperations(Catalog):
def __init__(self, name, datadir,
locator_uri="",
Modified: branches/tmckay/sage/rpc-defs/aviary/aviary-common.xsd
===================================================================
--- branches/tmckay/sage/rpc-defs/aviary/aviary-common.xsd 2013-01-18 23:19:32 UTC (rev
5662)
+++ branches/tmckay/sage/rpc-defs/aviary/aviary-common.xsd 2013-01-22 19:56:02 UTC (rev
5663)
@@ -48,8 +48,9 @@
<xs:complexType name="Attribute">
<xs:sequence>
<xs:element name="name" type="xs:string"/>
- <xs:element name="type" type="tns:AttributeType"/>
+ <xs:element name="type" type="tns:AttributeType"
minOccurs="0"/>
<xs:element name="value" type="xs:string"/>
+ <xs:element name="desc" type="xs:string"
minOccurs="0"/>
</xs:sequence>
</xs:complexType>
<!-- might just want to use the following just for returning data like in
getJobDetails -->
@@ -62,6 +63,8 @@
<xs:restriction base="xs:string">
<xs:enumeration value="LINUX"/>
<xs:enumeration value="WINDOWS"/>
+ <xs:enumeration value="OSX"/>
+ <xs:enumeration value="OTHER"/>
<!-- can expand later -->
</xs:restriction>
</xs:simpleType>
@@ -69,6 +72,7 @@
<xs:restriction base="xs:string">
<xs:enumeration value="INTEL"/>
<xs:enumeration value="X86_64"/>
+ <xs:enumeration value="OTHER"/>
<!-- can expand later -->
</xs:restriction>
</xs:simpleType>
@@ -105,8 +109,12 @@
<xs:sequence>
<xs:element name="name" type="xs:string"
minOccurs="0"/>
<xs:element name="owner" type="xs:string"
minOccurs="0"/>
- <xs:element name="qdate" type="xs:int"
minOccurs="0"/>
+ <xs:element name="qdate" type="xs:int"
minOccurs="0"/>
</xs:sequence>
+ <!-- where this submission was created in the condor space -->
+ <!-- TODO: tactical hiding for backward compat, revisit -->
+ <xs:attribute name="pool" type="xs:string"/>
+ <xs:attribute name="scheduler" type="xs:string"/>
</xs:complexType>
<xs:complexType name="JobID">
<xs:sequence>
@@ -150,7 +158,7 @@
<xs:element name="removed" type="xs:string"
minOccurs="0"/>
</xs:sequence>
</xs:complexType>
- <xs:complexType name="SubmissionSummary">
+ <xs:complexType name="SubmissionSummary">
<xs:sequence>
<xs:element name="id" type="tns:SubmissionID"/>
<xs:element name="status" type="tns:Status"/>
@@ -197,24 +205,151 @@
</xs:simpleType>
<xs:complexType name="ResourceID">
<xs:sequence>
- <xs:element name="resource" type="tns:ResourceType"/>
+ <xs:element name="resource" type="tns:ResourceType"
default="COLLECTOR"/>
<xs:element name="pool" type="xs:string"
minOccurs="0"/>
<xs:element name="name" type="xs:string"
minOccurs="0"/>
+ <xs:element name="address" type="xs:string"
minOccurs="0"/>
<xs:element name="sub_type" type="xs:string"
minOccurs="0"/>
+ <xs:element name="birthdate" type="xs:int"
minOccurs="0"/>
</xs:sequence>
</xs:complexType>
<xs:complexType name="ResourceLocation">
<xs:sequence>
<xs:element name="id" type="tns:ResourceID"/>
- <xs:element name="location" type="xs:anyURI"
minOccurs="0" maxOccurs="unbounded"/>
+ <xs:element name="location" type="xs:anyURI"
minOccurs="0" maxOccurs="unbounded"/>
</xs:sequence>
</xs:complexType>
<xs:simpleType name="ScanMode">
- <!-- a set of options for selecting how we scan a contiguous set of ids
-->
- <!-- only meaningful when used with an offset -->
<xs:restriction base="xs:string">
- <xs:enumeration value="AFTER"/>
- <xs:enumeration value="BEFORE"/>
- </xs:restriction>
- </xs:simpleType>
+ <xs:enumeration value="AFTER"/>
+ <xs:enumeration value="BEFORE"/>
+ </xs:restriction>
+ <!-- a set of options for selecting how we scan a contiguous set of ids -->
+ <!-- only meaningful when used with an offset -->
+ </xs:simpleType>
+ <xs:complexType name="CollectorSummary">
+ <xs:sequence>
+ <xs:element name="running_jobs" type="xs:int"/>
+ <xs:element name="idle_jobs" type="xs:int"/>
+ <xs:element name="total_hosts" type="xs:int"/>
+ <xs:element name="claimed_hosts" type="xs:int"/>
+ <xs:element name="unclaimed_hosts" type="xs:int"/>
+ <xs:element name="owner_hosts" type="xs:int"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="Collector">
+ <xs:sequence>
+ <xs:element name="id" type="tns:ResourceID"/>
+ <xs:element name="status" type="tns:Status"/>
+ <xs:element name="summary" type="tns:CollectorSummary"
minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="MasterSummary">
+ <xs:sequence>
+ <xs:element name="arch" type="tns:ArchType"/>
+ <xs:element name="os" type="tns:OSType"/>
+ <xs:element name="real_uid" type="xs:int"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="Master">
+ <xs:sequence>
+ <xs:element name="id" type="tns:ResourceID"/>
+ <xs:element name="status" type="tns:Status"/>
+ <xs:element name="summary" type="tns:MasterSummary"
minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="NegotiatorSummary">
+ <xs:sequence>
+ <xs:element name="match_rate" type="xs:double"/>
+ <xs:element name="matches" type="xs:int"/>
+ <xs:element name="duration" type="xs:int"/>
+ <xs:element name="schedulers" type="xs:int"/>
+ <xs:element name="active_submitters" type="xs:int"/>
+ <xs:element name="idle_jobs" type="xs:int"/>
+ <xs:element name="jobs_considered" type="xs:int"/>
+ <xs:element name="rejections" type="xs:int"/>
+ <xs:element name="total_slots" type="xs:int"/>
+ <xs:element name="candidate_slots" type="xs:int"/>
+ <xs:element name="trimmed_slots" type="xs:int"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="Negotiator">
+ <xs:sequence>
+ <xs:element name="id" type="tns:ResourceID"/>
+ <xs:element name="status" type="tns:Status"/>
+ <xs:element name="summary" type="tns:NegotiatorSummary"
minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="SchedulerSummary">
+ <xs:sequence>
+ <xs:element name="queue_created" type="xs:dateTime"/>
+ <xs:element name="max_jobs_running" type="xs:int"/>
+ <xs:element name="users" type="xs:int"/>
+ <xs:element name="ads" type="xs:int"/>
+ <xs:element name="running" type="xs:int"/>
+ <xs:element name="held" type="xs:int"/>
+ <xs:element name="idle" type="xs:int"/>
+ <xs:element name="removed" type="xs:int"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="Scheduler">
+ <xs:sequence>
+ <xs:element name="id" type="tns:ResourceID"/>
+ <xs:element name="status" type="tns:Status"/>
+ <xs:element name="summary" type="tns:SchedulerSummary"
minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:simpleType name="SlotType">
+ <xs:restriction base="xs:string">
+ <xs:enumeration value="STATIC"/>
+ <xs:enumeration value="PARTITIONABLE"/>
+ <xs:enumeration value="DYNAMIC"/>
+ </xs:restriction>
+ </xs:simpleType>
+ <xs:complexType name="SlotSummary">
+ <xs:sequence>
+ <xs:element name="arch" type="tns:ArchType"/>
+ <xs:element name="os" type="tns:OSType"/>
+ <xs:element name="activity" type="xs:string"/>
+ <xs:element name="state" type="xs:string"/>
+ <xs:element name="cpus" type="xs:int"/>
+ <xs:element name="disk" type="xs:int"/>
+ <xs:element name="memory" type="xs:int"/>
+ <xs:element name="swap" type="xs:int"/>
+ <xs:element name="mips" type="xs:int"/>
+ <xs:element name="load_avg" type="xs:double"/>
+ <xs:element name="start" type="xs:string"/>
+ <xs:element name="domain" type="xs:string"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="Slot">
+ <xs:sequence>
+ <xs:element name="id" type="tns:ResourceID"/>
+ <xs:element name="status" type="tns:Status"/>
+ <xs:element name="slot_type" type="tns:SlotType"
default="STATIC" minOccurs="0"/>
+ <xs:element name="summary" type="tns:SlotSummary"
minOccurs="0"/>
+ <xs:element name="dynamic_slots" type="tns:Slot"
minOccurs="0" maxOccurs="unbounded"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="SubmitterID">
+ <xs:sequence>
+ <xs:element name="name" type="xs:string"
minOccurs="0"/>
+ <xs:element name="machine" type="xs:string"
minOccurs="0"/>
+ <xs:element name="scheduler" type="xs:string"
minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="SubmitterSummary">
+ <xs:sequence>
+ <xs:element name="running" type="xs:int"/>
+ <xs:element name="held" type="xs:int"/>
+ <xs:element name="idle" type="xs:int"/>
+ </xs:sequence>
+ </xs:complexType>
+ <xs:complexType name="Submitter">
+ <xs:sequence>
+ <xs:element name="id" type="tns:SubmitterID"/>
+ <xs:element name="status" type="tns:Status"/>
+ <xs:element name="summary" type="tns:SubmitterSummary"
minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
</xs:schema>