-
Notifications
You must be signed in to change notification settings - Fork 116
[controller] Add OTel metrics to DeferredVersionSwapStats #2522
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
m-nagarajan
merged 3 commits into
linkedin:main
from
m-nagarajan:addOtelMetricsForDeferredVersionSwapService
Feb 25, 2026
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
195 changes: 158 additions & 37 deletions
195
...ntroller/src/main/java/com/linkedin/venice/controller/stats/DeferredVersionSwapStats.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,63 +1,184 @@ | ||
| package com.linkedin.venice.controller.stats; | ||
|
|
||
| import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_CLUSTER_NAME; | ||
| import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_STORE_NAME; | ||
| import static com.linkedin.venice.utils.Utils.setOf; | ||
|
|
||
| import com.google.common.collect.ImmutableMap; | ||
| import com.linkedin.venice.stats.AbstractVeniceStats; | ||
| import com.linkedin.venice.stats.OpenTelemetryMetricsSetup; | ||
| import com.linkedin.venice.stats.VeniceOpenTelemetryMetricsRepository; | ||
| import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions; | ||
| import com.linkedin.venice.stats.metrics.MetricEntity; | ||
| import com.linkedin.venice.stats.metrics.MetricEntityStateBase; | ||
| import com.linkedin.venice.stats.metrics.MetricEntityStateGeneric; | ||
| import com.linkedin.venice.stats.metrics.MetricType; | ||
| import com.linkedin.venice.stats.metrics.MetricUnit; | ||
| import com.linkedin.venice.stats.metrics.ModuleMetricEntityInterface; | ||
| import com.linkedin.venice.stats.metrics.TehutiMetricNameEnum; | ||
| import io.opentelemetry.api.common.Attributes; | ||
| import io.tehuti.metrics.MetricsRepository; | ||
| import io.tehuti.metrics.Sensor; | ||
| import io.tehuti.metrics.stats.Count; | ||
| import io.tehuti.metrics.stats.Gauge; | ||
| import java.util.Collections; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
|
|
||
|
|
||
| public class DeferredVersionSwapStats extends AbstractVeniceStats { | ||
| private final Sensor deferredVersionSwapErrorSensor; | ||
| private final Sensor deferredVersionSwapThrowableSensor; | ||
| private final Sensor deferredVersionSwapFailedRollForwardSensor; | ||
| private final Sensor deferredVersionSwapStalledVersionSwapSensor; | ||
| private final Sensor deferredVersionSwapParentChildStatusMismatchSensor; | ||
| private final Sensor deferredVersionSwapChildStatusMismatchSensor; | ||
| private final static String DEFERRED_VERSION_SWAP_ERROR = "deferred_version_swap_error"; | ||
| private final static String DEFERRED_VERSION_SWAP_THROWABLE = "deferred_version_swap_throwable"; | ||
| private final static String DEFERRED_VERSION_SWAP_FAILED_ROLL_FORWARD = "deferred_version_swap_failed_roll_forward"; | ||
| private static final String DEFERRED_VERSION_SWAP_STALLED_VERSION_SWAP = "deferred_version_swap_stalled_version_swap"; | ||
| private static final String DEFERRED_VERSION_SWAP_PARENT_CHILD_STATUS_MISMATCH_SENSOR = | ||
| "deferred_version_swap_parent_child_status_mismatch"; | ||
| private static final String DEFERRED_VERSION_SWAP_CHILD_STATUS_MISMATCH_SENSOR = | ||
| "deferred_version_swap_child_status_mismatch"; | ||
| private final MetricEntityStateGeneric deferredVersionSwapThrowableMetric; | ||
| private final MetricEntityStateGeneric deferredVersionSwapExceptionMetric; | ||
|
|
||
| private final MetricEntityStateGeneric deferredVersionSwapFailedRollForwardMetric; | ||
| private final MetricEntityStateBase deferredVersionSwapStalledVersionSwapMetric; | ||
| private final MetricEntityStateGeneric deferredVersionSwapParentChildStatusMismatchMetric; | ||
| private final MetricEntityStateGeneric deferredVersionSwapChildStatusMismatchMetric; | ||
|
|
||
| public DeferredVersionSwapStats(MetricsRepository metricsRepository) { | ||
| super(metricsRepository, "DeferredVersionSwap"); | ||
| deferredVersionSwapErrorSensor = registerSensorIfAbsent(DEFERRED_VERSION_SWAP_ERROR, new Count()); | ||
| deferredVersionSwapThrowableSensor = registerSensorIfAbsent(DEFERRED_VERSION_SWAP_THROWABLE, new Count()); | ||
| deferredVersionSwapFailedRollForwardSensor = | ||
| registerSensorIfAbsent(DEFERRED_VERSION_SWAP_FAILED_ROLL_FORWARD, new Count()); | ||
| deferredVersionSwapStalledVersionSwapSensor = | ||
| registerSensorIfAbsent(DEFERRED_VERSION_SWAP_STALLED_VERSION_SWAP, new Gauge()); | ||
| deferredVersionSwapParentChildStatusMismatchSensor = | ||
| registerSensorIfAbsent(DEFERRED_VERSION_SWAP_PARENT_CHILD_STATUS_MISMATCH_SENSOR, new Count()); | ||
| deferredVersionSwapChildStatusMismatchSensor = | ||
| registerSensorIfAbsent(DEFERRED_VERSION_SWAP_CHILD_STATUS_MISMATCH_SENSOR, new Count()); | ||
|
|
||
| OpenTelemetryMetricsSetup.OpenTelemetryMetricsSetupInfo otelData = | ||
| OpenTelemetryMetricsSetup.builder(metricsRepository).build(); | ||
| VeniceOpenTelemetryMetricsRepository otelRepository = otelData.getOtelRepository(); | ||
| Map<VeniceMetricsDimensions, String> baseDimensionsMap = otelData.getBaseDimensionsMap(); | ||
| Attributes baseAttributes = otelData.getBaseAttributes(); | ||
|
|
||
| deferredVersionSwapExceptionMetric = MetricEntityStateGeneric.create( | ||
| DeferredVersionSwapOtelMetricEntity.DEFERRED_VERSION_SWAP_PROCESSING_ERROR_COUNT.getMetricEntity(), | ||
| otelRepository, | ||
| this::registerSensorIfAbsent, | ||
| DeferredVersionSwapTehutiMetricNameEnum.DEFERRED_VERSION_SWAP_ERROR, | ||
| Collections.singletonList(new Count()), | ||
| baseDimensionsMap); | ||
|
|
||
| deferredVersionSwapThrowableMetric = MetricEntityStateGeneric.create( | ||
pthirun marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| DeferredVersionSwapOtelMetricEntity.DEFERRED_VERSION_SWAP_PROCESSING_ERROR_COUNT.getMetricEntity(), | ||
| otelRepository, | ||
| this::registerSensorIfAbsent, | ||
| DeferredVersionSwapTehutiMetricNameEnum.DEFERRED_VERSION_SWAP_THROWABLE, | ||
| Collections.singletonList(new Count()), | ||
| baseDimensionsMap); | ||
|
|
||
| deferredVersionSwapFailedRollForwardMetric = MetricEntityStateGeneric.create( | ||
| DeferredVersionSwapOtelMetricEntity.DEFERRED_VERSION_SWAP_ROLL_FORWARD_FAILURE_COUNT.getMetricEntity(), | ||
pthirun marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| otelRepository, | ||
| this::registerSensorIfAbsent, | ||
| DeferredVersionSwapTehutiMetricNameEnum.DEFERRED_VERSION_SWAP_FAILED_ROLL_FORWARD, | ||
| Collections.singletonList(new Count()), | ||
| baseDimensionsMap); | ||
|
|
||
| deferredVersionSwapStalledVersionSwapMetric = MetricEntityStateBase.create( | ||
| DeferredVersionSwapOtelMetricEntity.DEFERRED_VERSION_SWAP_STALLED_COUNT.getMetricEntity(), | ||
pthirun marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| otelRepository, | ||
| this::registerSensorIfAbsent, | ||
| DeferredVersionSwapTehutiMetricNameEnum.DEFERRED_VERSION_SWAP_STALLED_VERSION_SWAP, | ||
| Collections.singletonList(new Gauge()), | ||
| baseDimensionsMap, | ||
| baseAttributes); | ||
|
|
||
| deferredVersionSwapParentChildStatusMismatchMetric = MetricEntityStateGeneric.create( | ||
| DeferredVersionSwapOtelMetricEntity.DEFERRED_VERSION_SWAP_PARENT_STATUS_MISMATCH_COUNT.getMetricEntity(), | ||
| otelRepository, | ||
| this::registerSensorIfAbsent, | ||
| DeferredVersionSwapTehutiMetricNameEnum.DEFERRED_VERSION_SWAP_PARENT_CHILD_STATUS_MISMATCH, | ||
| Collections.singletonList(new Count()), | ||
| baseDimensionsMap); | ||
|
|
||
| deferredVersionSwapChildStatusMismatchMetric = MetricEntityStateGeneric.create( | ||
| DeferredVersionSwapOtelMetricEntity.DEFERRED_VERSION_SWAP_CHILD_STATUS_MISMATCH_COUNT.getMetricEntity(), | ||
| otelRepository, | ||
| this::registerSensorIfAbsent, | ||
| DeferredVersionSwapTehutiMetricNameEnum.DEFERRED_VERSION_SWAP_CHILD_STATUS_MISMATCH, | ||
| Collections.singletonList(new Count()), | ||
| baseDimensionsMap); | ||
| } | ||
|
|
||
| public void recordDeferredVersionSwapExceptionMetric(String clusterName) { | ||
| deferredVersionSwapExceptionMetric.record(1, clusterDimensions(clusterName)); | ||
| } | ||
|
|
||
| public void recordDeferredVersionSwapErrorSensor() { | ||
| deferredVersionSwapErrorSensor.record(); | ||
| public void recordDeferredVersionSwapThrowableMetric(String clusterName) { | ||
| deferredVersionSwapThrowableMetric.record(1, clusterDimensions(clusterName)); | ||
| } | ||
|
|
||
| public void recordDeferredVersionSwapThrowableSensor() { | ||
| deferredVersionSwapThrowableSensor.record(); | ||
| public void recordDeferredVersionSwapFailedRollForwardMetric(String clusterName, String storeName) { | ||
| deferredVersionSwapFailedRollForwardMetric.record(1, clusterAndStoreDimensions(clusterName, storeName)); | ||
| } | ||
|
|
||
| public void recordDeferredVersionSwapFailedRollForwardSensor() { | ||
| deferredVersionSwapFailedRollForwardSensor.record(); | ||
| public void recordDeferredVersionSwapStalledVersionSwapMetric(double value) { | ||
| deferredVersionSwapStalledVersionSwapMetric.record(value); | ||
| } | ||
|
|
||
| public void recordDeferredVersionSwapStalledVersionSwapSensor(double value) { | ||
| deferredVersionSwapStalledVersionSwapSensor.record(value); | ||
| public void recordDeferredVersionSwapParentChildStatusMismatchMetric(String clusterName, String storeName) { | ||
| deferredVersionSwapParentChildStatusMismatchMetric.record(1, clusterAndStoreDimensions(clusterName, storeName)); | ||
| } | ||
|
|
||
| public void recordDeferredVersionSwapParentChildStatusMismatchSensor() { | ||
| deferredVersionSwapParentChildStatusMismatchSensor.record(); | ||
| public void recordDeferredVersionSwapChildStatusMismatchMetric(String clusterName, String storeName) { | ||
| deferredVersionSwapChildStatusMismatchMetric.record(1, clusterAndStoreDimensions(clusterName, storeName)); | ||
| } | ||
|
|
||
| public void recordDeferredVersionSwapChildStatusMismatchSensor() { | ||
| deferredVersionSwapChildStatusMismatchSensor.record(); | ||
| private static Map<VeniceMetricsDimensions, String> clusterDimensions(String clusterName) { | ||
| return Collections.singletonMap(VENICE_CLUSTER_NAME, clusterName); | ||
| } | ||
|
|
||
| private static Map<VeniceMetricsDimensions, String> clusterAndStoreDimensions(String clusterName, String storeName) { | ||
| return ImmutableMap.of(VENICE_CLUSTER_NAME, clusterName, VENICE_STORE_NAME, storeName); | ||
| } | ||
|
|
||
| enum DeferredVersionSwapTehutiMetricNameEnum implements TehutiMetricNameEnum { | ||
| DEFERRED_VERSION_SWAP_ERROR, DEFERRED_VERSION_SWAP_THROWABLE, DEFERRED_VERSION_SWAP_FAILED_ROLL_FORWARD, | ||
| DEFERRED_VERSION_SWAP_STALLED_VERSION_SWAP, DEFERRED_VERSION_SWAP_PARENT_CHILD_STATUS_MISMATCH, | ||
| DEFERRED_VERSION_SWAP_CHILD_STATUS_MISMATCH | ||
| } | ||
|
|
||
| public enum DeferredVersionSwapOtelMetricEntity implements ModuleMetricEntityInterface { | ||
| /** Count of unexpected failures (both {@link Exception} and {@link Throwable}) in the per-cluster processing loop */ | ||
| DEFERRED_VERSION_SWAP_PROCESSING_ERROR_COUNT( | ||
| "deferred_version_swap.processing_error_count", MetricType.COUNTER, MetricUnit.NUMBER, | ||
| "Count of unexpected failures in the deferred version swap processing loop", setOf(VENICE_CLUSTER_NAME) | ||
| ), | ||
| /** Count of deferred version swap roll forward failures */ | ||
| DEFERRED_VERSION_SWAP_ROLL_FORWARD_FAILURE_COUNT( | ||
| "deferred_version_swap.roll_forward.failure_count", MetricType.COUNTER, MetricUnit.NUMBER, | ||
| "Count of deferred version swap roll forward failures", setOf(VENICE_CLUSTER_NAME, VENICE_STORE_NAME) | ||
| ), | ||
| /** Gauge of stalled deferred version swaps (global — stalledVersionSwapSet is shared across all clusters) */ | ||
| DEFERRED_VERSION_SWAP_STALLED_COUNT( | ||
| MetricEntity.createWithNoDimensions( | ||
| "deferred_version_swap.stalled_count", | ||
| MetricType.GAUGE, | ||
| MetricUnit.NUMBER, | ||
| "Count of stalled deferred version swaps across all clusters") | ||
| ), | ||
| /** Count of deferred version swap parent-child status mismatches */ | ||
| DEFERRED_VERSION_SWAP_PARENT_STATUS_MISMATCH_COUNT( | ||
| "deferred_version_swap.parent_status_mismatch_count", MetricType.COUNTER, MetricUnit.NUMBER, | ||
| "Count of deferred version swap parent-child status mismatches", setOf(VENICE_CLUSTER_NAME, VENICE_STORE_NAME) | ||
| ), | ||
| /** Count of deferred version swap child status mismatches */ | ||
| DEFERRED_VERSION_SWAP_CHILD_STATUS_MISMATCH_COUNT( | ||
| "deferred_version_swap.child_status_mismatch_count", MetricType.COUNTER, MetricUnit.NUMBER, | ||
| "Count of deferred version swap child status mismatches", setOf(VENICE_CLUSTER_NAME, VENICE_STORE_NAME) | ||
| ); | ||
|
|
||
| private final MetricEntity metricEntity; | ||
|
|
||
| DeferredVersionSwapOtelMetricEntity( | ||
| String metricName, | ||
| MetricType metricType, | ||
| MetricUnit unit, | ||
| String description, | ||
| Set<VeniceMetricsDimensions> dimensionsList) { | ||
| this.metricEntity = new MetricEntity(metricName, metricType, unit, description, dimensionsList); | ||
| } | ||
|
|
||
| DeferredVersionSwapOtelMetricEntity(MetricEntity metricEntity) { | ||
| this.metricEntity = metricEntity; | ||
| } | ||
|
|
||
| @Override | ||
| public MetricEntity getMetricEntity() { | ||
| return metricEntity; | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.