zuozhiw commented on code in PR #5376:
URL: https://github.com/apache/texera/pull/5376#discussion_r3680509406
##########
amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala:
##########
@@ -178,126 +180,157 @@ class WorkflowService(
new WorkflowContext(workflowId = workflowId, cuid = Some(computingUnitId))
}
+ /** Sets up and launches a workflow execution inside a run-level span so
+ * setup-path logs carry its trace id. The span covers the synchronous
+ * setup and the handoff to async execution via `executeWorkflow()`; it
+ * does not span the full async run. The real execution failure is
+ * recorded onto the current span from `errorHandler`.
+ */
def initExecutionService(
req: WorkflowExecuteRequest,
userOpt: Option[User],
sessionUri: URI
): Unit = {
+ val span = TexeraTracer.tracer
+ .spanBuilder("WorkflowService.initExecutionService")
+ .setAttribute("texera.workflow.id", workflowId.id.toString)
Review Comment:
minor: use standard keys instead of plain string
##########
amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala:
##########
@@ -183,6 +185,25 @@ class WorkflowService(
userOpt: Option[User],
sessionUri: URI
): Unit = {
+ TexeraTracer.withSpan(
+ "workflow.execute",
+ _.setAttribute("texera.workflow.id", workflowId.id.toString)
+ ) { span =>
+ initExecutionServiceSpanned(req, userOpt, sessionUri, span)
Review Comment:
👍
##########
amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala:
##########
@@ -183,6 +185,25 @@ class WorkflowService(
userOpt: Option[User],
sessionUri: URI
): Unit = {
+ TexeraTracer.withSpan(
+ "workflow.execute",
Review Comment:
👍
##########
amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala:
##########
@@ -178,126 +180,157 @@ class WorkflowService(
new WorkflowContext(workflowId = workflowId, cuid = Some(computingUnitId))
}
+ /** Sets up and launches a workflow execution inside a run-level span so
+ * setup-path logs carry its trace id. The span covers the synchronous
+ * setup and the handoff to async execution via `executeWorkflow()`; it
+ * does not span the full async run. The real execution failure is
+ * recorded onto the current span from `errorHandler`.
+ */
def initExecutionService(
req: WorkflowExecuteRequest,
userOpt: Option[User],
sessionUri: URI
): Unit = {
+ val span = TexeraTracer.tracer
+ .spanBuilder("WorkflowService.initExecutionService")
+ .setAttribute("texera.workflow.id", workflowId.id.toString)
+ .startSpan()
+ val scope = span.makeCurrent()
+ try {
- if (executionService.hasValue) {
- executionService.getValue.unsubscribeAll()
- }
+ if (executionService.hasValue) {
+ executionService.getValue.unsubscribeAll()
+ }
- val (uidOpt, userEmailOpt) = userOpt.map(user => (user.getUid,
user.getEmail)).unzip
+ val (uidOpt, userEmailOpt) = userOpt.map(user => (user.getUid,
user.getEmail)).unzip
- // uid is NOT NULL in the DB; fail early here rather than letting the
insert fail downstream.
- val uid = uidOpt.getOrElse(
- throw new IllegalArgumentException(
- "Cannot start execution: a user id (uid) is required but none was
provided."
+ // uid is NOT NULL in the DB; fail early here rather than letting the
insert fail downstream.
+ val uid = uidOpt.getOrElse(
+ throw new IllegalArgumentException(
+ "Cannot start execution: a user id (uid) is required but none was
provided."
+ )
)
- )
-
- val workflowContext: WorkflowContext = createWorkflowContext()
- var coordinatorConf = CoordinatorConfig.default
-
- // clean up results from previous run
- val previousExecutionId =
- WorkflowExecutionService.getLatestExecutionId(workflowId,
req.computingUnitId)
- previousExecutionId.foreach(eid => {
- clearExecutionResources(eid)
- }) // TODO: change this behavior after enabling cache.
-
- workflowContext.executionId =
ExecutionsMetadataPersistService.insertNewExecution(
- workflowContext.workflowId,
- uid,
- req.executionName,
- convertToJson(req.engineVersion),
- req.computingUnitId
- )
- if (ApplicationConfig.faultToleranceLogRootFolder.isDefined) {
- val writeLocation =
ApplicationConfig.faultToleranceLogRootFolder.get.resolve(
- s"${workflowContext.workflowId}/${workflowContext.executionId}/"
+ val workflowContext: WorkflowContext = createWorkflowContext()
+ var coordinatorConf = CoordinatorConfig.default
+
+ // clean up results from previous run
+ val previousExecutionId =
+ WorkflowExecutionService.getLatestExecutionId(workflowId,
req.computingUnitId)
+ previousExecutionId.foreach(eid => {
+ clearExecutionResources(eid)
+ }) // TODO: change this behavior after enabling cache.
+
+ workflowContext.executionId =
ExecutionsMetadataPersistService.insertNewExecution(
+ workflowContext.workflowId,
+ uid,
+ req.executionName,
+ convertToJson(req.engineVersion),
+ req.computingUnitId
)
-
ExecutionsMetadataPersistService.tryUpdateExistingExecution(workflowContext.executionId)
{
- execution => execution.setLogLocation(writeLocation.toString)
+ span.setAttribute("texera.execution.id",
workflowContext.executionId.id.toString)
+ // A run has started: record the start counter and stamp its start time.
+ org.apache.texera.web.observability.WorkflowMetricsRecorder
+ .onStart(workflowContext.executionId)
+
+ if (ApplicationConfig.faultToleranceLogRootFolder.isDefined) {
+ val writeLocation =
ApplicationConfig.faultToleranceLogRootFolder.get.resolve(
+ s"${workflowContext.workflowId}/${workflowContext.executionId}/"
+ )
+
ExecutionsMetadataPersistService.tryUpdateExistingExecution(workflowContext.executionId)
{
+ execution => execution.setLogLocation(writeLocation.toString)
+ }
+ coordinatorConf = coordinatorConf.copy(faultToleranceConfOpt =
+ Some(FaultToleranceConfig(writeTo = writeLocation))
+ )
}
- coordinatorConf = coordinatorConf.copy(faultToleranceConfOpt =
- Some(FaultToleranceConfig(writeTo = writeLocation))
- )
- }
- if (req.replayFromExecution.isDefined) {
- val replayInfo = req.replayFromExecution.get
- ExecutionsMetadataPersistService
- .tryGetExistingExecution(ExecutionIdentity(replayInfo.eid))
- .foreach { execution =>
- val readLocation = new URI(execution.getLogLocation)
- coordinatorConf = coordinatorConf.copy(stateRestoreConfOpt =
- Some(
- StateRestoreConfig(
- readFrom = readLocation,
- replayDestination =
EmbeddedControlMessageIdentity(replayInfo.interaction)
+ if (req.replayFromExecution.isDefined) {
+ val replayInfo = req.replayFromExecution.get
+ ExecutionsMetadataPersistService
+ .tryGetExistingExecution(ExecutionIdentity(replayInfo.eid))
+ .foreach { execution =>
+ val readLocation = new URI(execution.getLogLocation)
+ coordinatorConf = coordinatorConf.copy(stateRestoreConfOpt =
+ Some(
+ StateRestoreConfig(
+ readFrom = readLocation,
+ replayDestination =
EmbeddedControlMessageIdentity(replayInfo.interaction)
+ )
)
)
- )
- }
- }
+ }
+ }
- val executionStateStore = new ExecutionStateStore()
- // assign execution id to find the execution from DB in case the
constructor fails.
- executionStateStore.metadataStore.updateState(state =>
- state.withExecutionId(workflowContext.executionId)
- )
- val errorHandler: Throwable => Unit = { t =>
- {
- val fromActorOpt = t match {
- case ex: WorkflowRuntimeException =>
- ex.relatedWorkerId
- case other =>
- None
- }
- val (operatorId, workerId) = getOperatorFromActorIdOpt(fromActorOpt)
- logger.error("error during execution", t)
- executionStateStore.statsStore.updateState(stats =>
- stats.withEndTimeStamp(System.currentTimeMillis())
- )
- executionStateStore.metadataStore.updateState { metadataStore =>
- updateWorkflowState(FAILED, metadataStore).addFatalErrors(
- WorkflowFatalError(
- EXECUTION_FAILURE,
- Timestamp(Instant.now),
- t.toString,
- getStackTraceWithAllCauses(t),
- operatorId,
- workerId
- )
+ val executionStateStore = new ExecutionStateStore()
+ // assign execution id to find the execution from DB in case the
constructor fails.
+ executionStateStore.metadataStore.updateState(state =>
+ state.withExecutionId(workflowContext.executionId)
+ )
+ val errorHandler: Throwable => Unit = { t =>
+ {
Review Comment:
this error handler needs to be extra careful becaue it's passed around as an
argument to WorkflowExecutionService, and it will be called beyond this
function's lifecycle, we really need to check who calls it. when they call it
this specific span object we get here might already ended after this function
ends. this handler is also used locally, that is fine, but if other places call
it then we have troubles and the error is not attached to the correct span
this one is a bit tricky, as we need to inspect the places that invokes this
error handler and see what spans they have, then we either call span.current()
again inside this span (but need to make sure the callers has good spans...),
or has to pass in a span as optinal argument, please check how we are using
this handler and see what's the best way to handle this, maybe passing such
error handler around is not a good error handling method
##########
amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala:
##########
@@ -183,6 +185,25 @@ class WorkflowService(
userOpt: Option[User],
sessionUri: URI
): Unit = {
+ TexeraTracer.withSpan(
+ "workflow.execute",
+ _.setAttribute("texera.workflow.id", workflowId.id.toString)
+ ) { span =>
Review Comment:
left another comment on the error hanlder callback, please check
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
