You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2017/10/30 08:38:16 UTC
flink git commit: [FLINK-7933][metrics] Improve PrometheusReporter
tests
Repository: flink
Updated Branches:
refs/heads/master 6b8f7dc2d -> e22c777f4
[FLINK-7933][metrics] Improve PrometheusReporter tests
This closes #4908.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e22c777f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e22c777f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e22c777f
Branch: refs/heads/master
Commit: e22c777f44b3e24cf9f4509672f981c71d4eb715
Parents: 6b8f7dc
Author: zentol <ch...@apache.org>
Authored: Thu Oct 26 19:04:30 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Mon Oct 30 09:37:48 2017 +0100
----------------------------------------------------------------------
.../metrics/prometheus/PrometheusReporter.java | 9 +++
.../PrometheusReporterTaskScopeTest.java | 37 ++++++++----
.../prometheus/PrometheusReporterTest.java | 61 ++++++++++----------
3 files changed, 65 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e22c777f/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporter.java
----------------------------------------------------------------------
diff --git a/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporter.java b/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporter.java
index 1e44ab9..fad3ced 100644
--- a/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporter.java
+++ b/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporter.java
@@ -32,6 +32,7 @@ import org.apache.flink.metrics.reporter.MetricReporter;
import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup;
import org.apache.flink.runtime.metrics.groups.FrontMetricGroup;
import org.apache.flink.util.NetUtils;
+import org.apache.flink.util.Preconditions;
import io.prometheus.client.Collector;
import io.prometheus.client.CollectorRegistry;
@@ -73,9 +74,16 @@ public class PrometheusReporter implements MetricReporter {
private static final String SCOPE_PREFIX = "flink" + SCOPE_SEPARATOR;
private HTTPServer httpServer;
+ private int port;
private final Map<String, AbstractMap.SimpleImmutableEntry<Collector, Integer>> collectorsWithCountByMetricName = new HashMap<>();
@VisibleForTesting
+ int getPort() {
+ Preconditions.checkState(httpServer != null, "Server has not been initialized.");
+ return port;
+ }
+
+ @VisibleForTesting
static String replaceInvalidChars(final String input) {
// https://prometheus.io/docs/instrumenting/writing_exporters/
// Only [a-zA-Z0-9:_] are valid in metric names, any other characters should be sanitized to an underscore.
@@ -91,6 +99,7 @@ public class PrometheusReporter implements MetricReporter {
int port = ports.next();
try {
httpServer = new HTTPServer(port);
+ this.port = port;
LOG.info("Started PrometheusReporter HTTP server on port {}.", port);
break;
} catch (IOException ioe) { //assume port conflict
http://git-wip-us.apache.org/repos/asf/flink/blob/e22c777f/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTaskScopeTest.java
----------------------------------------------------------------------
diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTaskScopeTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTaskScopeTest.java
index c7d4040..0ae8fc7 100644
--- a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTaskScopeTest.java
+++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTaskScopeTest.java
@@ -37,6 +37,7 @@ import org.apache.flink.util.AbstractID;
import com.mashape.unirest.http.exceptions.UnirestException;
import io.prometheus.client.CollectorRegistry;
import org.junit.After;
+import org.junit.Before;
import org.junit.Test;
import java.util.Arrays;
@@ -62,7 +63,6 @@ public class PrometheusReporterTaskScopeTest {
private static final int SUBTASK_INDEX_1 = 0;
private static final int SUBTASK_INDEX_2 = 1;
- private final MetricRegistry registry = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "9429")));
private final JobID jobId = new JobID();
private final JobVertexID taskId1 = new JobVertexID();
@@ -72,10 +72,29 @@ public class PrometheusReporterTaskScopeTest {
private final AbstractID taskAttemptId2 = new AbstractID();
private final String[] labelValues2 = {jobId.toString(), taskId2.toString(), taskAttemptId2.toString(), TASK_MANAGER_HOST, TASK_NAME, "" + ATTEMPT_NUMBER, JOB_NAME, TASK_MANAGER_ID, "" + SUBTASK_INDEX_2};
- private final TaskManagerMetricGroup tmMetricGroup = new TaskManagerMetricGroup(registry, TASK_MANAGER_HOST, TASK_MANAGER_ID);
- private final TaskManagerJobMetricGroup tmJobMetricGroup = new TaskManagerJobMetricGroup(registry, tmMetricGroup, jobId, JOB_NAME);
- private final TaskMetricGroup taskMetricGroup1 = new TaskMetricGroup(registry, tmJobMetricGroup, taskId1, taskAttemptId1, TASK_NAME, SUBTASK_INDEX_1, ATTEMPT_NUMBER);
- private final TaskMetricGroup taskMetricGroup2 = new TaskMetricGroup(registry, tmJobMetricGroup, taskId2, taskAttemptId2, TASK_NAME, SUBTASK_INDEX_2, ATTEMPT_NUMBER);
+ private TaskMetricGroup taskMetricGroup1;
+ private TaskMetricGroup taskMetricGroup2;
+
+ private MetricRegistry registry;
+ private PrometheusReporter reporter;
+
+ @Before
+ public void setupReporter() {
+ registry = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "9400-9500")));
+ reporter = (PrometheusReporter) registry.getReporters().get(0);
+
+ TaskManagerMetricGroup tmMetricGroup = new TaskManagerMetricGroup(registry, TASK_MANAGER_HOST, TASK_MANAGER_ID);
+ TaskManagerJobMetricGroup tmJobMetricGroup = new TaskManagerJobMetricGroup(registry, tmMetricGroup, jobId, JOB_NAME);
+ taskMetricGroup1 = new TaskMetricGroup(registry, tmJobMetricGroup, taskId1, taskAttemptId1, TASK_NAME, SUBTASK_INDEX_1, ATTEMPT_NUMBER);
+ taskMetricGroup2 = new TaskMetricGroup(registry, tmJobMetricGroup, taskId2, taskAttemptId2, TASK_NAME, SUBTASK_INDEX_2, ATTEMPT_NUMBER);
+ }
+
+ @After
+ public void shutdownRegistry() {
+ if (registry != null) {
+ registry.shutdown();
+ }
+ }
@Test
public void countersCanBeAddedSeveralTimesIfTheyDifferInLabels() throws UnirestException {
@@ -137,7 +156,7 @@ public class PrometheusReporterTaskScopeTest {
taskMetricGroup1.histogram("my_histogram", histogram);
taskMetricGroup2.histogram("my_histogram", histogram);
- final String exportedMetrics = pollMetrics().getBody();
+ final String exportedMetrics = pollMetrics(reporter.getPort()).getBody();
assertThat(exportedMetrics, containsString("subtask_index=\"0\",quantile=\"0.5\",} 0.5")); // histogram
assertThat(exportedMetrics, containsString("subtask_index=\"1\",quantile=\"0.5\",} 0.5")); // histogram
@@ -179,10 +198,4 @@ public class PrometheusReporterTaskScopeTest {
labelNames[LABEL_NAMES.length] = element;
return labelNames;
}
-
- @After
- public void shutdownRegistry() {
- registry.shutdown();
- }
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e22c777f/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java
----------------------------------------------------------------------
diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java
index 956339b..0d7be6d 100644
--- a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java
+++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java
@@ -27,7 +27,6 @@ import org.apache.flink.metrics.Histogram;
import org.apache.flink.metrics.Meter;
import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.SimpleCounter;
-import org.apache.flink.metrics.reporter.MetricReporter;
import org.apache.flink.metrics.util.TestMeter;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
@@ -40,6 +39,7 @@ import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;
import com.mashape.unirest.http.exceptions.UnirestException;
import org.junit.After;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -56,7 +56,6 @@ import static org.junit.Assert.assertThat;
* Basic test for {@link PrometheusReporter}.
*/
public class PrometheusReporterTest extends TestLogger {
- private static final int NON_DEFAULT_PORT = 9429;
private static final String HOST_NAME = "hostname";
private static final String TASK_MANAGER = "tm";
@@ -70,9 +69,23 @@ public class PrometheusReporterTest extends TestLogger {
@Rule
public ExpectedException thrown = ExpectedException.none();
- private final MetricRegistry registry = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "" + NON_DEFAULT_PORT)));
- private final FrontMetricGroup<TaskManagerMetricGroup> metricGroup = new FrontMetricGroup<>(0, new TaskManagerMetricGroup(registry, HOST_NAME, TASK_MANAGER));
- private final MetricReporter reporter = registry.getReporters().get(0);
+ private MetricRegistry registry;
+ private FrontMetricGroup<TaskManagerMetricGroup> metricGroup;
+ private PrometheusReporter reporter;
+
+ @Before
+ public void setupReporter() {
+ registry = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "9400-9500")));
+ metricGroup = new FrontMetricGroup<>(0, new TaskManagerMetricGroup(registry, HOST_NAME, TASK_MANAGER));
+ reporter = (PrometheusReporter) registry.getReporters().get(0);
+ }
+
+ @After
+ public void shutdownRegistry() {
+ if (registry != null) {
+ registry.shutdown();
+ }
+ }
/**
* {@link io.prometheus.client.Counter} may not decrease, so report {@link Counter} as {@link io.prometheus.client.Gauge}.
@@ -145,9 +158,11 @@ public class PrometheusReporterTest extends TestLogger {
@Test
public void endpointIsUnavailableAfterReporterIsClosed() throws UnirestException {
+ MetricRegistry registry = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "9400-9500")));
+ PrometheusReporter reporter = (PrometheusReporter) registry.getReporters().get(0);
reporter.close();
thrown.expect(UnirestException.class);
- pollMetrics();
+ pollMetrics(reporter.getPort());
}
@Test
@@ -229,10 +244,12 @@ public class PrometheusReporterTest extends TestLogger {
@Test
public void cannotStartTwoReportersOnSamePort() {
- final MetricRegistry fixedPort1 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "12345")));
- final MetricRegistry fixedPort2 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test2", "12345")));
-
+ final MetricRegistry fixedPort1 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "9400-9500")));
assertThat(fixedPort1.getReporters(), hasSize(1));
+
+ PrometheusReporter firstReporter = (PrometheusReporter) fixedPort1.getReporters().get(0);
+
+ final MetricRegistry fixedPort2 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test2", String.valueOf(firstReporter.getPort()))));
assertThat(fixedPort2.getReporters(), hasSize(0));
fixedPort1.shutdown();
@@ -241,8 +258,8 @@ public class PrometheusReporterTest extends TestLogger {
@Test
public void canStartTwoReportersWhenUsingPortRange() {
- final MetricRegistry portRange1 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "9249-9252")));
- final MetricRegistry portRange2 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test2", "9249-9252")));
+ final MetricRegistry portRange1 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "9200-9300")));
+ final MetricRegistry portRange2 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test2", "9200-9300")));
assertThat(portRange1.getReporters(), hasSize(1));
assertThat(portRange2.getReporters(), hasSize(1));
@@ -251,28 +268,13 @@ public class PrometheusReporterTest extends TestLogger {
portRange2.shutdown();
}
- @Test
- public void cannotStartThreeReportersWhenPortRangeIsTooSmall() {
- final MetricRegistry smallPortRange1 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test1", "9253-9254")));
- final MetricRegistry smallPortRange2 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test2", "9253-9254")));
- final MetricRegistry smallPortRange3 = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(createConfigWithOneReporter("test3", "9253-9254")));
-
- assertThat(smallPortRange1.getReporters(), hasSize(1));
- assertThat(smallPortRange2.getReporters(), hasSize(1));
- assertThat(smallPortRange3.getReporters(), hasSize(0));
-
- smallPortRange1.shutdown();
- smallPortRange2.shutdown();
- smallPortRange3.shutdown();
- }
-
private String addMetricAndPollResponse(Metric metric, String metricName) throws UnirestException {
reporter.notifyOfAddedMetric(metric, metricName, metricGroup);
- return pollMetrics().getBody();
+ return pollMetrics(reporter.getPort()).getBody();
}
- static HttpResponse<String> pollMetrics() throws UnirestException {
- return Unirest.get("http://localhost:" + NON_DEFAULT_PORT + "/metrics").asString();
+ static HttpResponse<String> pollMetrics(int port) throws UnirestException {
+ return Unirest.get("http://localhost:" + port + "/metrics").asString();
}
static Configuration createConfigWithOneReporter(String reporterName, String portString) {
@@ -285,7 +287,6 @@ public class PrometheusReporterTest extends TestLogger {
@After
public void closeReporterAndShutdownRegistry() {
- reporter.close();
registry.shutdown();
}
}