Return-Path: X-Original-To: apmail-hadoop-common-commits-archive@www.apache.org Delivered-To: apmail-hadoop-common-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id 11A4410C27 for ; Sat, 10 Jan 2015 16:35:53 +0000 (UTC) Received: (qmail 16836 invoked by uid 500); 10 Jan 2015 16:35:54 -0000 Delivered-To: apmail-hadoop-common-commits-archive@hadoop.apache.org Received: (qmail 16765 invoked by uid 500); 10 Jan 2015 16:35:54 -0000 Mailing-List: contact common-commits-help@hadoop.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: common-dev@hadoop.apache.org Delivered-To: mailing list common-commits@hadoop.apache.org Received: (qmail 16756 invoked by uid 99); 10 Jan 2015 16:35:54 -0000 Received: from tyr.zones.apache.org (HELO tyr.zones.apache.org) (140.211.11.114) by apache.org (qpsmtpd/0.29) with ESMTP; Sat, 10 Jan 2015 16:35:54 +0000 Received: by tyr.zones.apache.org (Postfix, from userid 65534) id EB8929A7AED; Sat, 10 Jan 2015 16:35:53 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: raviprak@apache.org To: common-commits@hadoop.apache.org Message-Id: <87c6b0cfe7444a07b6b5e5e1faecb9d9@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: hadoop git commit: HADOOP-11400. GraphiteSink does not reconnect to Graphite after 'broken pipe' (Kamil Gorlo via raviprak) Date: Sat, 10 Jan 2015 16:35:53 +0000 (UTC) Repository: hadoop Updated Branches: refs/heads/trunk a26040626 -> 4d2914210 HADOOP-11400. GraphiteSink does not reconnect to Graphite after 'broken pipe' (Kamil Gorlo via raviprak) Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/4d291421 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/4d291421 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/4d291421 Branch: refs/heads/trunk Commit: 4d2914210053f28c95094aa59e48f8e84c55a2c7 Parents: a260406 Author: Ravi Prakash Authored: Sat Jan 10 08:35:40 2015 -0800 Committer: Ravi Prakash Committed: Sat Jan 10 08:35:40 2015 -0800 ---------------------------------------------------------------------- hadoop-common-project/hadoop-common/CHANGES.txt | 3 + .../hadoop/metrics2/sink/GraphiteSink.java | 141 ++++++++++++++----- .../metrics2/impl/TestGraphiteMetrics.java | 111 +++++++++------ 3 files changed, 175 insertions(+), 80 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d291421/hadoop-common-project/hadoop-common/CHANGES.txt ---------------------------------------------------------------------- diff --git a/hadoop-common-project/hadoop-common/CHANGES.txt b/hadoop-common-project/hadoop-common/CHANGES.txt index 90bfe3e..679989a 100644 --- a/hadoop-common-project/hadoop-common/CHANGES.txt +++ b/hadoop-common-project/hadoop-common/CHANGES.txt @@ -488,6 +488,9 @@ Release 2.7.0 - UNRELEASED BUG FIXES + HADOOP 11400. GraphiteSink does not reconnect to Graphite after 'broken pipe' + (Kamil Gorlo via raviprak) + HADOOP-11236. NFS: Fix javadoc warning in RpcProgram.java (Abhiraj Butala via harsh) HADOOP-11166. Remove ulimit from test-patch.sh. (wang) http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d291421/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/sink/GraphiteSink.java ---------------------------------------------------------------------- diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/sink/GraphiteSink.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/sink/GraphiteSink.java index e72fe24..e46a654 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/sink/GraphiteSink.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/sink/GraphiteSink.java @@ -18,25 +18,24 @@ package org.apache.hadoop.metrics2.sink; -import java.io.IOException; -import java.io.OutputStreamWriter; -import java.io.Writer; -import java.io.Closeable; -import java.net.Socket; - import org.apache.commons.configuration.SubsetConfiguration; import org.apache.commons.io.Charsets; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.metrics2.AbstractMetric; import org.apache.hadoop.metrics2.MetricsException; import org.apache.hadoop.metrics2.MetricsRecord; import org.apache.hadoop.metrics2.MetricsSink; import org.apache.hadoop.metrics2.MetricsTag; +import java.io.Closeable; +import java.io.IOException; +import java.io.OutputStreamWriter; +import java.io.Writer; +import java.net.Socket; + /** * A metrics sink that writes to a Graphite server */ @@ -47,30 +46,22 @@ public class GraphiteSink implements MetricsSink, Closeable { private static final String SERVER_HOST_KEY = "server_host"; private static final String SERVER_PORT_KEY = "server_port"; private static final String METRICS_PREFIX = "metrics_prefix"; - private Writer writer = null; private String metricsPrefix = null; - private Socket socket = null; + private Graphite graphite = null; @Override public void init(SubsetConfiguration conf) { // Get Graphite host configurations. - String serverHost = conf.getString(SERVER_HOST_KEY); - Integer serverPort = Integer.parseInt(conf.getString(SERVER_PORT_KEY)); + final String serverHost = conf.getString(SERVER_HOST_KEY); + final int serverPort = Integer.parseInt(conf.getString(SERVER_PORT_KEY)); // Get Graphite metrics graph prefix. metricsPrefix = conf.getString(METRICS_PREFIX); if (metricsPrefix == null) metricsPrefix = ""; - try { - // Open an connection to Graphite server. - socket = new Socket(serverHost, serverPort); - writer = new OutputStreamWriter( - socket.getOutputStream(), Charsets.UTF_8); - } catch (Exception e) { - throw new MetricsException("Error creating connection, " - + serverHost + ":" + serverPort, e); - } + graphite = new Graphite(serverHost, serverPort); + graphite.connect(); } @Override @@ -104,39 +95,111 @@ public class GraphiteSink implements MetricsSink, Closeable { } try { - if(writer != null){ - writer.write(lines.toString()); - } else { - throw new MetricsException("Writer in GraphiteSink is null!"); - } + graphite.write(lines.toString()); } catch (Exception e) { - throw new MetricsException("Error sending metrics", e); + LOG.warn("Error sending metrics to Graphite", e); + try { + graphite.close(); + } catch (Exception e1) { + throw new MetricsException("Error closing connection to Graphite", e1); + } } } @Override public void flush() { + try { + graphite.flush(); + } catch (Exception e) { + LOG.warn("Error flushing metrics to Graphite", e); try { - writer.flush(); - } catch (Exception e) { - throw new MetricsException("Error flushing metrics", e); + graphite.close(); + } catch (Exception e1) { + throw new MetricsException("Error closing connection to Graphite", e1); } + } } @Override public void close() throws IOException { - try { - IOUtils.closeStream(writer); - writer = null; - LOG.info("writer in GraphiteSink is closed!"); - } catch (Throwable e){ - throw new MetricsException("Error closing writer", e); - } finally { - if (socket != null && !socket.isClosed()) { - socket.close(); + graphite.close(); + } + + public static class Graphite { + private final static int MAX_CONNECTION_FAILURES = 5; + + private String serverHost; + private int serverPort; + private Writer writer = null; + private Socket socket = null; + private int connectionFailures = 0; + + public Graphite(String serverHost, int serverPort) { + this.serverHost = serverHost; + this.serverPort = serverPort; + } + + public void connect() { + if (isConnected()) { + throw new MetricsException("Already connected to Graphite"); + } + if (tooManyConnectionFailures()) { + // return silently (there was ERROR in logs when we reached limit for the first time) + return; + } + try { + // Open a connection to Graphite server. + socket = new Socket(serverHost, serverPort); + writer = new OutputStreamWriter(socket.getOutputStream(), Charsets.UTF_8); + } catch (Exception e) { + connectionFailures++; + if (tooManyConnectionFailures()) { + // first time when connection limit reached, report to logs + LOG.error("Too many connection failures, would not try to connect again."); + } + throw new MetricsException("Error creating connection, " + + serverHost + ":" + serverPort, e); + } + } + + public void write(String msg) throws IOException { + if (!isConnected()) { + connect(); + } + if (isConnected()) { + writer.write(msg); + } + } + + public void flush() throws IOException { + if (isConnected()) { + writer.flush(); + } + } + + public boolean isConnected() { + return socket != null && socket.isConnected() && !socket.isClosed(); + } + + public void close() throws IOException { + try { + if (writer != null) { + writer.close(); + } + } catch (IOException ex) { + if (socket != null) { + socket.close(); + } + } finally { socket = null; - LOG.info("socket in GraphiteSink is closed!"); + writer = null; } } + + private boolean tooManyConnectionFailures() { + return connectionFailures > MAX_CONNECTION_FAILURES; + } + } + } http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d291421/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/impl/TestGraphiteMetrics.java ---------------------------------------------------------------------- diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/impl/TestGraphiteMetrics.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/impl/TestGraphiteMetrics.java index 09f0081..5fac41e 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/impl/TestGraphiteMetrics.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/impl/TestGraphiteMetrics.java @@ -18,23 +18,7 @@ package org.apache.hadoop.metrics2.impl; -import static org.junit.Assert.assertEquals; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.OutputStreamWriter; -import java.io.Writer; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashSet; -import java.util.List; -import java.util.Set; - import org.apache.hadoop.metrics2.AbstractMetric; -import org.apache.hadoop.metrics2.MetricsException; import org.apache.hadoop.metrics2.MetricsRecord; import org.apache.hadoop.metrics2.MetricsTag; import org.apache.hadoop.metrics2.sink.GraphiteSink; @@ -42,6 +26,23 @@ import org.junit.Test; import org.mockito.ArgumentCaptor; import org.mockito.internal.util.reflection.Whitebox; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Matchers.anyString; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.reset; + + public class TestGraphiteMetrics { private AbstractMetric makeMetric(String name, Number value) { AbstractMetric metric = mock(AbstractMetric.class); @@ -50,6 +51,12 @@ public class TestGraphiteMetrics { return metric; } + private GraphiteSink.Graphite makeGraphite() { + GraphiteSink.Graphite mockGraphite = mock(GraphiteSink.Graphite.class); + when(mockGraphite.isConnected()).thenReturn(true); + return mockGraphite; + } + @Test public void testPutMetrics() { GraphiteSink sink = new GraphiteSink(); @@ -61,18 +68,18 @@ public class TestGraphiteMetrics { metrics.add(makeMetric("foo2", 2.25)); MetricsRecord record = new MetricsRecordImpl(MsInfo.Context, (long) 10000, tags, metrics); - OutputStreamWriter mockWriter = mock(OutputStreamWriter.class); ArgumentCaptor argument = ArgumentCaptor.forClass(String.class); - Whitebox.setInternalState(sink, "writer", mockWriter); + final GraphiteSink.Graphite mockGraphite = makeGraphite(); + Whitebox.setInternalState(sink, "graphite", mockGraphite); sink.putMetrics(record); try { - verify(mockWriter).write(argument.capture()); + verify(mockGraphite).write(argument.capture()); } catch (IOException e) { - e.printStackTrace(); + e.printStackTrace(); } - String result = argument.getValue().toString(); + String result = argument.getValue(); assertEquals(true, result.equals("null.all.Context.Context=all.Hostname=host.foo1 1.25 10\n" + @@ -86,24 +93,25 @@ public class TestGraphiteMetrics { GraphiteSink sink = new GraphiteSink(); List tags = new ArrayList(); tags.add(new MetricsTag(MsInfo.Context, "all")); - tags.add(new MetricsTag(MsInfo.Hostname, null)); + tags.add(new MetricsTag(MsInfo.Hostname, null)); Set metrics = new HashSet(); metrics.add(makeMetric("foo1", 1)); metrics.add(makeMetric("foo2", 2)); MetricsRecord record = new MetricsRecordImpl(MsInfo.Context, (long) 10000, tags, metrics); - OutputStreamWriter mockWriter = mock(OutputStreamWriter.class); + ArgumentCaptor argument = ArgumentCaptor.forClass(String.class); - Whitebox.setInternalState(sink, "writer", mockWriter); + final GraphiteSink.Graphite mockGraphite = makeGraphite(); + Whitebox.setInternalState(sink, "graphite", mockGraphite); sink.putMetrics(record); try { - verify(mockWriter).write(argument.capture()); + verify(mockGraphite).write(argument.capture()); } catch (IOException e) { e.printStackTrace(); } - String result = argument.getValue().toString(); + String result = argument.getValue(); assertEquals(true, result.equals("null.all.Context.Context=all.foo1 1 10\n" + @@ -120,8 +128,8 @@ public class TestGraphiteMetrics { // setup GraphiteSink GraphiteSink sink = new GraphiteSink(); - ByteArrayOutputStream out = new ByteArrayOutputStream(); - Whitebox.setInternalState(sink, "writer", new OutputStreamWriter(out)); + final GraphiteSink.Graphite mockGraphite = makeGraphite(); + Whitebox.setInternalState(sink, "graphite", mockGraphite); // given two metrics records with timestamps 1000 milliseconds apart. List tags = Collections.emptyList(); @@ -141,15 +149,16 @@ public class TestGraphiteMetrics { } // then the timestamps in the graphite stream should differ by one second. - String expectedOutput - = "null.default.Context.foo1 1 1000000000\n" - + "null.default.Context.foo1 1 1000000001\n"; - assertEquals(expectedOutput, out.toString()); + try { + verify(mockGraphite).write(eq("null.default.Context.foo1 1 1000000000\n")); + verify(mockGraphite).write(eq("null.default.Context.foo1 1 1000000001\n")); + } catch (IOException e) { + e.printStackTrace(); + } } - - @Test(expected=MetricsException.class) - public void testCloseAndWrite() throws IOException { + @Test + public void testFailureAndPutMetrics() throws IOException { GraphiteSink sink = new GraphiteSink(); List tags = new ArrayList(); tags.add(new MetricsTag(MsInfo.Context, "all")); @@ -159,18 +168,38 @@ public class TestGraphiteMetrics { metrics.add(makeMetric("foo2", 2.25)); MetricsRecord record = new MetricsRecordImpl(MsInfo.Context, (long) 10000, tags, metrics); - OutputStreamWriter writer = mock(OutputStreamWriter.class); + final GraphiteSink.Graphite mockGraphite = makeGraphite(); + Whitebox.setInternalState(sink, "graphite", mockGraphite); + + // throw exception when first try + doThrow(new IOException("IO exception")).when(mockGraphite).write(anyString()); - Whitebox.setInternalState(sink, "writer", writer); - sink.close(); sink.putMetrics(record); + verify(mockGraphite).write(anyString()); + verify(mockGraphite).close(); + + // reset mock and try again + reset(mockGraphite); + when(mockGraphite.isConnected()).thenReturn(false); + + ArgumentCaptor argument = ArgumentCaptor.forClass(String.class); + sink.putMetrics(record); + + verify(mockGraphite).write(argument.capture()); + String result = argument.getValue(); + + assertEquals(true, + result.equals("null.all.Context.Context=all.Hostname=host.foo1 1.25 10\n" + + "null.all.Context.Context=all.Hostname=host.foo2 2.25 10\n") || + result.equals("null.all.Context.Context=all.Hostname=host.foo2 2.25 10\n" + + "null.all.Context.Context=all.Hostname=host.foo1 1.25 10\n")); } @Test public void testClose(){ GraphiteSink sink = new GraphiteSink(); - Writer mockWriter = mock(Writer.class); - Whitebox.setInternalState(sink, "writer", mockWriter); + final GraphiteSink.Graphite mockGraphite = makeGraphite(); + Whitebox.setInternalState(sink, "graphite", mockGraphite); try { sink.close(); } catch (IOException ioe) { @@ -178,7 +207,7 @@ public class TestGraphiteMetrics { } try { - verify(mockWriter).close(); + verify(mockGraphite).close(); } catch (IOException ioe) { ioe.printStackTrace(); }