r4943 - in trunk/cumin: model python/cumin python/cumin/grid
by croberts＠fedoraproject.org
Author: croberts
Date: 2011-08-31 20:30:52 +0000 (Wed, 31 Aug 2011)
New Revision: 4943
Modified:
trunk/cumin/model/cumin.xml
trunk/cumin/python/cumin/grid/tags.py
trunk/cumin/python/cumin/util.py
Log:
First pass at assigning features to tags. Also, moved the truncate_text logic into util.
Modified: trunk/cumin/model/cumin.xml
===================================================================
--- trunk/cumin/model/cumin.xml 2011-08-31 18:09:29 UTC (rev 4942)
+++ trunk/cumin/model/cumin.xml 2011-08-31 20:30:52 UTC (rev 4943)
@@ -58,6 +58,7 @@
<property name="Name" type="sstr"/>
<property name="Tags" type="sstr"/>
<property name="Checkin" type="sstr"/>
+ <property name="Features" type="sstr"/>
</class>
Modified: trunk/cumin/python/cumin/grid/tags.py
===================================================================
--- trunk/cumin/python/cumin/grid/tags.py 2011-08-31 18:09:29 UTC (rev 4942)
+++ trunk/cumin/python/cumin/grid/tags.py 2011-08-31 20:30:52 UTC (rev 4943)
@@ -1,7 +1,7 @@
import logging
from wooly import Widget, Attribute
-from wooly.util import StringCatalog, Writer
+from wooly.util import StringCatalog, Writer, escape_amp, escape_entity
from wooly.datatable import DataAdapterOptions, DataAdapterField, DataTableColumn, DataTable, DataAdapter
from wooly.widgets import RadioModeSet, WidgetSet
from wooly.template import WidgetTemplate
@@ -85,9 +85,10 @@
data = []
try:
wallaby_tags = self.app.wallaby.get_data(WBTypes.TAGS)
+ wallaby_features = self.app.wallaby.get_data(WBTypes.FEATURES)
for i, tag in enumerate(wallaby_tags):
- data.append({'Tag':str(tag.name), 'Nodes':self.app.wallaby.get_node_names(tag)})
+ data.append({'Tag':str(tag.name), 'Nodes':self.app.wallaby.get_node_names(tag), 'Features':tag.features})
except Exception, e:
log.exception(e)
@@ -100,7 +101,7 @@
records = list()
if results:
for result in results:
- records.append([result['Tag'], result['Nodes']])
+ records.append([result['Tag'], result['Nodes'], result['Features']])
return records
@@ -142,6 +143,7 @@
super(TagsFrame, self).__init__(app, name, cls)
self.set_tags = TagsNodeEditTask(app, self)
self.set_nodes = TagsTagEditTask(app, self)
+ self.set_features = TagsFeatureEditTask(app, self)
class TagInventory(ObjectSelectorNoCheckboxes):
'''
@@ -160,10 +162,16 @@
link = TaskLink(app, "tag_remove", remove_node_tags_task)
self.links.add_child(link)
- col = self.TagColumn(app, "tagcol", cls.Tags)
+ col = self.TagColumn(app, "tagcol", cls.Tags)
+ col.width = "20%"
self.add_column(col)
- self.add_search_filter(col)
+ self.add_search_filter(col)
+ col = self.FeatureColumn(app, "featcol", cls.Features)
+ col.width = "80%"
+ self.add_column(col)
+ #self.add_search_filter(col)
+
def render_title(self, session):
return "Tags"
@@ -181,6 +189,28 @@
self.frame.tag.set_nodes.form.tags.set(session, tags)
href = self.frame.tag.set_nodes.get_href(session)
return fmt_link(href, tags)
+
+ class FeatureColumn(ObjectTableColumn):
+ '''
+ This column will display the features that are currently assigned to a tag.
+ If a tag has no features on it, <add features to this tag> will be displayed instead.
+ '''
+ def render_cell_content(self, session, data):
+ tags = data[0]
+ feature_list = data[2]
+ features = ""
+ if feature_list:
+ for feature in feature_list:
+ features = features + "," + str(feature)
+ features = features[1:]
+ #here we set some info that will be used in the display of the target form
+ self.frame.tag.set_features.form.feature_name.set(session, features)
+ self.frame.tag.set_features.form.tags.set(session, tags)
+ href = self.frame.tag.set_features.get_href(session)
+ if not features or features == "":
+ features = escape_entity("<add features to this tag>")
+ features = truncate_text(features, 50, True)
+ return fmt_link(href, features)
class NodeInventory(ObjectSelectorNoCheckboxes):
@@ -242,9 +272,7 @@
for tag in tags_list:
tags = tags + "," + str(tag)
tags = tags[1:]
- retval = len(tags) > 0 and tags or ""
- if(len(tags) > 50):
- retval = tags[:50] + "..." #indicate that we truncated the name
+ retval = truncate_text(tags, 50, True)
return retval
class NodeColumn(ObjectTableColumn):
@@ -342,7 +370,7 @@
tags_string = ""
for i, tag in enumerate(items):
- tags_string = tags_string + "<option id='" + str(i) + "' name='" + tag.title + "' value='" + tag.title + "'" + ">" + tag.title + "</option>"
+ tags_string = tags_string + "<option id='" + str(i) + "' name=\"" + tag.title + "\" value=\"" + tag.title + "\"" + ">" + tag.title + "</option>"
return tags_string
@@ -572,10 +600,8 @@
'''
Show the initial value passed-in, truncated to trunc_length characters
'''
- trunc_length = 50
value = self.input.get(session)
- if(len(value) > trunc_length):
- value = value[:trunc_length] + "..." #indicate truncated value
+ value = truncate_text(value, 50, True)
return value
class DisabledInput(StringInput):
@@ -665,9 +691,7 @@
def render_inputs(self, session, *args):
value = self.input.get(session)
- trunc_length = 50
- if(len(value) > trunc_length):
- value = value[:trunc_length] + "..." #indicate truncated value
+ value = truncate_text(value, 50, True)
return value
def render_title(self, session):
@@ -693,6 +717,117 @@
class DisabledInput(StringInput):
pass
+class EditTagFeaturesForm(ObjectFrameTaskFeedbackForm):
+ '''
+ This form will allow the editing of nodes for a single given tag
+ '''
+ def __init__(self, app, name, task):
+ super(EditTagFeaturesForm, self).__init__(app, name, task)
+
+ self.tags = self.Tags(app, "tags")
+ self.add_field(self.tags)
+
+ self.feature_name = self.FeatureName(app, "feature_name")
+ self.add_field(self.feature_name)
+
+ self.possible_features = self.FeatureList(app, "pfeat")
+ self.add_field(self.possible_features)
+
+ self.update_enabled = False
+
+
+ def process_submit(self, session):
+ tag = self.tags.input.get(session)
+ features_value = self.possible_features.get(session)
+
+ self.tags.validate(session)
+ if not self.errors.get(session):
+ self.tags.set(session, tag)
+
+ self.task.invoke(session, None, tag, features_value)
+ self.task.exit_with_redirect(session)
+
+ def render_form_class(self, session):
+ return " ".join((super(EditTagFeaturesForm, self).render_form_class(session), "mform"))
+
+ class FeatureList(PageableFilteredSelect):
+ def __init__(self, app, name):
+ super(EditTagFeaturesForm.FeatureList, self).__init__(app, name)
+
+ def render_items(self, session):
+ items = self.do_get_items(session)
+ features_string = ""
+
+ given_tag = self.form.tags.input.param.get(session)
+ tag_object = self.app.wallaby.get_tag_by_name(given_tag)
+ selected_features = tag_object.features
+
+ for i, feature in enumerate(items):
+ selected = ""
+ if(selected_features and feature.title in selected_features):
+ selected = " selected='selected' "
+ features_string = features_string + "<option id='" + str(i) + "' name='" + feature.title + "' value='" + feature.title + "'" + selected + ">" + feature.title + "</option>"
+
+ return features_string
+
+ def do_get_items(self, session):
+ features = fetchFeatures(self, session)
+ items = list()
+
+ if features:
+ for feature in features:
+ item = CheckboxItem(feature)
+ item.title = feature
+ items.append(item)
+
+ return items
+
+ def render_title(self, session):
+ return "Update features"
+
+ class FeatureName(StringField):
+ def __init__(self, app, name):
+ super(EditTagFeaturesForm.FeatureName, self).__init__(app, name)
+
+ self.input = self.DisabledInput(app, "input")
+ self.replace_child(self.input)
+
+ def get(self,session):
+ value = self.input.get(session)
+
+ if not value or value == "":
+ value = "No features currently selected"
+
+ return value
+
+ def render_inputs(self, session, *args):
+ value = self.input.get(session)
+ value = truncate_text(value, 50, True)
+ return value
+
+ def render_title(self, session):
+ return "Current features"
+
+ class DisabledInput(StringInput):
+ pass
+
+ class Tags(StringField):
+ def __init__(self, app, name):
+ super(EditTagFeaturesForm.Tags, self).__init__(app, name)
+
+ self.input = self.DisabledInput(app, "input")
+ self.replace_child(self.input)
+
+ def render_title(self, session):
+ return "Tag"
+
+ def get(self,session):
+ value = self.input.get(session)
+ return value
+
+ class DisabledInput(StringInput):
+ pass
+
class TagsNodeEditTask(ObjectFrameTask):
'''
On invocation, this task will take a node and reset the tags for that node to a
@@ -763,6 +898,38 @@
invoc.end()
+class TagsFeatureEditTask(ObjectFrameTask):
+ '''
+ This is the task that will take a tag and a set of nodes and make the
+ necessary assignments or unassignments.
+ '''
+ def __init__(self, app, frame):
+ super(TagsFeatureEditTask, self).__init__(app, frame)
+
+ self.form = EditTagFeaturesForm(app, self.name, self)
+
+ def get_title(self, session):
+ return "Change features for this tag "
+
+ def do_enter(self, session, osession):
+ self.form.tags.set(session, self.form.tags.get(osession))
+ self.form.feature_name.set(session, self.form.feature_name.get(osession))
+
+ def do_invoke(self, invoc, negotiator, tag, chosen_features):
+ '''
+ if the node is on the current_nodes list and is on the chosen_nodes passed in, nothing to do
+ ** if the node is NOT on the current_nodes list and is on the chosen_nodes passed in, add this tag to that node
+ ** if the node is on the current_nodes list and is NOT on the chosen_nodes passed in, update that node sans this tag
+ if the node is NOT on the current_nodes list and is NOT on the chosen_nodes passed in, nothing to do
+ '''
+ try:
+ call_async(invoc.make_callback(), self.app.wallaby.edit_features, tag, *chosen_features)
+ except Exception, e:
+ invoc.status = invoc.FAILED
+ log.exception(e)
+
+ invoc.end()
+
class AddTags(Task):
'''
This task is used to create a tag in wallaby without assigning it to any nodes.
@@ -888,4 +1055,18 @@
node_list.append(str(node.name))
node_list.sort()
- return node_list
\ No newline at end of file
+ return node_list
+
+def fetchFeatures(self, session):
+ '''
+ fetch the list of nodes from wallaby
+ '''
+ wallaby_features = self.app.wallaby.get_data(WBTypes.FEATURES)
+ feature_list = list()
+
+ for feature in wallaby_features:
+ feature_list.append(str(feature.name))
+ feature_list.sort()
+
+ return feature_list
+
\ No newline at end of file
Modified: trunk/cumin/python/cumin/util.py
===================================================================
--- trunk/cumin/python/cumin/util.py 2011-08-31 18:09:29 UTC (rev 4942)
+++ trunk/cumin/python/cumin/util.py 2011-08-31 20:30:52 UTC (rev 4943)
@@ -244,3 +244,17 @@
hList.append(n.rjust(2, '0').upper())
return "".join(hList)
+
+def truncate_text(text, length, indicator):
+ ''' Quick truncation of text at specified length
+
+ text: the text to be truncated
+ length: the length at which to truncate the text
+ indicator: If true, add "..." to the truncated text to indicate that truncation happened
+ '''
+ retval = len(text) > 0 and text or ""
+ if(len(retval) > length):
+ retval = retval[:length]
+ if indicator:
+ retval = retval + "..."
+ return retval
12 years, 7 months
- 1
- 0
r4942 - trunk/sage/python/sage/aviary
by tmckay＠fedoraproject.org
Author: tmckay
Date: 2011-08-31 18:09:29 +0000 (Wed, 31 Aug 2011)
New Revision: 4942
Modified:
trunk/sage/python/sage/aviary/https.py
Log:
Missing import of socket, breaks on Python 2.4
Modified: trunk/sage/python/sage/aviary/https.py
===================================================================
--- trunk/sage/python/sage/aviary/https.py 2011-08-30 21:03:38 UTC (rev 4941)
+++ trunk/sage/python/sage/aviary/https.py 2011-08-31 18:09:29 UTC (rev 4942)
@@ -20,6 +20,7 @@
import urllib2 as u2
from suds.transport.http import HttpTransport, Reply, TransportError
import httplib
+import socket
class HTTPSClientAuthHandler(u2.HTTPSHandler):
def __init__(self, key, cert):
12 years, 7 months
- 1
- 0
r4941 - trunk/sage/python/sage/aviary
by tmckay＠fedoraproject.org
Author: tmckay
Date: 2011-08-30 21:03:38 +0000 (Tue, 30 Aug 2011)
New Revision: 4941
Modified:
trunk/sage/python/sage/aviary/aviaryoperations.py
Log:
Remove commented section
Modified: trunk/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- trunk/sage/python/sage/aviary/aviaryoperations.py 2011-08-30 20:59:44 UTC (rev 4940)
+++ trunk/sage/python/sage/aviary/aviaryoperations.py 2011-08-30 21:03:38 UTC (rev 4941)
@@ -105,14 +105,6 @@
cb_args = self._cb_args_dataless(result)
callback(*cb_args)
-# job_client = self.job_client_pool.get_object()
-# scheme, host = self._get_host(scheduler.Machine, self.job_servers)
-# if host == "":
-# self._raise_no_host(scheduler.Machine)
-# service = host + "setJobAttribute"
- # Have to set the URL for the method. This might go away someday...
-# job_client.set_options(location=service)
-
job_client = self._setup_client(self.job_client_pool,
scheduler.Machine,
self.job_servers,
12 years, 7 months
- 1
- 0
r4940 - in trunk: cumin/bin cumin/python/cumin sage/python/sage/aviary
by tmckay＠fedoraproject.org
Author: tmckay
Date: 2011-08-30 20:59:44 +0000 (Tue, 30 Aug 2011)
New Revision: 4940
Added:
trunk/sage/python/sage/aviary/https.py
Modified:
trunk/cumin/bin/cumin-web
trunk/cumin/python/cumin/config.py
trunk/cumin/python/cumin/main.py
trunk/sage/python/sage/aviary/aviaryoperations.py
Log:
Add support for ssl connection to aviary.
Modified: trunk/cumin/bin/cumin-web
===================================================================
--- trunk/cumin/bin/cumin-web 2011-08-29 18:00:52 UTC (rev 4939)
+++ trunk/cumin/bin/cumin-web 2011-08-30 20:59:44 UTC (rev 4940)
@@ -18,6 +18,8 @@
cumin.use_aviary = values.use_aviary
cumin.aviary_job_servers = values.aviary_job_servers
cumin.aviary_query_servers = values.aviary_query_servers
+ cumin.aviary_key = values.aviary_key
+ cumin.aviary_cert = values.aviary_cert
def set_wallaby_configs(cumin, values, brokers):
if values.wallaby_broker == "":
Modified: trunk/cumin/python/cumin/config.py
===================================================================
--- trunk/cumin/python/cumin/config.py 2011-08-29 18:00:52 UTC (rev 4939)
+++ trunk/cumin/python/cumin/config.py 2011-08-30 20:59:44 UTC (rev 4940)
@@ -164,6 +164,12 @@
param = ConfigParameter(self, "aviary-query-servers", str)
param.default = "http://localhost:9091"
+ param = ConfigParameter(self, "aviary-key", str)
+ param.default = ""
+
+ param = ConfigParameter(self, "aviary-cert", str)
+ param.default = ""
+
self.log_file = ConfigParameter(self, "log-file", str)
param = ConfigParameter(self, "log-level", str)
Modified: trunk/cumin/python/cumin/main.py
===================================================================
--- trunk/cumin/python/cumin/main.py 2011-08-29 18:00:52 UTC (rev 4939)
+++ trunk/cumin/python/cumin/main.py 2011-08-30 20:59:44 UTC (rev 4940)
@@ -87,6 +87,8 @@
self.use_aviary = True
self.aviary_job_servers = ""
self.aviary_query_servers = ""
+ self.aviary_key = ""
+ self.aviary_cert = ""
self.wallaby = None
self.wallaby_broker = None
@@ -172,7 +174,8 @@
ops.insert(0, AviaryOperations("aviary", aviary_dir,
self.aviary_job_servers,
- self.aviary_query_servers))
+ self.aviary_query_servers,
+ self.aviary_key, self.aviary_cert))
self.remote.add_mechanisms(ops)
# Create RPC interface for Wallaby
Modified: trunk/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- trunk/sage/python/sage/aviary/aviaryoperations.py 2011-08-29 18:00:52 UTC (rev 4939)
+++ trunk/sage/python/sage/aviary/aviaryoperations.py 2011-08-30 20:59:44 UTC (rev 4940)
@@ -9,8 +9,10 @@
from suds import *
from suds.client import Client
+from suds.transport.https import HttpAuthenticated
from sage.util import CallSync, CallThread, ObjectPool, host_list
from datetime import datetime
+from sage.aviary.https import *
log = logging.getLogger("sage.aviary")
@@ -56,7 +58,8 @@
'''
class AviaryOperations(object):
- def __init__(self, name, datadir, job_servers, query_servers):
+ def __init__(self, name, datadir, job_servers, query_servers,
+ key="", cert=""):
self.name = name
self.datadir = datadir
@@ -87,29 +90,34 @@
self.type_to_aviary = self._type_to_aviary()
self.aviary_to_type = self._aviary_to_type()
+ self.key = key
+ self.cert = cert
+
# job server operations
def set_job_attribute(self, scheduler, job_id, name, value, callback, submission):
assert callback
- job_client = self.job_client_pool.get_object()
-
def my_callback(result):
self.job_client_pool.return_object(job_client)
result = self._pretty_result(result, scheduler.Machine)
# massage results for use by standard callback
cb_args = self._cb_args_dataless(result)
callback(*cb_args)
-
- host = self._get_host(scheduler.Machine, self.job_servers)
- if host == "":
- self._raise_no_host(scheduler.Machine)
-
- service = host + "setJobAttribute"
-
+
+# job_client = self.job_client_pool.get_object()
+# scheme, host = self._get_host(scheduler.Machine, self.job_servers)
+# if host == "":
+# self._raise_no_host(scheduler.Machine)
+# service = host + "setJobAttribute"
# Have to set the URL for the method. This might go away someday...
- job_client.set_options(location=service)
+# job_client.set_options(location=service)
+ job_client = self._setup_client(self.job_client_pool,
+ scheduler.Machine,
+ self.job_servers,
+ "setJobAttribute")
+
# Make a job id parameter (see job wsdl)
jobId = job_client.factory.create('ns0:JobID')
jobId.job = job_id
@@ -130,8 +138,6 @@
def submit_job(self, scheduler, ad, callback):
assert callback
- job_client = self.job_client_pool.get_object()
-
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...
@@ -151,15 +157,11 @@
id = None
callback(status, id)
- host = self._get_host(scheduler.Machine, self.job_servers)
- if host == "":
- self._raise_no_host(scheduler.Machine)
+ job_client = self._setup_client(self.job_client_pool,
+ scheduler.Machine,
+ self.job_servers,
+ "submitJob")
- service = host + "submitJob"
-
- # Have to set the URL for the method. This might go away someday...
- job_client.set_options(location=service)
-
# Set basic attributes in the order defined by aviary-job.wsdl.
args = list()
basic_attrs = ("Cmd", "Args", "Owner", "Iwd", "Submission")
@@ -277,8 +279,6 @@
# Aviary doesn't use the file name as does QMF, instead it
# specifies the file type and lets condor figure out the path.
- client = self.query_client_pool.get_object()
-
def my_process_results(result):
# Fix up the exception message if necessary
result = self._pretty_result(result, job_server.Machine)
@@ -295,16 +295,11 @@
data = None
return (status, data)
- host = self._get_host(job_server.Machine, self.query_servers)
- if host == "":
- self._raise_no_host(job_server.Machine)
+ client = self._setup_client(self.query_client_pool,
+ job_server.Machine,
+ self.query_servers,
+ "getJobData")
- service = host + "getJobData"
-
- # Have to set the URL for the method. This might go away someday...
- #print "Aviary control job " + url
- client.set_options(location=service)
-
# Make a job data parameter (see query wsdl)
jobData = client.factory.create('ns0:JobData')
jobData.id.job = job_id
@@ -345,8 +340,6 @@
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:
@@ -380,16 +373,11 @@
data = None
return (status, data)
- host = self._get_host(job_server.Machine, self.query_servers)
- if host == "":
- self._raise_no_host(job_server.Machine)
+ client = self._setup_client(self.query_client_pool,
+ job_server.Machine,
+ self.query_servers,
+ "getJobDetails")
- service = host + "getJobDetails"
-
- # Have to set the URL for the method. This might go away someday...
- #print "Aviary control job " + url
- client.set_options(location=service)
-
# Make a job id parameter (see job wsdl)
jobId = client.factory.create('ns0:JobID')
jobId.job = job_id
@@ -406,8 +394,6 @@
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
@@ -459,16 +445,12 @@
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"
+ query_client = self._setup_client(self.query_client_pool,
+ machine_name,
+ self.query_servers,
+ "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...
@@ -499,23 +481,51 @@
@classmethod
def _get_host(cls, name, servers):
+ scheme = ""
host = ""
if name in servers:
urls = servers[name]
if len(urls) > 0:
- host = str(random.sample(urls, 1)[0])
+ url = random.sample(urls, 1)[0]
+ scheme = url.scheme
+ host = str(url)
# A particular method name is going to be appended to path,
# so ensure the file "/" here. Since we supply default path
# values when the host list is parsed, we know the last portion
# of the host string has to be a path.
if not host.endswith("/"):
host += "/"
- return host
+ return scheme, host
@classmethod
def _raise_no_host(cls, name):
raise Exception("No aviary job servers specified for %s, check config files" % name)
+ def _setup_client(self, client_pool, host_name, host_list, method):
+ # Get a client object from the client_pool. Use host_name as
+ # a lookup in the host_list to find an initial URL and append the
+ # method name. Set up the client to point at that URL and return.
+ client = client_pool.get_object()
+ scheme, host = self._get_host(host_name, host_list)
+ if host == "":
+ self._raise_no_host(host_name)
+ # Have to set the URL for the method. This might go away someday...
+ client.set_options(location=host+method)
+ # Since we pool the clients and reuse them for different requests
+ # and since its possible to be using servers with different schemes,
+ # we have to always reset the transport here.
+ if scheme == "https":
+ if not os.path.isfile(self.key):
+ raise Exception("File not found for aviary-key, check cumin.conf settings")
+ if not os.path.isfile(self.cert):
+ raise Exception("File not found for aviary-cert, check cumin.conf settings")
+ the_transport = HTTPSClientCertTransport(self.key, self.cert)
+ else:
+ # this is the default transport when none is specified
+ the_transport = HttpAuthenticated()
+ client.set_options(transport=the_transport)
+ return client
+
@classmethod
def _get_status(cls, result):
# For Aviary operations, if the operation
@@ -560,19 +570,13 @@
meth_name,
callback, default, timeout):
- host = self._get_host(scheduler.Machine, self.job_servers)
- if host == "":
- self._raise_no_host(scheduler.Machine)
+ client = self._setup_client(self.job_client_pool,
+ scheduler.Machine,
+ self.job_servers,
+ meth_name)
- service = host + meth_name
-
- client = self.job_client_pool.get_object()
meth = getattr(client.service, meth_name)
- # Have to set the URL for the method. This might go away someday...
- #print "Aviary control job " + url
- client.set_options(location=service)
-
# Make a job id parameter (see job wsdl)
jobId = client.factory.create('ns0:JobID')
jobId.job = job_id
Added: trunk/sage/python/sage/aviary/https.py
===================================================================
--- trunk/sage/python/sage/aviary/https.py (rev 0)
+++ trunk/sage/python/sage/aviary/https.py 2011-08-30 20:59:44 UTC (rev 4940)
@@ -0,0 +1,59 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2009-2011 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# uses Suds - https://fedorahosted.org/suds/
+import urllib2 as u2
+from suds.transport.http import HttpTransport, Reply, TransportError
+import httplib
+
+class HTTPSClientAuthHandler(u2.HTTPSHandler):
+ def __init__(self, key, cert):
+ u2.HTTPSHandler.__init__(self)
+ self.key = key
+ self.cert = cert
+
+ def https_open(self, req):
+ #Rather than pass in a reference to a connection class, we pass in
+ # a reference to a function which, for all intents and purposes,
+ # will behave as a constructor
+ return self.do_open(self.getConnection, req)
+
+ def getConnection(self, host, timeout=300):
+ return httplib.HTTPSConnection(host, key_file=self.key, cert_file=self.cert)
+
+class HTTPSClientCertTransport(HttpTransport):
+ def __init__(self, key, cert, *args, **kwargs):
+ HttpTransport.__init__(self, *args, **kwargs)
+ self.key = key
+ self.cert = cert
+
+ def u2open(self, u2request):
+ """
+ Open a connection.
+ @param u2request: A urllib2 request.
+ @type u2request: urllib2.Request.
+ @return: The opened file-like urllib2 object.
+ @rtype: fp
+ """
+ tm = self.options.timeout
+ url = u2.build_opener(HTTPSClientAuthHandler(self.key, self.cert))
+ if self.u2ver() < 2.6:
+ socket.setdefaulttimeout(tm)
+ return url.open(u2request)
+ else:
+ return url.open(u2request, timeout=tm)
12 years, 7 months
- 1
- 0
r4939 - trunk/sage/python/sage/wallaby
by tmckay＠fedoraproject.org
Author: tmckay
Date: 2011-08-29 18:00:52 +0000 (Mon, 29 Aug 2011)
New Revision: 4939
Modified:
trunk/sage/python/sage/wallaby/wallabyoperations.py
Log:
Add try...finally blocks around any locks that are not completely trivial.
Change refresh to take *items
Modified: trunk/sage/python/sage/wallaby/wallabyoperations.py
===================================================================
--- trunk/sage/python/sage/wallaby/wallabyoperations.py 2011-08-29 15:47:22 UTC (rev 4938)
+++ trunk/sage/python/sage/wallaby/wallabyoperations.py 2011-08-29 18:00:52 UTC (rev 4939)
@@ -267,7 +267,7 @@
self._maintain_cache.join(timeout)
log.debug("WallabyOperations: stopped cache maintenance thread")
- def refresh(self, items=()):
+ def refresh(self, *items):
'''
Wake the caching thread if asleep and cause it to iterate.
@@ -276,21 +276,22 @@
Attributes of WBTypes define valid values for elements of "items"
'''
self._condition.acquire()
- if len(items) == 0:
- do_notify = True
- for attr, val in self._cache.iteritems():
- val.refresh = True
- else:
- do_notify = False
- for attr in items:
- if attr in self._cache:
- do_notify = True
- self._cache[attr].refresh = True
+ try:
+ if len(items) == 0:
+ do_notify = True
+ for attr, val in self._cache.iteritems():
+ val.refresh = True
+ else:
+ do_notify = False
+ for attr in items:
+ if attr in self._cache:
+ do_notify = True
+ self._cache[attr].refresh = True
+ if do_notify:
+ self._condition.notify()
+ finally:
+ self._condition.release()
- if do_notify:
- self._condition.notify()
- self._condition.release()
-
def get_data(self, which):
'''
Return a list of cached values for the specified category.
@@ -303,9 +304,11 @@
'''
d = []
self._lock.acquire()
- if which in self._cache:
- d = self._cache[which].data.values()
- self._lock.release()
+ try:
+ if which in self._cache:
+ d = self._cache[which].data.values()
+ finally:
+ self._lock.release()
return d
def get_names(self, which):
@@ -320,9 +323,11 @@
'''
d = []
self._lock.acquire()
- if which in self._cache:
- d = self._cache[which].data.keys()
- self._lock.release()
+ try:
+ if which in self._cache:
+ d = self._cache[which].data.keys()
+ finally:
+ self._lock.release()
return d
def get_node_by_name(self, name):
@@ -374,9 +379,11 @@
else:
n = tag.name
self._lock.acquire()
- if n in self._nodes_by_tag:
- names = self._nodes_by_tag[n]
- self._lock.release()
+ try:
+ if n in self._nodes_by_tag:
+ names = self._nodes_by_tag[n]
+ finally:
+ self._lock.release()
return names
def get_tag_names(self, node):
@@ -392,9 +399,11 @@
else:
n = node.name
self._lock.acquire()
- if n in self._cache[WBTypes.NODES].data:
- names = self._cache[WBTypes.NODES].data[n].getTags()
- self._lock.release()
+ try:
+ if n in self._cache[WBTypes.NODES].data:
+ names = self._cache[WBTypes.NODES].data[n].getTags()
+ finally:
+ self._lock.release()
return names
def create_tag(self, name):
@@ -405,7 +414,7 @@
'''
try:
self._store.addTag(name)
- self.refresh((WBTypes.GROUPS,WBTypes.TAGS))
+ self.refresh(WBTypes.GROUPS,WBTypes.TAGS)
except Exception, e:
log.debug("WallabyOperations: create_tag, exception suppressed, %s" % str(e))
return False
@@ -437,29 +446,36 @@
tags -- the new set of tags for the node,
list or tuple of strings
'''
+ status = False
if type(node) in (str, unicode):
n = node
else:
n = node.name
self._lock.acquire()
- if n in self._cache[WBTypes.NODES].data:
- n = self._cache[WBTypes.NODES].data[n]
- self._lock.release()
try:
- # cast to list in case tags is a tuple, list required
- start = time.time()
- n.modifyTags("REPLACE", list(tags), create_missing_tags=True)
- n.update()
- # refresh the groups and tags so that membership informatio
- # is up to date.
- self.refresh(WBTypes.GROUPS,WBTypes.TAGS)
- except Exception, e:
- log.debug("WallabyOperations: edit_tags, exception suppressed, %s" % str(e))
- return False
- delta = time.time() - start
- log.debug("WallabyOperations: edit_tags %s" % delta)
- return True
+ n = n in self._cache[WBTypes.NODES].data and \
+ self._cache[WBTypes.NODES].data[n] or None
+ finally:
+ self._lock.release()
+ if n is None:
+ log.debug("WallabyOperations: edit_tags, node not found %s" % str(n))
+ else:
+ try:
+ # cast to list in case tags is a tuple, list required
+ start = time.time()
+ n.modifyTags("REPLACE", list(tags), create_missing_tags=True)
+ n.update()
+ status = True
+ # refresh the groups and tags so that membership information
+ # is up to date.
+ self.refresh(WBTypes.GROUPS,WBTypes.TAGS)
+ delta = time.time() - start
+ log.debug("WallabyOperations: edit_tags %s" % delta)
+ except Exception, e:
+ log.debug("WallabyOperations: edit_tags, exception suppressed, %s" % str(e))
+ return status
+
def edit_features(self, group, *features):
'''
Replace existing features in a group with the specified features.
@@ -469,24 +485,30 @@
features -- the new set of features for the group,
list or tuple of strings
'''
+ status = False
if type(group) in (str, unicode):
g = group
else:
g = group.name
self._lock.acquire()
- if g in self._cache[WBTypes.GROUPS].data:
- g = self._cache[WBTypes.GROUPS].data[g]
- self._lock.release()
try:
- start = time.time()
- g.modifyFeatures("REPLACE", features)
- g.update()
- except Exception, e:
- log.debug("WallabyOperations: edit_features, exception suppressed, %s" % str(e))
- return False
- delta = time.time() - start
- log.debug("WallabyOperations: edit_features %s" % delta)
- return True
+ g = g in self._cache[WBTypes.GROUPS].data and \
+ self._cache[WBTypes.GROUPS].data[g] or None
+ finally:
+ self._lock.release()
+ if g is None:
+ log.debug("WallabyOperations: edit_features, group not found %s" % str(g))
+ else:
+ try:
+ start = time.time()
+ g.modifyFeatures("REPLACE", features)
+ g.update()
+ status = True
+ delta = time.time() - start
+ log.debug("WallabyOperations: edit_features %s" % delta)
+ except Exception, e:
+ log.debug("WallabyOperations: edit_features, exception suppressed, %s" % str(e))
+ return status
def set_interval(self, which, refresh):
'''
@@ -552,17 +574,21 @@
nodes_by_tag[g.name] = nodes
self._lock.acquire()
- self._cache[WBTypes.TAGS].data = self._to_dict(tags)
- self._cache[WBTypes.TAGS].reset_remaining(len(tags) == 0)
- self._nodes_by_tag = nodes_by_tag
- self._lock.release()
+ try:
+ self._cache[WBTypes.TAGS].data = self._to_dict(tags)
+ self._cache[WBTypes.TAGS].reset_remaining(len(tags) == 0)
+ self._nodes_by_tag = nodes_by_tag
+ finally:
+ self._lock.release()
log.debug("WallabyOperations: %s list updated (%s items)" % (WBTypes.TAGS, len(tags)))
def _set_cache(self, attr, data):
self._lock.acquire()
- self._cache[attr].data = self._to_dict(data)
- self._cache[attr].reset_remaining(len(data) == 0)
- self._lock.release()
+ try:
+ self._cache[attr].data = self._to_dict(data)
+ self._cache[attr].reset_remaining(len(data) == 0)
+ finally:
+ self._lock.release()
log.debug("WallabyOperations: %s list updated (%s items)" % (attr, len(data)))
def _to_dict(self, data):
@@ -571,9 +597,11 @@
def _lookup_by_name(self, which, name):
n = None
self._lock.acquire()
- if name in self._cache[which].data:
- n = self._cache[which].data[name]
- self._lock.release()
+ try:
+ if name in self._cache[which].data:
+ n = self._cache[which].data[name]
+ finally:
+ self._lock.release()
return n
class CacheData(object):
12 years, 7 months
- 1
- 0
r4938 - trunk/sage/python/sage/wallaby
by tmckay＠fedoraproject.org
Author: tmckay
Date: 2011-08-29 15:47:22 +0000 (Mon, 29 Aug 2011)
New Revision: 4938
Modified:
trunk/sage/python/sage/wallaby/wallabyoperations.py
Log:
Add edit_features routine to replace feature set on a group.
Modified: trunk/sage/python/sage/wallaby/wallabyoperations.py
===================================================================
--- trunk/sage/python/sage/wallaby/wallabyoperations.py 2011-08-26 18:22:16 UTC (rev 4937)
+++ trunk/sage/python/sage/wallaby/wallabyoperations.py 2011-08-29 15:47:22 UTC (rev 4938)
@@ -434,7 +434,8 @@
node -- a wallaby.Node object or the name of a wallaby.Node object
- tags -- the new set of tags for the node, list or tuple.
+ tags -- the new set of tags for the node,
+ list or tuple of strings
'''
if type(node) in (str, unicode):
n = node
@@ -449,6 +450,9 @@
start = time.time()
n.modifyTags("REPLACE", list(tags), create_missing_tags=True)
n.update()
+ # refresh the groups and tags so that membership informatio
+ # is up to date.
+ self.refresh(WBTypes.GROUPS,WBTypes.TAGS)
except Exception, e:
log.debug("WallabyOperations: edit_tags, exception suppressed, %s" % str(e))
return False
@@ -456,6 +460,34 @@
log.debug("WallabyOperations: edit_tags %s" % delta)
return True
+ def edit_features(self, group, *features):
+ '''
+ Replace existing features in a group with the specified features.
+
+ group -- a wallaby.Group object or the name of a wallaby.Group object
+
+ features -- the new set of features for the group,
+ list or tuple of strings
+ '''
+ if type(group) in (str, unicode):
+ g = group
+ else:
+ g = group.name
+ self._lock.acquire()
+ if g in self._cache[WBTypes.GROUPS].data:
+ g = self._cache[WBTypes.GROUPS].data[g]
+ self._lock.release()
+ try:
+ start = time.time()
+ g.modifyFeatures("REPLACE", features)
+ g.update()
+ except Exception, e:
+ log.debug("WallabyOperations: edit_features, exception suppressed, %s" % str(e))
+ return False
+ delta = time.time() - start
+ log.debug("WallabyOperations: edit_features %s" % delta)
+ return True
+
def set_interval(self, which, refresh):
'''
Set an individual refresh interval for a data item.
12 years, 7 months
- 1
- 0
r4937 - trunk/sage/python/sage/wallaby
by tmckay＠fedoraproject.org
Author: tmckay
Date: 2011-08-26 18:22:16 +0000 (Fri, 26 Aug 2011)
New Revision: 4937
Modified:
trunk/sage/python/sage/wallaby/wallabyoperations.py
Log:
Add basic facilities to cache feature lists, make use of new isTag() and addTag() routines to better support tagging.
BZ733365
BZ733367
Modified: trunk/sage/python/sage/wallaby/wallabyoperations.py
===================================================================
--- trunk/sage/python/sage/wallaby/wallabyoperations.py 2011-08-26 14:18:11 UTC (rev 4936)
+++ trunk/sage/python/sage/wallaby/wallabyoperations.py 2011-08-26 18:22:16 UTC (rev 4937)
@@ -25,6 +25,7 @@
NODES = "nodes"
GROUPS = "groups"
TAGS = "tags"
+ FEATURES = "features"
if imports_ok:
class WallabyOperations(object):
@@ -70,9 +71,10 @@
# an attribute on the Wallaby Store object, with the exception of
# WBTypes.TAGS. The TAGS data is a subset of the GROUPS produced
# in this module.
- self._cache = {WBTypes.NODES: self.CacheData(refresh_interval),
- WBTypes.GROUPS: self.CacheData(refresh_interval),
- WBTypes.TAGS: self.CacheData(refresh_interval,
+ self._cache = {WBTypes.NODES: self.CacheData(refresh_interval),
+ WBTypes.GROUPS: self.CacheData(refresh_interval),
+ WBTypes.FEATURES: self.CacheData(refresh_interval),
+ WBTypes.TAGS: self.CacheData(refresh_interval,
synthetic=self._generate_tag_data)}
# Cache a list of nodes that are members of a tag
@@ -350,6 +352,15 @@
'''
return self._lookup_by_name(WBTypes.TAGS, name)
+ def get_feature_by_name(self, name):
+ '''
+ Return a cached wallaby.Feature object by name.
+
+ If name does not designate a currently cached
+ object, None is returned.
+ '''
+ return self._lookup_by_name(WBTypes.FEATURES, name)
+
def get_node_names(self, tag):
'''
Return a list of node names associated with the tag.
@@ -393,7 +404,7 @@
Refresh the cached lists of groups and tags.
'''
try:
- self._store.addExplicitGroup(name)
+ self._store.addTag(name)
self.refresh((WBTypes.GROUPS,WBTypes.TAGS))
except Exception, e:
log.debug("WallabyOperations: create_tag, exception suppressed, %s" % str(e))
@@ -503,8 +514,7 @@
nodes_by_tag = dict()
for g in groups:
if not g.name.startswith("+++") and \
- g.name != self._partition_group and \
- g.getConfig() == {}:
+ self._store.isTag(g):
tags.append(g)
nodes = g.membership()
nodes_by_tag[g.name] = nodes
12 years, 7 months
- 1
- 0
r4936 - in trunk: cumin/python/cumin cumin/python/cumin/grid sage/python/sage/aviary sage/python/sage/qmf
by tmckay＠fedoraproject.org
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)
12 years, 7 months
- 1
- 0
r4935 - in trunk: cumin/python/cumin cumin/python/cumin/grid sage/python/sage/aviary sage/python/sage/qmf
by tmckay＠fedoraproject.org
Author: tmckay
Date: 2011-08-17 18:17:46 +0000 (Wed, 17 Aug 2011)
New Revision: 4935
Modified:
trunk/cumin/python/cumin/grid/job.py
trunk/cumin/python/cumin/model.py
trunk/sage/python/sage/aviary/aviaryoperations.py
trunk/sage/python/sage/qmf/qmfoperations.py
Log:
Initial ops using aviary query server (fetch_job_data and get_job_ad)
Modified: trunk/cumin/python/cumin/grid/job.py
===================================================================
--- trunk/cumin/python/cumin/grid/job.py 2011-08-17 15:52:36 UTC (rev 4934)
+++ trunk/cumin/python/cumin/grid/job.py 2011-08-17 18:17:46 UTC (rev 4935)
@@ -864,10 +864,9 @@
id = self.frame.id.get(session)
job_server = self.frame.get_job_server(session, id)
job_id = self.frame.job_id.get(session)
- file, start, end = self.get_file_args(session)
+ state, file, start, end = self.get_file_args(session)
if file:
- result = self.app.remote.fetch_job_data(job_server,
- job_id,
+ result = self.app.remote.fetch_job_data(job_server, job_id, state,
file, start, end, {'Data': ""})
if result.error:
return result.status
@@ -882,8 +881,9 @@
else:
start = 0
end = 2048
- file = self.which_file.get_current_file_name(session)
- return (file, start, end)
+ state = self.which_file.get_current_state(session)
+ file = self.which_file.get_file_name(session, state)
+ return (state, file, start, end)
def render_loading(self, session, *args):
file = self.which_file.get_current_file_name(session)
@@ -1010,6 +1010,9 @@
state = self.get(session)
return self.get_file_name(session, state)
+ def get_current_state(self, session):
+ return self.get(session)
+
def select_first_enabled(self, session):
states = self.get_items(session)
disabled = self.disabled.get(session)
Modified: trunk/cumin/python/cumin/model.py
===================================================================
--- trunk/cumin/python/cumin/model.py 2011-08-17 15:52:36 UTC (rev 4934)
+++ trunk/cumin/python/cumin/model.py 2011-08-17 18:17:46 UTC (rev 4935)
@@ -642,7 +642,9 @@
def render_universe(self, session, value):
try:
- return self.universes[value]
+ # might have a string version of universe
+ # value passed in here...
+ return self.universes[int(value)]
except KeyError:
return "Unknown (%s)" % str(value)
Modified: trunk/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- trunk/sage/python/sage/aviary/aviaryoperations.py 2011-08-17 15:52:36 UTC (rev 4934)
+++ trunk/sage/python/sage/aviary/aviaryoperations.py 2011-08-17 18:17:46 UTC (rev 4935)
@@ -85,9 +85,10 @@
self.job_client_pool = JobClientPool(job_wsdl, None)
self.query_client_pool = QueryClientPool(query_wsdl, None)
- self.type_map = self._type_map()
+ self.type_to_aviary = self._type_to_aviary()
+ self.aviary_to_type = self._aviary_to_type()
-# scheduler operations
+# job server operations
def set_job_attribute(self, scheduler, job_id, name, value, callback):
assert callback
@@ -184,7 +185,7 @@
extra.type = "EXPRESSION"
else:
try:
- extra.type = self.type_map[type(v)]
+ extra.type = self.type_to_aviary[type(v)]
except KeyError:
extra.type = "UNDEFINED"
extra.value = v
@@ -224,14 +225,130 @@
"removeJob",
callback, default, timeout)
+# query server ops
+
+ def fetch_job_data(self, job_server, job_id, ftype, file, start, end,
+ default=None, timeout=5):
+ # Aviary doesn't use the file name as does QMF, instead it
+ # specifies the file type and lets condor figure out the path.
+
+ client = self.query_client_pool.get_object()
+
+ def my_process_results(result):
+ # Fix up the exception message if necessary
+ result = self._pretty_result(result, job_server.Machine)
+ if isinstance(result, Exception):
+ status = result
+ 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}
+ return (status, data)
+
+ host = self._get_host(job_server.Machine, self.query_servers)
+ if host == "":
+ self._raise_no_host(job_server.Machine)
+
+ service = host + "getJobData"
+
+ # Have to set the URL for the method. This might go away someday...
+ #print "Aviary control job " + url
+ client.set_options(location=service)
+
+ # Make a job data parameter (see query wsdl)
+ jobData = client.factory.create('ns0:JobData')
+ jobData.id.job = job_id
+ jobData.id.pool = job_server.Pool
+
+ # Translate cumin file type to Aviary file type
+ if ftype == "e":
+ jobData.type = "ERR"
+ elif ftype == "o":
+ jobData.type = "OUT"
+ elif ftype == "u":
+ jobData.type = "LOG"
+ else:
+ # We can't translate the type.
+ # Let Aviary throw an error on this instead of us.
+ jobData.type = ftype
+
+ from_end = start < 0
+ max_bytes = abs(end - start)
+
+ res = self._call_sync(my_process_results, client.service.getJobData,
+ jobData, max_bytes, from_end )
+ self.query_client_pool.return_object(client)
+ return res;
+
+ def get_job_ad(self, job_server, job_id, default=None, timeout=5):
+
+ client = self.query_client_pool.get_object()
+
+ def make_tuple(attr):
+ if attr.type in self.aviary_to_type:
+ try:
+ v = self.aviary_to_type[attr.type](attr.value)
+ except:
+ v = attr.value
+ else:
+ v = attr.value
+ return (attr.name, v)
+
+ def make_dict(attrs):
+ return dict([make_tuple(attr) for attr in attrs])
+
+ def my_process_results(result):
+ # Fix up the exception message if necessary
+ result = self._pretty_result(result, job_server.Machine)
+ if isinstance(result, Exception):
+ status = result
+ 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}
+ return (status, data)
+
+ host = self._get_host(job_server.Machine, self.query_servers)
+ if host == "":
+ self._raise_no_host(job_server.Machine)
+
+ service = host + "getJobDetails"
+
+ # Have to set the URL for the method. This might go away someday...
+ #print "Aviary control job " + url
+ client.set_options(location=service)
+
+ # Make a job id parameter (see job wsdl)
+ 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
+
+ res = self._call_sync(my_process_results,
+ client.service.getJobDetails, jobId)
+ self.query_client_pool.return_object(client)
+ return res;
+
# Secret private implementation stuff, don't look!
@classmethod
- def _type_map(cls):
+ def _type_to_aviary(cls):
# Need to be able to turn simple Python types into Aviary types for attributes
return {int: "INTEGER", float: "FLOAT", str: "STRING", bool: "BOOLEAN"}
@classmethod
+ def _aviary_to_type(cls):
+ return {"INTEGER": int, "FLOAT": float, "STRING": str, "BOOLEAN": bool}
+
+ @classmethod
def _get_host(cls, name, servers):
host = ""
if name in servers:
Modified: trunk/sage/python/sage/qmf/qmfoperations.py
===================================================================
--- trunk/sage/python/sage/qmf/qmfoperations.py 2011-08-17 15:52:36 UTC (rev 4934)
+++ trunk/sage/python/sage/qmf/qmfoperations.py 2011-08-17 18:17:46 UTC (rev 4935)
@@ -119,8 +119,9 @@
def get_job_ad(self, job_server, job_id, default=None, timeout=5):
return self._call(job_server, "GetJobAd", 0, default, timeout, job_id)
- def fetch_job_data(self, job_server, job_id, file, start, end,
+ def fetch_job_data(self, job_server, job_id, ftype, file, start, end,
default=None, timeout=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)
12 years, 8 months
- 1
- 0
r4934 - trunk/cumin/python/cumin/grid
by croberts＠fedoraproject.org
Author: croberts
Date: 2011-08-17 15:52:36 +0000 (Wed, 17 Aug 2011)
New Revision: 4934
Modified:
trunk/cumin/python/cumin/grid/job.strings
Log:
Addressing BZ 731423 by removing extra space insice the textarea tag that was responsible for the extra space in the display.
Modified: trunk/cumin/python/cumin/grid/job.strings
===================================================================
--- trunk/cumin/python/cumin/grid/job.strings 2011-08-16 16:04:54 UTC (rev 4933)
+++ trunk/cumin/python/cumin/grid/job.strings 2011-08-17 15:52:36 UTC (rev 4934)
@@ -164,14 +164,10 @@
</script>
[OutputFile.html]
-<textarea id="{id}" name="job_output" class="job_output" rows="20" cols="80">
- {content}
-</textarea>
+<textarea id="{id}" name="job_output" class="job_output" rows="20" cols="80">{content}</textarea>
[OutputFile.deferred_html]
-<textarea id="{id}" name="job_output" class="job_output" rows="20" cols="80">
- {loading}
-</textarea>
+<textarea id="{id}" name="job_output" class="job_output" rows="20" cols="80">{loading}</textarea>
[JobSelectorTable.error_html]
<div id="{id}">
12 years, 8 months
- 1
- 0