[AN-142] Adding cost centaur test

centaur/src/main/resources/standardTestCases/cacheBetweenWF/cacheBetweenWFNoCost.wdl

@@ -0,0 +1,39 @@
+task getAverage {
+  Int base1 = 9
+  Int base2 = 13
+    command {
+        echo ${(base1*base2)/2}
+    }
+    output {
+        Float average = read_float(stdout())
+    }
+    runtime {
+       docker: "docker.io/ubuntu@sha256:71cd81252a3563a03ad8daee81047b62ab5d892ebbfbf71cf53415f29c130950"
+    }
+}
+
+task heightProduct{
+   Float baseAverage
+   Int height = 7
+
+   command {
+   		echo ${baseAverage*height}
+   }
+   output {
+		Float trapezoidalArea = read_float(stdout())
+   }
+   runtime {
+      docker: "ubuntu@sha256:71cd81252a3563a03ad8daee81047b62ab5d892ebbfbf71cf53415f29c130950"
+   }
+}
+
+workflow cacheBetweenWFNoCost {
+   call getAverage {
+   }
+   call heightProduct {
+      input: baseAverage = getAverage.average
+   }
+   output {
+        heightProduct.trapezoidalArea
+   }
+}

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

@@ -0,0 +1,16 @@
+name: cacheBetweenWFNoCost
+testFormat: runtwiceexpectingcallcachingnocost
+
+files {
+  workflow: cacheBetweenWF/cacheBetweenWFNoCost.wdl
+  options: common_options/cache_read_off_write_on.options
+  second-options: common_options/cache_read_on_write_on.options
+}
+
+metadata {
+  workflowName: cacheBetweenWFNoCost
+  status: Succeeded
+  "calls.cacheBetweenWFNoCost.getAverage.callCaching.result": "Cache Hit: <<CACHE_HIT_UUID>>:cacheBetweenWFNoCost.getAverage:-1"
+  "calls.cacheBetweenWFNoCost.heightProduct.callCaching.result": "Cache Hit: <<CACHE_HIT_UUID>>:cacheBetweenWFNoCost.heightProduct:-1"
+  "outputs.cacheBetweenWFNoCost.heightProduct.trapezoidalArea": 406.0
+}

centaur/src/main/resources/standardTestCases/recursive_imports/recursive_imports_cost.options

@@ -0,0 +1,3 @@
+{
+  "read_from_cache":false
+}

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

@@ -0,0 +1,22 @@
+name: recursive_imports_cost
+testFormat: workflowsuccessandverifycost
+
+files {
+  workflow: recursive_imports/recursive_imports.wdl
+  imports: [
+    forkjoin/forkjoin.wdl,
+    sub_workflow_hello_world/sub_workflow_hello_world.wdl,
+    sub_workflow_hello_world/sub_workflow_hello_world_import.wdl,
+    sub_workflow_interactions/sub_workflow_interactions_import.wdl,
+    sub_workflow_interactions/sub_workflow_interactions.wdl
+  ]
+  # Adds option to disable call-caching
+   options: recursive_imports/recursive_imports_cost.options
+}
+
+metadata {
+  workflowName: recursive_imports_cost
+  status: Succeeded
+}
+
+cost: 0.0016

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

@@ -14,3 +14,4 @@ metadata {
   status: Succeeded
   "outputs.main_workflow.main_output": "Hello sub world!"
 }
+

centaur/src/main/scala/centaur/api/CentaurCromwellClient.scala

@@ -122,6 +122,9 @@
   ): IO[WorkflowMetadata] =
     sendReceiveFutureCompletion(() => cromwellClient.metadata(id, args))
 
+  def cost(workflow: SubmittedWorkflow): IO[WorkflowCost] =
+    sendReceiveFutureCompletion(() => cromwellClient.cost(workflow.id))
+
   def archiveStatus(id: WorkflowId): IO[String] =
     sendReceiveFutureCompletion(() => cromwellClient.query(id)).map(_.results.head.metadataArchiveStatus)
 

centaur/src/main/scala/centaur/test/Test.scala

@@ -1060,4 +1060,51 @@
           IO.raiseError(CentaurTestException(message, workflow))
         }
     }
