[dvc][fc][tc] Lazily register client-side metrics

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

@@ -446,12 +446,6 @@ public StoreIngestionTask(
         getRecordTransformer != null ? getRecordTransformer.apply(store.getCurrentVersion()) : null;
     this.recordTransformer =
         clientRecordTransformer != null ? new BlockingDaVinciRecordTransformer(clientRecordTransformer) : null;
-    if (this.recordTransformer != null) {
-      versionedIngestionStats.registerTransformerLatencySensor(storeName, versionNumber);
-      versionedIngestionStats.registerTransformerLifecycleStartLatency(storeName, versionNumber);
-      versionedIngestionStats.registerTransformerLifecycleEndLatency(storeName, versionNumber);
-      versionedIngestionStats.registerTransformerErrorSensor(storeName, versionNumber);
-    }
 
     this.localKafkaServer = this.kafkaProps.getProperty(KAFKA_BOOTSTRAP_SERVERS);
     this.localKafkaServerSingletonSet = Collections.singleton(localKafkaServer);

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

@@ -235,24 +235,4 @@ public void recordTransformerError(String storeName, int version, double value,
   public void recordMaxIdleTime(String storeName, int version, long idleTimeMs) {
     getStats(storeName, version).recordIdleTime(idleTimeMs);
   }
-
-  public void registerTransformerLatencySensor(String storeName, int version) {
-    getStats(storeName, version).registerTransformerLatencySensor();
-    getTotalStats(storeName).registerTransformerLatencySensor();
-  }
-
-  public void registerTransformerLifecycleStartLatency(String storeName, int version) {
-    getStats(storeName, version).registerTransformerLifecycleStartLatencySensor();
-    getTotalStats(storeName).registerTransformerLifecycleStartLatencySensor();
-  }
-
-  public void registerTransformerLifecycleEndLatency(String storeName, int version) {
-    getStats(storeName, version).registerTransformerLifecycleEndLatencySensor();
-    getTotalStats(storeName).registerTransformerLifecycleEndLatencySensor();
-  }
-
-  public void registerTransformerErrorSensor(String storeName, int version) {
-    getStats(storeName, version).registerTransformerErrorSensor();
-    getTotalStats(storeName).registerTransformerErrorSensor();
-  }
 }

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

@@ -7,6 +7,7 @@
 import com.linkedin.davinci.kafka.consumer.StoreIngestionTask;
 import com.linkedin.venice.stats.LongAdderRateGauge;
 import com.linkedin.venice.utils.RegionUtils;
+import com.linkedin.venice.utils.lazy.Lazy;
 import io.tehuti.metrics.MetricConfig;
 import io.tehuti.metrics.MetricsRepository;
 import io.tehuti.metrics.Sensor;
@@ -93,9 +94,9 @@ public class IngestionStats {
   private final WritePathLatencySensor consumedRecordEndToEndProcessingLatencySensor;
   private final WritePathLatencySensor nearlineProducerToLocalBrokerLatencySensor;
   private final WritePathLatencySensor nearlineLocalBrokerToReadyToServeLatencySensor;
-  private WritePathLatencySensor transformerLatencySensor;
-  private WritePathLatencySensor transformerLifecycleStartLatencySensor;
-  private WritePathLatencySensor transformerLifecycleEndLatencySensor;
+  private final WritePathLatencySensor transformerLatencySensor;
+  private final WritePathLatencySensor transformerLifecycleStartLatencySensor;
+  private final WritePathLatencySensor transformerLifecycleEndLatencySensor;
   private final WritePathLatencySensor producerCallBackLatency;
   private final WritePathLatencySensor leaderPreprocessingLatency;
   private final WritePathLatencySensor internalPreprocessingLatency;
@@ -109,14 +110,14 @@ public class IngestionStats {
 
   /** Record a version-level offset rewind events for VTs across all stores. */
   private final Count versionTopicEndOffsetRewindCount = new Count();
-  private final Sensor versionTopicEndOffsetRewindSensor;
+  private final Lazy<Sensor> versionTopicEndOffsetRewindSensor;
   private final MetricsRepository localMetricRepository;
 
   // Measure the max idle time among partitions for a given the store on this host
   private final LongAdderRateGauge idleTimeSensor = new LongAdderRateGauge();
 
-  private Count transformerErrorCount = new Count();
-  private Sensor transformerErrorSensor;
+  private final Count transformerErrorCount = new Count();
+  private final Lazy<Sensor> transformerErrorSensor;
 
   public IngestionStats(VeniceServerConfig serverConfig) {
 
@@ -171,8 +172,11 @@ public IngestionStats(VeniceServerConfig serverConfig) {
     registerSensor(localMetricRepository, LEADER_RECORDS_PRODUCED_METRIC_NAME, leaderRecordsProducedSensor);
     registerSensor(localMetricRepository, LEADER_BYTES_PRODUCED_METRIC_NAME, leaderBytesProducedSensor);
 
-    versionTopicEndOffsetRewindSensor = localMetricRepository.sensor(VERSION_TOPIC_END_OFFSET_REWIND_COUNT);
-    versionTopicEndOffsetRewindSensor.add(VERSION_TOPIC_END_OFFSET_REWIND_COUNT, versionTopicEndOffsetRewindCount);
+    versionTopicEndOffsetRewindSensor = Lazy.of(() -> {
+      Sensor versionTopicEndOffsetRewindSensor = localMetricRepository.sensor(VERSION_TOPIC_END_OFFSET_REWIND_COUNT);
+      versionTopicEndOffsetRewindSensor.add(VERSION_TOPIC_END_OFFSET_REWIND_COUNT, versionTopicEndOffsetRewindCount);
+      return versionTopicEndOffsetRewindSensor;
+    });
 
     producerSourceBrokerLatencySensor =
         new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, "producer_to_source_broker_latency");
@@ -207,6 +211,17 @@ public IngestionStats(VeniceServerConfig serverConfig) {
     registerSensor(localMetricRepository, OFFSET_REGRESSION_DCR_ERROR, offsetRegressionDCRErrorSensor);
     registerSensor(localMetricRepository, TOMBSTONE_CREATION_DCR, tombstoneCreationDCRSensor);
     registerSensor(localMetricRepository, IDLE_TIME, idleTimeSensor);
+
+    transformerLatencySensor = new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, TRANSFORMER_LATENCY);
+    transformerLifecycleStartLatencySensor =
+        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, TRANSFORMER_LIFECYCLE_START_LATENCY);
+    transformerLifecycleEndLatencySensor =
+        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, TRANSFORMER_LIFECYCLE_END_LATENCY);
+    transformerErrorSensor = Lazy.of(() -> {
+      Sensor transformerErrorSensor = localMetricRepository.sensor(TRANSFORMER_ERROR_COUNT);
+      transformerErrorSensor.add(TRANSFORMER_ERROR_COUNT, transformerErrorCount);
+      return transformerErrorSensor;
+    });
   }
 
   private void registerSensor(MetricsRepository localMetricRepository, String sensorName, LongAdderRateGauge gauge) {
@@ -376,7 +391,7 @@ public void recordInternalPreprocessingLatency(double value, long currentTimeMs)
   }
 
   public void recordVersionTopicEndOffsetRewind() {
-    versionTopicEndOffsetRewindSensor.record();
+    versionTopicEndOffsetRewindSensor.get().record();
   }
 
   public double getVersionTopicEndOffsetRewindCount() {
@@ -568,55 +583,25 @@ public void recordNearlineLocalBrokerToReadyToServeLatency(double value, long cu
   }
 
   public void recordTransformerError(double value, long currentTimeMs) {
-    transformerErrorSensor.record(value, currentTimeMs);
-  }
-
-  public void registerTransformerErrorSensor() {
-    if (transformerErrorSensor == null) {
-      transformerErrorSensor = localMetricRepository.sensor(TRANSFORMER_ERROR_COUNT);
-      transformerErrorSensor.add(TRANSFORMER_ERROR_COUNT, transformerErrorCount);
-    }
+    transformerErrorSensor.get().record(value, currentTimeMs);
   }
 
   public double getTransformerErrorCount() {
-    if (transformerErrorCount != null) {
-      return transformerErrorCount.measure(METRIC_CONFIG, System.currentTimeMillis());
-    }
-    return 0;
+    return transformerErrorCount.measure(METRIC_CONFIG, System.currentTimeMillis());
   }
 
   public void recordTransformerLatency(double value, long currentTimeMs) {
     transformerLatencySensor.record(value, currentTimeMs);
   }
 
-  public void registerTransformerLatencySensor() {
-    if (transformerLatencySensor == null) {
-      transformerLatencySensor = new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, TRANSFORMER_LATENCY);
-    }
-  }
-
   public void recordTransformerLifecycleStartLatency(double value, long currentTimeMs) {
     transformerLifecycleStartLatencySensor.record(value, currentTimeMs);
   }
 
-  public void registerTransformerLifecycleStartLatencySensor() {
-    if (transformerLifecycleStartLatencySensor == null) {
-      transformerLifecycleStartLatencySensor =
-          new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, TRANSFORMER_LIFECYCLE_START_LATENCY);
-    }
-  }
-
   public void recordTransformerLifecycleEndLatency(double value, long currentTimeMs) {
     transformerLifecycleEndLatencySensor.record(value, currentTimeMs);
   }
 
