WX-1625 Refactor ahead of quota retry

backend/src/main/scala/cromwell/backend/standard/StandardAsyncExecutionActor.scala

@@ -24,6 +24,7 @@
 import cromwell.backend.async.AsyncBackendJobExecutionActor._
 import cromwell.backend.async._
 import cromwell.backend.standard.StandardAdHocValue._
+import cromwell.backend.standard.retry.memory.MemoryRetryResult
 import cromwell.backend.validation._
 import cromwell.core.io.{AsyncIoActorClient, DefaultIoCommandBuilder, IoCommandBuilder}
 import cromwell.core.path.Path
@@ -896,6 +897,16 @@
  /**
  * Returns true if the status represents a completion.
  *
+ * Select meanings by backend:
+ * - TES:
+ * `cromwell.backend.impl.tes.Complete` derived from "state": "COMPLETE"
+ * - Life Sciences:
+ * `com.google.api.services.genomics.v2alpha1.model.Operation.getDone` is true
+ * -- AND --
+ * `com.google.api.services.genomics.v2alpha1.model.Operation#getError` is empty
+ * - GCP Batch:
+ * `com.google.cloud.batch.v1.JobStatus.State` is `SUCCEEDED`
+ *
  * @param runStatus The run status.
  * @return True if the job is done.
  */
@@ -1054,7 +1065,7 @@
  * @return The execution handle.
  */
  def retryElseFail(backendExecutionStatus: Future[ExecutionHandle],
- retryWithMoreMemory: Boolean = false
+ memoryRetry: MemoryRetryResult = MemoryRetryResult.none
  ): Future[ExecutionHandle] =
  backendExecutionStatus flatMap {
  case failedRetryableOrNonRetryable: FailedExecutionHandle =>
@@ -1069,35 +1080,48 @@
  case None => Map.empty[String, KvPair]
  }
 
- val maxRetriesNotReachedYet = previousFailedRetries < maxRetries
  failedRetryableOrNonRetryable match {
- case failed: FailedNonRetryableExecutionHandle if maxRetriesNotReachedYet =>
- (retryWithMoreMemory, memoryRetryFactor, previousMemoryMultiplier) match {
- case (true, Some(retryFactor), Some(previousMultiplier)) =>
- val nextMemoryMultiplier = previousMultiplier * retryFactor.value
- saveAttrsAndRetry(failed,
- kvsFromPreviousAttempt,
- kvsForNextAttempt,
- incFailedCount = true,
- Option(nextMemoryMultiplier)
- )
- case (true, Some(retryFactor), None) =>
- saveAttrsAndRetry(failed,
- kvsFromPreviousAttempt,
- kvsForNextAttempt,
- incFailedCount = true,
- Option(retryFactor.value)
- )
- case (_, _, _) =>
- saveAttrsAndRetry(failed, kvsFromPreviousAttempt, kvsForNextAttempt, incFailedCount = true)
- }
- case failedNonRetryable: FailedNonRetryableExecutionHandle => Future.successful(failedNonRetryable)
+ case failedNonRetryable: FailedNonRetryableExecutionHandle if previousFailedRetries < maxRetries =>
+ // The user asked us to retry finitely for them, possibly with a memory modification
+ evaluateFailureRetry(failedNonRetryable, kvsFromPreviousAttempt, kvsForNextAttempt, memoryRetry)
+ case failedNonRetryable: FailedNonRetryableExecutionHandle =>
+ // No reason to retry
+ Future.successful(failedNonRetryable)
  case failedRetryable: FailedRetryableExecutionHandle =>
+ // Retry infinitely and unconditionally (!)
  saveAttrsAndRetry(failedRetryable, kvsFromPreviousAttempt, kvsForNextAttempt, incFailedCount = false)
  }
  case _ => backendExecutionStatus
  }
 
