diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java index 4863d5b2607..8013835d5d4 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java +++ b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java @@ -94,6 +94,7 @@ import com.linkedin.venice.serialization.avro.InternalAvroSpecificSerializer; import com.linkedin.venice.serializer.RecordDeserializer; import com.linkedin.venice.stats.StatsErrorCode; +import com.linkedin.venice.stats.dimensions.VeniceGlobalRtDivErrorType; import com.linkedin.venice.stats.dimensions.VeniceIngestionFailureReason; import com.linkedin.venice.stats.dimensions.VenicePartialUpdateOperation; import com.linkedin.venice.stats.dimensions.VeniceRegionLocality; @@ -4608,6 +4609,8 @@ CompletableFuture sendGlobalRtDivMessage( true); pcs.resetConsumedBytesSinceLastGlobalRtDivSync(brokerUrl); + versionedIngestionStats + .recordGlobalRtDivSent(storeName, versionNumber, valueBytes.length, rtDivPartitionStates.size()); return vtDivSyncedFuture; } @@ -4622,6 +4625,7 @@ private byte[] createGlobalRtDivValueBytes( try { valueBytes = compressor.get().compress(valueBytes); } catch (IOException e) { + versionedIngestionStats.recordGlobalRtDivError(storeName, versionNumber, VeniceGlobalRtDivErrorType.SEND); LOGGER.error( "Failed to compress GlobalRtDivState for replica: {}. Will proceed without {} compression.", topicPartition, @@ -4715,6 +4719,7 @@ GlobalRtDivState readGlobalRtDivState( // GlobalRtDivState loading is best-effort. Any failure (e.g. storage not initialized, // compressor dictionary not yet available before SOP is consumed on secondary-fabric leaders) // should not propagate and kill ingestion. + versionedIngestionStats.recordGlobalRtDivError(storeName, versionNumber, VeniceGlobalRtDivErrorType.LOAD); LOGGER.warn( "Unable to read Global RT DIV state for topic-partition: {}, brokerUrl: {}", topicPartition, @@ -4736,6 +4741,7 @@ private GlobalRtDivState deserializeGlobalRtDivState( return globalRtDivStateSerializer .deserialize(serializedValueBytes, AvroProtocolDefinition.GLOBAL_RT_DIV_STATE.getCurrentProtocolVersion()); } catch (Exception e) { + versionedIngestionStats.recordGlobalRtDivError(storeName, versionNumber, VeniceGlobalRtDivErrorType.LOAD); // TODO: evaluate whether these logs can be set to debug LOGGER.error( "Unable to deserialize stored value bytes for key: {}, topic-partition: {}", @@ -4854,6 +4860,7 @@ void loadGlobalRtDiv(int partition, String brokerUrl) { new ChunkedValueManifestContainer()); if (globalRtDivState == null) { + versionedIngestionStats.recordGlobalRtDivLoadNotFound(storeName, versionNumber); // If the GlobalRtDivState is not present, it could be acceptable if this could be the first leader to be elected // Object not existing could be problematic if this isn't the first leader (detected via nonzero leaderPosition) PubSubPosition leaderPosition = pcs.getLeaderPosition(brokerUrl, false); @@ -4868,6 +4875,7 @@ void loadGlobalRtDiv(int partition, String brokerUrl) { } final Map producerStates = globalRtDivState.getProducerStates(); + versionedIngestionStats.recordGlobalRtDivLoaded(storeName, versionNumber, producerStates.size()); PartitionTracker.TopicType realTimeTopicType = PartitionTracker.TopicType.of(REALTIME_TOPIC_TYPE, brokerUrl); getConsumerDiv().setPartitionState(realTimeTopicType, pcs.getPartition(), producerStates); ByteBuffer checkpointBytes = globalRtDivState.getLatestPubSubPosition(); // LCRP diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java index 90d8780c4f6..530cc8ffb02 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java +++ b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java @@ -128,6 +128,7 @@ 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.VeniceGlobalRtDivErrorType; import com.linkedin.venice.stats.dimensions.VeniceIngestionFailureReason; import com.linkedin.venice.stats.dimensions.VeniceRecordType; import com.linkedin.venice.storage.protocol.ChunkedValueManifest; @@ -2327,7 +2328,16 @@ protected void updateAndSyncOffsetFromSnapshot(PartitionTracker vtDivSnapshot, P } vtDivSnapshot.updateOffsetRecord(PartitionTracker.VERSION_TOPIC, pcs.getOffsetRecord()); updateOffsetMetadataInOffsetRecord(pcs); - syncOffset(pcs); + try { + syncOffset(pcs); + } catch (Exception e) { + versionedIngestionStats.recordGlobalRtDivError(storeName, versionNumber, VeniceGlobalRtDivErrorType.VT_SYNC); + throw e; + } + versionedIngestionStats.recordGlobalRtDivVtSynced( + storeName, + versionNumber, + vtDivSnapshot.getPartitionStates(PartitionTracker.VERSION_TOPIC).size()); } /** @@ -4818,9 +4828,15 @@ protected void putInStorageEngine(int partition, byte[] keyBytes, Put put) { } protected void putGlobalRtDivStateInMetadata(int partition, byte[] keyBytes, Put put) { - storageEngine.putGlobalRtDivMetadata( - keyBytes, - ByteUtils.prependIntHeaderToByteBuffer(put.putValue, put.schemaId, false).array()); + try { + storageEngine.putGlobalRtDivMetadata( + keyBytes, + ByteUtils.prependIntHeaderToByteBuffer(put.putValue, put.schemaId, false).array()); + } catch (Exception e) { + versionedIngestionStats.recordGlobalRtDivError(storeName, versionNumber, VeniceGlobalRtDivErrorType.PERSIST); + throw e; + } + versionedIngestionStats.recordGlobalRtDivPersisted(storeName, versionNumber); } protected void removeFromStorageEngine(int partition, byte[] keyBytes, Delete delete) { @@ -5421,7 +5437,12 @@ private int processKafkaDataMessage( keyLen = keyBytes.length; if (kafkaKey.isGlobalRtDiv()) { - storageEngine.deleteGlobalRtDivMetadata(keyBytes); + try { + storageEngine.deleteGlobalRtDivMetadata(keyBytes); + } catch (Exception e) { + versionedIngestionStats.recordGlobalRtDivError(storeName, versionNumber, VeniceGlobalRtDivErrorType.DELETE); + throw e; + } } else { deleteFromStorageEngine(producedPartition, keyBytes, delete); } diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/AggVersionedIngestionStats.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/AggVersionedIngestionStats.java index d299ef58233..e2ae266a077 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/AggVersionedIngestionStats.java +++ b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/AggVersionedIngestionStats.java @@ -9,6 +9,7 @@ import com.linkedin.venice.stats.dimensions.ReplicaType; import com.linkedin.venice.stats.dimensions.VeniceDCREvent; import com.linkedin.venice.stats.dimensions.VeniceDCROperation; +import com.linkedin.venice.stats.dimensions.VeniceGlobalRtDivErrorType; import com.linkedin.venice.stats.dimensions.VeniceIngestionDestinationComponent; import com.linkedin.venice.stats.dimensions.VeniceIngestionFailureReason; import com.linkedin.venice.stats.dimensions.VeniceIngestionSourceComponent; @@ -589,4 +590,28 @@ public void recordPartialUpdateAmplificationAlertCount(String storeName, int ver public void recordActiveKeyCountInvalidation(String storeName, int version) { getIngestionOtelStats(storeName).recordActiveKeyCountInvalidation(version); } + + public void recordGlobalRtDivSent(String storeName, int version, long payloadSizeBytes, int rtProducerCount) { + getIngestionOtelStats(storeName).recordGlobalRtDivSent(version, payloadSizeBytes, rtProducerCount); + } + + public void recordGlobalRtDivPersisted(String storeName, int version) { + getIngestionOtelStats(storeName).recordGlobalRtDivPersisted(version); + } + + public void recordGlobalRtDivVtSynced(String storeName, int version, int vtProducerCount) { + getIngestionOtelStats(storeName).recordGlobalRtDivVtSynced(version, vtProducerCount); + } + + public void recordGlobalRtDivError(String storeName, int version, VeniceGlobalRtDivErrorType errorType) { + getIngestionOtelStats(storeName).recordGlobalRtDivError(version, errorType); + } + + public void recordGlobalRtDivLoaded(String storeName, int version, int rtProducerCount) { + getIngestionOtelStats(storeName).recordGlobalRtDivLoaded(version, rtProducerCount); + } + + public void recordGlobalRtDivLoadNotFound(String storeName, int version) { + getIngestionOtelStats(storeName).recordGlobalRtDivLoadNotFound(version); + } } diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntity.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntity.java index b37aa352126..20b30310a09 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntity.java +++ b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntity.java @@ -4,6 +4,8 @@ import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DCR_EVENT; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DCR_OPERATION; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DESTINATION_REGION; +import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_ERROR_TYPE; +import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_DESTINATION_COMPONENT; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_FAILURE_REASON; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_SOURCE_COMPONENT; @@ -410,6 +412,50 @@ public enum IngestionOtelMetricEntity implements ModuleMetricEntityInterface { "ingestion.key.active_count_invalidation", MetricType.COUNTER, MetricUnit.NUMBER, "Count of active key count invalidations due to underflow drift, keyExists failures, leader-propagated invalidation, or corrupt kcs signals", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE) + ), + + GLOBAL_RT_DIV_SEND_SIZE( + "ingestion.global_rt_div.send_size", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.BYTES, + "Byte size of each Global RT DIV payload produced to the version topic by the leader. The count aggregation " + + "doubles as the number of Global RT DIV messages sent", + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE) + ), + + GLOBAL_RT_DIV_SEND_RT_PRODUCER_COUNT( + "ingestion.global_rt_div.send_rt_producer_count", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.NUMBER, + "Number of RT producers tracked per broker in each Global RT DIV payload sent by the leader", + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE) + ), + + GLOBAL_RT_DIV_PERSIST_COUNT( + "ingestion.global_rt_div.persist_count", MetricType.COUNTER, MetricUnit.NUMBER, + "Count of Global RT DIV states successfully persisted to metadata storage by the drainer", + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE) + ), + + GLOBAL_RT_DIV_VT_SYNC_PRODUCER_COUNT( + "ingestion.global_rt_div.vt_sync_producer_count", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.NUMBER, + "Number of VT producers in the snapshot checkpointed into the OffsetRecord. The count aggregation doubles as " + + "the number of VT position syncs", + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE) + ), + + GLOBAL_RT_DIV_ERROR_COUNT( + "ingestion.global_rt_div.error_count", MetricType.COUNTER, MetricUnit.NUMBER, + "Count of errors in any Global RT DIV operation phase (send, persist, vt_sync, delete, load)", + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_GLOBAL_RT_DIV_ERROR_TYPE) + ), + + GLOBAL_RT_DIV_LOAD_COUNT( + "ingestion.global_rt_div.load_count", MetricType.COUNTER, MetricUnit.NUMBER, + "Count of RT DIV load attempts on F→L leader promotion, dimensioned by whether state was found on disk", + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME) + ), + + GLOBAL_RT_DIV_LOAD_RT_PRODUCER_COUNT( + "ingestion.global_rt_div.load_rt_producer_count", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.NUMBER, + "Number of RT producers restored from disk when loading Global RT DIV state on leader promotion", + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE) ); private final MetricEntity metricEntity; diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStats.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStats.java index 8796c2fd7fa..c9c25ef3f43 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStats.java +++ b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStats.java @@ -20,6 +20,13 @@ import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.DCR_TOTAL_COUNT; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.DISK_QUOTA_USED; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.DUPLICATE_KEY_UPDATE_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_ERROR_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_LOAD_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_LOAD_RT_PRODUCER_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_PERSIST_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_SEND_RT_PRODUCER_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_SEND_SIZE; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_VT_SYNC_PRODUCER_COUNT; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.INGESTION_BYTES_CONSUMED; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.INGESTION_BYTES_PRODUCED; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.INGESTION_FAILURE_COUNT; @@ -71,6 +78,8 @@ import com.linkedin.venice.stats.dimensions.VeniceDCREvent; import com.linkedin.venice.stats.dimensions.VeniceDCROperation; import com.linkedin.venice.stats.dimensions.VeniceDimensionInterface; +import com.linkedin.venice.stats.dimensions.VeniceGlobalRtDivErrorType; +import com.linkedin.venice.stats.dimensions.VeniceGlobalRtDivLoadOutcome; import com.linkedin.venice.stats.dimensions.VeniceIngestionDestinationComponent; import com.linkedin.venice.stats.dimensions.VeniceIngestionFailureReason; import com.linkedin.venice.stats.dimensions.VeniceIngestionSourceComponent; @@ -201,6 +210,15 @@ public class IngestionOtelStats { private final MetricEntityStateTwoEnums recordAssembledSizeMetric; private final MetricEntityStateOneEnum recordAssembledSizeRatioMetric; + // Global RT DIV metrics + private final MetricEntityStateOneEnum globalRtDivSendSizeMetric; + private final MetricEntityStateOneEnum globalRtDivSendRtProducerCountMetric; + private final MetricEntityStateOneEnum globalRtDivPersistCountMetric; + private final MetricEntityStateOneEnum globalRtDivVtSyncProducerCountMetric; + private final MetricEntityStateTwoEnums globalRtDivErrorCountMetric; + private final MetricEntityStateTwoEnums globalRtDivLoadCountMetric; + private final MetricEntityStateOneEnum globalRtDivLoadRtProducerCountMetric; + // Async gauge metrics private final AsyncMetricEntityStateOneEnum ingestionTaskCountByRole; private final AsyncMetricEntityStateTwoEnums activeKeyCountByRoleAndReplicaType; @@ -273,6 +291,13 @@ public class IngestionOtelStats { this.recordValueSizeMetric = null; this.recordAssembledSizeMetric = null; this.recordAssembledSizeRatioMetric = null; + this.globalRtDivSendSizeMetric = null; + this.globalRtDivSendRtProducerCountMetric = null; + this.globalRtDivPersistCountMetric = null; + this.globalRtDivVtSyncProducerCountMetric = null; + this.globalRtDivErrorCountMetric = null; + this.globalRtDivLoadCountMetric = null; + this.globalRtDivLoadRtProducerCountMetric = null; this.ingestionTaskCountByRole = null; this.activeKeyCountByRoleAndReplicaType = null; this.uniqueIngestedKeyCountByRoleAndReplicaType = null; @@ -412,6 +437,17 @@ public IngestionOtelStats( recordAssembledSizeMetric = createTwoEnumMetric(RECORD_ASSEMBLED_SIZE.getMetricEntity(), VeniceRecordType.class); recordAssembledSizeRatioMetric = createOneEnumMetric(RECORD_ASSEMBLED_SIZE_RATIO.getMetricEntity()); + // Global RT DIV metrics + globalRtDivSendSizeMetric = createOneEnumMetric(GLOBAL_RT_DIV_SEND_SIZE.getMetricEntity()); + globalRtDivSendRtProducerCountMetric = createOneEnumMetric(GLOBAL_RT_DIV_SEND_RT_PRODUCER_COUNT.getMetricEntity()); + globalRtDivPersistCountMetric = createOneEnumMetric(GLOBAL_RT_DIV_PERSIST_COUNT.getMetricEntity()); + globalRtDivVtSyncProducerCountMetric = createOneEnumMetric(GLOBAL_RT_DIV_VT_SYNC_PRODUCER_COUNT.getMetricEntity()); + globalRtDivErrorCountMetric = + createTwoEnumMetric(GLOBAL_RT_DIV_ERROR_COUNT.getMetricEntity(), VeniceGlobalRtDivErrorType.class); + globalRtDivLoadCountMetric = + createTwoEnumMetric(GLOBAL_RT_DIV_LOAD_COUNT.getMetricEntity(), VeniceGlobalRtDivLoadOutcome.class); + globalRtDivLoadRtProducerCountMetric = createOneEnumMetric(GLOBAL_RT_DIV_LOAD_RT_PRODUCER_COUNT.getMetricEntity()); + ingestionTaskCountByRole = createAsyncByRole(INGESTION_TASK_COUNT.getMetricEntity(), this::getTaskForRole, (task, role) -> 1L); @@ -830,6 +866,36 @@ public void recordPartialUpdateAmplificationAlertCount(int version, long value) partialUpdateAmplificationAlertCountMetric.record(value, classifyVersion(version, versionInfo)); } + // Global RT DIV recording methods + + public void recordGlobalRtDivSent(int version, long payloadSizeBytes, int rtProducerCount) { + VersionRole versionRole = classifyVersion(version, versionInfo); + globalRtDivSendSizeMetric.record(payloadSizeBytes, versionRole); + globalRtDivSendRtProducerCountMetric.record(rtProducerCount, versionRole); + } + + public void recordGlobalRtDivPersisted(int version) { + globalRtDivPersistCountMetric.record(1, classifyVersion(version, versionInfo)); + } + + public void recordGlobalRtDivVtSynced(int version, int vtProducerCount) { + globalRtDivVtSyncProducerCountMetric.record(vtProducerCount, classifyVersion(version, versionInfo)); + } + + public void recordGlobalRtDivError(int version, VeniceGlobalRtDivErrorType errorType) { + globalRtDivErrorCountMetric.record(1, classifyVersion(version, versionInfo), errorType); + } + + public void recordGlobalRtDivLoaded(int version, int rtProducerCount) { + VersionRole versionRole = classifyVersion(version, versionInfo); + globalRtDivLoadCountMetric.record(1, versionRole, VeniceGlobalRtDivLoadOutcome.FOUND); + globalRtDivLoadRtProducerCountMetric.record(rtProducerCount, versionRole); + } + + public void recordGlobalRtDivLoadNotFound(int version) { + globalRtDivLoadCountMetric.record(1, classifyVersion(version, versionInfo), VeniceGlobalRtDivLoadOutcome.NOT_FOUND); + } + public void recordActiveKeyCountInvalidation(int version) { if (activeKeyCountInvalidationMetric != null) { activeKeyCountInvalidationMetric.record(1, classifyVersion(version, versionInfo)); diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/NoOpIngestionOtelStats.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/NoOpIngestionOtelStats.java index 16dead1d870..38ee0a07c03 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/NoOpIngestionOtelStats.java +++ b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/NoOpIngestionOtelStats.java @@ -4,6 +4,7 @@ import com.linkedin.venice.stats.dimensions.ReplicaType; import com.linkedin.venice.stats.dimensions.VeniceDCREvent; import com.linkedin.venice.stats.dimensions.VeniceDCROperation; +import com.linkedin.venice.stats.dimensions.VeniceGlobalRtDivErrorType; import com.linkedin.venice.stats.dimensions.VeniceIngestionDestinationComponent; import com.linkedin.venice.stats.dimensions.VeniceIngestionFailureReason; import com.linkedin.venice.stats.dimensions.VeniceIngestionSourceComponent; @@ -264,4 +265,28 @@ public void recordPartialUpdateAmplificationAlertCount(int version, long value) @Override public void recordActiveKeyCountInvalidation(int version) { } + + @Override + public void recordGlobalRtDivSent(int version, long payloadSizeBytes, int rtProducerCount) { + } + + @Override + public void recordGlobalRtDivPersisted(int version) { + } + + @Override + public void recordGlobalRtDivVtSynced(int version, int vtProducerCount) { + } + + @Override + public void recordGlobalRtDivError(int version, VeniceGlobalRtDivErrorType errorType) { + } + + @Override + public void recordGlobalRtDivLoaded(int version, int rtProducerCount) { + } + + @Override + public void recordGlobalRtDivLoadNotFound(int version) { + } } diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTaskTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTaskTest.java index b310fefaf17..561f42f2723 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTaskTest.java +++ b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTaskTest.java @@ -115,6 +115,7 @@ import com.linkedin.venice.serialization.avro.ChunkedValueManifestSerializer; import com.linkedin.venice.serialization.avro.InternalAvroSpecificSerializer; import com.linkedin.venice.server.VersionRole; +import com.linkedin.venice.stats.dimensions.VeniceGlobalRtDivErrorType; import com.linkedin.venice.stats.dimensions.VeniceRecordType; import com.linkedin.venice.storage.protocol.ChunkId; import com.linkedin.venice.storage.protocol.ChunkedKeySuffix; @@ -3120,6 +3121,8 @@ public void testPutGlobalRtDivStateInMetadata() throws Exception { DelegatingStorageEngine mockStorageEngine = mock(DelegatingStorageEngine.class); injectField(ingestionTask, StoreIngestionTask.class, "storageEngine", mockStorageEngine); + AggVersionedIngestionStats mockIngestionStats = mock(AggVersionedIngestionStats.class); + injectField(ingestionTask, StoreIngestionTask.class, "versionedIngestionStats", mockIngestionStats); byte[] testValueBytes = "test-value".getBytes(StandardCharsets.UTF_8); Put put = new Put(); @@ -3145,6 +3148,7 @@ public void testPutGlobalRtDivStateInMetadata() throws Exception { Assert.assertEquals(stored.length, ValueRecord.SCHEMA_HEADER_LENGTH + testValueBytes.length); Assert.assertEquals(ByteUtils.readInt(stored, 0), GLOBAL_RT_DIV_VERSION); Assert.assertEquals(Arrays.copyOfRange(stored, ValueRecord.SCHEMA_HEADER_LENGTH, stored.length), testValueBytes); + verify(mockIngestionStats, times(1)).recordGlobalRtDivPersisted(any(), anyInt()); } /** @@ -3190,6 +3194,8 @@ public void testReadGlobalRtDivStateMetadataPath() throws Exception { injectField(ingestionTask, StoreIngestionTask.class, "storageEngine", mockStorageEngine); injectField(ingestionTask, StoreIngestionTask.class, "kafkaVersionTopic", versionTopic); injectField(ingestionTask, StoreIngestionTask.class, "compressor", Lazy.of(() -> new NoopCompressor())); + AggVersionedIngestionStats mockIngestionStats = mock(AggVersionedIngestionStats.class); + injectField(ingestionTask, StoreIngestionTask.class, "versionedIngestionStats", mockIngestionStats); // Case 1: non-chunked value present → returns deserialized state doReturn(storedNonChunked).when(mockStorageEngine) @@ -3217,6 +3223,7 @@ public void testReadGlobalRtDivStateMetadataPath() throws Exception { doThrow(new VeniceException("storage not initialized")).when(mockStorageEngine).getGlobalRtDivMetadata(any()); Assert.assertNull( ingestionTask.readGlobalRtDivState(keyBytes, GLOBAL_RT_DIV_VERSION, topicPartition, manifestContainer)); + verify(mockIngestionStats, times(1)).recordGlobalRtDivError(any(), anyInt(), eq(VeniceGlobalRtDivErrorType.LOAD)); // Case 5: compressor.get() throws IllegalStateException (dictionary not yet available on // secondary-fabric leader before SOP is consumed) → returns null without propagating diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java index d4d60f3be06..c810f44afc5 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java +++ b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java @@ -6450,11 +6450,14 @@ public void testConsumedBytesSinceLastGlobalRtDivSyncTracking() throws Exception * Verify what happens when globalRtDiv() is called and simulate loading a GlobalRtDivState object from disk. */ @Test - public void testLoadGlobalRtDiv() { + public void testLoadGlobalRtDiv() throws NoSuchFieldException, IllegalAccessException { LeaderFollowerStoreIngestionTask ingestionTask = mock(LeaderFollowerStoreIngestionTask.class); doCallRealMethod().when(ingestionTask).restoreProducerStatesForLeaderConsumption(anyInt()); doCallRealMethod().when(ingestionTask).loadGlobalRtDiv(anyInt()); doCallRealMethod().when(ingestionTask).loadGlobalRtDiv(anyInt(), anyString()); + Field versionedIngestionStatsField = StoreIngestionTask.class.getDeclaredField("versionedIngestionStats"); + versionedIngestionStatsField.setAccessible(true); + versionedIngestionStatsField.set(ingestionTask, mockVersionedStorageIngestionStats); // Drive the real non-A/A accessor so the checkpoint is stored under the NON_AA key (the key the leader-start path // reads back), not the broker URL. doCallRealMethod().when(ingestionTask).updateDivRtCheckpointPosition(any(), anyString(), any()); @@ -6484,6 +6487,8 @@ public void testLoadGlobalRtDiv() { brokerIdToUrlMap.forEach((brokerId, url) -> { verify(ingestionTask, times(1)).loadGlobalRtDiv(eq(PARTITION_FOO), eq(url)); }); + verify(mockVersionedStorageIngestionStats, times(brokerIdToUrlMap.size())) + .recordGlobalRtDivLoaded(isNull(), eq(0), eq(globalRtDivState.getProducerStates().size())); ArgumentCaptor positionCaptor = ArgumentCaptor.forClass(PubSubPosition.class); ArgumentCaptor urlCaptor = ArgumentCaptor.forClass(String.class); verify(pcs, times(brokerIdToUrlMap.size())) diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ServerMetricEntityTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ServerMetricEntityTest.java index 8fee846e59d..d12c37bc26e 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ServerMetricEntityTest.java +++ b/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(), 185, "Expected 185 unique metric entities"); + assertEquals(SERVER_METRIC_ENTITIES.size(), 192, "Expected 192 unique metric entities"); } /** diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntityTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntityTest.java index 59be77f4ddf..51125c600ea 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntityTest.java +++ b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntityTest.java @@ -4,6 +4,8 @@ import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DCR_EVENT; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DCR_OPERATION; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DESTINATION_REGION; +import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_ERROR_TYPE; +import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_DESTINATION_COMPONENT; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_FAILURE_REASON; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_SOURCE_COMPONENT; @@ -63,6 +65,10 @@ private static Map expectedD setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_DCR_OPERATION); Set storeClusterVersionFailureReason = setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_INGESTION_FAILURE_REASON); + Set storeClusterVersionGlobalRtDivError = + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_GLOBAL_RT_DIV_ERROR_TYPE); + Set storeClusterVersionGlobalRtDivLoadOutcome = + setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME); // ASYNC_GAUGE metrics map.put( @@ -549,6 +555,64 @@ private static Map expectedD "Count of active key count invalidations due to underflow drift, keyExists failures, leader-propagated invalidation, or corrupt kcs signals", storeClusterVersion)); + // Global RT DIV metrics + map.put( + IngestionOtelMetricEntity.GLOBAL_RT_DIV_SEND_SIZE, + new MetricEntityExpectation( + "ingestion.global_rt_div.send_size", + MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, + MetricUnit.BYTES, + "Byte size of each Global RT DIV payload produced to the version topic by the leader. The count aggregation doubles as the number of Global RT DIV messages sent", + storeClusterVersion)); + map.put( + IngestionOtelMetricEntity.GLOBAL_RT_DIV_SEND_RT_PRODUCER_COUNT, + new MetricEntityExpectation( + "ingestion.global_rt_div.send_rt_producer_count", + MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, + MetricUnit.NUMBER, + "Number of RT producers tracked per broker in each Global RT DIV payload sent by the leader", + storeClusterVersion)); + map.put( + IngestionOtelMetricEntity.GLOBAL_RT_DIV_PERSIST_COUNT, + new MetricEntityExpectation( + "ingestion.global_rt_div.persist_count", + MetricType.COUNTER, + MetricUnit.NUMBER, + "Count of Global RT DIV states successfully persisted to metadata storage by the drainer", + storeClusterVersion)); + map.put( + IngestionOtelMetricEntity.GLOBAL_RT_DIV_VT_SYNC_PRODUCER_COUNT, + new MetricEntityExpectation( + "ingestion.global_rt_div.vt_sync_producer_count", + MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, + MetricUnit.NUMBER, + "Number of VT producers in the snapshot checkpointed into the OffsetRecord. The count aggregation doubles as the number of VT position syncs", + storeClusterVersion)); + map.put( + IngestionOtelMetricEntity.GLOBAL_RT_DIV_ERROR_COUNT, + new MetricEntityExpectation( + "ingestion.global_rt_div.error_count", + MetricType.COUNTER, + MetricUnit.NUMBER, + "Count of errors in any Global RT DIV operation phase (send, persist, vt_sync, delete, load)", + storeClusterVersionGlobalRtDivError)); + map.put( + IngestionOtelMetricEntity.GLOBAL_RT_DIV_LOAD_COUNT, + new MetricEntityExpectation( + "ingestion.global_rt_div.load_count", + MetricType.COUNTER, + MetricUnit.NUMBER, + "Count of RT DIV load attempts on F→L leader promotion, dimensioned by whether state was found on disk", + storeClusterVersionGlobalRtDivLoadOutcome)); + map.put( + IngestionOtelMetricEntity.GLOBAL_RT_DIV_LOAD_RT_PRODUCER_COUNT, + new MetricEntityExpectation( + "ingestion.global_rt_div.load_rt_producer_count", + MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, + MetricUnit.NUMBER, + "Number of RT producers restored from disk when loading Global RT DIV state on leader promotion", + storeClusterVersion)); + return map; } } diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStatsTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStatsTest.java index f64ed2cd520..1530ee6eedf 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStatsTest.java +++ b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStatsTest.java @@ -18,6 +18,13 @@ import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.DCR_MERGE_TIME; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.DCR_TOTAL_COUNT; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.DUPLICATE_KEY_UPDATE_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_ERROR_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_LOAD_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_LOAD_RT_PRODUCER_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_PERSIST_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_SEND_RT_PRODUCER_COUNT; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_SEND_SIZE; +import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.GLOBAL_RT_DIV_VT_SYNC_PRODUCER_COUNT; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.INGESTION_BYTES_CONSUMED; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.INGESTION_BYTES_PRODUCED; import static com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity.INGESTION_FAILURE_COUNT; @@ -56,6 +63,8 @@ import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DCR_EVENT; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DCR_OPERATION; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DESTINATION_REGION; +import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_ERROR_TYPE; +import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_DESTINATION_COMPONENT; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_FAILURE_REASON; import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_INGESTION_SOURCE_COMPONENT; @@ -88,6 +97,8 @@ import com.linkedin.venice.stats.dimensions.VeniceDCREvent; import com.linkedin.venice.stats.dimensions.VeniceDCROperation; import com.linkedin.venice.stats.dimensions.VeniceDimensionInterface; +import com.linkedin.venice.stats.dimensions.VeniceGlobalRtDivErrorType; +import com.linkedin.venice.stats.dimensions.VeniceGlobalRtDivLoadOutcome; import com.linkedin.venice.stats.dimensions.VeniceIngestionDestinationComponent; import com.linkedin.venice.stats.dimensions.VeniceIngestionFailureReason; import com.linkedin.venice.stats.dimensions.VeniceIngestionSourceComponent; @@ -515,6 +526,123 @@ public void testRecordBatchProcessingRequestErrorCount() { TEST_PREFIX); } + // Global RT DIV metrics + + @Test + public void testRecordGlobalRtDivSent() { + ingestionOtelStats.updateVersionInfo(CURRENT_VERSION, FUTURE_VERSION); + ingestionOtelStats.recordGlobalRtDivSent(CURRENT_VERSION, 1024L, 3); + + // The count aggregation of send_size doubles as the number of Global RT DIV messages sent + validateHistogramPointData( + inMemoryMetricReader, + 1024.0, + 1024.0, + 1, + 1024.0, + buildAttributesWithVersionRole(VersionRole.CURRENT), + GLOBAL_RT_DIV_SEND_SIZE.getMetricEntity().getMetricName(), + TEST_PREFIX); + + validateHistogramPointData( + inMemoryMetricReader, + 3.0, + 3.0, + 1, + 3.0, + buildAttributesWithVersionRole(VersionRole.CURRENT), + GLOBAL_RT_DIV_SEND_RT_PRODUCER_COUNT.getMetricEntity().getMetricName(), + TEST_PREFIX); + } + + @Test + public void testRecordGlobalRtDivPersisted() { + ingestionOtelStats.updateVersionInfo(CURRENT_VERSION, FUTURE_VERSION); + ingestionOtelStats.recordGlobalRtDivPersisted(CURRENT_VERSION); + validateLongPointDataFromCounter( + inMemoryMetricReader, + 1, + buildAttributesWithVersionRole(VersionRole.CURRENT), + GLOBAL_RT_DIV_PERSIST_COUNT.getMetricEntity().getMetricName(), + TEST_PREFIX); + } + + @Test + public void testRecordGlobalRtDivVtSynced() { + ingestionOtelStats.updateVersionInfo(CURRENT_VERSION, FUTURE_VERSION); + ingestionOtelStats.recordGlobalRtDivVtSynced(CURRENT_VERSION, 7); + // The count aggregation of vt_sync_producer_count doubles as the number of VT position syncs + validateHistogramPointData( + inMemoryMetricReader, + 7.0, + 7.0, + 1, + 7.0, + buildAttributesWithVersionRole(VersionRole.CURRENT), + GLOBAL_RT_DIV_VT_SYNC_PRODUCER_COUNT.getMetricEntity().getMetricName(), + TEST_PREFIX); + } + + @DataProvider(name = "globalRtDivErrorTypeProvider") + public Object[][] globalRtDivErrorTypeProvider() { + return new Object[][] { { VeniceGlobalRtDivErrorType.SEND }, { VeniceGlobalRtDivErrorType.PERSIST }, + { VeniceGlobalRtDivErrorType.VT_SYNC }, { VeniceGlobalRtDivErrorType.DELETE }, + { VeniceGlobalRtDivErrorType.LOAD } }; + } + + @Test(dataProvider = "globalRtDivErrorTypeProvider") + public void testAllGlobalRtDivErrorTypes(VeniceGlobalRtDivErrorType errorType) { + ingestionOtelStats.updateVersionInfo(CURRENT_VERSION, FUTURE_VERSION); + ingestionOtelStats.recordGlobalRtDivError(CURRENT_VERSION, errorType); + validateLongPointDataFromCounter( + inMemoryMetricReader, + 1, + buildAttributesWithVersionRoleAndSecondEnum(VersionRole.CURRENT, VENICE_GLOBAL_RT_DIV_ERROR_TYPE, errorType), + GLOBAL_RT_DIV_ERROR_COUNT.getMetricEntity().getMetricName(), + TEST_PREFIX); + } + + @Test + public void testRecordGlobalRtDivLoaded() { + ingestionOtelStats.updateVersionInfo(CURRENT_VERSION, FUTURE_VERSION); + ingestionOtelStats.recordGlobalRtDivLoaded(CURRENT_VERSION, 3); + + validateLongPointDataFromCounter( + inMemoryMetricReader, + 1, + buildAttributesWithVersionRoleAndSecondEnum( + VersionRole.CURRENT, + VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME, + VeniceGlobalRtDivLoadOutcome.FOUND), + GLOBAL_RT_DIV_LOAD_COUNT.getMetricEntity().getMetricName(), + TEST_PREFIX); + + validateHistogramPointData( + inMemoryMetricReader, + 3.0, + 3.0, + 1, + 3.0, + buildAttributesWithVersionRole(VersionRole.CURRENT), + GLOBAL_RT_DIV_LOAD_RT_PRODUCER_COUNT.getMetricEntity().getMetricName(), + TEST_PREFIX); + } + + @Test + public void testRecordGlobalRtDivLoadNotFound() { + ingestionOtelStats.updateVersionInfo(CURRENT_VERSION, FUTURE_VERSION); + ingestionOtelStats.recordGlobalRtDivLoadNotFound(CURRENT_VERSION); + validateLongPointDataFromCounter( + inMemoryMetricReader, + 1, + buildAttributesWithVersionRoleAndSecondEnum( + VersionRole.CURRENT, + VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME, + VeniceGlobalRtDivLoadOutcome.NOT_FOUND), + GLOBAL_RT_DIV_LOAD_COUNT.getMetricEntity().getMetricName(), + TEST_PREFIX); + } + @Test public void testRecordBatchProcessingRequestTime() { ingestionOtelStats.updateVersionInfo(CURRENT_VERSION, FUTURE_VERSION); diff --git a/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivErrorType.java b/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivErrorType.java new file mode 100644 index 00000000000..33c486bda30 --- /dev/null +++ b/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivErrorType.java @@ -0,0 +1,36 @@ +package com.linkedin.venice.stats.dimensions; + +import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_ERROR_TYPE; + + +/** + * Dimension values for the {@link VeniceMetricsDimensions#VENICE_GLOBAL_RT_DIV_ERROR_TYPE} dimension, + * representing the phase of a Global RT DIV operation that failed. + * + *

The Global RT DIV feature propagates RT DIV state from the leader to followers. Several of its + * phases are best-effort: errors are caught and logged rather than propagated. This dimension + * categorizes which phase failed, enabling targeted alerting. + * + * @see com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity#GLOBAL_RT_DIV_ERROR_COUNT + */ +public enum VeniceGlobalRtDivErrorType implements VeniceDimensionInterface { + /** Failed to serialize or compress the RT DIV snapshot before producing it to the version topic */ + SEND, + /** Failed to write the received RT DIV state to metadata storage */ + PERSIST, + /** Failed to sync the latest consumed VT position into the OffsetRecord */ + VT_SYNC, + /** Failed to delete RT DIV metadata during chunk cleanup */ + DELETE, + /** Failed to read or deserialize persisted RT DIV state on F→L leader promotion */ + LOAD; + + /** + * All instances of this enum share the same dimension name. + * Refer to {@link VeniceDimensionInterface#getDimensionName()} for more details. + */ + @Override + public VeniceMetricsDimensions getDimensionName() { + return VENICE_GLOBAL_RT_DIV_ERROR_TYPE; + } +} diff --git a/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivLoadOutcome.java b/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivLoadOutcome.java new file mode 100644 index 00000000000..d5463fe5bd6 --- /dev/null +++ b/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivLoadOutcome.java @@ -0,0 +1,30 @@ +package com.linkedin.venice.stats.dimensions; + +import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME; + + +/** + * Dimension values for the {@link VeniceMetricsDimensions#VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME} dimension, + * representing the outcome of loading Global RT DIV state during F→L leader promotion. + * + *

FOUND means state was on disk and the leader resumes from a saved checkpoint. NOT_FOUND means + * nothing usable was read, which is expected for the first leader ever elected for the partition but + * also covers a failed read; {@code error_count} with type {@code LOAD} distinguishes the latter. + * + * @see com.linkedin.davinci.stats.ingestion.IngestionOtelMetricEntity#GLOBAL_RT_DIV_LOAD_COUNT + */ +public enum VeniceGlobalRtDivLoadOutcome implements VeniceDimensionInterface { + /** State found on disk; leader resumes from saved checkpoint position */ + FOUND, + /** No usable state read from disk; leader starts from EARLIEST */ + NOT_FOUND; + + /** + * All instances of this enum share the same dimension name. + * Refer to {@link VeniceDimensionInterface#getDimensionName()} for more details. + */ + @Override + public VeniceMetricsDimensions getDimensionName() { + return VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME; + } +} diff --git a/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensions.java b/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensions.java index 7de901d655e..71433d2851c 100644 --- a/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensions.java +++ b/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensions.java @@ -122,6 +122,12 @@ public enum VeniceMetricsDimensions { /** {@link VeniceIngestionFailureReason} Categorized reason for ingestion failure */ VENICE_INGESTION_FAILURE_REASON("venice.ingestion.failure.reason"), + /** {@link VeniceGlobalRtDivErrorType} Error type for Global RT DIV best-effort operations */ + VENICE_GLOBAL_RT_DIV_ERROR_TYPE("venice.global_rt_div.error.type"), + + /** {@link VeniceGlobalRtDivLoadOutcome} Outcome of Global RT DIV state load on F→L transition */ + VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME("venice.global_rt_div.load.outcome"), + /** {@link VeniceChunkingStatus} */ VENICE_CHUNKING_STATUS("venice.chunking.status"), diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivErrorTypeTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivErrorTypeTest.java new file mode 100644 index 00000000000..bd020099916 --- /dev/null +++ b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivErrorTypeTest.java @@ -0,0 +1,24 @@ +package com.linkedin.venice.stats.dimensions; + +import com.linkedin.venice.utils.CollectionUtils; +import java.util.Map; +import org.testng.annotations.Test; + + +public class VeniceGlobalRtDivErrorTypeTest { + @Test + public void testDimensionInterface() { + Map expectedValues = + CollectionUtils.mapBuilder() + .put(VeniceGlobalRtDivErrorType.SEND, "send") + .put(VeniceGlobalRtDivErrorType.PERSIST, "persist") + .put(VeniceGlobalRtDivErrorType.VT_SYNC, "vt_sync") + .put(VeniceGlobalRtDivErrorType.DELETE, "delete") + .put(VeniceGlobalRtDivErrorType.LOAD, "load") + .build(); + new VeniceDimensionTestFixture<>( + VeniceGlobalRtDivErrorType.class, + VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_ERROR_TYPE, + expectedValues).assertAll(); + } +} diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivLoadOutcomeTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivLoadOutcomeTest.java new file mode 100644 index 00000000000..95ac85a5a46 --- /dev/null +++ b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceGlobalRtDivLoadOutcomeTest.java @@ -0,0 +1,21 @@ +package com.linkedin.venice.stats.dimensions; + +import com.linkedin.venice.utils.CollectionUtils; +import java.util.Map; +import org.testng.annotations.Test; + + +public class VeniceGlobalRtDivLoadOutcomeTest { + @Test + public void testDimensionInterface() { + Map expectedValues = + CollectionUtils.mapBuilder() + .put(VeniceGlobalRtDivLoadOutcome.FOUND, "found") + .put(VeniceGlobalRtDivLoadOutcome.NOT_FOUND, "not_found") + .build(); + new VeniceDimensionTestFixture<>( + VeniceGlobalRtDivLoadOutcome.class, + VeniceMetricsDimensions.VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME, + expectedValues).assertAll(); + } +} diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java index b60bdbe704f..b3dde532007 100644 --- a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java +++ b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java @@ -190,6 +190,12 @@ public void testGetDimensionNameInSnakeCase() { case VENICE_STORE_WRITE_TYPE: assertEquals(dimension.getDimensionName(format), "venice.store.write_type"); break; + case VENICE_GLOBAL_RT_DIV_ERROR_TYPE: + assertEquals(dimension.getDimensionName(format), "venice.global_rt_div.error.type"); + break; + case VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME: + assertEquals(dimension.getDimensionName(format), "venice.global_rt_div.load.outcome"); + break; default: throw new IllegalArgumentException("Unknown dimension: " + dimension); } @@ -378,6 +384,12 @@ public void testGetDimensionNameInCamelCase() { case VENICE_STORE_WRITE_TYPE: assertEquals(dimension.getDimensionName(format), "venice.store.writeType"); break; + case VENICE_GLOBAL_RT_DIV_ERROR_TYPE: + assertEquals(dimension.getDimensionName(format), "venice.globalRtDiv.error.type"); + break; + case VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME: + assertEquals(dimension.getDimensionName(format), "venice.globalRtDiv.load.outcome"); + break; default: throw new IllegalArgumentException("Unknown dimension: " + dimension); } @@ -566,6 +578,12 @@ public void testGetDimensionNameInPascalCase() { case VENICE_STORE_WRITE_TYPE: assertEquals(dimension.getDimensionName(format), "Venice.Store.WriteType"); break; + case VENICE_GLOBAL_RT_DIV_ERROR_TYPE: + assertEquals(dimension.getDimensionName(format), "Venice.GlobalRtDiv.Error.Type"); + break; + case VENICE_GLOBAL_RT_DIV_LOAD_OUTCOME: + assertEquals(dimension.getDimensionName(format), "Venice.GlobalRtDiv.Load.Outcome"); + break; default: throw new IllegalArgumentException("Unknown dimension: " + dimension); }