Package com.linkedin.davinci.stats.ingestion.heartbeat

Class HeartbeatMonitoringService

java.lang.Object
com.linkedin.venice.service.AbstractVeniceService
com.linkedin.davinci.stats.ingestion.heartbeat.HeartbeatMonitoringService
All Implemented Interfaces:
Closeable, AutoCloseable
public class HeartbeatMonitoringService extends AbstractVeniceService
This service monitors heartbeats. Heartbeats are only monitored if lagMonitors are added for leader or follower partitions. Once a lagMonitor is added, the service will being emitting a metric which grows linearly with time, only resetting to the timestamp of the last reported heartbeat for a given partition. Heartbeats are only monitored for stores which have a hybrid config. All other registrations for lag monitoring are ignored. Max and Average are reported per version of resource across partitions. If a heartbeat is invoked for a partition that we're NOT monitoring lag for, it is ignored. This class will monitor lag for a partition as a leader or follower, but never both. Whether we're reporting leader or follower depends on which monitor was set last. Lag will stop being reported for partitions which have the monitor removed. Each region gets a different lag monitor

  • Field Details

    • DEFAULT_REPORTER_THREAD_SLEEP_INTERVAL_SECONDS

      public static final int DEFAULT_REPORTER_THREAD_SLEEP_INTERVAL_SECONDS
      See Also:

    • DEFAULT_LAG_LOGGING_THREAD_SLEEP_INTERVAL_SECONDS

      public static final int DEFAULT_LAG_LOGGING_THREAD_SLEEP_INTERVAL_SECONDS
      See Also:

    • DEFAULT_STALE_HEARTBEAT_LOG_THRESHOLD_MILLIS

      public static final long DEFAULT_STALE_HEARTBEAT_LOG_THRESHOLD_MILLIS

    • INVALID_MESSAGE_TIMESTAMP

      public static final long INVALID_MESSAGE_TIMESTAMP
      See Also:

    • INVALID_HEARTBEAT_LAG

      public static final long INVALID_HEARTBEAT_LAG
      See Also:

    • DEFAULT_LAG_MONITOR_CLEANUP_CYCLE

      public static final int DEFAULT_LAG_MONITOR_CLEANUP_CYCLE
      See Also:

  • Constructor Details

  • Method Details

    • addFollowerLagMonitor

      public void addFollowerLagMonitor(Version version, int partition)
      Adds monitoring for a follower partition of a given version. This request is ignored if the version isn't hybrid.
      Parameters:
      version - the version to monitor lag for
      partition - the partition to monitor lag for

    • addLeaderLagMonitor

      public void addLeaderLagMonitor(Version version, int partition)
      Adds monitoring for a leader partition of a given version. This request is ignored if the version isn't hybrid.
      Parameters:
      version - the version to monitor lag for
      partition - the partition to monitor lag for

    • removeLagMonitor

      public void removeLagMonitor(Version version, int partition)
      Removes monitoring for a partition of a given version.
      Parameters:
      version - the version to remove monitoring for
      partition - the partition to remove monitoring for

    • getHeartbeatInfo

      public Map<String,ReplicaHeartbeatInfo> getHeartbeatInfo(String versionTopicName, int partitionFilter, boolean filterLagReplica)

    • startInner

      public boolean startInner() throws Exception
      Specified by:
      startInner in class AbstractVeniceService
      Returns:
      true if the service is completely started, false if it is still starting asynchronously (in this case, it is the implementer's responsibility to set AbstractVeniceService.serviceState to AbstractVeniceService.ServiceState.STARTED upon completion of the async work).
      Throws:
      Exception

    • stopInner

      public void stopInner() throws Exception
      Specified by:
      stopInner in class AbstractVeniceService
      Throws:
      Exception

    • updateLagMonitor

      public void updateLagMonitor(String resourceName, int partitionId, HeartbeatLagMonitorAction heartbeatLagMonitorAction)
      Update lag monitor for a given resource replica based on different heartbeat lag monitor action.

    • getReplicaLeaderMaxHeartbeatLag

      public long getReplicaLeaderMaxHeartbeatLag(PartitionConsumptionState partitionConsumptionState, boolean shouldLogLag)
      Get maximum heartbeat lag from all regions (except separate RT regions) for a given LEADER replica.
      Returns:
      Max leader heartbeat lag, or INVALID_HEARTBEAT_LAG if any region's heartbeat is unknown.

    • getReplicaLeaderMinHeartbeatTimestamp

      public long getReplicaLeaderMinHeartbeatTimestamp(PartitionConsumptionState partitionConsumptionState, boolean shouldLogLag)
      Get minimum heartbeat timestamp from all regions (except separate RT regions) for a given LEADER replica.
      Returns:
      Min leader heartbeat timestamp, or INVALID_MESSAGE_TIMESTAMP if any region's heartbeat is unknown.

    • getReplicaFollowerHeartbeatLag

      public long getReplicaFollowerHeartbeatLag(PartitionConsumptionState partitionConsumptionState, boolean shouldLogLag)
      Get heartbeat lag from local region for a given FOLLOWER replica.
      Returns:
      Follower heartbeat lag, or INVALID_HEARTBEAT_LAG if local region's heartbeat is unknown.

    • getReplicaFollowerHeartbeatTimestamp

      public long getReplicaFollowerHeartbeatTimestamp(PartitionConsumptionState partitionConsumptionState, boolean shouldLogLag)
      Get heartbeat timestamp from local region for a given FOLLOWER replica.
      Returns:
      Follower heartbeat timestamp, or INVALID_MESSAGE_TIMESTAMP if local region's heartbeat is unknown.

    • recordLeaderRecordTimestamp

      public void recordLeaderRecordTimestamp(HeartbeatKey key, Long timestamp, boolean isReadyToServe)
      Record a leader record-level timestamp using a pre-built HeartbeatKey. Avoids HeartbeatKey allocation and hash computation on the per-record hot path.

    • recordFollowerRecordTimestamp

      public void recordFollowerRecordTimestamp(HeartbeatKey key, Long timestamp, boolean isReadyToServe)
      Record a follower record-level timestamp using a pre-built HeartbeatKey. Avoids HeartbeatKey allocation and hash computation on the per-record hot path.

    • recordLeaderHeartbeat

      public void recordLeaderHeartbeat(HeartbeatKey key, Long timestamp, boolean isReadyToServe)
      Record a leader heartbeat timestamp using a pre-built HeartbeatKey. Avoids HeartbeatKey allocation and hash computation on the per-record hot path.

    • recordFollowerHeartbeat

      public void recordFollowerHeartbeat(HeartbeatKey key, Long timestamp, boolean isReadyToServe)
      Record a follower heartbeat timestamp using a pre-built HeartbeatKey. Avoids HeartbeatKey allocation and hash computation on the per-record hot path.

    • getLeaderHeartbeatTimeStamps

      protected Map<HeartbeatKey,IngestionTimestampEntry> getLeaderHeartbeatTimeStamps()

    • getFollowerHeartbeatTimeStamps

      protected Map<HeartbeatKey,IngestionTimestampEntry> getFollowerHeartbeatTimeStamps()

    • recordLags

      protected void recordLags(Map<HeartbeatKey,IngestionTimestampEntry> heartbeatTimestamps, com.linkedin.davinci.stats.ingestion.heartbeat.HeartbeatMonitoringService.ReportLagFunction lagFunction)

    • recordRecordLags

      protected void recordRecordLags(Map<HeartbeatKey,IngestionTimestampEntry> heartbeatTimestamps, com.linkedin.davinci.stats.ingestion.heartbeat.HeartbeatMonitoringService.ReportLagFunction lagFunction)

    • record

      protected void record()

    • checkAndMaybeLogHeartbeatDelay

      protected void checkAndMaybeLogHeartbeatDelay()

    • getMaxLeaderHeartbeatLag

      public AggregatedHeartbeatLagEntry getMaxLeaderHeartbeatLag()

    • getMaxFollowerHeartbeatLag

      public AggregatedHeartbeatLagEntry getMaxFollowerHeartbeatLag()

    • setKafkaStoreIngestionService

      public void setKafkaStoreIngestionService(KafkaStoreIngestionService kafkaStoreIngestionService)