Skip to content

Commit

Permalink
get necessary data from google
Browse files Browse the repository at this point in the history
  • Loading branch information
@THWiseman
THWiseman committed Sep 11, 2024
1 parent 550a9dd commit 44febb3
Show file tree
Hide file tree
Showing 5 changed files with 96 additions and 46 deletions.
4 changes: 4 additions & 0 deletions ...ommon/src/main/scala/cromwell/backend/google/pipelines/common/PapiCostPollingHelper.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,5 +18,9 @@ class PapiCostPollingHelper(tellMetadataFn: Map[String, Any] => Unit) extends Co
case event if event.name == CallMetadataKeys.VmEndTime => event.offsetDateTime
}

def hi(pollStatus: RunStatus) = {
pollStatus.eventList
}

override def tellMetadata(metadata: Map[String, Any]): Unit = tellMetadataFn(metadata)
}
31 changes: 20 additions & 11 deletions ...elines/common/src/main/scala/cromwell/backend/google/pipelines/common/api/RunStatus.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,20 +3,22 @@ package cromwell.backend.google.pipelines.common.api
import _root_.io.grpc.Status
import cromwell.backend.google.pipelines.common.PipelinesApiAsyncBackendJobExecutionActor
import cromwell.core.ExecutionEvent

import scala.util.Try

case class InstantiatedVmInfo(zone: String, machineType: String)
sealed trait RunStatus {
def eventList: Seq[ExecutionEvent]
def toString: String

val instantiatedVmInfo: Option[InstantiatedVmInfo]
}

