Add labels to subworkflows during Carbonite thawing [BA-5875] (broadinstitute#5223)

Author: mcovarr

build.sbt

@@ -164,7 +164,7 @@ lazy val services = project
   .dependsOn(ftpFileSystem % "test->test")
 
 lazy val hybridCarboniteMetadataService = project
-  .withLibrarySettings("hybrid-carbonite-metadata-service")
+  .withLibrarySettings("hybrid-carbonite-metadata-service", hybridCarboniteMetadataServiceDependencies)
   .dependsOn(services)
   .dependsOn(engine)
   .dependsOn(core % "test->test")

centaur/src/main/resources/standardTestCases/sub_sub.test

@@ -0,0 +1,18 @@
+# Used to generate JSON for testing label addition to sub workflows.
+name: sub_sub
+testFormat: workflowsuccess
+tags: [subworkflow]
+ignore: true
+
+files {
+  workflow: sub_sub/top.wdl
+  imports: [
+    sub_sub/sub.wdl,
+    sub_sub/sub_sub.wdl,
+    sub_sub/sub_sub_sub.wdl
+  ]
+}
+
+metadata {
+  status: Succeeded
+}

centaur/src/main/resources/standardTestCases/sub_sub/sub.wdl

@@ -0,0 +1,7 @@
+import "sub_sub.wdl" as sub_sub
+
+workflow wf {
+  scatter (i in range(2)) {
+    call sub_sub.wf
+  }
+}

centaur/src/main/resources/standardTestCases/sub_sub/sub_sub.wdl

@@ -0,0 +1,7 @@
+import "sub_sub_sub.wdl" as sub_sub_sub
+
+workflow wf {
+  scatter (i in range(2)) {
+    call sub_sub_sub.wf
+  }
+}

centaur/src/main/resources/standardTestCases/sub_sub/sub_sub_sub.wdl

@@ -0,0 +1,12 @@
+
+task foo {
+  command {
+    echo "foo"
+  }
+}
+
+workflow wf {
+  scatter (i in range(2)) {
+    call foo
+  }
+}

centaur/src/main/resources/standardTestCases/sub_sub/top.wdl

@@ -0,0 +1,7 @@
+import "sub.wdl" as sub
+
+workflow wf {
+  scatter (i in range(2)) {
+    call sub.wf
+  }
+}

common/src/main/scala/common/util/StringUtil.scala

@@ -44,5 +44,11 @@ object StringUtil {
       * e.g: /root/some/dir -> root/some/dir/
       */
     def relativeDirectory = string.ensureNoLeadingSlash.ensureSlashed
+
+    def elided(limit: Int): String = {
+      if (string.length > limit) {
+        s"(elided) ${string.take(limit)}..."
+      } else string
+    }
   }
 }

core/src/main/scala/cromwell/util/JsonEditor.scala

@@ -2,11 +2,13 @@ package cromwell.util
 
 import cats.data.NonEmptyList
 import common.collections.EnhancedCollections._
-import io.circe.{Json, JsonNumber, JsonObject}
+import cromwell.core.WorkflowId
+import io.circe.{Json, JsonNumber, JsonObject, Printer}
 import mouse.all._
 import io.circe.Json.Folder
 
 import scala.collection.immutable
+import common.util.StringUtil._
 
 object JsonEditor {
 
@@ -48,7 +50,7 @@ object JsonEditor {
             }
         }
         val jsonObject = Json.fromJsonObject(JsonObject.fromIterable(modified))
-        val keep = modified.size > 0
+        val keep = modified.nonEmpty
         (jsonObject, keep)
       }
     }
