drv: keep track of active requests

Motivation:
to handle dCache pool restarts driver should keep track of submitted
requests.

Modification:
added BerkeleyDB based db to keep track of submitted requests.

Result:
on start, the driver can find out which requests already submitted to CTA.

Author: kofemann

README.md

@@ -109,6 +109,7 @@ queue define class -expire=0 -pending=0 -total=0 -open <hsmType> *
 | io-port                  | The TCP port offered by dCache for IO by CTA                   |       no | -          |
 | use-dio                  | Use Direct-I/O                                                 |       no | `false`    |
 | restore-success-on-close | **obsolete**                                                   |        - | -          |
+| cleanup-journal          | The location of the journal file used for cleanup on start     |       no | -          |
 
 ### Load balancing and failover
 

pom.xml

@@ -135,6 +135,13 @@
             <version>2.2</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.sleepycat</groupId>
+            <artifactId>je</artifactId>
+            <version>7.5.11</version>
+            <!-- direct dependency in dCache -->
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -0,0 +1,26 @@
+package org.dcache.nearline.cta;
+
+import ch.cern.cta.rpc.CtaRpcGrpc;
+import ch.cern.cta.rpc.Request;
+
+import java.util.function.BiFunction;
+
+/**
+ * Interface to persistent storage for pending requests.
+ * <p>
+ * The new requests are added to the list and removed on completion.
+ * On submit, pool try to add the request to the list. And if already exist,
+ * the pending request will cancel (delete) in CTA before submission.
+ */
+public interface CleanupJournal extends AutoCloseable {
+    void cleanup(CtaRpcGrpc.CtaRpcBlockingStub cta);
+
+    void cleanup(CtaRpcGrpc.CtaRpcBlockingStub cta, BiFunction<String, Request, Boolean> function);
+
+    void put(String pnfsid, Request archiveResponse);
+
+    void remove(String pnfsid);
+
+    @Override
+    void close();
+}

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

