Ma77Ball commented on code in PR #5376:
URL: https://github.com/apache/texera/pull/5376#discussion_r4358315418
##########
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:
Switched the attribute keys from plain strings to the shared typed keys
(texera.workflow.id and texera.execution.id).
##########
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:
Kept the shared error handler from writing to the span, since it can run
later after the span has already closed. Setup failures that happen right away
are now recorded on the span by initExecutionService while it is still open.
##########
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:
executeWorkflow now hands back a setup failure that happens right away (like
a compile error that was already shown to the user) instead of swallowing it,
so initExecutionService can mark that run's trace as failed before it closes.
Failures that happen later during the run still go through the shared error
handler as before. The existing compilation-failure test now checks this.
--
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]
