AMBARI-1299. Bootstrap can hang indefinitely. (mahadev)

git-svn-id: https://svn.apache.org/repos/asf/incubator/ambari/trunk@1451494 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -688,6 +688,8 @@ Trunk (unreleased changes):
  AMBARI-1498. Hive service check fails on secure HDP cluster. (Siddharth Wagle
  via mahadev)
 
+ AMBARI-1299. Bootstrap can hang indefinitely. (mahadev)
+
 AMBARI-1.2.0 branch:
 
  INCOMPATIBLE CHANGES

ambari-server/src/main/java/org/apache/ambari/eventdb/webservice/WorkflowJsonService.java

@@ -273,16 +273,21 @@ public class WorkflowJsonService {
   @Produces(MediaType.APPLICATION_JSON)
   @Path("/tasklocality")
   public TaskLocalityData getTaskLocalitySummary(@QueryParam("jobId") String jobId, @DefaultValue("4") @QueryParam("minr") int minr,
-      @DefaultValue("24") @QueryParam("maxr") int maxr) {
+      @DefaultValue("24") @QueryParam("maxr") int maxr, @QueryParam("workflowId") String workflowId,
+      @DefaultValue("-1") @QueryParam("startTime") long startTime, @DefaultValue("-1") @QueryParam("endTime") long endTime) {
     if (maxr < minr)
       maxr = minr;
     TaskLocalityData data = new TaskLocalityData();
     PostgresConnector conn = null;
     try {
       conn = getConnector();
-      long[] times = conn.fetchJobStartStopTimes(jobId);
-      if (times != null) {
-        getTaskAttemptsByLocality(conn, jobId, times[0], times[1], data, minr, maxr);
+      if (jobId != null) {
+        long[] times = conn.fetchJobStartStopTimes(jobId);
+        if (times != null) {
+          getTaskAttemptsByLocality(conn.fetchJobTaskAttempts(jobId), times[0], times[1], data, minr, maxr);
+        }
+      } else if (workflowId != null) {
+        getTaskAttemptsByLocality(conn.fetchWorkflowTaskAttempts(workflowId), startTime, endTime, data, minr, maxr);
       }
     } catch (IOException e) {
       e.printStackTrace();
@@ -324,11 +329,10 @@ public class WorkflowJsonService {
     points.setReduceData(reducePoints);
   }
   
-  private static void getTaskAttemptsByLocality(PostgresConnector conn, String jobId, long submitTime, long finishTime, TaskLocalityData data, int minr,
+  private static void getTaskAttemptsByLocality(List<TaskAttempt> taskAttempts, long submitTime, long finishTime, TaskLocalityData data, int minr,
       int maxr) throws IOException {
     long submitTimeX = transformX(submitTime);
     long finishTimeX = transformX(finishTime);
-    List<TaskAttempt> taskAttempts = conn.fetchJobTaskAttempts(jobId);
     Set<Long> xPoints = getXPoints(taskAttempts, submitTimeX, finishTimeX);
     Long[] xList = xPoints.toArray(new Long[xPoints.size()]);
     MinMax io = new MinMax();

ambari-server/src/main/python/bootstrap.py

@@ -30,6 +30,7 @@ import traceback
 from pprint import pformat
 
 AMBARI_PASSPHRASE_VAR_NAME = "AMBARI_PASSPHRASE"
+HOST_BOOTSTRAP_TIMEOUT = 300
 
 class SCP(threading.Thread):
   """ SCP implementation that is thread based. The status can be returned using
@@ -42,6 +43,7 @@ class SCP(threading.Thread):
     self.bootdir = bootdir
     self.ret = {"exitstatus" : -1, "log" : "FAILED"}
     threading.Thread.__init__(self)
+    self.daemon = True
     pass
   
   def getStatus(self):
@@ -80,6 +82,7 @@ class SSH(threading.Thread):
     self.bootdir = bootdir
     self.ret = {"exitstatus" : -1, "log": "FAILED"}
     threading.Thread.__init__(self)
+    self.daemon = True
     pass
   
   def getHost(self):
@@ -173,8 +176,14 @@ class PSSH:
         chunkstats.append(ssh)
         pass
       # wait for the ssh's to complete
+      starttime = time.time()
       for chunkstat in chunkstats:
-        chunkstat.join()
+        elapsedtime = time.time() - starttime
+        if elapsedtime < HOST_BOOTSTRAP_TIMEOUT:
+          timeout = HOST_BOOTSTRAP_TIMEOUT - elapsedtime
+        else:
+          timeout = 0.0
+        chunkstat.join(timeout)
         self.ret[chunkstat.getHost()] = chunkstat.getStatus()
       pass
     pass
@@ -205,8 +214,14 @@ class PSCP:
         chunkstats.append(scp)
         pass
       # wait for the scp's to complete
+      starttime = time.time()
       for chunkstat in chunkstats:
-        chunkstat.join()
+        elapsedtime = time.time() - starttime
+        if elapsedtime < HOST_BOOTSTRAP_TIMEOUT:
+          timeout = HOST_BOOTSTRAP_TIMEOUT - elapsedtime
+        else:
+          timeout = 0.0
+        chunkstat.join(timeout)
         self.ret[chunkstat.getHost()] = chunkstat.getStatus()
       pass