From 3f6e1c605619ab1c3370d9ee943a6531e796eb01 Mon Sep 17 00:00:00 2001 From: emanuelpordes Date: Wed, 28 Nov 2012 14:32:30 -0500 Subject: [PATCH 1/3] Avoid binding DatagramSocket It's unnecessary and prevents running client on same machine as Statsd. --- .../metrics/reporting/StatsdReporter.java | 29 +++++++++++-------- 1 file changed, 17 insertions(+), 12 deletions(-) diff --git a/src/main/java/com/studyblue/metrics/reporting/StatsdReporter.java b/src/main/java/com/studyblue/metrics/reporting/StatsdReporter.java index d58d3d3..2f37603 100644 --- a/src/main/java/com/studyblue/metrics/reporting/StatsdReporter.java +++ b/src/main/java/com/studyblue/metrics/reporting/StatsdReporter.java @@ -26,6 +26,8 @@ import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.net.SocketException; import java.util.Locale; import java.util.Map; import java.util.SortedMap; @@ -49,8 +51,8 @@ public static enum StatType { COUNTER, TIMER, GAUGE } private boolean printVMMetrics = true; public interface UDPSocketProvider { - DatagramSocket get() throws Exception; - DatagramPacket newPacket(ByteArrayOutputStream out); + DatagramSocket get() throws SocketException; + DatagramPacket newPacket(ByteArrayOutputStream out) throws SocketException; } public StatsdReporter(String host, int port) throws IOException { @@ -78,7 +80,7 @@ public StatsdReporter(MetricsRegistry metricsRegistry, String prefix, MetricPred } public StatsdReporter(MetricsRegistry metricsRegistry, String prefix, MetricPredicate predicate, UDPSocketProvider socketProvider, Clock clock, VirtualMachineMetrics vm) throws IOException { - this(metricsRegistry, prefix, predicate, socketProvider, clock, vm, "graphite-reporter"); + this(metricsRegistry, prefix, predicate, socketProvider, clock, vm, "statsd-reporter"); } public StatsdReporter(MetricsRegistry metricsRegistry, String prefix, MetricPredicate predicate, UDPSocketProvider socketProvider, Clock clock, VirtualMachineMetrics vm, String name) throws IOException { @@ -128,9 +130,9 @@ public void run() { socket.send(packet); } catch (Exception e) { if (LOG.isDebugEnabled()) { - LOG.debug("Error writing to Graphite", e); + LOG.debug("Error writing to Statsd", e); } else { - LOG.warn("Error writing to Graphite: {}", e.getMessage()); + LOG.warn("Error writing to Statsd: {}", e.getMessage()); } if (writer != null) { try { @@ -304,7 +306,7 @@ protected void sendData(String name, String value, StatType statType) { statTypeStr = "ms"; break; } - + try { if (!prefix.isEmpty()) { writer.write(prefix); @@ -317,7 +319,7 @@ protected void sendData(String name, String value, StatType statType) { writer.write('\n'); writer.flush(); } catch (IOException e) { - LOG.error("Error sending to Graphite:", e); + LOG.error("Error sending to Statsd:", e); } } @@ -332,12 +334,12 @@ public DefaultSocketProvider(String host, int port) { } @Override - public DatagramSocket get() throws Exception { - return new DatagramSocket(new InetSocketAddress(this.host, this.port)); + public DatagramSocket get() throws SocketException { + return new DatagramSocket(); } - + @Override - public DatagramPacket newPacket(ByteArrayOutputStream out) { + public DatagramPacket newPacket(ByteArrayOutputStream out) throws SocketException { byte[] dataBuffer; if (out != null) { dataBuffer = out.toByteArray(); @@ -345,7 +347,10 @@ public DatagramPacket newPacket(ByteArrayOutputStream out) { else { dataBuffer = new byte[8192]; } - return new DatagramPacket(dataBuffer, dataBuffer.length); + + return new DatagramPacket(dataBuffer, dataBuffer.length, new InetSocketAddress(this.host, this.port)); } + + } } From 083b2cb9afd86693ba5c7da5cab373bffed78c83 Mon Sep 17 00:00:00 2001 From: emanuelpordes Date: Thu, 29 Nov 2012 09:48:24 -0500 Subject: [PATCH 2/3] bump version number --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 1995b2e..4ff9a9c 100644 --- a/pom.xml +++ b/pom.xml @@ -5,7 +5,7 @@ studyblue metrics-statsd Metrics Statsd Support - 2.1.3.0 + 2.1.3.1 jar From a9ac6907699f9e75adde14f072ca98cd2bf53220 Mon Sep 17 00:00:00 2001 From: Emanuel Pordes Date: Thu, 22 Jan 2015 13:30:03 -0500 Subject: [PATCH 3/3] Sync up with https://github.com/lithiumtech/metrics-statsd/tree/multiple-datagrams. Cleanup/update pom. --- pom.xml | 67 ++++----------- .../metrics/reporting/StatsdReporter.java | 81 +++++++++++++------ .../metrics/reporting/StatsdReporterTest.java | 3 +- 3 files changed, 75 insertions(+), 76 deletions(-) rename src/main/java/com/{studyblue => bealetech}/metrics/reporting/StatsdReporter.java (88%) rename src/test/java/com/{studyblue => bealetech}/metrics/reporting/StatsdReporterTest.java (99%) diff --git a/pom.xml b/pom.xml index 4ff9a9c..6f93231 100644 --- a/pom.xml +++ b/pom.xml @@ -2,25 +2,17 @@ 4.0.0 - studyblue + com.bealetech metrics-statsd Metrics Statsd Support - 2.1.3.1 + 2.3.0.1 jar - 2.1.3 - 1.6.4 + 2.2.0 + 1.7.10 - - - Sean Laurent - organicveggie@gmail.com - -6 - - - Apache License 2.0 @@ -29,34 +21,6 @@ - - scm:git:git://github.com/organicveggie/metrics-statsd.git - scm:git:git@github.com:organicveggie/metrics-statsd.git - http://github.com/organicveggie/metrics-statsd/ - - - - github - http://github.com/organicveggie/metrics-statsd/issues#issue/ - - - - - central - http://oss.sonatype.org/content/repositories/releases - - false - - - - snapshots - http://oss.sonatype.org/content/repositories/snapshots - - true - - - - com.yammer.metrics @@ -86,13 +50,13 @@ junit junit-dep - 4.10 + 4.11 test org.mockito mockito-all - 1.9.0 + 1.9.5 test @@ -102,30 +66,31 @@ org.apache.maven.plugins maven-compiler-plugin - 2.3.2 + 3.2 - 1.6 - 1.6 + 1.7 + 1.7 org.apache.felix maven-bundle-plugin - 2.3.7 + 2.5.3 true org.apache.maven.plugins maven-surefire-plugin - 2.8.1 + 2.18.1 - classes + 3 + true org.apache.maven.plugins maven-source-plugin - 2.1.2 + 2.4 attach-sources @@ -138,7 +103,7 @@ org.apache.maven.plugins maven-javadoc-plugin - 2.8.1 + 2.10.1 attach-javadocs @@ -151,7 +116,7 @@ org.apache.maven.plugins maven-release-plugin - 2.2.1 + 2.5.1 true forked-path diff --git a/src/main/java/com/studyblue/metrics/reporting/StatsdReporter.java b/src/main/java/com/bealetech/metrics/reporting/StatsdReporter.java similarity index 88% rename from src/main/java/com/studyblue/metrics/reporting/StatsdReporter.java rename to src/main/java/com/bealetech/metrics/reporting/StatsdReporter.java index 2f37603..eb457d0 100644 --- a/src/main/java/com/studyblue/metrics/reporting/StatsdReporter.java +++ b/src/main/java/com/bealetech/metrics/reporting/StatsdReporter.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package com.studyblue.metrics.reporting; +package com.bealetech.metrics.reporting; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.*; @@ -25,9 +25,7 @@ import java.io.*; import java.net.DatagramPacket; import java.net.DatagramSocket; -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.net.SocketException; +import java.net.InetAddress; import java.util.Locale; import java.util.Map; import java.util.SortedMap; @@ -35,6 +33,15 @@ public class StatsdReporter extends AbstractPollingReporter implements MetricProcessor { + /* + MTU size of the host handling the PDU (most of the case it will be 1500) - + size of the IP header (20 bytes) - + size of UDP header (8 bytes) + + 1500 MTU - 20 IP hdr - 8 UDP hdr = 1472 bytes + */ + private static final int MAX_UDPDATAGRAM_LENGTH = 1472; + public static enum StatType { COUNTER, TIMER, GAUGE } private static final Logger LOG = LoggerFactory.getLogger(StatsdReporter.class); @@ -43,16 +50,21 @@ public static enum StatType { COUNTER, TIMER, GAUGE } protected final MetricPredicate predicate; protected final Locale locale = Locale.US; protected final Clock clock; + protected final UDPSocketProvider socketProvider; + protected DatagramSocket currentSocket = null; + protected final VirtualMachineMetrics vm; + protected Writer writer; protected ByteArrayOutputStream outputData; + private boolean prependNewline = false; private boolean printVMMetrics = true; public interface UDPSocketProvider { - DatagramSocket get() throws SocketException; - DatagramPacket newPacket(ByteArrayOutputStream out) throws SocketException; + DatagramSocket get() throws Exception; + DatagramPacket newPacket(ByteArrayOutputStream out); } public StatsdReporter(String host, int port) throws IOException { @@ -111,23 +123,19 @@ public void setPrintVMMetrics(boolean printVMMetrics) { @Override public void run() { - DatagramSocket socket = null; + try { - socket = this.socketProvider.get(); - outputData.reset(); - writer = new BufferedWriter(new OutputStreamWriter(this.outputData)); + currentSocket = this.socketProvider.get(); + resetWriterState(); final long epoch = clock.time() / 1000; - if (this.printVMMetrics) { + if (printVMMetrics) { printVmMetrics(epoch); } printRegularMetrics(epoch); // Send UDP data - writer.flush(); - DatagramPacket packet = this.socketProvider.newPacket(outputData); - packet.setData(outputData.toByteArray()); - socket.send(packet); + sendDatagram(); } catch (Exception e) { if (LOG.isDebugEnabled()) { LOG.debug("Error writing to Statsd", e); @@ -142,13 +150,28 @@ public void run() { } } } finally { - if (socket != null) { - socket.close(); + if (currentSocket != null) { + currentSocket.close(); } writer = null; } } + private void resetWriterState() { + outputData.reset(); + prependNewline = false; + writer = new BufferedWriter(new OutputStreamWriter(this.outputData)); + } + + private void sendDatagram() throws IOException { + writer.flush(); + if (outputData.size() > 0) { // Don't send an empty datagram + DatagramPacket packet = this.socketProvider.newPacket(outputData); + packet.setData(outputData.toByteArray()); + currentSocket.send(packet); + } + } + protected void printVmMetrics(long epoch) { // Memory sendFloat("jvm.memory.totalInit", StatType.GAUGE, vm.totalInit()); @@ -308,6 +331,9 @@ protected void sendData(String name, String value, StatType statType) { } try { + if (prependNewline) { + writer.write("\n"); + } if (!prefix.isEmpty()) { writer.write(prefix); } @@ -316,8 +342,14 @@ protected void sendData(String name, String value, StatType statType) { writer.write(value); writer.write("|"); writer.write(statTypeStr); - writer.write('\n'); + prependNewline = true; writer.flush(); + + if (outputData.size() > MAX_UDPDATAGRAM_LENGTH) { + // Need to send our UDP packet now before it gets too big. + sendDatagram(); + resetWriterState(); + } } catch (IOException e) { LOG.error("Error sending to Statsd:", e); } @@ -334,13 +366,14 @@ public DefaultSocketProvider(String host, int port) { } @Override - public DatagramSocket get() throws SocketException { + public DatagramSocket get() throws Exception { return new DatagramSocket(); } @Override - public DatagramPacket newPacket(ByteArrayOutputStream out) throws SocketException { + public DatagramPacket newPacket(ByteArrayOutputStream out) { byte[] dataBuffer; + if (out != null) { dataBuffer = out.toByteArray(); } @@ -348,9 +381,11 @@ public DatagramPacket newPacket(ByteArrayOutputStream out) throws SocketExceptio dataBuffer = new byte[8192]; } - return new DatagramPacket(dataBuffer, dataBuffer.length, new InetSocketAddress(this.host, this.port)); + try { + return new DatagramPacket(dataBuffer, dataBuffer.length, InetAddress.getByName(this.host), this.port); + } catch (Exception e) { + return null; + } } - - } } diff --git a/src/test/java/com/studyblue/metrics/reporting/StatsdReporterTest.java b/src/test/java/com/bealetech/metrics/reporting/StatsdReporterTest.java similarity index 99% rename from src/test/java/com/studyblue/metrics/reporting/StatsdReporterTest.java rename to src/test/java/com/bealetech/metrics/reporting/StatsdReporterTest.java index 5e55da2..fd61497 100644 --- a/src/test/java/com/studyblue/metrics/reporting/StatsdReporterTest.java +++ b/src/test/java/com/bealetech/metrics/reporting/StatsdReporterTest.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package com.studyblue.metrics.reporting; +package com.bealetech.metrics.reporting; import com.yammer.metrics.core.*; import com.yammer.metrics.reporting.AbstractPollingReporter; @@ -33,7 +33,6 @@ import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; -import static org.mockito.Matchers.any; import static org.mockito.Mockito.*; public class StatsdReporterTest {