HBASE-22301 Consider rolling the WAL if the HDFS write pipeline is slow

Author: apurtell

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/wal/MetricsWALSource.java

@@ -58,10 +58,19 @@ public interface MetricsWALSource extends BaseSource {
   String SYNC_TIME = "syncTime";
   String SYNC_TIME_DESC = "The time it took to sync the WAL to HDFS.";
   String ROLL_REQUESTED = "rollRequest";
-  String ROLL_REQUESTED_DESC = "How many times a log roll has been requested total";
+  String ROLL_REQUESTED_DESC = "How many times a roll has been requested total";
+  String ERROR_ROLL_REQUESTED = "errorRollRequest";
+  String ERROR_ROLL_REQUESTED_DESC =
+      "How many times a roll was requested due to I/O or other errors.";
   String LOW_REPLICA_ROLL_REQUESTED = "lowReplicaRollRequest";
   String LOW_REPLICA_ROLL_REQUESTED_DESC =
-      "How many times a log roll was requested due to too few DN's in the write pipeline.";
+      "How many times a roll was requested due to too few datanodes in the write pipeline.";
+  String SLOW_SYNC_ROLL_REQUESTED = "slowSyncRollRequest";
+  String SLOW_SYNC_ROLL_REQUESTED_DESC =
+      "How many times a roll was requested due to sync too slow on the write pipeline.";
+  String SIZE_ROLL_REQUESTED = "sizeRollRequest";
+  String SIZE_ROLL_REQUESTED_DESC =
+      "How many times a roll was requested due to file size roll threshold.";
   String WRITTEN_BYTES = "writtenBytes";
   String WRITTEN_BYTES_DESC = "Size (in bytes) of the data written to the WAL.";
 
@@ -92,8 +101,14 @@ public interface MetricsWALSource extends BaseSource {
 
   void incrementLogRollRequested();
 
+  void incrementErrorLogRoll();
+
   void incrementLowReplicationLogRoll();
 
+  void incrementSlowSyncLogRoll();
+
+  void incrementSizeLogRoll();
+
   void incrementWrittenBytes(long val);
 
   long getWrittenBytes();

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/regionserver/wal/MetricsWALSourceImpl.java

@@ -39,7 +39,10 @@ public class MetricsWALSourceImpl extends BaseSourceImpl implements MetricsWALSo
   private final MutableFastCounter appendCount;
   private final MutableFastCounter slowAppendCount;
   private final MutableFastCounter logRollRequested;
-  private final MutableFastCounter lowReplicationLogRollRequested;
+  private final MutableFastCounter errorRollRequested;
+  private final MutableFastCounter lowReplicationRollRequested;
+  private final MutableFastCounter slowSyncRollRequested;
+  private final MutableFastCounter sizeRollRequested;
   private final MutableFastCounter writtenBytes;
 
   public MetricsWALSourceImpl() {
@@ -61,8 +64,14 @@ public MetricsWALSourceImpl(String metricsName,
     syncTimeHisto = this.getMetricsRegistry().newTimeHistogram(SYNC_TIME, SYNC_TIME_DESC);
     logRollRequested =
         this.getMetricsRegistry().newCounter(ROLL_REQUESTED, ROLL_REQUESTED_DESC, 0L);
-    lowReplicationLogRollRequested = this.getMetricsRegistry()
+    errorRollRequested = this.getMetricsRegistry()
+        .newCounter(ERROR_ROLL_REQUESTED, ERROR_ROLL_REQUESTED_DESC, 0L);
+    lowReplicationRollRequested = this.getMetricsRegistry()
         .newCounter(LOW_REPLICA_ROLL_REQUESTED, LOW_REPLICA_ROLL_REQUESTED_DESC, 0L);
+    slowSyncRollRequested = this.getMetricsRegistry()
+        .newCounter(SLOW_SYNC_ROLL_REQUESTED, SLOW_SYNC_ROLL_REQUESTED_DESC, 0L);
+    sizeRollRequested = this.getMetricsRegistry()
+        .newCounter(SIZE_ROLL_REQUESTED, SIZE_ROLL_REQUESTED_DESC, 0L);
     writtenBytes = this.getMetricsRegistry().newCounter(WRITTEN_BYTES, WRITTEN_BYTES_DESC, 0l);
   }
 
@@ -96,9 +105,24 @@ public void incrementLogRollRequested() {
     logRollRequested.incr();
   }
 
+  @Override
+  public void incrementErrorLogRoll() {
+    errorRollRequested.incr();
+  }
+
   @Override
   public void incrementLowReplicationLogRoll() {
-    lowReplicationLogRollRequested.incr();
+    lowReplicationRollRequested.incr();
+  }
+
+  @Override
+  public void incrementSlowSyncLogRoll() {
+    slowSyncRollRequested.incr();
+  }
+
+  @Override
+  public void incrementSizeLogRoll() {
+    sizeRollRequested.incr();
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java

@@ -71,7 +71,7 @@ public void addWAL(final WAL wal) {
     if (null == walNeedsRoll.putIfAbsent(wal, Boolean.FALSE)) {
       wal.registerWALActionsListener(new WALActionsListener.Base() {
         @Override
-        public void logRollRequested(boolean lowReplicas) {
+        public void logRollRequested(WALActionsListener.RollRequestReason reason) {
           walNeedsRoll.put(wal, Boolean.TRUE);
           // TODO logs will contend with each other here, replace with e.g. DelayedQueue
           synchronized(rollLog) {

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java

@@ -17,6 +17,10 @@
  */
 package org.apache.hadoop.hbase.regionserver.wal;
 
+import static org.apache.hadoop.hbase.regionserver.wal.WALActionsListener.RollRequestReason.ERROR;
+import static org.apache.hadoop.hbase.regionserver.wal.WALActionsListener.RollRequestReason.LOW_REPLICATION;
+import static org.apache.hadoop.hbase.regionserver.wal.WALActionsListener.RollRequestReason.SIZE;
+import static org.apache.hadoop.hbase.regionserver.wal.WALActionsListener.RollRequestReason.SLOW_SYNC;
 import static org.apache.hadoop.hbase.wal.DefaultWALProvider.WAL_FILE_NAME_DELIMITER;
 
 import java.io.FileNotFoundException;
@@ -160,9 +164,18 @@ public class FSHLog implements WAL {
 
   private static final Log LOG = LogFactory.getLog(FSHLog.class);
 
-  private static final int DEFAULT_SLOW_SYNC_TIME_MS = 100; // in ms
+  static final String SLOW_SYNC_TIME_MS ="hbase.regionserver.wal.slowsync.ms";
+  static final int DEFAULT_SLOW_SYNC_TIME_MS = 100; // in ms
+  static final String ROLL_ON_SYNC_TIME_MS = "hbase.regionserver.wal.roll.on.sync.ms";
+  static final int DEFAULT_ROLL_ON_SYNC_TIME_MS = 10000; // in ms
+  static final String SLOW_SYNC_ROLL_THRESHOLD = "hbase.regionserver.wal.slowsync.roll.threshold";
+  static final int DEFAULT_SLOW_SYNC_ROLL_THRESHOLD = 100; // 100 slow sync warnings
+  static final String SLOW_SYNC_ROLL_INTERVAL_MS =
+    "hbase.regionserver.wal.slowsync.roll.interval.ms";
+  static final int DEFAULT_SLOW_SYNC_ROLL_INTERVAL_MS = 60 * 1000; // in ms, 1 minute
 
-  private static final int DEFAULT_WAL_SYNC_TIMEOUT_MS = 5 * 60 * 1000; // in ms, 5min
+  static final String WAL_SYNC_TIMEOUT_MS = "hbase.regionserver.wal.sync.timeout";
+  static final int DEFAULT_WAL_SYNC_TIMEOUT_MS = 5 * 60 * 1000; // in ms, 5min
 
   /**
    * The nexus at which all incoming handlers meet.  Does appends and sync with an ordering.
@@ -280,7 +293,10 @@ public WALCoprocessorHost getCoprocessorHost() {
 
   private final boolean useHsync;
 
-  private final int slowSyncNs;
+  private final long slowSyncNs, rollOnSyncNs;
+  private final int slowSyncRollThreshold;
+  private final int slowSyncCheckInterval;
+  private final AtomicInteger slowSyncCount = new AtomicInteger();
 
   private final long walSyncTimeout;
 
@@ -350,9 +366,13 @@ public WALCoprocessorHost getCoprocessorHost() {
 
   private final AtomicInteger closeErrorCount = new AtomicInteger();
 
+  protected volatile boolean rollRequested;
+
   // Last time to check low replication on hlog's pipeline
   private volatile long lastTimeCheckLowReplication = EnvironmentEdgeManager.currentTime();
 
+  // Last time we asked to roll the log due to a slow sync
+  private volatile long lastTimeCheckSlowSync = EnvironmentEdgeManager.currentTime();
 
   /**
    * WAL Comparator; it compares the timestamp (log filenum), present in the log file name.
@@ -540,11 +560,16 @@ public boolean accept(final Path fileName) {
     // rollWriter sets this.hdfs_out if it can.
     rollWriter();
 
-    this.slowSyncNs =
-        1000000 * conf.getInt("hbase.regionserver.hlog.slowsync.ms",
-          DEFAULT_SLOW_SYNC_TIME_MS);
-    this.walSyncTimeout = conf.getLong("hbase.regionserver.hlog.sync.timeout",
-        DEFAULT_WAL_SYNC_TIMEOUT_MS);
+    this.slowSyncNs = TimeUnit.MILLISECONDS.toNanos(conf.getInt(SLOW_SYNC_TIME_MS,
+      conf.getInt("hbase.regionserver.hlog.slowsync.ms", DEFAULT_SLOW_SYNC_TIME_MS)));
+    this.rollOnSyncNs = TimeUnit.MILLISECONDS.toNanos(conf.getInt(ROLL_ON_SYNC_TIME_MS,
+      DEFAULT_ROLL_ON_SYNC_TIME_MS));
+    this.slowSyncRollThreshold = conf.getInt(SLOW_SYNC_ROLL_THRESHOLD,
+      DEFAULT_SLOW_SYNC_ROLL_THRESHOLD);
+    this.slowSyncCheckInterval = conf.getInt(SLOW_SYNC_ROLL_INTERVAL_MS,
+      DEFAULT_SLOW_SYNC_ROLL_INTERVAL_MS);
+    this.walSyncTimeout = conf.getLong(WAL_SYNC_TIMEOUT_MS,
+      conf.getLong("hbase.regionserver.hlog.sync.timeout", DEFAULT_WAL_SYNC_TIMEOUT_MS));
 
     // This is the 'writer' -- a single threaded executor.  This single thread 'consumes' what is
     // put on the ring buffer.
@@ -720,6 +745,11 @@ private void preemptiveSync(final ProtobufLogWriter nextWriter) {
         // NewPath could be equal to oldPath if replaceWriter fails.
         newPath = replaceWriter(oldPath, newPath, nextWriter, nextHdfsOut);
         tellListenersAboutPostLogRoll(oldPath, newPath);
+        // Reset rollRequested status
+        rollRequested = false;
+        // We got a new writer, so reset the slow sync count
+        lastTimeCheckSlowSync = EnvironmentEdgeManager.currentTime();
+        slowSyncCount.set(0);
         // Can we delete any of the old log files?
         if (getNumRolledLogFiles() > 0) {
           cleanOldLogs();
@@ -1303,8 +1333,11 @@ public void run() {
             syncCount += releaseSyncFuture(takeSyncFuture, currentSequence, lastException);
             // Can we release other syncs?
             syncCount += releaseSyncFutures(currentSequence, lastException);
-            if (lastException != null) requestLogRoll();
-            else checkLogRoll();
+            if (lastException != null) {
+              requestLogRoll();
+            } else {
+              checkLogRoll();
+            }
           }
           postSync(System.nanoTime() - start, syncCount);
         } catch (InterruptedException e) {
@@ -1321,24 +1354,37 @@ public void run() {
    * Schedule a log roll if needed.
    */
   public void checkLogRoll() {
+    // If we have already requested a roll, do nothing
+    if (isLogRollRequested()) {
+      return;
+    }
     // Will return immediately if we are in the middle of a WAL log roll currently.
-    if (!rollWriterLock.tryLock()) return;
-    boolean lowReplication;
-    try {
-      lowReplication = checkLowReplication();
-    } finally {
-      rollWriterLock.unlock();
+    if (!rollWriterLock.tryLock()) {
+      return;
     }
     try {
-      if (lowReplication || (writer != null && writer.getLength() > logrollsize)) {
-        requestLogRoll(lowReplication);
+      if (checkLowReplication()) {
+        LOG.warn("Requesting log roll because of low replication, current pipeline: " +
+            Arrays.toString(getPipeLine()));
+        requestLogRoll(LOW_REPLICATION);
+      } else if (writer != null && writer.getLength() > logrollsize) {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Requesting log roll because of file size threshold; length=" +
+            writer.getLength() + ", logrollsize=" + logrollsize);
+        }
+        requestLogRoll(SIZE);
+      } else if (checkSlowSync()) {
+        // We log this already in checkSlowSync
+        requestLogRoll(SLOW_SYNC);
       }
     } catch (IOException e) {
       LOG.warn("Writer.getLength() failed; continuing", e);
+    } finally {
+      rollWriterLock.unlock();
     }
   }
 
-  /*
+  /**
    * @return true if number of replicas for the WAL is lower than threshold
    */
   private boolean checkLowReplication() {
@@ -1389,6 +1435,41 @@ private boolean checkLowReplication() {
     return logRollNeeded;
   }
 
+  /**
+   * @return true if we exceeded the slow sync roll threshold over the last check
+   *              interval
+   */
+  private boolean checkSlowSync() {
+    boolean result = false;
+    long now = EnvironmentEdgeManager.currentTime();
+    long elapsedTime = now - lastTimeCheckSlowSync;
+    if (elapsedTime >= slowSyncCheckInterval) {
+      if (slowSyncCount.get() >= slowSyncRollThreshold) {
+        if (elapsedTime >= (2 * slowSyncCheckInterval)) {
+          // If two or more slowSyncCheckInterval have elapsed this is a corner case
+          // where a train of slow syncs almost triggered us but then there was a long
+          // interval from then until the one more that pushed us over. If so, we
+          // should do nothing and let the count reset.
+          if (LOG.isDebugEnabled()) {
+            LOG.debug("checkSlowSync triggered but we decided to ignore it; " +
+              "count=" + slowSyncCount.get() + ", threshold=" + slowSyncRollThreshold +
+              ", elapsedTime=" + elapsedTime +  " ms, slowSyncCheckInterval=" +
+              slowSyncCheckInterval + " ms");
+          }
+          // Fall through to count reset below
+        } else {
+          LOG.warn("Requesting log roll because we exceeded slow sync threshold; count=" +
+            slowSyncCount.get() + ", threshold=" + slowSyncRollThreshold +
+            ", current pipeline: " + Arrays.toString(getPipeLine()));
+          result = true;
+        }
+      }
+      lastTimeCheckSlowSync = now;
+      slowSyncCount.set(0);
+    }
+    return result;
+  }
+
   private SyncFuture publishSyncOnRingBuffer(long sequence) {
     return publishSyncOnRingBuffer(sequence, null, false);
   }
@@ -1452,10 +1533,23 @@ private void postSync(final long timeInNanos, final int handlerSyncs) {
     if (timeInNanos > this.slowSyncNs) {
       String msg =
           new StringBuilder().append("Slow sync cost: ")
-              .append(timeInNanos / 1000000).append(" ms, current pipeline: ")
+              .append(TimeUnit.NANOSECONDS.toMillis(timeInNanos))
+              .append(" ms, current pipeline: ")
               .append(Arrays.toString(getPipeLine())).toString();
       Trace.addTimelineAnnotation(msg);
       LOG.info(msg);
+      // A single sync took too long.
+      // Elsewhere in checkSlowSync, called from checkLogRoll, we will look at cumulative
+      // effects. Here we have a single data point that indicates we should take immediate
+      // action, so do so.
+      if (timeInNanos > this.rollOnSyncNs) {
+        LOG.warn("Requesting log roll because we exceeded slow sync threshold; time=" +
+          TimeUnit.NANOSECONDS.toMillis(timeInNanos) + " ms, threshold=" +
+          TimeUnit.NANOSECONDS.toMillis(rollOnSyncNs) + " ms, current pipeline: " +
+          Arrays.toString(getPipeLine()));
+        requestLogRoll(SLOW_SYNC);
+      }
+      slowSyncCount.incrementAndGet(); // it's fine to unconditionally increment this
     }
     if (!listeners.isEmpty()) {
       for (WALActionsListener listener : listeners) {
@@ -1539,15 +1633,24 @@ public void sync(long txid, boolean forceSync) throws IOException {
     }
   }
 
+  protected boolean isLogRollRequested() {
+    return rollRequested;
+  }
+
   // public only until class moves to o.a.h.h.wal
   public void requestLogRoll() {
-    requestLogRoll(false);
+    requestLogRoll(ERROR);
   }
 
-  private void requestLogRoll(boolean tooFewReplicas) {
+  private void requestLogRoll(final WALActionsListener.RollRequestReason reason) {
+    // If we have already requested a roll, don't do it again
+    if (rollRequested) {
+      return;
+    }
     if (!this.listeners.isEmpty()) {
+      rollRequested = true; // No point to assert this unless there is a registered listener
       for (WALActionsListener i: this.listeners) {
-        i.logRollRequested(tooFewReplicas);
+        i.logRollRequested(reason);
       }
     }
   }
@@ -1599,8 +1702,7 @@ boolean isLowReplicationRollEnabled() {
 
   public static final long FIXED_OVERHEAD = ClassSize.align(
     ClassSize.OBJECT + (5 * ClassSize.REFERENCE) +
-    ClassSize.ATOMIC_INTEGER + Bytes.SIZEOF_INT + (3 * Bytes.SIZEOF_LONG));
-
+    ClassSize.ATOMIC_INTEGER + (3 * Bytes.SIZEOF_INT) + (4 * Bytes.SIZEOF_LONG));
 
   private static void split(final Configuration conf, final Path p) throws IOException {
     FileSystem fs = FSUtils.getWALFileSystem(conf);
@@ -2083,4 +2185,14 @@ DatanodeInfo[] getPipeLine() {
   public long getLastTimeCheckLowReplication() {
     return this.lastTimeCheckLowReplication;
   }
+
+  @VisibleForTesting
+  Writer getWriter() {
+    return this.writer;
+  }
+
+  @VisibleForTesting
+  void setWriter(Writer writer) {
+    this.writer = writer;
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/MetricsWAL.java

@@ -72,10 +72,23 @@ public void postAppend(final long size, final long time, final WALKey logkey,
   }
 
   @Override
-  public void logRollRequested(boolean underReplicated) {
+  public void logRollRequested(WALActionsListener.RollRequestReason reason) {
     source.incrementLogRollRequested();
-    if (underReplicated) {
-      source.incrementLowReplicationLogRoll();
+    switch (reason) {
+      case ERROR:
+        source.incrementErrorLogRoll();
+        break;
+      case LOW_REPLICATION:
+        source.incrementLowReplicationLogRoll();
+        break;
+      case SIZE:
+        source.incrementSizeLogRoll();
+        break;
+      case SLOW_SYNC:
+        source.incrementSlowSyncLogRoll();
+        break;
+      default:
+        break;
     }
   }
 }