@@ -72,21 +74,77 @@ object JsonEditor {
 
   def logs(json: Json): Json = includeJson(json, NonEmptyList.of("stdout", "stderr", "backendLogs", "id"))
 
+  implicit class EnhancedJson(val json: Json) extends AnyVal {
+    def rootWorkflowId: Option[WorkflowId] = {
+      for {
+        o <- json.asObject
+        id <- o.kleisli("id")
+        s <- id.asString
+      } yield WorkflowId.fromString(s)
+    }
+  }
+
   /**
-    * We only return labels associated with top-level workflows.  Subworkflows don't include labels (as of 7/26/19).
-    *
-    * Thus this method puts the labels as a top-level field.
+    * In-memory upsert of labels into the base Json, handling root and sub workflows appropriately.
     *
     * @param json json blob with or without "labels" field
-    * @param labels a map of labels one would like to apply to a workflow json
+    * @param databaseLabels a map of workflow IDs to maps of labels one would like to apply to a workflow json
     * @return json with labels merged in.  Any prior non-object "labels" field will be overwritten and any object fields will be merged together and - again - any existing values overwritten.
     */
-  def augmentLabels(json: Json, labels: Map[String, String]): Json = {
-    val newData: Json = Json.fromFields(labels.safeMapValues(Json.fromString))
-    val newObj: Json = Json.fromFields(List(("labels", newData)))
-    //in the event of a key clash, the values in "newObj" will be favored over "json"
-    json deepMerge newObj
-  }
+  def updateLabels(json: Json, databaseLabels: Map[WorkflowId, Map[String, String]]): Json = {
+    val subWorkflowMetadataKey = "subWorkflowMetadata"
+
+    def doUpdateWorkflow(workflowJson: Json): Json = {
+      val id: String = workflowJson.rootWorkflowId.
+        getOrElse(throw new RuntimeException(s"did not find workflow id in ${workflowJson.printWith(Printer.spaces2).elided(100)}")).toString
+
+      // Look for an optional JsonObject keyed by "calls".
+      val callsObject: Option[JsonObject] = for {
+        wo <- workflowJson.asObject
+        callsJson <- wo("calls")
+        co <- callsJson.asObject
+      } yield co
+
+      val workflowWithUpdatedCalls: Json = callsObject match {
+        // If there were no calls just return the workflow JSON unmodified.
+        case None => workflowJson
+        case Some(calls) =>
+          val updatedCallsObject = calls.mapValues {
+            // The Json (a JSON array, really) corresponding to the array of call objects for a call name.
+            callValue: Json =>
+              // The object above converted to a List[Json].
+              val callArray: Vector[Json] = callValue.asArray.toVector.flatten
 
-  def removeSubworkflowData(json: Json): Json = excludeJson(json, NonEmptyList.of("subWorkflowMetadata"))
+              val updatedCallArray = callArray map { callJson =>
+                // If there is no subworkflow object this will be None.
+                val callAndSubworkflowObjects: Option[(JsonObject, Json)] = for {
+                  co <- callJson.asObject
+                  sub <- co(subWorkflowMetadataKey)
+                  _ <- sub.asObject
+                } yield (co, sub)
+
+                callAndSubworkflowObjects match {
+                  case None => callJson
+                  case Some((callObject, subworkflowObject)) =>
+                    // If the call contains a subWorkflowMetadata key, return a copy of the call with
+                    // its subworkflowMetadata updated via a recursive call to `doUpdateWorkflow`.
+                    val updatedSubworkflow = doUpdateWorkflow(subworkflowObject)
+                    Json.fromJsonObject(callObject.add(subWorkflowMetadataKey, updatedSubworkflow))
+                }
+              }
+              Json.fromValues(updatedCallArray)
+          }
+          Json.fromJsonObject(workflowJson.asObject.get.add("calls", Json.fromJsonObject(updatedCallsObject)))
+      }
+
+      databaseLabels.get(WorkflowId.fromString(id)) match {
+        case None => workflowWithUpdatedCalls
+        case Some(labels) =>
+          val labelsJson: Json = Json.fromFields(labels safeMapValues Json.fromString)
+          workflowWithUpdatedCalls deepMerge Json.fromFields(List(("labels", labelsJson)))
+      }
+    }
+
+    doUpdateWorkflow(workflowJson = json)
+  }
 }