Skip to content

Commit fc72e84

Browse files
luxiaolong-ctdet101
luxiaolong-ct
authored and
det101
committed
[Improvement-17330][K8s] Harden batch job informer per review feedback
Handle terminal status on onAdd, fail fast when informer stops with error via stopped().whenComplete, align event log format, and bump Fabric8 to 6.4.0 with kubernetes-client-bom. Expand K8sTaskExecutorTest coverage.
1 parent f3fc50b commit fc72e84

3 files changed

Lines changed: 176 additions & 68 deletions

File tree

‎dolphinscheduler-bom/pom.xml‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,7 @@
110110
<snowflake-jdbc.version>3.13.29</snowflake-jdbc.version>
111111
<google-cloud-storage.version>2.18.0</google-cloud-storage.version>
112112
<sshd.version>2.8.0</sshd.version>
113-
<fabric8.client.version>6.0.0</fabric8.client.version>
113+
<fabric8.client.version>6.4.0</fabric8.client.version>
114114
<casdoor.version>1.6.0</casdoor.version>
115115
<azure-sdk-bom.version>1.2.10</azure-sdk-bom.version>
116116
<protobuf.version>3.17.2</protobuf.version>
@@ -801,6 +801,14 @@
801801
<version>${joda-time.version}</version>
802802
</dependency>
803803

804+
<dependency>
805+
<groupId>io.fabric8</groupId>
806+
<artifactId>kubernetes-client-bom</artifactId>
807+
<version>${fabric8.client.version}</version>
808+
<type>pom</type>
809+
<scope>import</scope>
810+
</dependency>
811+
804812
<dependency>
805813
<groupId>org.springframework.cloud</groupId>
806814
<artifactId>spring-cloud-dependencies</artifactId>
@@ -904,6 +912,11 @@
904912
<artifactId>kubernetes-client</artifactId>
905913
<version>${fabric8.client.version}</version>
906914
</dependency>
915+
<dependency>
916+
<groupId>io.fabric8</groupId>
917+
<artifactId>kubernetes-client-api</artifactId>
918+
<version>${fabric8.client.version}</version>
919+
</dependency>
907920
<dependency>
908921
<groupId>org.casbin</groupId>
909922
<artifactId>casdoor-spring-boot-starter</artifactId>

‎dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java‎

