Author: tmckay
Date: 2013-06-20 14:56:58 +0000 (Thu, 20 Jun 2013)
New Revision: 5760
Modified:
trunk/sage/python/sage/aviary/aviaryoperations.py
Log:
Use Aviary api for addition of external name nodes and jt's
Modified: trunk/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- trunk/sage/python/sage/aviary/aviaryoperations.py 2013-06-20 11:52:20 UTC (rev 5759)
+++ trunk/sage/python/sage/aviary/aviaryoperations.py 2013-06-20 14:56:58 UTC (rev 5760)
@@ -1122,90 +1122,76 @@
def get_task_tracker_list(self, owner=None, callback=None):
return self._get_node_list(self.get_task_tracker, owner, callback)
- def add_external_name_node(self, ipc, url, callback=None):
- self._hadoop_lock.acquire()
- if not (ipc, url) in self._external_name_nodes:
- self._external_name_nodes.append((ipc, url))
- try:
- self._write_external(self._external_nn_path,
- self._external_name_nodes)
- except Exception:
- log.debug("Failed to write external name nodes")
- self._hadoop_lock.release()
- if callback:
- callback("OK")
- else:
- return "OK"
+ def add_external_name_node(self, host, ipc, url, callback):
+ # This is special because the HadoopID is special
+ assert callable(callback)
+
+ client = self.client_pool.get_object()
+ self._setup_client(client,
+ self.servers,
+ host,
+ "startNameNode")
+
+ 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, "ref"):
+ data = result.ref
+ return (status, data)
+
+ def my_callback(result):
+ self.client_pool.return_object(client)
+ callback(*result_tuple(result, host))
+
+ ref = client.factory.create("ns1:HadoopID")
+ ref.id = ""
+ ref.ipc = ipc
+ ref.http = url
+
+ t = CallThread(self.call_client_retry, my_callback,
+ client, "startNameNode", "", "",
"", ref)
+ t.start()
- def get_external_name_nodes(self, callback=None):
- status = "OK"
- res = []
- self._hadoop_lock.acquire()
- try:
- res = self._read_external(self._external_nn_path,
- self._external_name_nodes)
- except Exception:
- status = "FAILED"
- log.debug("Failed to read external name nodes")
- self._hadoop_lock.release()
- if callback:
- callback(status, res)
- else:
- return (status, res)
+ def add_external_job_tracker(self, host, ipc, url, callback):
+ # This is special because the HadoooID is special, and
+ # also because the id refers to the jt itself as an external
+ # and not a namenode
+ assert callable(callback)
- def add_external_job_tracker(self, ipc, url, callback=None):
- self._hadoop_lock.acquire()
- if not (ipc, url) in self._external_job_trackers:
- self._external_job_trackers.append((ipc, url))
- try:
- self._write_external(self._external_jt_path,
- self._external_job_trackers)
- except Exception:
- log.debug("Failed to write external job trackers")
- self._hadoop_lock.release()
- if callback:
- callback("OK")
- else:
- return "OK"
+ client = self.client_pool.get_object()
+ self._setup_client(client,
+ self.servers,
+ host,
+ "startJobTracker")
- def get_external_job_trackers(self, callback=None):
- status = "OK"
- res = []
- self._hadoop_lock.acquire()
- try:
- res = self._read_external(self._external_jt_path,
- self._external_job_trackers)
- except Exception:
- status = "FAILED"
- log.debug("Failed to read external job trackers")
- self._hadoop_lock.release()
- if callback:
- callback(status, res)
- else:
- return (status, res)
+ 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, "ref"):
+ data = result.ref
+ return (status, data)
- def _read_external(self, path, vals):
- res = []
- file = open(path, "r")
- lines = file.readlines()
- for l in lines:
- ipc, url = l.split()
- res.append((ipc, url))
- return res
+ def my_callback(result):
+ self.client_pool.return_object(client)
+ callback(*result_tuple(result, host))
- def _write_external(self, path, vals):
- p = os.path.split(path)[0]
- if os.path.isdir(p):
- try:
- file = open(path, "w+")
- for v in vals:
- file.write(v[0] + " " + v[1] + "\n")
- file.close()
- return True
- except Exception:
- pass
- return False
+ ref = client.factory.create("ns1:HadoopID")
+ ref.id = ""
+ ref.ipc = ipc
+ ref.http = url
+ t = CallThread(self.call_client_retry, my_callback,
+ client, "startJobTracker", ref)
+ t.start()
+
def _make_id(self, client, val):
def url(v):