r5658 - trunk/cumin/bin
by tmckay@fedoraproject.org
Author: tmckay
Date: 2013-01-18 18:02:31 +0000 (Fri, 18 Jan 2013)
New Revision: 5658
Modified:
trunk/cumin/bin/cumin
Log:
Re-insert SIGKILL for children after timeout.
Include child termination status in log messages.
Wait additional time after SIGKILL to report child termination before exiting.
BZ800065
Modified: trunk/cumin/bin/cumin
===================================================================
--- trunk/cumin/bin/cumin 2013-01-18 15:32:12 UTC (rev 5657)
+++ trunk/cumin/bin/cumin 2013-01-18 18:02:31 UTC (rev 5658)
@@ -320,17 +320,18 @@
% app[PROCESS].pid)
# Wait for children to exit so we can record the result and so that
- # the initd script can track the pid correctly. Exit regardless after
+ # the initd script can track the pid correctly. Issue SIGKILL after
# timeout+0.5 seconds, even if the children are still running. Each child
# will have a hard exit after timeout seconds from its own main thread,
# even if its shutdown activities have not completed, so in most
# cases the children *should* exit within timeout+0.5 seconds
then = datetime.now()
+ sent_kill = False
while complete != 0:
for app in apps:
poll = app[PROCESS] and app[PROCESS].poll()
if poll is not None:
- log.info("Subprocess (%s) exited", app[PROCESS].pid)
+ log.info("Subprocess (%s) exited (%s)" % (app[PROCESS].pid, poll))
app[PROCESS] = None
complete -= 1
@@ -340,12 +341,24 @@
sleep(0.25)
if datetime.now() - then > timedelta(seconds=options.timeout+0.5):
- log.warn("Timed out waiting for children, exiting")
- for app in apps: # just to be paranoid
- if app[PROCESS]:
- log.warn("Subprocess (%s) failed to stop"\
- % app[PROCESS].pid)
- break
+ if sent_kill:
+ log.warn("Timed out waiting for children, exiting")
+ for app in apps:
+ if app[PROCESS]:
+ log.warn("Subprocess (%s) failed to stop"\
+ % app[PROCESS].pid)
+ break
+ else:
+ log.warn("Timed out waiting for children, sending SIGKILL")
+ for app in apps: # just to be paranoid
+ if app[PROCESS]:
+ os.kill(app[PROCESS].pid, signal.SIGKILL)
+ log.warn("Subprocess (%s) failed to stop, "\
+ "sending SIGKILL" % (app[PROCESS].pid))
+
+ # Okay, let's loop one more time and report
+ sent_kill = True
+ then = datetime.now()
write_status_path(config.get_init_status_path(), "exit\n", append=True)
return return_code
11 years, 3 months
r5655 - in branches/tmckay: cumin/python/cumin cumin/python/cumin/grid mint/python/mint mint/python/mint/aviary sage/python/sage/aviary sage/python/sage/qmf
by tmckay@fedoraproject.org
Author: tmckay
Date: 2013-01-17 21:08:02 +0000 (Thu, 17 Jan 2013)
New Revision: 5655
Modified:
branches/tmckay/cumin/python/cumin/grid/job.py
branches/tmckay/cumin/python/cumin/main.py
branches/tmckay/mint/python/mint/aviary/endpoints.py
branches/tmckay/mint/python/mint/aviary/session.py
branches/tmckay/mint/python/mint/aviary/submissions.py
branches/tmckay/mint/python/mint/main.py
branches/tmckay/mint/python/mint/update.py
branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
branches/tmckay/sage/python/sage/qmf/qmfoperations.py
Log:
Initial version with Aviary submissions working
Modified: branches/tmckay/cumin/python/cumin/grid/job.py
===================================================================
--- branches/tmckay/cumin/python/cumin/grid/job.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/cumin/python/cumin/grid/job.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -106,10 +106,9 @@
def check_submission_membership(self, session, sub_id, job_id):
# Make sure that the job belongs to the designated submission.
- id = self.id.get(session)
frame = None
message = ""
- submission, sched = self.get_submission_sched(session, id)
+ submission, sched = self.get_submission_sched(session, sub_id)
summaries = self.app.model.get_submission_job_summaries(submission,
sched.Name)
Modified: branches/tmckay/cumin/python/cumin/main.py
===================================================================
--- branches/tmckay/cumin/python/cumin/main.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/cumin/python/cumin/main.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -206,7 +206,7 @@
self.aviary_job_servers or self.aviary_query_servers:
try:
from sage.aviary.aviaryoperations import \
- SudsLogging, AviaryOperationsFactory
+ SudsLogging, AviaryOperations
except:
imports_ok = False
if imports_ok:
@@ -224,7 +224,7 @@
# are empty strings. If locator is non empty, their actual
# values will be overridden but the presence of a value will
# still control enable/disable.
- aviary_itf = AviaryOperationsFactory("aviary", aviary_dir,
+ aviary_itf = AviaryOperations("aviary", aviary_dir,
self.aviary_locator,
self.aviary_job_servers,
self.aviary_query_servers,
Modified: branches/tmckay/mint/python/mint/aviary/endpoints.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/endpoints.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/mint/python/mint/aviary/endpoints.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -18,7 +18,7 @@
def __init__(self, url, id, factory):
super(SubmissionEndpoint, self).__init__(url, id)
self.size = 4 # do we want to control size per endpoint record?
- self.query_ops = factory.query_ops
+ self.app = factory.app
# Set up time spans for polling/history and initialize search indices
self.init_query_indices(factory.poll_delta,
@@ -58,11 +58,12 @@
QuerySubmissionsBefore(self),)
class EndpointFactory(object):
- def __init__(self, query_ops, poll_delta, history_delta):
+ def __init__(self, app, poll_delta, history_delta):
+ self.app = app
+
self.poll_delta = poll_delta
self.history_delta = history_delta
- self.query_ops = query_ops
self.type_map = \
{("CUSTOM","QUERY_SERVER"): self._make_submission_endpoint}
Modified: branches/tmckay/mint/python/mint/aviary/session.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/session.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/mint/python/mint/aviary/session.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -42,7 +42,7 @@
self.foreman = Foreman()
- endpoint_factory = EndpointFactory(self.app.aviary_ops,
+ endpoint_factory = EndpointFactory(self.app,
poll_delta=timedelta(days=64),
history_delta=timedelta(weeks=52))
Modified: branches/tmckay/mint/python/mint/aviary/submissions.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/submissions.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/mint/python/mint/aviary/submissions.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -2,20 +2,36 @@
from util import now_minus_duration
from datetime import datetime
+from mint.update import AviarySubmissionUpdate
+
+def max(stuff):
+ d = 0
+ for i in stuff:
+ if i.qdate > d:
+ d = i.qdate
+ return d
+
+def min(stuff):
+ d = None
+ for i in stuff:
+ if d is None or i.qdate < d:
+ d = i.qdate
+ return d
+
class QuerySubmissionsAfter(Work):
def __init__(self, ep):
super(QuerySubmissionsAfter, self).__init__()
self.ep = ep
def do_the_job(self, foreman):
- ids = self.ep.query_ops.get_submission_ids_by_qdate(
+ ids = self.ep.app.aviary_ops.get_submission_ids_by_qdate(
self.ep.url,
self.ep.size,
self.ep.after_index,
"AFTER")
- print ("querying", self.ep.after_index, datetime.fromtimestamp(self.ep.after_index))
- print (ids.status, ids.got_data)
+ #print ("querying after", self.ep.after_index, datetime.fromtimestamp(self.ep.after_index))
+ #print (ids.status, ids.got_data)
# if data.ids is empty (remaining == 0)
# then we should reset qdate to some point in
@@ -23,7 +39,7 @@
# yet. This is how we will accomplish polling,
# just looping over the data continuously.
if ids.status == 0 and ids.got_data:
- print ids.data.ids
+ #print ids.data.ids
# ids.data.ids is a list of SubmissionIDs
# Each of these ids needs to be polled to
@@ -43,10 +59,11 @@
# back an empty list but sets remaining...
# In that case just leave after_index as is
if ids.data.ids:
- self.ep.after_index = ids.data.ids[-1].qdate
+# self.ep.after_index = ids.data.ids[-1].qdate
+ self.ep.after_index = max(ids.data.ids)
else:
- print "query after reset"
- #self.ep.after_index = now_minus_duration(self.ep.poll_delta)
+ #print "query after reset"
+ self.ep.after_index = now_minus_duration(self.ep.poll_delta)
else:
print "query after failed"
@@ -63,14 +80,14 @@
# this should mark self.ep.after_queries_completed
# or self.ep.before_queries_completed as necessary
- ids = self.ep.query_ops.get_submission_ids_by_qdate(
+ ids = self.ep.app.aviary_ops.get_submission_ids_by_qdate(
self.ep.url,
self.ep.size,
self.ep.before_index,
"BEFORE")
- print ("querying", self.ep.before_index, datetime.fromtimestamp(self.ep.before_index))
- print (ids.status, ids.got_data)
+ #print ("querying before", self.ep.before_index, datetime.fromtimestamp(self.ep.before_index))
+ #print (ids.status, ids.got_data)
# if data.ids is empty (remaining == 0)
# then we should reset qdate to some point in
@@ -91,18 +108,20 @@
# the qdate of the last item gives us the
# starting index of the next query
if ids.data.remaining:
- self.ep.before_index = ids.data.ids[-1].qdate
+# self.ep.before_index = ids.data.ids[-1].qdate
+ self.ep.before_index = min(ids.data.ids)
+
if self.ep.before_index <= self.ep.before_cutoff:
- print ("hit before cutoff, finishing",
- datetime.fromtimestamp(self.ep.before_index),
- datetime.fromtimestamp(self.ep.before_cutoff))
+ #print ("hit before cutoff, finishing",
+ # datetime.fromtimestamp(self.ep.before_index),
+ # datetime.fromtimestamp(self.ep.before_cutoff))
self.ep.before_queries_completed = True
- #self.ep.before_index = now_minus_duration(self.ep.poll_delta)
+# self.ep.before_index = now_minus_duration(self.ep.poll_delta)
else:
- print "finished"
+ #print "query before reset"
self.ep.before_queries_completed = True
- #self.ep.before_index = now_minus_duration(self.ep.poll_delta)
+# self.ep.before_index = now_minus_duration(self.ep.poll_delta)
else:
print "query before failed"
@@ -124,7 +143,7 @@
id.Name = id.name
id.Owner = id.owner
- summaries = self.ep.query_ops.get_submission_summaries(
+ summaries = self.ep.app.aviary_ops.get_submission_summaries(
self.ids,
self.ep.url,
include_job_summaries=False)
@@ -141,7 +160,7 @@
for summary in summaries.data:
summary.Name = str(summary.id.name)
summary.Owner = str(summary.id.owner)
- summary.Qdate = summary.id.qdate
+ summary.QDate = summary.id.qdate
summary.Idle = summary.idle
summary.Running = summary.running
@@ -151,8 +170,12 @@
summary.TransferringOutput = summary.transferring_output
summary.Suspended = summary.suspended
- print summary
+ summary.endpoint_id = self.ep.id
+ #print summary.Name
+ a = AviarySubmissionUpdate(self.ep.app.model, summary)
+ self.ep.app.update_thread.enqueue(a)
+
# summary.statistics = {
# properties = Name, Owner, Qdate, jobserverRef, schedulerRef
Modified: branches/tmckay/mint/python/mint/main.py
===================================================================
--- branches/tmckay/mint/python/mint/main.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/mint/python/mint/main.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -130,6 +130,7 @@
else:
log.info("Imports failed for Aviary interface, disabling")
+ # This session type sets up discovery for data objects supported by Aviary
if self.aviary_ops:
self.aviary_session = MintAviarySession(self, self._aviary_classes)
Modified: branches/tmckay/mint/python/mint/update.py
===================================================================
--- branches/tmckay/mint/python/mint/update.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/mint/python/mint/update.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -705,36 +705,37 @@
self.submission = submission
self.cls = model.com_redhat_grid.Submission
- # the qmf type string for an attribute can be gotten
- # from the cls.attr_name.type
- # If you run that through the transforms, you can
- # get the sql type
-
def do_process(self, cursor, stats):
agent_id = self.get_agent_id()
object_id = self.get_object_id()
+
+ # For aviary stuff, the agent is inferred
+ # There really are no separate updates, so
+ # we just base the agent id on info in the submission
try:
agent = self.model.agents_by_id[agent_id]
except KeyError:
- raise UpdateDropped()
+ agent = MintAgent(self.model, agent_id)
+ stats.agents_created += 1
try:
- obj = agent.get_object(cursor, cls, object_id)
+ obj = agent.get_object(cursor, self.cls, object_id)
except RosemaryNotFound:
- obj = self.create_object(cursor, stats, cls)
+ obj = self.create_object(cursor, stats, self.cls)
return
self.update_object(cursor, stats, obj)
def get_agent_id(self):
- return self.cls._package._name + ":" + self.cls._name \
- + ":" + selfself.submission._scheduler
+ return self.submission.endpoint_id.resource + ":" + \
+ self.submission.endpoint_id.sub_type + ":" + \
+ self.submission.endpoint_id.name
def get_object_id(self):
- return self.submission.name
+ return self.submission.id.name
def create_object(self, cursor, stats, cls):
- update_time = datetime.datetime.now()
+ update_time = datetime.now()
create_time = update_time
obj = cls.create_object(cursor)
@@ -773,8 +774,6 @@
sql = "; ".join(statements)
self.execute_sql(cursor, sql, obj.__dict__)
- self.process_deferred_links(cursor, obj)
-
obj._save_time = datetime.now()
self.model.print_event(3, "Created %s", obj)
@@ -784,21 +783,8 @@
return obj
- def process_deferred_links(self, cursor, obj):
- agent = self.model.agents_by_id[obj._qmf_agent_id]
-
- if obj._qmf_object_id not in agent.deferred_links_by_id:
- return
-
- links = agent.deferred_links_by_id[obj._qmf_object_id]
-
- for link in links:
- link.realize(cursor, obj)
-
- del agent.deferred_links_by_id[obj._qmf_object_id]
-
def update_object(self, cursor, stats, obj):
- update_time = datetime.datetime.now()
+ update_time = datetime.now()
create_time = update_time
obj._qmf_update_time = update_time
@@ -844,7 +830,12 @@
#stats.objects_updated_by_class[cls] += 1
def transform_value(self, prop, value):
- t = qmf_type_code_by_string[attr.type]
+ # For some reason the timestamps in QMF values
+ # are in 10s of microseconds, so we need to
+ # handle our own conversion here
+ if prop.type == "absTime":
+ return datetime.fromtimestamp(value)
+ t = qmf_type_code_by_string[prop.type]
return transformers[t](value)
def process_properties(self, obj, columns, cursor):
@@ -859,22 +850,25 @@
class FakeOID(object):
def __init__(self, agent, objname):
- self.isV2 = True
self.agentName = agent
self.objectName = objname
- val = FakeOID("com.redhat.grid:scheduler:"+self.submission.id._scheduler,
- self.submission.id._scheduler)
+ val = FakeOID("com.redhat.grid:scheduler:" + \
+ self.submission.id._scheduler,
+ self.submission.id._scheduler)
prop = FakeProp("schedulerRef")
- col, nvalue = self.process_reference(obj, prop, val, cursor)
- if nvalue != getattr(obj, col.name):
- setattr(obj, col.name, nvalue)
- columns.append(col)
+ try:
+ col, nvalue = self.process_reference(obj, prop, val, cursor)
+ if nvalue != getattr(obj, col.name):
+ setattr(obj, col.name, nvalue)
+ columns.append(col)
+ except MappingException, e:
+ log.debug(e)
for name, prop in cls._properties_by_name.iteritems():
value = getattr(self.submission, name)
if value is not None:
- value = transform_value(prop, value)
+ value = self.transform_value(prop, value)
if value == getattr(obj, prop.sql_column.name):
continue
setattr(obj, prop.sql_column.name, value)
@@ -889,44 +883,39 @@
if not ref.sql_column:
raise MappingException("Reference %s has no column" % ref.name)
+ # Okay, for submissions coming from Aviary we may actually see
+ # a submission before we see its scheduler. So, let's go ahead
+ # and assume the agent and create a deferred link.
+ # If the agent never shows up, no harm done. We should probably
+ # add a timeout to the agent and delete it if it does not show up.
value = None
+ try:
+ agent = self.model.agents_by_id[oid.agentName]
+ except KeyError:
+ agent = MintAgent(self.model, oid.agentName)
- if oid:
- if oid.isV2:
- agent_id = oid.agentName
- else:
- # Not much we can do but assume same agent
- agent_id = self.get_agent_id()
+ object_id = oid.objectName
+ try:
+ that = agent.get_object(cursor, ref.that_cls, object_id)
+ except RosemaryNotFound:
+ link = DeferredLink(obj, ref)
+ agent.deferred_links_by_id[object_id].append(link)
- try:
- agent = self.model.agents_by_id[agent_id]
- except KeyError:
- raise MappingException("Agent %s is unknown" % agent_id)
+ msg = "Deferring link to object %s %s"
+ raise MappingException(msg % (ref.that_cls, object_id))
- object_id = oid.objectName
+ value = that._id
- try:
- that = agent.get_object(cursor, ref.that_cls, object_id)
- except RosemaryNotFound:
- link = DeferredLink(obj, ref)
- agent.deferred_links_by_id[object_id].append(link)
-
- msg = "Deferring link to object %s %s"
- raise MappingException(msg % (ref.that_cls, object_id))
-
- value = that._id
-
return ref.sql_column, value
def process_statistics(self, obj, update_columns, insert_columns):
build_columns = list()
saw_change = False
-
# get a list of statistic names from the cls
# loop through the list of properties, look for
# the value in the object
- for name, stat in cls._statistics_by_name.iteritems():
+ for name, stat in self.cls._statistics_by_name.iteritems():
value = getattr(self.submission, name)
if value is not None:
value = self.transform_value(stat, value)
@@ -973,10 +962,9 @@
def __repr__(self):
name = self.__class__.__name__
agent_id = self.get_agent_id()
- cls = self.qmf_object.getClassKey().getClassName()
obj_id = self.get_object_id()
- return "%s(%s,%s,%s)" % (name, agent_id, cls, obj_id)
+ return "%s(%s,%s,%s)" % (name, agent_id, self.cls, obj_id)
class UpdateDropped(Exception):
pass
@@ -1001,5 +989,5 @@
transformers[14] = str
transformers[15] = transform_pickle
-def transform_value(attr, value):
+def transform_value(attr, value):
return transformers[attr.type](value)
Modified: branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- branches/tmckay/sage/python/sage/aviary/aviaryoperations.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/sage/python/sage/aviary/aviaryoperations.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -771,7 +771,7 @@
self.client_pool.return_object(client)
return res;
- def get_job_ad(self, job_server, job_id, scheduler_name, submission,
+ def get_job_ad(self, sched, job_id, submission,
*args, **kwargs):
'''
kwargs will be searched for 'default' and 'timeout' arguments.
@@ -802,7 +802,7 @@
def my_process_results(result):
data = None
# Fix up the exception message if necessary
- result = self._pretty_result(result, job_server.Machine)
+ result = self._pretty_result(result, sched.Machine)
if isinstance(result, Exception):
status = result
else:
@@ -819,14 +819,14 @@
client = self.client_pool.get_object()
self._setup_client(client,
self.servers, # server lookup object
- job_server.Machine, # host we want
+ sched.Machine, # host we want
"getJobDetails")
# Make a job id parameter (see job wsdl)
jobId = client.factory.create('ns0:JobID')
jobId.job = job_id
- jobId.pool = job_server.Pool
- jobId.scheduler = scheduler_name
+ jobId.pool = sched.Pool
+ jobId.scheduler = sched.Name
jobId.submission.name = submission.Name
jobId.submission.owner = submission.Owner
Modified: branches/tmckay/sage/python/sage/qmf/qmfoperations.py
===================================================================
--- branches/tmckay/sage/python/sage/qmf/qmfoperations.py 2013-01-17 19:41:02 UTC (rev 5654)
+++ branches/tmckay/sage/python/sage/qmf/qmfoperations.py 2013-01-17 21:08:02 UTC (rev 5655)
@@ -164,21 +164,6 @@
assert callback
return self._call(master, "Stop", callback, 0, 0, daemon)
-# submission operations
-
- def get_job_summaries(self, submission, callback, *args):
- '''
- This method is asynchronous iff 'callback' is supplied.
-
- kwargs will be searched for 'callback', 'default' and 'timeout' arguments.
- '''
-
- 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(submission, "GetJobSummaries", callback, default, timeout)
-
# Secret private implementation stuff, don't look!
def _call(self, obj, meth, cb, dflt, tout, *args):
11 years, 3 months
r5654 - branches/elephant/cumin/python/cumin/gridhadoop
by croberts@fedoraproject.org
Author: croberts
Date: 2013-01-17 19:41:02 +0000 (Thu, 17 Jan 2013)
New Revision: 5654
Modified:
branches/elephant/cumin/python/cumin/gridhadoop/datanode.py
branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py
branches/elephant/cumin/python/cumin/gridhadoop/main.py
branches/elephant/cumin/python/cumin/gridhadoop/namenode.py
branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py
Log:
Another gridoop UI pit stop.
Modified: branches/elephant/cumin/python/cumin/gridhadoop/datanode.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/datanode.py 2013-01-16 19:27:04 UTC (rev 5653)
+++ branches/elephant/cumin/python/cumin/gridhadoop/datanode.py 2013-01-17 19:41:02 UTC (rev 5654)
@@ -8,6 +8,7 @@
from wooly.forms import StringField, ScalarField, OptionInputSet
from hadoop import *
+from hadoopobjectframe import *
from sage.util import *
@@ -44,7 +45,7 @@
def __init__(self, app, name, cls):
super(DataNodeTable, self).__init__(app, name, cls)
- self.id_col = ObjectTableColumn(app, "idcol", cls.Id)
+ self.id_col = self.DataNodeColumn(app, "idcol", cls.Id, cls.Id, ".datanodeframe")
self.sub_col = SubmittedColumn(app, "sub", cls.Submitted)
self.state_col = ObjectTableColumn(app, "statecol", cls.State)
self.uptime_col = UptimeColumn(app, "uptimecol", cls.Uptime)
@@ -56,6 +57,14 @@
self.add_column(self.uptime_col)
self.add_column(self.owner_col)
+ class DataNodeColumn(ObjectLinkColumn):
+ def render_cell_href(self, session, record):
+ id = unescape_entity(record[self.id_field.index])
+ #need to make this work for user/grid-hadoop path too somehow
+ # this is way less than ideal
+ frame = self.page.page_widgets_by_path[self.path.rsplit(".", 11)[0] + self.frame_path]
+ return frame.get_href(session, id, "grid2.lab.bos.redhat.com")
+
class DataNodeDelete(ObjectSelectorTask):
def __init__(self, app, selector, name):
super(DataNodeDelete, self).__init__(app, selector)
@@ -189,5 +198,59 @@
def render_title(self, session):
return "Number of data nodes to start"
+class DataNodeFrame(HadoopObjectFrame):
+ def __init__(self, app, name):
+ super(DataNodeFrame, self).__init__(app, name)
+ self.view = DataNodeObjectView(app, "view", self.object)
+ self.replace_child(self.view)
+ self.overview = DataNodeOverview(app, self.name + ".dnoverview", self.object)
+ self.view.add_tab(self.overview)
+
+ def get_title(self, session):
+ retval = ""
+ try:
+ retval = "Data node:'%s' (%s)" % (xml_escape(self.object.get(session).name), xml_escape(self.object.get(session).id))
+ except Exception, e:
+ pass
+ return retval
+
+ def get_object(self, session, id):
+ class Object(object):
+ pass
+ fake = Object()
+ fake.name = "fake data node object"
+ fake.id = id
+ return fake
+
+class DataNodeObjectView(HadoopObjectView):
+ def add_details_tab(self):
+ self.add_tab(DataNodeDetails(self.app, self.name + ".details", self.object))
+
+ def render_title(self, session):
+ obj = self.object.get(session)
+ if obj is not None:
+ return "Data node: %s" % xml_escape(obj.id)
+ else:
+ return ""
+
+class DataNodeOverview(Widget):
+ def __init__(self, app, name, node):
+ super(DataNodeOverview, self).__init__(app, name)
+
+ def render_title(self, session):
+ return "Overview"
+
+class DataNodeDetails(Widget):
+ def __init__(self, app, name, object):
+ super(DataNodeDetails, self).__init__(app, name)
+ self.object = object
+
+ def render_title(self, session):
+ return "Details"
+
+ def render_name(self, session):
+ '''sample....would only be called if DataNodeDetails display was overridden to say {name}'''
+ obj = self.object.get(session)
+ return obj.name
\ No newline at end of file
Modified: branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py 2013-01-16 19:27:04 UTC (rev 5653)
+++ branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py 2013-01-17 19:41:02 UTC (rev 5654)
@@ -10,7 +10,7 @@
from hadoop import *
from hadoopobjectframe import *
-from tasktracker import TaskTrackerSelector
+from tasktracker import TaskTrackerSelector, TaskTrackerFrame
from sage.util import *
@@ -177,6 +177,9 @@
self.view = JobTrackerObjectView(app, "view", self.object)
self.replace_child(self.view)
+ self.task_tracker_frame = TaskTrackerFrame(app, "tasktrackerframe")
+ self.add_mode(self.task_tracker_frame)
+
self.overview = JobTrackerOverview(app, self.name + ".jtoverview", self.object)
self.view.add_tab(self.overview)
Modified: branches/elephant/cumin/python/cumin/gridhadoop/main.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/main.py 2013-01-16 19:27:04 UTC (rev 5653)
+++ branches/elephant/cumin/python/cumin/gridhadoop/main.py 2013-01-17 19:41:02 UTC (rev 5654)
@@ -7,6 +7,7 @@
from cumin.util import *
from namenode import NameNodeFrame
+from datanode import DataNodeFrame
from jobtracker import JobTrackerFrame
from namenode import NameNodeSelector
Modified: branches/elephant/cumin/python/cumin/gridhadoop/namenode.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/namenode.py 2013-01-16 19:27:04 UTC (rev 5653)
+++ branches/elephant/cumin/python/cumin/gridhadoop/namenode.py 2013-01-17 19:41:02 UTC (rev 5654)
@@ -8,7 +8,7 @@
from hadoop import *
from hadoopobjectframe import *
-from datanode import DataNodeSelector
+from datanode import DataNodeSelector, DataNodeFrame
from sage.util import *
@@ -67,11 +67,9 @@
#need to make this work for user/grid-hadoop path too somehow
# this is way less than ideal
frame = self.page.page_widgets_by_path[self.path.rsplit(".", 7)[0] + self.frame_path]
-
- #if isinstance(frame, TagObjectFrame):
+
return frame.get_href(session, id, "grid2.lab.bos.redhat.com")
-# else:
-# return id #not able to get a link
+
class NameNodeCreate(Task):
def __init__(self, app, name):
@@ -162,12 +160,12 @@
def __init__(self, app, name):
super(NameNodeFrame, self).__init__(app, name)
-# self.set_tags = TagsNodeEditTask(app, self)
-# self.set_nodes = TagsTagEditTask(app, self)
-# self.set_features = TagsFeatureEditTask(app, self)
self.view = NameNodeObjectView(app, "view", self.object)
self.replace_child(self.view)
+ self.data_node_frame = DataNodeFrame(app, "datanodeframe")
+ self.add_mode(self.data_node_frame)
+
self.overview = NameNodeOverview(app, self.name + ".nnoverview", self.object)
self.view.add_tab(self.overview)
Modified: branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py 2013-01-16 19:27:04 UTC (rev 5653)
+++ branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py 2013-01-17 19:41:02 UTC (rev 5654)
@@ -8,6 +8,7 @@
from wooly.forms import StringField
from hadoop import *
+from hadoopobjectframe import *
from sage.util import *
@@ -44,7 +45,7 @@
def __init__(self, app, name, cls):
super(TaskTrackerTable, self).__init__(app, name, cls)
- self.id_col = ObjectTableColumn(app, "idcol", cls.Id)
+ self.id_col = self.TaskTrackerColumn(app, "idcol", cls.Id, cls.Id, ".tasktrackerframe")
self.sub_col = SubmittedColumn(app, "sub", cls.Submitted)
self.state_col = ObjectTableColumn(app, "statecol", cls.State)
self.uptime_col = UptimeColumn(app, "uptimecol", cls.Uptime)
@@ -56,6 +57,14 @@
self.add_column(self.uptime_col)
self.add_column(self.owner_col)
+ class TaskTrackerColumn(ObjectLinkColumn):
+ def render_cell_href(self, session, record):
+ id = unescape_entity(record[self.id_field.index])
+ #need to make this work for user/grid-hadoop path too somehow
+ # this is way less than ideal
+ frame = self.page.page_widgets_by_path[self.path.rsplit(".", 11)[0] + self.frame_path]
+ return frame.get_href(session, id, "grid2.lab.bos.redhat.com")
+
class TaskTrackerDelete(ObjectSelectorTask):
def __init__(self, app, selector, name):
super(TaskTrackerDelete, self).__init__(app, selector)
@@ -248,3 +257,61 @@
def render_item_selected_attr(self, session, item):
if item == self.param.get(session):
return "selected=\"selected\""
+
+
+class TaskTrackerFrame(HadoopObjectFrame):
+ def __init__(self, app, name):
+ super(TaskTrackerFrame, self).__init__(app, name)
+
+ self.view = TaskTrackerObjectView(app, "view", self.object)
+ self.replace_child(self.view)
+
+ self.overview = TaskTrackerOverview(app, self.name + ".ttoverview", self.object)
+ self.view.add_tab(self.overview)
+
+ def get_title(self, session):
+ retval = ""
+ try:
+ retval = "Task tracker:'%s' (%s)" % (xml_escape(self.object.get(session).name), xml_escape(self.object.get(session).id))
+ except Exception, e:
+ pass
+ return retval
+
+ def get_object(self, session, id):
+ class Object(object):
+ pass
+ fake = Object()
+ fake.name = "fake task tracker object"
+ fake.id = id
+ return fake
+
+class TaskTrackerObjectView(HadoopObjectView):
+ def add_details_tab(self):
+ self.add_tab(TaskTrackerDetails(self.app, self.name + ".details", self.object))
+
+ def render_title(self, session):
+ obj = self.object.get(session)
+ if obj is not None:
+ return "Task tracker: %s" % xml_escape(obj.id)
+ else:
+ return ""
+
+class TaskTrackerOverview(Widget):
+ def __init__(self, app, name, node):
+ super(TaskTrackerOverview, self).__init__(app, name)
+
+ def render_title(self, session):
+ return "Overview"
+
+class TaskTrackerDetails(Widget):
+ def __init__(self, app, name, object):
+ super(TaskTrackerDetails, self).__init__(app, name)
+ self.object = object
+
+ def render_title(self, session):
+ return "Details"
+
+ def render_name(self, session):
+ '''sample....would only be called if TaskTrackerDetails display was overridden to say {name}'''
+ obj = self.object.get(session)
+ return obj.name
\ No newline at end of file
11 years, 3 months
r5653 - branches/elephant/cumin/python/cumin/gridhadoop
by croberts@fedoraproject.org
Author: croberts
Date: 2013-01-16 19:27:04 +0000 (Wed, 16 Jan 2013)
New Revision: 5653
Added:
branches/elephant/cumin/python/cumin/gridhadoop/hadoopobjectframe.py
Modified:
branches/elephant/cumin/python/cumin/gridhadoop/datanode.py
branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py
branches/elephant/cumin/python/cumin/gridhadoop/main.py
branches/elephant/cumin/python/cumin/gridhadoop/namenode.py
branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py
Log:
Starting rework of gridoop UI
Modified: branches/elephant/cumin/python/cumin/gridhadoop/datanode.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/datanode.py 2013-01-15 22:40:03 UTC (rev 5652)
+++ branches/elephant/cumin/python/cumin/gridhadoop/datanode.py 2013-01-16 19:27:04 UTC (rev 5653)
@@ -39,7 +39,7 @@
def render_title(self, session):
return "HDFS data nodes"
-
+
class DataNodeTable(ObjectQmfSelectorTable):
def __init__(self, app, name, cls):
super(DataNodeTable, self).__init__(app, name, cls)
@@ -97,10 +97,9 @@
super(DataNodeCreate, self).__init__(app, name)
cls = app.model.com_redhat_cumin_grid_hadoop.DataNode
- self.form = DataNodeCreateForm(app, self.name, self, cls)
+ self.form = DataNodeCreateForm(app, self.name + ".createform", self, cls)
self.invoc = None
-
-
+
def callback(self, result):
if result == False:
self.invoc.status = self.invoc.FAILED
@@ -124,23 +123,42 @@
def get_title(self, session, x):
return "Create data nodes"
+ def enter(self, session, obj):
+ form_session = wooly.Session(self.app.form_page)
+ if obj:
+ self.form.id.set(form_session, obj._id)
+
+ ## TODO find a better way to do this
+ self.form.nn_id.set(form_session, self.app.gridhadoop.frame.children_by_name["gridhadoop.namenodeframe"].id.get(session))
+ self.form.return_url.set(form_session, session.marshal())
+ self.form.show(form_session)
+
+ self.do_enter(session, obj, form_session)
+ return form_session
+
+ def get_href(self, session, obj):
+
+ return self.enter(session, obj).marshal()
+
+
class DataNodeCreateForm(HadoopNodeCreateForm):
def __init__(self, app, name, task, cls):
super(DataNodeCreateForm, self).__init__(app, name, task, cls)
- self.nameNode = NameNodeField(app, "namenode")
- self.add_field(self.nameNode)
-
- self.binfile = BinFileLoc(app, "binfile")
+ self.binfile = BinFileLoc(app, name + ".binfile")
self.add_field(self.binfile)
- self.count = self.CountField(app, "count")
+ self.count = self.CountField(app, name + ".count")
self.add_field(self.count)
- self.hadoophost = HadoopHostField(app, "hadoophost")
+ self.hadoophost = HadoopHostField(app, name + ".hadoophost")
self.add_field(self.hadoophost)
-
+
+ self.nn_id = Parameter(app, "nn_id")
+ self.add_parameter(self.nn_id)
+
+
def process_display(self, session):
self.scheduler.validate(session)
@@ -155,7 +173,7 @@
owner = session.client_session.username()
hadoophost = self.hadoophost.get(session)
count = self.count.get(session)
- name_node = self.nameNode.get(session)
+ name_node = self.nn_id.get(session)
self.task.invoke(session, None, (binfile, owner, hadoophost, count, name_node))
self.task.exit_with_redirect(session, url)
Added: branches/elephant/cumin/python/cumin/gridhadoop/hadoopobjectframe.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/hadoopobjectframe.py (rev 0)
+++ branches/elephant/cumin/python/cumin/gridhadoop/hadoopobjectframe.py 2013-01-16 19:27:04 UTC (rev 5653)
@@ -0,0 +1,60 @@
+from cumin.widgets import *
+from cumin.parameters import *
+from cumin.objectframe import ObjectView
+
+class HadoopObjectFrame(CuminFrame):
+ '''
+ This frame houses the view page for a hadoop object
+ '''
+ def __init__(self, app, name):
+ super(HadoopObjectFrame, self).__init__(app, name)
+
+ # This will hold the id of the object that should be assigned
+ # to self.object. The id and the object are typically determined during
+ # the "process" pass before a page is rendered.
+ self.id = StringParameter(app, "id")
+ self.add_parameter(self.id)
+
+ self.host = StringParameter(app, "host")
+ self.add_parameter(self.host)
+
+ # This will be given a value during the "process" pass after
+ # self.id is determined (lookup by id)
+ self.object = Attribute(app, "object")
+ self.add_attribute(self.object)
+ self.view = HadoopObjectView(app, "view", self.object)
+ self.add_child(self.view)
+ self.icon_href = "resource?name=action-36.png"
+ self.tasks = list()
+
+ def init(self):
+ super(HadoopObjectFrame, self).init()
+ assert self
+ for task in self.tasks:
+ task.init()
+
+ def get_href(self, session, id, host):
+ branch = session.branch()
+ self.id.set(branch, id)
+ self.host.set(branch, host)
+ self.view.show(branch)
+ return branch.marshal()
+
+ def do_process(self, session):
+ id = self.id.get(session)
+ assert id
+ obj = self.get_object(session, id)
+ self.object.set(session, obj)
+ super(HadoopObjectFrame, self).do_process(session)
+
+
+class HadoopObjectView(ObjectView):
+ def render_title(self, session):
+ obj = self.object.get(session)
+ if obj is not None:
+ return "%s" % xml_escape(obj.id)
+ else:
+ return ""
+
+ def add_details_tab(self):
+ pass
\ No newline at end of file
Property changes on: branches/elephant/cumin/python/cumin/gridhadoop/hadoopobjectframe.py
___________________________________________________________________
Added: svn:mime-type
+ text/plain
Modified: branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py 2013-01-15 22:40:03 UTC (rev 5652)
+++ branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py 2013-01-16 19:27:04 UTC (rev 5653)
@@ -8,7 +8,10 @@
from wooly.forms import StringField
from hadoop import *
+from hadoopobjectframe import *
+from tasktracker import TaskTrackerSelector
+
from sage.util import *
class JobTrackerAdapter(HadoopAdapter):
@@ -44,7 +47,7 @@
def __init__(self, app, name, cls):
super(JobTrackerTable, self).__init__(app, name, cls)
- self.id_col = ObjectTableColumn(app, "idcol", cls.Id)
+ self.id_col = self.JobTrackerColumn(app, "idcol", cls.Id, cls.Id, ".jobtrackerframe")
self.ipc_col = ObjectTableColumn(app, "ipc", cls.Ipc)
self.ipc_col.width = "20%"
self.sub_col = SubmittedColumn(app, "sub", cls.Submitted)
@@ -59,6 +62,15 @@
self.add_column(self.uptime_col)
self.add_column(self.owner_col)
+ class JobTrackerColumn(ObjectLinkColumn):
+ def render_cell_href(self, session, record):
+ id = unescape_entity(record[self.id_field.index])
+ #need to make this work for user/grid-hadoop path too somehow
+ frame = self.page.page_widgets_by_path[self.path.rsplit(".", 7)[0] + self.frame_path]
+
+ #if isinstance(frame, TagObjectFrame):
+ return frame.get_href(session, id, "grid2.lab.bos.redhat.com")
+
class JobTrackerDelete(ObjectSelectorTask):
def __init__(self, app, selector, name):
super(JobTrackerDelete, self).__init__(app, selector)
@@ -157,3 +169,54 @@
def render_title(self, session):
return "Create a job tracker"
+
+class JobTrackerFrame(HadoopObjectFrame):
+ def __init__(self, app, name):
+ super(JobTrackerFrame, self).__init__(app, name)
+
+ self.view = JobTrackerObjectView(app, "view", self.object)
+ self.replace_child(self.view)
+
+ self.overview = JobTrackerOverview(app, self.name + ".jtoverview", self.object)
+ self.view.add_tab(self.overview)
+
+ def get_title(self, session):
+ retval = ""
+ try:
+ retval = "Job tracker:'%s' (%s)" % (xml_escape(self.object.get(session).name), xml_escape(self.object.get(session).id))
+ except Exception, e:
+ pass
+ return retval
+
+ def get_object(self, session, id):
+ class Object(object):
+ pass
+ fake = Object()
+ fake.name = "fake object"
+ fake.id = id
+ return fake
+
+class JobTrackerObjectView(HadoopObjectView):
+ def add_details_tab(self):
+ self.add_tab(JobTrackerDetails(self.app, "details", self.object))
+
+class JobTrackerOverview(Widget):
+ def __init__(self, app, name, node):
+ super(JobTrackerOverview, self).__init__(app, name)
+ self.add_child(TaskTrackerSelector(app, self.name + ".ttforjt", False))
+
+ def render_title(self, session):
+ return "Task trackers"
+
+class JobTrackerDetails(Widget):
+ def __init__(self, app, name, object):
+ super(JobTrackerDetails, self).__init__(app, name)
+ self.object = object
+
+ def render_title(self, session):
+ return "Details"
+
+ def render_name(self, session):
+ '''sample....would only be called if JobTrackerDetails display was overridden to say {name}'''
+ obj = self.object.get(session)
+ return obj.name
\ No newline at end of file
Modified: branches/elephant/cumin/python/cumin/gridhadoop/main.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/main.py 2013-01-15 22:40:03 UTC (rev 5652)
+++ branches/elephant/cumin/python/cumin/gridhadoop/main.py 2013-01-16 19:27:04 UTC (rev 5653)
@@ -6,6 +6,9 @@
from cumin.widgets import *
from cumin.util import *
+from namenode import NameNodeFrame
+from jobtracker import JobTrackerFrame
+
from namenode import NameNodeSelector
from datanode import DataNodeSelector
from jobtracker import JobTrackerSelector
@@ -34,6 +37,12 @@
self.view = GridHadoopView(app, name + ".view", check_owner)
self.add_mode(self.view)
+ self.name_node_frame = NameNodeFrame(app, name + ".namenodeframe")
+ self.add_mode(self.name_node_frame)
+
+ self.job_tracker_frame = JobTrackerFrame(app, name + ".jobtrackerframe")
+ self.add_mode(self.job_tracker_frame)
+
def render_title(self, session):
return "Grid-hadoop"
@@ -48,7 +57,4 @@
self.add_child(self.tabs)
self.tabs.add_tab(NameNodeSelector(app, name + ".namenodes", check_owner))
- self.tabs.add_tab(DataNodeSelector(app, name + ".datanodes", check_owner))
- self.tabs.add_tab(JobTrackerSelector(app, name + ".jobtrackers", check_owner))
- self.tabs.add_tab(TaskTrackerSelector(app, name + ".tasktrackers", check_owner))
-
+ self.tabs.add_tab(JobTrackerSelector(app, name + ".jobtrackers", check_owner))
Modified: branches/elephant/cumin/python/cumin/gridhadoop/namenode.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/namenode.py 2013-01-15 22:40:03 UTC (rev 5652)
+++ branches/elephant/cumin/python/cumin/gridhadoop/namenode.py 2013-01-16 19:27:04 UTC (rev 5653)
@@ -7,6 +7,8 @@
from wooly.forms import StringField
from hadoop import *
+from hadoopobjectframe import *
+from datanode import DataNodeSelector
from sage.util import *
@@ -43,7 +45,8 @@
def __init__(self, app, name, cls):
super(NameNodeTable, self).__init__(app, name, cls)
- self.id_col = ObjectTableColumn(app, "idcol", cls.Id)
+ #probably doesnt' work for user/grid-hadoop path
+ self.id_col = self.NameNodeColumn(app, "idcol", cls.Id, cls.Id, ".namenodeframe")
self.ipc_col = ObjectTableColumn(app, "ipc", cls.Ipc)
self.ipc_col.width = "20%"
self.sub_col = SubmittedColumn(app, "sub", cls.Submitted)
@@ -58,6 +61,18 @@
self.add_column(self.uptime_col)
self.add_column(self.owner_col)
+ class NameNodeColumn(ObjectLinkColumn):
+ def render_cell_href(self, session, record):
+ id = unescape_entity(record[self.id_field.index])
+ #need to make this work for user/grid-hadoop path too somehow
+ # this is way less than ideal
+ frame = self.page.page_widgets_by_path[self.path.rsplit(".", 7)[0] + self.frame_path]
+
+ #if isinstance(frame, TagObjectFrame):
+ return frame.get_href(session, id, "grid2.lab.bos.redhat.com")
+# else:
+# return id #not able to get a link
+
class NameNodeCreate(Task):
def __init__(self, app, name):
super(NameNodeCreate, self).__init__(app, name)
@@ -143,3 +158,56 @@
def render_title(self, session):
return "Remove name nodes"
+class NameNodeFrame(HadoopObjectFrame):
+ def __init__(self, app, name):
+ super(NameNodeFrame, self).__init__(app, name)
+
+# self.set_tags = TagsNodeEditTask(app, self)
+# self.set_nodes = TagsTagEditTask(app, self)
+# self.set_features = TagsFeatureEditTask(app, self)
+ self.view = NameNodeObjectView(app, "view", self.object)
+ self.replace_child(self.view)
+
+ self.overview = NameNodeOverview(app, self.name + ".nnoverview", self.object)
+ self.view.add_tab(self.overview)
+
+ def get_title(self, session):
+ retval = ""
+ try:
+ retval = "Name node:'%s' (%s)" % (xml_escape(self.object.get(session).name), xml_escape(self.object.get(session).id))
+ except Exception, e:
+ pass
+ return retval
+
+ def get_object(self, session, id):
+ class Object(object):
+ pass
+ fake = Object()
+ fake.name = "fake object"
+ fake.id = id
+ return fake
+
+class NameNodeObjectView(HadoopObjectView):
+ def add_details_tab(self):
+ self.add_tab(NameNodeDetails(self.app, self.name + ".details", self.object))
+
+class NameNodeOverview(Widget):
+ def __init__(self, app, name, node):
+ super(NameNodeOverview, self).__init__(app, name)
+ self.add_child(DataNodeSelector(app, self.name + ".dnfornn", False))
+
+ def render_title(self, session):
+ return "Data nodes"
+
+class NameNodeDetails(Widget):
+ def __init__(self, app, name, object):
+ super(NameNodeDetails, self).__init__(app, name)
+ self.object = object
+
+ def render_title(self, session):
+ return "Details"
+
+ def render_name(self, session):
+ '''sample....would only be called if NameNodeDetails display was overridden to say {name}'''
+ obj = self.object.get(session)
+ return obj.name
\ No newline at end of file
Modified: branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py 2013-01-15 22:40:03 UTC (rev 5652)
+++ branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py 2013-01-16 19:27:04 UTC (rev 5653)
@@ -115,8 +115,22 @@
except Exception, e:
invoc.status = invoc.FAILED
log.debug("Creating task trackers failed", exc_info=True)
- invoc.end()
+ invoc.end()
+
+ def enter(self, session, obj):
+ form_session = wooly.Session(self.app.form_page)
+ if obj:
+ self.form.id.set(form_session, obj._id)
+
+ ## TODO find a better way to do this
+ self.form.jt_id.set(form_session, self.app.gridhadoop.frame.children_by_name["gridhadoop.jobtrackerframe"].id.get(session))
+ self.form.return_url.set(form_session, session.marshal())
+ self.form.show(form_session)
+
+ self.do_enter(session, obj, form_session)
+ return form_session
+
def get_title(self, session, x):
return "Create task trackers"
@@ -125,9 +139,6 @@
def __init__(self, app, name, task, cls):
super(TaskTrackerCreateForm, self).__init__(app, name, task, cls)
- self.jobTracker = JobTrackerField(app, "jobtracker")
- self.add_field(self.jobTracker)
-
self.binfile = BinFileLoc(app, "binfile")
self.add_field(self.binfile)
@@ -136,6 +147,9 @@
self.hadoophost = HadoopHostField(app, "hadoophost")
self.add_field(self.hadoophost)
+
+ self.jt_id = Parameter(app, "jt_id")
+ self.add_parameter(self.jt_id)
def process_display(self, session):
self.scheduler.validate(session)
@@ -151,7 +165,7 @@
owner = session.client_session.username()
hadoophost = self.hadoophost.get(session)
count = self.count.get(session)
- job_tracker = self.jobTracker.get(session)
+ job_tracker = self.jt_id.get(session)
self.task.invoke(session, None, (binfile, owner, hadoophost, count, job_tracker))
self.task.exit_with_redirect(session, url)
11 years, 3 months
r5652 - in branches/tmckay: cumin/bin mint/python/mint/aviary mint/python/mint/plumage
by tmckay@fedoraproject.org
Author: tmckay
Date: 2013-01-15 22:40:03 +0000 (Tue, 15 Jan 2013)
New Revision: 5652
Modified:
branches/tmckay/cumin/bin/cumin-report
branches/tmckay/mint/python/mint/aviary/endpoints.py
branches/tmckay/mint/python/mint/aviary/locator.py
branches/tmckay/mint/python/mint/aviary/session.py
branches/tmckay/mint/python/mint/plumage/main.py
Log:
Start up locator thread from mint/aviary/session.py
Modified: branches/tmckay/cumin/bin/cumin-report
===================================================================
--- branches/tmckay/cumin/bin/cumin-report 2013-01-15 00:01:24 UTC (rev 5651)
+++ branches/tmckay/cumin/bin/cumin-report 2013-01-15 22:40:03 UTC (rev 5652)
@@ -33,7 +33,7 @@
from mint.plumage.main import Plumage
mint = Plumage(home,
- ("model/plumage", "model/admin")
+ ("model/plumage", "model/admin"),
values.plumage_host,
values.plumage_port,
opts.database)
Modified: branches/tmckay/mint/python/mint/aviary/endpoints.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/endpoints.py 2013-01-15 00:01:24 UTC (rev 5651)
+++ branches/tmckay/mint/python/mint/aviary/endpoints.py 2013-01-15 22:40:03 UTC (rev 5652)
@@ -54,7 +54,7 @@
def get_work_list(self):
- return ( #QuerySubmissionsAfter(self),
+ return (QuerySubmissionsAfter(self),
QuerySubmissionsBefore(self),)
class EndpointFactory(object):
Modified: branches/tmckay/mint/python/mint/aviary/locator.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/locator.py 2013-01-15 00:01:24 UTC (rev 5651)
+++ branches/tmckay/mint/python/mint/aviary/locator.py 2013-01-15 22:40:03 UTC (rev 5652)
@@ -1,10 +1,19 @@
+import logging
+
from threading import Thread
from time import sleep
+from mint.util import MintBlockingDaemonThread
-class LocatorThread(Thread):
+log = logging.getLogger("mint.aviary")
+
+class LocatorThread(MintBlockingDaemonThread):
def __init__(self, aviary_ops, foreman, endpoint_factory,
resource="ANY", subtype=""):
- Thread.__init__(self)
+
+ # We don't care about app for this thread,
+ # at least not currently
+ super(LocatorThread, self).__init__(None)
+
self.aviary_ops = aviary_ops
self.foreman = foreman
self.ep_fact = endpoint_factory
@@ -16,17 +25,24 @@
self.start()
- def stop(self):
- self.stop_requested = True
+ def run(self):
- def run(self):
+ def block():
+ self._condition.acquire()
+ if not self.stop_requested:
+ self._condition.wait(self.interval_secs)
+ self._condition.release()
+
endpoints = dict()
while not self.stop_requested:
try:
ends = self.aviary_ops.get_endpoints(self.resource,
self.subtype)
except Exception:
- sleep(1)
+ import traceback
+ message = traceback.format_exc()
+ log.debug("LocatorThread: %s" % message)
+ block()
continue
locs = dict([[e.location[0], e.id] for e in ends])
@@ -55,4 +71,4 @@
endpoints[e] = self.ep_fact.make_endpoint(e, locs[e])
self.foreman.add_work(endpoints[e].get_work_list())
- sleep(self.interval_secs)
+ block()
Modified: branches/tmckay/mint/python/mint/aviary/session.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/session.py 2013-01-15 00:01:24 UTC (rev 5651)
+++ branches/tmckay/mint/python/mint/aviary/session.py 2013-01-15 22:40:03 UTC (rev 5652)
@@ -1,5 +1,10 @@
import logging
+from work import Foreman
+from endpoints import EndpointFactory
+from locator import LocatorThread
+from datetime import timedelta
+
log = logging.getLogger("mint.aviary.session")
class MintAviarySession(object):
@@ -7,6 +12,8 @@
self.app = app
self.classes = classes
+ self.foreman = None
+ self.locatorThread = None
# need to know the location of the aviary query server(s)
# get this from sage?
@@ -31,15 +38,24 @@
pname = cls._package._name
cname = cls._name
- # is this where the launch goes?
-
log.info("Querying for Aviary class %s.%s" % (pname, cname))
- # launch all the threads/objects that we need to handle the classes
- # and build a list so that we can call stop on them
+ self.foreman = Foreman()
+ endpoint_factory = EndpointFactory(self.app.aviary_ops,
+ poll_delta=timedelta(days=64),
+ history_delta=timedelta(weeks=52))
+
+ self.locatorThread = LocatorThread(self.app.aviary_ops,
+ self.foreman, endpoint_factory)
+
def stop(self):
log.info("Stopping %s", self)
- # This will have to be a thread or object list that we
- # can call stop on
+ if self.foreman:
+ self.foreman.end_work()
+
+ if self.locatorThread:
+ self.locatorThread.stop()
+ self.locatorThread.join()
+ log.info("Stopped %s", self)
Modified: branches/tmckay/mint/python/mint/plumage/main.py
===================================================================
--- branches/tmckay/mint/python/mint/plumage/main.py 2013-01-15 00:01:24 UTC (rev 5651)
+++ branches/tmckay/mint/python/mint/plumage/main.py 2013-01-15 22:40:03 UTC (rev 5652)
@@ -1,3 +1,5 @@
+import os
+
from mint.database import MintDatabase
from mint.expire import ExpireThread
from mint.model import MintModel
@@ -12,8 +14,11 @@
log = logging.getLogger("mint.plumage")
class Plumage(object):
- def __init__(self, model_dir, server_host, server_port, database_dsn):
+ def __init__(self, home, model_paths,
+ server_host, server_port, database_dsn):
+ model_dir = [os.path.join(home, x) for x in model_paths]
+
# Can we just use a different model dir here?
# That would control visible packages, vacuuming, etc.
self.model = MintModel(self, model_dir)
11 years, 3 months
r5651 - in branches/tmckay: mint/python/mint/aviary sage/python/sage/aviary
by tmckay@fedoraproject.org
Author: tmckay
Date: 2013-01-15 00:01:24 +0000 (Tue, 15 Jan 2013)
New Revision: 5651
Modified:
branches/tmckay/mint/python/mint/aviary/locator.py
branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
Log:
Add general support for fixed vs dynamic service list
Modified: branches/tmckay/mint/python/mint/aviary/locator.py
===================================================================
--- branches/tmckay/mint/python/mint/aviary/locator.py 2013-01-11 19:43:42 UTC (rev 5650)
+++ branches/tmckay/mint/python/mint/aviary/locator.py 2013-01-15 00:01:24 UTC (rev 5651)
@@ -2,13 +2,18 @@
from time import sleep
class LocatorThread(Thread):
- def __init__(self, locator, foreman, endpoint_factory):
+ def __init__(self, aviary_ops, foreman, endpoint_factory,
+ resource="ANY", subtype=""):
Thread.__init__(self)
- self.locator = locator
+ self.aviary_ops = aviary_ops
self.foreman = foreman
self.ep_fact = endpoint_factory
+ self.resource = resource
+ self.subtype = subtype
self.daemon = True
self.stop_requested = False
+ self.interval_secs = 60
+
self.start()
def stop(self):
@@ -18,41 +23,36 @@
endpoints = dict()
while not self.stop_requested:
try:
- ends = self.locator.get_endpoints("ANY","")
+ ends = self.aviary_ops.get_endpoints(self.resource,
+ self.subtype)
except Exception:
sleep(1)
continue
- if ends.status.code in ("OK", "NO_MATCH"):
+ locs = dict([[e.location[0], e.id] for e in ends])
+ loc_keys = set(locs.keys())
- # generate a set of keys for endpoints returned from the locator
- if hasattr(ends, "resources"):
- locs = dict([[e.location[0], e.id] for e in ends.resources])
- loc_keys = set(locs.keys())
- else:
- loc_keys = set()
+ # the endpoints that we already know about
+ keys = set(endpoints.keys())
- # the endpoints that we already know about
- keys = set(endpoints.keys())
+ # loc_keys - keys gives the items returned from Aviary that are
+ # not in the current dictionary, ie new stuff
+ new_endpoints = loc_keys - keys
- # loc_keys - keys gives the items returned from Aviary that are
- # not in the current dictionary, ie new stuff
- new_endpoints = loc_keys - keys
+ # keys - loc_keys gives the items that are not currently in Aviary, ie
+ # old stuff that has been deleted
+ stale_endpoints = keys - loc_keys
- # keys - loc_keys gives the items that are not currently in Aviary, ie
- # old stuff that has been deleted
- stale_endpoints = keys - loc_keys
+ for e in stale_endpoints:
+ # if a thread is currently processing the dictionary entry,
+ # that's okay because with Python reference counting the record
+ # will still be valid. When the thread is done processing, the record
+ # will disappear.
+ endpoints[e].abandon()
+ del endpoints[e]
- for e in stale_endpoints:
- # if a thread is currently processing the dictionary entry,
- # that's okay because with Python reference counting the record
- # will still be valid. When the thread is done processing, the record
- # will disappear.
- endpoints[e].abandon()
- del endpoints[e]
+ for e in new_endpoints:
+ endpoints[e] = self.ep_fact.make_endpoint(e, locs[e])
+ self.foreman.add_work(endpoints[e].get_work_list())
- for e in new_endpoints:
- endpoints[e] = self.ep_fact.make_endpoint(e, locs[e])
- self.foreman.add_work(endpoints[e].get_work_list())
-
- sleep(1)
+ sleep(self.interval_secs)
Modified: branches/tmckay/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- branches/tmckay/sage/python/sage/aviary/aviaryoperations.py 2013-01-11 19:43:42 UTC (rev 5650)
+++ branches/tmckay/sage/python/sage/aviary/aviaryoperations.py 2013-01-15 00:01:24 UTC (rev 5651)
@@ -78,10 +78,11 @@
"QUERY_SERVER": "query service",
"HADOOP": "hadoop service"}
- def __init__(self, subtype):
+ def __init__(self, resource, subtype):
self.servers = None
self.should_retry = False
-
+ self.resource = resource
+ self.subtype = subtype
try:
self.nice = BaseServerList._nice[subtype]
except:
@@ -137,6 +138,12 @@
def get_hosts(self):
return self._get_host_set()
+ def get_endpoints(self):
+ # For ServerList types using a locator, there is no need
+ # for this. We can go straight to the locator (or this
+ # could be overloaded to do that).
+ raise Exception("Not implemented")
+
class ServerList(BaseServerList):
'''
Query an Aviary locator object for endpoints by reource and subtype.
@@ -144,7 +151,7 @@
Allow the cached list to be refreshed on demand.
'''
def __init__(self, locator, resource, subtype):
- super(ServerList, self).__init__(subtype)
+ super(ServerList, self).__init__(resource, subtype)
# Since we have a dynamic server list, failed operations
# may be retried
@@ -152,8 +159,6 @@
self._lock = Lock()
self.locator = locator
- self.resource = resource
- self.subtype = subtype
def _refresh(self):
log.debug("AviaryOperations: refresh server list for %s %s" \
@@ -238,23 +243,35 @@
self._lock.release()
return res
+class Fake_Resource_ID(object):
+ def __init__(self, resource, sub_type, name="", pool=""):
+ self.name = name
+ self.pool = pool
+ self.resource = resource
+ self.sub_type = sub_type
+
+class Fake_Endpoint(object):
+ def __init__(self, resource_id, url_list):
+ self.id = resource_id
+ self.location = url_list
+
class FixedServerList(BaseServerList):
'''
Query a fixed server list for Aviary endpoints.
Lookup an endpoint in the fixed list by hostname or hostname:port.
The fixed server list is an alternative to using an Aviary locator.
'''
- def __init__(self, servers, port, path, subtype):
+ def __init__(self, servers, port, path, resource, subtype):
'''
Initialize a fixed server list of Aviary endpoints.
servers -- a comma separated list of (partial) URLs
port -- default port value if a URL does not specify a port
path -- default path value if a URL does not specify a path
- subtype -- this is used to identify the service type in error
- messages. Predefind values are "JOB" and "QUERY_SERVER".
+ resource -- corresponds to Aviary resource type
+ subtype -- corresponds to Aviary sub_type.
The scheme for a URL will be "http" if not specified.
'''
- super(FixedServerList, self).__init__(subtype)
+ super(FixedServerList, self).__init__(resource, subtype)
# Replace any occurrence of locahost with output of gethostname()
# before parsing to match Machine fields of QMF objects later on.
@@ -280,6 +297,15 @@
"settings in cumin.conf" % (self.nice, machine))
return scheme, host
+ def get_endpoints(self):
+ endpoints = []
+ for host, urls in self.servers.iteritems():
+ res_id = Fake_Resource_ID(self.resource, self.subtype, host)
+ url_strs = [str(u) for u in urls]
+ ep = Fake_Endpoint(res_id, url_strs)
+ endpoints.append(ep)
+ return endpoints
+
class MockScheduler(object):
'''
Pseudo scheduler object.
@@ -336,11 +362,19 @@
self.use_MethodResult_for_sync_calls = True
def get_hosts(self, resource, subtype):
- if resource == self.resource and \
- (not subtype or self.subtype == subtype):
+ if resource == "ANY" or \
+ (resource == self.resource and \
+ (not subtype or self.subtype == subtype)):
return self.servers.get_hosts()
return set()
+ def get_endpoints(self, resource, subtype):
+ if resource == "ANY" or \
+ (resource == self.resource and \
+ (not subtype or self.subtype == subtype)):
+ return self.servers.get_endpoints()
+ return []
+
def _set_client_info(self, client, refresh=False):
# See if the server_name is a full URL. If so, we'll use it.
url = parse_URL(client.server_name)
@@ -460,6 +494,7 @@
servers = FixedServerList(job_servers,
"9090",
"/services/job/",
+ resource,
subtype)
else:
servers = None
@@ -658,7 +693,8 @@
elif query_servers and type(query_servers) == str:
servers = FixedServerList(query_servers,
"9091",
- "/services/query/",
+ "/services/query/",
+ resource,
subtype)
else:
servers = None
@@ -991,6 +1027,7 @@
servers = FixedServerList(hadoop_servers,
"9090",
"/services/hadoop/",
+ resource,
subtype)
else:
servers = None
@@ -1238,6 +1275,23 @@
hosts.update(m.get_hosts(resource, subtype))
return list(hosts)
+ def get_endpoints(self, resource, subtype=""):
+ # This should return a list of objects with id/location
+ # id location
+ # [ ( (name,pool,resource,sub_type), [full urls] ) ]
+ if self.locator:
+ ep = self.locator.get_endpoints(resource, subtype)
+ if ep.status.code in ("OK", "NO_MATCH"):
+ if hasattr(ep, "resources"):
+ return ep.resources
+ return []
+ else:
+ # hmmm, no locator to use.
+ # Make something that looks like locator responses.
+ endpoints = []
+ for m in self.mechs:
+ endpoints.extend(m.get_endpoints(resource, subtype))
+ return endpoints
11 years, 3 months
r5649 - in branches/elephant: cumin/python/cumin/gridhadoop wooly/python/wooly
by tmckay@fedoraproject.org
Author: tmckay
Date: 2013-01-10 16:02:34 +0000 (Thu, 10 Jan 2013)
New Revision: 5649
Modified:
branches/elephant/cumin/python/cumin/gridhadoop/datanode.py
branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py
branches/elephant/cumin/python/cumin/gridhadoop/namenode.py
branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py
branches/elephant/wooly/python/wooly/server.py
Log:
Add a username() wrapper to ClientSession and use for owner in haddoop forms
Modified: branches/elephant/cumin/python/cumin/gridhadoop/datanode.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/datanode.py 2013-01-09 18:26:08 UTC (rev 5648)
+++ branches/elephant/cumin/python/cumin/gridhadoop/datanode.py 2013-01-10 16:02:34 UTC (rev 5649)
@@ -152,7 +152,7 @@
if not self.errors.get(session):
binfile = self.binfile.get(session)
- owner = session.app.default_user
+ owner = session.client_session.username()
hadoophost = self.hadoophost.get(session)
count = self.count.get(session)
name_node = self.nameNode.get(session)
Modified: branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py 2013-01-09 18:26:08 UTC (rev 5648)
+++ branches/elephant/cumin/python/cumin/gridhadoop/jobtracker.py 2013-01-10 16:02:34 UTC (rev 5649)
@@ -148,7 +148,7 @@
if not self.errors.get(session):
binfile = self.binfile.get(session)
- owner = session.app.default_user
+ owner = session.client_session.username()
hadoophost = self.hadoophost.get(session)
name_node = self.nameNode.get(session)
count = 1
Modified: branches/elephant/cumin/python/cumin/gridhadoop/namenode.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/namenode.py 2013-01-09 18:26:08 UTC (rev 5648)
+++ branches/elephant/cumin/python/cumin/gridhadoop/namenode.py 2013-01-10 16:02:34 UTC (rev 5649)
@@ -102,7 +102,7 @@
if not self.errors.get(session):
binfile = self.binfile.get(session)
hadoophost = self.hadoophost.get(session)
- owner = session.app.default_user
+ owner = session.client_session.username()
self.task.invoke(session, None, (binfile, owner, hadoophost))
self.task.exit_with_redirect(session, url)
Modified: branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py
===================================================================
--- branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py 2013-01-09 18:26:08 UTC (rev 5648)
+++ branches/elephant/cumin/python/cumin/gridhadoop/tasktracker.py 2013-01-10 16:02:34 UTC (rev 5649)
@@ -148,7 +148,7 @@
if not self.errors.get(session):
binfile = self.binfile.get(session)
- owner = session.app.default_user
+ owner = session.client_session.username()
hadoophost = self.hadoophost.get(session)
count = self.count.get(session)
job_tracker = self.jobTracker.get(session)
Modified: branches/elephant/wooly/python/wooly/server.py
===================================================================
--- branches/elephant/wooly/python/wooly/server.py 2013-01-09 18:26:08 UTC (rev 5648)
+++ branches/elephant/wooly/python/wooly/server.py 2013-01-10 16:02:34 UTC (rev 5649)
@@ -364,6 +364,9 @@
user = self.attributes["login_session"].user.name
return owner == user
+ def username(self):
+ return self.attributes["login_session"].user.name
+
def get_csrf(self):
return self.csrf
11 years, 3 months
r5648 - trunk/cumin/etc
by tmckay@fedoraproject.org
Author: tmckay
Date: 2013-01-09 18:26:08 +0000 (Wed, 09 Jan 2013)
New Revision: 5648
Modified:
trunk/cumin/etc/sysvinit-cumin
Log:
Incrementally impove path information in messages from /etc/init.rd/cumin
BZ892971
Modified: trunk/cumin/etc/sysvinit-cumin
===================================================================
--- trunk/cumin/etc/sysvinit-cumin 2013-01-09 16:26:34 UTC (rev 5647)
+++ trunk/cumin/etc/sysvinit-cumin 2013-01-09 18:26:08 UTC (rev 5648)
@@ -52,26 +52,26 @@
}
# Initial checks for the start case
- cumin-database check-started &> /dev/null || {
+ /usr/sbin/cumin-database check-started &> /dev/null || {
echo_failure; echo
echo "Cumin's database server is not running"
- echo "Run 'cumin-database start' as root"
+ echo "Run '/usr/sbin/cumin-database start' as root"
exit 1
}
- cumin-database check &> /dev/null || {
+ /usr/sbin/cumin-database check &> /dev/null || {
# Run again with output left on for user.
echo_failure; echo
- echo "(detailed output from cumin-database check:)"
- cumin-database check
+ echo "(detailed output from /usr/sbin/cumin-database check:)"
+ /usr/sbin/cumin-database check
exit 1
}
- cumin-admin check-schema &> /dev/null || {
+ /usr/sbin/cumin-admin check-schema &> /dev/null || {
# Run again with output left on for user.
echo_failure; echo
- echo "(detailed output from cumin-admin check-schema:)"
- cumin-admin check-schema
+ echo "(detailed output from /usr/sbin/cumin-admin check-schema:)"
+ /usr/sbin/cumin-admin check-schema
exit 1
}
11 years, 4 months
r5647 - trunk/cumin/python/cumin
by croberts@fedoraproject.org
Author: croberts
Date: 2013-01-09 16:26:34 +0000 (Wed, 09 Jan 2013)
New Revision: 5647
Modified:
trunk/cumin/python/cumin/widgets.py
Log:
Per BZ 768298, bumping the supported Internet Explorer version to 9.
Modified: trunk/cumin/python/cumin/widgets.py
===================================================================
--- trunk/cumin/python/cumin/widgets.py 2013-01-09 16:01:48 UTC (rev 5646)
+++ trunk/cumin/python/cumin/widgets.py 2013-01-09 16:26:34 UTC (rev 5647)
@@ -1689,7 +1689,7 @@
super(AboutSupportedBrowsers, self).__init__(app, name)
self.browsers = ["Firefox 17", \
- "Internet Explorer 8", \
+ "Internet Explorer 9", \
"Safari 5"]
def render_browser_list(self, session):
11 years, 4 months
r5646 - trunk/cumin/python/cumin
by croberts@fedoraproject.org
Author: croberts
Date: 2013-01-09 16:01:48 +0000 (Wed, 09 Jan 2013)
New Revision: 5646
Modified:
trunk/cumin/python/cumin/widgets.py
Log:
Changing supported Firefox version from 10 to 17 per BZ 891919.
Modified: trunk/cumin/python/cumin/widgets.py
===================================================================
--- trunk/cumin/python/cumin/widgets.py 2013-01-09 14:53:37 UTC (rev 5645)
+++ trunk/cumin/python/cumin/widgets.py 2013-01-09 16:01:48 UTC (rev 5646)
@@ -1688,7 +1688,7 @@
def __init__(self, app, name):
super(AboutSupportedBrowsers, self).__init__(app, name)
- self.browsers = ["Firefox 10", \
+ self.browsers = ["Firefox 17", \
"Internet Explorer 8", \
"Safari 5"]
11 years, 4 months