Lines changed: 103 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -201,78 +201,117 @@ public void registerBatchJobWatcher(Job job, String taskInstanceId, TaskResponse
201201
final String jobName = job.getMetadata().getName();
202202
final String namespace = job.getMetadata().getNamespace();
203203
final CountDownLatch countDownLatch = new CountDownLatch(1);
204-
SharedIndexInformer<Job> informer = k8sUtils.createBatchJobInformer(jobName, namespace,
205-
new ResourceEventHandler<Job>() {
206-
207-
@Override
208-
public void onAdd(Job watchedJob) {
209-
try {
210-
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
211-
taskRequest.getTaskInstanceId());
212-
LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
213-
log.info("event received : job:{} action:ADD", watchedJob.getMetadata().getName());
214-
} finally {
215-
LogUtils.removeTaskInstanceLogFullPathMDC();
216-
LogUtils.removeWorkflowAndTaskInstanceIdMDC();
217-
}
218-
}
219-
220-
@Override
221-
public void onUpdate(Job oldJob, Job watchedJob) {
222-
try {
223-
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
224-
taskRequest.getTaskInstanceId());
225-
LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
226-
log.info("event received : job:{} action:UPDATE", watchedJob.getMetadata().getName());
227-
int jobStatus = getK8sJobStatus(watchedJob);
228-
log.info("job {} status {}", watchedJob.getMetadata().getName(), jobStatus);
229-
if (jobStatus == TaskConstants.RUNNING_CODE) {
230-
return;
231-
}
232-
setTaskStatus(jobStatus, taskInstanceId, taskResponse);
233-
countDownLatch.countDown();
234-
} finally {
235-
LogUtils.removeTaskInstanceLogFullPathMDC();
236-
LogUtils.removeWorkflowAndTaskInstanceIdMDC();
237-
}
238-
}
239-
240-
@Override
241-
public void onDelete(Job watchedJob, boolean deletedFinalStateUnknown) {
242-
try {
243-
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
244-
taskRequest.getTaskInstanceId());
245-
LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
246-
log.error("[K8sJobExecutor-{}] fail in k8s", watchedJob.getMetadata().getName());
247-
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
248-
countDownLatch.countDown();
249-
} finally {
250-
LogUtils.removeTaskInstanceLogFullPathMDC();
251-
LogUtils.removeWorkflowAndTaskInstanceIdMDC();
252-
}
253-
}
254-
});
204+
SharedIndexInformer<Job> informer = null;
255205
try {
256-
boolean timeoutFlag = taskRequest.getTaskTimeoutStrategy() == TaskTimeoutStrategy.FAILED
257-
|| taskRequest.getTaskTimeoutStrategy() == TaskTimeoutStrategy.WARNFAILED;
258-
if (timeoutFlag) {
259-
Boolean timeout = !(countDownLatch.await(taskRequest.getTaskTimeout(), TimeUnit.SECONDS));
260-
waitTimeout(timeout);
261-
} else {
262-
countDownLatch.await();
263-
}
206+
informer = k8sUtils.createBatchJobInformer(jobName, namespace,
207+
createBatchJobEventHandler(taskInstanceId, taskResponse, countDownLatch));
208+
informer.stopped().whenComplete((v, ex) -> {
209+
if (ex != null && countDownLatch.getCount() > 0) {
210+
failInformerAndCountDown(jobName, ex.getMessage(), ex, taskResponse, countDownLatch);
211+
}
212+
});
213+
awaitJobCompletion(countDownLatch);
264214
} catch (InterruptedException e) {
265-
log.error("job failed in k8s: {}", e.getMessage(), e);
266215
Thread.currentThread().interrupt();
267-
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
216+
withTaskLogContext(() -> {
217+
log.error("job failed in k8s: {}", e.getMessage(), e);
218+
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
219+
countDownLatch.countDown();
220+
});
268221
} catch (Exception e) {
269-
log.error("job failed in k8s: {}", e.getMessage(), e);
270-
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
222+
if (countDownLatch.getCount() > 0) {
223+
withTaskLogContext(() -> {
224+
log.error("job failed in k8s: {}", e.getMessage(), e);
225+
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
226+
countDownLatch.countDown();
227+
});
228+
}
271229
} finally {
272-
informer.stop();
230+
if (informer != null) {
231+
informer.stop();
232+
}
273233
}
274234
}
275235

