[da-vinci][client-common][test] Add per-metric attribution + per-replica-type

ingestion gauges

OTel ingestion gauges:
- New entities: ingestion.task.count, ingestion.key.active_count,
  ingestion.key.unique_ingested_count, dimensioned by version-role and
  (where applicable) replica-type. Wired through multi-emit gauges via
  createAsyncByRole / createAsyncByRoleAndReplicaType helpers.
- IngestionOtelStats.resolveByRole consolidates version-role lookup;
  setIngestionTaskPushTimeoutGauge / recordIdleTime gate writes on task
  registration via computeIfPresent.
- HostLevelIngestionStats uses the new no-arg aggregators on
  StoreIngestionTask to avoid duplicating partition iteration.

StoreIngestionTask aggregation:
- getActiveKeyCount(ReplicaType) and getEstimatedUniqueIngestedKeyCount
  (ReplicaType) backed by getKeyCountByReplicaType + matchesReplicaType
  (LEADER<->LEADER, FOLLOWER<->STANDBY; transitional states excluded from
  per-replica aggregates).
- Consolidated active-key-count invalidation into a
recordActiveKeyCountInvalidation helper.

recordFailureMetric per-metric attribution:
- New VENICE_METRIC_NAME dimension. recordFailureMetric switched to
  MetricEntityStateGeneric and tags every failure with venice.metric.name.
- New MetricEntity.createWithCustomPrefix factory (dimensions + custom
  prefix); prefix validation centralized in a private helper. Removed
  unused 5-arg createWithNoDimensions overload.
- Defensive re-entry guard in incrementFailureCounter to short-circuit
  recursion if recording the failure metric itself ever throws.
- Extracted INTERNAL_METRIC_PREFIX constant.

Tests:
- New StoreIngestionTaskAggregationTest covers per-replica-type aggregators
  (LEADER-only, FOLLOWER-only, IN_TRANSITION/PAUSE_TRANSITION exclusion,
  untracked, no-arg total, LEADER+FOLLOWER<total invariant).
- VeniceOpenTelemetryMetricsRepositoryTest: per-metric attribution,
  accumulation, null-message exception tolerance, both Exception/String
  overloads accumulating into the same data point.
- MetricEntityTest: createWithCustomPrefix happy path + validation
  branches (empty prefix, "venice." prefix, empty dimensions, RATIO +
  long-only metric type rejection).
- VeniceMetricsDimensionsTest: VENICE_METRIC_NAME in snake_case, camelCase,
  and PascalCase switches.

Author: m-nagarajan

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/ActiveActiveStoreIngestionTask.java

@@ -939,8 +939,7 @@ private PubSubMessageHeaders computeActiveKeyCountSignal(
       }
       // Underflow detected (count was 0 but got a delete) — count drifted, now invalidated to
       // ACTIVE_KEY_COUNT_NOT_TRACKED.
-      aggVersionedIngestionStats.recordActiveKeyCountInvalidation(storeName, versionNumber);
-      getHostLevelIngestionStats().recordActiveKeyCountInvalidation();
+      recordActiveKeyCountInvalidation();
       return new PubSubMessageHeaders().add(KEY_COUNT_INVALIDATE_SIGNAL);
     }
     return EmptyPubSubMessageHeaders.SINGLETON;
