[da-vinci][server] Redesign ServerReadQuotaUsageStats OTel metrics as

counters with outcome dimension

Replace 6 OTel metrics (ASYNC_GAUGE + COUNTER) with 3 consolidated metrics:
- read.quota.request.count (ASYNC_COUNTER_FOR_HIGH_PERF_CASES)
- read.quota.key.count (ASYNC_COUNTER_FOR_HIGH_PERF_CASES)
- read.quota.usage_ratio (ASYNC_DOUBLE_GAUGE, unchanged)

New QuotaRequestOutcome dimension (ALLOWED/REJECTED/ALLOWED_UNINTENTIONALLY)
combined with VersionRole provides per-outcome, per-role breakdown.

Also adds ASYNC_DOUBLE_GAUGE MetricType and RATIO MetricUnit to the
metric infra framework, with DoubleSupplier support in
AsyncMetricEntityState/Base/OneEnum.

API change: recordAllowedUnintentionally now takes a version parameter
for OTel version role classification.

Author: m-nagarajan

clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ServerMetricEntity.java

@@ -38,7 +38,8 @@ public static List<Class<? extends ModuleMetricEntityInterface>> getMetricEntity
         HeartbeatMonitoringOtelMetricEntity.class,
         BlobTransferOtelMetricEntity.class,
         KafkaConsumerServiceOtelMetricEntity.class,
-        RocksDBMemoryOtelMetricEntity.class);
+        RocksDBMemoryOtelMetricEntity.class,
+        ServerReadQuotaOtelMetricEntity.class);
   }
 
   public static final Collection<MetricEntity> SERVER_METRIC_ENTITIES =

clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ServerReadQuotaOtelMetricEntity.java

@@ -0,0 +1,59 @@
+package com.linkedin.davinci.stats;
+
+import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_CLUSTER_NAME;
+import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_QUOTA_REQUEST_OUTCOME;
+import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_STORE_NAME;
+import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_VERSION_ROLE;
+import static com.linkedin.venice.utils.Utils.setOf;
+
+import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
+import com.linkedin.venice.stats.metrics.MetricEntity;
+import com.linkedin.venice.stats.metrics.MetricType;
+import com.linkedin.venice.stats.metrics.MetricUnit;
+import com.linkedin.venice.stats.metrics.ModuleMetricEntityInterface;
+import java.util.Set;
+
+
+/**
+ * OTel metric entities for server read quota usage stats.
+ *
+ * <p>Consolidates Tehuti sensors into 3 OTel metrics:
+ * <ul>
+ *   <li>{@link #READ_QUOTA_REQUEST_COUNT} — high-perf counter with outcome and version role dimensions</li>
+ *   <li>{@link #READ_QUOTA_KEY_COUNT} — high-perf counter with outcome and version role dimensions</li>
+ *   <li>{@link #READ_QUOTA_USAGE_RATIO} — async double gauge (unchanged)</li>
+ * </ul>
+ */
+public enum ServerReadQuotaOtelMetricEntity implements ModuleMetricEntityInterface {
+  READ_QUOTA_REQUEST_COUNT(
+      "read.quota.request.count", MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES, MetricUnit.NUMBER,
+      "Count of read quota requests per outcome and version role",
+      setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_QUOTA_REQUEST_OUTCOME)
+  ),
+  READ_QUOTA_KEY_COUNT(
+      "read.quota.key.count", MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES, MetricUnit.NUMBER,
+      "Count of read quota keys (RCU) per outcome and version role",
+      setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_QUOTA_REQUEST_OUTCOME)
+  ),
+  READ_QUOTA_USAGE_RATIO(
+      "read.quota.usage_ratio", MetricType.ASYNC_DOUBLE_GAUGE, MetricUnit.RATIO,
+      "Ratio of read quota used, based on requested keys per second relative to the node's quota responsibility",
+      setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
+  );
+
+  private final MetricEntity metricEntity;
+
+  ServerReadQuotaOtelMetricEntity(
+      String metricName,
+      MetricType metricType,
+      MetricUnit unit,
+      String description,
+      Set<VeniceMetricsDimensions> dimensions) {
+    this.metricEntity = new MetricEntity(metricName, metricType, unit, description, dimensions);
+  }
+
+  @Override
+  public MetricEntity getMetricEntity() {
+    return metricEntity;
+  }
+}

clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntity.java

@@ -37,8 +37,8 @@ public enum IngestionOtelMetricEntity implements ModuleMetricEntityInterface {
   ),
 
   DISK_QUOTA_USED(
-      "ingestion.disk_quota.used", MetricType.ASYNC_GAUGE, MetricUnit.RATIO, "Disk quota used for the store version",
-      setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE)
+      "ingestion.disk_quota.used", MetricType.ASYNC_DOUBLE_GAUGE, MetricUnit.RATIO,
+      "Disk quota usage ratio for the store version", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE)
   ),
 
   INGESTION_RECORDS_CONSUMED(

clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStats.java

@@ -441,8 +441,8 @@ private long getPushTimeoutCountForRole(VersionRole role) {
     return pushTimeoutByVersion.getOrDefault(version, 0);
   }
 
-  private long getDiskQuotaUsedForRole(VersionRole role) {
-    return (long) (IngestionStatsUtils.getStorageQuotaUsed(getTaskForRole(role)) * 100);
+  private double getDiskQuotaUsedForRole(VersionRole role) {
+    return IngestionStatsUtils.getStorageQuotaUsed(getTaskForRole(role));
   }
 
   private long getIdleTimeForRole(VersionRole role) {

clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ServerMetricEntityTest.java

@@ -22,7 +22,7 @@
 public class ServerMetricEntityTest {
   @Test
   public void testServerMetricEntitiesCount() {
-    assertEquals(SERVER_METRIC_ENTITIES.size(), 132, "Expected 132 unique metric entities");
+    assertEquals(SERVER_METRIC_ENTITIES.size(), 135, "Expected 135 unique metric entities");
   }
 
   /**

clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntityTest.java

@@ -85,9 +85,9 @@ private static Map<IngestionOtelMetricEntity, MetricEntityExpectation> expectedD
         IngestionOtelMetricEntity.DISK_QUOTA_USED,
         new MetricEntityExpectation(
             "ingestion.disk_quota.used",
-            MetricType.ASYNC_GAUGE,
+            MetricType.ASYNC_DOUBLE_GAUGE,
             MetricUnit.RATIO,
-            "Disk quota used for the store version",
+            "Disk quota usage ratio for the store version",
             storeClusterVersion));
     map.put(
         IngestionOtelMetricEntity.CONSUMER_IDLE_TIME,

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/VeniceOpenTelemetryMetricsRepository.java

@@ -26,6 +26,7 @@
 import io.opentelemetry.api.metrics.LongUpDownCounterBuilder;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.api.metrics.ObservableDoubleGauge;
 import io.opentelemetry.api.metrics.ObservableLongCounter;
 import io.opentelemetry.api.metrics.ObservableLongGauge;
 import io.opentelemetry.api.metrics.ObservableLongMeasurement;
@@ -55,6 +56,7 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
+import java.util.function.DoubleSupplier;
 import java.util.function.LongSupplier;
 import javax.annotation.Nonnull;
 import org.apache.logging.log4j.LogManager;
@@ -423,6 +425,45 @@ public ObservableLongGauge createAsyncLongGauge(
         });
   }
 
+  /**
+   * Asynchronous double gauge that will call the callback during metrics collection.
+   * Use this for metrics requiring fractional precision (e.g., ratios in [0.0, 1.0]).
+   *
+   * <p>Each call creates a new SDK instrument handle via {@code buildWithCallback} — there is no
+   * deduplication. Multiple callers can register callbacks for the same metric name; the OTel SDK
+   * natively aggregates all their data points during collection.
+   */
+  public ObservableDoubleGauge createAsyncDoubleGauge(
+      MetricEntity metricEntity,
+      @Nonnull DoubleSupplier asyncCallback,
+      @Nonnull Attributes attributes) {
+    if (!emitOpenTelemetryMetrics()) {
+      return null;
+    }
+    return meter.gaugeBuilder(getFullMetricName(metricEntity))
+        .setUnit(metricEntity.getUnit().name())
+        .setDescription(getMetricDescription(metricEntity, metricsConfig))
+        .buildWithCallback(measurement -> {
+          double v;
+          try {
+            v = asyncCallback.getAsDouble();
+          } catch (Exception e) {
+            recordFailureMetric(metricEntity, e);
+            return;
+          }
+          measurement.record(v, attributes);
+        });
+  }
+
+  public Object createInstrument(MetricEntity metricEntity, DoubleSupplier asyncDoubleCallback, Attributes attributes) {
+    if (metricEntity.getMetricType() != MetricType.ASYNC_DOUBLE_GAUGE) {
+      throw new IllegalArgumentException(
+          "DoubleSupplier callback requires ASYNC_DOUBLE_GAUGE metric type, but got: " + metricEntity.getMetricType()
+              + " for metric: " + metricEntity.getMetricName());
+    }
+    return createAsyncDoubleGauge(metricEntity, asyncDoubleCallback, attributes);
+  }
+
   public Object createInstrument(MetricEntity metricEntity, LongSupplier asyncCallback, Attributes attributes) {
     MetricType metricType = metricEntity.getMetricType();
     switch (metricType) {
@@ -442,6 +483,11 @@ public Object createInstrument(MetricEntity metricEntity, LongSupplier asyncCall
       case ASYNC_GAUGE:
         return createAsyncLongGauge(metricEntity, asyncCallback, attributes);
 
+      case ASYNC_DOUBLE_GAUGE:
+        throw new IllegalArgumentException(
+            "ASYNC_DOUBLE_GAUGE requires DoubleSupplier callback. Use createInstrument(MetricEntity, DoubleSupplier, Attributes) instead. Metric: "
+                + metricEntity.getMetricName());
+
       case ASYNC_COUNTER_FOR_HIGH_PERF_CASES:
       case ASYNC_UP_DOWN_COUNTER_FOR_HIGH_PERF_CASES:
         /**
@@ -458,7 +504,7 @@ public Object createInstrument(MetricEntity metricEntity, LongSupplier asyncCall
 
   @VisibleForTesting
   public Object createInstrument(MetricEntity metricEntity) {
-    return createInstrument(metricEntity, null, null);
+    return createInstrument(metricEntity, (LongSupplier) null, null);
   }
 
   /**

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/QuotaRequestOutcome.java

@@ -0,0 +1,24 @@
+package com.linkedin.venice.stats.dimensions;
+
+/**
+ * Outcome of a read quota enforcement decision: whether the request was allowed,
+ * rejected, or allowed without quota check (unintentionally).
+ */
+public enum QuotaRequestOutcome implements VeniceDimensionInterface {
+  /** Request passed quota check and was served normally. */
+  ALLOWED,
+
+  /** Request exceeded the store-version quota and was rejected. */
+  REJECTED,
+
+  /**
+   * Request was allowed without a quota check — either the quota enforcer had not yet
+   * initialized, or no rate limiter was allocated for the requested resource.
+   */
+  ALLOWED_UNINTENTIONALLY;
+
+  @Override
+  public VeniceMetricsDimensions getDimensionName() {
+    return VeniceMetricsDimensions.VENICE_QUOTA_REQUEST_OUTCOME;
+  }
+}

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensions.java

@@ -135,7 +135,10 @@ public enum VeniceMetricsDimensions {
   VENICE_HEARTBEAT_COMPONENT("venice.heartbeat.component"),
 
   /** {@link VeniceConsumerPoolAction} Consumer pool action (subscribe, update_assignment). */
-  VENICE_CONSUMER_POOL_ACTION("venice.consumer_pool.action");
+  VENICE_CONSUMER_POOL_ACTION("venice.consumer_pool.action"),
+
+  /** {@link QuotaRequestOutcome} Outcome of read quota enforcement (allowed, rejected, allowed_unintentionally). */
+  VENICE_QUOTA_REQUEST_OUTCOME("venice.quota.request.outcome");
 
   private final String[] dimensionName = new String[VeniceOpenTelemetryMetricNamingFormat.SIZE];
 

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/metrics/AsyncMetricEntityState.java

@@ -14,6 +14,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.DoubleSupplier;
 import java.util.function.LongSupplier;
 
 
@@ -56,13 +57,59 @@ public AsyncMetricEntityState(
       List<MeasurableStat> tehutiMetricStats,
       LongSupplier asyncCallback,
       Attributes asyncAttributes) {
+    this(
+        metricEntity,
+        otelRepository,
+        baseDimensionsMap,
+        registerTehutiSensorFn,
+        tehutiMetricNameEnum,
+        tehutiMetricStats);
+    validateAsyncCallback(asyncCallback);
+    if (emitOpenTelemetryMetrics()) {
+      setOtelMetric(otelRepository.createInstrument(this.metricEntity, asyncCallback, asyncAttributes));
+    }
+    registerTehutiSensor(registerTehutiSensorFn, tehutiMetricNameEnum, tehutiMetricStats);
+  }
+
+  /** Constructor for {@link MetricType#ASYNC_DOUBLE_GAUGE} metrics that use a {@link DoubleSupplier} callback. */
+  public AsyncMetricEntityState(
+      MetricEntity metricEntity,
+      VeniceOpenTelemetryMetricsRepository otelRepository,
+      Map<VeniceMetricsDimensions, String> baseDimensionsMap,
+      TehutiSensorRegistrationFunction registerTehutiSensorFn,
+      TehutiMetricNameEnum tehutiMetricNameEnum,
+      List<MeasurableStat> tehutiMetricStats,
+      DoubleSupplier asyncDoubleCallback,
+      Attributes asyncAttributes) {
+    this(
+        metricEntity,
+        otelRepository,
+        baseDimensionsMap,
+        registerTehutiSensorFn,
+        tehutiMetricNameEnum,
+        tehutiMetricStats);
+    validateAsyncCallback(asyncDoubleCallback != null);
+    if (emitOpenTelemetryMetrics()) {
+      setOtelMetric(otelRepository.createInstrument(this.metricEntity, asyncDoubleCallback, asyncAttributes));
+    }
+    registerTehutiSensor(registerTehutiSensorFn, tehutiMetricNameEnum, tehutiMetricStats);
+  }
+
+  /** Common field initialization for all constructors. */
+  private AsyncMetricEntityState(
+      MetricEntity metricEntity,
+      VeniceOpenTelemetryMetricsRepository otelRepository,
+      Map<VeniceMetricsDimensions, String> baseDimensionsMap,
+      TehutiSensorRegistrationFunction registerTehutiSensorFn,
+      TehutiMetricNameEnum tehutiMetricNameEnum,
+      List<MeasurableStat> tehutiMetricStats) {
     this.metricEntity = metricEntity;
     this.emitOpenTelemetryMetrics = otelRepository != null && otelRepository.emitOpenTelemetryMetrics();
     this.emitTehutiMetrics =
         shouldEmitTehutiMetrics(otelRepository, registerTehutiSensorFn, tehutiMetricNameEnum, tehutiMetricStats);
     this.otelRepository = otelRepository;
     this.baseDimensionsMap = baseDimensionsMap;
-    createMetric(tehutiMetricNameEnum, tehutiMetricStats, registerTehutiSensorFn, asyncCallback, asyncAttributes);
+    validateTehutiAsyncGaugeConsistency(tehutiMetricStats);
   }
 
   /**
@@ -96,60 +143,65 @@ public interface TehutiSensorRegistrationFunction {
   }
 
   /**
-   * Validates
-   * 1. whether an async callback is provided for an async metric
-   * 2. only when tehutiMetricStats contains AsyncGauge, the metric type should be ASYNC_GAUGE and tehutiMetricStats
-   *    should contain only one stat.
+   * Validates that async callback presence is consistent with the metric type.
+   * Accepts a {@link LongSupplier} directly for the common case.
+   */
+  private void validateAsyncCallback(LongSupplier asyncCallback) {
+    validateAsyncCallback(asyncCallback != null);
+  }
+
+  /**
+   * Validates that async callback presence is consistent with the metric type.
    *
-   * @param tehutiMetricStats the tehuti metrics stats for the given metric entity.
-   * @param asyncCallback the async callback function to be used for async metrics.
+   * @param hasAsyncCallback whether an async callback (LongSupplier or DoubleSupplier) was provided
    */
-  private void validateMetric(List<MeasurableStat> tehutiMetricStats, LongSupplier asyncCallback) {
-    if (asyncCallback != null && !metricEntity.getMetricType().isAsyncMetric()) {
+  private void validateAsyncCallback(boolean hasAsyncCallback) {
+    if (hasAsyncCallback && !metricEntity.getMetricType().isAsyncMetric()) {
       throw new IllegalArgumentException(
           "Async callback is provided, but the metric type is not async for metric: " + metricEntity.getMetricName());
-    } else if (metricEntity.getMetricType().isAsyncMetric() && asyncCallback == null
+    } else if (metricEntity.getMetricType().isAsyncMetric() && !hasAsyncCallback
         && !metricEntity.getMetricType().isObservableCounterType()) {
       // Observable counter types (ASYNC_COUNTER_FOR_HIGH_PERF_CASES and ASYNC_UP_DOWN_COUNTER_FOR_HIGH_PERF_CASES)
       // are async but handle callback registration internally via registerObservableLongCounter/UpDownCounter(),
       // so they don't need a callback passed in here.
       throw new IllegalArgumentException(
           "Async callback is not provided, but the metric type is async for metric: " + metricEntity.getMetricName());
     }
+  }
 
-    // ASYNC_GAUGE specific: If both tehuti and Otel are present, validate if all are nothing is async
+  /**
+   * Validates that Tehuti AsyncGauge stats are consistent with the OTel metric type.
+   * Called once during construction after field initialization.
+   */
+  private void validateTehutiAsyncGaugeConsistency(List<MeasurableStat> tehutiMetricStats) {
     if (tehutiMetricStats == null || tehutiMetricStats.isEmpty()) {
       return;
     }
-    // if tehutiMetricStats has AsyncGauge() then the metric type should be ASYNC_GAUGE
     if (tehutiMetricStats.stream().anyMatch(stat -> stat instanceof AsyncGauge)) {
       if (tehutiMetricStats.size() > 1) {
         throw new IllegalArgumentException(
             "Tehuti metric stats contains AsyncGauge, but it should be the only stat for metric: "
                 + metricEntity.getMetricName());
       }
-      if (metricEntity.getMetricType() != MetricType.ASYNC_GAUGE) {
+      MetricType mt = metricEntity.getMetricType();
+      if (mt != MetricType.ASYNC_GAUGE && mt != MetricType.ASYNC_DOUBLE_GAUGE) {
         throw new IllegalArgumentException(
-            "Tehuti metric stats contains AsyncGauge, but the otel metric type is not ASYNC_GAUGE for metric: "
+            "Tehuti metric stats contains AsyncGauge, but the otel metric type is not ASYNC_GAUGE/ASYNC_DOUBLE_GAUGE for metric: "
                 + metricEntity.getMetricName());
       }
-    } else if (metricEntity.getMetricType() == MetricType.ASYNC_GAUGE) {
+    } else if (metricEntity.getMetricType() == MetricType.ASYNC_GAUGE
+        || metricEntity.getMetricType() == MetricType.ASYNC_DOUBLE_GAUGE) {
       throw new IllegalArgumentException(
-          "Tehuti metric stats does not contain AsyncGauge, but the otel metric type is ASYNC_GAUGE for metric: "
+          "Tehuti metric stats does not contain AsyncGauge, but the otel metric type is ASYNC_GAUGE/ASYNC_DOUBLE_GAUGE for metric: "
               + metricEntity.getMetricName());
     }
   }
 
-  private void createMetric(
-      TehutiMetricNameEnum tehutiMetricNameEnum,
-      List<MeasurableStat> tehutiMetricStats,
+  /** Registers the Tehuti sensor if Tehuti metrics are enabled. */
+  private void registerTehutiSensor(
       TehutiSensorRegistrationFunction registerTehutiSensorFn,
-      LongSupplier asyncCallback,
-      Attributes asyncAttributes) {
-    validateMetric(tehutiMetricStats, asyncCallback);
-    if (emitOpenTelemetryMetrics()) {
-      setOtelMetric(otelRepository.createInstrument(this.metricEntity, asyncCallback, asyncAttributes));
-    }
+      TehutiMetricNameEnum tehutiMetricNameEnum,
+      List<MeasurableStat> tehutiMetricStats) {
     if (emitTehutiMetrics()) {
       setTehutiSensor(
           registerTehutiSensorFn