return task status reported by peon#14040
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
|
||
| Optional<InputStream> inputStreamOptional = s3TaskLogs.streamTaskStatus(KEY_1); | ||
| String report = new BufferedReader( | ||
| new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) |
Check warning
Code scanning / CodeQL
Potential input resource leak
…tisfy code coverage
clintropolis
left a comment
clintropolis
left a comment
There was a problem hiding this comment.
this seems fine to me 👍, and while the interface changes are to things marked as @ExtensionPoint the default implementation makes it not too disruptive.
| { | ||
| } | ||
|
|
||
| default void pushTaskStatus(String taskid, File reportFile) throws IOException |
There was a problem hiding this comment.
these new methods need javadocs
There was a problem hiding this comment.
Can we put together a follow up PR that adds javadocs to all of these methods.
| return Optional.absent(); | ||
| } | ||
|
|
||
| default Optional<InputStream> streamTaskStatus(final String taskid) throws IOException |
There was a problem hiding this comment.
javadocs for devs who would want to implement this method
There was a problem hiding this comment.
Can we put together a follow up PR that adds javadocs to all of these methods.
| @Override | ||
| public void pushTaskStatus(String taskid, File statusFile) throws IOException | ||
| { | ||
| final String taskKey = getTaskLogKey(taskid, "status.json"); | ||
| log.info("Pushing task status %s to: %s", statusFile, taskKey); | ||
| pushTaskFile(statusFile, taskKey); | ||
| } |
There was a problem hiding this comment.
could this be the default implementation?
| log.debug("No task reports file exists to push"); | ||
| } | ||
| } else { | ||
| if (!toolbox.getConfig().isEncapsulatedTask()) { |
There was a problem hiding this comment.
@churromorales @nlippis - does the druid.indexer.task.encapsulatedTask has to be a user-facing setting? seems like we could just set it to true always if the task runner is k8s.
There was a problem hiding this comment.
just to clarify, it's not a blocker comment for this PR.
There was a problem hiding this comment.
We could, especially since the peon depends on the druid.indexer.runner.type setting.
* return task status reported by peon * Write TaskStatus to file in AbstractTask.cleanUp * Get TaskStatus from task log * Fix merge conflicts in AbstractTaskTest * Add unit tests for TaskLogPusher, TaskLogStreamer, NoopTaskLogs to satisfy code coverage * Add license headerss * Fix style * Remove unknown exception declarations
4 participants
Return task status generated by Peon in KuberentesTaskRunner
The KubernetesTaskRunner determines the status of a completed task by looking at the status of the containers the job ran (Kubernetes marks a container as failed if it exits with a non zero exit code). If any of the containers succeeded, the task is reported as successful. If any of the containers failed, the task is reported as failed.
This way of determining final task status is incorrect for the following reasons
Adds a method to push and stream task status from object storage
Peon pushes task status to object storage
KubernetesTaskRunner returns task status read from object storage
Release note
If the Druid cluster is upgraded from a version that does not have this change to a version that does, peons running the old version of the code will not write their task status to object storage. When the task completes the KubernetesTaskRunner will use the previous logic in order to preserve backwards compatibility.
Key changed/added classes in this PR
KubernetesTaskRunnerAbstractTaskThis PR has: