WX-1217 Workflow completion callback

cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureCredentials.scala

@@ -33,7 +33,7 @@ case object AzureCredentials {
     new DefaultAzureCredentialBuilder()
       .authorityHost(azureProfile.getEnvironment.getActiveDirectoryEndpoint)
 
-  def getAccessToken(identityClientId: Option[String]): ErrorOr[String] = {
+  def getAccessToken(identityClientId: Option[String] = None): ErrorOr[String] = {
     val credentials = identityClientId.foldLeft(defaultCredentialBuilder) {
       (builder, clientId) => builder.managedIdentityClientId(clientId)
     }.build()

core/src/main/resources/reference.conf

@@ -626,3 +626,17 @@ ga4gh {
     contact-info-url = "https://cromwell.readthedocs.io/en/stable/"
   }
 }
+
+workflow-state-callback {
+  enabled: false
+  # endpoint: "http://example.com/foo" # Can be overridden in workflow options
+  # auth.azure: true
+  # Users can override default retry behavior if desired
+  # request-backoff {
+  #   min: "3 seconds"
+  #   max: "5 minutes"
+  #   multiplier: 1.1
+  # }
+  # max-retries = 10
+
+}

core/src/main/scala/cromwell/core/WorkflowOptions.scala

@@ -62,6 +62,7 @@ object WorkflowOptions {
   case object WorkflowFailureMode extends WorkflowOption("workflow_failure_mode")
   case object UseReferenceDisks extends WorkflowOption("use_reference_disks")
   case object MemoryRetryMultiplier extends WorkflowOption("memory_retry_multiplier")
+  case object WorkflowCallbackUri extends WorkflowOption("workflow_callback_uri")
 
   private lazy val WorkflowOptionsConf = ConfigFactory.load.getConfig("workflow-options")
   private lazy val EncryptedFields: Seq[String] = WorkflowOptionsConf.getStringList("encrypted-fields").asScala.toList

engine/src/main/scala/cromwell/engine/workflow/WorkflowActor.scala

@@ -1,7 +1,6 @@
 package cromwell.engine.workflow
 
 import java.util.concurrent.atomic.AtomicInteger
-
 import akka.actor.SupervisorStrategy.Stop
 import akka.actor._
 import com.typesafe.config.Config
@@ -23,6 +22,7 @@ import cromwell.engine.workflow.lifecycle.deletion.DeleteWorkflowFilesActor
 import cromwell.engine.workflow.lifecycle.deletion.DeleteWorkflowFilesActor.{DeleteWorkflowFilesFailedResponse, DeleteWorkflowFilesSucceededResponse, StartWorkflowFilesDeletion}
 import cromwell.engine.workflow.lifecycle.execution.WorkflowExecutionActor
 import cromwell.engine.workflow.lifecycle.execution.WorkflowExecutionActor._
+import cromwell.engine.workflow.lifecycle.finalization.WorkflowCallbackActor.PerformCallbackCommand
 import cromwell.engine.workflow.lifecycle.finalization.WorkflowFinalizationActor.{StartFinalizationCommand, WorkflowFinalizationFailedResponse, WorkflowFinalizationSucceededResponse}
 import cromwell.engine.workflow.lifecycle.finalization.{CopyWorkflowLogsActor, CopyWorkflowOutputsActor, WorkflowFinalizationActor}
 import cromwell.engine.workflow.lifecycle.initialization.WorkflowInitializationActor
@@ -52,6 +52,7 @@ object WorkflowActor {
   final case class AbortWorkflowWithExceptionCommand(exception: Throwable) extends WorkflowActorCommand
   case object SendWorkflowHeartbeatCommand extends WorkflowActorCommand
   case object AwaitMetadataIntegrity
+  case class PerformWorkflowCallback(uri: Option[String], workflowState: WorkflowState)
 
   case class WorkflowFailedResponse(workflowId: WorkflowId, inState: WorkflowActorState, reasons: Seq[Throwable])
 
@@ -149,7 +150,7 @@ object WorkflowActor {
                                initializationData: AllBackendInitializationData,
                                lastStateReached: StateCheckpoint,
                                effectiveStartableState: StartableState,
-                               workflowFinalOutputs: Set[WomValue] = Set.empty,
+                               workflowFinalOutputs: Option[CallOutputs] = None,
                                workflowAllOutputs: Set[WomValue] = Set.empty,
                                rootAndSubworkflowIds: Set[WorkflowId] = Set.empty,
                                failedInitializationAttempts: Int = 0)
@@ -176,6 +177,7 @@ object WorkflowActor {
             ioActor: ActorRef,
             serviceRegistryActor: ActorRef,
             workflowLogCopyRouter: ActorRef,
+            workflowCallbackActor: Option[ActorRef],
             jobStoreActor: ActorRef,
             subWorkflowStoreActor: ActorRef,
             callCacheReadActor: ActorRef,
@@ -199,6 +201,7 @@ object WorkflowActor {
         ioActor = ioActor,
         serviceRegistryActor = serviceRegistryActor,
         workflowLogCopyRouter = workflowLogCopyRouter,
+        workflowCallbackActor = workflowCallbackActor,
         jobStoreActor = jobStoreActor,
         subWorkflowStoreActor = subWorkflowStoreActor,
         callCacheReadActor = callCacheReadActor,
@@ -226,6 +229,7 @@ class WorkflowActor(workflowToStart: WorkflowToStart,
                     ioActor: ActorRef,
                     override val serviceRegistryActor: ActorRef,
                     workflowLogCopyRouter: ActorRef,
+                    workflowCallbackActor: Option[ActorRef],
                     jobStoreActor: ActorRef,
                     subWorkflowStoreActor: ActorRef,
                     callCacheReadActor: ActorRef,
@@ -516,6 +520,11 @@ class WorkflowActor(workflowToStart: WorkflowToStart,
       stay()
     case Event(AwaitMetadataIntegrity, data) =>
       goto(MetadataIntegrityValidationState) using data.copy(lastStateReached = data.lastStateReached.copy(state = stateName))
+    case Event(PerformWorkflowCallback(uri, workflowState), data) =>
+      workflowCallbackActor.foreach { wca =>
+        wca ! PerformCallbackCommand(workflowId, uri, workflowState, data.workflowFinalOutputs.getOrElse(CallOutputs.empty))
+      }
+      stay()
   }
 
   onTransition {
@@ -534,26 +543,28 @@ class WorkflowActor(workflowToStart: WorkflowToStart,
         case _ => // The WMA is waiting for the WorkflowActorWorkComplete message. No extra information needed here.
       }
 
-      // Copy/Delete workflow logs
-      if (WorkflowLogger.isEnabled) {
-        /*
-          * The submitted workflow options have been previously validated by the CromwellApiHandler. These are
-          * being recreated so that in case MaterializeWorkflowDescriptor fails, the workflow logs can still
-          * be copied by accessing the workflow options outside of the EngineWorkflowDescriptor.
-          */
-        def bruteForceWorkflowOptions: WorkflowOptions = sources.workflowOptions
-        val system = context.system
-        val ec = context.system.dispatcher
-        def bruteForcePathBuilders: Future[List[PathBuilder]] = {
-          // Protect against path builders that may throw an exception instead of returning a failed future
-          Future(EngineFilesystems.pathBuildersForWorkflow(bruteForceWorkflowOptions, pathBuilderFactories)(system))(ec).flatten
-        }
+      /*
+       * The submitted workflow options have been previously validated by the CromwellApiHandler. These are
+       * being recreated so that in case MaterializeWorkflowDescriptor fails, the workflow logs can still
+       * be copied by accessing the workflow options outside of the EngineWorkflowDescriptor. Used for both
+       * copying workflow log and sending workflow callback.
+       */
+      def bruteForceWorkflowOptions: WorkflowOptions = sources.workflowOptions
+
+      val system = context.system
+      val ec = context.system.dispatcher
+      def bruteForcePathBuilders: Future[List[PathBuilder]] = {
+        // Protect against path builders that may throw an exception instead of returning a failed future
+        Future(EngineFilesystems.pathBuildersForWorkflow(bruteForceWorkflowOptions, pathBuilderFactories)(system))(ec).flatten
+      }
 
-        val (workflowOptions, pathBuilders) = stateData.workflowDescriptor match {
-          case Some(wd) => (wd.backendDescriptor.workflowOptions, Future.successful(wd.pathBuilders))
-          case None => (bruteForceWorkflowOptions, bruteForcePathBuilders)
-        }
+      val (workflowOptions, pathBuilders) = stateData.workflowDescriptor match {
+        case Some(wd) => (wd.backendDescriptor.workflowOptions, Future.successful(wd.pathBuilders))
+        case None => (bruteForceWorkflowOptions, bruteForcePathBuilders)
+      }
 
+      // Copy/Delete workflow logs
+      if (WorkflowLogger.isEnabled) {
         workflowOptions.get(FinalWorkflowLogDir).toOption match {
           case Some(destinationDir) =>
             pathBuilders
@@ -566,6 +577,10 @@ class WorkflowActor(workflowToStart: WorkflowToStart,
         }
       }
 
+      // Attempt to perform workflow completion callback
+      val callbackUri = workflowOptions.get(WorkflowOptions.WorkflowCallbackUri).toOption
+      self ! PerformWorkflowCallback(callbackUri, terminalState.workflowState)
+
       // We can't transition from within another transition function, but we can instruct ourselves to with a message:
       self ! AwaitMetadataIntegrity
 
@@ -627,7 +642,7 @@ class WorkflowActor(workflowToStart: WorkflowToStart,
         rootWorkflowId = rootWorkflowId,
         rootWorkflowRootPaths = data.initializationData.getWorkflowRoots(),
         rootAndSubworkflowIds = data.rootAndSubworkflowIds,
-        workflowFinalOutputs = data.workflowFinalOutputs,
+        workflowFinalOutputs = data.workflowFinalOutputs.map(out => out.outputs.values.toSet).getOrElse(Set.empty),
         workflowAllOutputs = data.workflowAllOutputs,
         pathBuilders = data.workflowDescriptor.get.pathBuilders,
         serviceRegistryActor = serviceRegistryActor,
@@ -689,7 +704,7 @@ class WorkflowActor(workflowToStart: WorkflowToStart,
     finalizationActor ! StartFinalizationCommand
     goto(FinalizingWorkflowState) using data.copy(
       lastStateReached = StateCheckpoint (lastStateOverride.getOrElse(stateName), failures),
-      workflowFinalOutputs = workflowFinalOutputs.outputs.values.toSet,
+      workflowFinalOutputs = Option(workflowFinalOutputs),
       workflowAllOutputs = workflowAllOutputs,
       rootAndSubworkflowIds = rootAndSubworkflowIds
     )

engine/src/main/scala/cromwell/engine/workflow/WorkflowManagerActor.scala

@@ -57,6 +57,7 @@ object WorkflowManagerActor {
             ioActor: ActorRef,
             serviceRegistryActor: ActorRef,
             workflowLogCopyRouter: ActorRef,
+            workflowCallbackActor: Option[ActorRef],
             jobStoreActor: ActorRef,
             subWorkflowStoreActor: ActorRef,
             callCacheReadActor: ActorRef,
@@ -75,6 +76,7 @@ object WorkflowManagerActor {
       ioActor = ioActor,
       serviceRegistryActor = serviceRegistryActor,
       workflowLogCopyRouter = workflowLogCopyRouter,
+      workflowCallbackActor = workflowCallbackActor,
       jobStoreActor = jobStoreActor,
       subWorkflowStoreActor = subWorkflowStoreActor,
       callCacheReadActor = callCacheReadActor,
@@ -124,6 +126,7 @@ case class WorkflowManagerActorParams(config: Config,
                                       ioActor: ActorRef,
                                       serviceRegistryActor: ActorRef,
                                       workflowLogCopyRouter: ActorRef,
+                                      workflowCallbackActor: Option[ActorRef],
                                       jobStoreActor: ActorRef,
                                       subWorkflowStoreActor: ActorRef,
                                       callCacheReadActor: ActorRef,
@@ -313,6 +316,7 @@ class WorkflowManagerActor(params: WorkflowManagerActorParams)
       ioActor = params.ioActor,
       serviceRegistryActor = params.serviceRegistryActor,
       workflowLogCopyRouter = params.workflowLogCopyRouter,
+      workflowCallbackActor = params.workflowCallbackActor,
       jobStoreActor = params.jobStoreActor,
       subWorkflowStoreActor = params.subWorkflowStoreActor,
       callCacheReadActor = params.callCacheReadActor,