-  public void registerTransformerLifecycleEndLatencySensor() {
-    if (transformerLifecycleEndLatencySensor == null) {
-      transformerLifecycleEndLatencySensor =
-          new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, TRANSFORMER_LIFECYCLE_END_LATENCY);
-    }
-  }
-
   public void recordIdleTime(long value) {
     idleTimeSensor.record(value);
   }

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

@@ -1,6 +1,7 @@
 package com.linkedin.davinci.stats;
 
 import com.linkedin.venice.stats.AbstractVeniceStats;
+import com.linkedin.venice.utils.lazy.Lazy;
 import io.tehuti.metrics.MetricsRepository;
 import io.tehuti.metrics.Sensor;
 import io.tehuti.metrics.stats.Gauge;
@@ -10,17 +11,17 @@
 public class IsolatedIngestionProcessHeartbeatStats extends AbstractVeniceStats {
   private static final String METRICS_PREFIX = "ingestion_isolation_heartbeat";
   // Delay in millis since last successful heartbeat query.
-  private final Sensor heartbeatAgeSensor;
+  private final Lazy<Sensor> heartbeatAgeSensor;
   private final Sensor forkedProcessRestartSensor;
 
   public IsolatedIngestionProcessHeartbeatStats(MetricsRepository metricsRepository) {
     super(metricsRepository, METRICS_PREFIX);
-    heartbeatAgeSensor = registerSensor("heartbeat_age", new Gauge());
+    heartbeatAgeSensor = registerLazySensor("heartbeat_age", new Gauge());
     forkedProcessRestartSensor = registerSensor("forked_process_restart", new OccurrenceRate());
   }
 
   public void recordHeartbeatAge(long heartbeatAgeInMs) {
-    heartbeatAgeSensor.record(heartbeatAgeInMs);
+    heartbeatAgeSensor.get().record(heartbeatAgeInMs);
   }
 
   public void recordForkedProcessRestart() {

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

@@ -3,6 +3,7 @@
 import com.linkedin.venice.stats.AbstractVeniceStats;
 import com.linkedin.venice.stats.LongAdderRateGauge;
 import com.linkedin.venice.utils.Time;
+import com.linkedin.venice.utils.lazy.Lazy;
 import io.tehuti.metrics.MetricsRepository;
 import io.tehuti.metrics.Sensor;
 import io.tehuti.metrics.stats.AsyncGauge;
@@ -24,25 +25,25 @@
 
 public class KafkaConsumerServiceStats extends AbstractVeniceStats {
   private final LongAdderRateGauge pollRequestSensor;
-  private final Sensor pollRequestLatencySensor;
+  private final Lazy<Sensor> pollRequestLatencySensor;
   private final Sensor pollResultNumSensor;
   private final LongAdderRateGauge pollNonZeroResultNumSensor;
 
   private final Sensor pollRequestError;
-  private final Sensor consumerRecordsProducingToWriterBufferLatencySensor;
-  private final Sensor detectedDeletedTopicNumSensor;
-  private final Sensor detectedNoRunningIngestionTopicPartitionNumSensor;
-  private final Sensor delegateSubscribeLatencySensor;
-  private final Sensor updateCurrentAssignmentLatencySensor;
-  private final Sensor maxPartitionsPerConsumer;
-  private final Sensor minPartitionsPerConsumer;
-  private final Sensor avgPartitionsPerConsumer;
+  private final Lazy<Sensor> consumerRecordsProducingToWriterBufferLatencySensor;
+  private final Lazy<Sensor> detectedDeletedTopicNumSensor;
+  private final Lazy<Sensor> detectedNoRunningIngestionTopicPartitionNumSensor;
+  private final Lazy<Sensor> delegateSubscribeLatencySensor;
+  private final Lazy<Sensor> updateCurrentAssignmentLatencySensor;
+  private final Lazy<Sensor> maxPartitionsPerConsumer;
+  private final Lazy<Sensor> minPartitionsPerConsumer;
+  private final Lazy<Sensor> avgPartitionsPerConsumer;
   private final Sensor getOffsetLagIsAbsentSensor;
   private final Sensor getOffsetLagIsPresentSensor;
   private final Sensor getLatestOffsetIsAbsentSensor;
   private final Sensor getLatestOffsetIsPresentSensor;
   private final Sensor byteSizeSensor;
-  private final Sensor idleTimeSensor;
+  private final Lazy<Sensor> idleTimeSensor;
 
   public KafkaConsumerServiceStats(
       MetricsRepository metricsRepository,
@@ -71,12 +72,12 @@ public KafkaConsumerServiceStats(
      */
 
     // the consumer idle time
-    idleTimeSensor = registerSensor("idle_time", new Max());
+    idleTimeSensor = registerLazySensor("idle_time", new Max());
     // the number of poll requests
     pollRequestSensor =
         registerOnlyTotalRate("consumer_poll_request", totalStats, () -> totalStats.pollRequestSensor, time);
     // Notice that "pollRequestLatencySensor" only reports correct data when consumer task threads are not stuck
-    pollRequestLatencySensor = registerSensor("consumer_poll_request_latency", new Avg(), new Max());
+    pollRequestLatencySensor = registerLazySensor("consumer_poll_request_latency", new Avg(), new Max());
     pollNonZeroResultNumSensor = registerOnlyTotalRate(
         "consumer_poll_non_zero_result_num",
         totalStats,
@@ -96,16 +97,17 @@ public KafkaConsumerServiceStats(
     pollRequestError = registerSensor("consumer_poll_error", new OccurrenceRate());
     // To measure 'put' latency of consumer records blocking queue
     consumerRecordsProducingToWriterBufferLatencySensor =
-        registerSensor("consumer_records_producing_to_write_buffer_latency", new Avg(), new Max());
-    detectedDeletedTopicNumSensor = registerSensor("detected_deleted_topic_num", new Total());
+        registerLazySensor("consumer_records_producing_to_write_buffer_latency", new Avg(), new Max());
+    detectedDeletedTopicNumSensor = registerLazySensor("detected_deleted_topic_num", new Total());
     detectedNoRunningIngestionTopicPartitionNumSensor =
-        registerSensor("detected_no_running_ingestion_topic_partition_num", new Total());
-    delegateSubscribeLatencySensor = registerSensor("delegate_subscribe_latency", new Avg(), new Max());
-    updateCurrentAssignmentLatencySensor = registerSensor("update_current_assignment_latency", new Avg(), new Max());
+        registerLazySensor("detected_no_running_ingestion_topic_partition_num", new Total());
+    delegateSubscribeLatencySensor = registerLazySensor("delegate_subscribe_latency", new Avg(), new Max());
+    updateCurrentAssignmentLatencySensor =
+        registerLazySensor("update_current_assignment_latency", new Avg(), new Max());
 
-    minPartitionsPerConsumer = registerSensor("min_partitions_per_consumer", new Gauge());
-    maxPartitionsPerConsumer = registerSensor("max_partitions_per_consumer", new Gauge());
-    avgPartitionsPerConsumer = registerSensor("avg_partitions_per_consumer", new Gauge());
+    minPartitionsPerConsumer = registerLazySensor("min_partitions_per_consumer", new Gauge());
+    maxPartitionsPerConsumer = registerLazySensor("max_partitions_per_consumer", new Gauge());
+    avgPartitionsPerConsumer = registerLazySensor("avg_partitions_per_consumer", new Gauge());
 
     Sensor getOffsetLagSensor = registerSensor("getOffsetLag", new OccurrenceRate());
     Sensor[] offsetLagParent = new Sensor[] { getOffsetLagSensor };
@@ -122,7 +124,7 @@ public KafkaConsumerServiceStats(
 
   public void recordPollRequestLatency(double latency) {
     pollRequestSensor.record();
-    pollRequestLatencySensor.record(latency);
+    pollRequestLatencySensor.get().record(latency);
   }
 
   public void recordPollResultNum(int count) {
@@ -134,39 +136,39 @@ public void recordNonZeroPollResultNum(int count) {
   }
 
   public void recordConsumerRecordsProducingToWriterBufferLatency(double latency) {
-    consumerRecordsProducingToWriterBufferLatencySensor.record(latency);
+    consumerRecordsProducingToWriterBufferLatencySensor.get().record(latency);
   }
 
   public void recordPollError() {
     pollRequestError.record();
   }
 
   public void recordDetectedDeletedTopicNum(int count) {
-    detectedDeletedTopicNumSensor.record(count);
+    detectedDeletedTopicNumSensor.get().record(count);
   }
 
   public void recordDetectedNoRunningIngestionTopicPartitionNum(int count) {
-    detectedNoRunningIngestionTopicPartitionNumSensor.record(count);
+    detectedNoRunningIngestionTopicPartitionNumSensor.get().record(count);
   }
 
   public void recordDelegateSubscribeLatency(double value) {
-    delegateSubscribeLatencySensor.record(value);
+    delegateSubscribeLatencySensor.get().record(value);
   }
 
   public void recordUpdateCurrentAssignmentLatency(double value) {
-    updateCurrentAssignmentLatencySensor.record(value);
+    updateCurrentAssignmentLatencySensor.get().record(value);
   }
 
   public void recordMinPartitionsPerConsumer(int count) {
-    minPartitionsPerConsumer.record(count);
+    minPartitionsPerConsumer.get().record(count);
   }
 
   public void recordMaxPartitionsPerConsumer(int count) {
-    maxPartitionsPerConsumer.record(count);
+    maxPartitionsPerConsumer.get().record(count);
   }
 
   public void recordAvgPartitionsPerConsumer(int count) {
-    avgPartitionsPerConsumer.record(count);
+    avgPartitionsPerConsumer.get().record(count);
   }
 
   public void recordOffsetLagIsAbsent() {
@@ -190,6 +192,6 @@ public void recordByteSizePerPoll(double count) {
   }
 
   public void recordConsumerIdleTime(double time) {
-    idleTimeSensor.record(time);
+    idleTimeSensor.get().record(time);
   }
 }

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

@@ -2,6 +2,7 @@
 
 import com.linkedin.davinci.ingestion.main.MainIngestionStorageMetadataService;
 import com.linkedin.venice.stats.AbstractVeniceStats;
+import com.linkedin.venice.utils.lazy.Lazy;
 import io.tehuti.metrics.MetricsRepository;
 import io.tehuti.metrics.Sensor;
 import io.tehuti.metrics.stats.Gauge;
@@ -14,23 +15,23 @@ public class MetadataUpdateStats extends AbstractVeniceStats {
   private static final String METRICS_PREFIX = "ingestion_isolation_metadata_updates";
 
   // Number of remaining elements inside metadata update queue.
-  private final Sensor metadataUpdateQueueLengthSensor;
+  private final Lazy<Sensor> metadataUpdateQueueLengthSensor;
   // If we encountered unknown exception during metadata update, we will set the Gauge value to 1
-  private final Sensor metadataUpdateQueueErrorSensor;
+  private final Lazy<Sensor> metadataUpdateQueueErrorSensor;
 
   public MetadataUpdateStats(MetricsRepository metricsRepository) {
     super(metricsRepository, METRICS_PREFIX);
-    metadataUpdateQueueLengthSensor = registerSensor("queue_length", new Gauge());
-    metadataUpdateQueueErrorSensor = registerSensor("queue_update_error", new Gauge());
+    metadataUpdateQueueLengthSensor = registerLazySensor("queue_length", new Gauge());
+    metadataUpdateQueueErrorSensor = registerLazySensor("queue_update_error", new Gauge());
     // Reset metadata update queue error Gauge.
     recordMetadataQueueUpdateError(0.0);
   }
 
   public void recordMetadataUpdateQueueLength(int queueLength) {
-    metadataUpdateQueueLengthSensor.record(queueLength);
+    metadataUpdateQueueLengthSensor.get().record(queueLength);
   }
 
   public void recordMetadataQueueUpdateError(double value) {
-    metadataUpdateQueueErrorSensor.record(value);
+    metadataUpdateQueueErrorSensor.get().record(value);
   }
 }