You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "georgew5656 (via GitHub)" <gi...@apache.org> on 2024/11/04 22:58:06 UTC
[PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
georgew5656 opened a new pull request, #17446:
URL: https://github.com/apache/druid/pull/17446
### Description
It's possible for the KubenretesTaskRunner.start method to return before the threads in the exec pool running (KubernetesPeonLifecycle.join) have finished gathering information about kubernetes jobs. This can be a problem because other services on the overlord (like supervisors) expect the KubenretesTaskRunner to have information about tasks once it has returned from start (for example each task's location).
This diff adds a wait in the start() method that attempts to wait for all the tasks that have been discovered to go into RUNNING state. This state indicates the KubernetesTaskRunner knows all the information that it needs about a task.
#### Release note
Bugfix that helps mitigate unexpected behavior when running k8s based ingestion during overlord restarts.
##### Key changed/added classes in this PR
* `KubernetesTaskRunner`
* `KubernetesWorkItem`
* `KubernetesPeonLifecycle`
This PR has:
- [X] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [X] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1833113432
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java:
##########
@@ -321,16 +333,40 @@ public List<Pair<Task, ListenableFuture<TaskStatus>>> restore()
public void start()
{
log.info("Starting K8sTaskRunner...");
- // Load tasks from previously running jobs and wait for their statuses to be updated asynchronously.
- for (Job job : client.getPeonJobs()) {
+ // Load tasks from previously running jobs and wait for their statuses to start running.
+ final List<ListenableFuture<Boolean>> taskStatusActiveList = new ArrayList<>();
+ final List<Job> peonJobs = client.getPeonJobs();
+
+ log.info("Locating [%,d] active tasks.", peonJobs.size());
+ for (Job job : peonJobs) {
try {
- joinAsync(adapter.toTask(job));
+ KubernetesWorkItem kubernetesWorkItem = joinAsync(adapter.toTask(job));
+ taskStatusActiveList.add(kubernetesWorkItem.getPeonLifeycle().getTaskStartedSuccessfullyFuture());
}
catch (IOException e) {
log.error(e, "Error deserializing task from job [%s]", job.getMetadata().getName());
}
}
- log.info("Loaded %,d tasks from previous run", tasks.size());
+
+ try {
+ final DateTime nowUtc = DateTimes.nowUtc();
+ final long timeoutMs = nowUtc.plus(config.getTaskJoinTimeout()).getMillis() - nowUtc.getMillis();
+ if (timeoutMs > 0) {
+ FutureUtils.coalesce(taskStatusActiveList).get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+ log.info("Located [%,d] active tasks.", taskStatusActiveList.size());
+ }
+ catch (Exception e) {
+ final long numInitialized =
+ tasks.values().stream().filter(item -> item.getPeonLifeycle().getTaskStartedSuccessfullyFuture().isDone()).count();
Review Comment:
yeah i think looking for ones that eval to true makes sense, updated this.
on the second comment, the main logic of the task lifecycle is performed by the exec thread.
` return tasks.computeIfAbsent(task.getId(), k -> new KubernetesWorkItem(
task,
exec.submit(() -> joinTask(task)),
peonLifecycleFactory.build(
task,
this::emitTaskStateMetrics
)
));`
so if the task fails to join the KubernetesWorkItem.getResult() future will complete as failed and the taskQueue will mark the task as failed and shut it down, i don't believe we need to do this in the start() method too.
for the new "did task start successfully future"
i updated the logic so the catch blocks in run/join catch any errors during run/join and mark the future as failed if it didn't start properly. also added a test for the scenario
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 merged PR #17446:
URL: https://github.com/apache/druid/pull/17446
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on PR #17446:
URL: https://github.com/apache/druid/pull/17446#issuecomment-2465149616
i don't think the failing unit test is related so i'm going to go ahead and merge this
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1833177081
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -136,14 +141,16 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout,
launchTimeout,
TimeUnit.MILLISECONDS
);
-
return join(timeout);
}
catch (Exception e) {
log.info("Failed to run task: %s", taskId.getOriginalTaskId());
throw e;
}
finally {
+ if (!taskStartedSuccessfullyFuture.isDone()) {
Review Comment:
ah, I see, OK.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1831883883
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -174,11 +176,13 @@ private void writeTaskPayload(Task task) throws IOException
* @return
* @throws IllegalStateException
*/
- protected synchronized TaskStatus join(long timeout) throws IllegalStateException
+ protected synchronized TaskStatus join(long timeout, SettableFuture<Boolean> taskActiveStatusFuture) throws IllegalStateException
Review Comment:
i added a function that just returns a future containing a boolean with info on whether the task started running successfully. in order to get this to work i made KubernetesPeonLifecycle a regular argument for KubernetesWorkItem (instead of setting it with setKubernetesPeonLifecycle afterwards). this seems fine to me becuase KubernetesPeonLifecycle is just a simple POJO until you call run or join on it, i don't really see why it needs to be instantiated in doTask instead of run/joinAsync.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1833076128
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -136,14 +141,16 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout,
launchTimeout,
TimeUnit.MILLISECONDS
);
-
return join(timeout);
}
catch (Exception e) {
log.info("Failed to run task: %s", taskId.getOriginalTaskId());
throw e;
}
finally {
+ if (!taskStartedSuccessfullyFuture.isDone()) {
Review Comment:
run always calls into join (line 140) so it'll get set there
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1831670326
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -174,11 +176,13 @@ private void writeTaskPayload(Task task) throws IOException
* @return
* @throws IllegalStateException
*/
- protected synchronized TaskStatus join(long timeout) throws IllegalStateException
+ protected synchronized TaskStatus join(long timeout, SettableFuture<Boolean> taskActiveStatusFuture) throws IllegalStateException
Review Comment:
hmm doing it this way kinda requires the entire lifecycle logic to be rewritten, i think i have a simpler semi-hacky way of accomplishing the same thing.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1831620805
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -174,11 +176,13 @@ private void writeTaskPayload(Task task) throws IOException
* @return
* @throws IllegalStateException
*/
- protected synchronized TaskStatus join(long timeout) throws IllegalStateException
+ protected synchronized TaskStatus join(long timeout, SettableFuture<Boolean> taskActiveStatusFuture) throws IllegalStateException
Review Comment:
i guess it would make more sense to maybe split this up into a joinUntilStart method that calls back into the wait method, i can rewrite it a bit
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1833035823
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java:
##########
@@ -321,16 +333,40 @@ public List<Pair<Task, ListenableFuture<TaskStatus>>> restore()
public void start()
{
log.info("Starting K8sTaskRunner...");
- // Load tasks from previously running jobs and wait for their statuses to be updated asynchronously.
- for (Job job : client.getPeonJobs()) {
+ // Load tasks from previously running jobs and wait for their statuses to start running.
+ final List<ListenableFuture<Boolean>> taskStatusActiveList = new ArrayList<>();
+ final List<Job> peonJobs = client.getPeonJobs();
+
+ log.info("Locating [%,d] active tasks.", peonJobs.size());
+ for (Job job : peonJobs) {
try {
- joinAsync(adapter.toTask(job));
+ KubernetesWorkItem kubernetesWorkItem = joinAsync(adapter.toTask(job));
+ taskStatusActiveList.add(kubernetesWorkItem.getPeonLifeycle().getTaskStartedSuccessfullyFuture());
}
catch (IOException e) {
log.error(e, "Error deserializing task from job [%s]", job.getMetadata().getName());
}
}
- log.info("Loaded %,d tasks from previous run", tasks.size());
+
+ try {
+ final DateTime nowUtc = DateTimes.nowUtc();
+ final long timeoutMs = nowUtc.plus(config.getTaskJoinTimeout()).getMillis() - nowUtc.getMillis();
+ if (timeoutMs > 0) {
+ FutureUtils.coalesce(taskStatusActiveList).get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+ log.info("Located [%,d] active tasks.", taskStatusActiveList.size());
+ }
+ catch (Exception e) {
+ final long numInitialized =
+ tasks.values().stream().filter(item -> item.getPeonLifeycle().getTaskStartedSuccessfullyFuture().isDone()).count();
Review Comment:
I don't see anything that recovers from failed `join`, so I suppose there's three cases?
- future is done and `true` — the task was joined
- future is done and `false` — error joining the task, i suppose it's never going to be joined? at this point should we mark it failed?
- future is not yet done — the task is still being joined async
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1833113432
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java:
##########
@@ -321,16 +333,40 @@ public List<Pair<Task, ListenableFuture<TaskStatus>>> restore()
public void start()
{
log.info("Starting K8sTaskRunner...");
- // Load tasks from previously running jobs and wait for their statuses to be updated asynchronously.
- for (Job job : client.getPeonJobs()) {
+ // Load tasks from previously running jobs and wait for their statuses to start running.
+ final List<ListenableFuture<Boolean>> taskStatusActiveList = new ArrayList<>();
+ final List<Job> peonJobs = client.getPeonJobs();
+
+ log.info("Locating [%,d] active tasks.", peonJobs.size());
+ for (Job job : peonJobs) {
try {
- joinAsync(adapter.toTask(job));
+ KubernetesWorkItem kubernetesWorkItem = joinAsync(adapter.toTask(job));
+ taskStatusActiveList.add(kubernetesWorkItem.getPeonLifeycle().getTaskStartedSuccessfullyFuture());
}
catch (IOException e) {
log.error(e, "Error deserializing task from job [%s]", job.getMetadata().getName());
}
}
- log.info("Loaded %,d tasks from previous run", tasks.size());
+
+ try {
+ final DateTime nowUtc = DateTimes.nowUtc();
+ final long timeoutMs = nowUtc.plus(config.getTaskJoinTimeout()).getMillis() - nowUtc.getMillis();
+ if (timeoutMs > 0) {
+ FutureUtils.coalesce(taskStatusActiveList).get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+ log.info("Located [%,d] active tasks.", taskStatusActiveList.size());
+ }
+ catch (Exception e) {
+ final long numInitialized =
+ tasks.values().stream().filter(item -> item.getPeonLifeycle().getTaskStartedSuccessfullyFuture().isDone()).count();
Review Comment:
yeah i think looking for ones that eval to true makes sense, updated this.
on the second comment, the main logic of the task lifecycle is performed by the exec thread.
` return tasks.computeIfAbsent(task.getId(), k -> new KubernetesWorkItem(
task,
exec.submit(() -> joinTask(task)),
peonLifecycleFactory.build(
task,
this::emitTaskStateMetrics
)
));`
so if the task fails to join the KubernetesWorkItem.getResult() future will complete as failed and the taskQueue will mark the task as failed and shut it down, we don't need to do this in the start() method
for the new "did task start successfully future"
i updated the logic so the catch blocks in run/join catch any errors during run/join and mark the future as failed if it didn't start properly. also added a test for the scenario
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1833021788
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java:
##########
@@ -321,16 +333,40 @@ public List<Pair<Task, ListenableFuture<TaskStatus>>> restore()
public void start()
{
log.info("Starting K8sTaskRunner...");
- // Load tasks from previously running jobs and wait for their statuses to be updated asynchronously.
- for (Job job : client.getPeonJobs()) {
+ // Load tasks from previously running jobs and wait for their statuses to start running.
+ final List<ListenableFuture<Boolean>> taskStatusActiveList = new ArrayList<>();
+ final List<Job> peonJobs = client.getPeonJobs();
+
+ log.info("Locating [%,d] active tasks.", peonJobs.size());
+ for (Job job : peonJobs) {
try {
- joinAsync(adapter.toTask(job));
+ KubernetesWorkItem kubernetesWorkItem = joinAsync(adapter.toTask(job));
+ taskStatusActiveList.add(kubernetesWorkItem.getPeonLifeycle().getTaskStartedSuccessfullyFuture());
}
catch (IOException e) {
log.error(e, "Error deserializing task from job [%s]", job.getMetadata().getName());
}
}
- log.info("Loaded %,d tasks from previous run", tasks.size());
+
+ try {
+ final DateTime nowUtc = DateTimes.nowUtc();
+ final long timeoutMs = nowUtc.plus(config.getTaskJoinTimeout()).getMillis() - nowUtc.getMillis();
+ if (timeoutMs > 0) {
+ FutureUtils.coalesce(taskStatusActiveList).get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+ log.info("Located [%,d] active tasks.", taskStatusActiveList.size());
+ }
+ catch (Exception e) {
+ final long numInitialized =
+ tasks.values().stream().filter(item -> item.getPeonLifeycle().getTaskStartedSuccessfullyFuture().isDone()).count();
Review Comment:
This is just looking for how many of the futures finished. Should it be looking for ones that finish *and* eval to `true`? I am wondering what the boolean is for.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -136,14 +141,16 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout,
launchTimeout,
TimeUnit.MILLISECONDS
);
-
return join(timeout);
}
catch (Exception e) {
log.info("Failed to run task: %s", taskId.getOriginalTaskId());
throw e;
}
finally {
+ if (!taskStartedSuccessfullyFuture.isDone()) {
Review Comment:
taskStartedSuccessfullyFuture is not set to `true` in `run` as far as I can tell. is that intentional?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1833176108
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java:
##########
@@ -321,16 +333,40 @@ public List<Pair<Task, ListenableFuture<TaskStatus>>> restore()
public void start()
{
log.info("Starting K8sTaskRunner...");
- // Load tasks from previously running jobs and wait for their statuses to be updated asynchronously.
- for (Job job : client.getPeonJobs()) {
+ // Load tasks from previously running jobs and wait for their statuses to start running.
+ final List<ListenableFuture<Boolean>> taskStatusActiveList = new ArrayList<>();
+ final List<Job> peonJobs = client.getPeonJobs();
+
+ log.info("Locating [%,d] active tasks.", peonJobs.size());
+ for (Job job : peonJobs) {
try {
- joinAsync(adapter.toTask(job));
+ KubernetesWorkItem kubernetesWorkItem = joinAsync(adapter.toTask(job));
+ taskStatusActiveList.add(kubernetesWorkItem.getPeonLifeycle().getTaskStartedSuccessfullyFuture());
}
catch (IOException e) {
log.error(e, "Error deserializing task from job [%s]", job.getMetadata().getName());
}
}
- log.info("Loaded %,d tasks from previous run", tasks.size());
+
+ try {
+ final DateTime nowUtc = DateTimes.nowUtc();
+ final long timeoutMs = nowUtc.plus(config.getTaskJoinTimeout()).getMillis() - nowUtc.getMillis();
+ if (timeoutMs > 0) {
+ FutureUtils.coalesce(taskStatusActiveList).get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+ log.info("Located [%,d] active tasks.", taskStatusActiveList.size());
+ }
+ catch (Exception e) {
+ final long numInitialized =
+ tasks.values().stream().filter(item -> item.getPeonLifeycle().getTaskStartedSuccessfullyFuture().isDone()).count();
Review Comment:
Ok, thanks. I missed that there was a different future that did have a callback for failure.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] k8s-based-ingestion: Wait for task lifecycles to enter RUNNING state before returning from KubernetesTaskRunner.start (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #17446:
URL: https://github.com/apache/druid/pull/17446#discussion_r1831582139
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -256,13 +268,18 @@ protected TaskLocation getTaskLocation()
if (taskLocation == null) {
Optional<Pod> maybePod = kubernetesClient.getPeonPod(taskId.getK8sJobName());
if (!maybePod.isPresent()) {
+ /* Arguably we should throw a exception here but leaving it as a warn log to prevent unexpected errors.
+ If there is strange behavior during overlord restarts the operator should look for this warn log.
+ */
+ log.warn("Could not get task location from k8s.");
return TaskLocation.unknown();
}
Pod pod = maybePod.get();
PodStatus podStatus = pod.getStatus();
if (podStatus == null || podStatus.getPodIP() == null) {
+ log.warn("Could not get task location from k8s.");
Review Comment:
Log the task id please.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java:
##########
@@ -119,4 +122,9 @@ public Task getTask()
{
return task;
}
+
+ public SettableFuture<Boolean> getTaskActiveStatusFuture()
Review Comment:
Javadoc please about what this represents.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -256,13 +268,18 @@ protected TaskLocation getTaskLocation()
if (taskLocation == null) {
Optional<Pod> maybePod = kubernetesClient.getPeonPod(taskId.getK8sJobName());
if (!maybePod.isPresent()) {
+ /* Arguably we should throw a exception here but leaving it as a warn log to prevent unexpected errors.
+ If there is strange behavior during overlord restarts the operator should look for this warn log.
+ */
+ log.warn("Could not get task location from k8s.");
Review Comment:
Log the task id please.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -174,11 +176,13 @@ private void writeTaskPayload(Task task) throws IOException
* @return
* @throws IllegalStateException
*/
- protected synchronized TaskStatus join(long timeout) throws IllegalStateException
+ protected synchronized TaskStatus join(long timeout, SettableFuture<Boolean> taskActiveStatusFuture) throws IllegalStateException
Review Comment:
This is an odd pattern; it's more typical for a method to return `ListenableFuture<T>`. If the caller needs to chain that onto a `SettableFuture<T>`, the caller should attach a callback that sets the settable future.
Please also add javadoc explaining what the returned future means.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org