Emit metrics about recovery operations from server processes (#6081)

Author: dlmarion

core/src/main/java/org/apache/accumulo/core/metrics/Metric.java

@@ -306,7 +306,17 @@ public enum Metric {
       MetricDocSection.MANAGER),
   MANAGER_GOAL_STATE("accumulo.manager.goal.state", MetricType.GAUGE,
       "Manager goal state: -1=unknown, 0=CLEAN_STOP, 1=SAFE_MODE, 2=NORMAL.",
-      MetricDocSection.MANAGER);
+      MetricDocSection.MANAGER),
+
+  // Recovery Metrics
+  RECOVERIES_IN_PROGRESS("accumulo.recoveries.in.progress", MetricType.GAUGE,
+      "The number of recoveries in progress.", MetricDocSection.GENERAL_SERVER),
+  RECOVERIES_LONGEST_RUNTIME("accumulo.recoveries.runtime.longest", MetricType.GAUGE,
+      "The time (in milliseconds) of the longest running recovery.",
+      MetricDocSection.GENERAL_SERVER),
+  RECOVERIES_AVG_PROGRESS("accumulo.recoveries.avg.progress", MetricType.GAUGE,
+      "The average percentage (0.0 - 99.9) of the in progress recoveries.",
+      MetricDocSection.GENERAL_SERVER);
 
   private final String name;
   private final MetricType type;

server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java