+
+  def getExpectedCost(workflowCost: Option[BigDecimal]): IO[BigDecimal] =
+    workflowCost match {
+      case Some(cost) if cost == 0 => IO.raiseError(new Exception("Expected cost cannot be 0"))
+      case Some(cost) => IO.pure(cost)
+      case None =>
+        IO.raiseError(new Exception("Expected 'cost' is required in the test config to validate the workflow cost"))
+    }
+
+  /**
+    * Validate that the actual cost is within 10% of the expected cost
+    */
+  def validateCost(actualCost: BigDecimal, expectedCost: BigDecimal): IO[Unit] = {
+    val costDiff = (actualCost - expectedCost).abs / expectedCost
+    if (costDiff > 0.1) {
+      IO.raiseError(
+        new Exception(s"Expected cost $expectedCost but got $actualCost, which is outside the 10% threshold")
+      )
+    } else {
+      IO.unit
+    }
+  }
+
+  def fetchAndValidateCost(workflowSpec: Workflow, submittedWorkflow: SubmittedWorkflow): Test[Unit] =
+    new Test[Unit] {
+
+      override def run: IO[Unit] =
+        for {
+          actualCost <- CentaurCromwellClient.cost(submittedWorkflow)
+          expectedCost <- getExpectedCost(workflowSpec.cost)
+          _ <- validateCost(actualCost.cost, expectedCost)
+        } yield ()
+    }
+
+  def validateNoCost(submittedWorkflow: SubmittedWorkflow): Test[Unit] =
+    new Test[Unit] {
+
+      override def run: IO[Unit] =
+        for {
+          actualCost <- CentaurCromwellClient.cost(submittedWorkflow)
+          _ =
+            if (actualCost.cost != 0)
+              IO.raiseError(new Exception(s"When using call caching, the cost must be 0, not ${actualCost.cost}"))
+            else IO.unit
+        } yield ()
+    }
+
 }

centaur/src/main/scala/centaur/test/formulas/TestFormulas.scala

@@ -130,6 +130,26 @@
       _ <- validateDirectoryContentsCounts(workflowDefinition, secondWf, metadata)
     } yield SubmitResponse(secondWf)
 
+  def runWorkflowTwiceExpectingCachingNoCost(
+    workflowDefinition: Workflow
+  )(implicit cromwellTracker: Option[CromwellTracker]): Test[SubmitResponse] =
+    for {
+      _ <- checkDescription(workflowDefinition, validityExpectation = Option(true))
+      _ <- timingVerificationNotSupported(workflowDefinition.maximumAllowedTime)
+      firstWF <- runSuccessfulWorkflow(workflowDefinition)
+      secondWf <- runSuccessfulWorkflow(workflowDefinition.secondRun)
+      _ <- printHashDifferential(firstWF, secondWf)
+      metadata <- fetchAndValidateNonSubworkflowMetadata(secondWf, workflowDefinition, Option(firstWF.id.id))
+      _ <- fetchAndValidateJobManagerStyleMetadata(secondWf,
+                                                   workflowDefinition,
+                                                   prefetchedOriginalNonSubWorkflowMetadata = None
+      )
+      _ = cromwellTracker.track(metadata)
+      _ <- validateNoCacheMisses(secondWf, metadata, workflowDefinition)
+      _ <- validateDirectoryContentsCounts(workflowDefinition, secondWf, metadata)
+      _ <- validateNoCost(secondWf)
+    } yield SubmitResponse(secondWf)
+
   def runWorkflowThriceExpectingCaching(
     workflowDefinition: Workflow
   )(implicit cromwellTracker: Option[CromwellTracker]): Test[SubmitResponse] =
@@ -337,6 +357,15 @@
       case _ => Test.invalidTestDefinition("Configuration not supported by PapiUpgradeTest", workflowDefinition)
     }
 
+  def runSuccessfulWorkflowAndVerifyCost(
+    workflowSpec: Workflow
+  ): Test[SubmitResponse] = for {
+    _ <- checkDescription(workflowSpec, validityExpectation = Option(true))
+    _ <- timingVerificationNotSupported(workflowSpec.maximumAllowedTime)
+    submittedWorkflow <- runSuccessfulWorkflow(workflowSpec)
+    _ <- fetchAndValidateCost(workflowSpec, submittedWorkflow)
+  } yield SubmitResponse(submittedWorkflow)
+
   implicit class EnhancedCromwellTracker(val tracker: Option[CromwellTracker]) extends AnyVal {
     def track(metadata: WorkflowMetadata): Unit = tracker foreach { _.track(metadata) }
   }