@@ -58,6 +58,11 @@ public class CtaNearlineStorage implements NearlineStorage {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(CtaNearlineStorage.class);
 
+    /**
+     * Unicode character for screaming face.
+     */
+    public static final String MWAI_SCREAMING = Character.toString(0x1F631);
+
     public static final String CTA_INSTANCE = "cta-instance-name";
     public static final String CTA_ENDPOINT = "cta-frontend-addr";
     public static final String CTA_USER = "cta-user";
@@ -68,6 +73,7 @@ public class CtaNearlineStorage implements NearlineStorage {
     public static final String IO_PORT = "io-port";
     public static final String CTA_REQUEST_TIMEOUT = "cta-frontend-timeout";
     public static final String DIO = "use-dio";
+    public static final String CTA_REQUEST_JOURNAL = "cleanup-journal";
 
     protected final String type;
     protected final String name;
@@ -166,6 +172,11 @@ enum Action {
      */
     private final RequestCounters<Action> requestCounters = new RequestCounters<>("CTA gRPC Requests counters");
 
+    /**
+     * Pending request to CTA.
+     */
+    private CleanupJournal pendingRequestsTracker;
+
     public CtaNearlineStorage(String type, String name) {
 
         Objects.requireNonNull(type, "HSM type is not provided");
@@ -251,18 +262,21 @@ protected FlushRequest delegate() {
             @Override
             public void failed(Exception e) {
                 pendingRequests.remove(id);
+                pendingRequestsTracker.remove(id);
                 super.failed(e);
             }
 
             @Override
             public void failed(int i, String s) {
                 pendingRequests.remove(id);
+                pendingRequestsTracker.remove(id);
                 super.failed(i, s);
             }
 
             @Override
             public void completed(Set<URI> uris) {
                 pendingRequests.remove(id);
+                pendingRequestsTracker.remove(id);
                 super.completed(uris);
             }
         };
@@ -280,7 +294,9 @@ public void completed(Set<URI> uris) {
 
         var cancelRequest = ctaRequestFactory.getAbortStoreRequest(ar, response);
 
+        pendingRequestsTracker.put(id, cancelRequest);
         pendingRequests.put(id, new PendingRequest(r, response.getRequestObjectstoreId(), PendingRequest.Type.FLUSH) {
+
                     @Override
                     public void cancel() {
                         try {
@@ -313,18 +329,21 @@ protected StageRequest delegate() {
                 @Override
                 public void failed(Exception e) {
                     pendingRequests.remove(id);
+                    pendingRequestsTracker.remove(id);
                     super.failed(e);
                 }
 
                 @Override
                 public void failed(int i, String s) {
                     pendingRequests.remove(id);
+                    pendingRequestsTracker.remove(id);
                     super.failed(i, s);
                 }
 
                 @Override
                 public void completed(Set<Checksum> checksums) {
                     pendingRequests.remove(id);
+                    pendingRequestsTracker.remove(id);
                     super.completed(checksums);
                 }
             };
@@ -349,7 +368,9 @@ public void completed(Set<Checksum> checksums) {
                 );
 
                 var cancelRequest = ctaRequestFactory.getAbortStageRequest(rr, response);
+                pendingRequestsTracker.put(id, cancelRequest);
                 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
@@ -442,6 +463,7 @@ public void configure(Map<String, String> properties) throws IllegalArgumentExce
         String user = properties.get(CTA_USER);
         String group = properties.get(CTA_GROUP);
         String timeoutString = properties.get(CTA_REQUEST_TIMEOUT);
+        String journal = properties.get(CTA_REQUEST_JOURNAL);
 
         checkArgument(instance != null, "dCache instance name is not set.");
         checkArgument(endpoint != null, "CTA frontend is not set.");
@@ -485,6 +507,12 @@ public void configure(Map<String, String> properties) throws IllegalArgumentExce
         }
 
         dio = Boolean.parseBoolean(properties.getOrDefault(DIO, "false"));
+
+        if (journal != null) {
+            pendingRequestsTracker = new DigitalMwai(journal);
+        } else {
+            pendingRequestsTracker = new NopCleanupJournal();
+        }
     }
 
     @Override
@@ -526,6 +554,12 @@ public void start() {
                     LOGGER.info("Connected to CTA frontend");
                 }
         );
+
+        pendingRequestsTracker.cleanup(cta, (p, r) -> {
+            LOGGER.warn("[Mwai {}]: Found pending entry for {} : {}", MWAI_SCREAMING, p, r.getNotification().getFile().getRequestObjectstoreId());
+            return Boolean.TRUE;
+        });
+
         ctaRequestFactory = new RequestsFactory(instanceName, ctaUser, ctaGroup, dataMover);
     }
 

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

@@ -0,0 +1,176 @@
+package org.dcache.nearline.cta;
+
+
+import static io.grpc.Status.Code.NOT_FOUND;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import ch.cern.cta.rpc.CtaRpcGrpc.CtaRpcBlockingStub;
+import ch.cern.cta.rpc.Request;
+import com.sleepycat.je.Cursor;
+import com.sleepycat.je.CursorConfig;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseConfig;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.EnvironmentConfig;
+import com.sleepycat.je.OperationStatus;
+
+import com.sleepycat.je.Transaction;
+import io.grpc.StatusRuntimeException;
+
+import java.io.File;
+import java.util.function.BiFunction;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class provided a persistent storage for pending requests.
+ * <p>
+ * The new requests are added to the list and removed on completion.
+ * On submit, pool try to add the request to the list. And if already exist,
+ * the pending request will cancel (delete) in CTA before submission.
+ */
+public class DigitalMwai implements CleanupJournal {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DigitalMwai.class);
+
+    private static final String CTA_PENDING_REQUEST = "archive-requests";
+
+    private final CursorConfig config = new CursorConfig();
+    private final Environment env;
+    private final DatabaseConfig dbConfig;
+    private Database pendingRequests;
+
+
+    /**
+     * Create a new instance of {@link DigitalMwai}.
+     *
+     * @param path the path to the directory where the persistent storage will be created.
+     */
+    DigitalMwai(String path) {
+
+        File dir = new File(path);
+
+        if (!dir.exists()) {
+            dir.mkdirs();
+        }
+
+        EnvironmentConfig envConfig = new EnvironmentConfig();
+        envConfig.setAllowCreate(true);
+        envConfig.setTransactional(true);
+        envConfig.setReadOnly(false);
+
+        env = new Environment(dir, envConfig);
+
+        dbConfig = new DatabaseConfig();
+        dbConfig.setTransactional(true);
+        dbConfig.setAllowCreate(true);
+        dbConfig.setReadOnly(false);
+
+        pendingRequests = env.openDatabase(null, CTA_PENDING_REQUEST, dbConfig);
+    }
+
+
+    @Override
+    public void cleanup(CtaRpcBlockingStub cta) {
+
+        Transaction tx = env.beginTransaction(null, null);
+        try (Cursor cursor = pendingRequests.openCursor(tx, config)) {
+
+            DatabaseEntry key = new DatabaseEntry();
+            DatabaseEntry data = new DatabaseEntry();
+
+            /*
+             * Remove all pending requests that are submitted before restart.
+             */
+            while (cursor.getNext(key, data, null) == OperationStatus.SUCCESS) {
+                boolean canDelete = false;
+                try {
+                    var cancelRequest = Request.parseFrom(data.getData());
+                    switch (cancelRequest.getNotification().getWf().getEvent()) {
+                        case CLOSEW:
+                            cta.delete(cancelRequest);
+                            break;
+                        case PREPARE:
+                            cta.cancelRetrieve(cancelRequest);
+                            break;
+                        default:
+                            LOGGER.warn("Unexpected request type: {}", cancelRequest.getNotification().getWf().getEvent());
+                    }
+                    canDelete = true;
+                } catch (StatusRuntimeException e) {
+                    if (e.getStatus().getCode() == NOT_FOUND) {
+                        canDelete = true;
+                    } else {
+                        LOGGER.error("Failed to remove pending request: {}", e.getStatus());
+                    }
+                } catch (Exception e) {
+                    LOGGER.error("Unexpected error while removing pending request: {}", e.getMessage());
+                }
+
+                if (canDelete) {
+                    cursor.delete();
+                }
+            }
+        } finally {
+            tx.commit();
+        }
+    }
+
+    @Override
+    public void cleanup(CtaRpcBlockingStub cta, BiFunction<String, Request, Boolean> consumer) {
+
+        Transaction tx = env.beginTransaction(null, null);
+        try (Cursor cursor = pendingRequests.openCursor(tx, config)) {
+
+            DatabaseEntry key = new DatabaseEntry();
+            DatabaseEntry data = new DatabaseEntry();
+
+            /*
+             * Remove all pending requests that are submitted before restart.
+             */
+            while (cursor.getNext(key, data, null) == OperationStatus.SUCCESS) {
+
+                try {
+                    String id = new String(key.getData(), UTF_8);
+                    var cancelRequest = Request.parseFrom(data.getData());
+                    if (consumer.apply(id, cancelRequest)) {
+                        cursor.delete();
+                    }
+                } catch (Exception e) {
+                    LOGGER.error("Unexpected error while removing pending request: {}", e.getMessage());
+                }
+            }
+        } finally {
+            tx.commit();
+        }
+    }
+
+    @Override
+    public void put(String pnfsid, Request archiveResponse) {
+        DatabaseEntry key = new DatabaseEntry(pnfsid.getBytes(UTF_8));
+        DatabaseEntry data = new DatabaseEntry(archiveResponse.toByteArray());
+
+        var status = pendingRequests.putNoOverwrite(null, key, data);
+        if (status != OperationStatus.SUCCESS) {
+            throw new IllegalStateException("Unexpected status: " + status);
+        }
+    }
+
+    @Override
+    public void remove(String pnfsid) {
+        DatabaseEntry key = new DatabaseEntry(pnfsid.getBytes(UTF_8));
+        var status = pendingRequests.delete(null, key);
+        if (status != OperationStatus.SUCCESS) {
+            LOGGER.error("Failed to remove persistent for {} entry: {}", pnfsid, status);
+        }
+    }
+
+    @Override
+    public void close() {
+        pendingRequests.close();
+        env.close();
+    }
+
+}

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

@@ -0,0 +1,34 @@
+package org.dcache.nearline.cta;
+
+import ch.cern.cta.rpc.CtaRpcGrpc;
+import ch.cern.cta.rpc.Request;
+
+import java.util.function.BiFunction;
+
+public class NopCleanupJournal implements CleanupJournal {
+
+    @Override
+    public void cleanup(CtaRpcGrpc.CtaRpcBlockingStub cta) {
+        // NOP
+    }
+
+    @Override
+    public void cleanup(CtaRpcGrpc.CtaRpcBlockingStub cta, BiFunction<String, Request, Boolean> function) {
+        // NOP
+    }
+
+    @Override
+    public void put(String pnfsid, Request archiveResponse) {
+        // NOP
+    }
+
+    @Override
+    public void remove(String pnfsid) {
+        // NOP
+    }
+
+    @Override
+    public void close() {
+        // nop
+    }
+}

src/main/java/org/dcache/nearline/cta/xrootd/DataMover.java

@@ -231,4 +231,8 @@ public Transport getTransport(String id, long ctaArchiveId) {
               .setReportUrl(reporterUrl)
               .build();
     }
+
+    public String getId() {
+        return url;
+    }
 }