Fix HADOOP-148. Maintain a task failure count per tasktracker and display it in the web ui. Contributed by Owen.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@395284 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -61,6 +61,9 @@ Trunk (unreleased)
 17. Fix HADOOP-142.  Avoid re-running a task on a host where it has
     previously failed.  (omalley via cutting)
 
+18. Fix HADOOP-148.  Maintain a task failure count for each
+    tasktracker and display it in the web ui.  (omalley via cutting)
+
 
 Release 0.1.1 - 2006-04-08
 

src/java/org/apache/hadoop/dfs/FSNamesystem.java

@@ -243,7 +243,9 @@ class FSNamesystem implements FSConstants {
                                             short replication 
                                           ) throws IOException {
         if (pendingCreates.get(src) != null) {
-          LOG.warning("Cannot start file because pendingCreates is non-null. src=" + src);
+          LOG.warning("Cannot create file " + src + " for " + holder +
+                       " on " + clientMachine + 
+                       " because pendingCreates is non-null.");
           return null;
         }
 

src/java/org/apache/hadoop/mapred/TaskTracker.java

@@ -69,6 +69,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
     private MapOutputFile mapOutputFile;
 
     private int maxCurrentTasks;
+    private int failures;
 
     class MapOutputServer extends RPC.Server {
       private MapOutputServer(int port, int threads) {
@@ -255,7 +256,10 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
                 this.fs = FileSystem.getNamed(jobClient.getFilesystemName(), this.fConf);
             }
             
-            int resultCode = jobClient.emitHeartbeat(new TaskTrackerStatus(taskTrackerName, localHostname, mapOutputPort, taskReports), justStarted);
+            TaskTrackerStatus status = 
+              new TaskTrackerStatus(taskTrackerName, localHostname, 
+                                    mapOutputPort, taskReports, failures); 
+            int resultCode = jobClient.emitHeartbeat(status, justStarted);
             justStarted = false;
               
             if (resultCode == InterTrackerProtocol.UNKNOWN_TASKTRACKER) {
@@ -279,10 +283,11 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
                 for (Iterator it = runningTasks.values().iterator(); it.hasNext(); ) {
                     TaskInProgress tip = (TaskInProgress) it.next();
                     if ((tip.getRunState() == TaskStatus.RUNNING) &&
-                        (System.currentTimeMillis() - tip.getLastProgressReport() > this.taskTimeout)) {
+                        (System.currentTimeMillis() - tip.getLastProgressReport() > this.taskTimeout) &&
+                        !tip.wasKilled) {
                         LOG.info("Task " + tip.getTask().getTaskId() + " timed out.  Killing.");
                         tip.reportDiagnosticInfo("Timed out.");
-                        tip.killAndCleanup();
+                        tip.killAndCleanup(true);
                     }
                 }
             }
@@ -531,6 +536,9 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
             if (done) {
                 runstate = TaskStatus.SUCCEEDED;
             } else {
+                if (!wasKilled) {
+                  failures += 1;
+                }
                 runstate = TaskStatus.FAILED;
             }
 
@@ -554,7 +562,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
          */
         public synchronized void jobHasFinished() throws IOException {
             if (getRunState() == TaskStatus.RUNNING) {
-                killAndCleanup();
+                killAndCleanup(false);
             } else {
                 cleanup();
             }
@@ -563,9 +571,13 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
         /**
          * This task has run on too long, and should be killed.
          */
-        public synchronized void killAndCleanup() throws IOException {
+        public synchronized void killAndCleanup(boolean wasFailure
+                                                ) throws IOException {
             if (runstate == TaskStatus.RUNNING) {
                 wasKilled = true;
+                if (wasFailure) {
+                  failures += 1;
+                }
                 runner.kill();
             }
         }

src/java/org/apache/hadoop/mapred/TaskTrackerStatus.java

@@ -41,6 +41,7 @@ class TaskTrackerStatus implements Writable {
     String trackerName;
     String host;
     int port;
+    int failures;
     Vector taskReports;
     
     volatile long lastSeen;
@@ -52,13 +53,15 @@ class TaskTrackerStatus implements Writable {
 
     /**
      */
-    public TaskTrackerStatus(String trackerName, String host, int port, Vector taskReports) {
+    public TaskTrackerStatus(String trackerName, String host, int port, 
+                             Vector taskReports, int failures) {
         this.trackerName = trackerName;
         this.host = host;
         this.port = port;
 
         this.taskReports = new Vector();
         this.taskReports.addAll(taskReports);
+        this.failures = failures;
     }
 
     /**
@@ -77,6 +80,14 @@ class TaskTrackerStatus implements Writable {
         return port;
     }
 
+    /**
+     * Get the number of tasks that have failed on this tracker.
+     * @return The number of failed tasks
+     */
+    public int getFailures() {
+      return failures;
+    }
+    
     /**
      * All current tasks at the TaskTracker.  
      *
@@ -127,6 +138,7 @@ class TaskTrackerStatus implements Writable {
         out.writeInt(port);
 
         out.writeInt(taskReports.size());
+        out.writeInt(failures);
         for (Iterator it = taskReports.iterator(); it.hasNext(); ) {
             ((TaskStatus) it.next()).write(out);
         }
@@ -143,6 +155,7 @@ class TaskTrackerStatus implements Writable {
         taskReports.clear();
 
         int numTasks = in.readInt();
+        this.failures = in.readInt();
         for (int i = 0; i < numTasks; i++) {
             TaskStatus tmp = new TaskStatus();
             tmp.readFields(in);

src/webapps/mapred/jobtracker.jsp

@@ -20,9 +20,12 @@
     } else {
       out.print("<center>\n");
       out.print("<table border=\"2\" cellpadding=\"5\" cellspacing=\"2\">\n");
-      out.print("<tr><td align=\"center\" colspan=\"4\"><b>Task Trackers</b></td></tr>\n");
-      out.print("<tr><td><b>Name</b></td><td><b>Host</b></td><td><b># running tasks</b></td><td><b>Secs since heartbeat</b></td></tr>\n");
-
+      out.print("<tr><td align=\"center\" colspan=\"5\"><b>Task Trackers</b></td></tr>\n");
+      out.print("<tr><td><b>Name</b></td><td><b>Host</b></td>" +
+                "<td><b># running tasks</b></td><td><b>Failures</b></td>" +
+                "<td><b>Secs since heartbeat</b></td></tr>\n");
+      int maxFailures = 0;
+      String failureKing = null;
       for (Iterator it = c.iterator(); it.hasNext(); ) {
         TaskTrackerStatus tt = (TaskTrackerStatus) it.next();
         long sinceHeartbeat = System.currentTimeMillis() - tt.getLastSeen();
@@ -34,11 +37,23 @@
           it2.next();
           numCurTasks++;
         }
+        int numFailures = tt.getFailures();
+        if (numFailures > maxFailures) {
+          maxFailures = numFailures;
+          failureKing = tt.getTrackerName();
+        }
 
-        out.print("<tr><td>" + tt.getTrackerName() + "</td><td>" + tt.getHost() + "</td><td>" + numCurTasks + "</td><td>" + sinceHeartbeat + "</td></tr>\n");
+        out.print("<tr><td>" + tt.getTrackerName() + "</td><td>" + 
+                  tt.getHost() + "</td><td>" + numCurTasks +
+                  "</td><td>" + numFailures + 
+                  "</td><td>" + sinceHeartbeat + "</td></tr>\n");
       }
       out.print("</table>\n");
       out.print("</center>\n");
+      if (maxFailures > 0) {
+        out.print("Highest Failures: " + failureKing + " with " + maxFailures + 
+                  " failures<br>\n");
+      }
     }
   }