object RunStatus {
case class Initializing(eventList: Seq[ExecutionEvent]) extends RunStatus { override def toString = "Initializing" }
case class AwaitingCloudQuota(eventList: Seq[ExecutionEvent]) extends RunStatus {
case class Initializing(eventList: Seq[ExecutionEvent], instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty) extends RunStatus { override def toString = "Initializing" }
case class AwaitingCloudQuota(eventList: Seq[ExecutionEvent], instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty) extends RunStatus {
override def toString = "AwaitingCloudQuota"
}
case class Running(eventList: Seq[ExecutionEvent]) extends RunStatus { override def toString = "Running" }
case class Running(eventList: Seq[ExecutionEvent], instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty) extends RunStatus { override def toString = "Running" }

sealed trait TerminalRunStatus extends RunStatus {
def machineType: Option[String]
Expand All @@ -38,7 +40,8 @@ object RunStatus {
case class Success(eventList: Seq[ExecutionEvent],
machineType: Option[String],
zone: Option[String],
instanceName: Option[String]
instanceName: Option[String],
instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty
) extends TerminalRunStatus {
override def toString = "Success"
}
Expand All @@ -60,7 +63,8 @@ object RunStatus {
machineType: Option[String],
zone: Option[String],
instanceName: Option[String],
wasPreemptible: Boolean
wasPreemptible: Boolean,
instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty
): UnsuccessfulRunStatus = {
val jesCode: Option[Int] = errorMessage flatMap { em => Try(em.substring(0, em.indexOf(':')).toInt).toOption }

Expand Down Expand Up @@ -88,7 +92,8 @@ object RunStatus {
eventList,
machineType,
zone,
instanceName
instanceName,
instantiatedVmInfo
)
}
}
Expand All @@ -99,7 +104,8 @@ object RunStatus {
eventList: Seq[ExecutionEvent],
machineType: Option[String],
zone: Option[String],
instanceName: Option[String]
instanceName: Option[String],
instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty
) extends UnsuccessfulRunStatus {
override def toString = "Failed"
}
Expand All @@ -113,7 +119,8 @@ object RunStatus {
eventList: Seq[ExecutionEvent],
machineType: Option[String],
zone: Option[String],
instanceName: Option[String]
instanceName: Option[String],
instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty
) extends UnsuccessfulRunStatus {
override def toString = "Cancelled"
}
Expand All @@ -124,7 +131,8 @@ object RunStatus {
eventList: Seq[ExecutionEvent],
machineType: Option[String],
zone: Option[String],
instanceName: Option[String]
instanceName: Option[String],
instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty
) extends UnsuccessfulRunStatus {
override def toString = "Preempted"
}
Expand All @@ -139,7 +147,8 @@ object RunStatus {
eventList: Seq[ExecutionEvent],
machineType: Option[String],
zone: Option[String],
instanceName: Option[String]
instanceName: Option[String],
instantiatedVmInfo: Option[InstantiatedVmInfo] = Option.empty
) extends UnsuccessfulRunStatus {
override def toString = "QuotaFailed"
}
Expand Down
37 changes: 34 additions & 3 deletions ...lines/common/src/test/scala/cromwell/backend/google/pipelines/common/CostLookupSpec.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,14 +25,45 @@ class CostLookupSpec

val testCatalogService = constructTestActor

it should "find a CPU from Runtime Attributes" in {
it should "find a CPU sku" in {
val machineType = Some(N2)
val usageType = Some(OnDemand)
val customization = Some(Custom)
val resourceGroup = Some(Cpu)
val region = "europe-west9"
val key = CostCatalogKey(machineType, usageType, customization, resourceGroup, region)
val result = testCatalogService.getSku(key)
result shouldBe true
val result = testCatalogService.getSku(key).get.catalogObject.getDescription
result shouldBe "N2 Custom Instance Core running in Paris"
}

it should "find a RAM sku" in {
val machineType = Some(N2)
val usageType = Some(OnDemand)
val customization = Some(Custom)
val resourceGroup = Some(Ram)
val region = "europe-west9"
val key = CostCatalogKey(machineType, usageType, customization, resourceGroup, region)
val result = testCatalogService.getSku(key).get.catalogObject.getDescription
result shouldBe "N2 Custom Instance Ram running in Paris"
}

it should "find CPU skus for all supported machine types" in {
val legalMachineTypes: List[MachineType] = List(N1, N2, N2d)
val legalUsageTypes: List[UsageType] = List(Preemptible, OnDemand)
val legalCustomizations: List[MachineCustomization] = List(Custom, Predefined)
val resourceGroup: Option[ResourceGroup] = Some(Cpu)
val region = "us-west1"
for (machineType <- legalMachineTypes) {
for(usageType <- legalUsageTypes) {
for(customization <- legalCustomizations) {
val key = CostCatalogKey(Some(machineType), Some(usageType), Some(customization), resourceGroup, region)
val result = testCatalogService.getSku(key)
if(!result.isEmpty){
println("Success")
}
result.isEmpty shouldBe false
}
}
}
}
}
2 changes: 1 addition & 1 deletion ...a/src/main/scala/cromwell/backend/google/pipelines/v2beta/api/request/ErrorReporter.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ class ErrorReporter(machineType: Option[String],
// Reverse the list because the first failure (likely the most relevant, will appear last otherwise)
val unexpectedExitEvents: List[String] = unexpectedExitStatusErrorStrings(events, actions).reverse

builder(status, None, failed.toList ++ unexpectedExitEvents, executionEvents, machineType, zone, instanceName)
builder(status, None, failed.toList ++ unexpectedExitEvents, executionEvents, machineType, zone, instanceName, Option.empty)
}

// There's maybe one FailedEvent per operation with a summary error message
Expand Down
68 changes: 37 additions & 31 deletions ...c/main/scala/cromwell/backend/google/pipelines/v2beta/api/request/GetRequestHandler.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import com.google.api.services.lifesciences.v2beta.model._
import common.validation.Validation._
import cromwell.backend.google.pipelines.common.action.ActionLabels._
import cromwell.backend.google.pipelines.common.api.PipelinesApiRequestManager._
import cromwell.backend.google.pipelines.common.api.RunStatus
import cromwell.backend.google.pipelines.common.api.{InstantiatedVmInfo, RunStatus}
import cromwell.backend.google.pipelines.common.api.RunStatus.{
AwaitingCloudQuota,
Initializing,
Expand Down Expand Up @@ -81,33 +81,39 @@ trait GetRequestHandler { this: RequestHandler =>
.toList
.flatten
val executionEvents = getEventList(metadata, events, actions)
val workerAssignedEvent: Option[WorkerAssignedEvent] =
events.collectFirst {
case event if event.getWorkerAssigned != null => event.getWorkerAssigned
}
val virtualMachineOption = for {
pipelineValue <- pipeline
resources <- Option(pipelineValue.getResources)
virtualMachine <- Option(resources.getVirtualMachine)
} yield virtualMachine

// Correlate `executionEvents` to `actions` to potentially assign a grouping into the appropriate events.
val machineType = virtualMachineOption.flatMap(virtualMachine => Option(virtualMachine.getMachineType))
/*
preemptible is only used if the job fails, as a heuristic to guess if the VM was preempted.
If we can't get the value of preempted we still need to return something, returning false will not make the
failure count as a preemption which seems better than saying that it was preemptible when we really don't know
*/
val preemptibleOption = for {
pipelineValue <- pipeline
resources <- Option(pipelineValue.getResources)
virtualMachine <- Option(resources.getVirtualMachine)
preemptible <- Option(virtualMachine.getPreemptible)
} yield preemptible
val preemptible = preemptibleOption.exists(_.booleanValue)
val instanceName =
workerAssignedEvent.flatMap(workerAssignedEvent => Option(workerAssignedEvent.getInstance()))
val zone = workerAssignedEvent.flatMap(workerAssignedEvent => Option(workerAssignedEvent.getZone))
val instantiatedVmInfo: Option[InstantiatedVmInfo] = (zone, machineType) match {
case (Some(instantiatedZone), Some(instantiatedMachineType)) =>
Option(InstantiatedVmInfo(instantiatedZone, instantiatedMachineType))
case _ => Option.empty
}
if (operation.getDone) {
val workerAssignedEvent: Option[WorkerAssignedEvent] =
events.collectFirst {
case event if event.getWorkerAssigned != null => event.getWorkerAssigned
}
val virtualMachineOption = for {
pipelineValue <- pipeline
resources <- Option(pipelineValue.getResources)
virtualMachine <- Option(resources.getVirtualMachine)
} yield virtualMachine
// Correlate `executionEvents` to `actions` to potentially assign a grouping into the appropriate events.
val machineType = virtualMachineOption.flatMap(virtualMachine => Option(virtualMachine.getMachineType))
/*
preemptible is only used if the job fails, as a heuristic to guess if the VM was preempted.
If we can't get the value of preempted we still need to return something, returning false will not make the
failure count as a preemption which seems better than saying that it was preemptible when we really don't know
*/
val preemptibleOption = for {
pipelineValue <- pipeline
resources <- Option(pipelineValue.getResources)
virtualMachine <- Option(resources.getVirtualMachine)
preemptible <- Option(virtualMachine.getPreemptible)
} yield preemptible
val preemptible = preemptibleOption.exists(_.booleanValue)
val instanceName =
workerAssignedEvent.flatMap(workerAssignedEvent => Option(workerAssignedEvent.getInstance()))
val zone = workerAssignedEvent.flatMap(workerAssignedEvent => Option(workerAssignedEvent.getZone))
// If there's an error, generate an unsuccessful status. Otherwise, we were successful!
Option(operation.getError) match {
case Some(error) =>
Expand All @@ -122,14 +128,14 @@ trait GetRequestHandler { this: RequestHandler =>
pollingRequest.workflowId
)
errorReporter.toUnsuccessfulRunStatus(error, events)
case None => Success(executionEvents, machineType, zone, instanceName)
case None => Success(executionEvents, machineType, zone, instanceName, instantiatedVmInfo)
}
} else if (isQuotaDelayed(events)) {
AwaitingCloudQuota(executionEvents)
AwaitingCloudQuota(executionEvents, instantiatedVmInfo)
} else if (operation.hasStarted) {
Running(executionEvents)
Running(executionEvents, instantiatedVmInfo)
} else {
Initializing(executionEvents)
Initializing(executionEvents, instantiatedVmInfo)
}
} catch {
case nullPointerException: NullPointerException =>
Expand Down

0 comments on commit 44febb3

Please sign in to comment.