[WX-1792] Helper to Actor #7544
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| package cromwell.backend.standard.pollmonitoring | ||
| import akka.actor.Actor | ||
| import cromwell.services.metadata.CallMetadataKeys | ||
| import java.time.OffsetDateTime | ||
| trait PollResultMessage | ||
| case class ProcessThisPollResult[PollResultType](pollResultType: PollResultType) extends PollResultMessage | ||
| case class AsyncJobHasFinished(terminalStateName: String) extends PollResultMessage | ||
|
|
||
| /** | ||
| * Processes poll results from backends and sends messages to other actors based on their contents. | ||
| * Primarily concerned with reporting start times, end times, and cost data to both the bard and cromwell metadata services. | ||
| */ | ||
| trait PollResultMonitorActor[PollResultType] extends Actor { | ||
| // Time that Cromwell (but not necessarily the cloud) started working on this job. | ||
| def extractEarliestEventTimeFromRunState(pollStatus: PollResultType): Option[OffsetDateTime] | ||
|
|
||
| // Time that the user VM started spending money. | ||
| def extractStartTimeFromRunState(pollStatus: PollResultType): Option[OffsetDateTime] | ||
|
|
||
| // Time that the user VM stopped spending money. | ||
| def extractEndTimeFromRunState(pollStatus: PollResultType): Option[OffsetDateTime] | ||
|
|
||
| // Function to emit metadata that is associated with a specific call attempt. | ||
| def tellMetadata(metadata: Map[String, Any]): Unit | ||
|
|
||
| // Function that reports metrics to bard, called when a specific call attempt terminates. | ||
| def tellBard(terminalStateName: String, | ||
|
There was a problem hiding this comment. Moved (almost) verbatim from |
||
| jobStart: OffsetDateTime, | ||
| vmStartTime: Option[OffsetDateTime], | ||
| vmEndTime: OffsetDateTime | ||
| ): Unit | ||
|
|
||
| private var jobStartTime: Option[OffsetDateTime] = | ||
| Option.empty | ||
| private var vmStartTime: Option[OffsetDateTime] = Option.empty | ||
| private var vmEndTime: Option[OffsetDateTime] = Option.empty | ||
|
|
||
| def processPollResult(pollStatus: PollResultType): Unit = { | ||
| // Make sure jobStartTime remains the earliest event time ever seen | ||
| extractEarliestEventTimeFromRunState(pollStatus).foreach { earliestTime => | ||
| if (earliestTime.isBefore(jobStartTime.getOrElse(OffsetDateTime.now()))) { | ||
| jobStartTime = Option(earliestTime) | ||
| } | ||
| } | ||
| // If vm start time is reported, record it to metadata and stop trying | ||
| if (vmStartTime.isEmpty) { | ||
| extractStartTimeFromRunState(pollStatus).foreach { start => | ||
| vmStartTime = Option(start) | ||
| tellMetadata(Map(CallMetadataKeys.VmStartTime -> start)) | ||
| } | ||
| } | ||
| // If vm end time is reported, (or for some weird reason we see an end time after our recorded one), | ||
| // record it to metadata. | ||
| extractEndTimeFromRunState(pollStatus).foreach { end => | ||
| if (vmEndTime.isEmpty || end.isAfter(vmEndTime.get)) { | ||
| vmEndTime = Option(end) | ||
| tellMetadata(Map(CallMetadataKeys.VmEndTime -> end)) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // When a job finishes, the bard actor needs to know about the timing in order to record metrics. | ||
| // Cost related metadata should already have been handled in processPollResult. | ||
| def handleAsyncJobFinish(terminalStateName: String): Unit = | ||
| jobStartTime.foreach(jobStart => | ||
| tellBard(terminalStateName, jobStart, vmStartTime, vmEndTime.getOrElse(OffsetDateTime.now())) | ||
| ) | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| package cromwell.backend.google.pipelines.common | ||
|
|
||
| import akka.actor.Props | ||
| import cromwell.backend.google.pipelines.common.api.RunStatus | ||
| import cromwell.backend.standard.pollmonitoring.{ | ||
| AsyncJobHasFinished, | ||
| PollResultMessage, | ||
| PollResultMonitorActor, | ||
| ProcessThisPollResult | ||
| } | ||
| import cromwell.services.metadata.CallMetadataKeys | ||
|
|
||
| import java.time.OffsetDateTime | ||
|
|
||
| object PapiPollResultMonitorActor { | ||
| def props(tellMetadataFn: Map[String, Any] => Unit, | ||
| tellBardFn: (String, OffsetDateTime, Option[OffsetDateTime], OffsetDateTime) => Unit | ||
| ): Props = Props(new PapiPollResultMonitorActor(tellMetadataFn, tellBardFn)) | ||
| } | ||
|
|
||
| class PapiPollResultMonitorActor(tellMetadataFn: Map[String, Any] => Unit, | ||
| tellBardFn: (String, OffsetDateTime, Option[OffsetDateTime], OffsetDateTime) => Unit | ||
|
There was a problem hiding this comment. Might be worth bringing back the |
||
| ) extends PollResultMonitorActor[RunStatus] { | ||
|
|
||
| override def extractEarliestEventTimeFromRunState(pollStatus: RunStatus): Option[OffsetDateTime] = | ||
| pollStatus.eventList.minByOption(_.offsetDateTime).map(e => e.offsetDateTime) | ||
| override def extractStartTimeFromRunState(pollStatus: RunStatus): Option[OffsetDateTime] = | ||
| pollStatus.eventList.collectFirst { | ||
| case event if event.name == CallMetadataKeys.VmStartTime => event.offsetDateTime | ||
| } | ||
|
|
||
| override def extractEndTimeFromRunState(pollStatus: RunStatus): Option[OffsetDateTime] = | ||
| pollStatus.eventList.collectFirst { | ||
| case event if event.name == CallMetadataKeys.VmEndTime => event.offsetDateTime | ||
| } | ||
|
|
||
| override def tellMetadata(metadata: Map[String, Any]): Unit = tellMetadataFn(metadata) | ||
|
There was a problem hiding this comment. Can we give this actor a handle to the metadata service registry rather than having all the backends pass in (I assume identical?) implementations? Similar question for Bard - can we put the actual message-passing to the Bard service in the parent There was a problem hiding this comment. That is certainly possible, but I'm not sure if that's a big improvement since we would also need to pass in a lot of information about the particular task (runtime attributes, job descriptor, etc...) in order to have all the information necessary to send the message in the right way. All that other stuff doesn't seem relevant to tracking start and end times, yet we still want to send messages that include more context about their task. The passing of the function object is more about capturing necessary context from the parent class than it is about sharing implementation. There was a problem hiding this comment. Yeah, I see what you mean, for Bard. Maybe we could have a method that takes the current inputs and creates and returns a There was a problem hiding this comment. Updated branch to do something along these lines. We no longer pass in callbacks from individual backends, but instead pass in some extra data so that the poll monitors can call their own implementations of |
||
| override def tellBard(terminalStateName: String, | ||
| jobStart: OffsetDateTime, | ||
| vmStartTime: Option[OffsetDateTime], | ||
| vmEndTime: OffsetDateTime | ||
| ): Unit = | ||
| tellBardFn(terminalStateName, jobStart, vmStartTime, vmEndTime) | ||
| override def receive: Receive = { | ||
| case message: PollResultMessage => | ||
| message match { | ||
| case ProcessThisPollResult(pollResult: RunStatus) => processPollResult(pollResult) | ||
| case ProcessThisPollResult(_) => println("Programmer error: Received Poll Result of unknown type.") | ||
|
Check warning on line 48 in supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/PapiPollResultMonitorActor.scala
|
||
| case AsyncJobHasFinished(terminalStateName) => handleAsyncJobFinish(terminalStateName) | ||
| } | ||
| case _ => | ||
| println("Programmer error: Cost Helper received message of type other than CostPollingMessage") | ||
|
Check warning on line 52 in supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/PapiPollResultMonitorActor.scala
|
||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we pass
runStatushere rather than stringifying it?