hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cdoug...@apache.org
Subject svn commit: r1131736 - in /hadoop/mapreduce/trunk: CHANGES.txt src/java/org/apache/hadoop/mapred/TaskTracker.java src/java/org/apache/hadoop/mapreduce/server/jobtracker/JTConfig.java src/test/mapred/org/apache/hadoop/mapred/TestShuffleExceptionCount.java
Date Sun, 05 Jun 2011 05:22:41 GMT
Author: cdouglas
Date: Sun Jun  5 05:22:41 2011
New Revision: 1131736

URL: http://svn.apache.org/viewvc?rev=1131736&view=rev
Log:
MAPREDUCE-2529. Add support for regex-based shuffle metric counting
exceptions. Contributed by Thomas Graves

Added:
    hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestShuffleExceptionCount.java
Modified:
    hadoop/mapreduce/trunk/CHANGES.txt
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/server/jobtracker/JTConfig.java

Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=1131736&r1=1131735&r2=1131736&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Sun Jun  5 05:22:41 2011
@@ -272,6 +272,9 @@ Trunk (unreleased changes)
     MAPREDUCE-2536. Update FsShell -mv command usage in TestMRCLI.  (Daryn
     Sharp via szetszwo)
 
+    MAPREDUCE-2529. Add support for regex-based shuffle metric counting
+    exceptions. (Thomas Graves via cdouglas)
+
 Release 0.22.0 - Unreleased
 
   INCOMPATIBLE CHANGES

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java?rev=1131736&r1=1131735&r2=1131736&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java Sun Jun  5 05:22:41
2011
@@ -327,12 +327,13 @@ public class TaskTracker 
    * the specific metrics for shuffle. The TaskTracker is actually a server for
    * the shuffle and hence the name ShuffleServerMetrics.
    */
