Author: tmckay
Date: 2012-12-14 15:19:48 +0000 (Fri, 14 Dec 2012)
New Revision: 5595
Modified:
branches/elephant/sage/python/sage/aviary/aviaryoperations.py
Log:
Add get_name_node
Modified: branches/elephant/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- branches/elephant/sage/python/sage/aviary/aviaryoperations.py 2012-12-14 14:58:59 UTC
(rev 5594)
+++ branches/elephant/sage/python/sage/aviary/aviaryoperations.py 2012-12-14 15:19:48 UTC
(rev 5595)
@@ -975,6 +975,59 @@
client, "stopNameNode", refs)
t.start()
+ def get_name_node(self, host, ids, callback=None):
+ if callback:
+ assert callable(callback)
+
+ if type(ids) not in (list, tuple):
+ ids = (ids)
+
+ client = self.client_pool.get_object()
+ self._setup_client(client,
+ self.servers,
+ host,
+ "getNameNode")
+
+ 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 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)
+
+ if callback:
+ def my_callback(result):
+ self.client_pool.return_object(client)
+ callback(*result_tuple(result, host))
+
+ t = CallThread(self.call_client_retry, my_callback,
+ client, "getNameNode", 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)
+ self.client_pool.return_object(client)
+ return res;
+
class AviaryOperations(Catalog):
def __init__(self, name, datadir,
locator_uri="",