Added monitor activity page for Fate transaction details

server/monitor/src/main/java/org/apache/accumulo/monitor/next/Endpoints.java

@@ -57,6 +57,7 @@
 import org.apache.accumulo.monitor.next.InformationFetcher.InstanceSummary;
 import org.apache.accumulo.monitor.next.SystemInformation.CompactionGroupSummary;
 import org.apache.accumulo.monitor.next.SystemInformation.CompactionTableSummary;
+import org.apache.accumulo.monitor.next.SystemInformation.FateTransaction;
 import org.apache.accumulo.monitor.next.SystemInformation.MessageCategory;
 import org.apache.accumulo.monitor.next.SystemInformation.MessagePriority;
 import org.apache.accumulo.monitor.next.SystemInformation.RecoveryInformation;
@@ -386,6 +387,14 @@ public CompactorsSummary getExternalCompactors() {
     return new CompactorsSummary(summary.getCompactorServers(), summary.getTimestamp());
   }
 
+  @GET
+  @Path("fate")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Description("Returns a list of fate transaction details")
+  public List<FateTransaction> getFateTransactions() {
+    return monitor.getInformationFetcher().getSummaryForEndpoint().getFateTransactions();
+  }
+
   @GET
   @Path("tables")
   @Produces(MediaType.APPLICATION_JSON)

server/monitor/src/main/java/org/apache/accumulo/monitor/next/InformationFetcher.java

@@ -28,8 +28,10 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CancellationException;
@@ -53,6 +55,14 @@
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.RowRange;
 import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.fate.AdminUtil;
+import org.apache.accumulo.core.fate.AdminUtil.FateStatus;
+import org.apache.accumulo.core.fate.FateId;
+import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.fate.ReadOnlyFateStore;
+import org.apache.accumulo.core.fate.user.UserFateStore;
+import org.apache.accumulo.core.fate.zookeeper.LockRange;
+import org.apache.accumulo.core.fate.zookeeper.MetaFateStore;
 import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
 import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
 import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
@@ -74,6 +84,8 @@
 import org.apache.accumulo.core.util.threads.ThreadPools;
 import org.apache.accumulo.server.ServerContext;
 import org.apache.accumulo.server.compaction.CompactionPluginUtils;
+import org.apache.accumulo.server.util.adminCommand.Fate;
+import org.apache.zookeeper.KeeperException;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.eclipse.jetty.util.NanoTime;
 import org.slf4j.Logger;
@@ -173,7 +185,7 @@ void add(UpdateTaskFuture f) {
   }
 
   enum UpdateType {
-    COMPACTION, COMPACTION_RGS, METRIC, TABLE;
+    COMPACTION, COMPACTION_RGS, FATE, METRIC, TABLE;
   }
 
   interface UpdateTask<T extends Object> extends Runnable, Comparable<UpdateTask<T>> {
@@ -412,6 +424,67 @@ public void run() {
     }
   }
 
