zuozhiw commented on code in PR #5376:
URL: https://github.com/apache/texera/pull/5376#discussion_r4346754018
##########
amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala:
##########
@@ -229,126 +231,152 @@ class WorkflowService(
userOpt: Option[User],
sessionUri: URI
): Unit = {
+ val span = TexeraTracer.tracer
+ .spanBuilder("WorkflowService.initExecutionService")
+ .setAttribute(SpanAttrs.WorkflowId, Long.box(workflowId.id))
+ .startSpan()
+ val scope = span.makeCurrent()
+ try {
- val (uidOpt, userEmailOpt) = userOpt.map(user => (user.getUid,
user.getEmail)).unzip
+ val (uidOpt, userEmailOpt) = userOpt.map(user => (user.getUid,
user.getEmail)).unzip
- // Validate before touching the execution already in flight: a request
that is
- // going to be refused must not take the running one's subscriptions with
it.
- // 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."
+ // Validate before touching the execution already in flight: a request
that is
+ // going to be refused must not take the running one's subscriptions
with it.
+ // 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 warehouseName =
WorkflowService.resolveLakekeeperWarehouseName(req.warehouseId, uid)
+ val warehouseName =
WorkflowService.resolveLakekeeperWarehouseName(req.warehouseId, uid)
- if (executionService.hasValue) {
- executionService.getValue.unsubscribeAll()
- }
-
- val workflowContext: WorkflowContext = createWorkflowContext()
- workflowContext.warehouse = warehouseName
- 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,
- req.warehouseId
- )
+ if (executionService.hasValue) {
+ executionService.getValue.unsubscribeAll()
+ }
- if (ApplicationConfig.faultToleranceLogRootFolder.isDefined) {
- val writeLocation =
ApplicationConfig.faultToleranceLogRootFolder.get.resolve(
- s"${workflowContext.workflowId}/${workflowContext.executionId}/"
+ val workflowContext: WorkflowContext = createWorkflowContext()
+ workflowContext.warehouse = warehouseName
+ 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,
+ req.warehouseId
)
-
ExecutionsMetadataPersistService.tryUpdateExistingExecution(workflowContext.executionId)
{
- execution => execution.setLogLocation(writeLocation.toString)
+ span.setAttribute(SpanAttrs.ExecutionId,
Long.box(workflowContext.executionId.id))
+ // 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 =>
+ {
+ val fromActorOpt = t match {
+ case ex: WorkflowRuntimeException =>
+ ex.relatedWorkerId
+ case other =>
+ None
+ }
+ val (operatorId, workerId) = getOperatorFromActorIdOpt(fromActorOpt)
+ logger.error("error during execution", t)
+ // Do NOT touch `span` here: this handler is passed into
+ // WorkflowExecutionService and invoked asynchronously (runtime,
+ // websocket, startWorkflow callbacks) after initExecutionService has
+ // returned and the setup span has already ended, so recording onto
it
+ // would be a silent no-op. The failure is surfaced via the metadata
+ // store below; setup-span errors are recorded in the catch block.
+ 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
+ )
+ )
+ }
}
}
- }
- // WorkflowExecutionService construction does no external work and cannot
- // throw; it registers its error/state diff handler up front. Once
published
- // via `executionService.onNext`, any failure in `executeWorkflow()` is
- // recorded by `errorHandler` into the metadata store, whose handler emits
a
- // WorkflowErrorEvent that `connectToExecution` forwards.
- try {
- val execution = new WorkflowExecutionService(
- coordinatorConf,
- workflowContext,
- resultService,
- req,
- executionStateStore,
- errorHandler,
- userEmailOpt,
- sessionUri
- )
- lifeCycleManager.registerCleanUpOnStateChange(executionStateStore)
- executionService.onNext(execution)
- execution.executeWorkflow()
+ // WorkflowExecutionService construction does no external work and cannot
+ // throw; it registers its error/state diff handler up front. Once
published
+ // via `executionService.onNext`, any failure in `executeWorkflow()` is
+ // recorded by `errorHandler` into the metadata store, whose handler
emits a
+ // WorkflowErrorEvent that `connectToExecution` forwards.
+ try {
+ val execution = new WorkflowExecutionService(
+ coordinatorConf,
+ workflowContext,
+ resultService,
+ req,
+ executionStateStore,
+ errorHandler,
+ userEmailOpt,
+ sessionUri
+ )
+ lifeCycleManager.registerCleanUpOnStateChange(executionStateStore)
+ executionService.onNext(execution)
+ execution.executeWorkflow()
+ } catch {
+ case e: Throwable => errorHandler(e)
Review Comment:
Avoiding writes to an ended setup span is the right change, but synchronous
setup errors now miss span error recording too. From reading the code:
- `executeWorkflow()` catches compilation failures, calls `errorHandler`,
and returns.
- This inner catch also calls `errorHandler` without rethrowing.
Neither path reaches the outer catch that records the exception and sets
`ERROR`, so failed setup can leave its span unmarked. I have not reproduced
this with a workflow test.
Could we separate synchronous setup failures from later runtime callbacks?
Record synchronous failures on the live setup span while preserving UI/state
updates. Record later failures on the execution/worker span that is still
active. This is a common async tracing pitfall; giving setup and execution
spans separate, explicit lifetimes should prevent callbacks from using an ended
span.
--
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]
