Author: tmckay
Date: 2012-12-14 15:47:58 +0000 (Fri, 14 Dec 2012)
New Revision: 5597
Modified:
branches/elephant/sage/python/sage/aviary/aviaryoperations.py
Log:
Add stop and get for all types.
Modified: branches/elephant/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- branches/elephant/sage/python/sage/aviary/aviaryoperations.py 2012-12-14 15:47:00 UTC
(rev 5596)
+++ branches/elephant/sage/python/sage/aviary/aviaryoperations.py 2012-12-14 15:47:58 UTC
(rev 5597)
@@ -933,49 +933,31 @@
t.start()
def stop_name_node(self, host, ids, callback):
- assert callable(callback)
+ assert callback
+ self._operate_on_ids(host, ids, callback, "stopNameNode")
- if type(ids) not in (list, tuple):
- ids = (ids)
+ def get_name_node(self, host, ids, callback=None):
+ return self._operate_on_ids(self, host, ids, callback, "getNameNode")
- client = self.client_pool.get_object()
- self._setup_client(client,
- self.servers,
- host,
- "stopNameNode")
+ def stop_data_node(self, host, ids, callback):
+ self._operate_on_ids(host, ids, callback, "stopDataNode")
- refs = []
- for id in ids:
- new_ref = client.factory.create("ns1:HadoopID")
- try:
- if type(id) in (str,unicode):
- new_ref.ipc = id
- else:
- new_ref.id = int(id)
- except Exception:
- raise Exception("Invalid HadoopID value")
- refs.append(new_ref)
+ def get_data_node(self, host, ids, callback=None):
+ return self._operate_on_ids(self, host, ids, callback, "getDataNode")
- def result_tuple(result, host):
- data = None
- result = self._pretty_result(result, host)
- if isinstance(result, Exception):
- status = result
- else:
- status = _AviaryCommon._get_status(result.status)
- if status == "OK" and hasattr(result, "results"):
- data = result.results
- return (status, data)
+ def stop_job_tracker(self, host, ids, callback):
+ self._operate_on_ids(host, ids, callback, "stopJobTracker")
- def my_callback(result):
- self.client_pool.return_object(client)
- callback(*result_tuple(result, host))
+ def get_job_tracker(self, host, ids, callback=None):
+ return self._operate_on_ids(self, host, ids, callback,
"getJobTracker")
- t = CallThread(self.call_client_retry, my_callback,
- client, "stopNameNode", refs)
- t.start()
+ def stop_task_tracker(self, host, ids, callback):
+ self._operate_on_ids(host, ids, callback, "stopTaskTracker")
- def get_name_node(self, host, ids, callback=None):
+ def get_task_tracker(self, host, ids, callback=None):
+ return self._operate_on_ids(self, host, ids, callback,
"getTaskTracker")
+
+ def _operate_on_ids(self, host, ids, callback, meth_name):
if callback:
assert callable(callback)
@@ -986,7 +968,7 @@
self._setup_client(client,
self.servers,
host,
- "getNameNode")
+ meth_name)
refs = []
for id in ids:
@@ -1017,14 +999,14 @@
callback(*result_tuple(result, host))
t = CallThread(self.call_client_retry, my_callback,
- client, "getNameNode", refs)
+ client, meth_name, refs)
t.start()
else:
def my_process_results(result):
return result_tuple(result, host)
res = self._call_sync(self.call_client_retry, my_process_results,
- client, "getNameNode", refs)
+ client, meth_name, refs)
self.client_pool.return_object(client)
return res;