+ private def evaluateFailureRetry(handle: FailedNonRetryableExecutionHandle,
+ kvsFromPreviousAttempt: Map[String, KvPair],
+ kvsForNextAttempt: Map[String, KvPair],
+ memoryRetry: MemoryRetryResult
+ ): Future[FailedRetryableExecutionHandle] =
+ (memoryRetry.oomDetected, memoryRetry.factor, memoryRetry.previousMultiplier) match {
+ case (true, Some(retryFactor), Some(previousMultiplier)) =>
+ // Subsequent memory retry attempt
+ val nextMemoryMultiplier = previousMultiplier * retryFactor.value
+ saveAttrsAndRetry(handle,
+ kvsFromPreviousAttempt,
+ kvsForNextAttempt,
+ incFailedCount = true,
+ Option(nextMemoryMultiplier)
+ )
+ case (true, Some(retryFactor), None) =>
+ // First memory retry attempt
+ saveAttrsAndRetry(handle,
+ kvsFromPreviousAttempt,
+ kvsForNextAttempt,
+ incFailedCount = true,
+ Option(retryFactor.value)
+ )
+ case (_, _, _) =>
+ // Not an OOM
+ saveAttrsAndRetry(handle, kvsFromPreviousAttempt, kvsForNextAttempt, incFailedCount = true)
+ }
+
  private def saveAttrsAndRetry(failedExecHandle: FailedExecutionHandle,
  kvPrev: Map[String, KvPair],
  kvNext: Map[String, KvPair],
@@ -1400,7 +1424,9 @@
  None
  )
  )
- retryElseFail(executionHandle, outOfMemoryDetected)
+ retryElseFail(executionHandle,
+ MemoryRetryResult(outOfMemoryDetected, memoryRetryFactor, previousMemoryMultiplier)
+ )
  case Success(returnCodeAsInt) if isAbort(returnCodeAsInt) =>
  Future.successful(AbortedExecutionHandle)
  case Success(returnCodeAsInt) =>
@@ -1430,7 +1456,9 @@
  None
  )
  )
- retryElseFail(executionHandle, outOfMemoryDetected)
+ retryElseFail(executionHandle,
+ MemoryRetryResult(outOfMemoryDetected, memoryRetryFactor, previousMemoryMultiplier)
+ )
  case _ =>
  val failureStatus = handleExecutionFailure(status, tryReturnCodeAsInt.toOption)
  retryElseFail(failureStatus)

backend/src/main/scala/cromwell/backend/standard/retry/memory/MemoryRetryResult.scala

@@ -0,0 +1,20 @@
+package cromwell.backend.standard.retry.memory
+
+import common.validation.Validation.MemoryRetryMultiplierRefined
+
+/**
+ * Result of evaluating an attempt as a memory-retry candidate, encapsulating instructions
+ * for configuring the next attempt if applicable.
+ *
+ * @param oomDetected Did the previous attempt OOM?
+ * @param factor User-configured factor
+ * @param previousMultiplier Multiplier used for the previous attempt
+ */
+case class MemoryRetryResult(oomDetected: Boolean,
+ factor: Option[MemoryRetryMultiplierRefined],
+ previousMultiplier: Option[Double]
+)
+
+case object MemoryRetryResult {
+ def none: MemoryRetryResult = MemoryRetryResult(oomDetected = false, None, None)
+}

supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/errors/package.scala

@@ -0,0 +1,16 @@
+package cromwell.backend.google.pipelines.common
+
+package object errors {
+
+ private def quotaMessages = List(
+ "A resource limit has delayed the operation",
+ "usage too high",
+ "no available zones",
+ "resource_exhausted",
+ "quota too low"
+ )
+
+ def isQuotaMessage(msg: String): Boolean =
+ quotaMessages.exists(msg.contains)
+
+}

supportedBackends/google/pipelines/v2beta/src/main/scala/cromwell/backend/google/pipelines/v2beta/api/request/GetRequestHandler.scala

@@ -17,6 +17,7 @@ import cromwell.backend.google.pipelines.common.api.RunStatus.{
  Success,
  UnsuccessfulRunStatus
 }
+import cromwell.backend.google.pipelines.common.errors.isQuotaMessage
 import cromwell.backend.google.pipelines.v2beta.PipelinesConversions._
 import cromwell.backend.google.pipelines.v2beta.api.Deserialization._
 import cromwell.backend.google.pipelines.v2beta.api.request.ErrorReporter._
@@ -197,16 +198,9 @@ trait GetRequestHandler { this: RequestHandler =>
  private def isQuotaDelayed(events: List[Event]): Boolean =
  events.sortBy(_.getTimestamp).reverse.headOption match {
  case Some(event) =>
- quotaMessages.exists(event.getDescription.contains)
+ isQuotaMessage(event.getDescription)
  case None =>
  // If the events list is empty, we're not waiting for quota yet
  false
  }
-
- private val quotaMessages = List(
- "A resource limit has delayed the operation",
- "usage too high",
- "no available zones",
- "resource_exhausted"
- )
 }