PHOENIX-7992 Add replay-side performance metrics for HA replication

phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java

@@ -145,8 +145,9 @@ public abstract class ReplicationLogDiscovery {
   @GuardedBy("this")
   protected long lastAlignedTargetMillis = Long.MIN_VALUE;
   /**
-   * One-shot guard so a misconfigured {@link #REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY} logs its
-   * fall-back WARN once rather than every scheduling cycle (the epsilon is read live per cycle).
+   * One-shot guard so a misconfigured {@link #REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY} logs
+   * its fall-back WARN once rather than every scheduling cycle (the epsilon is read live per
+   * cycle).
    */
   private final AtomicBoolean warnedInvalidEpsilon = new AtomicBoolean(false);
 
@@ -279,11 +280,11 @@ protected void scheduleNextReplay() {
    * Runs one replay pass and, unless the service has been stopped, schedules the next aligned pass.
    * A recoverable {@link Exception} from {@link #replay()} is logged and the chain continues, so a
    * single failed round does not break it. A fatal {@link Error} (OOM, stack overflow, linkage) is
-   * logged, tears the chain down (marking the service not-running so the supervisor can rebuild it),
-   * and is rethrown -- never rescheduled onto a potentially corrupted JVM.
-   * The reschedule is guarded by the same lock stop() uses; if stop() shut the scheduler down
-   * first, {@link #isRunning} is false and we do not reschedule (and a concurrent shutdown that
-   * rejects the submission is caught and treated as "stop the chain").
+   * logged, tears the chain down (marking the service not-running so the supervisor can rebuild
+   * it), and is rethrown -- never rescheduled onto a potentially corrupted JVM. The reschedule is
+   * guarded by the same lock stop() uses; if stop() shut the scheduler down first,
+   * {@link #isRunning} is false and we do not reschedule (and a concurrent shutdown that rejects
+   * the submission is caught and treated as "stop the chain").
    * @param owner the scheduler this cycle was launched on. If a stop()->start() restart has since
    *              swapped in a new scheduler, {@code owner} no longer equals {@link #scheduler} and
    *              this stale cycle must not reschedule onto the new generation (which would create a
@@ -438,13 +439,16 @@ protected void processNewFilesForRound(ReplicationRound replicationRound) throws
     List<Path> files = replicationLogTracker.getNewFilesForRound(replicationRound);
     LOG.info("Number of new files for round {} is {}", replicationRound, files.size());
     while (!files.isEmpty() && isRunning()) {
-      processOneRandomFile(files);
+      processOneRandomFile(files, true);
       files = replicationLogTracker.getNewFilesForRound(replicationRound);
     }
     long duration = EnvironmentEdgeManager.currentTime() - startTime;
     LOG.info("Finished new files processing for round: {} in {}ms for haGroup: {}",
       replicationRound, duration, haGroupName);
     getMetrics().updateTimeToProcessNewFiles(duration);
+    if (duration > roundTimeMills) {
+      getMetrics().incrementRoundsExceedingRoundTime();
+    }
   }
 
   /**
@@ -468,7 +472,7 @@ protected void processInProgressDirectory() throws IOException {
       replicationLogTracker.getInProgressLogSubDirectoryName(), renameTimestampThreshold,
       files.size(), haGroupName);
     while (!files.isEmpty() && isRunning()) {
-      Optional<Path> failedFile = processOneRandomFile(files);
+      Optional<Path> failedFile = processOneRandomFile(files, false);
       if (failedFile.isPresent()) {
         String prefix = replicationLogTracker.getFilePrefix(failedFile.get());
         int count = failureCount.merge(prefix, 1, Integer::sum);
@@ -494,17 +498,23 @@ protected void processInProgressDirectory() throws IOException {
   /**
    * Processes a single random file from the provided list. Marks the file as in-progress, processes
    * it, and marks it as completed or failed.
-   * @param files - List of files from which to select and process one randomly
+   * @param files      - List of files from which to select and process one randomly
+   * @param firstClaim - true when {@code files} are new files being claimed for the first time
+   *                   (from {@link #processNewFilesForRound}); false when they are already in the
+   *                   in-progress directory and are being reprocessed (from
+   *                   {@link #processInProgressDirectory}). Forwarded to
+   *                   {@link #processFile(Path, boolean)}.
    * @return the original path of the file that failed, or empty if processing succeeded
    */
-  private Optional<Path> processOneRandomFile(final List<Path> files) throws IOException {
+  private Optional<Path> processOneRandomFile(final List<Path> files, final boolean firstClaim)
+    throws IOException {
     // Pick a random file and process it
     Path file = files.get(ThreadLocalRandom.current().nextInt(files.size()));
     Optional<Path> optionalInProgressFilePath = Optional.empty();
     try {
       optionalInProgressFilePath = replicationLogTracker.markInProgress(file);
       if (optionalInProgressFilePath.isPresent()) {
-        processFile(optionalInProgressFilePath.get());
+        processFile(optionalInProgressFilePath.get(), firstClaim);
         replicationLogTracker.markCompleted(optionalInProgressFilePath.get());
       }
     } catch (IOException exception) {
@@ -518,10 +528,15 @@ private Optional<Path> processOneRandomFile(final List<Path> files) throws IOExc
 
   /**
    * Handles the processing of a single file.
-   * @param path - The file to be processed
+   * @param path       - The file to be processed
+   * @param firstClaim - true when this invocation is the file's first claim (it was just moved into
+   *                   the in-progress directory from the new-files path); false when reprocessing a
+   *                   file that was already in the in-progress directory. Lets subclasses record
+   *                   first-claim-only signals (e.g. pickup lag) without double counting on
+   *                   retries.
    * @throws IOException if there's an error during file processing
    */
-  protected abstract void processFile(Path path) throws IOException;
+  protected abstract void processFile(Path path, boolean firstClaim) throws IOException;
 
   /** Creates a new metrics source for monitoring operations. */
   protected abstract MetricsReplicationLogDiscovery createMetricsSource();
@@ -696,11 +711,12 @@ public int getInProgressFileMinAgeSeconds() {
   /**
    * Returns the epsilon margin (milliseconds) added to the aligned scheduler wake instant. Guards
    * against a misconfigured value: a non-numeric string, a negative value (which would move wakes
-   * <em>before</em> eligibility and reintroduce missed rounds), or a value {@code >= roundTimeMills}
-   * (which wraps through {@link Math#floorMod} and no longer represents the documented "epsilon
-   * after the boundary") all fall back to {@link #DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS} with a
-   * one-shot WARN. Riding over the misconfiguration keeps replay polling on a safe default rather
-   * than wedging the group, while the WARN makes the bad config visible.
+   * <em>before</em> eligibility and reintroduce missed rounds), or a value
+   * {@code >= roundTimeMills} (which wraps through {@link Math#floorMod} and no longer represents
+   * the documented "epsilon after the boundary") all fall back to
+   * {@link #DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS} with a one-shot WARN. Riding over the
+   * misconfiguration keeps replay polling on a safe default rather than wedging the group, while
+   * the WARN makes the bad config visible.
    * @return the epsilon margin in milliseconds, always within {@code [0, roundTimeMills)}.
    */
   public long getAlignedDelayEpsilonMillis() {
@@ -711,8 +727,8 @@ public long getAlignedDelayEpsilonMillis() {
     } catch (NumberFormatException e) {
       // Hadoop's getLong throws (rather than returning the default) when the key is present but
       // not parseable as a number.
-      warnInvalidEpsilon("non-numeric value \""
-        + conf.get(REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY) + "\"");
+      warnInvalidEpsilon(
+        "non-numeric value \"" + conf.get(REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY) + "\"");
       return DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS;
     }
     if (epsilon < 0 || epsilon >= roundTimeMills) {

phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java

@@ -104,8 +104,14 @@ public void init() throws IOException {
     super.init();
   }
 
+  /**
+   * Forwards a single log file to the peer cluster. The {@code firstClaim} flag (true when the file
+   * was just claimed from the new-files directory, false on an in-progress reclaim) is unused here:
+   * forwarding is idempotent on the stable (ts, originServerName) destination and records no
+   * claim-latency metrics, so first claims and reclaims are handled identically.
+   */
   @Override
-  protected void processFile(Path src) throws IOException {
+  protected void processFile(Path src, boolean firstClaim) throws IOException {
     FileSystem srcFS = replicationLogTracker.getFileSystem();
     FileStatus srcStat = srcFS.getFileStatus(src);
     long ts = replicationLogTracker.getFileTimestamp(srcStat.getPath());

phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java

@@ -384,6 +384,11 @@ protected Optional<Path> markInProgress(final Path file) {
         return Optional.of(newPath);
       } else {
         LOG.warn("Failed to rename file for in-progress marking: {}", file);
+        // Claim rename returned false. Usually another process claimed the same file first (the
+        // common collision case), but any other rename() == false outcome lands here too, so the
+        // counter is named for the observable condition (rename failed), not the presumed cause. A
+        // strict subset of the request count incremented in the finally block below.
+        getMetrics().incrementMarkFileInProgressRenameFailedCount();
         return Optional.empty();
       }
     } catch (IOException e) {

phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscovery.java

@@ -37,6 +37,9 @@ public interface MetricsReplicationLogDiscovery extends BaseSource {
   String TIME_TO_PROCESS_IN_PROGRESS_FILES = "timeToProcessInProgressFilesMs";
   String TIME_TO_PROCESS_IN_PROGRESS_FILES_DESC =
     "Histogram of time taken to process in progress files in milliseconds";
+  String ROUNDS_EXCEEDING_ROUND_TIME = "roundsExceedingRoundTime";
+  String ROUNDS_EXCEEDING_ROUND_TIME_DESC =
+    "Number of rounds whose new-file processing time exceeded the round time";
 
   /**
    * Increments the counter for rounds processed. This counter tracks the number of rounds processed
@@ -62,6 +65,13 @@ public interface MetricsReplicationLogDiscovery extends BaseSource {
    */
   void updateTimeToProcessInProgressFiles(long timeMs);
 
+  /**
+   * Increments the counter for rounds whose new-file processing time exceeded the round time. A
+   * rising rate of such rounds signals that replay is falling behind the cadence at which new
+   * rounds become eligible.
+   */
+  void incrementRoundsExceedingRoundTime();
+
   /**
    * Unregister this metrics source.
    */

phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryImpl.java

@@ -31,6 +31,7 @@ public class MetricsReplicationLogDiscoveryImpl extends BaseSourceImpl
   protected final MutableFastCounter numInProgressDirectoryProcessed;
   protected final MutableHistogram timeToProcessNewFiles;
   protected final MutableHistogram timeToProcessInProgressFiles;
+  protected final MutableFastCounter roundsExceedingRoundTime;
 
   public MetricsReplicationLogDiscoveryImpl(String metricsName, String metricsDescription,
     String metricsContext, String metricsJmxContext) {
@@ -43,6 +44,8 @@ public MetricsReplicationLogDiscoveryImpl(String metricsName, String metricsDesc
       getMetricsRegistry().newHistogram(TIME_TO_PROCESS_NEW_FILES, TIME_TO_PROCESS_NEW_FILES_DESC);
     timeToProcessInProgressFiles = getMetricsRegistry()
       .newHistogram(TIME_TO_PROCESS_IN_PROGRESS_FILES, TIME_TO_PROCESS_IN_PROGRESS_FILES_DESC);
+    roundsExceedingRoundTime = getMetricsRegistry().newCounter(ROUNDS_EXCEEDING_ROUND_TIME,
+      ROUNDS_EXCEEDING_ROUND_TIME_DESC, 0L);
   }
 
   @Override
@@ -65,6 +68,11 @@ public void updateTimeToProcessInProgressFiles(long timeMs) {
     timeToProcessInProgressFiles.add(timeMs);
   }
 
+  @Override
+  public void incrementRoundsExceedingRoundTime() {
+    roundsExceedingRoundTime.incr();
+  }
+
   @Override
   public void close() {
     // Unregister this metrics source
@@ -75,7 +83,7 @@ public void close() {
   public ReplicationLogDiscoveryMetricValues getCurrentMetricValues() {
     return new ReplicationLogDiscoveryMetricValues(numRoundsProcessed.value(),
       numInProgressDirectoryProcessed.value(), timeToProcessNewFiles.getMax(),
-      timeToProcessInProgressFiles.getMax());
+      timeToProcessInProgressFiles.getMax(), roundsExceedingRoundTime.value());
   }
 
   @Override

phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplay.java

@@ -26,6 +26,14 @@ public interface MetricsReplicationLogDiscoveryReplay extends MetricsReplication
   String CONSISTENCY_POINT = "consistencyPoint";
   String CONSISTENCY_POINT_DESC =
     "Consistency point timestamp in milliseconds for the HA Group during replay";
+  String END_TO_END_REPLAY_LAG = "endToEndReplayLagMs";
+  String END_TO_END_REPLAY_LAG_DESC =
+    "Histogram of end-to-end replay lag, from when a file's round became eligible for processing "
+      + "to when the file finished replaying, in milliseconds";
+  String PICKUP_LAG = "pickupLagMs";
+  String PICKUP_LAG_DESC =
+    "Histogram of pickup lag, from when a file's round became eligible for processing to when the "
+      + "file was claimed (renamed into the in-progress directory), in milliseconds";
 
   /**
    * Updates the consistency point metric. The consistency point represents the timestamp up to
@@ -34,4 +42,19 @@ public interface MetricsReplicationLogDiscoveryReplay extends MetricsReplication
    * @param consistencyPointMs The consistency point timestamp in milliseconds
    */
   void updateConsistencyPoint(long consistencyPointMs);
+
+  /**
+   * Records a sample into the end-to-end replay lag histogram: the elapsed time from when a file's
+   * round became eligible for processing to when the file finished replaying.
+   * @param lagMs The end-to-end replay lag in milliseconds
+   */
+  void updateEndToEndReplayLag(long lagMs);
+
+  /**
+   * Records a sample into the pickup lag histogram: the elapsed time from when a file's round
+   * became eligible for processing to when the file was claimed (renamed into the in-progress
+   * directory).
+   * @param lagMs The pickup lag in milliseconds
+   */
+  void updatePickupLag(long lagMs);
 }

phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplayImpl.java

@@ -18,6 +18,9 @@
 package org.apache.phoenix.replication.metrics;
 
 import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
+import org.apache.hadoop.metrics2.lib.MutableHistogram;
+
+import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
 
 /** Implementation of metrics source for ReplicationLogDiscoveryReplay operations. */
 public class MetricsReplicationLogDiscoveryReplayImpl extends MetricsReplicationLogDiscoveryImpl
@@ -29,17 +32,42 @@ public class MetricsReplicationLogDiscoveryReplayImpl extends MetricsReplication
   private static final String METRICS_JMX_CONTEXT = "RegionServer,sub=" + METRICS_NAME;
 
   private final MutableGaugeLong consistencyPoint;
+  private final MutableHistogram endToEndReplayLag;
+  private final MutableHistogram pickupLag;
 
   public MetricsReplicationLogDiscoveryReplayImpl(final String haGroupName) {
     super(MetricsReplicationLogDiscoveryReplayImpl.METRICS_NAME,
       MetricsReplicationLogDiscoveryReplayImpl.METRICS_DESCRIPTION,
       MetricsReplicationLogDiscoveryImpl.METRICS_CONTEXT,
       MetricsReplicationLogDiscoveryReplayImpl.METRICS_JMX_CONTEXT + ",haGroup=" + haGroupName);
     consistencyPoint = getMetricsRegistry().newGauge(CONSISTENCY_POINT, CONSISTENCY_POINT_DESC, 0L);
+    endToEndReplayLag =
+      getMetricsRegistry().newHistogram(END_TO_END_REPLAY_LAG, END_TO_END_REPLAY_LAG_DESC);
+    pickupLag = getMetricsRegistry().newHistogram(PICKUP_LAG, PICKUP_LAG_DESC);
   }
 
   @Override
   public void updateConsistencyPoint(long consistencyPointMs) {
     consistencyPoint.set(consistencyPointMs);
   }
+
+  @Override
+  public void updateEndToEndReplayLag(long lagMs) {
+    endToEndReplayLag.add(lagMs);
+  }
+
+  @Override
+  public void updatePickupLag(long lagMs) {
+    pickupLag.add(lagMs);
+  }
+
+  @VisibleForTesting
+  MutableHistogram getEndToEndReplayLagHistogram() {
+    return endToEndReplayLag;
+  }
+
+  @VisibleForTesting
+  MutableHistogram getPickupLagHistogram() {
+    return pickupLag;
+  }
 }

phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java

@@ -41,6 +41,13 @@ public interface MetricsReplicationLogProcessor extends BaseSource {
   String LOG_FILE_REPLAY_TIME = "logFileReplayTimeMs";
   String LOG_FILE_REPLAY_TIME_DESC =
     "Histogram of time taken for replaying a log file in milliseconds";
+  String SUCCESSFUL_FILE_MUTATIONS_REPLAYED_COUNT = "successfulFileMutationsReplayedCount";
+  String SUCCESSFUL_FILE_MUTATIONS_REPLAYED_COUNT_DESC =
+    "Total number of mutations replayed across log files that completed successfully; mutations "
+      + "already applied by a file that then fails mid-replay are not counted";
+  String MUTATIONS_PER_FILE = "mutationsPerFile";
+  String MUTATIONS_PER_FILE_DESC =
+    "Histogram of the number of mutations replayed per log file (files with mutations only)";
 
   /**
    * Increments the counter for failed mutations. This counter tracks the number of mutations that
@@ -78,6 +85,22 @@ public interface MetricsReplicationLogProcessor extends BaseSource {
    */
   void updateLogFileReplayTime(long timeMs);
 
+  /**
+   * Increments the count of mutations replayed by files that completed successfully. Called once
+   * per file after its replay returns without error, so mutations already applied by a file that
+   * then fails mid-replay are not counted. Tracks replay throughput across fully-successful files.
+   * @param delta the number of mutations replayed for a successfully completed log file
+   */
+  void incrementSuccessfulFileMutationsReplayedCount(long delta);
+
+  /**
+   * Records a sample into the per-file mutation-count histogram. Called only for files that
+   * replayed at least one mutation, so rotation-only (zero-mutation) files do not skew the
+   * distribution toward zero.
+   * @param mutationsCount the number of mutations replayed for a single log file
+   */
+  void updateMutationsPerFile(long mutationsCount);
+
   /**
    * Unregister this metrics source.
    */