This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new f1387f72 [FLINK-29619] Remove redundant MeterView updater thread from
KubernetesClientMetrics
f1387f72 is described below
commit f1387f723da3f274bcb3268d401d91042064255d
Author: Matyas Orhidi <[email protected]>
AuthorDate: Thu Oct 13 09:36:17 2022 +0200
[FLINK-29619] Remove redundant MeterView updater thread from
KubernetesClientMetrics
---
.../operator/metrics/KubernetesClientMetrics.java | 11 --
.../metrics/KubernetesClientMetricsTest.java | 184 +++++++++++++++------
.../operator/metrics/TestingMetricListener.java | 22 ++-
3 files changed, 152 insertions(+), 65 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetrics.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetrics.java
index fe582260..1dc6d3d1 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetrics.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetrics.java
@@ -32,8 +32,6 @@ import okhttp3.Response;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
/** Kubernetes client metrics. */
public class KubernetesClientMetrics implements Interceptor {
@@ -90,9 +88,6 @@ public class KubernetesClientMetrics implements Interceptor {
this.responseLatency =
responseMetricGroup.histogram(
HISTO,
OperatorMetricUtils.createHistogram(flinkOperatorConfiguration));
-
- Executors.newSingleThreadScheduledExecutor()
- .scheduleAtFixedRate(this::updateMeters, 0, 1,
TimeUnit.SECONDS);
}
@Override
@@ -140,10 +135,4 @@ public class KubernetesClientMetrics implements
Interceptor {
OperatorMetricUtils.synchronizedCounter(
responseMetricGroup.addGroup(key).counter(COUNTER)));
}
-
- private void updateMeters() {
- this.requestRateMeter.update();
- this.requestFailedRateMeter.update();
- this.responseRateMeter.update();
- }
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetricsTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetricsTest.java
index 70ef11d9..9a87f699 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetricsTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetricsTest.java
@@ -22,9 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.utils.KubernetesClientUtils;
-import org.apache.flink.metrics.testutils.MetricListener;
-import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
@@ -53,7 +51,6 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@TestMethodOrder(OrderAnnotation.class)
public class KubernetesClientMetricsTest {
private KubernetesMockServer mockServer;
- private final MetricListener listener = new MetricListener();
private static final String REQUEST_COUNTER_ID =
String.join(".", KUBE_CLIENT_GROUP, HTTP_REQUEST_GROUP, COUNTER);
@@ -83,65 +80,154 @@ public class KubernetesClientMetricsTest {
@Test
@Order(1)
public void testMetricsDisabled() {
+ var configuration = new Configuration();
+ configuration.set(
+
KubernetesOperatorMetricOptions.OPERATOR_KUBERNETES_CLIENT_METRICS_ENABLED,
false);
+ var flinkConfigManager = new FlinkConfigManager(configuration);
+ var listener = new
TestingMetricListener(flinkConfigManager.getDefaultConfig());
+ var kubernetesClient =
+ KubernetesClientUtils.getKubernetesClient(
+ flinkConfigManager.getOperatorConfiguration(),
+ listener.getMetricGroup(),
+ mockServer.createClient().getConfiguration());
+
var deployment = TestUtils.buildApplicationCluster();
- KubernetesClient noMetricsClient = getKubernetesClient(false);
- noMetricsClient.resource(deployment).get();
- assertFalse(listener.getCounter(REQUEST_COUNTER_ID).isPresent());
- assertFalse(listener.getMeter(REQUEST_METER_ID).isPresent());
-
assertFalse(listener.getCounter(REQUEST_FAILED_COUNTER_ID).isPresent());
- assertFalse(listener.getMeter(REQUEST_FAILED_METER_ID).isPresent());
- assertFalse(listener.getCounter(RESPONSE_COUNTER_ID).isPresent());
- assertFalse(listener.getMeter(RESPONSE_METER_ID).isPresent());
- assertFalse(listener.getHistogram(RESPONSE_LATENCY_ID).isPresent());
- assertFalse(listener.getHistogram(RESPONSE_LATENCY_ID).isPresent());
+ kubernetesClient.resource(deployment).get();
+
assertFalse(listener.getCounter(listener.getMetricId(REQUEST_COUNTER_ID)).isPresent());
+
assertFalse(listener.getMeter(listener.getMetricId(REQUEST_METER_ID)).isPresent());
+ assertFalse(
+
listener.getCounter(listener.getMetricId(REQUEST_FAILED_COUNTER_ID)).isPresent());
+
assertFalse(listener.getMeter(listener.getMetricId(REQUEST_FAILED_METER_ID)).isPresent());
+
assertFalse(listener.getCounter(listener.getMetricId(RESPONSE_COUNTER_ID)).isPresent());
+
assertFalse(listener.getMeter(listener.getMetricId(RESPONSE_METER_ID)).isPresent());
+
assertFalse(listener.getHistogram(listener.getMetricId(RESPONSE_LATENCY_ID)).isPresent());
+
assertFalse(listener.getHistogram(listener.getMetricId(RESPONSE_LATENCY_ID)).isPresent());
}
@Test
@Order(2)
public void testMetricsEnabled() {
- KubernetesClient kubernetesClient = getKubernetesClient(true);
+ var flinkConfigManager = new FlinkConfigManager(new Configuration());
+ var listener = new
TestingMetricListener(flinkConfigManager.getDefaultConfig());
+ var kubernetesClient =
+ KubernetesClientUtils.getKubernetesClient(
+ flinkConfigManager.getOperatorConfiguration(),
+ listener.getMetricGroup(),
+ mockServer.createClient().getConfiguration());
+
var deployment = TestUtils.buildApplicationCluster();
- assertEquals(0,
listener.getCounter(REQUEST_COUNTER_ID).get().getCount());
- assertEquals(0.0, listener.getMeter(REQUEST_METER_ID).get().getRate());
- assertEquals(0,
listener.getCounter(REQUEST_FAILED_COUNTER_ID).get().getCount());
- assertEquals(0.0,
listener.getMeter(REQUEST_FAILED_METER_ID).get().getRate());
- assertEquals(0,
listener.getCounter(RESPONSE_COUNTER_ID).get().getCount());
- assertEquals(0.0,
listener.getMeter(RESPONSE_METER_ID).get().getRate());
- assertEquals(0,
listener.getHistogram(RESPONSE_LATENCY_ID).get().getStatistics().getMin());
- assertEquals(0,
listener.getHistogram(RESPONSE_LATENCY_ID).get().getStatistics().getMax());
+ assertEquals(
+ 0,
listener.getCounter(listener.getMetricId(REQUEST_COUNTER_ID)).get().getCount());
+ assertEquals(
+ 0.0,
listener.getMeter(listener.getMetricId(REQUEST_METER_ID)).get().getRate());
+ assertEquals(
+ 0,
+
listener.getCounter(listener.getMetricId(REQUEST_FAILED_COUNTER_ID))
+ .get()
+ .getCount());
+ assertEquals(
+ 0.0,
+
listener.getMeter(listener.getMetricId(REQUEST_FAILED_METER_ID)).get().getRate());
+ assertEquals(
+ 0,
listener.getCounter(listener.getMetricId(RESPONSE_COUNTER_ID)).get().getCount());
+ assertEquals(
+ 0.0,
listener.getMeter(listener.getMetricId(RESPONSE_METER_ID)).get().getRate());
+ assertEquals(
+ 0,
+
listener.getHistogram(listener.getMetricId(RESPONSE_LATENCY_ID))
+ .get()
+ .getStatistics()
+ .getMin());
+ assertEquals(
+ 0,
+
listener.getHistogram(listener.getMetricId(RESPONSE_LATENCY_ID))
+ .get()
+ .getStatistics()
+ .getMax());
kubernetesClient.resource(deployment).createOrReplace();
- assertEquals(1,
listener.getCounter(REQUEST_COUNTER_ID).get().getCount());
- assertEquals(1,
listener.getCounter(REQUEST_POST_COUNTER_ID).get().getCount());
- assertEquals(1,
listener.getCounter(RESPONSE_COUNTER_ID).get().getCount());
- assertEquals(1,
listener.getCounter(RESPONSE_200_COUNTER_ID).get().getCount());
-
assertTrue(listener.getHistogram(RESPONSE_LATENCY_ID).get().getStatistics().getMin()
> 0);
-
assertTrue(listener.getHistogram(RESPONSE_LATENCY_ID).get().getStatistics().getMax()
> 0);
+ assertEquals(
+ 1,
listener.getCounter(listener.getMetricId(REQUEST_COUNTER_ID)).get().getCount());
+ assertEquals(
+ 1,
+
listener.getCounter(listener.getMetricId(REQUEST_POST_COUNTER_ID))
+ .get()
+ .getCount());
+ assertEquals(
+ 1,
listener.getCounter(listener.getMetricId(RESPONSE_COUNTER_ID)).get().getCount());
+ assertEquals(
+ 1,
+
listener.getCounter(listener.getMetricId(RESPONSE_200_COUNTER_ID))
+ .get()
+ .getCount());
+ assertTrue(
+
listener.getHistogram(listener.getMetricId(RESPONSE_LATENCY_ID))
+ .get()
+ .getStatistics()
+ .getMin()
+ > 0);
+ assertTrue(
+
listener.getHistogram(listener.getMetricId(RESPONSE_LATENCY_ID))
+ .get()
+ .getStatistics()
+ .getMax()
+ > 0);
kubernetesClient.resource(deployment).delete();
- assertEquals(1,
listener.getCounter(REQUEST_DELETE_COUNTER_ID).get().getCount());
+ assertEquals(
+ 1,
+
listener.getCounter(listener.getMetricId(REQUEST_DELETE_COUNTER_ID))
+ .get()
+ .getCount());
kubernetesClient.resource(deployment).delete();
- assertEquals(2,
listener.getCounter(REQUEST_DELETE_COUNTER_ID).get().getCount());
- assertEquals(1,
listener.getCounter(RESPONSE_404_COUNTER_ID).get().getCount());
+ assertEquals(
+ 2,
+
listener.getCounter(listener.getMetricId(REQUEST_DELETE_COUNTER_ID))
+ .get()
+ .getCount());
+ assertEquals(
+ 1,
+
listener.getCounter(listener.getMetricId(RESPONSE_404_COUNTER_ID))
+ .get()
+ .getCount());
Awaitility.await()
.atMost(1, TimeUnit.MINUTES)
.until(
() -> {
kubernetesClient.resource(deployment).createOrReplace();
- return
listener.getMeter(REQUEST_METER_ID).get().getRate() > 0.1
- &&
listener.getMeter(RESPONSE_METER_ID).get().getRate() > 0.1;
+ return
listener.getMeter(listener.getMetricId(REQUEST_METER_ID))
+ .get()
+ .getRate()
+ > 0.01
+ &&
listener.getMeter(listener.getMetricId(RESPONSE_METER_ID))
+ .get()
+ .getRate()
+ > 0.01;
});
}
@Test
@Order(3)
public void testAPIServerIsDown() {
+ var flinkConfigManager = new FlinkConfigManager(new Configuration());
+ var listener = new
TestingMetricListener(flinkConfigManager.getDefaultConfig());
+ var kubernetesClient =
+ KubernetesClientUtils.getKubernetesClient(
+ flinkConfigManager.getOperatorConfiguration(),
+ listener.getMetricGroup(),
+ mockServer.createClient().getConfiguration());
+
var deployment = TestUtils.buildApplicationCluster();
- KubernetesClient kubernetesClient = getKubernetesClient(true);
mockServer.shutdown();
- assertEquals(0,
listener.getCounter(REQUEST_FAILED_COUNTER_ID).get().getCount());
- assertEquals(0.0,
listener.getMeter(REQUEST_FAILED_METER_ID).get().getRate());
+ assertEquals(
+ 0,
+
listener.getCounter(listener.getMetricId(REQUEST_FAILED_COUNTER_ID))
+ .get()
+ .getCount());
+ assertEquals(
+ 0.0,
+
listener.getMeter(listener.getMetricId(REQUEST_FAILED_METER_ID)).get().getRate());
Awaitility.await()
.atMost(1, TimeUnit.MINUTES)
.until(
@@ -149,22 +235,18 @@ public class KubernetesClientMetricsTest {
assertThrows(
KubernetesClientException.class,
() ->
kubernetesClient.resource(deployment).createOrReplace());
- return
listener.getCounter(REQUEST_FAILED_COUNTER_ID).get().getCount()
+ return listener.getCounter(
+
listener.getMetricId(
+
REQUEST_FAILED_COUNTER_ID))
+ .get()
+ .getCount()
> 0
- &&
listener.getMeter(REQUEST_FAILED_METER_ID).get().getRate()
- > 0.1;
+ && listener.getMeter(
+
listener.getMetricId(
+
REQUEST_FAILED_METER_ID))
+ .get()
+ .getRate()
+ > 0.01;
});
}
-
- private KubernetesClient getKubernetesClient(boolean enableMetrics) {
- var configuration = new Configuration();
- configuration.setBoolean(
-
KubernetesOperatorMetricOptions.OPERATOR_KUBERNETES_CLIENT_METRICS_ENABLED,
- enableMetrics);
- var configManager = new FlinkConfigManager(configuration);
- return KubernetesClientUtils.getKubernetesClient(
- configManager.getOperatorConfiguration(),
- listener.getMetricGroup(),
- mockServer.createClient().getConfiguration());
- }
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/TestingMetricListener.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/TestingMetricListener.java
index 3ce5f21b..40c72efe 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/TestingMetricListener.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/TestingMetricListener.java
@@ -24,30 +24,46 @@ import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Histogram;
import org.apache.flink.metrics.Meter;
import org.apache.flink.metrics.Metric;
+import org.apache.flink.metrics.View;
+import org.apache.flink.runtime.metrics.ViewUpdater;
import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
/** Utility class for metrics testing. */
public class TestingMetricListener {
-
public static final String DELIMITER = ".";
- private final KubernetesOperatorMetricGroup metricGroup;
- private final Map<String, Metric> metrics = new HashMap();
private static final String NAMESPACE = "test-op-ns";
private static final String NAME = "test-op-name";
private static final String HOST = "test-op-host";
+ private final KubernetesOperatorMetricGroup metricGroup;
+ private final Map<String, Metric> metrics = new HashMap();
+ private final ScheduledExecutorService executor;
private Configuration configuration;
+ private ViewUpdater viewUpdater;
public TestingMetricListener(Configuration configuration) {
+ this.executor =
+ Executors.newSingleThreadScheduledExecutor(
+ new
ExecutorThreadFactory("Flink-TestingMetricRegistry"));
+
TestingMetricRegistry registry =
TestingMetricRegistry.builder()
.setDelimiter(DELIMITER.charAt(0))
.setRegisterConsumer(
(metric, name, group) -> {
this.metrics.put(group.getMetricIdentifier(name), metric);
+ if (metric instanceof View) {
+ if (viewUpdater == null) {
+ viewUpdater = new
ViewUpdater(executor);
+ }
+ viewUpdater.notifyOfAddedView((View)
metric);
+ }
})
.build();
this.metricGroup =