236+
private ResourceEventHandler<Job> createBatchJobEventHandler(String taskInstanceId, TaskResponse taskResponse,
237+
CountDownLatch countDownLatch) {
238+
return new ResourceEventHandler<Job>() {
239+
240+
@Override
241+
public void onAdd(Job watchedJob) {
242+
withTaskLogContext(() -> {
243+
log.info("event received, job: {}, action: ADD", watchedJob.getMetadata().getName());
244+
handleBatchJobTerminalStatus(watchedJob, taskInstanceId, taskResponse, countDownLatch);
245+
});
246+
}
247+
248+
@Override
249+
public void onUpdate(Job oldJob, Job watchedJob) {
250+
withTaskLogContext(() -> {
251+
log.info("event received, job: {}, action: UPDATE", watchedJob.getMetadata().getName());
252+
handleBatchJobTerminalStatus(watchedJob, taskInstanceId, taskResponse, countDownLatch);
253+
});
254+
}
255+
256+
@Override
257+
public void onDelete(Job watchedJob, boolean deletedFinalStateUnknown) {
258+
withTaskLogContext(() -> {
259+
log.info("event received, job: {}, action: DELETE", watchedJob.getMetadata().getName());
260+
log.error("[K8sJobExecutor-{}] fail in k8s", watchedJob.getMetadata().getName());
261+
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
262+
countDownLatch.countDown();
263+
});
264+
}
265+
};
266+
}
267+
268+
private void awaitJobCompletion(CountDownLatch countDownLatch) throws InterruptedException {
269+
boolean timeoutFlag = taskRequest.getTaskTimeoutStrategy() == TaskTimeoutStrategy.FAILED
270+
|| taskRequest.getTaskTimeoutStrategy() == TaskTimeoutStrategy.WARNFAILED;
271+
if (timeoutFlag) {
272+
if (!countDownLatch.await(taskRequest.getTaskTimeout(), TimeUnit.SECONDS)) {
273+
waitTimeout(true);
274+
}
275+
} else {
276+
countDownLatch.await();
277+
}
278+
}
279+
280+
private void withTaskLogContext(Runnable action) {
281+
try {
282+
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
283+
taskRequest.getTaskInstanceId());
284+
LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
285+
action.run();
286+
} finally {
287+
LogUtils.removeTaskInstanceLogFullPathMDC();
288+
LogUtils.removeWorkflowAndTaskInstanceIdMDC();
289+
}
290+
}
291+
292+
private void handleBatchJobTerminalStatus(Job watchedJob, String taskInstanceId, TaskResponse taskResponse,
293+
CountDownLatch countDownLatch) {
294+
int jobStatus = getK8sJobStatus(watchedJob);
295+
log.info("job {} status {}", watchedJob.getMetadata().getName(), jobStatus);
296+
if (jobStatus == TaskConstants.RUNNING_CODE) {
297+
return;
298+
}
299+
setTaskStatus(jobStatus, taskInstanceId, taskResponse);
300+
countDownLatch.countDown();
301+
}
302+
303+
private void failInformerAndCountDown(String jobName, String message, Throwable ex, TaskResponse taskResponse,
304+
CountDownLatch countDownLatch) {
305+
if (countDownLatch.getCount() == 0) {
306+
return;
307+
}
308+
withTaskLogContext(() -> {
309+
log.error("[K8sJobExecutor-{}] fail in k8s: {}", jobName, message, ex);
310+
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
311+
countDownLatch.countDown();
312+
});
313+
}
314+
276315
private void parsePodLogOutput() {
277316
ExecutorService collectPodLogExecutorService = ThreadUtils
278317
.newSingleDaemonScheduledExecutorService("CollectPodLogOutput-thread-" + taskRequest.getTaskName());

‎dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/k8s/K8sTaskExecutorTest.java‎

Lines changed: 59 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import static org.junit.jupiter.api.Assertions.assertEquals;
2323
import static org.mockito.ArgumentMatchers.any;
2424
import static org.mockito.ArgumentMatchers.eq;
25+
import static org.mockito.Mockito.doThrow;
2526
import static org.mockito.Mockito.mock;
2627
import static org.mockito.Mockito.verify;
2728
import static org.mockito.Mockito.when;
@@ -38,6 +39,7 @@
3839
import java.util.HashMap;
3940
import java.util.List;
4041
import java.util.Map;
42+
import java.util.concurrent.CompletableFuture;
4143
import java.util.concurrent.CountDownLatch;
4244
import java.util.concurrent.TimeUnit;
4345
import java.util.concurrent.atomic.AtomicReference;
@@ -106,9 +108,19 @@ private TaskExecutionContext getTaskRequest() throws Exception {
106108
return (TaskExecutionContext) field.get(k8sTaskExecutor);
107109
}
108110

111+
private void mockInformer(WatcherHarness harness, CompletableFuture<Void> stoppedFuture) {
112+
harness.informer = mock(SharedIndexInformer.class);
113+
when(harness.informer.stopped()).thenReturn(stoppedFuture);
114+
}
115+
109116
private WatcherHarness startBatchJobWatcher(TaskResponse taskResponse) throws InterruptedException {
110117
WatcherHarness harness = new WatcherHarness();
111-
harness.informer = mock(SharedIndexInformer.class);
118+
mockInformer(harness, new CompletableFuture<>());
119+
return startWatcherThread(harness, taskResponse);
120+
}
121+
122+
private WatcherHarness startWatcherThread(WatcherHarness harness,
123+
TaskResponse taskResponse) throws InterruptedException {
112124
CountDownLatch handlerReady = new CountDownLatch(1);
113125
AtomicReference<ResourceEventHandler<Job>> handlerRef = new AtomicReference<>();
114126
when(k8sUtils.createBatchJobInformer(eq(job.getMetadata().getName()), eq(namespace), any()))
@@ -127,7 +139,9 @@ private WatcherHarness startBatchJobWatcher(TaskResponse taskResponse) throws In
127139

128140
private void finishWatcher(WatcherHarness harness) throws InterruptedException {
129141
harness.thread.join(5000);
130-
verify(harness.informer).stop();
142+
if (harness.informer != null) {
143+
verify(harness.informer).stop();
144+
}
131145
}
132146

133147
private Job jobWithStatus(Integer succeeded, Integer failed) {
@@ -168,10 +182,28 @@ public void testRegisterBatchJobInformerOnDelete() throws Exception {
168182
}
169183

170184
@Test
171-
public void testRegisterBatchJobInformerIgnoreOnAdd() throws Exception {
185+
public void testRegisterBatchJobInformerOnAddSuccess() throws Exception {
172186
TaskResponse taskResponse = new TaskResponse();
173187
WatcherHarness harness = startBatchJobWatcher(taskResponse);
174188
harness.handler.onAdd(jobWithStatus(1, null));
189+
finishWatcher(harness);
190+
assertEquals(EXIT_CODE_SUCCESS, taskResponse.getExitStatusCode());
191+
}
192+
193+
@Test
194+
public void testRegisterBatchJobInformerOnAddFailed() throws Exception {
195+
TaskResponse taskResponse = new TaskResponse();
196+
WatcherHarness harness = startBatchJobWatcher(taskResponse);
197+
harness.handler.onAdd(jobWithStatus(null, 1));
198+
finishWatcher(harness);
199+
assertEquals(EXIT_CODE_FAILURE, taskResponse.getExitStatusCode());
200+
}
201+
202+
@Test
203+
public void testRegisterBatchJobInformerOnAddRunning() throws Exception {
204+
TaskResponse taskResponse = new TaskResponse();
205+
WatcherHarness harness = startBatchJobWatcher(taskResponse);
206+
harness.handler.onAdd(jobWithStatus(null, null));
175207
Assertions.assertTrue(harness.thread.isAlive());
176208
harness.handler.onUpdate(job, jobWithStatus(1, null));
177209
finishWatcher(harness);
@@ -189,6 +221,30 @@ public void testRegisterBatchJobInformerIgnoreRunningUpdate() throws Exception {
189221
assertEquals(EXIT_CODE_SUCCESS, taskResponse.getExitStatusCode());
190222
}
191223

224+
@Test
225+
public void testRegisterBatchJobInformerInformerStartFailed() throws Exception {
226+
TaskResponse taskResponse = new TaskResponse();
227+
doThrow(new RuntimeException("informer start failed")).when(k8sUtils)
228+
.createBatchJobInformer(eq(job.getMetadata().getName()), eq(namespace), any());
229+
Thread thread = new Thread(() -> k8sTaskExecutor.registerBatchJobWatcher(job,
230+
String.valueOf(taskInstanceId), taskResponse));
231+
thread.start();
232+
thread.join(5000);
233+
assertEquals(EXIT_CODE_FAILURE, taskResponse.getExitStatusCode());
234+
}
235+
236+
@Test
237+
public void testRegisterBatchJobInformerInformerStopped() throws Exception {
238+
TaskResponse taskResponse = new TaskResponse();
239+
WatcherHarness harness = new WatcherHarness();
240+
CompletableFuture<Void> stoppedFuture = new CompletableFuture<>();
241+
mockInformer(harness, stoppedFuture);
242+
startWatcherThread(harness, taskResponse);
243+
stoppedFuture.completeExceptionally(new RuntimeException("informer stopped"));
244+
finishWatcher(harness);
245+
assertEquals(EXIT_CODE_FAILURE, taskResponse.getExitStatusCode());
246+
}
247+
192248
@Test
193249
public void testRegisterBatchJobInformerTimeout() throws Exception {
194250
TaskExecutionContext taskRequest = getTaskRequest();

0 commit comments

Comments
 (0)