This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new b7d3f1515f [Improve][Zeta] Add ReportMetricsOperation observability
exports (#11104)
b7d3f1515f is described below
commit b7d3f1515f015dd2bba498ddcd6218a69ccfdade
Author: Doyeon Kim <[email protected]>
AuthorDate: Sun Jun 21 11:48:54 2026 +0900
[Improve][Zeta] Add ReportMetricsOperation observability exports (#11104)
---
docs/en/engines/zeta/telemetry.md | 9 ++
docs/zh/engines/zeta/telemetry.md | 9 ++
.../engine/server/TaskExecutionService.java | 72 +++++++++++++-
.../metrics/ExportsInstanceInitializer.java | 3 +
.../entity/ReportMetricsOperationStats.java | 39 ++++++++
.../exports/ReportMetricsOperationExports.java | 90 +++++++++++++++++
.../TelemetryCollectorCoordinatorGuardTest.java | 110 +++++++++++++++++++++
7 files changed, 329 insertions(+), 3 deletions(-)
diff --git a/docs/en/engines/zeta/telemetry.md
b/docs/en/engines/zeta/telemetry.md
index a749785a18..7c91348a1d 100644
--- a/docs/en/engines/zeta/telemetry.md
+++ b/docs/en/engines/zeta/telemetry.md
@@ -146,6 +146,15 @@
engine_state_store_connector_jar_total_references{backend="hazelcast"}
| job_thread_pool_task_total | Counter | **address**, server instance
address,for example: "127.0.0.1:5801" | The taskCount of seatunnel coordinator
job's executor cached thread pool |
| job_thread_pool_rejection_total | Counter | **address**, server instance
address,for example: "127.0.0.1:5801" | The rejectionCount of seatunnel
coordinator job's executor cached thread pool |
|
+### Report Metrics Operation
+
+| MetricName | Type | Labels
| DESCRIPTION
|
+|---------------------------------------------------|---------|-----------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------|
+| report_metrics_operation_total | Counter | **address**,
worker instance address,for example: "127.0.0.1:5801". **result**, one of
"success" "failure" "interrupted" | The total number of
`ReportMetricsOperation` invocations sent by a worker
|
+| report_metrics_operation_last_payload_task_count | Gauge | **address**,
worker instance address,for example: "127.0.0.1:5801"
| The number of task metrics included in the most recent
`ReportMetricsOperation` payload sent by a worker |
+| report_metrics_operation_last_invocation_latency_ms | Gauge | **address**,
worker instance address,for example: "127.0.0.1:5801"
| The most recent worker-side `ReportMetricsOperation` reporting latency
in milliseconds, including local metrics collection and worker-to-master
invocation |
+| report_metrics_operation_max_invocation_latency_ms | Gauge | **address**,
worker instance address,for example: "127.0.0.1:5801"
| The maximum observed worker-side `ReportMetricsOperation` reporting
latency in milliseconds since the worker started, including local metrics
collection and worker-to-master invocation |
+
### Job info detail
| MetricName | Type | Labels
|
DESCRIPTION |
diff --git a/docs/zh/engines/zeta/telemetry.md
b/docs/zh/engines/zeta/telemetry.md
index 0da7c4e1b8..e90a320be4 100644
--- a/docs/zh/engines/zeta/telemetry.md
+++ b/docs/zh/engines/zeta/telemetry.md
@@ -145,6 +145,15 @@
engine_state_store_connector_jar_total_references{backend="hazelcast"}
| job_thread_pool_task_total | Counter |
**address**，服务器实例地址，例如："127.0.0.1:5801" | seatunnel 协调器作业执行器缓存线程池的总任务数 |
| job_thread_pool_rejection_total | Counter |
**address**，服务器实例地址，例如："127.0.0.1:5801" | seatunnel 协调器作业执行器缓存线程池的拒绝任务总数 |
+### ReportMetricsOperation 指标
+
+| MetricName | Type | Labels
|
描述
|
+|---------------------------------------------------|---------|--------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------|
+| report_metrics_operation_total | Counter |
**address**，worker 实例地址，例如："127.0.0.1:5801"。**result**，取值包括："success" "failure"
"interrupted" | worker 发送的 `ReportMetricsOperation` 调用总次数
|
+| report_metrics_operation_last_payload_task_count | Gauge |
**address**，worker 实例地址，例如："127.0.0.1:5801"
| 最近一次 `ReportMetricsOperation` payload 中包含的 task metrics 数量
|
+| report_metrics_operation_last_invocation_latency_ms | Gauge |
**address**，worker 实例地址，例如："127.0.0.1:5801"
| worker 侧最近一次 `ReportMetricsOperation` 上报耗时，单位为毫秒，包含本地 metrics 收集和 worker 到
master 的调用时间 |
+| report_metrics_operation_max_invocation_latency_ms | Gauge |
**address**，worker 实例地址，例如："127.0.0.1:5801"
| worker 启动以来观测到的 `ReportMetricsOperation` 最大上报耗时，单位为毫秒，包含本地 metrics 收集和
worker 到 master 的调用时间 |
+
### 作业信息详细
| MetricName | Type | Labels
| 描述 |
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
index a7b02766dc..5e011d5391 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
@@ -53,6 +53,7 @@ import org.apache.seatunnel.engine.server.task.SeaTunnelTask;
import org.apache.seatunnel.engine.server.task.TaskGroupImmutableInformation;
import
org.apache.seatunnel.engine.server.task.operation.NotifyTaskStatusOperation;
import
org.apache.seatunnel.engine.server.task.operation.ReportMetricsOperation;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.ReportMetricsOperationStats;
import org.apache.commons.collections4.CollectionUtils;
@@ -96,6 +97,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
@@ -196,6 +198,13 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
/** SeaTunnel configuration for this engine. */
private final SeaTunnelConfig seaTunnelConfig;
+ // Track worker-side metrics reporting cost without changing the report
path semantics.
+ private final AtomicLong reportMetricsOperationSuccessCount = new
AtomicLong();
+ private final AtomicLong reportMetricsOperationFailureCount = new
AtomicLong();
+ private final AtomicLong reportMetricsOperationInterruptedCount = new
AtomicLong();
+ private final AtomicLong reportMetricsOperationLastPayloadTaskCount = new
AtomicLong();
+ private final AtomicLong reportMetricsOperationLastInvocationLatencyMs =
new AtomicLong();
+ private final AtomicLong reportMetricsOperationMaxInvocationLatencyMs =
new AtomicLong();
/** Scheduled executor for periodic tasks like metrics backup. */
private final ScheduledExecutorService scheduledExecutorService;
@@ -753,26 +762,83 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
return;
}
+ long invocationStartNanos = System.nanoTime();
+ HashMap<TaskLocation, SeaTunnelMetricsContext> localMetricsMap =
collectLocalMetricsMap();
+ int payloadTaskCount = localMetricsMap.size();
InvocationFuture<Object> invoke =
nodeEngine
.getOperationService()
.createInvocationBuilder(
SeaTunnelServer.SERVICE_NAME,
- new
ReportMetricsOperation(collectLocalMetricsMap()),
+ new ReportMetricsOperation(localMetricsMap),
nodeEngine.getMasterAddress())
.invoke();
try {
invoke.get();
+ recordReportMetricsOperationSuccess(
+ payloadTaskCount,
elapsedMillisSince(invocationStartNanos));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- logger.severe("update metrics context stopped due to thread
interruption.", e);
+ long elapsedMillis = elapsedMillisSince(invocationStartNanos);
+ recordReportMetricsOperationInterruption(payloadTaskCount,
elapsedMillis);
+ logger.severe(
+ String.format(
+ "update metrics context stopped due to thread
interruption, "
+ + "payloadTaskCount=%d,
invocationLatencyMs=%d.",
+ payloadTaskCount, elapsedMillis),
+ e);
} catch (Exception e) {
- logger.severe("failed to update metrics", e);
+ long elapsedMillis = elapsedMillisSince(invocationStartNanos);
+ recordReportMetricsOperationFailure(payloadTaskCount,
elapsedMillis);
+ logger.severe(
+ String.format(
+ "failed to update metrics, payloadTaskCount=%d, "
+ + "invocationLatencyMs=%d.",
+ payloadTaskCount, elapsedMillis),
+ e);
}
this.printTaskExecutionRuntimeInfo();
}
+ private void recordReportMetricsOperationSuccess(int payloadTaskCount,
long elapsedMillis) {
+ updateReportMetricsOperationObservability(payloadTaskCount,
elapsedMillis);
+ reportMetricsOperationSuccessCount.incrementAndGet();
+ }
+
+ private void recordReportMetricsOperationFailure(int payloadTaskCount,
long elapsedMillis) {
+ updateReportMetricsOperationObservability(payloadTaskCount,
elapsedMillis);
+ reportMetricsOperationFailureCount.incrementAndGet();
+ }
+
+ private void recordReportMetricsOperationInterruption(
+ int payloadTaskCount, long elapsedMillis) {
+ updateReportMetricsOperationObservability(payloadTaskCount,
elapsedMillis);
+ reportMetricsOperationInterruptedCount.incrementAndGet();
+ }
+
+ private void updateReportMetricsOperationObservability(
+ int payloadTaskCount, long elapsedMillis) {
+ reportMetricsOperationLastPayloadTaskCount.set(payloadTaskCount);
+ reportMetricsOperationLastInvocationLatencyMs.set(elapsedMillis);
+
reportMetricsOperationMaxInvocationLatencyMs.accumulateAndGet(elapsedMillis,
Math::max);
+ }
+
+ private long elapsedMillisSince(long startNanos) {
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ }
+
+ /** Returns the latest worker-side ReportMetricsOperation observability
snapshot. */
+ public ReportMetricsOperationStats getReportMetricsOperationStats() {
+ return new ReportMetricsOperationStats(
+ reportMetricsOperationSuccessCount.get(),
+ reportMetricsOperationFailureCount.get(),
+ reportMetricsOperationInterruptedCount.get(),
+ reportMetricsOperationLastPayloadTaskCount.get(),
+ reportMetricsOperationLastInvocationLatencyMs.get(),
+ reportMetricsOperationMaxInvocationLatencyMs.get());
+ }
+
private HashMap<TaskLocation, SeaTunnelMetricsContext>
collectLocalMetricsMap() {
Map<TaskGroupLocation, TaskGroupContext> contextMap = new HashMap<>();
contextMap.putAll(finishedExecutionContexts);
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/ExportsInstanceInitializer.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/ExportsInstanceInitializer.java
index d4b6cba6ec..4b759aba5f 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/ExportsInstanceInitializer.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/ExportsInstanceInitializer.java
@@ -23,6 +23,7 @@ import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.EngineStateS
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobMetricExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobThreadPoolStatusExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.NodeMetricExports;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.ReportMetricsOperationExports;
import com.hazelcast.instance.impl.Node;
import io.prometheus.client.CollectorRegistry;
@@ -47,6 +48,8 @@ public final class ExportsInstanceInitializer {
new JobThreadPoolStatusExports(node).register(collectorRegistry);
// Node metrics
new NodeMetricExports(node).register(collectorRegistry);
+ // ReportMetricsOperation metrics
+ new
ReportMetricsOperationExports(node).register(collectorRegistry);
// Engine state store metrics
new
EngineStateStoreMetricExports(node).register(collectorRegistry);
// Engine state store logical metrics
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/entity/ReportMetricsOperationStats.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/entity/ReportMetricsOperationStats.java
new file mode 100644
index 0000000000..ddd80fe89d
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/entity/ReportMetricsOperationStats.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.engine.server.telemetry.metrics.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+/** Snapshot of worker-side ReportMetricsOperation observability state. */
+@Data
+@AllArgsConstructor
+public class ReportMetricsOperationStats {
+ /** Total successful report invocations sent by this worker. */
+ private long successCount;
+ /** Total failed report invocations sent by this worker. */
+ private long failureCount;
+ /** Total interrupted report invocations sent by this worker. */
+ private long interruptedCount;
+ /** Task metric count in the most recent report payload. */
+ private long lastPayloadTaskCount;
+ /** Most recent worker-side reporting latency in milliseconds. */
+ private long lastInvocationLatencyMs;
+ /** Maximum observed worker-side reporting latency in milliseconds. */
+ private long maxInvocationLatencyMs;
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/ReportMetricsOperationExports.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/ReportMetricsOperationExports.java
new file mode 100644
index 0000000000..0f46c64765
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/ReportMetricsOperationExports.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.engine.server.telemetry.metrics.exports;
+
+import org.apache.seatunnel.engine.server.TaskExecutionService;
+import org.apache.seatunnel.engine.server.telemetry.metrics.AbstractCollector;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.ReportMetricsOperationStats;
+
+import com.hazelcast.instance.impl.Node;
+import io.prometheus.client.CounterMetricFamily;
+import io.prometheus.client.GaugeMetricFamily;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ReportMetricsOperationExports extends AbstractCollector {
+
+ public ReportMetricsOperationExports(Node node) {
+ super(node);
+ }
+
+ @Override
+ public List<MetricFamilySamples> collect() {
+ List<MetricFamilySamples> mfs = new ArrayList();
+ TaskExecutionService taskExecutionService =
getServer().getTaskExecutionService();
+ if (taskExecutionService == null) {
+ return mfs;
+ }
+
+ String address = localAddress();
+ ReportMetricsOperationStats stats =
taskExecutionService.getReportMetricsOperationStats();
+
+ CounterMetricFamily totalMetricFamily =
+ new CounterMetricFamily(
+ "report_metrics_operation",
+ "The total number of ReportMetricsOperation
invocations sent by a worker",
+ clusterLabelNames(ADDRESS, "result"));
+ totalMetricFamily.addMetric(labelValues(address, "success"),
stats.getSuccessCount());
+ totalMetricFamily.addMetric(labelValues(address, "failure"),
stats.getFailureCount());
+ totalMetricFamily.addMetric(
+ labelValues(address, "interrupted"),
stats.getInterruptedCount());
+ mfs.add(totalMetricFamily);
+
+ GaugeMetricFamily payloadMetricFamily =
+ new GaugeMetricFamily(
+ "report_metrics_operation_last_payload_task_count",
+ "The number of task metrics included in the most
recent "
+ + "ReportMetricsOperation payload sent by a
worker",
+ clusterLabelNames(ADDRESS));
+ payloadMetricFamily.addMetric(labelValues(address),
stats.getLastPayloadTaskCount());
+ mfs.add(payloadMetricFamily);
+
+ GaugeMetricFamily lastLatencyMetricFamily =
+ new GaugeMetricFamily(
+ "report_metrics_operation_last_invocation_latency_ms",
+ "The most recent worker-side ReportMetricsOperation
reporting latency "
+ + "in milliseconds, including local metrics
collection and "
+ + "worker-to-master invocation",
+ clusterLabelNames(ADDRESS));
+ lastLatencyMetricFamily.addMetric(labelValues(address),
stats.getLastInvocationLatencyMs());
+ mfs.add(lastLatencyMetricFamily);
+
+ GaugeMetricFamily maxLatencyMetricFamily =
+ new GaugeMetricFamily(
+ "report_metrics_operation_max_invocation_latency_ms",
+ "The maximum observed worker-side
ReportMetricsOperation reporting "
+ + "latency in milliseconds since the worker
started, "
+ + "including local metrics collection and
worker-to-master "
+ + "invocation",
+ clusterLabelNames(ADDRESS));
+ maxLatencyMetricFamily.addMetric(labelValues(address),
stats.getMaxInvocationLatencyMs());
+ mfs.add(maxLatencyMetricFamily);
+ return mfs;
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
index c41a923580..03edb148a4 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
@@ -19,11 +19,14 @@ package
org.apache.seatunnel.engine.server.telemetry.metrics;
import org.apache.seatunnel.engine.server.CoordinatorService;
import org.apache.seatunnel.engine.server.SeaTunnelServer;
+import org.apache.seatunnel.engine.server.TaskExecutionService;
import org.apache.seatunnel.engine.server.telemetry.metrics.entity.JobCounter;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.ReportMetricsOperationStats;
import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.ThreadPoolStatus;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.ClusterMetricExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobMetricExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobThreadPoolStatusExports;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.ReportMetricsOperationExports;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -181,6 +184,76 @@ public class TelemetryCollectorCoordinatorGuardTest {
Mockito.verify(mockServer, Mockito.never()).isCoordinatorActive();
}
+ @Test
+ void
testReportMetricsOperationExportsReturnsEmptyWhenTaskExecutionServiceMissing() {
+ Mockito.when(mockServer.getTaskExecutionService()).thenReturn(null);
+
+ ReportMetricsOperationExports exports = new
ReportMetricsOperationExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertTrue(
+ result.isEmpty(),
+ "collect() must return empty when task execution service is
unavailable");
+ }
+
+ @Test
+ void
testReportMetricsOperationExportsReturnsMetricsWhenTaskExecutionServiceAvailable()
{
+ TaskExecutionService taskExecutionService =
Mockito.mock(TaskExecutionService.class);
+
Mockito.when(mockServer.getTaskExecutionService()).thenReturn(taskExecutionService);
+ Mockito.when(taskExecutionService.getReportMetricsOperationStats())
+ .thenReturn(new ReportMetricsOperationStats(3L, 1L, 1L, 9L,
15L, 27L));
+
+ ReportMetricsOperationExports exports = new
ReportMetricsOperationExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertFalse(
+ result.isEmpty(),
+ "collect() must return metrics when task execution service
exists");
+ Collector.MetricFamilySamples totalMetric =
+ result.stream()
+ .filter(s -> "report_metrics_operation".equals(s.name))
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(totalMetric);
+ Assertions.assertEquals(3, totalMetric.samples.size());
+ assertMetricSample(totalMetric, "report_metrics_operation_total",
"success", 3D);
+ assertMetricSample(totalMetric, "report_metrics_operation_total",
"failure", 1D);
+ assertMetricSample(totalMetric, "report_metrics_operation_total",
"interrupted", 1D);
+
+ Collector.MetricFamilySamples payloadMetric =
+ result.stream()
+ .filter(
+ s ->
+
"report_metrics_operation_last_payload_task_count"
+ .equals(s.name))
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(payloadMetric);
+ assertSingleMetricSample(payloadMetric, 9D);
+
+ Collector.MetricFamilySamples lastLatencyMetric =
+ result.stream()
+ .filter(
+ s ->
+
"report_metrics_operation_last_invocation_latency_ms"
+ .equals(s.name))
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(lastLatencyMetric);
+ assertSingleMetricSample(lastLatencyMetric, 15D);
+
+ Collector.MetricFamilySamples maxLatencyMetric =
+ result.stream()
+ .filter(
+ s ->
+
"report_metrics_operation_max_invocation_latency_ms"
+ .equals(s.name))
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(maxLatencyMetric);
+ assertSingleMetricSample(maxLatencyMetric, 27D);
+ }
+
//
-------------------------------------------------------------------------
// ClusterMetricExports
//
-------------------------------------------------------------------------
@@ -237,4 +310,41 @@ public class TelemetryCollectorCoordinatorGuardTest {
Mockito.eq("Skip cluster_info metric: unable to
resolve master address"),
Mockito.any(UnknownHostException.class));
}
+
+ private void assertMetricSample(
+ Collector.MetricFamilySamples metricFamilySamples,
+ String expectedSampleName,
+ String result,
+ double expectedValue) {
+ Collector.MetricFamilySamples.Sample sample =
+ metricFamilySamples.samples.stream()
+ .filter(
+ s -> {
+ if (!expectedSampleName.equals(s.name)) {
+ return false;
+ }
+ int resultLabelIndex =
s.labelNames.indexOf("result");
+ return resultLabelIndex >= 0
+ &&
result.equals(s.labelValues.get(resultLabelIndex));
+ })
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(sample);
+ assertAddressLabel(sample);
+ Assertions.assertEquals(expectedValue, sample.value);
+ }
+
+ private void assertSingleMetricSample(
+ Collector.MetricFamilySamples metricFamilySamples, double
expectedValue) {
+ Assertions.assertEquals(1, metricFamilySamples.samples.size());
+ Collector.MetricFamilySamples.Sample sample =
metricFamilySamples.samples.get(0);
+ assertAddressLabel(sample);
+ Assertions.assertEquals(expectedValue, sample.value);
+ }
+
+ private void assertAddressLabel(Collector.MetricFamilySamples.Sample
sample) {
+ int addressLabelIndex = sample.labelNames.indexOf("address");
+ Assertions.assertTrue(addressLabelIndex >= 0, "metric sample must
contain 'address' label");
+ Assertions.assertEquals("127.0.0.1:5801",
sample.labelValues.get(addressLabelIndex));
+ }
}
