MetricsHttpSink implements MetricsSink { @Override public void writeMetrics(MetricQueryResults metricQueryResults) throws Exception { URL url = new URL(urlString); String metrics = serializeMetrics(metricQueryResults); byte[] postData = metrics.getBytes(StandardCharsets.UTF_8); HttpURLConnection connection = (HttpURLConnection) url.openConnection(); connection.setDoOutput(true); connection.setInstanceFollowRedirects(false); connection.setRequestMethod("POST"); connection.setRequestProperty("Content-Type", "application/json"); connection.setRequestProperty("charset", "utf-8"); connection.setRequestProperty("Content-Length", Integer.toString(postData.length)); connection.setUseCaches(false); try (DataOutputStream connectionOuputStream = new DataOutputStream(connection.getOutputStream())) { connectionOuputStream.write(postData); } int responseCode = connection.getResponseCode(); if (responseCode != 200) { throw new IOException( "Expected HTTP 200 OK response while writing metrics to MetricsHttpSink but received " + responseCode); } } MetricsHttpSink(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.setMetricsHttpSinkUrl(String.format("http: MetricsHttpSink metricsHttpSink = new MetricsHttpSink(pipelineOptions); countDownLatch = new CountDownLatch(1); metricsHttpSink.writeMetrics(metricQueryResults); countDownLatch.await(); String expected = "{\"counters\":[{\"attempted\":20,\"committed\":10,\"name\":{\"name\":\"n1\"," + "\"namespace\":\"ns1\"},\"step\":\"s1\"}],\"distributions\":[{\"attempted\":" + "{\"count\":4,\"max\":9,\"mean\":6.25,\"min\":3,\"sum\":25},\"committed\":" + "{\"count\":2,\"max\":8,\"mean\":5.0,\"min\":5,\"sum\":10},\"name\":{\"name\":\"n2\"," + "\"namespace\":\"ns1\"},\"step\":\"s2\"}],\"gauges\":[{\"attempted\":{\"timestamp\":" + "\"1970-01-05T00:04:22.800Z\",\"value\":120},\"committed\":{\"timestamp\":" + "\"1970-01-05T00:04:22.800Z\",\"value\":100},\"name\":{\"name\":\"n3\",\"namespace\":" + "\"ns1\"},\"step\":\"s3\"}]}"; assertEquals("Wrong number of messages sent to HTTP server", 1, messages.size()); assertEquals("Wrong messages sent to HTTP server", expected, messages.get(0)); }
@Test public void testWriteMetricsWithCommittedUnSupported() throws Exception { MetricQueryResults metricQueryResults = new CustomMetricQueryResults(false); MetricsOptions pipelineOptions = PipelineOptionsFactory.create().as(MetricsOptions.class); pipelineOptions.setMetricsHttpSinkUrl(String.format("http: MetricsHttpSink metricsHttpSink = new MetricsHttpSink(pipelineOptions); countDownLatch = new CountDownLatch(1); metricsHttpSink.writeMetrics(metricQueryResults); countDownLatch.await(); String expected = "{\"counters\":[{\"attempted\":20,\"name\":{\"name\":\"n1\"," + "\"namespace\":\"ns1\"},\"step\":\"s1\"}],\"distributions\":[{\"attempted\":" + "{\"count\":4,\"max\":9,\"mean\":6.25,\"min\":3,\"sum\":25},\"name\":{\"name\":\"n2\"" + ",\"namespace\":\"ns1\"},\"step\":\"s2\"}],\"gauges\":[{\"attempted\":{\"timestamp\":" + "\"1970-01-05T00:04:22.800Z\",\"value\":120},\"name\":{\"name\":\"n3\",\"namespace\":" + "\"ns1\"},\"step\":\"s3\"}]}"; assertEquals("Wrong number of messages sent to HTTP server", 1, messages.size()); assertEquals("Wrong messages sent to HTTP server", expected, messages.get(0)); }