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 2023/08/21 16:10:59 UTC
[PR] Use task manager for task payload (druid)
georgew5656 opened a new pull request, #14887:
URL: https://github.com/apache/druid/pull/14887
This change is meant to fix a issue where passing too large of a task payload to the mm-less task runner will cause the peon to fail to startup because the payload is passed (compressed) as a environment variable (TASK_JSON). In linux systems the limit for a environment variable is commonly 128KB, for windows systems less than this. Setting a env variable longer than this results in a bunch of "Argument list too long" errors.
### Description
The goal of this patch is to prevent larger tasks from failing with mm-less ingestion.
To address the immediate problem (setting environment variables that are too large), I added a additional config for the KubernetesTaskRunner (druid.indexer.runner.taskPayloadAsEnvVariable) that defaults to true but can be optionally set to false. Setting this config to false will cause the K8s adapters to not set the task payload as the TASK_JSON env variable. This prevents the Jobs from failing to come up.
We still need to pass the task.json payload to the peons somehow. I explored three options for this, and ended up going with option 3.
1. Using configmaps to store the task payload and then mounting them onto the created peon pods.
I decided not to go with this option because configmaps still have a 1MB size limit and I was concerned with the KubernetesTaskRunner having to manage a bunch of configmaps in addition to jobs. Having this many configmaps also pollutes K8s metadata, making it hard to see anything else going on.
2. Updating CliPeon to use the getTaskPayload endpoint on the overlord to pull the task.json payload on startup. This didn't work because we currently have a guice injector in the peon that requires the task.json be available at injection time. In order to pull the task.json from the overlord, we need to use the ServiceLocator class which is only available once the peon lifecycle has already started (after injection). Changing this would have required many changes to the code so I didn't want to do it. Additionally, I would have had to deprecate the toTask interface on the overlords since there would be no way for the overlord to turn a K8s Job into a task definition.
3. Push the task payload into task logs deep storage and have the peon read the payload.
I ended up going with this option because it was the most simple to implement and the most future-proof (no worry about task payloads getting larger than 1MB). The task logs killer will automatically delete the task.json in deep storage alongside he task logs whenever it is run.
To implement option 3 I included several changes
- Introduced a new interface (TaskPayloadManager) that exposes two methods (push task payload, pull task payload) and is implemented by TaskLogs. (I only implemented it for S3TaskLogs, but it should be easy to implement for other deep storage systems).
- Introduce a new config in TaskConfig (druid.indexer.task.enableTaskPayloadManagerPerTask). When set on the overlord, mm-less ingestion will push the task.json payload to deep storage before launching a k8s job. When set on the peon, the CliPeon task injector will read the task.json from deep storage instead of assuming it is available on the file system. This is currently only useable by mm-less ingestion but it technically could be used in any task running scenario.
- The K8s adapters in mm-less ingestion will check deep storage if the config is set when converting K8s jobs to tasks in toTask
#### Release note
- Support compressed task payloads larger than 128KB in mm-less ingestion.
##### Key changed/added classes in this PR
* `CliPeon`
* `KubernetesPeonLifecycle `
* `PodTemplateTaskAdapter`
* `K8sTaskAdapter`
* `S3TaskLogs` `TaskLogs`
I can add some more documentation to this PR later but I wanted to get some feedback on this approach before doing so.
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)
- [X] 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.
- [X] 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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1308040993
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -145,6 +153,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
+ Path file = Files.createTempFile(taskId.getOriginalTaskId(), "task.json");
+ try {
Review Comment:
the issue is the deleteIfExists call throws a IOException so I didn't want to have the whole function do that
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1303311963
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -145,6 +153,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
+ Path file = Files.createTempFile(taskId.getOriginalTaskId(), "task.json");
+ try {
Review Comment:
how do you reccommend simplifying this? the createTempFile can throw a IOException
--
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] Use task manager for task payload (druid)
Posted by "YongGang (via GitHub)" <gi...@apache.org>.
YongGang commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1307805224
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerConfig.java:
##########
@@ -403,6 +419,12 @@ public Builder withCapacity(@Min(0) @Max(Integer.MAX_VALUE) Integer capacity)
return this;
}
+ public Builder withTaskPayloadAsEnvVariable(Boolean taskPayloadAsEnvVariable)
Review Comment:
I don't see where this method is called.
And also could we consolidate the two new config items to one? The current two configs are excluding each other thus can be misconfigured. Maybe like this:
| Config Name | Type | Description | Default | Required |
|-------------|------|-------------|---------|----------|
| `druid.indexer.task.taskPayloadHandlingMethod` | `String` | Determines how task payloads should be passed to K8s jobs. Use `ENV_VARIABLE` if you want task payloads to be passed as a compressed environment variable (suitable for payloads <= 1 MB). Use `PAYLOAD_MANAGER` if you plan on using task payloads > 1 MB. | `ENV_VARIABLE` | No |
##########
docs/development/extensions-contrib/k8s-jobs.md:
##########
@@ -216,23 +216,25 @@ data:
```
### Properties
-|Property| Possible Values | Description |Default|required|
-|--------|-----------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------|--------|
-|`druid.indexer.runner.debugJobs`| `boolean` | Clean up K8s jobs after tasks complete. |False|No|
-|`druid.indexer.runner.sidecarSupport`| `boolean` | Deprecated, specify adapter type as runtime property `druid.indexer.runner.k8s.adapter.type: overlordMultiContainer` instead. If your overlord pod has sidecars, this will attempt to start the task with the same sidecars as the overlord pod. |False|No|
-|`druid.indexer.runner.primaryContainerName`| `String` | If running with sidecars, the `primaryContainerName` should be that of your druid container like `druid-overlord`. |First container in `podSpec` list|No|
-|`druid.indexer.runner.kubexitImage`| `String` | Used kubexit project to help shutdown sidecars when the main pod completes. Otherwise jobs with sidecars never terminate. |karlkfi/kubexit:latest|No|
-|`druid.indexer.runner.disableClientProxy`| `boolean` | Use this if you have a global http(s) proxy and you wish to bypass it. |false|No|
-|`druid.indexer.runner.maxTaskDuration`| `Duration` | Max time a task is allowed to run for before getting killed |`PT4H`|No|
-|`druid.indexer.runner.taskCleanupDelay`| `Duration` | How long do jobs stay around before getting reaped from K8s |`P2D`|No|
-|`druid.indexer.runner.taskCleanupInterval`| `Duration` | How often to check for jobs to be reaped |`PT10M`|No|
-|`druid.indexer.runner.K8sjobLaunchTimeout`| `Duration` | How long to wait to launch a K8s task before marking it as failed, on a resource constrained cluster it may take some time. |`PT1H`|No|
-|`druid.indexer.runner.javaOptsArray`| `JsonArray` | java opts for the task. |`-Xmx1g`|No|
-|`druid.indexer.runner.labels`| `JsonObject` | Additional labels you want to add to peon pod |`{}`|No|
-|`druid.indexer.runner.annotations`| `JsonObject` | Additional annotations you want to add to peon pod |`{}`|No|
-|`druid.indexer.runner.peonMonitors`| `JsonArray` | Overrides `druid.monitoring.monitors`. Use this property if you don't want to inherit monitors from the Overlord. |`[]`|No|
-|`druid.indexer.runner.graceTerminationPeriodSeconds`| `Long` | Number of seconds you want to wait after a sigterm for container lifecycle hooks to complete. Keep at a smaller value if you want tasks to hold locks for shorter periods. |`PT30S` (K8s default)|No|
-|`druid.indexer.runner.capacity`| `Integer` | Number of concurrent jobs that can be sent to Kubernetes. |`2147483647`|No|
+|Property| Possible Values | Description | Default |required|
Review Comment:
As I have been told the doc shouldn't be reformatted by the IDE (you can uncheck the `Reformat table when typing` option).
--
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] Use task manager for task payload (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1691123834
Thanks @churromorales and @georgew5656 for the responses. I was more worried about adding another config which needs to be set for the end users.
The current opt out by default lgtm.
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1334679880
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -132,11 +140,36 @@ public Task toTask(Job from) throws IOException
Optional<EnvVar> taskJson = envVars.stream().filter(x -> "TASK_JSON".equals(x.getName())).findFirst();
String contents = taskJson.map(envVar -> taskJson.get().getValue()).orElse(null);
if (contents == null) {
- throw new IOException("No TASK_JSON environment variable found in pod: " + from.getMetadata().getName());
+ log.info("No TASK_JSON environment variable found in pod: %s. Trying to load task payload from deep storage.", from.getMetadata().getName());
+ return toTaskUsingDeepStorage(from);
}
return mapper.readValue(Base64Compression.decompressBase64(contents), Task.class);
}
+ private Task toTaskUsingDeepStorage(Job from) throws IOException
+ {
+ com.google.common.base.Optional<InputStream> taskBody = taskLogs.streamTaskPayload(getTaskId(from).getOriginalTaskId());
+ if (!taskBody.isPresent()) {
+ throw new IOE("Could not load task payload for job [%s]", from.getMetadata().getName());
Review Comment:
the druid exception classes all look like query exceptions to me, are u suggesting adding a new internal exception or something?
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1704823400
I have concerns about the config change as well. In a single PR adding one config doesn't seem much, but after a year worth of work, you suddenly realize that the feature has become very complex to tune and use. Is there a way to check if the deep storage supports storing payload and if not, then using the environment variable? Because then we can change the default easily.
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1334668034
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -144,6 +148,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
+ Path file = Files.createTempFile(taskId.getOriginalTaskId(), "task.json");
+ try {
+ FileUtils.writeStringToFile(file.toFile(), mapper.writeValueAsString(task), Charset.defaultCharset());
+ taskLogs.pushTaskPayload(task.getId(), file.toFile());
Review Comment:
the task reads the file to disk on startup so i wouldn't be worried about the log cleanup job to clean it up that soon. we are relying on the log cleanup job to cleanup deep storage
--
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] Use task manager for task payload (druid)
Posted by "YongGang (via GitHub)" <gi...@apache.org>.
YongGang commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1303568359
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -145,6 +153,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
+ Path file = Files.createTempFile(taskId.getOriginalTaskId(), "task.json");
+ try {
Review Comment:
Don't think we need to handle the IOException explicitly as it is wrapped to RuntimeException anyway.
So the code can be like:
```
Path file = null;
try {
file = Files.createTempFile(taskId.getOriginalTaskId(), "task.json");
FileUtils.writeStringToFile(file.toFile(), mapper.writeValueAsString(task), Charset.defaultCharset());
taskLogs.pushTaskPayload(task.getId(), file.toFile());
}
catch (Exception e) {
log.error("Failed to write task payload for task: %s", taskId.getOriginalTaskId());
throw new RuntimeException(e);
}
finally {
if (file ! = null) {
Files.deleteIfExists(file);
}
}
}
```
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1744479658
@georgew5656 - I will let you merge with a nicer commit summary that what github is generating.
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1300399696
##########
processing/src/main/java/org/apache/druid/tasklogs/TaskPayloadManager.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.tasklogs;
+
+import com.google.common.base.Optional;
+import org.apache.commons.lang.NotImplementedException;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+
+public interface TaskPayloadManager
+{
+ default void pushTaskPayload(String taskid, File taskPayloadFile) throws IOException
+ {
+ throw new NotImplementedException("Managing task payloads is not implemented for this druid.indexer.logs.type");
Review Comment:
The error message here is written with a developer persona in mind. As a user, following error will help me
```
this druid.indexer.logs.type [%s] does not support managing task payloads yet. You will have to switch to using environment variables, getClass()
```
something along those lines.
##########
processing/src/main/java/org/apache/druid/tasklogs/TaskPayloadManager.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.tasklogs;
+
+import com.google.common.base.Optional;
+import org.apache.commons.lang.NotImplementedException;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+
+public interface TaskPayloadManager
Review Comment:
Please add javadocs.
--
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] Use task manager for task payload (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1687271207
@georgew5656
The current approach also SGTM but I would not have if behind a feature flag. If you want to have a feature flag, I would optionally turn it on by default .
This also means we implement that TaskPayloadManager for other deep storage mainly 'GCS,AZURE,HDFS`. If the implementation is not found, we should through a nice error message instructing the user to set `druid.indexer.runner.taskPayloadAsEnvVariable` to true or if we are not using a flag, instructing the user to implement the methods to implement 'TaskPayloadManager` for there custom deep storage impl.
--
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] Use task manager for task payload (druid)
Posted by "churromorales (via GitHub)" <gi...@apache.org>.
churromorales commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1688853183
@cryptoe I think the way it is implemented now behind the feature flag is better. For most usecases it is much better to just pass the task.json directly. It is much faster, also for our customers we use deep-storage which is slow and less featured than s3, thus this would be a feature we would only turn on if necessary. Additionally, our deep storage provider wont allow us to do batch deletes so the cleaner approach to remove the task files is not great. I know its a special case for us, but it is always better to pass something directly if possible than use indirection...disregarding our usecase.
LGTM overall, but one feature request I would like to request is that at the end of the task in AbstractTask to delete the task.json file, not leave it up to the cleaner. Quite a few folks that have their own k8s which they launch in their datacenters. While this works for cloud providers, it might not work for everyone else. This can be done in another PR or I can one up after this PR is merged if needed.
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1690328968
> @georgew5656 The current approach also SGTM but I would not have it behind a feature flag. If you want to have a feature flag, I would optionally turn it on by default . This also means we implement that TaskPayloadManager for other deep storage mainly 'GCS,AZURE,HDFS`. If the implementation is not found, we should through a nice error message instructing the user to set `druid.indexer.runner.taskPayloadAsEnvVariable` to true or if we are not using a flag, instructing the user to implement the methods to implement 'TaskPayloadManager` for there custom deep storage impl.
i didn't really want to break anyone who didn't want to use deep storage for task payloads for whatever reason. i think maybe later on if this gets used in production a bit more and the performance is okay maybe we could consider flipping it on as a default
--
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] Use task manager for task payload (druid)
Posted by "YongGang (via GitHub)" <gi...@apache.org>.
YongGang commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1300510528
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -145,6 +153,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
+ Path file = Files.createTempFile(taskId.getOriginalTaskId(), "task.json");
+ try {
Review Comment:
The nested try-catch seems can be simplified.
##########
services/src/main/java/org/apache/druid/cli/CliPeon.java:
##########
@@ -282,9 +290,15 @@ public Supplier<Map<String, Object>> heartbeatDimensions(Task task)
@Provides
@LazySingleton
- public Task readTask(@Json ObjectMapper mapper, ExecutorLifecycleConfig config)
+ public Task readTask(@Json ObjectMapper mapper, @Smile ObjectMapper smileMapper, ExecutorLifecycleConfig config, TaskPayloadManager taskPayloadManager, TaskConfig taskConfig)
Review Comment:
What's the reason to use a Smile mapper? Seems same as Json mapper.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/TaskAdapter.java:
##########
@@ -31,4 +32,5 @@ public interface TaskAdapter
Task toTask(Job from) throws IOException;
+ K8sTaskId getTaskId(Job from) throws IOException;
Review Comment:
Seems this method has same implementation in the subclasses, maybe move here as a default method.
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1334669694
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -119,11 +119,15 @@ protected KubernetesPeonLifecycle(
* @return
* @throws IllegalStateException
*/
- protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout) throws IllegalStateException
+ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout, Boolean useDeepStorageForTaskPayload) throws IllegalStateException
Review Comment:
none really
--
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] Use task manager for task payload (druid)
Posted by "churromorales (via GitHub)" <gi...@apache.org>.
churromorales commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1716148715
Why not just check the size of the task.json, if its larger than then MAX_SIZE do it in deep storage, if it is smaller then just use the env? I agree 100% with Druid in general having too many configuration options, makes it hard to remember everything you need include. When creating this feature, the whole goal I had in mind was to have this work with as few configuration options as possible.
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1334928374
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -144,6 +148,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
Review Comment:
Yeah. I think so. Unless you don't want to handle it up the stack.
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1334682414
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -433,5 +475,18 @@ static ResourceRequirements getResourceRequirements(ResourceRequirements require
}
return requirements;
}
+
+ @Override
+ public Boolean shouldUseDeepStorageForTaskPayload(Task task)
+ {
+ try {
+ String compressedTaskPayload = Base64Compression.compressBase64(mapper.writeValueAsString(task));
+ return compressedTaskPayload.length() > DruidK8sConstants.MAX_ENV_VARIABLE_KBS;
+ }
+ catch (Exception e) {
Review Comment:
the compiler is forcing the call, but I guess we can make this method throw IOException since all the other methods in the interface do
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1341005123
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -132,11 +141,39 @@ public Task toTask(Job from) throws IOException
Optional<EnvVar> taskJson = envVars.stream().filter(x -> "TASK_JSON".equals(x.getName())).findFirst();
String contents = taskJson.map(envVar -> taskJson.get().getValue()).orElse(null);
if (contents == null) {
- throw new IOException("No TASK_JSON environment variable found in pod: " + from.getMetadata().getName());
+ log.info("No TASK_JSON environment variable found in pod: %s. Trying to load task payload from deep storage.", from.getMetadata().getName());
+ return toTaskUsingDeepStorage(from);
}
return mapper.readValue(Base64Compression.decompressBase64(contents), Task.class);
}
+ private Task toTaskUsingDeepStorage(Job from) throws IOException
+ {
+ com.google.common.base.Optional<InputStream> taskBody = taskLogs.streamTaskPayload(getTaskId(from).getOriginalTaskId());
+ if (!taskBody.isPresent()) {
+ throw InternalServerError.exception(
+ "Could not load task payload from deep storage for job [%s]. Check the overlord logs for errors uploading task payloads to deep storage.",
Review Comment:
```suggestion
"Could not load task payload from deep storage for job [%s]. Check the overlord logs for any errors in uploading task payload to deep storage.",
```
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1303311446
##########
services/src/main/java/org/apache/druid/cli/CliPeon.java:
##########
@@ -282,9 +290,15 @@ public Supplier<Map<String, Object>> heartbeatDimensions(Task task)
@Provides
@LazySingleton
- public Task readTask(@Json ObjectMapper mapper, ExecutorLifecycleConfig config)
+ public Task readTask(@Json ObjectMapper mapper, @Smile ObjectMapper smileMapper, ExecutorLifecycleConfig config, TaskPayloadManager taskPayloadManager, TaskConfig taskConfig)
Review Comment:
we use the smile mapper to serialize the payload so i figured we should use the same one to deserialize
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1334148644
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -119,11 +119,15 @@ protected KubernetesPeonLifecycle(
* @return
* @throws IllegalStateException
*/
- protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout) throws IllegalStateException
+ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout, Boolean useDeepStorageForTaskPayload) throws IllegalStateException
Review Comment:
it seems inconsistent that some arguments are primitives (long) and some are objects (Boolean). Whats the reason?
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -144,6 +148,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
Review Comment:
nit - you could have avoided nested tries and handling exception twice by just declaring a null file object and deleting the file only when the said object is not null.
##########
processing/src/main/java/org/apache/druid/tasklogs/TaskPayloadManager.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.tasklogs;
+
+import com.google.common.base.Optional;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.druid.guice.annotations.ExtensionPoint;
+import org.apache.druid.java.util.common.StringUtils;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Something that knows how to manage task json paylods.
Review Comment:
The javadocs can be more descriptive. Like how it is used, where it is used etc. "manage task json payloads" is very broad description. It's only used in k8s task management.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -132,11 +140,36 @@ public Task toTask(Job from) throws IOException
Optional<EnvVar> taskJson = envVars.stream().filter(x -> "TASK_JSON".equals(x.getName())).findFirst();
String contents = taskJson.map(envVar -> taskJson.get().getValue()).orElse(null);
if (contents == null) {
- throw new IOException("No TASK_JSON environment variable found in pod: " + from.getMetadata().getName());
+ log.info("No TASK_JSON environment variable found in pod: %s. Trying to load task payload from deep storage.", from.getMetadata().getName());
+ return toTaskUsingDeepStorage(from);
}
return mapper.readValue(Base64Compression.decompressBase64(contents), Task.class);
}
+ private Task toTaskUsingDeepStorage(Job from) throws IOException
+ {
+ com.google.common.base.Optional<InputStream> taskBody = taskLogs.streamTaskPayload(getTaskId(from).getOriginalTaskId());
+ if (!taskBody.isPresent()) {
+ throw new IOE("Could not load task payload for job [%s]", from.getMetadata().getName());
Review Comment:
we should start using DruidException classes instead of IOE.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -89,14 +94,16 @@ public abstract class K8sTaskAdapter implements TaskAdapter
protected final StartupLoggingConfig startupLoggingConfig;
protected final DruidNode node;
protected final ObjectMapper mapper;
+ protected final TaskLogs taskLogs;
public K8sTaskAdapter(
KubernetesClientApi client,
KubernetesTaskRunnerConfig taskRunnerConfig,
TaskConfig taskConfig,
StartupLoggingConfig startupLoggingConfig,
DruidNode node,
- ObjectMapper mapper
+ ObjectMapper mapper,
+ TaskLogs taskLogs
Review Comment:
the name of this class has been confusing since it does so much more than dealing with task logs. could be fixed in some other PR someday.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -144,6 +148,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
+ Path file = Files.createTempFile(taskId.getOriginalTaskId(), "task.json");
+ try {
+ FileUtils.writeStringToFile(file.toFile(), mapper.writeValueAsString(task), Charset.defaultCharset());
+ taskLogs.pushTaskPayload(task.getId(), file.toFile());
Review Comment:
is it possible that a log cleanup job removes the task payload from the deep storage while task is still in progress? How are these payloads cleaned up from deep storage?
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -433,5 +475,18 @@ static ResourceRequirements getResourceRequirements(ResourceRequirements require
}
return requirements;
}
+
+ @Override
+ public Boolean shouldUseDeepStorageForTaskPayload(Task task)
+ {
+ try {
+ String compressedTaskPayload = Base64Compression.compressBase64(mapper.writeValueAsString(task));
+ return compressedTaskPayload.length() > DruidK8sConstants.MAX_ENV_VARIABLE_KBS;
+ }
+ catch (Exception e) {
Review Comment:
is any of the call in the method actually throwing an exception here? The task should be serializable. If compiler is not forcing you to catch Exception, I would suggest you avoid doing it. At least avoid it here, let it bubble up and then handle at the top of the call stack. Otherwise, your call will be littered with unnecessary catch blocks.
##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3TaskLogs.java:
##########
@@ -83,6 +83,21 @@ public Optional<InputStream> streamTaskStatus(String taskid) throws IOException
return streamTaskFileWithRetry(0, taskKey);
}
+ @Override
+ public void pushTaskPayload(String taskid, File taskPayloadFile) throws IOException
+ {
+ final String taskKey = getTaskLogKey(taskid, "task.json");
+ log.info("Pushing task payload %s to: %s", taskPayloadFile, taskKey);
Review Comment:
```suggestion
log.info("Pushing task payload [%s] to location [%s]", taskPayloadFile, taskKey);
```
##########
distribution/docker/peon.sh:
##########
@@ -150,6 +150,6 @@ then
fi
# take the ${TASK_JSON} environment variable and base64 decode, unzip and throw it in ${TASK_DIR}/task.json
Review Comment:
the comment needs an update?
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1741477991
> Almost there. Can you look into the test failures?
i think the only failing tests are for coverage now so we should be good
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1721434953
> Why not just check the size of the task.json, if its larger than then MAX_SIZE do it in deep storage, if it is smaller then just use the env? I agree 100% with Druid in general having too many configuration options, makes it hard to remember everything you need include. When creating this feature, the whole goal I had in mind was to have this work with as few configuration options as possible.
i think MAX_SIZE depends on the underlying OS but I don't mind using the 100KB default for linux and doing it that way
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1721756551
@abhishekagarwal87 @churromorales I just put up a new version of the PR that removes the configs and uses deep storage to pass the task payload if the task size is too large. This is more of a "opinionated" choice but in this case I think it makes sense, lmk what you guys think
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1339808899
##########
processing/src/main/java/org/apache/druid/error/InternalError.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.error;
+
+public class InternalError extends DruidException.Failure
Review Comment:
how about InternalServerError?
##########
processing/src/main/java/org/apache/druid/error/InternalError.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.error;
+
+public class InternalError extends DruidException.Failure
+{
+
+ public static DruidException exception()
+ {
+ return exception("Internal Error");
+ }
+
+ public static DruidException exception(String msg, Object... args)
+ {
+ return exception(null, msg, args);
+ }
+
+ public static DruidException exception(Throwable t, String msg, Object... args)
+ {
+ return DruidException.fromFailure(new InternalError(t, msg, args));
+ }
+
+ private final Throwable t;
+ private final String msg;
+ private final Object[] args;
+
+ private InternalError(
+ Throwable t,
+ String msg,
+ Object... args
+ )
+ {
+ super("canceled");
+ this.t = t;
+ this.msg = msg;
+ this.args = args;
+ }
+
+ @Override
+ public DruidException makeException(DruidException.DruidExceptionBuilder bob)
+ {
+ bob = bob.forPersona(DruidException.Persona.OPERATOR)
+ .ofCategory(DruidException.Category.CANCELED);
Review Comment:
it should be runtime_failure category.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -132,11 +141,36 @@ public Task toTask(Job from) throws IOException
Optional<EnvVar> taskJson = envVars.stream().filter(x -> "TASK_JSON".equals(x.getName())).findFirst();
String contents = taskJson.map(envVar -> taskJson.get().getValue()).orElse(null);
if (contents == null) {
- throw new IOException("No TASK_JSON environment variable found in pod: " + from.getMetadata().getName());
+ log.info("No TASK_JSON environment variable found in pod: %s. Trying to load task payload from deep storage.", from.getMetadata().getName());
+ return toTaskUsingDeepStorage(from);
}
return mapper.readValue(Base64Compression.decompressBase64(contents), Task.class);
}
+ private Task toTaskUsingDeepStorage(Job from) throws IOException
+ {
+ com.google.common.base.Optional<InputStream> taskBody = taskLogs.streamTaskPayload(getTaskId(from).getOriginalTaskId());
+ if (!taskBody.isPresent()) {
+ throw InternalError.exception("Could not load task payload for job [%s]", from.getMetadata().getName());
Review Comment:
is there an action you can associate with this error message? Like should they verify that overlord is successfully uploading task jsons to deep storage.
##########
processing/src/main/java/org/apache/druid/error/InternalError.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.error;
+
+public class InternalError extends DruidException.Failure
+{
+
+ public static DruidException exception()
+ {
+ return exception("Internal Error");
+ }
+
+ public static DruidException exception(String msg, Object... args)
+ {
+ return exception(null, msg, args);
+ }
+
+ public static DruidException exception(Throwable t, String msg, Object... args)
+ {
+ return DruidException.fromFailure(new InternalError(t, msg, args));
+ }
+
+ private final Throwable t;
+ private final String msg;
+ private final Object[] args;
+
+ private InternalError(
+ Throwable t,
+ String msg,
+ Object... args
+ )
+ {
+ super("canceled");
Review Comment:
Why canceled? Instead, you can force the caller to pass on an error code.
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -132,11 +141,36 @@ public Task toTask(Job from) throws IOException
Optional<EnvVar> taskJson = envVars.stream().filter(x -> "TASK_JSON".equals(x.getName())).findFirst();
String contents = taskJson.map(envVar -> taskJson.get().getValue()).orElse(null);
if (contents == null) {
- throw new IOException("No TASK_JSON environment variable found in pod: " + from.getMetadata().getName());
+ log.info("No TASK_JSON environment variable found in pod: %s. Trying to load task payload from deep storage.", from.getMetadata().getName());
+ return toTaskUsingDeepStorage(from);
}
return mapper.readValue(Base64Compression.decompressBase64(contents), Task.class);
}
+ private Task toTaskUsingDeepStorage(Job from) throws IOException
+ {
+ com.google.common.base.Optional<InputStream> taskBody = taskLogs.streamTaskPayload(getTaskId(from).getOriginalTaskId());
+ if (!taskBody.isPresent()) {
+ throw InternalError.exception("Could not load task payload for job [%s]", from.getMetadata().getName());
+ }
+ String task = IOUtils.toString(taskBody.get(), Charset.defaultCharset());
+ return mapper.readValue(task, Task.class);
+ }
+
+ @Override
+ public K8sTaskId getTaskId(Job from) throws IOException
+ {
+ Map<String, String> annotations = from.getSpec().getTemplate().getMetadata().getAnnotations();
+ if (annotations == null) {
+ throw new IOE("No annotations found on pod spec for job [%s]", from.getMetadata().getName());
Review Comment:
can this be replaced with DruidException.defensive()?
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1340147947
##########
processing/src/main/java/org/apache/druid/error/InternalError.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.error;
+
+public class InternalError extends DruidException.Failure
+{
+
+ public static DruidException exception()
+ {
+ return exception("Internal Error");
+ }
+
+ public static DruidException exception(String msg, Object... args)
+ {
+ return exception(null, msg, args);
+ }
+
+ public static DruidException exception(Throwable t, String msg, Object... args)
+ {
+ return DruidException.fromFailure(new InternalError(t, msg, args));
+ }
+
+ private final Throwable t;
+ private final String msg;
+ private final Object[] args;
+
+ private InternalError(
+ Throwable t,
+ String msg,
+ Object... args
+ )
+ {
+ super("canceled");
Review Comment:
i just figured this was the default 500 error code
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1307811855
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerConfig.java:
##########
@@ -403,6 +419,12 @@ public Builder withCapacity(@Min(0) @Max(Integer.MAX_VALUE) Integer capacity)
return this;
}
+ public Builder withTaskPayloadAsEnvVariable(Boolean taskPayloadAsEnvVariable)
Review Comment:
this is called in the tests. we can't separate them because they are doing different things. taskPayloadAsEnvVariable is enabling/disabling the TASK_JSON env variable, and the other one is enabling/disabling managing the payload via deep storage. If we make these one config then there is the issue of migration/backwards compatibility.
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1303314485
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/TaskAdapter.java:
##########
@@ -31,4 +32,5 @@ public interface TaskAdapter
Task toTask(Job from) throws IOException;
+ K8sTaskId getTaskId(Job from) throws IOException;
Review Comment:
i didn't want to do this because the implementation depends on logic in fromTask, since both adapters happen to add the same annotation for taskId the implementation for getTaskId is the same but this isn't necessarily the case
--
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] Use task manager for task payload (druid)
Posted by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1327741763
##########
services/src/main/java/org/apache/druid/cli/CliPeon.java:
##########
@@ -282,9 +290,15 @@
@Provides
@LazySingleton
- public Task readTask(@Json ObjectMapper mapper, ExecutorLifecycleConfig config)
+ public Task readTask(@Json ObjectMapper mapper, @Smile ObjectMapper smileMapper, ExecutorLifecycleConfig config, TaskPayloadManager taskPayloadManager, TaskConfig taskConfig)
Review Comment:
## Useless parameter
The parameter 'smileMapper' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/5795)
##########
services/src/main/java/org/apache/druid/cli/CliPeon.java:
##########
@@ -282,9 +290,15 @@
@Provides
@LazySingleton
- public Task readTask(@Json ObjectMapper mapper, ExecutorLifecycleConfig config)
+ public Task readTask(@Json ObjectMapper mapper, @Smile ObjectMapper smileMapper, ExecutorLifecycleConfig config, TaskPayloadManager taskPayloadManager, TaskConfig taskConfig)
Review Comment:
## Useless parameter
The parameter 'taskConfig' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/5796)
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1334681055
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -144,6 +148,28 @@ protected synchronized TaskStatus run(Job job, long launchTimeout, long timeout)
}
}
+ private void writeTaskPayload(Task task)
+ {
+ try {
Review Comment:
i would still have to add a IOException to method signature since the delete would happen in the finally block, but maybe that's okay?
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1334928532
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -132,11 +140,36 @@ public Task toTask(Job from) throws IOException
Optional<EnvVar> taskJson = envVars.stream().filter(x -> "TASK_JSON".equals(x.getName())).findFirst();
String contents = taskJson.map(envVar -> taskJson.get().getValue()).orElse(null);
if (contents == null) {
- throw new IOException("No TASK_JSON environment variable found in pod: " + from.getMetadata().getName());
+ log.info("No TASK_JSON environment variable found in pod: %s. Trying to load task payload from deep storage.", from.getMetadata().getName());
+ return toTaskUsingDeepStorage(from);
}
return mapper.readValue(Base64Compression.decompressBase64(contents), Task.class);
}
+ private Task toTaskUsingDeepStorage(Job from) throws IOException
+ {
+ com.google.common.base.Optional<InputStream> taskBody = taskLogs.streamTaskPayload(getTaskId(from).getOriginalTaskId());
+ if (!taskBody.isPresent()) {
+ throw new IOE("Could not load task payload for job [%s]", from.getMetadata().getName());
Review Comment:
They are not just query exceptions. They are used for generating exceptions that will be returned to the end user. So it could either be a query user, an ingestion user, or a cluster admin.
--
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] Use task manager for task payload (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on PR #14887:
URL: https://github.com/apache/druid/pull/14887#issuecomment-1732225327
thank you for addressing comments @georgew5656. Looks good to me except for the exception handling in some places.
--
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] Allow users to pass task payload via deep storage instead of environment variable (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 merged PR #14887:
URL: https://github.com/apache/druid/pull/14887
--
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] Use task manager for task payload (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #14887:
URL: https://github.com/apache/druid/pull/14887#discussion_r1340410459
##########
extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/taskadapter/K8sTaskAdapter.java:
##########
@@ -132,11 +141,36 @@ public Task toTask(Job from) throws IOException
Optional<EnvVar> taskJson = envVars.stream().filter(x -> "TASK_JSON".equals(x.getName())).findFirst();
String contents = taskJson.map(envVar -> taskJson.get().getValue()).orElse(null);
if (contents == null) {
- throw new IOException("No TASK_JSON environment variable found in pod: " + from.getMetadata().getName());
+ log.info("No TASK_JSON environment variable found in pod: %s. Trying to load task payload from deep storage.", from.getMetadata().getName());
+ return toTaskUsingDeepStorage(from);
}
return mapper.readValue(Base64Compression.decompressBase64(contents), Task.class);
}
+ private Task toTaskUsingDeepStorage(Job from) throws IOException
+ {
+ com.google.common.base.Optional<InputStream> taskBody = taskLogs.streamTaskPayload(getTaskId(from).getOriginalTaskId());
+ if (!taskBody.isPresent()) {
+ throw InternalError.exception("Could not load task payload for job [%s]", from.getMetadata().getName());
Review Comment:
updated the message
--
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