github-advanced-security[bot] commented on code in PR #18358:
URL:
https://github.com/apache/dolphinscheduler/pull/18358#discussion_r3464440238
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java:
##########
@@ -199,62 +201,161 @@
}
public void registerBatchJobWatcher(Job job, String taskInstanceId,
TaskResponse taskResponse) {
- CountDownLatch countDownLatch = new CountDownLatch(1);
- Watcher<Job> watcher = new Watcher<Job>() {
-
- @Override
- public void eventReceived(Action action, Job job) {
- try {
-
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
- taskRequest.getTaskInstanceId());
-
LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
- log.info("event received : job:{} action:{}",
job.getMetadata().getName(), action);
- if (action == Action.DELETED) {
- log.error("[K8sJobExecutor-{}] fail in k8s",
job.getMetadata().getName());
+ final String jobName = job.getMetadata().getName();
+ final String namespace = job.getMetadata().getNamespace();
+ final CountDownLatch countDownLatch = new CountDownLatch(1);
+ SharedIndexInformer<Job> informer = null;
+ ScheduledExecutorService jobStatusPoller = null;
+ try {
+ informer = k8sUtils.createBatchJobInformer(jobName, namespace,
+ createBatchJobEventHandler(taskInstanceId, taskResponse,
countDownLatch));
+ informer.start().whenComplete((v, ex) -> {
+ if (ex != null && countDownLatch.getCount() > 0) {
+ withTaskLogContext(() -> {
+ log.error("[K8sJobExecutor-{}] fail in k8s: {}",
jobName, ex.getMessage(), ex);
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
countDownLatch.countDown();
- } else if (action != Action.ADDED) {
- int jobStatus = getK8sJobStatus(job);
- log.info("job {} status {}",
job.getMetadata().getName(), jobStatus);
- if (jobStatus == TaskConstants.RUNNING_CODE) {
- return;
- }
- setTaskStatus(jobStatus, taskInstanceId, taskResponse);
- countDownLatch.countDown();
- }
- } finally {
- LogUtils.removeTaskInstanceLogFullPathMDC();
- LogUtils.removeWorkflowAndTaskInstanceIdMDC();
+ });
}
- }
-
- @Override
- public void onClose(WatcherException e) {
-
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
- taskRequest.getTaskInstanceId());
- log.error("[K8sJobExecutor-{}] fail in k8s: {}",
job.getMetadata().getName(), e.getMessage());
- taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
- countDownLatch.countDown();
- LogUtils.removeWorkflowAndTaskInstanceIdMDC();
- }
- };
- try (Watch watch =
k8sUtils.createBatchJobWatcher(job.getMetadata().getName(), watcher)) {
- boolean timeoutFlag = taskRequest.getTaskTimeoutStrategy() ==
TaskTimeoutStrategy.FAILED
- || taskRequest.getTaskTimeoutStrategy() ==
TaskTimeoutStrategy.WARNFAILED;
- if (timeoutFlag) {
- Boolean timeout =
!(countDownLatch.await(taskRequest.getTaskTimeout(), TimeUnit.SECONDS));
- waitTimeout(timeout);
- } else {
- countDownLatch.await();
- }
+ });
+ jobStatusPoller = startJobStatusPoller(jobName, namespace,
taskInstanceId, taskResponse, countDownLatch);
+ awaitJobCompletion(countDownLatch);
} catch (InterruptedException e) {
log.error("job failed in k8s: {}", e.getMessage(), e);
Thread.currentThread().interrupt();
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
} catch (Exception e) {
log.error("job failed in k8s: {}", e.getMessage(), e);
taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
+ } finally {
+ if (jobStatusPoller != null) {
+ jobStatusPoller.shutdownNow();
+ }
+ if (informer != null) {
+ informer.stop();
+ }
+ }
+ }
+
+ /**
+ * Returns the interval for polling Job status. Protected so unit tests
can override with a shorter value.
+ */
+ protected long getJobStatusPollIntervalSeconds() {
+ return JOB_STATUS_POLL_INTERVAL_SECONDS;
+ }
+
+ private ScheduledExecutorService startJobStatusPoller(String jobName,
+ String namespace,
+ String
taskInstanceId,
+ TaskResponse
taskResponse,
+ CountDownLatch
countDownLatch) {
+ ScheduledExecutorService jobStatusPoller =
ThreadUtils.newSingleDaemonScheduledExecutorService(
+ "K8sJobStatusPoller-" + jobName);
+ long pollIntervalSeconds = getJobStatusPollIntervalSeconds();
+ jobStatusPoller.scheduleAtFixedRate(
+ () -> pollBatchJobStatus(jobName, namespace, taskInstanceId,
taskResponse, countDownLatch),
+ pollIntervalSeconds,
+ pollIntervalSeconds,
+ TimeUnit.SECONDS);
+ return jobStatusPoller;
+ }
+
+ private void pollBatchJobStatus(String jobName,
+ String namespace,
+ String taskInstanceId,
+ TaskResponse taskResponse,
+ CountDownLatch countDownLatch) {
+ if (countDownLatch.getCount() == 0) {
+ return;
+ }
+ withTaskLogContext(() -> {
+ try {
+ log.info("[K8sJobExecutor-{}] polling job status via GET",
jobName);
+ Job polledJob = k8sUtils.getJob(jobName, namespace);
+ if (polledJob == null) {
+ log.error("[K8sJobExecutor-{}] job not found during status
polling", jobName);
+ taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
+ countDownLatch.countDown();
+ return;
+ }
+ handleBatchJobTerminalStatus(polledJob, taskInstanceId,
taskResponse, countDownLatch);
+ } catch (Exception e) {
+ log.warn("[K8sJobExecutor-{}] failed to poll job status: {}",
jobName, e.getMessage());
+ }
+ });
+ }
+
+ private ResourceEventHandler<Job> createBatchJobEventHandler(String
taskInstanceId, TaskResponse taskResponse,
+
CountDownLatch countDownLatch) {
+ return new ResourceEventHandler<Job>() {
+
+ @Override
+ public void onAdd(Job watchedJob) {
+ withTaskLogContext(() -> {
+ log.info("event received, job: {}, action: ADD",
watchedJob.getMetadata().getName());
+ handleBatchJobTerminalStatus(watchedJob, taskInstanceId,
taskResponse, countDownLatch);
+ });
+ }
+
+ @Override
+ public void onUpdate(Job oldJob, Job watchedJob) {
+ withTaskLogContext(() -> {
+ log.info("event received, job: {}, action: UPDATE",
watchedJob.getMetadata().getName());
+ handleBatchJobTerminalStatus(watchedJob, taskInstanceId,
taskResponse, countDownLatch);
+ });
+ }
+
+ @Override
+ public void onDelete(Job watchedJob, boolean
deletedFinalStateUnknown) {
+ withTaskLogContext(() -> {
+ if (countDownLatch.getCount() == 0) {
+ return;
+ }
+ log.info("event received, job: {}, action: DELETE",
watchedJob.getMetadata().getName());
+ log.error("[K8sJobExecutor-{}] fail in k8s",
watchedJob.getMetadata().getName());
+ taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
+ countDownLatch.countDown();
+ });
+ }
+ };
+ }
+
+ private void awaitJobCompletion(CountDownLatch countDownLatch) throws
InterruptedException {
+ boolean timeoutFlag = taskRequest.getTaskTimeoutStrategy() ==
TaskTimeoutStrategy.FAILED
+ || taskRequest.getTaskTimeoutStrategy() ==
TaskTimeoutStrategy.WARNFAILED;
+ if (timeoutFlag) {
+ if (!countDownLatch.await(taskRequest.getTaskTimeout(),
TimeUnit.SECONDS)) {
+ waitTimeout(true);
+ }
+ } else {
+ countDownLatch.await();
+ }
+ }
+
+ private void withTaskLogContext(Runnable action) {
+ try {
+
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
+ taskRequest.getTaskInstanceId());
+ LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
+ action.run();
+ } finally {
+ LogUtils.removeTaskInstanceLogFullPathMDC();
+ LogUtils.removeWorkflowAndTaskInstanceIdMDC();
+ }
+ }
+
+ private void handleBatchJobTerminalStatus(Job watchedJob, String
taskInstanceId, TaskResponse taskResponse,
Review Comment:
## CodeQL / Useless parameter
The parameter 'taskInstanceId' is never used.
[Show more
details](https://github.com/apache/dolphinscheduler/security/code-scanning/5806)
--
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]
