Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore(controller): ignore k8s job fail reason and refine task finish time #2912

Merged
merged 1 commit into from
Oct 27, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -139,23 +139,16 @@ private void reportRunStatus(V1Job job) {
log.warn("no pod stop time found for job {}, use now", jobName);
}

// prefer using the pod failed reason, it has more details
// use the pod failed reason, it has more details
// we have disabled the job retry, so the job failed reason is not useful
var podFailedReason = getPodFailedReason(jobName);

var failedReason = "";
if (StringUtils.hasText(jobFailedReason)) {
failedReason = "job failed: " + jobFailedReason;
}
if (StringUtils.hasText(podFailedReason)) {
failedReason = failedReason + "\npod failed: " + podFailedReason;
}

var report = ReportedRun.builder()
.id(runId)
.status(runStatus)
.startTimeMillis(startTime)
.stopTimeMillis(stopTime)
.failedReason(StringUtils.hasText(failedReason) ? failedReason : null)
.failedReason(StringUtils.hasText(podFailedReason) ? podFailedReason : null)
.build();
runReportReceiver.receive(report);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,8 @@ public void onRunUpdate(Run run) {
retryNum = null == retryNum ? 0 : retryNum;
Integer userRetryLimit = task.getStep().getSpec().getBackoffLimit();
Integer backOffLimit = userRetryLimit == null ? this.backOffLimit : userRetryLimit;

var updateFinishedTime = true;
if (
run.getStatus() == RunStatus.FAILED
&& retryNum < backOffLimit
Expand All @@ -108,6 +110,8 @@ public void onRunUpdate(Run run) {
Task ot = ((WatchableTask) task).unwrap();
ot.updateStatus(TaskStatus.READY);
}
// do not update finished time for retrying task
updateFinishedTime = false;
} else {
taskNewStatus = taskStatusMachine.transfer(task.getStatus(), run.getStatus());
}
Expand All @@ -118,7 +122,7 @@ public void onRunUpdate(Run run) {
if (run.getStartTime() != null) {
taskMapper.updateTaskStartedTimeIfNotSet(run.getTaskId(), new Date(run.getStartTime()));
}
if (run.getFinishTime() != null) {
if (run.getFinishTime() != null && updateFinishedTime) {
taskMapper.updateTaskFinishedTimeIfNotSet(run.getTaskId(), new Date(run.getFinishTime()));
}
if (run.getIp() != null) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@ public void testOnAddFail() {
.stopTimeMillis(endTime.toInstant().toEpochMilli())
.build();
verify(runReportReceiver).receive(expected);
reset(runReportReceiver);

// test with reason and message
var con = new V1JobCondition()
Expand All @@ -112,7 +113,7 @@ public void testOnAddFail() {
.status(RunStatus.FAILED)
.startTimeMillis(startTime.toInstant().toEpochMilli())
.stopTimeMillis(endTime.toInstant().toEpochMilli())
.failedReason("job failed: reason, message")
.failedReason(null)
.build();
verify(runReportReceiver).receive(expected2);

Expand All @@ -128,7 +129,7 @@ public void testOnAddFail() {
.status(RunStatus.FAILED)
.startTimeMillis(startTime.toInstant().toEpochMilli())
.stopTimeMillis(endTime.toInstant().toEpochMilli())
.failedReason("job failed: reason, message\npod failed: foo, bar\nfoo, bar")
.failedReason("foo, bar\nfoo, bar")
.build();
verify(runReportReceiver).receive(expected3);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
Expand All @@ -32,6 +33,7 @@
import ai.starwhale.mlops.domain.task.status.TaskStatus;
import ai.starwhale.mlops.domain.task.status.TaskStatusMachine;
import ai.starwhale.mlops.schedule.SwTaskScheduler;
import java.util.Date;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
Expand Down Expand Up @@ -103,4 +105,32 @@ public void testOnUpdate() {

}

@Test
public void testFinishTime() {
var task = Task.builder()
.retryNum(0)
.step(Step.builder().spec(StepSpec.builder().backoffLimit(2).build()).build())
.currentRun(Run.builder().id(1L).status(RunStatus.RUNNING).build())
.build();
when(hotJobHolder.taskWithId(1L)).thenReturn(task);
var reportdRun = Run.builder()
.id(1L)
.taskId(1L)
.status(RunStatus.FAILED)
.finishTime(42L)
.ip("1.1.1.1")
.build();
runReportReceiver.onRunUpdate(reportdRun);
verify(this.taskMapper, never()).updateTaskFinishedTimeIfNotSet(any(), any());

// fail again
task.getCurrentRun().setStatus(RunStatus.RUNNING);
runReportReceiver.onRunUpdate(reportdRun);
verify(this.taskMapper, never()).updateTaskFinishedTimeIfNotSet(any(), any());

// the last attempt
task.getCurrentRun().setStatus(RunStatus.RUNNING);
runReportReceiver.onRunUpdate(reportdRun);
verify(this.taskMapper, times(1)).updateTaskFinishedTimeIfNotSet(1L, new Date(42L));
}
}
Loading