Merge pull request #2269 from HubSpot/retry-tasks-with-new-deploys

Retry failed tasks with new deploys

Author: WH77

SingularityService/src/main/java/com/hubspot/singularity/scheduler/SingularityScheduler.java

@@ -428,7 +428,7 @@ private void handlePendingRequestsForDeployKey(
     );
 
     if (!isRequestActive(maybeRequest)) {
-      LOG.debug(
+      LOG.info(
         "Pending request {} was obsolete (request {})",
         requestId,
         SingularityRequestWithState.getRequestState(maybeRequest)
@@ -475,7 +475,7 @@ private void handlePendingRequestsForDeployKey(
           maybeRequestDeployState
         )
       ) {
-        LOG.debug(
+        LOG.info(
           "Pending request {} was obsolete (request {})",
           pendingRequest,
           SingularityRequestWithState.getRequestState(maybeRequest)
@@ -1014,12 +1014,28 @@ private Optional<PendingType> handleCompletedTaskWithStatistics(
     );
 
     if (!isDeployInUse(requestDeployState, taskId.getDeployId(), true)) {
-      LOG.debug(
-        "Task {} completed, but it didn't match active deploy state {} - ignoring",
-        taskId.getId(),
-        requestDeployState
-      );
-      return Optional.empty();
+      // failed tasks with remaining retries cannot short circuit here
+      // isFailed is false for lost and other weird task states
+      if (
+        !state.isSuccess() &&
+        (
+          deployStatistics.getNumSequentialRetries() <
+          request.getNumRetriesOnFailure().orElse(0)
+        )
+      ) {
+        LOG.info(
+          "Task {} failed, but it didn't match active deploy state {} - may retry with current deploy state",
+          taskId.getId(),
+          requestDeployState
+        );
+      } else {
+        LOG.info(
+          "Task {} completed, but it didn't match active deploy state {} - ignoring",
+          taskId.getId(),
+          requestDeployState
+        );
+        return Optional.empty();
+      }
     }
 
     if (

SingularityService/src/test/java/com/hubspot/singularity/scheduler/SingularitySchedulerTest.java

@@ -1255,6 +1255,98 @@ public void testRetriesWithOverrides() {
     Assertions.assertTrue(taskManager.getActiveTaskIds().isEmpty());
   }
 
+  @Test
+  public void testRetriesWithNewDeploys() {
+    SingularityRequestBuilder bldr = new SingularityRequestBuilder(
+      requestId,
+      RequestType.RUN_ONCE
+    );
+    request = bldr.setNumRetriesOnFailure(Optional.of(2)).build();
+    saveRequest(request);
+
+    deployResource.deploy(
+      new SingularityDeployRequest(
+        new SingularityDeployBuilder(requestId, "d1")
+          .setCommand(Optional.of("cmd"))
+          .build(),
+        Optional.empty(),
+        Optional.empty()
+      ),
+      singularityUser
+    );
+
+    scheduler.drainPendingQueue();
+    deployChecker.checkDeploys();
+    scheduler.drainPendingQueue();
+    resourceOffers();
+
+    Assertions.assertEquals(1, taskManager.getActiveTaskIds().size());
+
+    deployResource.deploy(
+      new SingularityDeployRequest(
+        new SingularityDeployBuilder(requestId, "d2")
+          .setCommand(Optional.of("cmd"))
+          .build(),
+        Optional.empty(),
+        Optional.empty()
+      ),
+      singularityUser
+    );
+
+    statusUpdate(taskManager.getActiveTasks().get(0), TaskState.TASK_LOST);
+
+    scheduler.drainPendingQueue();
+    deployChecker.checkDeploys();
+    scheduler.drainPendingQueue();
+    resourceOffers();
+
+    Assertions.assertEquals(1, taskManager.getActiveTaskIds().size());
+    Assertions.assertEquals("d2", taskManager.getActiveTaskIds().get(0).getDeployId());
+
+    deployResource.deploy(
+      new SingularityDeployRequest(
+        new SingularityDeployBuilder(requestId, "d3")
+          .setCommand(Optional.of("cmd"))
+          .build(),
+        Optional.empty(),
+        Optional.empty()
+      ),
+      singularityUser
+    );
+
+    statusUpdate(taskManager.getActiveTasks().get(0), TaskState.TASK_LOST);
+
+    scheduler.drainPendingQueue();
+    deployChecker.checkDeploys();
+    scheduler.drainPendingQueue();
+    resourceOffers();
+
+    Assertions.assertEquals(1, taskManager.getActiveTaskIds().size());
+    Assertions.assertEquals("d3", taskManager.getActiveTaskIds().get(0).getDeployId());
+
+    deployResource.deploy(
+      new SingularityDeployRequest(
+        new SingularityDeployBuilder(requestId, "d4")
+          .setCommand(Optional.of("cmd"))
+          .build(),
+        Optional.empty(),
+        Optional.empty()
+      ),
+      singularityUser
+    );
+
+    statusUpdate(taskManager.getActiveTasks().get(0), TaskState.TASK_LOST);
+
+    scheduler.drainPendingQueue();
+    deployChecker.checkDeploys();
+    scheduler.drainPendingQueue();
+    resourceOffers();
+
+    // TODO - new deploys have new deploy statistics, so we lose track of the # of retries
+    Assertions.assertEquals(1, taskManager.getActiveTaskIds().size());
+    Assertions.assertEquals("d4", taskManager.getActiveTaskIds().get(0).getDeployId());
+  }
+
   /* @Test
   public void testCooldownAfterSequentialFailures() {
     initRequest();