diff --git a/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java b/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java index eefb5e52a3..a9d525d364 100644 --- a/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java +++ b/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java @@ -507,6 +507,9 @@ protected void handleJobTimeout(JobContext jobCtx, WorkflowContext workflowCtx, _clusterStatusMonitor.updateJobCounters(jobCfg, TaskState.TIMED_OUT); _rebalanceScheduler.removeScheduledRebalance(jobResource); TaskUtil.cleanupJobIdealStateExtView(_manager.getHelixDataAccessor(), jobResource); + // Aggregate the terminal state of every task into a job-level summary (computed after INIT + // tasks are marked TASK_ABORTED above so the counts reflect the final states). + jobCtx.updateTaskStatusSummary(); // New pipeline trigger for workflow status update // TODO: Enhance the pipeline and remove this because this operation is expansive RebalanceUtil.scheduleOnDemandPipeline(_manager.getClusterName(),0L,false); @@ -529,6 +532,10 @@ protected void failJob(String jobName, WorkflowContext workflowContext, JobConte // New pipeline trigger for workflow status update // TODO: Enhance the pipeline and remove this because this operation is expansive RebalanceUtil.scheduleOnDemandPipeline(_manager.getClusterName(),0L,false); + + // Aggregate the terminal state of every task into a job-level summary (computed after INIT + // tasks are marked TASK_ABORTED above so the counts reflect the final states). + jobContext.updateTaskStatusSummary(); } // Compute real assignment from theoretical calculation with applied throttling @@ -935,6 +942,10 @@ protected void markJobComplete(final String jobName, final JobContext jobContext reportControllerInducedDelay(dataProvider, _clusterStatusMonitor, workflowConfig, jobConfig, currentTime); } + + // Aggregate the terminal state of every task into a job-level summary so that partial failures + // remain visible even though the job itself is COMPLETED (e.g. FailureThreshold set high). + jobContext.updateTaskStatusSummary(); } protected void markJobFailed(String jobName, JobContext jobContext, WorkflowConfig workflowConfig, diff --git a/helix-core/src/main/java/org/apache/helix/task/JobContext.java b/helix-core/src/main/java/org/apache/helix/task/JobContext.java index 83ec551748..7f28acb37c 100644 --- a/helix-core/src/main/java/org/apache/helix/task/JobContext.java +++ b/helix-core/src/main/java/org/apache/helix/task/JobContext.java @@ -20,22 +20,48 @@ */ import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; import java.util.TreeMap; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.Lists; import com.google.common.collect.Maps; import com.google.common.collect.Sets; import org.apache.helix.HelixProperty; import org.apache.helix.zookeeper.datamodel.ZNRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Provides a typed interface to the context information stored by {@link TaskRebalancer} in the * Helix property store. */ public class JobContext extends HelixProperty { + private static final Logger LOG = LoggerFactory.getLogger(JobContext.class); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + // Field names used in the aggregated task status summary JSON produced by + // updateTaskStatusSummary(). Kept as named constants (rather than inline string literals) so the + // producer here and every consumer that mirrors this shape stay in agreement on the exact keys. + static final String SUMMARY_KEY_TOTAL = "total"; + static final String SUMMARY_KEY_COMPLETED = "completed"; + static final String SUMMARY_KEY_FAILED = "failed"; + static final String SUMMARY_KEY_TIMED_OUT = "timedOut"; + static final String SUMMARY_KEY_IN_PROGRESS = "inProgress"; + static final String SUMMARY_KEY_PENDING = "pending"; + static final String SUMMARY_KEY_OTHER = "other"; + static final String SUMMARY_KEY_BY_STATE = "byState"; + static final String SUMMARY_KEY_FAILED_TASKS = "failedTasks"; + static final String SUMMARY_KEY_TIMED_OUT_TASKS = "timedOutTasks"; + static final String SUMMARY_KEY_IN_PROGRESS_TASKS = "inProgressTasks"; + static final String SUMMARY_KEY_PENDING_TASKS = "pendingTasks"; + // byState bucket used for a partition that has no task state yet (getPartitionState == null). + static final String SUMMARY_STATE_UNSCHEDULED = "UNSCHEDULED"; + private enum ContextProperties { START_TIME, // Time at which this JobContext was created STATE, @@ -48,6 +74,9 @@ private enum ContextProperties { INFO, NAME, EXECUTION_START_TIME, // Time at which the first task of this job got scheduled + TASK_STATUS_SUMMARY, // Aggregated per-task terminal status summary (JSON), set when the job + // reaches a terminal state. Surfaces partial failures even when the job + // itself is COMPLETED (e.g. FailureThreshold set high so all tasks run). } // Note: This field needs to be set if any of the job context fields have been changed. @@ -337,6 +366,131 @@ public long getExecutionStartTime() { return Long.parseLong(tStr); } + /** + * Task partition states that represent a terminal failure (the task was given up and will not be + * retried, or it errored/timed out). Used to compute the aggregated task status summary. + */ + private static boolean isFailedState(TaskPartitionState state) { + return state == TaskPartitionState.TASK_ERROR || state == TaskPartitionState.TASK_ABORTED + || state == TaskPartitionState.TIMED_OUT || state == TaskPartitionState.ERROR; + } + + /** + * Task partition state for a task that is actively running (RUNNING). Surfaced explicitly instead + * of being lumped into "other" so still-running tasks are visible, which matters because the + * summary is refreshed live while the job runs, not only when it reaches a terminal state. + */ + private static boolean isInProgressState(TaskPartitionState state) { + return state == TaskPartitionState.RUNNING; + } + + /** + * Task partition state for a task that has been scheduled but has not started running yet (INIT). + * Surfaced as its own "pending" count, distinct from "inProgress" (RUNNING), so operators can + * tell tasks that are waiting to start apart from tasks that are actively executing. + */ + private static boolean isPendingState(TaskPartitionState state) { + return state == TaskPartitionState.INIT; + } + + /** + * Aggregates the state of every scheduled task in this job into a compact, job-level summary and + * stores it as a simple field (JSON). This lets operators observe partial failures even when the + * job's own status flag is COMPLETED, which happens when FailureThreshold is set high so that all + * partition tasks are allowed to run to completion. Intended to be called by the controller when + * the job reaches a terminal state (completed / failed / timed out). + * The summary shape is: + * {@code {"total":N,"completed":X,"failed":Y,"timedOut":T,"inProgress":R,"pending":P,"other":Z, + * "byState":{...},"failedTasks":[...],"timedOutTasks":[...],"inProgressTasks":[...], + * "pendingTasks":[...]}}. + * The top-level counts partition the tasks as {@code total = completed + failed + inProgress + + * pending + other}. {@code failed} is the count of tasks that did not succeed (errored, aborted, + * or timed out); {@code timedOut} is the subset of {@code failed} that specifically timed out, + * called out separately so operators can distinguish a task that ran out of time from one that + * errored or aborted. {@code inProgress} is the count of tasks actively running (RUNNING) and + * {@code pending} is the count of tasks scheduled but not yet started (INIT); both are pulled out + * of {@code other} so still-active tasks are visible when the summary is recomputed on demand + * (for example when the job detail page is opened) while the job is still running. + * {@code byState} is a histogram mapping each raw task partition state name (or + * {@code "UNSCHEDULED"} for a partition that has not been assigned a state yet) to the number of + * tasks in that state; it is the fine-grained breakdown behind the coarser counts above. The + * {@code failedTasks} / {@code timedOutTasks} / {@code inProgressTasks} / {@code pendingTasks} + * arrays list the partition ids that fall into each of those buckets. + */ + public void updateTaskStatusSummary() { + Set partitions = getPartitionSet(); + Map byState = new TreeMap<>(); + List failedTasks = Lists.newArrayList(); + List timedOutTasks = Lists.newArrayList(); + List inProgressTasks = Lists.newArrayList(); + List pendingTasks = Lists.newArrayList(); + int completed = 0; + int failed = 0; + int timedOut = 0; + int inProgress = 0; + int pending = 0; + for (int p : partitions) { + TaskPartitionState state = getPartitionState(p); + String key = (state == null) ? SUMMARY_STATE_UNSCHEDULED : state.name(); + byState.merge(key, 1, Integer::sum); + if (state == TaskPartitionState.COMPLETED) { + completed++; + } else if (isFailedState(state)) { + failed++; + failedTasks.add(p); + if (state == TaskPartitionState.TIMED_OUT) { + timedOut++; + timedOutTasks.add(p); + } + } else if (isInProgressState(state)) { + inProgress++; + inProgressTasks.add(p); + } else if (isPendingState(state)) { + pending++; + pendingTasks.add(p); + } + } + failedTasks.sort(null); + timedOutTasks.sort(null); + inProgressTasks.sort(null); + pendingTasks.sort(null); + + Map summary = new LinkedHashMap<>(); + summary.put(SUMMARY_KEY_TOTAL, partitions.size()); + summary.put(SUMMARY_KEY_COMPLETED, completed); + summary.put(SUMMARY_KEY_FAILED, failed); + summary.put(SUMMARY_KEY_TIMED_OUT, timedOut); + summary.put(SUMMARY_KEY_IN_PROGRESS, inProgress); + summary.put(SUMMARY_KEY_PENDING, pending); + summary.put(SUMMARY_KEY_OTHER, partitions.size() - completed - failed - inProgress - pending); + summary.put(SUMMARY_KEY_BY_STATE, byState); + summary.put(SUMMARY_KEY_FAILED_TASKS, failedTasks); + summary.put(SUMMARY_KEY_TIMED_OUT_TASKS, timedOutTasks); + summary.put(SUMMARY_KEY_IN_PROGRESS_TASKS, inProgressTasks); + summary.put(SUMMARY_KEY_PENDING_TASKS, pendingTasks); + + try { + String json = OBJECT_MAPPER.writeValueAsString(summary); + if (!json.equals(getTaskStatusSummary())) { + _record.setSimpleField(ContextProperties.TASK_STATUS_SUMMARY.name(), json); + markJobContextAsModified(); + } + } catch (JsonProcessingException e) { + // The summary is a best-effort convenience field; never let it disrupt the job status + // update path. + LOG.warn("Failed to serialize task status summary for job {}", getName(), e); + } + } + + /** + * @return the aggregated per-task status summary as a JSON string, or null if it has not been + * computed yet (i.e. the job has not reached a terminal state). See + * {@link #updateTaskStatusSummary()} for the shape. + */ + public String getTaskStatusSummary() { + return _record.getSimpleField(ContextProperties.TASK_STATUS_SUMMARY.name()); + } + /** * Get MapField for the given partition. * @param p diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestJobTaskStatusSummary.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestJobTaskStatusSummary.java new file mode 100644 index 0000000000..c7d1b9534a --- /dev/null +++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestJobTaskStatusSummary.java @@ -0,0 +1,144 @@ +package org.apache.helix.integration.task; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import java.util.ArrayList; +import java.util.List; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.helix.TestHelper; +import org.apache.helix.task.JobConfig; +import org.apache.helix.task.JobContext; +import org.apache.helix.task.TaskConfig; +import org.apache.helix.task.TaskPartitionState; +import org.apache.helix.task.TaskResult; +import org.apache.helix.task.TaskState; +import org.apache.helix.task.TaskUtil; +import org.apache.helix.task.Workflow; +import org.testng.Assert; +import org.testng.annotations.Test; + +/** + * Verifies the aggregated per-task status summary that Helix writes into the JobContext when a job + * reaches a terminal state. The scenario mirrors a common validation setup: one job per table, one + * task per partition, and a high FailureThreshold so every task runs even if some fail. In that + * setup the job's own status flag is COMPLETED, which previously masked partition level failures. + * The summary must surface those failures. + */ +public class TestJobTaskStatusSummary extends TaskTestBase { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + @Test + public void testSummaryReflectsPartialFailureOnCompletedJob() throws Exception { + int numTasks = 6; + // Indices of the tasks that should fail terminally. + final int fatalTask = 1; // FATAL_FAILED -> given up immediately -> TASK_ABORTED + final int exceptionTask = 3; // throws -> retried to exhaustion -> TASK_ERROR + final int expectedFailed = 2; + final int expectedCompleted = numTasks - expectedFailed; + + String jobResource = TestHelper.getTestMethodName(); + JobConfig.Builder jobBuilder = new JobConfig.Builder(); + // FailureThreshold high (a common operator workaround) so all tasks run and the job still + // ends COMPLETED. + jobBuilder.setCommand(MockTask.TASK_COMMAND).setTimeoutPerTask(10000).setMaxAttemptsPerTask(2) + .setFailureThreshold(Integer.MAX_VALUE); + + List taskConfigs = new ArrayList<>(); + for (int j = 0; j < numTasks; j++) { + TaskConfig.Builder configBuilder = new TaskConfig.Builder().setTaskId("task_" + j); + if (j == fatalTask) { + configBuilder.addConfig(MockTask.TASK_RESULT_STATUS, TaskResult.Status.FATAL_FAILED.name()); + } else if (j == exceptionTask) { + configBuilder.addConfig(MockTask.THROW_EXCEPTION, Boolean.TRUE.toString()); + } + configBuilder.setTargetPartition(String.valueOf(j)); + taskConfigs.add(configBuilder.build()); + } + jobBuilder.addTaskConfigs(taskConfigs); + + Workflow flow = + WorkflowGenerator.generateSingleJobWorkflowBuilder(jobResource, jobBuilder).build(); + _driver.start(flow); + + // The job completes even though tasks failed, because FailureThreshold is high. + _driver.pollForWorkflowState(jobResource, TaskState.COMPLETED); + + String namespacedJob = TaskUtil.getNamespacedJobName(jobResource); + Assert.assertEquals(_driver.getWorkflowContext(jobResource).getJobState(namespacedJob), + TaskState.COMPLETED, "Job status flag should be COMPLETED (this is what masked failures)."); + + JobContext ctx = _driver.getJobContext(namespacedJob); + + // Cross-check the ground truth directly from per-partition states. + int actualCompleted = 0; + int actualFailed = 0; + for (int pId : ctx.getPartitionSet()) { + TaskPartitionState state = ctx.getPartitionState(pId); + if (state == TaskPartitionState.COMPLETED) { + actualCompleted++; + } else if (state == TaskPartitionState.TASK_ABORTED || state == TaskPartitionState.TASK_ERROR + || state == TaskPartitionState.TIMED_OUT || state == TaskPartitionState.ERROR) { + actualFailed++; + } + } + Assert.assertEquals(actualCompleted, expectedCompleted); + Assert.assertEquals(actualFailed, expectedFailed); + + // Now the crux: the summary must be present and must match the ground truth. + String summaryJson = ctx.getTaskStatusSummary(); + Assert.assertNotNull(summaryJson, "Task status summary should be populated on a terminal job."); + JsonNode summary = OBJECT_MAPPER.readTree(summaryJson); + + Assert.assertEquals(summary.get("total").asInt(), numTasks); + Assert.assertEquals(summary.get("completed").asInt(), expectedCompleted); + Assert.assertEquals(summary.get("failed").asInt(), expectedFailed); + Assert.assertEquals(summary.get("other").asInt(), 0); + Assert.assertEquals(summary.get("failedTasks").size(), expectedFailed); + + // No task timed out or stayed in-flight in this scenario, but the counts must be present so + // operators can rely on them being part of the summary shape. + Assert.assertEquals(summary.get("timedOut").asInt(), 0); + Assert.assertEquals(summary.get("inProgress").asInt(), 0); + Assert.assertEquals(summary.get("pending").asInt(), 0); + Assert.assertEquals(summary.get("timedOutTasks").size(), 0); + Assert.assertEquals(summary.get("inProgressTasks").size(), 0); + Assert.assertEquals(summary.get("pendingTasks").size(), 0); + // The top-level counts partition the tasks: + // total = completed + failed + inProgress + pending + other. + Assert.assertEquals(summary.get("completed").asInt() + summary.get("failed").asInt() + + summary.get("inProgress").asInt() + summary.get("pending").asInt() + + summary.get("other").asInt(), numTasks); + + JsonNode byState = summary.get("byState"); + Assert.assertEquals(byState.get(TaskPartitionState.COMPLETED.name()).asInt(), expectedCompleted); + Assert.assertEquals(byState.get(TaskPartitionState.TASK_ABORTED.name()).asInt(), 1); + Assert.assertEquals(byState.get(TaskPartitionState.TASK_ERROR.name()).asInt(), 1); + + // The failed partition ids reported in the summary must actually be failed partitions. + for (JsonNode failedPidNode : summary.get("failedTasks")) { + TaskPartitionState state = ctx.getPartitionState(failedPidNode.asInt()); + Assert.assertTrue(state != TaskPartitionState.COMPLETED, + "Partition " + failedPidNode.asInt() + " reported as failed but state is " + state); + } + } +} diff --git a/helix-core/src/test/java/org/apache/helix/task/TestJobContextTaskStatusSummary.java b/helix-core/src/test/java/org/apache/helix/task/TestJobContextTaskStatusSummary.java new file mode 100644 index 0000000000..ba0fb52237 --- /dev/null +++ b/helix-core/src/test/java/org/apache/helix/task/TestJobContextTaskStatusSummary.java @@ -0,0 +1,115 @@ +package org.apache.helix.task; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.helix.zookeeper.datamodel.ZNRecord; +import org.testng.Assert; +import org.testng.annotations.Test; + +/** + * Deterministic unit test for {@link JobContext#updateTaskStatusSummary()}. It builds a JobContext + * with a fixed mix of per-partition task states and asserts the aggregated summary buckets every + * state correctly, without needing a live cluster. This is the same aggregation the job detail page + * recomputes on demand from the per-partition states, so it locks the contract both consumers rely + * on. + */ +public class TestJobContextTaskStatusSummary { + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + @Test + public void testSummaryBucketsEveryState() throws Exception { + JobContext ctx = new JobContext(new ZNRecord("TestJob")); + // A representative mix covering every summary bucket. + ctx.setPartitionState(0, TaskPartitionState.COMPLETED); + ctx.setPartitionState(1, TaskPartitionState.COMPLETED); + ctx.setPartitionState(2, TaskPartitionState.COMPLETED); + ctx.setPartitionState(3, TaskPartitionState.TASK_ERROR); + ctx.setPartitionState(4, TaskPartitionState.TASK_ABORTED); + ctx.setPartitionState(5, TaskPartitionState.TIMED_OUT); + ctx.setPartitionState(6, TaskPartitionState.RUNNING); + ctx.setPartitionState(7, TaskPartitionState.INIT); + ctx.setPartitionState(8, TaskPartitionState.STOPPED); + + ctx.updateTaskStatusSummary(); + + String summaryJson = ctx.getTaskStatusSummary(); + Assert.assertNotNull(summaryJson, "Summary should be populated after updateTaskStatusSummary()."); + JsonNode summary = OBJECT_MAPPER.readTree(summaryJson); + + Assert.assertEquals(summary.get("total").asInt(), 9); + Assert.assertEquals(summary.get("completed").asInt(), 3); + // failed = TASK_ERROR + TASK_ABORTED + TIMED_OUT + Assert.assertEquals(summary.get("failed").asInt(), 3); + // timedOut is the subset of failed that specifically timed out. + Assert.assertEquals(summary.get("timedOut").asInt(), 1); + Assert.assertEquals(summary.get("inProgress").asInt(), 1); + Assert.assertEquals(summary.get("pending").asInt(), 1); + // other = STOPPED only. + Assert.assertEquals(summary.get("other").asInt(), 1); + + // The top-level counts partition the tasks. + Assert.assertEquals(summary.get("completed").asInt() + summary.get("failed").asInt() + + summary.get("inProgress").asInt() + summary.get("pending").asInt() + + summary.get("other").asInt(), 9); + + assertIntList(summary.get("failedTasks"), 3, 4, 5); + assertIntList(summary.get("timedOutTasks"), 5); + assertIntList(summary.get("inProgressTasks"), 6); + assertIntList(summary.get("pendingTasks"), 7); + + JsonNode byState = summary.get("byState"); + Assert.assertEquals(byState.get(TaskPartitionState.COMPLETED.name()).asInt(), 3); + Assert.assertEquals(byState.get(TaskPartitionState.TASK_ERROR.name()).asInt(), 1); + Assert.assertEquals(byState.get(TaskPartitionState.TASK_ABORTED.name()).asInt(), 1); + Assert.assertEquals(byState.get(TaskPartitionState.TIMED_OUT.name()).asInt(), 1); + Assert.assertEquals(byState.get(TaskPartitionState.RUNNING.name()).asInt(), 1); + Assert.assertEquals(byState.get(TaskPartitionState.INIT.name()).asInt(), 1); + Assert.assertEquals(byState.get(TaskPartitionState.STOPPED.name()).asInt(), 1); + } + + @Test + public void testSummaryOnAllCompletedJob() throws Exception { + JobContext ctx = new JobContext(new ZNRecord("TestJob")); + for (int p = 0; p < 4; p++) { + ctx.setPartitionState(p, TaskPartitionState.COMPLETED); + } + ctx.updateTaskStatusSummary(); + + JsonNode summary = OBJECT_MAPPER.readTree(ctx.getTaskStatusSummary()); + Assert.assertEquals(summary.get("total").asInt(), 4); + Assert.assertEquals(summary.get("completed").asInt(), 4); + Assert.assertEquals(summary.get("failed").asInt(), 0); + Assert.assertEquals(summary.get("timedOut").asInt(), 0); + Assert.assertEquals(summary.get("inProgress").asInt(), 0); + Assert.assertEquals(summary.get("pending").asInt(), 0); + Assert.assertEquals(summary.get("other").asInt(), 0); + Assert.assertEquals(summary.get("failedTasks").size(), 0); + Assert.assertEquals(summary.get("pendingTasks").size(), 0); + } + + private static void assertIntList(JsonNode arrayNode, int... expected) { + Assert.assertEquals(arrayNode.size(), expected.length); + for (int i = 0; i < expected.length; i++) { + Assert.assertEquals(arrayNode.get(i).asInt(), expected[i]); + } + } +} diff --git a/helix-front/src/app/workflow/job-detail/job-detail.component.html b/helix-front/src/app/workflow/job-detail/job-detail.component.html index 8409e72e7a..41bbc8cf8a 100644 --- a/helix-front/src/app/workflow/job-detail/job-detail.component.html +++ b/helix-front/src/app/workflow/job-detail/job-detail.component.html @@ -18,6 +18,53 @@ --> + + +
+ +
+ Total: {{ taskSummary.total }} + Completed: {{ taskSummary.completed }} + Failed: {{ taskSummary.failed }} + Timed Out: {{ taskSummary.timedOut || 0 }} + In Progress: {{ taskSummary.inProgress }} + Pending: {{ taskSummary.pending }} + Other: {{ taskSummary.other }} +
+
+ Failed task partitions: + {{ taskSummary.failedTasks?.join(', ') }} +
+
+ Timed out task partitions: + {{ taskSummary.timedOutTasks?.join(', ') }} +
+
+ In progress task partitions: + {{ taskSummary.inProgressTasks?.join(', ') }} +
+
+ Pending task partitions: + {{ taskSummary.pendingTasks?.join(', ') }} +
+ + + +
+ No task status summary available yet. It is computed from the job's per-task states each + time this page is opened, and appears once the job has scheduled its tasks. +
+ +
+ diff --git a/helix-front/src/app/workflow/job-detail/job-detail.component.scss b/helix-front/src/app/workflow/job-detail/job-detail.component.scss index e69de29bb2..53c3e8c59e 100644 --- a/helix-front/src/app/workflow/job-detail/job-detail.component.scss +++ b/helix-front/src/app/workflow/job-detail/job-detail.component.scss @@ -0,0 +1,69 @@ +.task-summary { + padding: 16px; + + .summary-counts { + display: flex; + flex-wrap: wrap; + gap: 8px; + margin-bottom: 12px; + + .chip { + padding: 4px 10px; + border-radius: 12px; + font-size: 13px; + background: #eeeeee; + + &.completed { + background: #e6f4ea; + color: #1e7e34; + } + + &.failed.has-failures { + background: #fdecea; + color: #c5221f; + font-weight: 600; + } + + &.timed-out.has-timed-out { + background: #fef7e0; + color: #b06000; + font-weight: 600; + } + + &.in-progress { + background: #e8f0fe; + color: #1967d2; + } + + &.pending { + background: #f1f3f4; + color: #5f6368; + } + } + } + + .failed-tasks { + margin-bottom: 12px; + color: #c5221f; + } + + .timed-out-tasks { + margin-bottom: 12px; + color: #b06000; + } + + .in-progress-tasks { + margin-bottom: 12px; + color: #1967d2; + } + + .pending-tasks { + margin-bottom: 12px; + color: #5f6368; + } + + .no-summary { + color: rgba(0, 0, 0, 0.54); + font-style: italic; + } +} diff --git a/helix-front/src/app/workflow/job-detail/job-detail.component.spec.ts b/helix-front/src/app/workflow/job-detail/job-detail.component.spec.ts index 8a151f9a49..c1f28edf09 100644 --- a/helix-front/src/app/workflow/job-detail/job-detail.component.spec.ts +++ b/helix-front/src/app/workflow/job-detail/job-detail.component.spec.ts @@ -48,4 +48,132 @@ describe('JobDetailComponent', () => { it('should create', () => { expect(component).toBeTruthy(); }); + + it('taskSummary should be null when there is no summary field', () => { + component.job = { context: { simpleFields: {} } } as any; + expect(component.taskSummary).toBeNull(); + expect(component.hasTaskFailures).toBe(false); + }); + + it('taskSummary should parse the TASK_STATUS_SUMMARY JSON string', () => { + component.job = { + context: { + simpleFields: { + TASK_STATUS_SUMMARY: + '{"total":6,"completed":4,"failed":2,"timedOut":0,"inProgress":0,"other":0,"byState":{"COMPLETED":4,"TASK_ABORTED":1,"TASK_ERROR":1},"failedTasks":[3,5],"timedOutTasks":[],"inProgressTasks":[]}', + }, + }, + } as any; + const summary = component.taskSummary; + expect(summary).not.toBeNull(); + expect(summary.total).toBe(6); + expect(summary.completed).toBe(4); + expect(summary.failed).toBe(2); + expect(summary.timedOut).toBe(0); + expect(summary.inProgress).toBe(0); + expect(summary.failedTasks).toEqual([3, 5]); + expect(component.hasTaskFailures).toBe(true); + expect(component.hasTimedOut).toBe(false); + expect(component.hasInProgress).toBe(false); + }); + + it('taskSummary should expose timed out and in progress counts', () => { + component.job = { + context: { + simpleFields: { + TASK_STATUS_SUMMARY: + '{"total":5,"completed":2,"failed":2,"timedOut":1,"inProgress":1,"other":0,"byState":{"COMPLETED":2,"TASK_ERROR":1,"TIMED_OUT":1,"RUNNING":1},"failedTasks":[1,4],"timedOutTasks":[4],"inProgressTasks":[3]}', + }, + }, + } as any; + const summary = component.taskSummary; + expect(summary).not.toBeNull(); + expect(summary.timedOut).toBe(1); + expect(summary.inProgress).toBe(1); + expect(summary.timedOutTasks).toEqual([4]); + expect(summary.inProgressTasks).toEqual([3]); + // A timed-out task is still a failure, so failure detection stays true. + expect(component.hasTaskFailures).toBe(true); + expect(component.hasTimedOut).toBe(true); + expect(component.hasInProgress).toBe(true); + }); + + it('taskSummary should return null for malformed JSON', () => { + component.job = { + context: { simpleFields: { TASK_STATUS_SUMMARY: 'not-json' } }, + } as any; + expect(component.taskSummary).toBeNull(); + }); + + it('taskSummary should be computed live from per-partition states on page entry', () => { + component.job = { + context: { + simpleFields: {}, + mapFields: { + '0': { STATE: 'COMPLETED' }, + '1': { STATE: 'COMPLETED' }, + '2': { STATE: 'TASK_ERROR' }, + '3': { STATE: 'TIMED_OUT' }, + '4': { STATE: 'RUNNING' }, + '5': { STATE: 'INIT' }, + }, + }, + } as any; + const summary = component.taskSummary; + expect(summary).not.toBeNull(); + expect(summary.total).toBe(6); + expect(summary.completed).toBe(2); + expect(summary.failed).toBe(2); + expect(summary.timedOut).toBe(1); + expect(summary.inProgress).toBe(1); + expect(summary.pending).toBe(1); + expect(summary.other).toBe(0); + expect(summary.failedTasks).toEqual([2, 3]); + expect(summary.timedOutTasks).toEqual([3]); + expect(summary.inProgressTasks).toEqual([4]); + expect(summary.pendingTasks).toEqual([5]); + expect(component.hasTaskFailures).toBe(true); + expect(component.hasTimedOut).toBe(true); + expect(component.hasInProgress).toBe(true); + expect(component.hasPending).toBe(true); + }); + + it('taskSummary should prefer live per-partition states over a stale stored snapshot', () => { + component.job = { + context: { + // A stale snapshot that no longer matches the current per-partition states. + simpleFields: { + TASK_STATUS_SUMMARY: + '{"total":2,"completed":2,"failed":0,"timedOut":0,"inProgress":0,"pending":0,"other":0,"byState":{"COMPLETED":2},"failedTasks":[],"timedOutTasks":[],"inProgressTasks":[],"pendingTasks":[]}', + }, + mapFields: { + '0': { STATE: 'COMPLETED' }, + '1': { STATE: 'RUNNING' }, + }, + }, + } as any; + const summary = component.taskSummary; + // The live compute (1 running) must win over the stale snapshot (all completed). + expect(summary.total).toBe(2); + expect(summary.completed).toBe(1); + expect(summary.inProgress).toBe(1); + expect(component.hasInProgress).toBe(true); + }); + + it('taskSummary should fall back to the stored snapshot when no per-partition states exist', () => { + component.job = { + context: { + simpleFields: { + TASK_STATUS_SUMMARY: + '{"total":3,"completed":3,"failed":0,"timedOut":0,"inProgress":0,"pending":0,"other":0,"byState":{"COMPLETED":3},"failedTasks":[],"timedOutTasks":[],"inProgressTasks":[],"pendingTasks":[]}', + }, + mapFields: {}, + }, + } as any; + const summary = component.taskSummary; + expect(summary).not.toBeNull(); + expect(summary.total).toBe(3); + expect(summary.completed).toBe(3); + expect(component.hasPending).toBe(false); + }); }); diff --git a/helix-front/src/app/workflow/job-detail/job-detail.component.ts b/helix-front/src/app/workflow/job-detail/job-detail.component.ts index fd73ecc441..3a41f486ba 100644 --- a/helix-front/src/app/workflow/job-detail/job-detail.component.ts +++ b/helix-front/src/app/workflow/job-detail/job-detail.component.ts @@ -24,4 +24,119 @@ export class JobDetailComponent implements OnInit { () => (this.isLoading = false) ); } + + // Aggregated per-task status summary. It is computed on demand from the per-partition task + // states carried in the JobContext, so it is refreshed every time this page is opened (ngOnInit + // re-fetches the JobContext) rather than in the background. This surfaces partial failures even + // when the job's own state flag is COMPLETED. If the per-partition states are not available (for + // example the context was trimmed), it falls back to the TASK_STATUS_SUMMARY snapshot that the + // controller materializes into the JobContext when the job reaches a terminal state. + get taskSummary(): any { + const live = this.computeSummaryFromStates(); + if (live) { + return live; + } + const raw = + this.job && + this.job.context && + this.job.context.simpleFields && + this.job.context.simpleFields.TASK_STATUS_SUMMARY; + if (!raw) { + return null; + } + try { + return typeof raw === 'string' ? JSON.parse(raw) : raw; + } catch (e) { + return null; + } + } + + // Task partition states, mirrored from helix-core TaskPartitionState, that count as a terminal + // failure (given up / errored / timed out). + private static readonly FAILED_STATES = ['TASK_ERROR', 'TASK_ABORTED', 'TIMED_OUT', 'ERROR']; + + // Aggregates the raw per-partition task states (JobContext mapFields) into the same shape the + // controller writes, so the running-job view stays accurate on each page open without any + // background refresh. Returns null when no per-partition states are present. + private computeSummaryFromStates(): any { + const mapFields = + this.job && this.job.context && this.job.context.mapFields; + if (!mapFields || typeof mapFields !== 'object') { + return null; + } + const partitions = Object.keys(mapFields).filter((k) => !isNaN(Number(k))); + if (partitions.length === 0) { + return null; + } + + const byState: { [state: string]: number } = {}; + const failedTasks: number[] = []; + const timedOutTasks: number[] = []; + const inProgressTasks: number[] = []; + const pendingTasks: number[] = []; + let completed = 0; + let failed = 0; + let timedOut = 0; + let inProgress = 0; + let pending = 0; + + for (const key of partitions) { + const p = Number(key); + const state = (mapFields[key] && mapFields[key].STATE) || 'UNSCHEDULED'; + byState[state] = (byState[state] || 0) + 1; + if (state === 'COMPLETED') { + completed++; + } else if (JobDetailComponent.FAILED_STATES.indexOf(state) !== -1) { + failed++; + failedTasks.push(p); + if (state === 'TIMED_OUT') { + timedOut++; + timedOutTasks.push(p); + } + } else if (state === 'RUNNING') { + inProgress++; + inProgressTasks.push(p); + } else if (state === 'INIT') { + pending++; + pendingTasks.push(p); + } + } + + const total = partitions.length; + const numericAsc = (a: number, b: number) => a - b; + return { + total, + completed, + failed, + timedOut, + inProgress, + pending, + other: total - completed - failed - inProgress - pending, + byState, + failedTasks: failedTasks.sort(numericAsc), + timedOutTasks: timedOutTasks.sort(numericAsc), + inProgressTasks: inProgressTasks.sort(numericAsc), + pendingTasks: pendingTasks.sort(numericAsc), + }; + } + + get hasTaskFailures(): boolean { + const summary = this.taskSummary; + return !!summary && summary.failed > 0; + } + + get hasTimedOut(): boolean { + const summary = this.taskSummary; + return !!summary && summary.timedOut > 0; + } + + get hasInProgress(): boolean { + const summary = this.taskSummary; + return !!summary && summary.inProgress > 0; + } + + get hasPending(): boolean { + const summary = this.taskSummary; + return !!summary && summary.pending > 0; + } }