+  class FateTransactionFetcher implements UpdateTask<Void> {
+
+    private final SystemInformation summary;
+    private Map<FateId,List<String>> emptyLocks = new HashMap<>();
+    private Map<FateId,LockRange> emptyRanges = new HashMap<>();
+
+    public FateTransactionFetcher(SystemInformation summary) {
+      this.summary = summary;
+    }
+
+    @Override
+    public void run() {
+      FateStatus status = AdminUtil.getTransactionStatus(stores, null, null, null, emptyLocks,
+          emptyLocks, emptyRanges);
+      summary.processFateTransactions(status.getTransactions());
+    }
+
+    @Override
+    public int hashCode() {
+      final int prime = 31;
+      int result = 1;
+      result = prime * result + Objects.hash(getType());
+      return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj) {
+        return true;
+      }
+      if (obj == null) {
+        return false;
+      }
+      if (getClass() != obj.getClass()) {
+        return false;
+      }
+      FateTransactionFetcher other = (FateTransactionFetcher) obj;
+      return Objects.equals(getType(), other.getType());
+    }
+
+    @Override
+    public int compareTo(UpdateTask<Void> other) {
+      return this.getType().compareTo(other.getType());
+    }
+
+    @Override
+    public UpdateType getType() {
+      return UpdateType.FATE;
+    }
+
+    @Override
+    public Void getResource() {
+      return null;
+    }
+
+    @Override
+    public String getFailureMessage() {
+      return "Error fetching fate transaction details";
+    }
+  }
+
   class ConfiguredCompactionResourceGroupFetcher implements UpdateTask<Void> {
 
     private final SystemInformation summary;
@@ -486,6 +559,9 @@ public String getFailureMessage() {
   private final Cache<ServerId,MetricResponse> allMetrics;
   private final Cache<ServerId,Boolean> retainedProblemServers;
   private final AtomicReference<SystemInformation> summaryRef = new AtomicReference<>();
+  private final ReadOnlyFateStore<Fate> readOnlyMFS;
+  private final ReadOnlyFateStore<Fate> readOnlyUFS;
+  private final Map<FateInstanceType,ReadOnlyFateStore<Fate>> stores;
   private final TabletMetadataFilter noLocation = new NoCurrentLocationFilter();
 
   public InformationFetcher(ServerContext ctx, Supplier<Long> connectionCount) {
@@ -495,6 +571,13 @@ public InformationFetcher(ServerContext ctx, Supplier<Long> connectionCount) {
         .expireAfterWrite(Duration.ofMinutes(10)).evictionListener(this::onRemoval).build();
     this.retainedProblemServers = Caffeine.newBuilder().executor(pool)
         .scheduler(Scheduler.systemScheduler()).expireAfterWrite(Duration.ofMinutes(10)).build();
+    try {
+      this.readOnlyMFS = new MetaFateStore<>(ctx.getZooSession(), null, null);
+    } catch (KeeperException | InterruptedException e) {
+      throw new RuntimeException("Exception creating MetaFateStore", e);
+    }
+    this.readOnlyUFS = new UserFateStore<>(ctx, SystemTables.FATE.tableName(), null, null);
+    this.stores = Map.of(FateInstanceType.META, readOnlyMFS, FateInstanceType.USER, readOnlyUFS);
   }
 
   public void newConnectionEvent() {
@@ -659,9 +742,16 @@ public void run() {
 
       final UpdateTasks futures = new UpdateTasks();
       final SystemInformation summary = new SystemInformation(allMetrics, this.ctx);
+
+      // Fetch set of registered compactors
       Set<ServerId> compactors = this.ctx.instanceOperations().getServers(Type.COMPACTOR);
       summary.processExternalCompactionInventory(compactors);
 
+      // Fetch Fate transaction information
+      FateTransactionFetcher fateFetcher = new FateTransactionFetcher(summary);
+      Future<?> fff = this.pool.submit(fateFetcher);
+      futures.add(new UpdateTaskFuture(fff, fateFetcher));
+
       // Fetch metrics from the other server processes. This
       // makes an RPC call to AbstractServer.getMetrics
       for (ServerId.Type type : ServerId.Type.values()) {
@@ -674,7 +764,6 @@ public void run() {
           futures.add(new UpdateTaskFuture(mff, mf));
         }
       }
-      ThreadPools.resizePool(pool, () -> Math.max(20, (futures.size() / 20)), poolName);
 
       // Fetch external compaction information from the Compactors
       RunningCompactionFetcher rcf = new RunningCompactionFetcher(summary, pool);
@@ -692,6 +781,8 @@ public void run() {
       Future<?> f = this.pool.submit(r);
       futures.add(new UpdateTaskFuture(f, r));
 
+      ThreadPools.resizePool(pool, () -> Math.max(20, (futures.size() / 20)), poolName);
+
       final long monitorFetchTimeout =
           ctx.getConfiguration().getTimeInMillis(Property.MONITOR_FETCH_TIMEOUT);
       final long allFuturesAdded = NanoTime.now();

server/monitor/src/main/java/org/apache/accumulo/monitor/next/SystemInformation.java

@@ -67,6 +67,10 @@
 import org.apache.accumulo.core.data.TabletId;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.dataImpl.TabletIdImpl;
+import org.apache.accumulo.core.fate.AdminUtil.TransactionStatus;
+import org.apache.accumulo.core.fate.Fate.FateOperation;
+import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.fate.ReadOnlyFateStore.TStatus;
 import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
 import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
 import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
@@ -457,6 +461,10 @@ public List<TabletNeedingRecovery> getTabletsNeedingRecovery() {
     }
   }
 
+  public record FateTransaction(FateInstanceType type, FateOperation op, String id, TStatus status,
+      long created, List<String> heldLocks, List<String> waitingLocks, String lockRange) {
+  }
+
   private static final Logger LOG = LoggerFactory.getLogger(SystemInformation.class);
 
   private final DistributionStatisticConfig DSC =
@@ -524,6 +532,8 @@ public List<TabletNeedingRecovery> getTabletsNeedingRecovery() {
 
   private final Set<String> configuredCompactionResourceGroups = ConcurrentHashMap.newKeySet();
 
+  private final List<FateTransaction> fateTransactions = new ArrayList<>();
+
   private final AtomicLong timestamp = new AtomicLong(0);
   private final EnumMap<ServerId.Type,Status> componentStatuses =
       new EnumMap<>(ServerId.Type.class);
@@ -570,6 +580,7 @@ public void clear() {
     componentStatuses.clear();
     managerGoalState = null;
     serverMetricsView.clear();
+    fateTransactions.clear();
   }
 
   public void addMessage(MessagePriority pri, MessageCategory cat, String msg) {
@@ -887,6 +898,14 @@ public void processTabletInformation(TableId tableId, String tableName, TabletIn
     }
   }
 
+  public void processFateTransactions(List<TransactionStatus> transactions) {
+    transactions.forEach(t -> {
+      fateTransactions.add(new FateTransaction(t.getInstanceType(), t.getFateOp(),
+          t.getFateId().getTxUUIDStr(), t.getStatus(), t.getTimeCreated(), t.getHeldLocks(),
+          t.getWaitingLocks(), t.getLockRange().toString()));
+    });
+  }
+
   public void processError(ServerId server) {
     problemHosts.add(server);
   }
@@ -1352,6 +1371,10 @@ public Map<MessagePriority,Map<MessageCategory,Set<String>>> getMessages() {
     return this.messages;
   }
 
+  public List<FateTransaction> getFateTransactions() {
+    return this.fateTransactions;
+  }
+
   public long getTimestamp() {
     return this.timestamp.get();
   }

server/monitor/src/main/java/org/apache/accumulo/monitor/view/WebViews.java

@@ -346,6 +346,24 @@ public Map<String,Object> getBulkImports() {
     return model;
   }
 
+  /**
+   * Returns the garbage collector template
+   *
+   * @return GC model
+   */
+  @GET
+  @Path("fate")
+  @Template(name = "/default.ftl")
+  public Map<String,Object> getFate() {
+
+    Map<String,Object> model = getModel();
+    model.put("title", "Fate Transaction Details");
+    model.put("template", "fate.ftl");
+    model.put("js", "fate.js");
+
+    return model;
+  }
+
   /**
    * Returns the garbage collector template
    *

server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/js/fate.js

@@ -0,0 +1,107 @@
+/*
+ * 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.
+ */
+"use strict";
+
+const fateHtmlTable = '#fateTable';
+
+var dataTableRef;
+
+function getTableData() {
+  return getStoredArray(FATE);
+}
+
+function createDataTable() {
+  $(fateHtmlTable).find('thead').remove();
+  $(fateHtmlTable).find('tbody').remove();
+  dataTableRef = $(fateHtmlTable).DataTable({
+    "autoWidth": false,
+    "ajax": function (data, callback) {
+      callback({
+        data: getTableData()
+      });
+    },
+    "stateSave": true,
+    "colReorder": true,
+    "columnDefs": [{
+      targets: '_all',
+      defaultContent: '-'
+    }],
+    "columns": [{
+        "data": "type",
+        "title": "Type"
+      },
+      {
+        "data": "op",
+        "title": "Operation"
+      },
+      {
+        "data": "id",
+        "title": "Id"
+      },
+      {
+        "data": "status",
+        "title": "State"
+      },
+      {
+        "data": "created",
+        "title": "Created",
+        "render": function (data, type, row) {
+          if (type === 'display') data = dateFormat(data);
+          return data;
+        }
+      },
+      {
+        "data": "created",
+        "title": "Age",
+        "render": function (data, type, row) {
+          var dur = Date.now() - data;
+          if (type === 'display') dur = timeDuration(dur);
+          return dur;
+        }
+      },
+      {
+        "data": "heldLocks",
+        "title": "Locks Held"
+      },
+      {
+        "data": "waitingLocks",
+        "title": "Locks Waiting On"
+      },
+      {
+        "data": "lockRange",
+        "title": "Lock Range"
+      }
+    ]
+  });
+}
+
+function refresh() {
+  return getFate().then(function () {
+    if (dataTableRef) {
+      ajaxReloadTable(dataTableRef);
+    }
+  });
+}
+
+
+$(function () {
+  getFate().then(function () {
+    createDataTable();
+  });
+});