dcache-bulk: add admin command and query to reset all requests with f…

…ailed targets

Motivation:

There exists an admin command to reset requests to be
rerun.  The command takes various options based on
the fields in the bulk request table.   However,
it is currently not possible to rerun only the
requests with failed targets that have completed.
This may sometimes be useful to the admin.
(Note that for safety we do not automatically
do this on service restart/reload, where only
incomplete requests are reset).

Modifications:

Add the command and a few extra pieces of
support in the dao/store layers.

Result:

We can now retry completed requests which
have failed targets.

Target: master
Request: 9.2
Patch: https://rb.dcache.org/r/14117/
Requires-notes: yes
Acked-by: Tigran

modules/dcache-bulk/src/main/java/org/dcache/services/bulk/BulkServiceCommands.java

@@ -1062,6 +1062,26 @@ public String call() throws Exception {
         }
     }
 
+    @Command(name = "request retry failed",
+          hint = "Retry only those requests which have failed targets.",
+          description = "Calls reset on the requests matching this criterion.")
+    class RequestRetryFailed implements Callable<String> {
+
+        @Override
+        public String call() throws Exception {
+            executor.submit(()-> {
+                try {
+                    int count = requestStore.retryFailed();
+                    LOGGER.info("{} requests with failed targets have been reset.", count);
+                } catch (BulkStorageException e) {
+                    LOGGER.error("could not reset failed: {}.", e.toString());
+                }
+            });
+
+            return "Resetting requests with failed targets.";
+        }
+    }
+
     @Command(name = "request submit",
           hint = "Launch a bulk request.",
           description = "Command-line version of the RESTful request.")

modules/dcache-bulk/src/main/java/org/dcache/services/bulk/store/BulkRequestStore.java

@@ -277,6 +277,13 @@ List<BulkRequest> next(Optional<String> sortedBy, Optional<Boolean> reverse, lon
      */
     void reset(String uid) throws BulkStorageException;
 
+    /**
+     * Retry all requests that have FAILED targets.
+     *
+     * @throws BulkStorageException
+     */
+    int retryFailed() throws BulkStorageException;
+
     /**
      * Store the request.
      *

modules/dcache-bulk/src/main/java/org/dcache/services/bulk/store/jdbc/request/JdbcBulkRequestStore.java

@@ -71,6 +71,7 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 import static org.dcache.services.bulk.util.BulkRequestTarget.PLACEHOLDER_PNFSID;
 import static org.dcache.services.bulk.util.BulkRequestTarget.ROOT_REQUEST_PATH;
 import static org.dcache.services.bulk.util.BulkRequestTarget.State.CREATED;
+import static org.dcache.services.bulk.util.BulkRequestTarget.State.FAILED;
 
 import com.google.common.base.Strings;
 import com.google.common.base.Throwables;
@@ -93,6 +94,7 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import javax.security.auth.Subject;
@@ -120,7 +122,6 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 import org.dcache.services.bulk.util.BulkRequestFilter;
 import org.dcache.services.bulk.util.BulkRequestTarget;
 import org.dcache.services.bulk.util.BulkRequestTarget.PID;
-import org.dcache.services.bulk.util.BulkRequestTarget.State;
 import org.dcache.services.bulk.util.BulkRequestTargetBuilder;
 import org.dcache.services.bulk.util.BulkServiceStatistics;
 import org.dcache.vehicles.FileAttributes;
@@ -547,6 +548,17 @@ public void reset(String uid) throws BulkStorageException {
         }
     }
 
+    @Override
+    public int retryFailed() throws BulkStorageException {
+        AtomicInteger count = new AtomicInteger(0);
+        List<String> uids = requestTargetDao.getRequestsOfFailed();
+        for (String uid: uids) {
+            reset(uid);
+            count.incrementAndGet();
+        }
+        return count.get();
+    }
+
     @Required
     public void setArchiveDao(JdbcBulkArchiveDao archiveDao) {
         this.archiveDao = archiveDao;
@@ -783,7 +795,7 @@ private String checkRequestPermissions(Subject subject, String uid)
 
     private void conditionallyClearTerminalRequest(BulkRequest stored) {
         Long rid = stored.getId();
-        if (requestTargetDao.count(requestTargetDao.where().rid(rid).state(State.FAILED)) > 0) {
+        if (requestTargetDao.count(requestTargetDao.where().rid(rid).state(FAILED)) > 0) {
             if (stored.isClearOnFailure()) {
                 clear(stored.getUid());
             }

modules/dcache-bulk/src/main/java/org/dcache/services/bulk/store/jdbc/rtarget/JdbcRequestTargetDao.java

@@ -117,6 +117,10 @@ static class TargetPlaceholder {
 
     static final String JOINED_TABLE_NAMES_FOR_SELECT = SECONDARY_TABLE_NAME + ", " + TABLE_NAME;
 
+    static final String UIDS_OF_FAILED
+          = "SELECT r.uid FROM bulk_request r WHERE r.status = 'COMPLETED' AND EXISTS " +
+          "(SELECT * FROM request_target t WHERE r.id = t.rid AND t.state = 'FAILED')";
+
     static final ParameterizedPreparedStatementSetter<TargetPlaceholder> SETTER = (ps, target) -> {
         Instant now = Instant.now();
         ps.setInt(1, PID.INITIAL.ordinal());
@@ -164,6 +168,10 @@ public List<BulkRequestTarget> get(JdbcRequestTargetCriterion criterion, int lim
               this, criterion.isJoined() ? this::toFullRequestTarget : this::toRequestTarget);
     }
 
+    public List<String> getRequestsOfFailed() {
+        return getJdbcTemplate().queryForList(UIDS_OF_FAILED, String.class);
+    }
+
     public Optional<KeyHolder> insert(JdbcRequestTargetUpdate update) {
         return utils.insert(update, TABLE_NAME, this);
     }