centaur/src/main/scala/centaur/test/standard/CentaurTestCase.scala

@@ -25,8 +25,10 @@
   def testFunction: Test[SubmitResponse] = this.testFormat match {
     case WorkflowSuccessTest => TestFormulas.runSuccessfulWorkflowAndVerifyMetadata(workflow)
     case WorkflowSuccessAndTimedOutputsTest => TestFormulas.runSuccessfulWorkflowAndVerifyTimeAndOutputs(workflow)
+    case WorkflowSuccessAndVerifyCostTest => TestFormulas.runSuccessfulWorkflowAndVerifyCost(workflow)
     case WorkflowFailureTest => TestFormulas.runFailingWorkflowAndVerifyMetadata(workflow)
     case RunTwiceExpectingCallCachingTest => TestFormulas.runWorkflowTwiceExpectingCaching(workflow)
+    case RunTwiceExpectingCallCachingNoCostTest => TestFormulas.runWorkflowTwiceExpectingCachingNoCost(workflow)
     case RunThriceExpectingCallCachingTest => TestFormulas.runWorkflowThriceExpectingCaching(workflow)
     case RunTwiceExpectingNoCallCachingTest => TestFormulas.runWorkflowTwiceExpectingNoCaching(workflow)
     case RunFailingTwiceExpectingNoCallCachingTest => TestFormulas.runFailingWorkflowTwiceExpectingNoCaching(workflow)

centaur/src/main/scala/centaur/test/standard/CentaurTestFormat.scala

@@ -13,8 +13,10 @@ sealed abstract class CentaurTestFormat(val name: String) {
   def testSpecString: String = this match {
     case WorkflowSuccessTest => "successfully run"
     case WorkflowSuccessAndTimedOutputsTest => "successfully run"
+    case WorkflowSuccessAndVerifyCostTest => "successfully run and verify cost"
     case WorkflowFailureTest => "fail during execution"
     case RunTwiceExpectingCallCachingTest => "call cache the second run of"
+    case RunTwiceExpectingCallCachingNoCostTest => "call cache the second run and verify no cost of"
     case RunThriceExpectingCallCachingTest => "call cache the third run of"
     case RunTwiceExpectingNoCallCachingTest => "NOT call cache the second run of"
     case RunFailingTwiceExpectingNoCallCachingTest => "Fail the first run and NOT call cache the second run of"
@@ -49,8 +51,10 @@ object CentaurTestFormat {
 
   case object WorkflowSuccessTest extends CentaurTestFormat("WorkflowSuccess")
   case object WorkflowSuccessAndTimedOutputsTest extends CentaurTestFormat("WorkflowSuccessAndTimedOutputs")
+  case object WorkflowSuccessAndVerifyCostTest extends CentaurTestFormat("WorkflowSuccessAndVerifyCost")
   case object WorkflowFailureTest extends CentaurTestFormat("WorkflowFailure")
   case object RunTwiceExpectingCallCachingTest extends CentaurTestFormat("RunTwiceExpectingCallCaching")
+  case object RunTwiceExpectingCallCachingNoCostTest extends CentaurTestFormat("RunTwiceExpectingCallCachingNoCost")
   case object RunThriceExpectingCallCachingTest extends CentaurTestFormat(name = "RunThriceExpectingCallCaching")
   case object RunTwiceExpectingNoCallCachingTest extends CentaurTestFormat("RunTwiceExpectingNoCallCaching")
   case object RunFailingTwiceExpectingNoCallCachingTest
@@ -127,8 +131,10 @@ object CentaurTestFormat {
     List(
       WorkflowSuccessTest,
       WorkflowSuccessAndTimedOutputsTest,
+      WorkflowSuccessAndVerifyCostTest,
       WorkflowFailureTest,
       RunTwiceExpectingCallCachingTest,
+      RunTwiceExpectingCallCachingNoCostTest,
       RunThriceExpectingCallCachingTest,
       RunTwiceExpectingNoCallCachingTest,
       RunFailingTwiceExpectingNoCallCachingTest,

centaur/src/main/scala/centaur/test/workflow/Workflow.scala

@@ -25,7 +25,8 @@ final case class Workflow private (testName: String,
                                    allowOtherOutputs: Boolean,
                                    skipDescribeEndpointValidation: Boolean,
                                    submittedWorkflowTracker: SubmittedWorkflowTracker,
-                                   maximumAllowedTime: Option[FiniteDuration]
+                                   maximumAllowedTime: Option[FiniteDuration],
+                                   cost: Option[BigDecimal] = None
 ) {
 
   def toWorkflowSubmission: WorkflowSingleSubmission = WorkflowSingleSubmission(
@@ -94,6 +95,7 @@ object Workflow {
         val validateDescription: Boolean = conf.get[Boolean]("skipDescribeEndpointValidation").valueOrElse(false)
 
         val maximumTime: Option[FiniteDuration] = conf.get[Option[FiniteDuration]]("maximumTime").value
+        val cost: Option[BigDecimal] = conf.get[Option[BigDecimal]]("cost").value
 
         (files, directoryContentCheckValidation, metadata, retryTestFailuresErrorOr) mapN {
           (f, d, m, retryTestFailures) =>
@@ -107,7 +109,8 @@ object Workflow {
                      allowOtherOutputs,
                      validateDescription,
                      submittedWorkflowTracker,
-                     maximumTime
+                     maximumTime,
+                     cost
             )
         }
 

core/src/main/resources/reference.conf

@@ -616,7 +616,7 @@ services {
   GcpCostCatalogService {
     class = "cromwell.services.cost.GcpCostCatalogService"
     config {
-      enabled = false
+      enabled = true
       catalogExpirySeconds = 86400
     }
   }

cromwellApiClient/src/main/scala/cromwell/api/CromwellClient.scala

@@ -18,6 +18,16 @@
 
 import scala.concurrent.ExecutionContext
 
+import model.CallCacheDiffJsonSupport._
+import model.CromwellBackendsJsonSupport._
+import model.CromwellStatusJsonSupport._
+import model.CromwellVersionJsonSupport._
+import model.WorkflowLabelsJsonSupport._
+import model.WorkflowOutputsJsonSupport._
+import model.WorkflowDescriptionJsonSupport._
+import model.CromwellQueryResultJsonSupport._
+import model.WorkflowCostJsonSupport._
+
 class CromwellClient(val cromwellUrl: URL,
                      val apiVersion: String,
                      val defaultCredentials: Option[HttpCredentials] = None
@@ -49,6 +59,7 @@
 
   def abortEndpoint(workflowId: WorkflowId): Uri = workflowSpecificGetEndpoint(workflowsEndpoint, workflowId, "abort")
   def statusEndpoint(workflowId: WorkflowId): Uri = workflowSpecificGetEndpoint(workflowsEndpoint, workflowId, "status")
+  def costEndpoint(workflowId: WorkflowId): Uri = workflowSpecificGetEndpoint(workflowsEndpoint, workflowId, "cost")
   def metadataEndpoint(workflowId: WorkflowId, args: Option[Map[String, List[String]]] = None): Uri =
     workflowSpecificGetEndpoint(workflowsEndpoint, workflowId, "metadata", args)
   def outputsEndpoint(workflowId: WorkflowId, args: Option[Map[String, List[String]]] = None): Uri =
@@ -69,15 +80,6 @@
   lazy val backendsEndpoint = s"$workflowsEndpoint/backends"
   lazy val versionEndpoint = s"$engineEndpoint/version"
 
-  import model.CallCacheDiffJsonSupport._
-  import model.CromwellBackendsJsonSupport._
-  import model.CromwellStatusJsonSupport._
-  import model.CromwellVersionJsonSupport._
-  import model.WorkflowLabelsJsonSupport._
-  import model.WorkflowOutputsJsonSupport._
-  import model.WorkflowDescriptionJsonSupport._
-  import model.CromwellQueryResultJsonSupport._
-
   def submit(workflow: WorkflowSubmission)(implicit ec: ExecutionContext): FailureResponseOrT[SubmittedWorkflow] = {
     val requestEntity = requestEntityForSubmit(workflow)
 
@@ -133,6 +135,9 @@
   def status(workflowId: WorkflowId)(implicit ec: ExecutionContext): FailureResponseOrT[WorkflowStatus] =
     simpleRequest[CromwellStatus](statusEndpoint(workflowId)) map WorkflowStatus.apply
 
+  def cost(workflowId: WorkflowId)(implicit ec: ExecutionContext): FailureResponseOrT[WorkflowCost] =
+    simpleRequest[WorkflowCost](costEndpoint(workflowId))
+
   def metadata(workflowId: WorkflowId,
                args: Option[Map[String, List[String]]] = None,
                headers: List[HttpHeader] = defaultHeaders

cromwellApiClient/src/main/scala/cromwell/api/model/WorkflowCost.scala

@@ -0,0 +1,9 @@
+package cromwell.api.model
+
+import spray.json.DefaultJsonProtocol
+
+object WorkflowCostJsonSupport extends DefaultJsonProtocol {
+  implicit val WorkflowCostJsonFormat = jsonFormat5(WorkflowCost)
+}
+
+final case class WorkflowCost(errors: List[String], id: String, cost: BigDecimal, status: String, currency: String)

docs/developers/Centaur.md

@@ -14,7 +14,9 @@ You can now run the tests from another terminal.
 
 There are two ways to invoke the integration tests:
 
-* `sbt "centaur / IntegrationTest / test"` - compiles Centaur and runs all tests via sbt directly. Tests are expected to be in the `centaur/src/main/standardTestCases` directory. This can be changed by modifying `reference.conf`.
+* `sbt "centaur / IntegrationTest / test"` - compiles Centaur and runs all tests via sbt directly. 
+*  Can also just run `centaur/ IntegrationTest / test` from the sbt terminal.
+Tests are expected to be in the `centaur/src/main/standardTestCases` directory. In order to run a set of specific tests, create a new subdirectory that contains the tests to run, then modify the path in the centaur `reference.conf`.
 
 * `src/ci/bin/testCentaurLocal.sh` - runs the same tests using the continuous integration pipeline configuration.
 
@@ -77,6 +79,9 @@ metadata {
   "failures.0.causedBy": "BetweenKeyboardAndChairException"
 }
 
+// Optional, the expected cost of running the workflow
+cost: 0.0000
+
 filesystemcheck: "local" // possible values: "local", "gcs". Used in conjunction with outputExpectations to define files we expect to exist after running this workflow.
 outputExpectations: {
     "/path/to/my/output/file1": 1
@@ -91,6 +96,7 @@ The `basePath` field is optional, but if supplied all paths will be resolved fro
 The `testFormat` field can be one of the following, case insensitive:
 * `workflowsuccess`: The workflow being supplied is expected to successfully complete
 * `workflowfailure`: The workflow being supplied is expected to fail
+* `workflowsuccessandverifycost`: The workflow being supplied is expected to complete and the expected cost will be verified
 
 The `metadata` is optional. If supplied, Centaur will retrieve the metadata from the successfully completed workflow and compare the values retrieved to those supplied. At the moment the only fields supported are strings, numbers and booleans.
 
@@ -105,7 +111,10 @@ In case the absolute path the cromwell root is used (for example: `/home/my_user
 * `"calls.hello.hello.exit_code": "<<WORKFLOW_ROOT>>/call-hello/execution/exit_code"`
 
 In case testing of the caching is required `<<CACHE_HIT_UUID>>` can be used. 
-The testFormat should be `runtwiceexpectingcallcaching`.
+The testFormat should be `runtwiceexpectingcallcaching`. To verify that the cost is 0 when using call-caching the testFormat should be `runtwiceexpectingcallcachingnocost`
+
+The cost is optional. If supplied, Centaur will retrieve the cost of the successfully completed workflow and compare it to the cost supplied. 
+The expected range is within 10% of the estimated cost. If evaluating the cost, the test format must be `WorkflowSuccessAndVerifyCost` and the call-caching option must be disabled for that test (example can be found in the `recursive_imports_cost.test`)
 
 
 ## Centaur Test Types

docs/developers/Running.md

@@ -0,0 +1,17 @@
+In order to run a Cromwell instance locally, there are the some prerequisites:
+- Have docker and IntelliJ installed
+- Start a local mysql instance by running the `start_publish_mysql_docker.sh` script
+
+To run the backend locally, set up your preferred run configuration. 
+More information on backends can be found [here](Backend.md)
+![](select-run-config.png)
+
+If you've created a new config, you may need to edit your run configuration to point to the correct config file or edit the environment variables.
+Example for batch:
+![](edit-run-config.png)
+
+
+Upon startup, the Cromwell swagger will be available at `http://localhost:8000`.
+(You can change the port and other configurations by following [these instructions](../Configuring.md)
+
+**Note**: To verify that workflows are being submitted to the correct backend, look for logs  with `Call-to-Backend assignments` and check that the tasks are being directed to `-> GCPBATCH` or `-> PAPI` etc.