You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/17 20:20:02 UTC
(pulsar) 08/22: [feat][monitor] PIP-320: Add OpenTelemetry metrics for the Function Worker (#26823)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8180b59e4dfb6ddd366306735bba38ecea70d478
Author: Dream95 <86...@qq.com>
AuthorDate: Sat Oct 17 21:45:40 2026 +0800
[feat][monitor] PIP-320: Add OpenTelemetry metrics for the Function Worker (#26823)
Signed-off-by: Dream95 <zh...@163.com>
(cherry picked from commit 846f21bc578c54108272d0351d58d640471874cb)
---
pulsar-functions/worker/build.gradle.kts | 1 +
.../functions/worker/OpenTelemetryWorkerStats.java | 241 +++++++++++++++++++++
.../functions/worker/PulsarWorkerService.java | 8 +-
.../functions/worker/WorkerStatsManager.java | 88 ++++++--
.../pulsar/functions/worker/rest/WorkerServer.java | 7 +-
.../worker/OpenTelemetryWorkerStatsTest.java | 218 +++++++++++++++++++
6 files changed, 538 insertions(+), 25 deletions(-)
diff --git a/pulsar-functions/worker/build.gradle.kts b/pulsar-functions/worker/build.gradle.kts
index d0cc152ea5e9..079ab95afc5e 100644
--- a/pulsar-functions/worker/build.gradle.kts
+++ b/pulsar-functions/worker/build.gradle.kts
@@ -86,6 +86,7 @@ dependencies {
compileOnly(libs.swagger.annotations)
testImplementation(libs.protobuf.java.util)
+ testImplementation(libs.opentelemetry.sdk.testing)
testImplementation(project(":pulsar-functions:pulsar-functions-api-examples"))
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/OpenTelemetryWorkerStats.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/OpenTelemetryWorkerStats.java
new file mode 100644
index 000000000000..425ca882d3e1
--- /dev/null
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/OpenTelemetryWorkerStats.java
@@ -0,0 +1,241 @@
+/*
+ * 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.pulsar.functions.worker;
+
+import io.opentelemetry.api.metrics.DoubleHistogram;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.ObservableDoubleGauge;
+import io.opentelemetry.api.metrics.ObservableLongGauge;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import lombok.Setter;
+import org.apache.pulsar.common.stats.MetricsUtil;
+import org.apache.pulsar.functions.proto.FunctionMetaData;
+
+/**
+ * OpenTelemetry instruments for Function Worker operational metrics.
+ *
+ * <p>Registers instruments on the Function Worker Meter from {@link PulsarWorkerOpenTelemetry}.
+ */
+public class OpenTelemetryWorkerStats implements AutoCloseable {
+
+ // Replaces pulsar_function_worker_start_up_time_ms
+ public static final String STARTUP_DURATION = "pulsar.function_worker.startup.duration";
+
+ // Replaces pulsar_function_worker_instance_count
+ public static final String INSTANCE_COUNT = "pulsar.function_worker.instance.count";
+
+ // Replaces pulsar_function_worker_total_expected_instance_count
+ public static final String EXPECTED_INSTANCE_COUNT = "pulsar.function_worker.instance.expected.count";
+
+ // Replaces pulsar_function_worker_total_function_count
+ public static final String FUNCTION_COUNT = "pulsar.function_worker.function.count";
+
+ // Replaces pulsar_function_worker_schedule_execution_time_total_ms
+ public static final String SCHEDULE_DURATION = "pulsar.function_worker.schedule.duration";
+
+ // Replaces pulsar_function_worker_schedule_strategy_execution_time_ms
+ public static final String SCHEDULE_STRATEGY_DURATION = "pulsar.function_worker.schedule.strategy.duration";
+
+ // Replaces pulsar_function_worker_rebalance_execution_time_total_ms
+ public static final String REBALANCE_DURATION = "pulsar.function_worker.rebalance.duration";
+
+ // Replaces pulsar_function_worker_rebalance_strategy_execution_time_ms
+ public static final String REBALANCE_STRATEGY_DURATION = "pulsar.function_worker.rebalance.strategy.duration";
+
+ // Replaces pulsar_function_worker_stop_instance_process_time_ms
+ public static final String INSTANCE_STOP_DURATION = "pulsar.function_worker.instance.stop.duration";
+
+ // Replaces pulsar_function_worker_start_instance_process_time_ms
+ public static final String INSTANCE_START_DURATION = "pulsar.function_worker.instance.start.duration";
+
+ // Replaces pulsar_function_worker_drain_execution_time_total_ms
+ public static final String DRAIN_DURATION = "pulsar.function_worker.drain.duration";
+
+ // Replaces pulsar_function_worker_is_leader
+ public static final String LEADER = "pulsar.function_worker.leader";
+
+ private static final List<Double> DURATION_BUCKETS = Arrays.asList(
+ .0005, .001, .0025, .005, .01, .025, .05, .1, .25, .5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0);
+
+ @Setter
+ private volatile FunctionRuntimeManager functionRuntimeManager;
+
+ @Setter
+ private volatile FunctionMetaDataManager functionMetaDataManager;
+
+ @Setter
+ private volatile Supplier<Boolean> isLeader;
+
+ private volatile double startupDurationSeconds = Double.NaN;
+
+ private final DoubleHistogram scheduleDuration;
+ private final DoubleHistogram scheduleStrategyDuration;
+ private final DoubleHistogram rebalanceDuration;
+ private final DoubleHistogram rebalanceStrategyDuration;
+ private final DoubleHistogram instanceStopDuration;
+ private final DoubleHistogram instanceStartDuration;
+ private final DoubleHistogram drainDuration;
+
+ private final ObservableDoubleGauge startupDurationGauge;
+ private final ObservableLongGauge instanceCountGauge;
+ private final ObservableLongGauge expectedInstanceCountGauge;
+ private final ObservableLongGauge functionCountGauge;
+ private final ObservableLongGauge leaderGauge;
+
+ public OpenTelemetryWorkerStats(Meter meter) {
+ scheduleDuration = durationHistogram(meter, SCHEDULE_DURATION,
+ "Total execution time of a schedule cycle.");
+ scheduleStrategyDuration = durationHistogram(meter, SCHEDULE_STRATEGY_DURATION,
+ "Execution time of the schedule strategy.");
+ rebalanceDuration = durationHistogram(meter, REBALANCE_DURATION,
+ "Total execution time of a rebalance cycle.");
+ rebalanceStrategyDuration = durationHistogram(meter, REBALANCE_STRATEGY_DURATION,
+ "Execution time of the rebalance strategy.");
+ instanceStopDuration = durationHistogram(meter, INSTANCE_STOP_DURATION,
+ "Time taken to stop a function instance.");
+ instanceStartDuration = durationHistogram(meter, INSTANCE_START_DURATION,
+ "Time taken to start a function instance.");
+ drainDuration = durationHistogram(meter, DRAIN_DURATION,
+ "Total execution time of a drain cycle.");
+
+ startupDurationGauge = meter.gaugeBuilder(STARTUP_DURATION)
+ .setDescription("Worker service startup time.")
+ .setUnit("s")
+ .buildWithCallback(measurement -> {
+ double duration = startupDurationSeconds;
+ if (!Double.isNaN(duration)) {
+ measurement.record(duration);
+ }
+ });
+
+ instanceCountGauge = meter.gaugeBuilder(INSTANCE_COUNT)
+ .ofLongs()
+ .setDescription("Number of function instances running on this worker.")
+ .setUnit("{instance}")
+ .buildWithCallback(measurement -> {
+ FunctionRuntimeManager manager = functionRuntimeManager;
+ if (manager != null) {
+ measurement.record(manager.getMyInstances());
+ }
+ });
+
+ expectedInstanceCountGauge = meter.gaugeBuilder(EXPECTED_INSTANCE_COUNT)
+ .ofLongs()
+ .setDescription("Total expected function instance count in the cluster.")
+ .setUnit("{instance}")
+ .buildWithCallback(measurement -> {
+ if (!isClusterLeader()) {
+ return;
+ }
+ FunctionMetaDataManager manager = functionMetaDataManager;
+ if (manager != null) {
+ measurement.record(expectedInstanceCount(manager));
+ }
+ });
+
+ functionCountGauge = meter.gaugeBuilder(FUNCTION_COUNT)
+ .ofLongs()
+ .setDescription("Total number of functions in the cluster.")
+ .setUnit("{function}")
+ .buildWithCallback(measurement -> {
+ if (!isClusterLeader()) {
+ return;
+ }
+ FunctionMetaDataManager manager = functionMetaDataManager;
+ if (manager != null) {
+ measurement.record(manager.getAllFunctionMetaData().size());
+ }
+ });
+
+ leaderGauge = meter.gaugeBuilder(LEADER)
+ .ofLongs()
+ .setDescription("Whether this worker is the functions cluster leader.")
+ .setUnit("1")
+ .buildWithCallback(measurement -> measurement.record(isClusterLeader() ? 1 : 0));
+ }
+
+ public void recordStartupDuration(long durationNanos) {
+ startupDurationSeconds = MetricsUtil.convertToSeconds(durationNanos, TimeUnit.NANOSECONDS);
+ }
+
+ public void recordScheduleDuration(long durationNanos) {
+ scheduleDuration.record(toSeconds(durationNanos));
+ }
+
+ public void recordScheduleStrategyDuration(long durationNanos) {
+ scheduleStrategyDuration.record(toSeconds(durationNanos));
+ }
+
+ public void recordRebalanceDuration(long durationNanos) {
+ rebalanceDuration.record(toSeconds(durationNanos));
+ }
+
+ public void recordRebalanceStrategyDuration(long durationNanos) {
+ rebalanceStrategyDuration.record(toSeconds(durationNanos));
+ }
+
+ public void recordInstanceStopDuration(long durationNanos) {
+ instanceStopDuration.record(toSeconds(durationNanos));
+ }
+
+ public void recordInstanceStartDuration(long durationNanos) {
+ instanceStartDuration.record(toSeconds(durationNanos));
+ }
+
+ public void recordDrainDuration(long durationNanos) {
+ drainDuration.record(toSeconds(durationNanos));
+ }
+
+ @Override
+ public void close() {
+ startupDurationGauge.close();
+ instanceCountGauge.close();
+ expectedInstanceCountGauge.close();
+ functionCountGauge.close();
+ leaderGauge.close();
+ }
+
+ private boolean isClusterLeader() {
+ Supplier<Boolean> leader = isLeader;
+ return leader != null && Boolean.TRUE.equals(leader.get());
+ }
+
+ private static long expectedInstanceCount(FunctionMetaDataManager manager) {
+ long total = 0;
+ for (FunctionMetaData entry : manager.getAllFunctionMetaData()) {
+ total += entry.getFunctionDetails().getParallelism();
+ }
+ return total;
+ }
+
+ private static double toSeconds(long durationNanos) {
+ return MetricsUtil.convertToSeconds(durationNanos, TimeUnit.NANOSECONDS);
+ }
+
+ private static DoubleHistogram durationHistogram(Meter meter, String name, String description) {
+ return meter.histogramBuilder(name)
+ .setDescription(description)
+ .setUnit("s")
+ .setExplicitBucketBoundariesAdvice(DURATION_BUCKETS)
+ .build();
+ }
+}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java
index 6c96ba488735..ed963f6cd571 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java
@@ -190,7 +190,7 @@ public class PulsarWorkerService implements WorkerService {
this.openTelemetry = new PulsarWorkerOpenTelemetry(workerConfig);
this.workerConfig = workerConfig;
this.dlogUri = dlogUri;
- this.workerStatsManager = new WorkerStatsManager(workerConfig, runAsStandalone);
+ this.workerStatsManager = new WorkerStatsManager(workerConfig, runAsStandalone, openTelemetry.getMeter());
this.functions = new FunctionsImpl(() -> PulsarWorkerService.this);
this.functionsV2 = new FunctionsImplV2(() -> PulsarWorkerService.this);
this.sinks = new SinksImpl(() -> PulsarWorkerService.this);
@@ -610,7 +610,7 @@ public class PulsarWorkerService implements WorkerService {
workerStatsManager.setFunctionRuntimeManager(functionRuntimeManager);
workerStatsManager.setFunctionMetaDataManager(functionMetaDataManager);
workerStatsManager.setLeaderService(leaderService);
- workerStatsManager.setIsLeader(checkIsStillLeader);
+ workerStatsManager.setIsLeader(leaderService::isLeader);
workerStatsManager.startupTimeEnd();
} catch (Throwable t) {
log.error().exception(t).log("Error Starting up in worker");
@@ -692,6 +692,10 @@ public class PulsarWorkerService implements WorkerService {
stateStoreProvider.close();
}
+ if (null != workerStatsManager) {
+ workerStatsManager.close();
+ }
+
if (null != openTelemetry) {
openTelemetry.close();
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerStatsManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerStatsManager.java
index 82b96ae45208..d287ff115200 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerStatsManager.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerStatsManager.java
@@ -19,6 +19,8 @@
package org.apache.pulsar.functions.worker;
import static org.apache.pulsar.common.stats.JvmMetrics.getJvmDirectMemoryUsed;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.metrics.Meter;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
@@ -31,8 +33,9 @@ import lombok.Setter;
import org.apache.pulsar.common.util.DirectMemoryUtils;
import org.apache.pulsar.functions.instance.stats.PrometheusTextFormat;
import org.apache.pulsar.functions.proto.FunctionMetaData;
+import org.apache.pulsar.opentelemetry.annotations.PulsarDeprecatedMetric;
-public class WorkerStatsManager {
+public class WorkerStatsManager implements AutoCloseable {
static {
DefaultExports.initialize();
@@ -41,7 +44,9 @@ public class WorkerStatsManager {
private static final String PULSAR_FUNCTION_WORKER_METRICS_PREFIX = "pulsar_function_worker_";
private static final String START_UP_TIME = "start_up_time_ms";
private static final String INSTANCE_COUNT = "instance_count";
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.EXPECTED_INSTANCE_COUNT)
private static final String TOTAL_EXPECTED_INSTANCE_COUNT = "total_expected_instance_count";
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.FUNCTION_COUNT)
private static final String TOTAL_FUNCTIONS_COUNT = "total_function_count";
private static final String SCHEDULE_TOTAL_EXEC_TIME = "schedule_execution_time_total_ms";
private static final String SCHEDULE_STRATEGY_EXEC_TIME = "schedule_strategy_execution_time_ms";
@@ -50,34 +55,43 @@ public class WorkerStatsManager {
private static final String STOPPING_INSTANCE_PROCESS_TIME = "stop_instance_process_time_ms";
private static final String STARTING_INSTANCE_PROCESS_TIME = "start_instance_process_time_ms";
private static final String DRAIN_TOTAL_EXEC_TIME = "drain_execution_time_total_ms";
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.LEADER)
private static final String IS_LEADER = "is_leader";
private static final String[] metricsLabelNames = {"cluster"};
private final String[] metricsLabels;
- @Setter
private FunctionRuntimeManager functionRuntimeManager;
- @Setter
private FunctionMetaDataManager functionMetaDataManager;
@Setter
private LeaderService leaderService;
- @Setter
private Supplier<Boolean> isLeader;
private CollectorRegistry collectorRegistry = new CollectorRegistry();
+ private final OpenTelemetryWorkerStats openTelemetryWorkerStats;
+
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.STARTUP_DURATION)
private final Summary statWorkerStartupTime;
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.INSTANCE_COUNT)
private final Gauge statNumInstances;
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.SCHEDULE_DURATION)
private final Summary scheduleTotalExecutionTime;
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.SCHEDULE_STRATEGY_DURATION)
private final Summary scheduleStrategyExecutionTime;
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.REBALANCE_DURATION)
private final Summary rebalanceTotalExecutionTime;
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.REBALANCE_STRATEGY_DURATION)
private final Summary rebalanceStrategyExecutionTime;
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.INSTANCE_STOP_DURATION)
private final Summary stopInstanceProcessTime;
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.INSTANCE_START_DURATION)
private final Summary startInstanceProcessTime;
+ @PulsarDeprecatedMetric(newMetricName = OpenTelemetryWorkerStats.DRAIN_DURATION)
private final Summary drainTotalExecutionTime;
// As an optimization
@@ -92,8 +106,14 @@ public class WorkerStatsManager {
private final Summary.Child drainTotalExecutionTimeChild;
public WorkerStatsManager(WorkerConfig workerConfig, boolean runAsStandalone) {
+ this(workerConfig, runAsStandalone,
+ OpenTelemetry.noop().getMeter(PulsarWorkerOpenTelemetry.INSTRUMENTATION_SCOPE_NAME));
+ }
+
+ public WorkerStatsManager(WorkerConfig workerConfig, boolean runAsStandalone, Meter meter) {
metricsLabels = new String[]{workerConfig.getPulsarFunctionsCluster()};
+ this.openTelemetryWorkerStats = new OpenTelemetryWorkerStats(meter);
statWorkerStartupTime = Summary.build()
.name(PULSAR_FUNCTION_WORKER_METRICS_PREFIX + START_UP_TIME)
@@ -196,6 +216,21 @@ public class WorkerStatsManager {
}
}
+ public void setFunctionRuntimeManager(FunctionRuntimeManager functionRuntimeManager) {
+ this.functionRuntimeManager = functionRuntimeManager;
+ openTelemetryWorkerStats.setFunctionRuntimeManager(functionRuntimeManager);
+ }
+
+ public void setFunctionMetaDataManager(FunctionMetaDataManager functionMetaDataManager) {
+ this.functionMetaDataManager = functionMetaDataManager;
+ openTelemetryWorkerStats.setFunctionMetaDataManager(functionMetaDataManager);
+ }
+
+ public void setIsLeader(Supplier<Boolean> isLeader) {
+ this.isLeader = isLeader;
+ openTelemetryWorkerStats.setIsLeader(isLeader);
+ }
+
private Long startupTimeStart;
public void startupTimeStart() {
startupTimeStart = System.nanoTime();
@@ -203,8 +238,9 @@ public class WorkerStatsManager {
public void startupTimeEnd() {
if (startupTimeStart != null) {
- double endTimeMs = ((double) System.nanoTime() - startupTimeStart) / 1.0E6D;
- statWorkerStartupTimeChild.observe(endTimeMs);
+ long elapsedNanos = System.nanoTime() - startupTimeStart;
+ statWorkerStartupTimeChild.observe(elapsedNanos / 1.0E6D);
+ openTelemetryWorkerStats.recordStartupDuration(elapsedNanos);
}
}
@@ -215,8 +251,9 @@ public class WorkerStatsManager {
public void scheduleTotalExecTimeEnd() {
if (scheduleTotalExecTimeStart != null) {
- double endTimeMs = ((double) System.nanoTime() - scheduleTotalExecTimeStart) / 1.0E6D;
- scheduleTotalExecutionTimeChild.observe(endTimeMs);
+ long elapsedNanos = System.nanoTime() - scheduleTotalExecTimeStart;
+ scheduleTotalExecutionTimeChild.observe(elapsedNanos / 1.0E6D);
+ openTelemetryWorkerStats.recordScheduleDuration(elapsedNanos);
}
}
@@ -227,8 +264,9 @@ public class WorkerStatsManager {
public void scheduleStrategyExecTimeStartEnd() {
if (scheduleStrategyExecTimeStart != null) {
- double endTimeMs = ((double) System.nanoTime() - scheduleStrategyExecTimeStart) / 1.0E6D;
- scheduleStrategyExecutionTimeChild.observe(endTimeMs);
+ long elapsedNanos = System.nanoTime() - scheduleStrategyExecTimeStart;
+ scheduleStrategyExecutionTimeChild.observe(elapsedNanos / 1.0E6D);
+ openTelemetryWorkerStats.recordScheduleStrategyDuration(elapsedNanos);
}
}
@@ -239,8 +277,9 @@ public class WorkerStatsManager {
public void rebalanceTotalExecTimeEnd() {
if (rebalanceTotalExecTimeStart != null) {
- double endTimeMs = ((double) System.nanoTime() - rebalanceTotalExecTimeStart) / 1.0E6D;
- rebalanceTotalExecutionTimeChild.observe(endTimeMs);
+ long elapsedNanos = System.nanoTime() - rebalanceTotalExecTimeStart;
+ rebalanceTotalExecutionTimeChild.observe(elapsedNanos / 1.0E6D);
+ openTelemetryWorkerStats.recordRebalanceDuration(elapsedNanos);
}
}
@@ -251,8 +290,9 @@ public class WorkerStatsManager {
public void rebalanceStrategyExecTimeEnd() {
if (rebalanceStrategyExecTimeStart != null) {
- double endTimeMs = ((double) System.nanoTime() - rebalanceStrategyExecTimeStart) / 1.0E6D;
- rebalanceStrategyExecutionTimeChild.observe(endTimeMs);
+ long elapsedNanos = System.nanoTime() - rebalanceStrategyExecTimeStart;
+ rebalanceStrategyExecutionTimeChild.observe(elapsedNanos / 1.0E6D);
+ openTelemetryWorkerStats.recordRebalanceStrategyDuration(elapsedNanos);
}
}
@@ -263,8 +303,9 @@ public class WorkerStatsManager {
public void drainTotalExecTimeEnd() {
if (drainTotalExecTimeStart != null) {
- double endTimeMs = ((double) System.nanoTime() - drainTotalExecTimeStart) / 1.0E6D;
- drainTotalExecutionTimeChild.observe(endTimeMs);
+ long elapsedNanos = System.nanoTime() - drainTotalExecTimeStart;
+ drainTotalExecutionTimeChild.observe(elapsedNanos / 1.0E6D);
+ openTelemetryWorkerStats.recordDrainDuration(elapsedNanos);
}
}
@@ -275,8 +316,9 @@ public class WorkerStatsManager {
public void stopInstanceProcessTimeEnd() {
if (stopInstanceProcessTimeStart != null) {
- double endTimeMs = ((double) System.nanoTime() - stopInstanceProcessTimeStart) / 1.0E6D;
- stopInstanceProcessTimeChild.observe(endTimeMs);
+ long elapsedNanos = System.nanoTime() - stopInstanceProcessTimeStart;
+ stopInstanceProcessTimeChild.observe(elapsedNanos / 1.0E6D);
+ openTelemetryWorkerStats.recordInstanceStopDuration(elapsedNanos);
}
}
@@ -287,11 +329,17 @@ public class WorkerStatsManager {
public void startInstanceProcessTimeEnd() {
if (startInstanceProcessTimeStart != null) {
- double endTimeMs = ((double) System.nanoTime() - startInstanceProcessTimeStart) / 1.0E6D;
- startInstanceProcessTimeChild.observe(endTimeMs);
+ long elapsedNanos = System.nanoTime() - startInstanceProcessTimeStart;
+ startInstanceProcessTimeChild.observe(elapsedNanos / 1.0E6D);
+ openTelemetryWorkerStats.recordInstanceStartDuration(elapsedNanos);
}
}
+ @Override
+ public void close() {
+ openTelemetryWorkerStats.close();
+ }
+
public String getStatsAsString() throws IOException {
statNumInstancesChild.set(functionRuntimeManager.getMyInstances());
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java
index 80e123c622b7..12b586769afa 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java
@@ -98,7 +98,7 @@ public class WorkerServer {
this.authenticationService = authenticationService;
this.webServerExecutor = new WebExecutorThreadPool(this.workerConfig.getNumHttpServerThreads(), "function-web",
this.workerConfig.getHttpServerThreadPoolQueueSize());
- this.filterInitializer = new FilterInitializer(workerConfig, authenticationService);
+ this.filterInitializer = new FilterInitializer(workerConfig, authenticationService, workerOpenTelemetry());
init();
}
@@ -216,11 +216,12 @@ public class WorkerServer {
private final List<FilterHolder> filterHolders = new ArrayList<>();
private final FilterHolder authenticationFilterHolder;
- FilterInitializer(WorkerConfig config, AuthenticationService authenticationService) {
+ FilterInitializer(WorkerConfig config, AuthenticationService authenticationService,
+ OpenTelemetry openTelemetry) {
if (config.isHttpRequestsLimitEnabled()) {
filterHolders.add(new FilterHolder(
new RateLimitingFilter(config.getHttpRequestsMaxPerSecond(),
- OpenTelemetry.noop().getMeter(PulsarWorkerOpenTelemetry.INSTRUMENTATION_SCOPE_NAME))));
+ openTelemetry.getMeter(PulsarWorkerOpenTelemetry.INSTRUMENTATION_SCOPE_NAME))));
}
if (config.isAuthenticationEnabled()) {
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/OpenTelemetryWorkerStatsTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/OpenTelemetryWorkerStatsTest.java
new file mode 100644
index 000000000000..ea0c1c0ebefb
--- /dev/null
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/OpenTelemetryWorkerStatsTest.java
@@ -0,0 +1,218 @@
+/*
+ * 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.pulsar.functions.worker;
+
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.pulsar.functions.proto.FunctionMetaData;
+import org.apache.pulsar.opentelemetry.OpenTelemetryService;
+import org.assertj.core.api.Assertions;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class OpenTelemetryWorkerStatsTest {
+
+ private static final long FIVE_MILLIS_NANOS = TimeUnit.MILLISECONDS.toNanos(5);
+
+ private OpenTelemetryService openTelemetryService;
+ private InMemoryMetricReader reader;
+ private OpenTelemetryWorkerStats stats;
+
+ @BeforeMethod
+ public void setup() {
+ reader = InMemoryMetricReader.create();
+ openTelemetryService = OpenTelemetryService.builder()
+ .clusterName("test-cluster")
+ .serviceName(PulsarWorkerOpenTelemetry.SERVICE_NAME)
+ .builderCustomizer(builder -> builder
+ .disableShutdownHook()
+ .addMeterProviderCustomizer((providerBuilder, config) ->
+ providerBuilder.registerMetricReader(reader))
+ .addPropertiesSupplier(() -> Map.of(
+ OpenTelemetryService.OTEL_SDK_DISABLED_KEY, "false",
+ "otel.metrics.exporter", "none",
+ "otel.traces.exporter", "none",
+ "otel.logs.exporter", "none")))
+ .build();
+ var meter = openTelemetryService.getOpenTelemetry()
+ .getMeter(PulsarWorkerOpenTelemetry.INSTRUMENTATION_SCOPE_NAME);
+ stats = new OpenTelemetryWorkerStats(meter);
+ }
+
+ @AfterMethod
+ public void teardown() throws Exception {
+ stats.close();
+ openTelemetryService.close();
+ reader.close();
+ }
+
+ @Test
+ public void testDurationHistograms() {
+ stats.recordScheduleDuration(FIVE_MILLIS_NANOS);
+ stats.recordScheduleStrategyDuration(FIVE_MILLIS_NANOS);
+ stats.recordRebalanceDuration(FIVE_MILLIS_NANOS);
+ stats.recordRebalanceStrategyDuration(FIVE_MILLIS_NANOS);
+ stats.recordInstanceStartDuration(FIVE_MILLIS_NANOS);
+ stats.recordInstanceStopDuration(FIVE_MILLIS_NANOS);
+ stats.recordDrainDuration(FIVE_MILLIS_NANOS);
+
+ var metrics = reader.collectAllMetrics();
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.SCHEDULE_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.SCHEDULE_STRATEGY_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.REBALANCE_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.REBALANCE_STRATEGY_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.INSTANCE_START_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.INSTANCE_STOP_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.DRAIN_DURATION);
+ }
+
+ @Test
+ public void testGaugesWhenLeader() {
+ FunctionRuntimeManager runtimeManager = mock(FunctionRuntimeManager.class);
+ when(runtimeManager.getMyInstances()).thenReturn(4);
+
+ FunctionMetaData function1 = new FunctionMetaData();
+ function1.setFunctionDetails().setName("fn-1").setParallelism(2);
+ FunctionMetaData function2 = new FunctionMetaData();
+ function2.setFunctionDetails().setName("fn-2").setParallelism(3);
+ FunctionMetaDataManager metaDataManager = mock(FunctionMetaDataManager.class);
+ when(metaDataManager.getAllFunctionMetaData()).thenReturn(List.of(function1, function2));
+
+ stats.setFunctionRuntimeManager(runtimeManager);
+ stats.setFunctionMetaDataManager(metaDataManager);
+ stats.setIsLeader(() -> true);
+ stats.recordStartupDuration(FIVE_MILLIS_NANOS);
+
+ var metrics = reader.collectAllMetrics();
+ assertLongGauge(metrics, OpenTelemetryWorkerStats.INSTANCE_COUNT, 4);
+ assertLongGauge(metrics, OpenTelemetryWorkerStats.FUNCTION_COUNT, 2);
+ assertLongGauge(metrics, OpenTelemetryWorkerStats.EXPECTED_INSTANCE_COUNT, 5);
+ assertLongGauge(metrics, OpenTelemetryWorkerStats.LEADER, 1);
+ assertThat(metrics).anySatisfy(metric -> assertThat(metric)
+ .hasName(OpenTelemetryWorkerStats.STARTUP_DURATION)
+ .hasDoubleGaugeSatisfying(gauge ->
+ gauge.hasPointsSatisfying(point -> point.hasValue(0.005))));
+ }
+
+ @Test
+ public void testGaugesWhenNotLeader() {
+ FunctionRuntimeManager runtimeManager = mock(FunctionRuntimeManager.class);
+ when(runtimeManager.getMyInstances()).thenReturn(1);
+ FunctionMetaData function = new FunctionMetaData();
+ function.setFunctionDetails().setName("fn-1").setParallelism(1);
+ FunctionMetaDataManager metaDataManager = mock(FunctionMetaDataManager.class);
+ when(metaDataManager.getAllFunctionMetaData()).thenReturn(List.of(function));
+ stats.setFunctionRuntimeManager(runtimeManager);
+ stats.setFunctionMetaDataManager(metaDataManager);
+ stats.setIsLeader(() -> false);
+
+ var metrics = reader.collectAllMetrics();
+ assertLongGauge(metrics, OpenTelemetryWorkerStats.INSTANCE_COUNT, 1);
+ assertLongGauge(metrics, OpenTelemetryWorkerStats.LEADER, 0);
+ Assertions.assertThat(metrics)
+ .noneMatch(metric -> OpenTelemetryWorkerStats.FUNCTION_COUNT.equals(metric.getName()))
+ .noneMatch(metric -> OpenTelemetryWorkerStats.EXPECTED_INSTANCE_COUNT.equals(metric.getName()));
+ }
+
+ @Test
+ public void testWorkerStatsManagerDualPublish() throws Exception {
+ var localReader = InMemoryMetricReader.create();
+ var localOpenTelemetryService = OpenTelemetryService.builder()
+ .clusterName("test-cluster")
+ .serviceName(PulsarWorkerOpenTelemetry.SERVICE_NAME)
+ .builderCustomizer(builder -> builder
+ .disableShutdownHook()
+ .addMeterProviderCustomizer((providerBuilder, config) ->
+ providerBuilder.registerMetricReader(localReader))
+ .addPropertiesSupplier(() -> Map.of(
+ OpenTelemetryService.OTEL_SDK_DISABLED_KEY, "false",
+ "otel.metrics.exporter", "none",
+ "otel.traces.exporter", "none",
+ "otel.logs.exporter", "none")))
+ .build();
+ WorkerConfig workerConfig = new WorkerConfig().setPulsarFunctionsCluster("test-cluster");
+ var meter = localOpenTelemetryService.getOpenTelemetry()
+ .getMeter(PulsarWorkerOpenTelemetry.INSTRUMENTATION_SCOPE_NAME);
+ WorkerStatsManager workerStatsManager = new WorkerStatsManager(workerConfig, false, meter);
+
+ FunctionRuntimeManager runtimeManager = mock(FunctionRuntimeManager.class);
+ when(runtimeManager.getMyInstances()).thenReturn(2);
+ workerStatsManager.setFunctionRuntimeManager(runtimeManager);
+ workerStatsManager.setIsLeader(new AtomicBoolean(true)::get);
+
+ workerStatsManager.startupTimeStart();
+ workerStatsManager.startupTimeEnd();
+ workerStatsManager.scheduleTotalExecTimeStart();
+ workerStatsManager.scheduleTotalExecTimeEnd();
+ workerStatsManager.scheduleStrategyExecTimeStartStart();
+ workerStatsManager.scheduleStrategyExecTimeStartEnd();
+ workerStatsManager.rebalanceTotalExecTimeStart();
+ workerStatsManager.rebalanceTotalExecTimeEnd();
+ workerStatsManager.rebalanceStrategyExecTimeStart();
+ workerStatsManager.rebalanceStrategyExecTimeEnd();
+ workerStatsManager.startInstanceProcessTimeStart();
+ workerStatsManager.startInstanceProcessTimeEnd();
+ workerStatsManager.stopInstanceProcessTimeStart();
+ workerStatsManager.stopInstanceProcessTimeEnd();
+ workerStatsManager.drainTotalExecTimeStart();
+ workerStatsManager.drainTotalExecTimeEnd();
+
+ try {
+ var metrics = localReader.collectAllMetrics();
+ assertThat(metrics).anySatisfy(metric ->
+ assertThat(metric).hasName(OpenTelemetryWorkerStats.STARTUP_DURATION));
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.SCHEDULE_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.SCHEDULE_STRATEGY_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.REBALANCE_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.REBALANCE_STRATEGY_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.INSTANCE_START_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.INSTANCE_STOP_DURATION);
+ assertHistogramCount(metrics, OpenTelemetryWorkerStats.DRAIN_DURATION);
+ assertLongGauge(metrics, OpenTelemetryWorkerStats.INSTANCE_COUNT, 2);
+ assertLongGauge(metrics, OpenTelemetryWorkerStats.LEADER, 1);
+ } finally {
+ workerStatsManager.close();
+ localOpenTelemetryService.close();
+ localReader.close();
+ }
+ }
+
+ private static void assertHistogramCount(Collection<MetricData> metrics, String name) {
+ assertThat(metrics).anySatisfy(metric -> assertThat(metric)
+ .hasName(name)
+ .hasHistogramSatisfying(histogram ->
+ histogram.hasPointsSatisfying(point -> point.hasCount(1))));
+ }
+
+ private static void assertLongGauge(Collection<MetricData> metrics, String name, long value) {
+ assertThat(metrics).anySatisfy(metric -> assertThat(metric)
+ .hasName(name)
+ .hasLongGaugeSatisfying(gauge ->
+ gauge.hasPointsSatisfying(point -> point.hasValue(value))));
+ }
+}