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:
A synchronous setup failure can leave the setup span without an error status
or exception, even though the workflow failed. For example, if compilation
throws, the UI receives the error through `errorHandler`, but the trace will
not identify setup as the failing operation. Searching traces for errors can
then miss this failure.
There are two paths that bypass the outer catch's `recordException` and
`ERROR` status:
- `executeWorkflow()` catches compilation failures, calls `errorHandler`,
and returns.
- This inner catch also calls `errorHandler` without rethrowing.
Please separate synchronous setup failures from later runtime callbacks.
Record synchronous failures on the setup span while it is still open,
preserving the existing UI/state updates. Later failures belong on an
execution/worker span that remains open for that work.
Avoid adding span writes back to the shared `errorHandler`: a worker can
call it after setup has returned and its span has ended. Explicit setup and
execution lifetimes, with the correct context carried into callbacks, avoid
both losing these errors and attaching them to the wrong operation.
--
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]
