WX-1217 Workflow completion callback

CHANGELOG.md

@@ -10,6 +10,11 @@ WDL `size` engine function now works for HTTP files.
 Cromwell can now send logs to Azure Application Insights. To enable, set environment 
 variable `APPLICATIONINSIGHTS_INSTRUMENTATIONKEY` to your account's key. [See here for information.](https://learn.microsoft.com/en-us/azure/azure-monitor/app/sdk-connection-string)
 
+### Workflow Completion Callback
+
+Cromwell can be configured to send a POST request to a specified URL when a workflow completes. The request body
+includes the workflow id, terminal state, and (if workflow succeeded) final outputs or (if workflow failed) error message.
+See `WorkflowCallback` in [ReadTheDocs](https://cromwell.readthedocs.io/en/stable/) for more information.
 
 ## 85 Release Notes
 

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

docs/cromwell_features/WorkflowCallback.md

@@ -0,0 +1,55 @@
+The workflow callback is a simple way to integrate Cromwell with an external system. When each workflow reaches a terminal
+state, Cromwell will POST a message to a provided URL (see below for schema of this message). Messages are sent for root
+workflows only, not subworkflows.
+
+### Configuration
+
+This feature will only be used if enabled via config. All config items except `enabled` are optional.
+
+```
+workflow-state-callback {
+  enabled: true
+  endpoint: "http://example.com"
+  auth.azure: true
+  request-backoff {
+    min: "3 seconds",
+    max: "5 minutes",
+    multiplier: 1.1
+  }
+  max-retries = 10
+}
+```
+
+ * `enabled`: This boolean controls whether a callback will be attempted or not.
+ * `endpoint`: This is the default URL to send the message to. If this is unset, and no URL is set in workflow options, no callback will be sent.
+ * `auth.azure`: If true, and if Cromwell is running in an Azure environment, Cromwell will include an auth header with bearer token generated from local Azure credentials.
+ * `request-backoff` and `max-retries`: Include these to override the default retry behavior (default behavior shown here).
+
+### Workflow Options
+
+You may choose to override the `endpoint` set in config by including this workflow option:
+```
+{
+  "workflow_callback_uri": "http://mywebsite.com"
+}
+```
+
+### Callback schema
+
+Below is an example of a callback request body.
+
+```
+{
+  "workflowId": "00001111-2222-3333-4444-555566667777",
+  "state": "Succeeded",
+  "outputs": {
+    "task1.out": 5,
+    "task2.out": "/some/file.txt"
+  }
+}
+```
+
+ * `workflowId`: The UUID of the workflow
+ * `state`: The terminal state of the workflow. The list of possible values is: `Succeeded`, `Failed`, `Aborted`
+ * `outputs`: The final outputs of the workflow, as would be returned from the `api/workflows/{version}/{id}/outputs` endpoint. This field will ONLY be included when the workflow was successful.
+ * `failures`: A list of strings describing the workflow's failures. This field will ONLY be included if the workflow failed.

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,17 @@ 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),
+          data.lastStateReached.failures.toList.flatMap(_.map(_.getMessage))
+        )
+      }
+      stay()
   }
 
   onTransition {
@@ -534,26 +549,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 +583,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 +648,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 +710,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,