MetricsGraphiteSink implements MetricsSink { @Override public void writeMetrics(MetricQueryResults metricQueryResults) throws Exception { final long metricTimestamp = System.currentTimeMillis() / 1000L; Socket socket = new Socket(InetAddress.getByName(address), port); BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(), charset)); StringBuilder messagePayload = new StringBuilder(); Iterable<MetricResult<Long>> counters = metricQueryResults.getCounters(); Iterable<MetricResult<GaugeResult>> gauges = metricQueryResults.getGauges(); Iterable<MetricResult<DistributionResult>> distributions = metricQueryResults.getDistributions(); for (MetricResult<Long> counter : counters) { messagePayload.append(new CounterMetricMessage(counter, "value", metricTimestamp).toString()); } for (MetricResult<GaugeResult> gauge : gauges) { messagePayload.append(new GaugeMetricMessage(gauge, "value").toString()); } for (MetricResult<DistributionResult> distribution : distributions) { messagePayload.append( new DistributionMetricMessage(distribution, "min", metricTimestamp).toString()); messagePayload.append( new DistributionMetricMessage(distribution, "max", metricTimestamp).toString()); messagePayload.append( new DistributionMetricMessage(distribution, "count", metricTimestamp).toString()); messagePayload.append( new DistributionMetricMessage(distribution, "sum", metricTimestamp).toString()); messagePayload.append( new DistributionMetricMessage(distribution, "mean", metricTimestamp).toString()); } writer.write(messagePayload.toString()); writer.flush(); writer.close(); socket.close(); } MetricsGraphiteSink(MetricsOptions pipelineOptions); @Override void writeMetrics(MetricQueryResults metricQueryResults);  }
@Test public void testWriteMetricsWithCommittedSupported() throws Exception { MetricQueryResults metricQueryResults = new CustomMetricQueryResults(true); MetricsOptions pipelineOptions = PipelineOptionsFactory.create().as(MetricsOptions.class); pipelineOptions.setMetricsGraphitePort(port); pipelineOptions.setMetricsGraphiteHost("127.0.0.1"); MetricsGraphiteSink metricsGraphiteSink = new MetricsGraphiteSink(pipelineOptions); CountDownLatch countDownLatch = new CountDownLatch(1); graphiteServer.setCountDownLatch(countDownLatch); metricsGraphiteSink.writeMetrics(metricQueryResults); countDownLatch.await(); String join = String.join("\n", graphiteServer.getMessages()); String regexpr = "beam.counter.ns1.n1.s1.committed.value 10 [0-9]+\\n" + "beam.counter.ns1.n1.s1.attempted.value 20 [0-9]+\\n" + "beam.gauge.ns1.n3.s3.committed.value 100 [0-9]+\\n" + "beam.gauge.ns1.n3.s3.attempted.value 120 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.committed.min 5 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.min 3 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.committed.max 8 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.max 9 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.committed.count 2 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.count 4 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.committed.sum 10 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.sum 25 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.committed.mean 5.0 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.mean 6.25 [0-9]+"; assertTrue(join.matches(regexpr)); }
@Test public void testWriteMetricsWithCommittedUnSupported() throws Exception { MetricQueryResults metricQueryResults = new CustomMetricQueryResults(false); MetricsOptions pipelineOptions = PipelineOptionsFactory.create().as(MetricsOptions.class); pipelineOptions.setMetricsGraphitePort(port); pipelineOptions.setMetricsGraphiteHost("127.0.0.1"); MetricsGraphiteSink metricsGraphiteSink = new MetricsGraphiteSink(pipelineOptions); CountDownLatch countDownLatch = new CountDownLatch(1); graphiteServer.setCountDownLatch(countDownLatch); metricsGraphiteSink.writeMetrics(metricQueryResults); countDownLatch.await(); String join = String.join("\n", graphiteServer.getMessages()); String regexpr = "beam.counter.ns1.n1.s1.attempted.value 20 [0-9]+\\n" + "beam.gauge.ns1.n3.s3.attempted.value 120 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.min 3 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.max 9 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.count 4 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.sum 25 [0-9]+\\n" + "beam.distribution.ns1.n2.s2.attempted.mean 6.25 [0-9]+"; assertTrue(join.matches(regexpr)); }
@Test public void testWriteMetricsWithCommittedSupported() throws Exception { MetricQueryResults metricQueryResults = new CustomMetricQueryResults(true); MetricsOptions pipelineOptions = PipelineOptionsFactory.create().as(MetricsOptions.class); pipelineOptions.setMetricsGraphitePort(port); pipelineOptions.setMetricsGraphiteHost("127.0.0.1"); MetricsGraphiteSink metricsGraphiteSink = new MetricsGraphiteSink(pipelineOptions); CountDownLatch countDownLatch = new CountDownLatch(1); graphiteServer.setCountDownLatch(countDownLatch); metricsGraphiteSink.writeMetrics(metricQueryResults); countDownLatch.await(); String join = String.join("\n", graphiteServer.getMessages()); String regexpr = "beam.counter.ns1.n1.committed.value 10 [0-9]+\\n" + "beam.counter.ns1.n1.attempted.value 20 [0-9]+\\n" + "beam.gauge.ns1.n3.committed.value 100 [0-9]+\\n" + "beam.gauge.ns1.n3.attempted.value 120 [0-9]+\\n" + "beam.distribution.ns1.n2.committed.min 5 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.min 3 [0-9]+\\n" + "beam.distribution.ns1.n2.committed.max 8 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.max 9 [0-9]+\\n" + "beam.distribution.ns1.n2.committed.count 2 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.count 4 [0-9]+\\n" + "beam.distribution.ns1.n2.committed.sum 10 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.sum 25 [0-9]+\\n" + "beam.distribution.ns1.n2.committed.mean 5.0 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.mean 6.25 [0-9]+"; assertTrue(join.matches(regexpr)); }
@Test public void testWriteMetricsWithCommittedUnSupported() throws Exception { MetricQueryResults metricQueryResults = new CustomMetricQueryResults(false); MetricsOptions pipelineOptions = PipelineOptionsFactory.create().as(MetricsOptions.class); pipelineOptions.setMetricsGraphitePort(port); pipelineOptions.setMetricsGraphiteHost("127.0.0.1"); MetricsGraphiteSink metricsGraphiteSink = new MetricsGraphiteSink(pipelineOptions); CountDownLatch countDownLatch = new CountDownLatch(1); graphiteServer.setCountDownLatch(countDownLatch); metricsGraphiteSink.writeMetrics(metricQueryResults); countDownLatch.await(); String join = String.join("\n", graphiteServer.getMessages()); String regexpr = "beam.counter.ns1.n1.attempted.value 20 [0-9]+\\n" + "beam.gauge.ns1.n3.attempted.value 120 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.min 3 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.max 9 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.count 4 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.sum 25 [0-9]+\\n" + "beam.distribution.ns1.n2.attempted.mean 6.25 [0-9]+"; assertTrue(join.matches(regexpr)); }