Added new scan executor metric

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

@@ -418,6 +418,13 @@
  * <td>Cache</td>
  * <td>scan server tablet cache metrics</td>
  * </tr>
+ * <tr>
+ * <td>N/A</td>
+ * <td>N/A</td>
+ * <td>{@value #METRICS_SCAN_EXCEPTIONS}</td>
+ * <td>Counter</td>
+ * <td>Count the number of exceptions that occur within scan executors</td>
+ * </tr>
  * <!-- scans -->
  * <tr>
  * <td>scan</td>
@@ -725,6 +732,7 @@ public interface MetricsProducer {
   String METRICS_SCAN_SCANNED_ENTRIES = METRICS_SCAN_PREFIX + "query.scanned.entries";
   String METRICS_SCAN_ZOMBIE_THREADS = METRICS_SCAN_PREFIX + "zombie.threads";
   String METRICS_SCAN_TABLET_METADATA_CACHE = METRICS_SCAN_PREFIX + "tablet.metadata.cache";
+  String METRICS_SCAN_EXCEPTIONS = METRICS_SCAN_PREFIX + "exceptions";
 
   String METRICS_TSERVER_PREFIX = "accumulo.tserver.";
   String METRICS_TSERVER_ENTRIES = METRICS_TSERVER_PREFIX + "entries";

server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java

@@ -19,6 +19,7 @@
 package org.apache.accumulo.tserver.metrics;
 
 import java.time.Duration;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.IntSupplier;
@@ -28,6 +29,7 @@
 
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.FunctionCounter;
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Timer;
@@ -47,6 +49,9 @@ public class TabletServerScanMetrics implements MetricsProducer {
   private final LongAdder queryResultCount = new LongAdder();
   private final LongAdder queryResultBytes = new LongAdder();
   private final LongAdder scannedCount = new LongAdder();
+  private final ConcurrentHashMap<String,AtomicLong> executorExceptionCounts =
+      new ConcurrentHashMap<>();
+  private volatile MeterRegistry registry = null;
 
   public void incrementLookupCount(long amount) {
     this.lookupCount.add(amount);
@@ -124,8 +129,28 @@ public TabletServerScanMetrics(IntSupplier openFileSupplier) {
     openFiles = openFileSupplier;
   }
 
+  public void incrementExecutorExceptions(String executorName) {
+    executorExceptionCounts.computeIfAbsent(executorName, k -> {
+      AtomicLong counter = new AtomicLong(0);
+      // Register the counter if the registry is already available
+      if (registry != null) {
+        registerExecutorExceptionCounter(executorName, counter);
+      }
+      return counter;
+    }).incrementAndGet();
+  }
+
+  private void registerExecutorExceptionCounter(String executorName, AtomicLong counter) {
+    FunctionCounter.builder(METRICS_SCAN_EXCEPTIONS, counter, AtomicLong::get)
+        .tags("executor", executorName)
+        .description(
+            "Number of exceptions thrown from the iterator stack during scan execution, tagged by executor name")
+        .register(registry);
+  }
+
   @Override
   public void registerMetrics(MeterRegistry registry) {
+    this.registry = registry;
     Gauge.builder(METRICS_SCAN_OPEN_FILES, openFiles::getAsInt)
         .description("Number of files open for scans").register(registry);
     scans = Timer.builder(METRICS_SCAN_TIMES).description("Scans").register(registry);
@@ -156,6 +181,7 @@ public void registerMetrics(MeterRegistry registry) {
     Gauge.builder(METRICS_SCAN_ZOMBIE_THREADS, this, TabletServerScanMetrics::getZombieThreadsCount)
         .description("Number of scan threads that have no associated client session")
         .register(registry);
+    executorExceptionCounts.forEach(this::registerExecutorExceptionCounter);
   }
 
 }

server/tserver/src/main/java/org/apache/accumulo/tserver/scan/LookupTask.java

@@ -179,9 +179,11 @@ public void run() {
         addResult(iie);
       }
     } catch (SampleNotPresentException e) {
+      recordException(session);
       addResult(e);
     } catch (Exception e) {
       log.warn("exception while doing multi-scan ", e);
+      recordException(session);
       addResult(e);
     } finally {
       transitionFromRunning();

server/tserver/src/main/java/org/apache/accumulo/tserver/scan/NextBatchTask.java

@@ -93,10 +93,12 @@ public void run() {
         addResult(iie);
       }
     } catch (TooManyFilesException | SampleNotPresentException e) {
+      recordException(scanSession);
       addResult(e);
     } catch (IOException | RuntimeException e) {
       log.warn("exception while scanning tablet {} for {}", scanSession.extent, scanSession.client,
           e);
+      recordException(scanSession);
       addResult(e);
     } finally {
       transitionFromRunning();

server/tserver/src/main/java/org/apache/accumulo/tserver/scan/ScanTask.java

@@ -30,6 +30,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.accumulo.tserver.TabletHostingServer;
+import org.apache.accumulo.tserver.session.ScanSession;
 
 import com.google.common.base.Preconditions;
 
@@ -167,6 +168,13 @@ private String stateString(int state) {
     return stateStr;
   }
 
+  protected void recordException(ScanSession<?> scanSession) {
+    if (scanSession != null && server.getScanMetrics() != null) {
+      String executorName = scanSession.scanParams.getScanDispatch().getExecutorName();
+      server.getScanMetrics().incrementExecutorExceptions(executorName);
+    }
+  }
+
   /**
    * @param busyTimeout when this less than 0 it has no impact. When its greater than 0 and the task
    *        is queued, then get() will sleep for the specified busyTimeout and if after sleeping its

test/src/main/java/org/apache/accumulo/test/metrics/MetricsIT.java

@@ -120,7 +120,8 @@ public void confirmMetricsPublished() throws Exception {
             METRICS_SCAN_RESERVATION_TOTAL_TIMER,
             METRICS_SCAN_RESERVATION_WRITEOUT_TIMER,
             METRICS_SCAN_TABLET_METADATA_CACHE,
-            METRICS_SERVER_IDLE);
+            METRICS_SERVER_IDLE,
+            METRICS_SCAN_EXCEPTIONS);
     // @formatter:on
 
     Map<String,String> expectedMetricNames = this.getMetricFields();

test/src/main/java/org/apache/accumulo/test/metrics/ScanExecutorExceptionsIT.java

@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.metrics;
+
+import static org.apache.accumulo.core.metrics.MetricsProducer.METRICS_SCAN_EXCEPTIONS;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.iterators.IteratorUtil;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.spi.metrics.LoggingMeterRegistryFactory;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.functional.ConfigurableMacBase;
+import org.apache.accumulo.test.functional.ErrorThrowingIterator;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ScanExecutorExceptionsIT extends ConfigurableMacBase {
+
+  private static final Logger log = LoggerFactory.getLogger(ScanExecutorExceptionsIT.class);
+  private static TestStatsDSink sink;
+
+  @Override
+  protected Duration defaultTimeout() {
+    return Duration.ofMinutes(5);
+  }
+
+  @BeforeAll
+  public static void setup() throws Exception {
+    sink = new TestStatsDSink();
+  }
+
+  @AfterAll
+  public static void teardown() throws Exception {
+    if (sink != null) {
+      sink.close();
+    }
+  }
+
+  @Override
+  protected void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
+    cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+    cfg.setProperty(Property.GENERAL_MICROMETER_JVM_METRICS_ENABLED, "true");
+    cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+    cfg.setProperty("general.custom.metrics.opts.logging.step", "5s");
+    String clazzList = LoggingMeterRegistryFactory.class.getName() + ","
+        + TestStatsDRegistryFactory.class.getName();
+    cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY, clazzList);
+    Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+        TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+    cfg.setSystemProperties(sysProps);
+  }
+
+  @Test
+  public void testScanExecutorExceptions() throws Exception {
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProperties()).build()) {
+
+      String tableName = getUniqueNames(1)[0];
+      log.info("Creating table: {}", tableName);
+
+      final int numEntries = 10;
+      final int totalExceptions = 6;
+
+      NewTableConfiguration ntc = new NewTableConfiguration();
+      IteratorSetting errorIterSetting =
+          new IteratorSetting(50, "error", ErrorThrowingIterator.class);
+      errorIterSetting.addOption(ErrorThrowingIterator.TIMES, String.valueOf(totalExceptions));
+      ntc.attachIterator(errorIterSetting, EnumSet.of(IteratorUtil.IteratorScope.scan));
+      client.tableOperations().create(tableName, ntc);
+
+      try (BatchWriter writer = client.createBatchWriter(tableName)) {
+        for (int i = 0; i < numEntries; i++) {
+          Mutation m = new Mutation(new Text("row" + i));
+          m.put("cf", "cq", "value" + i);
+          writer.addMutation(m);
+        }
+      }
+
+      client.tableOperations().flush(tableName, null, null, true);
+
+      log.info("Performing regular scan");
+      int scanCount = performScanCountingEntries(client, tableName);
+      log.info("Regular scan returned {} entries", scanCount);
+
+      boolean regularScanHadExceptions = checkMetricsForExceptions("regular scan");
+      assertTrue(regularScanHadExceptions, "Regular scan should have produced exception metrics");
+
+      log.info("Performing batch scan");
+      int batchScanCount = performBatchScanCountingEntries(client, tableName);
+      log.info("Batch scan returned {} entries", batchScanCount);
+
+      boolean batchScanHadExceptions = checkMetricsForExceptions("batch scan");
+      assertTrue(batchScanHadExceptions, "Batch scan should have produced exception metrics");
+    }
+  }
+
+  private boolean checkMetricsForExceptions(String scanType) throws InterruptedException {
+    List<String> statsDMetrics;
+    boolean foundMetric = false;
+    long startTime = System.currentTimeMillis();
+    long timeout = 30_000;
+
+    while (!foundMetric && (System.currentTimeMillis() - startTime) < timeout) {
+      statsDMetrics = sink.getLines();
+
+      if (!statsDMetrics.isEmpty()) {
+        for (String line : statsDMetrics) {
+          if (line.startsWith(METRICS_SCAN_EXCEPTIONS)) {
+            TestStatsDSink.Metric metric = TestStatsDSink.parseStatsDMetric(line);
+            String executor = metric.getTags().get("executor");
+            assertNotNull(executor,
+                "Executor tag should always be present in scan exception metrics");
+            long val = Long.parseLong(metric.getValue());
+            if (val > 0) {
+              foundMetric = true;
+              log.info("Found {} scan exception metric for executor '{}': {}", scanType, executor,
+                  val);
+            }
+          }
+        }
+      }
+
+      if (!foundMetric) {
+        Thread.sleep(1_000);
+      }
+    }
+
+    return foundMetric;
+  }
+
+  private int performScanCountingEntries(AccumuloClient client, String table) {
+    int count = 0;
+    try (Scanner scanner = client.createScanner(table, Authorizations.EMPTY)) {
+      scanner.setTimeout(10, java.util.concurrent.TimeUnit.SECONDS);
+      for (var entry : scanner) {
+        count++;
+        log.debug("Scan entry {}: {}", count, entry.getKey());
+      }
+      log.info("Scan completed successfully with {} entries", count);
+    } catch (Exception e) {
+      log.info("Exception during regular scan after {} entries: {}", count, e.getMessage());
+    }
+    return count;
+  }
+
+  private int performBatchScanCountingEntries(AccumuloClient client, String table) {
+    int count = 0;
+    try (BatchScanner batchScanner = client.createBatchScanner(table, Authorizations.EMPTY, 2)) {
+      batchScanner.setTimeout(10, java.util.concurrent.TimeUnit.SECONDS);
+      batchScanner.setRanges(List.of(new Range()));
+      for (var entry : batchScanner) {
+        count++;
+        log.debug("Batch scan entry {}: {}", count, entry.getKey());
+      }
+      log.info("Batch scan completed successfully with {} entries", count);
+    } catch (Exception e) {
+      log.info("Exception during batch scan after {} entries: {}", count, e.getMessage());
+    }
+    return count;
+  }
+}