Author: tmckay
Date: 2011-08-26 14:18:11 +0000 (Fri, 26 Aug 2011)
New Revision: 4936
Modified:
trunk/cumin/python/cumin/grid/job.py
trunk/cumin/python/cumin/model.py
trunk/cumin/python/cumin/util.py
trunk/sage/python/sage/aviary/aviaryoperations.py
trunk/sage/python/sage/qmf/qmfoperations.py
Log:
Add get_job_summaries in aviaryoperations, pass necessary objects to
methods in aviaryoperations to fully qualify all id fields in arguments
to aviary
Modified: trunk/cumin/python/cumin/grid/job.py
===================================================================
--- trunk/cumin/python/cumin/grid/job.py 2011-08-17 18:17:46 UTC (rev 4935)
+++ trunk/cumin/python/cumin/grid/job.py 2011-08-26 14:18:11 UTC (rev 4936)
@@ -67,6 +67,17 @@
cls = self.app.model.com_redhat_grid.JobServer
return cls.get_object(session.cursor, _id=submission._jobserverRef_id)
+ def get_trifecta(self, session, id):
+ # return submission, job_server, and scheduler too!
+ submission = self.get_submission(session, id)
+
+ cls = self.app.model.com_redhat_grid.JobServer
+ js = cls.get_object(session.cursor, _id=submission._jobserverRef_id)
+
+ cls = self.app.model.com_redhat_grid.Scheduler
+ sched = cls.get_object(session.cursor, _id=js._schedulerRef_id)
+ return (submission, js, sched)
+
def get_scheduler(self, session, id):
job_server = self.get_job_server(session, id)
@@ -121,9 +132,13 @@
def get_qmf_results(self, values):
session = values['session']
submission = self.submission.get(session)
+ try:
+ js = self.app.model.get_jobserver_from_submission(session,
+ submission).Machine
+ except:
+ js = ""
+ return self.app.model.get_submission_job_summaries(submission, js)
- return self.app.model.get_submission_job_summaries(submission)
-
def do_get_data(self, values):
results = self.get_qmf_results(values)
summaries = results.data
@@ -285,10 +300,10 @@
self.form = JobObjectSelectorTaskForm(app, self.name, self, verb)
self.method = None
- def do_invoke(self, invoc, job_id, scheduler, reason):
+ def do_invoke(self, invoc, job_id, scheduler, reason, submission):
assert self.method
- results = self.method(scheduler, job_id, reason)
+ results = self.method(scheduler, job_id, reason, submission)
if results.error:
raise results.error
@@ -385,12 +400,12 @@
selection = self.selection.get(session)
reason = self.get_reason(session, self.verb)
- scheduler = self.get_scheduler(session)
+ submission, scheduler = self.get_submission_sched(session)
- self.task.invoke(session, selection, scheduler, reason)
+ self.task.invoke(session, selection, scheduler, reason, submission)
self.task.exit_with_redirect(session)
- def get_scheduler(self, session):
+ def get_submission_sched(self, session):
submission_id = self.submission_id.get(session)
cls = self.app.model.com_redhat_grid.Submission
submission = cls.get_object_by_id(session.cursor, submission_id)
@@ -399,7 +414,8 @@
job_server = cls.get_object(session.cursor, _id=submission._jobserverRef_id)
cls = self.app.model.com_redhat_grid.Scheduler
- return cls.get_object(session.cursor, _id=job_server._schedulerRef_id)
+ sched = cls.get_object(session.cursor, _id=job_server._schedulerRef_id)
+ return (submission, sched)
def render_content(self, session, *args):
content = super(JobObjectSelectorTaskForm, self).render_content(session, *args)
@@ -428,9 +444,11 @@
if not ad_list and not error:
ad_list = list()
id = self.frame.id.get(session)
- job_server = self.frame.get_job_server(session, id)
+ submission, job_server, sched = self.frame.get_trifecta(session, id)
job_id = self.frame.job_id.get(session)
- results = self.app.remote.get_job_ad(job_server, job_id, {'JobAd': {}})
+ results = self.app.remote.get_job_ad(job_server, job_id,
+ sched.Name, submission,
+ default={'JobAd': {}})
error = results.error
self.qmf_error.set(session, error)
ads = results.data['JobAd']
@@ -514,9 +532,11 @@
if not ad_list and not error:
ad_list = list()
id = self.frame.id.get(session)
- job_server = self.frame.get_job_server(session, id)
+ submission, job_server, sched = self.frame.get_trifecta(session, id)
job_id = self.frame.job_id.get(session)
- results = self.app.remote.get_job_ad(job_server, job_id, {'JobAd': {}})
+ results = self.app.remote.get_job_ad(job_server, job_id,
+ sched.Name, submission,
+ default={'JobAd': {}})
error = results.error
self.qmf_error.set(session, error)
ads = results.data['JobAd']
@@ -844,10 +864,13 @@
id = self.frame.id.get(session)
scheduler = self.frame.get_scheduler(session, id)
job_id = self.frame.job_id.get(session)
+ submission = self.frame.get_submission(session, id)
task = self.frame.set_ad_task
for field in just_ads:
- task.invoke(session, scheduler, job_id, field, str(just_ads[field]), ads[field]["error"])
+ task.invoke(session, scheduler, job_id, field,
+ str(just_ads[field]), ads[field]["error"],
+ submission)
self.process_cancel(session)
class OutputFile(Widget):
@@ -862,12 +885,16 @@
def render_content(self, session):
id = self.frame.id.get(session)
- job_server = self.frame.get_job_server(session, id)
+ submission, job_server, scheduler = self.frame.get_trifecta(session, id)
job_id = self.frame.job_id.get(session)
state, file, start, end = self.get_file_args(session)
+
if file:
result = self.app.remote.fetch_job_data(job_server, job_id, state,
- file, start, end, {'Data': ""})
+ file, start, end,
+ scheduler.Name,
+ submission,
+ default={'Data': ""})
if result.error:
return result.status
return escape_entity(result.data['Data'])
@@ -921,7 +948,8 @@
def add_path(path, filename):
# If filename does not begin with an absolute
# path, prepend the path value to filename
- if not filename.startswith("/"):
+ if filename is not None and \
+ not filename.startswith("/"):
if not path.endswith("/"):
path += "/"
filename = path + filename
@@ -1085,8 +1113,8 @@
scheduler = self.task.frame.get_scheduler(session, id)
reason = self.get_reason(session, self.verb)
job_id = self.job_id.get(session)
-
- self.task.invoke(session, scheduler, job_id, reason)
+ submission = self.task.frame.get_submission(session, id)
+ self.task.invoke(session, scheduler, job_id, reason, submission)
self.task.exit_with_redirect(session)
class ReasonField(StringField):
@@ -1103,15 +1131,16 @@
job_id = self.frame.job_id.get(osession)
self.form.job_id.set(session, job_id)
- def do_invoke(self, invoc, scheduler, job_id, reason):
+ def do_invoke(self, invoc, scheduler, job_id, reason, submission):
raise Exception("oops, not implemented!")
class JobHold(JobAction):
def __init__(self, app, frame):
super(JobHold, self).__init__(app, frame, "held")
- def do_invoke(self, invoc, scheduler, job_id, reason):
- self.app.remote.hold_job(scheduler, job_id, reason, invoc.make_callback())
+ def do_invoke(self, invoc, scheduler, job_id, reason, submission):
+ self.app.remote.hold_job(scheduler, job_id, reason, submission,
+ callback=invoc.make_callback())
def get_title(self, session):
return "Hold Job"
@@ -1120,8 +1149,9 @@
def __init__(self, app, frame):
super(JobRelease, self).__init__(app, frame, "released")
- def do_invoke(self, invoc, scheduler, job_id, reason):
- self.app.remote.release_job(scheduler, job_id, reason, invoc.make_callback())
+ def do_invoke(self, invoc, scheduler, job_id, reason, submission):
+ self.app.remote.release_job(scheduler, job_id, reason, submission,
+ callback=invoc.make_callback())
def get_title(self, session):
return "Release Job"
@@ -1130,8 +1160,9 @@
def __init__(self, app, frame):
super(JobRemove, self).__init__(app, frame, "removed")
- def do_invoke(self, invoc, scheduler, job_id, reason):
- self.app.remote.remove_job(scheduler, job_id, reason, invoc.make_callback())
+ def do_invoke(self, invoc, scheduler, job_id, reason, submission):
+ self.app.remote.remove_job(scheduler, job_id, reason, submission,
+ callback=invoc.make_callback())
def get_title(self, session):
return "Remove Job"
@@ -1161,12 +1192,14 @@
def get_description(self, session):
return "Edit Ad"
- def do_invoke(self, invoc, scheduler, job_id, name, value, error):
+ def do_invoke(self, invoc, scheduler, job_id, name, value, error, submission):
# Don't make the call, but we want to see the banner
# so invoke the callback directly with the error as status
if error:
invoc.make_callback()(name + ": " + str(error), None)
else:
- self.app.remote.set_job_attribute(scheduler, job_id, name, value,
- invoc.make_callback())
+ self.app.remote.set_job_attribute(scheduler,
+ job_id, name, value,
+ invoc.make_callback(),
+ submission)
Modified: trunk/cumin/python/cumin/model.py
===================================================================
--- trunk/cumin/python/cumin/model.py 2011-08-17 18:17:46 UTC (rev 4935)
+++ trunk/cumin/python/cumin/model.py 2011-08-26 14:18:11 UTC (rev 4936)
@@ -179,7 +179,7 @@
finally:
self.lock.release()
- def get_submission_job_summaries(self, submission):
+ def get_submission_job_summaries(self, submission, machine_name=""):
if not submission:
store = SubmissionJobSummaryStore(self, None)
store.exception = Exception("Missing Submission")
@@ -192,7 +192,7 @@
store = self.job_summaries_by_submission[submission._id]
store.extend_updates()
except KeyError:
- store = SubmissionJobSummaryStore(self, submission)
+ store = SubmissionJobSummaryStore(self, submission, machine_name)
store.start_updates()
self.job_summaries_by_submission[submission._id] = store
@@ -201,6 +201,15 @@
finally:
self.lock.release()
+ def get_jobserver_from_submission(self, session, submission):
+ cls = self.com_redhat_grid.JobServer
+ job_server = cls.get_object(session.cursor, _id=submission._jobserverRef_id)
+ return job_server
+
+ def get_scheduler_from_jobserver(self, session, job_server):
+ cls = self.app.model.com_redhat_grid.Scheduler
+ return cls.get_object(session.cursor, _id=job_server._schedulerRef_id)
+
class CuminProperty(object):
def __init__(self, cls, name):
self.model = cls.model
@@ -757,20 +766,24 @@
super(NegotiatorLimitStore, self).delete()
class SubmissionJobSummaryStore(ObjectStore):
- def __init__(self, model, submission):
+ def __init__(self, model, submission, machine_name=""):
super(SubmissionJobSummaryStore, self).__init__(model)
self.submission = submission
+ self.machine_name = machine_name
def update(self, cursor):
def completion(status, data):
self.status = status
try:
- self.data = data["Jobs"]
+ if data is not None:
+ self.data = data["Jobs"]
except KeyError:
pass
- self.model.app.remote.get_job_summaries(self.submission, completion)
+ self.model.app.remote.get_job_summaries(self.submission,
+ completion,
+ self.machine_name)
def delete(self):
del self.model.job_summaries_by_submission[self.submission._id]
Modified: trunk/cumin/python/cumin/util.py
===================================================================
--- trunk/cumin/python/cumin/util.py 2011-08-17 18:17:46 UTC (rev 4935)
+++ trunk/cumin/python/cumin/util.py 2011-08-26 14:18:11 UTC (rev 4936)
@@ -1,6 +1,7 @@
import sys
import os
import logging
+import string
from crypt import crypt
from datetime import datetime, timedelta
@@ -160,14 +161,20 @@
return dvalue
class JobStatusInfo(object):
- stat_strings = ["Unexpanded", "Idle", "Running", "Removed", "Completed", "Held", "Submission Error"]
+ stat_strings = ["Unexpanded", "Idle", "Running", "Removed", "Completed", "Held", "Submission error"]
stat_colors = ["red", "clear", "green", "black", "blue", "yellow", "red"]
@classmethod
def get_status_string(cls, stat):
- try:
- return cls.stat_strings[stat]
- except:
- return ""
+ # Allow strings to be passed in for flexibility.
+ # Enforcing cumin case should make strings
+ # match stat_strings for expected values
+ if type(stat) in (str, unicode):
+ return string.capitalize(stat)
+ else:
+ try:
+ return cls.stat_strings[stat]
+ except:
+ return "Unrecognized"
@classmethod
def get_status_color(cls, stat):
Modified: trunk/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- trunk/sage/python/sage/aviary/aviaryoperations.py 2011-08-17 18:17:46 UTC (rev 4935)
+++ trunk/sage/python/sage/aviary/aviaryoperations.py 2011-08-26 14:18:11 UTC (rev 4936)
@@ -5,12 +5,13 @@
import urllib2
import socket
import string
+import time
from suds import *
from suds.client import Client
from sage.util import CallSync, CallThread, ObjectPool, host_list
+from datetime import datetime
-
log = logging.getLogger("sage.aviary")
#f = open("./suds.client.log", 'a+')
@@ -41,19 +42,17 @@
- Add logging to Aviary
-- See if we can fill in the scheduler/pool information.
-What is cumin passing? (the scheduler object has this stuff,
-but what about hierarchical collectors?)
+- Test pool/schedd/stuff with hierarchical collectors
-- get the submission objects passed down if possible
-for job control functions. This would be extra stuff for QMF, but helpful with
-Aviary.
-
- Add a summary comment at the top like QMF opreations
- can we use default/timeout with suds?
-yes, there is at least a timeout that can be set once,
-it might be possible to change the timeout per call.
+looks like the way to do this is through the Transport object that
+is set in the client. The timeout is set on the Transport itself,
+and the Transport can be changed on the client with set_options.
+Alternatively, we could retain a reference to the Transport in
+the clients that we pool so that we can call set_options on the
+transport.
'''
class AviaryOperations(object):
@@ -62,7 +61,7 @@
self.datadir = datadir
# job_servers and query_servers are comma separated lists of
- # network locations. See comments on host_port_list for format.
+ # network locations. See comments on host_list() for format.
# Replace any occurrence of locahost with output of gethostname()
# before parsing to match Machine fields of QMF objects later on.
host = socket.gethostname()
@@ -90,7 +89,7 @@
# job server operations
- def set_job_attribute(self, scheduler, job_id, name, value, callback):
+ def set_job_attribute(self, scheduler, job_id, name, value, callback, submission):
assert callback
job_client = self.job_client_pool.get_object()
@@ -116,6 +115,8 @@
jobId.job = job_id
jobId.pool = scheduler.Pool
jobId.scheduler = scheduler.Name
+ jobId.submission.name = submission.Name
+ jobId.submission.owner = submission.Owner
# Make attribute parameter from name and value
aviary_attr = job_client.factory.create('ns0:Attribute')
@@ -134,7 +135,7 @@
def my_callback(result):
# Turn this back off before we put it back in the pool
# so allow_overrides isn't set for someone else...
- job_client.set_allow_overrides(False)
+ job_client.set_enable_attributes(False)
self.job_client_pool.return_object(job_client)
result = self._pretty_result(result, scheduler.Machine)
if isinstance(result, Exception):
@@ -144,7 +145,11 @@
# we'll pass it anyway even though Cumin does not care
# at the present time
status = AviaryOperations._get_status(result.status)
- callback(status, result.id)
+ if status == "OK" and hasattr(result, "id"):
+ id = result.id
+ else:
+ id = None
+ callback(status, id)
host = self._get_host(scheduler.Machine, self.job_servers)
if host == "":
@@ -170,7 +175,8 @@
# "extras" fields and set allowOverrides to True.
# (otherwise, Requirements will be limited to particular
# resource constraint types defined by aviary)
- job_client.set_allow_overrides(True)
+ job_client.set_enable_attributes(True)
+ job_client.set_attributes({"allowOverrides": True})
extras = list()
for k, v in ad.iteritems():
# We don't need to send descriptors down to aviary
@@ -198,37 +204,76 @@
t = CallThread(job_client.service.submitJob, my_callback, *args)
t.start()
- def hold_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
+ def hold_job(self, scheduler, job_id, reason, submission, *args, **kwargs):
'''
This method is asynchronous iff 'callback' is supplied.
+
+ kwargs will be searched for 'callback', 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
'''
- return self._control_job(scheduler, job_id, reason,
+
+ callback = "callback" in kwargs and kwargs["callback"] or None
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
+ return self._control_job(scheduler, job_id, reason, submission,
"holdJob",
callback, default, timeout)
- def release_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
+ def release_job(self, scheduler, job_id, reason, submission, *args, **kwargs):
'''
This method is asynchronous iff 'callback' is supplied.
+
+ kwargs will be searched for 'callback', 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
'''
- return self._control_job(scheduler, job_id, reason,
+
+ callback = "callback" in kwargs and kwargs["callback"] or None
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
+ return self._control_job(scheduler, job_id, reason, submission,
"releaseJob",
callback, default, timeout)
- def remove_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
+ def remove_job(self, scheduler, job_id, reason, submission, *args, **kwargs):
'''
This method is asynchronous iff 'callback' is supplied.
+
+ kwargs will be searched for 'callback', 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
'''
- return self._control_job(scheduler, job_id, reason,
+
+ callback = "callback" in kwargs and kwargs["callback"] or None
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
+ return self._control_job(scheduler, job_id, reason, submission,
"removeJob",
callback, default, timeout)
+#########################
# query server ops
+########################
def fetch_job_data(self, job_server, job_id, ftype, file, start, end,
- default=None, timeout=5):
+ scheduler_name, submission, *args, **kwargs):
+ '''
+ kwargs will be searched for 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
+ '''
+
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
# Aviary doesn't use the file name as does QMF, instead it
# specifies the file type and lets condor figure out the path.
@@ -242,9 +287,12 @@
data = None
else:
status = AviaryOperations._get_status(result.status)
- # Match the format expected by Cumin. This is
- # the format used by the QMF call...
- data = {'Data': result.content}
+ if status == "OK" and hasattr(result, "content"):
+ # Match the format expected by Cumin. This is
+ # the format used by the QMF call...
+ data = {'Data': result.content}
+ else:
+ data = None
return (status, data)
host = self._get_host(job_server.Machine, self.query_servers)
@@ -261,6 +309,9 @@
jobData = client.factory.create('ns0:JobData')
jobData.id.job = job_id
jobData.id.pool = job_server.Pool
+ jobData.id.scheduler = scheduler_name
+ jobData.id.submission.name = submission.Name
+ jobData.id.submission.owner = submission.Owner
# Translate cumin file type to Aviary file type
if ftype == "e":
@@ -282,11 +333,22 @@
self.query_client_pool.return_object(client)
return res;
- def get_job_ad(self, job_server, job_id, default=None, timeout=5):
-
+ def get_job_ad(self, job_server, job_id, scheduler_name, submission,
+ *args, **kwargs):
+ '''
+ kwargs will be searched for 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
+ '''
+
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
client = self.query_client_pool.get_object()
def make_tuple(attr):
+ # Attempt to cast the value into the specified type
if attr.type in self.aviary_to_type:
try:
v = self.aviary_to_type[attr.type](attr.value)
@@ -307,12 +369,15 @@
data = None
else:
status = AviaryOperations._get_status(result[0].status)
- # Match the format expected by Cumin. This is
- # the format used by the QMF call. We have a list
- # of attributes in attrs that we need to make into
- # a dictionary
- ads = make_dict(result[0].details.attrs)
- data = {'JobAd': ads}
+ if status == "OK":
+ # Match the format expected by Cumin. This is
+ # the format used by the QMF call. We have a list
+ # of attributes in attrs that we need to make into
+ # a dictionary
+ ads = make_dict(result[0].details.attrs)
+ data = {'JobAd': ads}
+ else:
+ data = None
return (status, data)
host = self._get_host(job_server.Machine, self.query_servers)
@@ -329,15 +394,99 @@
jobId = client.factory.create('ns0:JobID')
jobId.job = job_id
jobId.pool = job_server.Pool
- # still don't have the scheduler filled in here
- # or the submission. Ditto with other jobId values in this module
+ jobId.scheduler = scheduler_name
+ jobId.submission.name = submission.Name
+ jobId.submission.owner = submission.Owner
res = self._call_sync(my_process_results,
client.service.getJobDetails, jobId)
self.query_client_pool.return_object(client)
return res;
+ def get_job_summaries(self, submission, callback, machine_name):
+ assert callback
+
+ query_client = self.query_client_pool.get_object()
+
+ def to_int_seconds(dt):
+ # Change a datetime.datetime into int seconds since epoch
+ # Note, this works nicely if the datetime happens to include microseconds
+ # since the call to timetuple will drop them. Stuff coming back from
+ # condor should not have microseconds anyway.
+ return int(time.mktime(dt.timetuple()))
+
+ def get_string(job, attr):
+ # Cast suds text types into str so we have standard Py types
+ # Handles optional strings as well
+ if hasattr(job, attr):
+ return str(getattr(job, attr))
+ return ""
+
+ def adapt(jobs):
+ # Make an aviary job summary look like the canonical form
+ # that cumin is expecting (actually the QMF form because of history).
+ result = list()
+ for job in jobs:
+ cluster, proc = job.id.job.split(".")
+ j = dict()
+ j["ClusterId"] = int(cluster)
+ j["Cmd"] = str(job.cmd)
+ j["EnteredCurrentStatus"] = to_int_seconds(job.last_update)
+ j["GlobalJobId"] = job.id.scheduler + \
+ "#" + job.id.job + \
+ "#" + str(to_int_seconds(job.queued))
+ j["JobStatus"] = str(job.job_status)
+ j["ProcId"] = int(proc)
+ j["QDate"] = to_int_seconds(job.queued)
+
+ # These may be null...
+ j["Args"] = get_string(job, "args1")
+ j["ReleaseReason"] = get_string(job, "released")
+ j["HoldReason"] = get_string(job, "held")
+ result.append(j)
+ return result
+
+ def my_callback(result):
+ query_client.set_enable_attributes(False)
+ self.query_client_pool.return_object(query_client)
+ result = self._pretty_result(result, machine_name)
+ if isinstance(result, Exception):
+ callback(result, None)
+ else:
+ status = AviaryOperations._get_status(result[0].status)
+ if status == "OK" and hasattr(result[0], "jobs"):
+ data = {"Jobs": adapt(result[0].jobs)}
+ else:
+ data = {"Jobs": None}
+ callback(status, data)
+
+ host = self._get_host(machine_name, self.query_servers)
+ if host == "":
+ self._raise_no_host(machine_name)
+
+ service = host + "getSubmissionSummary"
+
+ # Have to set the URL for the method. This might go away someday...
+ query_client.set_options(location=service)
+
+ # What we really want here is the job summaries from the
+ # submission summary response. To get those, we have to
+ # set an extra attribute on the client...
+ query_client.set_enable_attributes(True)
+ query_client.set_attributes({"includeJobSummaries": "true"})
+
+ # Make a submission id. (see query wsdl)
+ subId = query_client.factory.create('ns0:SubmissionID')
+ subId.name = submission.Name
+ subId.owner = submission.Owner
+
+ t = CallThread(query_client.service.getSubmissionSummary,
+ my_callback, subId)
+ t.start()
+
+########################################################
# Secret private implementation stuff, don't look!
+########################################################
@classmethod
def _type_to_aviary(cls):
@@ -407,7 +556,8 @@
sync.get_completion()(*cb_args)
return sync
- def _control_job(self, scheduler, job_id, reason, meth_name,
+ def _control_job(self, scheduler, job_id, reason, submission,
+ meth_name,
callback, default, timeout):
host = self._get_host(scheduler.Machine, self.job_servers)
@@ -428,6 +578,8 @@
jobId.job = job_id
jobId.pool = scheduler.Pool
jobId.scheduler = scheduler.Name
+ jobId.submission.name = submission.Name
+ jobId.submission.owner = submission.Owner
if callback:
def my_callback(result):
@@ -461,16 +613,19 @@
to the suds message after marshalling.
'''
def __init__(self):
- self.allow = False
+ self.set_attributes = False
+ self.attributes = dict()
def marshalled(self, context):
- if self.allow:
+ if self.set_attributes:
sj_body = context.envelope.getChild('Body')[0]
- sj_body.attributes.append(Attribute("allowOverrides", "true"))
+ for k,v in self.attributes.iteritems():
+ sj_body.attributes.append(Attribute(k, v))
+
except:
class OverridesPlugin(object):
def __init__(self):
- self.allow = False
+ self.set_attributes = False
class OverrideClient(Client):
'''
@@ -484,9 +639,28 @@
except:
super(OverrideClient, self).__init__(*args)
- def set_allow_overrides(self, truth):
- self.override.allow = truth
+ def set_enable_attributes(self, truth):
+ '''
+ Set the flag that controls whether extra attributes are marshalled.
+ The attribute set is specified through the set_attributes() method.
+ If 'truth' is True, then service calls made through this client
+ will have the extra attributes appended to the attribute set of service
+ request.
+ '''
+ self.override.set_attributes = truth
+
+ def set_attributes(self, attrs):
+ '''
+ Specify extra attributes to append to the attribute set
+ of service requests made through this client.
+
+ The extra attributes will be added if set_enable_attributes()
+ has been called with a value of True sometime prior to the
+ time of a request.
+ '''
+ self.override.attributes = attrs
+
class JobClientPool(ObjectPool):
def __init__(self, job_wsdl, max_size):
super(JobClientPool, self).__init__(max_size)
@@ -501,4 +675,4 @@
self.query_wsdl = query_wsdl
def create_object(self):
- return Client(self.query_wsdl)
+ return OverrideClient(self.query_wsdl)
Modified: trunk/sage/python/sage/qmf/qmfoperations.py
===================================================================
--- trunk/sage/python/sage/qmf/qmfoperations.py 2011-08-17 18:17:46 UTC (rev 4935)
+++ trunk/sage/python/sage/qmf/qmfoperations.py 2011-08-26 14:18:11 UTC (rev 4936)
@@ -81,7 +81,7 @@
# scheduler operations
- def set_job_attribute(self, scheduler, job_id, name, value, callback):
+ def set_job_attribute(self, scheduler, job_id, name, value, callback, *args):
assert callback
self._call(scheduler, "SetJobAttribute", callback, 0, 0,
job_id, name, value)
@@ -90,37 +90,84 @@
assert callback
self._call(scheduler, "SubmitJob", callback, 0, 0, ad)
- def hold_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
+ def hold_job(self, scheduler, job_id, reason, *args, **kwargs):
'''
This method is asynchronous iff 'callback' is supplied.
+
+ kwargs will be searched for 'callback', 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
'''
+
+ callback = "callback" in kwargs and kwargs["callback"] or None
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
return self._call(scheduler, "HoldJob", callback, default,
timeout, job_id, reason)
- def release_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
+ def release_job(self, scheduler, job_id, reason, *args, **kwargs):
'''
This method is asynchronous iff 'callback' is supplied.
+
+ kwargs will be searched for 'callback', 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
'''
+
+ callback = "callback" in kwargs and kwargs["callback"] or None
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
return self._call(scheduler, "ReleaseJob", callback, default,
timeout, job_id, reason)
- def remove_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
+ def remove_job(self, scheduler, job_id, reason, *args, **kwargs):
'''
This method is asynchronous iff 'callback' is supplied.
+
+ kwargs will be searched for 'callback', 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
'''
+
+ callback = "callback" in kwargs and kwargs["callback"] or None
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
return self._call(scheduler, "RemoveJob", callback, default,
timeout, job_id, reason)
# jobserver operations
- def get_job_ad(self, job_server, job_id, default=None, timeout=5):
+ def get_job_ad(self, job_server, job_id, *args, **kwargs):
+ '''
+ kwargs will be searched for 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
+ '''
+
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
return self._call(job_server, "GetJobAd", 0, default, timeout, job_id)
def fetch_job_data(self, job_server, job_id, ftype, file, start, end,
- default=None, timeout=5):
+ *args, **kwargs):
+ '''
+ kwargs will be searched for 'default' and 'timeout' arguments.
+
+ These optional arguments were moved to kwargs for compatibility
+ with another implementation of the same routine.
+ '''
+
+ default = "default" in kwargs and kwargs["default"] or None
+ timeout = "timeout" in kwargs and kwargs["timeout"] or 5
+
# QMF doesn't use the ftype value, it just uses the filename
return self._call(job_server, "FetchJobData", 0, default, timeout,
job_id, file, start, end)
@@ -158,7 +205,7 @@
# submission operations
- def get_job_summaries(self, submission, callback):
+ def get_job_summaries(self, submission, callback, *args):
assert callback
return self._call(submission, "GetJobSummaries", callback, 0, 0)
