drv: add dCache admin command to show panding requests

easy way to match dcache to cta requests

Author: kofemann

src/main/java/org/dcache/nearline/cta/CtaNearlineStorage.java

@@ -9,6 +9,7 @@
 import com.google.common.net.HostAndPort;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import diskCacheV111.util.CacheException;
+import dmg.util.command.Command;
 import io.grpc.ChannelCredentials;
 import io.grpc.ConnectivityState;
 import io.grpc.Deadline;
@@ -27,13 +28,15 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
-
+import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
-import ch.cern.cta.rpc.CtaRpcGrpc.CtaRpcBlockingStub;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import ch.cern.cta.rpc.CtaRpcGrpc.CtaRpcBlockingStub;
 import org.dcache.namespace.FileAttribute;
 import org.dcache.nearline.cta.xrootd.DataMover;
 import org.dcache.pool.nearline.spi.FlushRequest;
@@ -249,7 +252,7 @@ public void completed(Set<URI> uris) {
 
         var cancelRequest = ctaRequestFactory.getAbortStoreRequest(ar, response);
 
-        pendingRequests.put(id, new PendingRequest(r) {
+        pendingRequests.put(id, new PendingRequest(r, response.getRequestObjectstoreId(), PendingRequest.Type.FLUSH) {
                     @Override
                     public void cancel() {
                         try {
@@ -318,7 +321,7 @@ public void completed(Set<Checksum> checksums) {
                 );
 
                 var cancelRequest = ctaRequestFactory.getAbortStageRequest(rr, response);
-                pendingRequests.put(id, new PendingRequest(r) {
+                pendingRequests.put(id, new PendingRequest(r, response.getRequestObjectstoreId(), PendingRequest.Type.STAGE) {
                             @Override
                             public void cancel() {
                                 // on cancel send the request to CTA; on success cancel the requests
@@ -543,4 +546,33 @@ private CacheException asCacheException(Throwable e) {
     private URI createZeroFileUri(FileAttributes attrs) {
         return URI.create(type + "://" + name + "/" + attrs.getPnfsId() + "?archiveid=*");
     }
+
+    @Command(name="show requests")
+    public class ShowRequestsCommand implements Callable<String> {
+        @Override
+        public String call() {
+            if (pendingRequests.isEmpty()) {
+                return "No pending requests";
+            }
+
+            StringBuilder sb = new StringBuilder();
+            sb.append("Pending requests:\n");
+
+            pendingRequests.entrySet().stream()
+                    .collect(Collectors.groupingBy(e -> e.getValue().getAction()))
+                    .forEach((type, entries) -> {
+                        sb.append(type).append(":\n");
+                        sb.append(" count: ").append(entries.size()).append("\n");
+                        for (var entry : entries) {
+                            sb.append("   ")
+                                    .append(entry.getKey())
+                                    .append(" -> ")
+                                    .append(entry.getValue().getCtaRequestId()).append("\n");
+                        }
+                        sb.append("\n");
+                    });
+
+            return sb.toString();
+        }
+    }
 }

src/main/java/org/dcache/nearline/cta/PendingRequest.java

@@ -10,6 +10,21 @@
  */
 public class PendingRequest {
 
+    /**
+     * The type of the request.
+     * <p>
+     * STAGE - request to stage file from nearline storage to online storage.
+     * FLUSH - request to flush file from online storage to nearline storage.
+     * CANCEL - request to cancel staged file.
+     * DELETE - request to delete archived file.
+     */
+    public enum Type {
+        STAGE, // request to stage file
+        FLUSH, // request to flush file
+        CANCEL, // request to cancel staged file
+        DELETE // request to delete arcived file
+    }
+
     /**
      * Point on the time-line when request was submitted into pending queue.
      */
@@ -20,9 +35,28 @@ public class PendingRequest {
      */
     private final NearlineRequest request;
 
-    public PendingRequest(NearlineRequest request) {
+    /**
+     * CTA scheduler request ID (aka object store ID).
+     */
+    private final String ctaRequestId;
+
+    /**
+     * The type of the request.
+     */
+    private Type action;
+
+    /**
+     * Constructs a new pending request.
+     *
+     * @param request the nearline request.
+     * @param ctaRequestId the CTA request ID (aka object store ID).
+     * @param action the type of the request.
+     */
+    public PendingRequest(NearlineRequest request, String ctaRequestId, Type action) {
         this.submissionTime = Instant.now();
         this.request = request;
+        this.ctaRequestId  = ctaRequestId;
+        this.action = action;
     }
 
     /**
@@ -52,6 +86,24 @@ public UUID getRequestId() {
         return request.getId();
     }
 
+    /**
+     * Returns CTA request ID (aka object store ID).
+     *
+     * @return CTA request ID.
+     */
+    public String getCtaRequestId() {
+        return ctaRequestId;
+    }
+
+    /**
+     * Returns the request type.
+     *
+     * @return the request type.
+     */
+    public Type getAction() {
+        return action;
+    }
+
     public void cancel() {
         request.failed(new CancellationException("Canceled by dCache"));
     }

src/test/java/org/dcache/nearline/cta/xrootd/DataServerHandlerTest.java

@@ -531,7 +531,7 @@ private StageRequest mockedStageRequest() throws IOException {
 
         when(request.getReplicaUri()).thenReturn(f.toURI());
 
-        requests.put("0000C9B4E3768770452E8B1B8E0232584872", new PendingRequest(request));
+        requests.put("0000C9B4E3768770452E8B1B8E0232584872", new PendingRequest(request, "stage", PendingRequest.Type.STAGE));
 
         waitForComplete = new CompletableFuture<>();
 
@@ -573,7 +573,7 @@ private FlushRequest mockedFlushRequest() throws IOException {
 
         when(request.getReplicaUri()).thenReturn(f.toURI());
 
-        requests.put("0000C9B4E3768770452E8B1B8E0232584872", new PendingRequest(request));
+        requests.put("0000C9B4E3768770452E8B1B8E0232584872", new PendingRequest(request, "flush", PendingRequest.Type.FLUSH));
 
         return request;
     }