@@ -1056,8 +1055,7 @@ private boolean isValuePresentForKey(
       // A transient RocksDB I/O failure must not halt ingestion. Invalidate the count so we stop
       // publishing a wrong number, and return false (assume key absent) to skip the signal.
       partitionConsumptionState.setActiveKeyCount(ACTIVE_KEY_COUNT_NOT_TRACKED);
-      aggVersionedIngestionStats.recordActiveKeyCountInvalidation(storeName, versionNumber);
-      getHostLevelIngestionStats().recordActiveKeyCountInvalidation();
+      recordActiveKeyCountInvalidation();
       String msg =
           "keyExists failed for replica " + partitionConsumptionState.getReplicaId() + "; invalidating activeKeyCount.";
       if (!REDUNDANT_LOGGING_FILTER.isRedundantException(msg)) {

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java

@@ -124,6 +124,7 @@
 import com.linkedin.venice.serializer.FastSerializerDeserializerFactory;
 import com.linkedin.venice.serializer.RecordDeserializer;
 import com.linkedin.venice.server.VersionRole;
+import com.linkedin.venice.stats.dimensions.ReplicaType;
 import com.linkedin.venice.stats.dimensions.VeniceIngestionFailureReason;
 import com.linkedin.venice.stats.dimensions.VeniceRecordType;
 import com.linkedin.venice.storage.protocol.ChunkedValueManifest;
@@ -186,8 +187,10 @@
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.LongPredicate;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
+import java.util.function.ToLongFunction;
 import javax.annotation.Nonnull;
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.Schema;
@@ -4696,8 +4699,8 @@ private void trackActiveKeyCount(
 
     // Follower RT: apply "kcs" signal from leader
     if (activeKeyCountForHybridStoreEnabled && isActiveActiveReplicationEnabled
-        && partitionConsumptionState.getActiveKeyCount() >= 0 && partitionConsumptionState.isEndOfPushReceived()
-        && leaderProducedRecordContext == null
+        && partitionConsumptionState.getActiveKeyCount() != ACTIVE_KEY_COUNT_NOT_TRACKED
+        && partitionConsumptionState.isEndOfPushReceived() && leaderProducedRecordContext == null
         && (messageType == MessageType.PUT ? !isChunkFragment(writerSchemaId) : messageType == MessageType.DELETE)) {
       PubSubMessageHeader signalHeader = consumerRecord.getPubSubMessageHeaders().get(KEY_COUNT_SIGNAL_HEADER);
       if (signalHeader != null && signalHeader.value() != null && signalHeader.value().length == 1) {
@@ -4706,8 +4709,7 @@ private void trackActiveKeyCount(
           partitionConsumptionState.incrementActiveKeyCount();
         } else if (signal == ActiveActiveStoreIngestionTask.KEY_DELETED_SIGNAL_VALUE) {
           if (!partitionConsumptionState.decrementActiveKeyCount()) {
-            versionedIngestionStats.recordActiveKeyCountInvalidation(storeName, versionNumber);
-            hostLevelIngestionStats.recordActiveKeyCountInvalidation();
+            recordActiveKeyCountInvalidation();
           }
         } else if (signal == ActiveActiveStoreIngestionTask.KEY_COUNT_INVALIDATE_SIGNAL_VALUE) {
           invalidateActiveKeyCount(partitionConsumptionState);
@@ -4735,6 +4737,11 @@ private void trackActiveKeyCount(
   /** Invalidates the active key count and records invalidation metrics (both OTel and Tehuti). */
   private void invalidateActiveKeyCount(PartitionConsumptionState partitionConsumptionState) {
     partitionConsumptionState.setActiveKeyCount(ACTIVE_KEY_COUNT_NOT_TRACKED);
+    recordActiveKeyCountInvalidation();
+  }
+
+  /** Records active-key-count invalidation on both the OTel (per-version) and Tehuti (host-level) paths. */
+  protected final void recordActiveKeyCountInvalidation() {
     versionedIngestionStats.recordActiveKeyCountInvalidation(storeName, versionNumber);
     hostLevelIngestionStats.recordActiveKeyCountInvalidation();
   }
@@ -5784,18 +5791,78 @@ public long getEstimatedUniqueIngestedKeyCount() {
   }
 
   /**
-   * Returns the estimated count of unique keys ever put or deleted across partitions on this host.
-   * If stateFilter is provided, only partitions matching that state are summed.
-   * If stateFilter is null, all partitions are summed.
+   * HLL estimate of unique keys ever put or deleted, filtered by {@code replicaType}
+   * ({@code null} = all partitions). HLL has no "untracked" sentinel — every matching partition
+   * contributes (zero if empty), and the no-match fallback is {@code 0}.
+   */
+  public long getEstimatedUniqueIngestedKeyCount(ReplicaType replicaType) {
+    return getKeyCountByReplicaType(
+        replicaType,
+        PartitionConsumptionState::getEstimatedUniqueIngestedKeyCount,
+        v -> true,
+        0L);
+  }
+
+  /** Sums {@link #getActiveKeyCount(ReplicaType)} across all partitions regardless of replica type. */
+  public long getActiveKeyCount() {
+    return getActiveKeyCount(null);
+  }
+
+  /**
+   * Exact count of currently active (alive) keys, filtered by {@code replicaType} ({@code null} =
+   * all partitions). Non-monotonic. Returns {@link OffsetRecord#ACTIVE_KEY_COUNT_NOT_TRACKED} when
+   * no matching partition has tracking active (no batch baseline); 0 means tracked but empty
+   * (e.g., empty push). The {@link OffsetRecord#ACTIVE_KEY_COUNT_NOT_TRACKED} vs 0 distinction is
+   * intentional — unlike HLL which has no "untracked" state, the active count uses
+   * {@link OffsetRecord#ACTIVE_KEY_COUNT_NOT_TRACKED} to signal that tracking was never initialized.
+   */
+  public long getActiveKeyCount(ReplicaType replicaType) {
+    return getKeyCountByReplicaType(
+        replicaType,
+        PartitionConsumptionState::getActiveKeyCount,
+        v -> v != ACTIVE_KEY_COUNT_NOT_TRACKED,
+        ACTIVE_KEY_COUNT_NOT_TRACKED);
+  }
+
+  /**
+   * Iterates partitions filtered by {@code replicaType} (null = all), reads a per-partition
+   * count via {@code keyCountFn}, and sums values that pass {@code valueInclusionCheck}. Returns
+   * {@code emptyReturnValue} when no value passed (used as a "not tracked" sentinel by callers
+   * that distinguish "no contributions" from "zero").
    */
-  public long getEstimatedUniqueIngestedKeyCount(LeaderFollowerStateType stateFilter) {
+  private long getKeyCountByReplicaType(
+      ReplicaType replicaType,
+      ToLongFunction<PartitionConsumptionState> keyCountFn,
+      LongPredicate valueInclusionCheck,
+      long emptyReturnValue) {
     long total = 0;
-    for (PartitionConsumptionState pcs: partitionConsumptionStateMap.values()) {
-      if (stateFilter == null || pcs.getLeaderFollowerState() == stateFilter) {
-        total += pcs.getEstimatedUniqueIngestedKeyCount();
+    boolean hasValidKeyCount = false;
+    for (PartitionConsumptionState pcs: getPartitionConsumptionStates()) {
+      if (replicaType != null && !matchesReplicaType(pcs, replicaType)) {
+        continue;
       }
+      long value = keyCountFn.applyAsLong(pcs);
+      if (valueInclusionCheck.test(value)) {
+        total += value;
+        hasValidKeyCount = true;
+      }
+    }
+    return hasValidKeyCount ? total : emptyReturnValue;
+  }
+
+  /**
+   * LEADER matches {@link LeaderFollowerStateType#LEADER}; FOLLOWER matches
+   * {@link LeaderFollowerStateType#STANDBY}; anything else returns {@code false}.
+   */
+  private static boolean matchesReplicaType(PartitionConsumptionState pcs, ReplicaType replicaType) {
+    switch (replicaType) {
+      case LEADER:
+        return pcs.getLeaderFollowerState() == LeaderFollowerStateType.LEADER;
+      case FOLLOWER:
+        return pcs.getLeaderFollowerState() == LeaderFollowerStateType.STANDBY;
+      default:
+        return false;
     }
-    return total;
   }
 
   @VisibleForTesting

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

@@ -5,6 +5,7 @@
 import static com.linkedin.davinci.stats.IngestionStats.BATCH_PROCESSING_REQUEST_LATENCY;
 import static com.linkedin.davinci.stats.IngestionStats.BATCH_PROCESSING_REQUEST_RECORDS;
 import static com.linkedin.davinci.stats.IngestionStats.BATCH_PROCESSING_REQUEST_SIZE;
+import static com.linkedin.venice.offsets.OffsetRecord.ACTIVE_KEY_COUNT_NOT_TRACKED;
 
 import com.linkedin.davinci.config.VeniceServerConfig;
 import com.linkedin.davinci.kafka.consumer.PartitionConsumptionState;
@@ -310,25 +311,25 @@ public HostLevelIngestionStats(
       registerSensor(new AsyncGauge((ignored, ignored2) -> ingestionTaskMap.size(), "ingestion_task_count"));
     }
 
-    // Active key count gauge. -1 = not tracked, 0 = tracked but empty.
+    // Active key count gauge. ACTIVE_KEY_COUNT_NOT_TRACKED = not tracked, 0 = tracked but empty.
     // Cannot use measurable() because its 0-fallback conflates "untracked" with "empty".
     if (isTotalStats) {
       registerSensor(new AsyncGauge((ignored, ignored2) -> {
         long total = 0;
         boolean anyTracked = false;
         for (StoreIngestionTask task: ingestionTaskMap.values()) {
-          long storeCount = sumActiveKeyCount(task);
-          if (storeCount >= 0) {
+          long storeCount = task.getActiveKeyCount();
+          if (storeCount != ACTIVE_KEY_COUNT_NOT_TRACKED) {
             anyTracked = true;
             total += storeCount;
           }
         }
-        return anyTracked ? total : -1;
+        return anyTracked ? total : ACTIVE_KEY_COUNT_NOT_TRACKED;
       }, "active_key_count"));
     } else {
       registerSensor(new AsyncGauge((ignored, ignored2) -> {
         StoreIngestionTask sit = ingestionTaskMap.get(storeName);
-        return sit == null ? -1 : sumActiveKeyCount(sit);
+        return sit == null ? ACTIVE_KEY_COUNT_NOT_TRACKED : sit.getActiveKeyCount();
       }, "active_key_count"));
     }
 
@@ -582,19 +583,6 @@ private Measurable measurable(
     };
   }
 
-  private static long sumActiveKeyCount(StoreIngestionTask task) {
-    long sum = 0;
-    boolean anyTracked = false;
-    for (PartitionConsumptionState pcs: task.getPartitionConsumptionStates()) {
-      long count = pcs.getActiveKeyCount();
-      if (count >= 0) {
-        anyTracked = true;
-        sum += count;
-      }
-    }
-    return anyTracked ? sum : -1;
-  }
-
   /** Record a host-level byte consumption rate across all store versions */
   public void recordTotalBytesConsumed(long bytes) {
     totalBytesConsumedRate.record(bytes);

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

@@ -16,6 +16,7 @@
 import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_VERSION_ROLE;
 import static com.linkedin.venice.utils.Utils.setOf;
 
+import com.linkedin.venice.offsets.OffsetRecord;
 import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
 import com.linkedin.venice.stats.metrics.MetricEntity;
 import com.linkedin.venice.stats.metrics.MetricType;
@@ -323,21 +324,43 @@ public enum IngestionOtelMetricEntity implements ModuleMetricEntityInterface {
       setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE)
   ),
 
+  /**
+   * HLL estimate of unique keys ever put or deleted, per replica type, for a store version on this
+   * host. Monotonically increasing within a version; resets on new version push.
+   *
+   * <p>LEADER aggregates partitions in LEADER state; FOLLOWER aggregates partitions in STANDBY
+   * state. Partitions in any IN_TRANSITION_* state contribute to neither aggregate. No data point
+   * is emitted when no task exists for the role.
+   */
   UNIQUE_INGESTED_KEY_COUNT(
       "ingestion.key.unique_ingested_count", MetricType.ASYNC_GAUGE, MetricUnit.NUMBER,
-      "Estimated unique keys ever put or deleted per replica type for a store version on this host (HLL-based, monotonically increasing, resets on new version push)",
+      "Estimated unique keys ever put or deleted per replica type for a store version on this host "
+          + "(HLL-based, monotonically increasing, resets on new version push), for leaders and followers.",
       setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_REPLICA_TYPE)
   ),
 
   INGESTION_TASK_COUNT(
       "ingestion.task.count", MetricType.ASYNC_GAUGE, MetricUnit.NUMBER,
-      "Whether an active ingestion task exists for this store version (0 or 1)",
+      "Emits 1 when an active ingestion task exists for this store version and role; no data point is emitted otherwise.",
       setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE)
   ),
 
+  /**
+   * Point-in-time count of unique active keys across partitions of this store version on this
+   * host. Non-monotonic (tracks creates and deletes).
+   *
+   * <p>LEADER aggregates partitions in LEADER state; FOLLOWER aggregates partitions in STANDBY
+   * state. Partitions in any IN_TRANSITION_* state contribute to neither aggregate.
+   *
+   * <p>Sentinels: {@link OffsetRecord#ACTIVE_KEY_COUNT_NOT_TRACKED} = no partition of this replica
+   * type has tracking active; 0 = tracked but empty (e.g., empty push). No data point is emitted
+   * when no task exists for the role.
+   */
   ACTIVE_KEY_COUNT(
       "ingestion.key.active_count", MetricType.ASYNC_GAUGE, MetricUnit.NUMBER,
-      "Point-in-time count of unique active keys across partitions of this store version on this host. Non-monotonic (tracks creates and deletes). -1 = not tracked, 0 = tracked but empty",
+      "Point-in-time count of unique active keys across partitions of this store version on this host. "
+          + "Non-monotonic (tracks creates and deletes). -1 = not tracked, 0 = tracked but empty, "
+          + "for leaders and followers.",
       setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_REPLICA_TYPE)
   ),