DATA-WF-002 — Bind Resource Monitoring to task executions, not workflows

Field	Value
Status	Open; source-verified
Architecture state	Current schema and API are workflow-selected but only partially task-keyed
Severity	High
Affected surface	Workflow Resources task selector, current metrics, history, and recommendations
Components	sparqd/workflow-resource-monitoring, sparqd/spark-recommendation, sparqd/workflow-backend, sparqd/bff-api, sparqd/oqullus, Airflow, Spark Operator
Associated issue	Not opened

Where the problem appears

Open Resources from a workflow containing multiple Spark tasks or multiple runs of the same task. The modal is selected by Airflow DAG ID and offers a task selector, but it has no workflow-run, task-attempt, or Spark-application selector.

Concrete observed case

The dedicated monitoring record contains dag_id, task_id, snapshot fields, and measurements. It does not contain the managed Workflow UUID, workflow revision, Airflow dag_run_id, task attempt, or Spark application ID.

The Delta writer removes duplicates and merges on:

snapshot_date + dag_id + task_id

Consequently, two executions of the same task on the same date match the same row. A later collection updates the earlier row and increments version; it does not preserve two task executions. Retries are also indistinguishable. The latest API ranks rows per DAG/task, while history and the browser continue to group by task and calendar day, so neither can reconstruct the lost execution identity.

The API path /workflow-resource-monitoring/{dag_id} makes the workflow the lookup boundary, but the actual measured workload is a particular Spark application launched by a particular task attempt.

Expected condition

The canonical monitoring identity is a task execution:

workflow_id
+ workflow_revision
+ workflow_run_id
+ task_id
+ task_attempt
+ spark_application_id
+ observation_id

Each observation also records an authoritative observation timestamp and source. A Workflow page may aggregate these records, but it must require or clearly default a workflow run before presenting task measurements. Non-Spark tasks explicitly show Resource monitoring not applicable.

Recommendations reference the exact task execution and the configuration or workflow revision that produced it. They must not attach only to a mutable workflow or task name.

Cause and evidence

• The collector derives dag_id and task_id, but not a run, attempt, or Spark-application identity in its persisted schema.
• The Delta merge condition is (snapshot_date, dag_id, task_id).
• The response model exposes only DAG/task identity plus snapshot values.
• Oqullus requests monitoring by Airflow DAG ID, selects task_id, and reduces history to the newest record per calendar day.
• Current KCL supplies the monitoring table to Spark Recommendation, so the selected canonical producer must preserve the same execution identity before recommendations can target a trustworthy input.

Why it matters

Users cannot determine which run produced a measurement. Same-day executions, retries, workflow revisions, and renamed workflows can be conflated. A recommendation derived from one execution can then appear applicable to a different task configuration.

Existing controls and workarounds

Treat current records as coarse daily task snapshots, not per-run evidence. Confirm the relevant Airflow run and Spark application separately before using the values for troubleshooting, configuration changes, or cost decisions.

Alternatives considered

1. Recommended: store immutable task-execution observations and derive latest, workflow summaries, and daily rollups as projections.
2. Use (dag_run_id, task_id, task_attempt) without Spark application identity. This improves Airflow traceability but cannot reliably distinguish a relaunched or replaced Spark application within one attempt.
3. Keep one daily DAG/task row and add more version values. Rejected because an update counter does not preserve the overwritten measurements or their execution identity.
4. Store only workflow-level totals. Rejected because workflows can contain multiple Spark tasks and non-Spark tasks with different applicability and configurations.

Component involvement

• sparqd/workflow-resource-monitoring captures and persists immutable task-execution and Spark-application identities, then exposes run-scoped current and bounded-history reads.
• The selected producer in sparqd/spark-recommendation or sparqd/workflow-resource-monitoring writes the canonical schema; it must not maintain a competing identity model.
• sparqd/workflow-backend resolves managed Workflow UUID/revision to the authorized Airflow run and task attempt.
• sparqd/bff-api exposes the managed, authorized run/task resource path.
• sparqd/oqullus adds run context, marks non-Spark tasks not applicable, and treats workflow totals as derived aggregation.
• Airflow and Spark Operator supply dag_run_id, task attempt, and Spark application identity at execution time.

Reproduction and verification

Follow RM-S006.

Resolution conditions

Close when two same-day runs and a retried task produce distinct immutable records, each record resolves to one managed workflow revision and Spark application, the UI selects the intended run/task execution, non-Spark tasks show not applicable, and recommendations preserve the same identity.

Validation and blockers

Workflow backend, Airflow, Spark, monitoring, recommendation, frontend, data, and security reviewers must validate the canonical identifiers, lifecycle, and retention. Migration behavior for existing daily DAG/task rows remains open.