-  private class ShuffleServerMetrics implements Updater {
+  class ShuffleServerMetrics implements Updater {
     private MetricsRecord shuffleMetricsRecord = null;
     private int serverHandlerBusy = 0;
     private long outputBytes = 0;
     private int failedOutputs = 0;
     private int successOutputs = 0;
+    private int exceptionsCaught = 0;
     ShuffleServerMetrics(JobConf conf) {
       MetricsContext context = MetricsUtil.getContext("mapred");
       shuffleMetricsRecord = 
@@ -355,6 +356,9 @@ public class TaskTracker 
     synchronized void successOutput() {
       ++successOutputs;
     }
+    synchronized void exceptionsCaught() {
+      ++exceptionsCaught;
+    }
     public void doUpdates(MetricsContext unused) {
       synchronized (this) {
         if (workerThreads != 0) {
@@ -369,9 +373,12 @@ public class TaskTracker 
                                         failedOutputs);
         shuffleMetricsRecord.incrMetric("shuffle_success_outputs", 
                                         successOutputs);
+        shuffleMetricsRecord.incrMetric("shuffle_exceptions_caught",
+                                        exceptionsCaught);
         outputBytes = 0;
         failedOutputs = 0;
         successOutputs = 0;
+        exceptionsCaught = 0;
       }
       shuffleMetricsRecord.update();
     }
@@ -1376,6 +1383,10 @@ public class TaskTracker 
     server.setAttribute("log", LOG);
     server.setAttribute("localDirAllocator", localDirAllocator);
     server.setAttribute("shuffleServerMetrics", shuffleServerMetrics);
+    String exceptionStackRegex = conf.get(JTConfig.SHUFFLE_EXCEPTION_STACK_REGEX);
+    String exceptionMsgRegex = conf.get(JTConfig.SHUFFLE_EXCEPTION_MSG_REGEX);
+    server.setAttribute("exceptionStackRegex", exceptionStackRegex);
+    server.setAttribute("exceptionMsgRegex", exceptionMsgRegex);
     server.addInternalServlet("mapOutput", "/mapOutput", MapOutputServlet.class);
     server.addServlet("taskLog", "/tasklog", TaskLogServlet.class);
     server.start();
@@ -3660,6 +3671,10 @@ public class TaskTracker 
         (ShuffleServerMetrics) context.getAttribute("shuffleServerMetrics");
       TaskTracker tracker = 
         (TaskTracker) context.getAttribute("task.tracker");
+      String exceptionStackRegex =
+        (String) context.getAttribute("exceptionStackRegex");
+      String exceptionMsgRegex =
+        (String) context.getAttribute("exceptionMsgRegex");
 
       verifyRequest(request, response, tracker, jobId);
       
@@ -3690,6 +3705,7 @@ public class TaskTracker 
         String errorMsg = ("getMapOutputs(" + mapIds + "," + reduceId + 
                            ") failed");
         log.warn(errorMsg, ie);
+        checkException(ie, exceptionMsgRegex, exceptionStackRegex, shuffleMetrics);
         response.sendError(HttpServletResponse.SC_GONE, errorMsg);
         shuffleMetrics.failedOutput();
         throw ie;
@@ -3712,6 +3728,37 @@ public class TaskTracker 
       }
     }
 
+    protected void checkException(IOException ie, String exceptionMsgRegex,
+        String exceptionStackRegex, ShuffleServerMetrics shuffleMetrics) {
+      // parse exception to see if it looks like a regular expression you
+      // configure. If both msgRegex and StackRegex set then make sure both
+      // match, otherwise only the one set has to match.
+      if (exceptionMsgRegex != null) {
+        String msg = ie.getMessage();
+        if (msg == null || !msg.matches(exceptionMsgRegex)) {
+          return;
+        }
+      }
+      if (exceptionStackRegex != null
+          && !checkStackException(ie, exceptionStackRegex)) {
+        return;
+      }
+      shuffleMetrics.exceptionsCaught();
+    }
+
+    private boolean checkStackException(IOException ie,
+        String exceptionStackRegex) {
+      StackTraceElement[] stack = ie.getStackTrace();
+
+      for (StackTraceElement elem : stack) {
+        String stacktrace = elem.toString();
+        if (stacktrace.matches(exceptionStackRegex)) {
+          return true;
+        }
+      }
+      return false;
+    }
+
     private void sendMapFile(String jobId, String mapId,
                              int reduce,
                              Configuration conf,

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/server/jobtracker/JTConfig.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/server/jobtracker/JTConfig.java?rev=1131736&r1=1131735&r2=1131736&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/server/jobtracker/JTConfig.java
(original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/server/jobtracker/JTConfig.java
Sun Jun  5 05:22:41 2011
@@ -111,4 +111,9 @@ public interface JTConfig extends MRConf
      "mapreduce.jobtracker.webinterface.trusted";
   public static final String JT_PLUGINS = 
     "mapreduce.jobtracker.plugins";
+  public static final String SHUFFLE_EXCEPTION_STACK_REGEX =
+    "mapreduce.reduce.shuffle.catch.exception.stack.regex";
+  public static final String SHUFFLE_EXCEPTION_MSG_REGEX =
+    "mapreduce.reduce.shuffle.catch.exception.message.regex";
+
 }

Added: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestShuffleExceptionCount.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestShuffleExceptionCount.java?rev=1131736&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestShuffleExceptionCount.java
(added)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestShuffleExceptionCount.java
Sun Jun  5 05:22:41 2011
@@ -0,0 +1,245 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapred;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.hadoop.mapred.TaskTracker.ShuffleServerMetrics;
+import org.apache.hadoop.metrics.ContextFactory;
+import org.apache.hadoop.metrics.MetricsContext;
+import org.apache.hadoop.metrics.spi.OutputRecord;
+import org.junit.Test;
+
+public class TestShuffleExceptionCount {
+
+  public static class TestMapOutputServlet extends TaskTracker.MapOutputServlet {
+
+    public void checkException(IOException ie, String exceptionMsgRegex,
+        String exceptionStackRegex, ShuffleServerMetrics shuffleMetrics) {
+      super.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+          shuffleMetrics);
+    }
+
+  }
+
+  @Test
+  public void testCheckException() throws IOException, InterruptedException,
+      ClassNotFoundException, InstantiationException, IllegalAccessException {
+    TestMapOutputServlet testServlet = new TestMapOutputServlet();
+    JobConf conf = new JobConf();
+    conf.setUser("testuser");
+    conf.setJobName("testJob");
+    conf.setSessionId("testSession");
+
+    // setup metrics context factory
+    ContextFactory factory = ContextFactory.getFactory();
+    factory.setAttribute("mapred.class",
+        "org.apache.hadoop.metrics.spi.NoEmitMetricsContext");
+
+    TaskTracker tt = new TaskTracker();
+    tt.setConf(conf);
+    ShuffleServerMetrics shuffleMetrics = tt.new ShuffleServerMetrics(conf);
+
+    // first test with only MsgRegex set but doesn't match
+    String exceptionMsgRegex = "Broken pipe";
+    String exceptionStackRegex = null;
+    IOException ie = new IOException("EOFException");
+    testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+        shuffleMetrics);
+
+    MetricsContext context = factory.getContext("mapred");
+    shuffleMetrics.doUpdates(context);
+    Map<String, Collection<OutputRecord>> records = context.getAllRecords();
+    Collection<OutputRecord> col = records.get("shuffleOutput");
+    OutputRecord outputRecord = col.iterator().next();
+    assertEquals(0, outputRecord.getMetric("shuffle_exceptions_caught")
+        .intValue());
+
+    // test with only MsgRegex set that does match
+    ie = new IOException("Broken pipe");
+    testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+        shuffleMetrics);
+
+    shuffleMetrics.doUpdates(context);
+    assertEquals(1, outputRecord.getMetric("shuffle_exceptions_caught")
+        .intValue());
+
+    // test with neither set, make sure incremented
+    exceptionStackRegex = null;
+    exceptionMsgRegex = null;
+    testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+        shuffleMetrics);
+    shuffleMetrics.doUpdates(context);
+    assertEquals(2, outputRecord.getMetric("shuffle_exceptions_caught")
+        .intValue());
+
+    // test with only StackRegex set doesn't match
+    exceptionStackRegex = ".*\\.doesnt\\$SelectSet\\.wakeup.*";
+    exceptionMsgRegex = null;
+    ie.setStackTrace(constructStackTrace());
+    testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+        shuffleMetrics);
+    shuffleMetrics.doUpdates(context);
+    assertEquals(2, outputRecord.getMetric("shuffle_exceptions_caught")
+        .intValue());
+
+    // test with only StackRegex set does match
+    exceptionStackRegex = ".*\\.SelectorManager\\$SelectSet\\.wakeup.*";
+    testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+        shuffleMetrics);
+    shuffleMetrics.doUpdates(context);
+    assertEquals(3, outputRecord.getMetric("shuffle_exceptions_caught")
+        .intValue());
+
+    // test with both regex set and matches
+    exceptionMsgRegex = "Broken pipe";
+    ie.setStackTrace(constructStackTraceTwo());
+    testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+        shuffleMetrics);
+    shuffleMetrics.doUpdates(context);
+    assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
+        .intValue());
+
+    // test with both regex set and only msg matches
+    exceptionStackRegex = ".*[1-9]+BOGUSREGEX";
+    testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+        shuffleMetrics);
+    shuffleMetrics.doUpdates(context);
+    assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
+        .intValue());
+
+    // test with both regex set and only stack matches
+    exceptionStackRegex = ".*\\.SelectorManager\\$SelectSet\\.wakeup.*";
+    exceptionMsgRegex = "EOFException";
+    testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
+        shuffleMetrics);
+    shuffleMetrics.doUpdates(context);
+    assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
+        .intValue());
+  }
+
+  /*
+   * Construction exception like: java.io.IOException: Broken pipe at
+   * sun.nio.ch.EPollArrayWrapper.interrupt(Native Method) at
+   * sun.nio.ch.EPollArrayWrapper.interrupt(EPollArrayWrapper.java:256) at
+   * sun.nio.ch.EPollSelectorImpl.wakeup(EPollSelectorImpl.java:175) at
+   * org.mortbay
+   * .io.nio.SelectorManager$SelectSet.wakeup(SelectorManager.java:831) at
+   * org.mortbay
+   * .io.nio.SelectorManager$SelectSet.doSelect(SelectorManager.java:709) at
+   * org.mortbay.io.nio.SelectorManager.doSelect(SelectorManager.java:192) at
+   * org
+   * .mortbay.jetty.nio.SelectChannelConnector.accept(SelectChannelConnector.java
+   * :124) at
+   * org.mortbay.jetty.AbstractConnector$Acceptor.run(AbstractConnector.
+   * java:708) at
+   * org.mortbay.thread.QueuedThreadPool$PoolThread.run(QueuedThreadPool
+   * .java:582)
+   */
+  private StackTraceElement[] constructStackTrace() {
+    StackTraceElement[] stack = new StackTraceElement[9];
+    stack[0] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
+        "interrupt", "", -2);
+    stack[1] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
+        "interrupt", "EPollArrayWrapper.java", 256);
+    stack[2] = new StackTraceElement("sun.nio.ch.EPollSelectorImpl", "wakeup",
+        "EPollSelectorImpl.java", 175);
+    stack[3] = new StackTraceElement(
+        "org.mortbay.io.nio.SelectorManager$SelectSet", "wakeup",
+        "SelectorManager.java", 831);
+    stack[4] = new StackTraceElement(
+        "org.mortbay.io.nio.SelectorManager$SelectSet", "doSelect",
+        "SelectorManager.java", 709);
+    stack[5] = new StackTraceElement("org.mortbay.io.nio.SelectorManager",
+        "doSelect", "SelectorManager.java", 192);
+    stack[6] = new StackTraceElement(
+        "org.mortbay.jetty.nio.SelectChannelConnector", "accept",
+        "SelectChannelConnector.java", 124);
+    stack[7] = new StackTraceElement(
+        "org.mortbay.jetty.AbstractConnector$Acceptor", "run",
+        "AbstractConnector.java", 708);
+    stack[8] = new StackTraceElement(
+        "org.mortbay.thread.QueuedThreadPool$PoolThread", "run",
+        "QueuedThreadPool.java", 582);
+
+    return stack;
+  }
+
+  /*
+   * java.io.IOException: Broken pipe at
+   * sun.nio.ch.EPollArrayWrapper.interrupt(Native Method) at
+   * sun.nio.ch.EPollArrayWrapper.interrupt(EPollArrayWrapper.java:256) at
+   * sun.nio.ch.EPollSelectorImpl.wakeup(EPollSelectorImpl.java:175) at
+   * org.mortbay
+   * .io.nio.SelectorManager$SelectSet.wakeup(SelectorManager.java:831) at
+   * org.mortbay
+   * .io.nio.SelectChannelEndPoint.updateKey(SelectChannelEndPoint.java:335) at
+   * org
+   * .mortbay.io.nio.SelectChannelEndPoint.blockWritable(SelectChannelEndPoint
+   * .java:278) at
+   * org.mortbay.jetty.AbstractGenerator$Output.blockForOutput(AbstractGenerator
+   * .java:545) at
+   * org.mortbay.jetty.AbstractGenerator$Output.flush(AbstractGenerator
+   * .java:572) at
+   * org.mortbay.jetty.HttpConnection$Output.flush(HttpConnection.java:1012) at
+   * org
+   * .mortbay.jetty.AbstractGenerator$Output.write(AbstractGenerator.java:651)at
+   * org
+   * .mortbay.jetty.AbstractGenerator$Output.write(AbstractGenerator.java:580)
+   * at
+   */
+  private StackTraceElement[] constructStackTraceTwo() {
+    StackTraceElement[] stack = new StackTraceElement[11];
+    stack[0] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
+        "interrupt", "", -2);
+    stack[1] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
+        "interrupt", "EPollArrayWrapper.java", 256);
+    stack[2] = new StackTraceElement("sun.nio.ch.EPollSelectorImpl", "wakeup",
+        "EPollSelectorImpl.java", 175);
+    stack[3] = new StackTraceElement(
+        "org.mortbay.io.nio.SelectorManager$SelectSet", "wakeup",
+        "SelectorManager.java", 831);
+    stack[4] = new StackTraceElement(
+        "org.mortbay.io.nio.SelectChannelEndPoint", "updateKey",
+        "SelectChannelEndPoint.java", 335);
+    stack[5] = new StackTraceElement(
+        "org.mortbay.io.nio.SelectChannelEndPoint", "blockWritable",
+        "SelectChannelEndPoint.java", 278);
+    stack[6] = new StackTraceElement(
+        "org.mortbay.jetty.AbstractGenerator$Output", "blockForOutput",
+        "AbstractGenerator.java", 545);
+    stack[7] = new StackTraceElement(
+        "org.mortbay.jetty.AbstractGenerator$Output", "flush",
+        "AbstractGenerator.java", 572);
+    stack[8] = new StackTraceElement("org.mortbay.jetty.HttpConnection$Output",
+        "flush", "HttpConnection.java", 1012);
+    stack[9] = new StackTraceElement(
+        "org.mortbay.jetty.AbstractGenerator$Output", "write",
+        "AbstractGenerator.java", 651);
+    stack[10] = new StackTraceElement(
+        "org.mortbay.jetty.AbstractGenerator$Output", "write",
+        "AbstractGenerator.java", 580);
+
+    return stack;
+  }
+
+}



Mime
View raw message