@@ -819,7 +819,8 @@ public void run() {
 
     MetricsInfo metricsInfo = getContext().getMetricsInfo();
 
-    metricsInfo.addMetricsProducers(this, pausedMetrics);
+    final LogSorter logSorter = new LogSorter(this);
+    metricsInfo.addMetricsProducers(this, pausedMetrics, logSorter);
     metricsInfo.init(getServiceTags(clientAddress));
 
     var watcher = new CompactionWatcher(getConfiguration());
@@ -831,7 +832,6 @@ public void run() {
     LOG.info("Compactor started, waiting for work");
 
     final AtomicReference<Throwable> err = new AtomicReference<>();
-    final LogSorter logSorter = new LogSorter(this);
     long nextSortLogsCheckTime = System.currentTimeMillis();
 
     while (!isShutdownRequested()) {

server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java

@@ -377,16 +377,14 @@ public void run() {
         resourceManager.getDataCache(), resourceManager.getSummaryCache());
 
     metricsInfo.addMetricsProducers(this, scanMetrics, scanServerMetrics, blockCacheMetrics);
-    metricsInfo.init(MetricsInfo.serviceTags(getContext().getInstanceName(), getApplicationName(),
-        getAdvertiseAddress(), getResourceGroup()));
-    // We need to set the compaction manager so that we don't get an NPE in CompactableImpl.close
 
     ServiceLock lock = announceExistence();
     this.getContext().setServiceLock(lock);
 
     int threadPoolSize = getConfiguration().getCount(Property.SSERV_WAL_SORT_MAX_CONCURRENT);
     if (threadPoolSize > 0) {
       final LogSorter logSorter = new LogSorter(this);
+      metricsInfo.addMetricsProducers(logSorter);
       try {
         // Attempt to process all existing log sorting work and start a background
         // thread to look for log sorting work in the future
@@ -400,6 +398,9 @@ public void run() {
           "Log sorting for tablet recovery is disabled, SSERV_WAL_SORT_MAX_CONCURRENT is less than 1.");
     }
 
+    metricsInfo.init(MetricsInfo.serviceTags(getContext().getInstanceName(), getApplicationName(),
+        getAdvertiseAddress(), getResourceGroup()));
+
     while (!isShutdownRequested()) {
       if (Thread.currentThread().isInterrupted()) {
         LOG.info("Server process thread has been interrupted, shutting down");

server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java

@@ -558,7 +558,7 @@ public void run() {
         this.resourceManager.getDataCache(), this.resourceManager.getSummaryCache());
 
     metricsInfo.addMetricsProducers(this, metrics, updateMetrics, scanMetrics, mincMetrics,
-        pausedMetrics, blockCacheMetrics);
+        pausedMetrics, blockCacheMetrics, logSorter);
     metricsInfo.init(MetricsInfo.serviceTags(context.getInstanceName(), getApplicationName(),
         getAdvertiseAddress(), getResourceGroup()));
 

server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java

@@ -19,6 +19,9 @@
 package org.apache.accumulo.tserver.log;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.accumulo.core.metrics.Metric.RECOVERIES_AVG_PROGRESS;
+import static org.apache.accumulo.core.metrics.Metric.RECOVERIES_IN_PROGRESS;
+import static org.apache.accumulo.core.metrics.Metric.RECOVERIES_LONGEST_RUNTIME;
 import static org.apache.accumulo.core.util.threads.ThreadPoolNames.TSERVER_WAL_SORT_CONCURRENT_POOL;
 
 import java.io.DataInputStream;
@@ -32,6 +35,8 @@
 import java.util.Map.Entry;
 import java.util.TreeMap;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
@@ -43,6 +48,7 @@
 import org.apache.accumulo.core.file.FileOperations;
 import org.apache.accumulo.core.manager.thrift.RecoveryStatus;
 import org.apache.accumulo.core.metadata.UnreferencedTabletFile;
+import org.apache.accumulo.core.metrics.MetricsProducer;
 import org.apache.accumulo.core.spi.crypto.CryptoEnvironment;
 import org.apache.accumulo.core.spi.crypto.CryptoService;
 import org.apache.accumulo.core.util.Pair;
@@ -64,9 +70,13 @@
 import org.slf4j.LoggerFactory;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.AtomicDouble;
 import com.google.common.util.concurrent.MoreExecutors;
 
-public class LogSorter {
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+
+public class LogSorter implements MetricsProducer {
 
   private static final Logger log = LoggerFactory.getLogger(LogSorter.class);
 
@@ -229,6 +239,9 @@ synchronized long getBytesCopied() throws IOException {
   private final double walBlockSize;
   private final CryptoService cryptoService;
   private final AccumuloConfiguration sortedLogConf;
+  private final AtomicLong recoveriesInProgress = new AtomicLong(0);
+  private final AtomicLong recoveryRuntime = new AtomicLong(0);
+  private final AtomicDouble recoveryAvgProgress = new AtomicDouble(0.0D);
 
   public LogSorter(AbstractServer server) {
     this.server = server;
@@ -239,6 +252,8 @@ public LogSorter(AbstractServer server) {
     CryptoEnvironment env = new CryptoEnvironmentImpl(CryptoEnvironment.Scope.RECOVERY);
     this.cryptoService =
         context.getCryptoFactory().getService(env, this.conf.getAllCryptoProperties());
+    ThreadPools.watchNonCriticalScheduledTask(context.getScheduledExecutor()
+        .scheduleWithFixedDelay(() -> updateMetrics(), 60, 60, TimeUnit.SECONDS));
   }
 
   /**
@@ -335,4 +350,41 @@ public List<RecoveryStatus> getLogSorts() {
       return result;
     }
   }
+
+  private void updateMetrics() {
+    synchronized (currentWork) {
+      recoveriesInProgress.set(currentWork.size());
+      if (recoveriesInProgress.get() == 0) {
+        recoveryRuntime.set(0);
+        recoveryAvgProgress.set(0.0D);
+      } else {
+        long runtime = 0;
+        long progress = 0;
+        for (LogProcessor processor : currentWork.values()) {
+          long start = processor.getSortTime();
+          if (start > 0) { // may not have started yet
+            runtime = Math.max(start, runtime);
+          }
+          try {
+            progress += processor.getBytesCopied();
+          } catch (IOException e) {
+            log.warn("Error getting bytes read");
+          }
+        }
+        recoveryRuntime.set(runtime);
+        recoveryAvgProgress
+            .set(Math.min(progress / (walBlockSize * recoveriesInProgress.get()), 99.9));
+      }
+    }
+  }
+
+  @Override
+  public void registerMetrics(MeterRegistry registry) {
+    Gauge.builder(RECOVERIES_IN_PROGRESS.getName(), recoveriesInProgress, AtomicLong::get)
+        .description(RECOVERIES_IN_PROGRESS.getDescription()).register(registry);
+    Gauge.builder(RECOVERIES_LONGEST_RUNTIME.getName(), recoveryRuntime, AtomicLong::get)
+        .description(RECOVERIES_LONGEST_RUNTIME.getDescription()).register(registry);
+    Gauge.builder(RECOVERIES_AVG_PROGRESS.getName(), recoveryAvgProgress, AtomicDouble::get)
+        .description(RECOVERIES_AVG_PROGRESS.getDescription()).register(registry);
+  }
 }

server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java

@@ -37,6 +37,7 @@
 import java.util.Objects;
 import java.util.TreeMap;
 import java.util.UUID;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 import org.apache.accumulo.core.conf.DefaultConfiguration;
 import org.apache.accumulo.core.data.TableId;
@@ -54,6 +55,7 @@
 import org.apache.accumulo.tserver.logger.LogFileValue;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -64,6 +66,7 @@
 @SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "paths not set by user input")
 public class RecoveryLogsIteratorTest extends WithTestNames {
 
+  private static final ScheduledThreadPoolExecutor EXECUTOR = new ScheduledThreadPoolExecutor(1);
   private VolumeManager fs;
   private java.nio.file.Path workDir;
   static final KeyExtent extent = new KeyExtent(TableId.of("table"), null, null);
@@ -85,6 +88,7 @@ public void setUp() throws Exception {
     expect(context.getCryptoFactory()).andReturn(new GenericCryptoServiceFactory()).anyTimes();
     expect(context.getVolumeManager()).andReturn(fs).anyTimes();
     expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
+    expect(context.getScheduledExecutor()).andReturn(EXECUTOR).anyTimes();
     replay(server, context);
 
     logSorter = new LogSorter(server);
@@ -96,6 +100,11 @@ public void tearDown() throws Exception {
     verify(server, context);
   }
 
+  @AfterAll
+  public static void shutdown() {
+    EXECUTOR.shutdownNow();
+  }
+
   static class KeyValue implements Comparable<KeyValue> {
     public final LogFileKey key;
     public final LogFileValue value;

server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java

@@ -44,6 +44,7 @@
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.UUID;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 import org.apache.accumulo.core.conf.ConfigurationCopy;
 import org.apache.accumulo.core.conf.DefaultConfiguration;
@@ -80,6 +81,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
 import org.easymock.EasyMock;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
@@ -92,6 +94,7 @@
 @SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "paths not set by user input")
 public class SortedLogRecoveryTest extends WithTestNames {
 
+  private static final ScheduledThreadPoolExecutor EXECUTOR = new ScheduledThreadPoolExecutor(1);
   static final int bufferSize = 5;
   static final KeyExtent extent = new KeyExtent(TableId.of("table"), null, null);
   static final Text cf = new Text("cf");
@@ -135,6 +138,11 @@ public int compareTo(KeyValue o) {
     }
   }
 
+  @AfterAll
+  public static void shutdown() {
+    EXECUTOR.shutdownNow();
+  }
+
   private static KeyValue createKeyValue(LogEvents type, long seq, int tid,
       Object fileExtentMutation) {
     KeyValue result = new KeyValue();
@@ -193,6 +201,7 @@ private List<Mutation> recover(Map<String,KeyValue[]> logs, Set<String> files, K
       expect(context.getVolumeManager()).andReturn(fs).anyTimes();
       expect(context.getCryptoFactory()).andReturn(cryptoFactory).anyTimes();
       expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
+      expect(context.getScheduledExecutor()).andReturn(EXECUTOR).anyTimes();
       replay(server, context);
       logSorter = new LogSorter(server);
 
@@ -1133,6 +1142,7 @@ public void testLogSortedProperties() throws Exception {
       expect(context.getCryptoFactory()).andReturn(cryptoFactory).anyTimes();
       expect(context.getVolumeManager()).andReturn(vm).anyTimes();
       expect(context.getConfiguration()).andReturn(testConfig).anyTimes();
+      expect(context.getScheduledExecutor()).andReturn(EXECUTOR).anyTimes();
       replay(server, context);
       LogSorter sorter = new LogSorter(server);
 

server/tserver/src/test/java/org/apache/accumulo/tserver/log/TestUpgradePathForWALogs.java

@@ -31,6 +31,7 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.file.Files;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 import org.apache.accumulo.core.conf.DefaultConfiguration;
 import org.apache.accumulo.core.spi.crypto.GenericCryptoServiceFactory;
@@ -40,6 +41,7 @@
 import org.apache.accumulo.tserver.TabletServer;
 import org.apache.accumulo.tserver.WithTestNames;
 import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -58,6 +60,8 @@ public class TestUpgradePathForWALogs extends WithTestNames {
   // logs from 2.0 were changed for improved crypto
   private static final String WALOG_FROM_20 = "walog-from-20.walog";
 
+  private static final ScheduledThreadPoolExecutor EXECUTOR = new ScheduledThreadPoolExecutor(1);
+
   private ServerContext context;
   private TabletServer server;
 
@@ -85,6 +89,7 @@ public void setUp() throws Exception {
     expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
     expect(context.getCryptoFactory()).andReturn(new GenericCryptoServiceFactory()).anyTimes();
     expect(context.getVolumeManager()).andReturn(fs).anyTimes();
+    expect(context.getScheduledExecutor()).andReturn(EXECUTOR).anyTimes();
     replay(server, context);
   }
 
@@ -93,6 +98,11 @@ public void tearDown() {
     verify(server, context);
   }
 
+  @AfterAll
+  public static void shutdown() {
+    EXECUTOR.shutdownNow();
+  }
+
   /**
    * Since 2.0 this version of WAL is no longer compatible.
    */