Repository navigation
Improve task retrieval APIs on Overlord - #5801
Conversation
It takes 4 optional query params * state(pending/running/waiting/compelte) * dataSource * interval (applies to completed tasks) * maxCompletedTasks (applies to completed tasks) If all params are null, the api returns all the tasks
* divide active tasks into waiting, pending or running * Add more unit tests
| public enum TaskState | ||
| { | ||
| RUNNING, | ||
| WAITING, |
There was a problem hiding this comment.
Would you add descriptions for each state to here?
There was a problem hiding this comment.
And I wonder why these new states are needed in this PR. Would you share your thoughts more about this?
There was a problem hiding this comment.
This is an incompatible change. This PR should be tagged with Release notes and a note for rolling update should be added to the release notes if this change is needed.
There was a problem hiding this comment.
The tasks API returns all tasks if no state query param is passed, so the user of this API wants to know statusCode for each of the tasks. Current statusCode consist of running, success and failed. We need to also send appropriate statusCode for waiting and pending. That's why added those here. We can sure mark this PR as Release notes if you agree to this change.
There was a problem hiding this comment.
IMO, TaskState is not the right place to add these things. That's because TaskState's purpose is to model task status as stored in the database (see TaskStorage.insert and TaskStorage.setStatus), where there is no concept of "waiting" or "pending" (those are both stored as "running"). Splitting "running" into waiting/pending/really-really-running is more of a task runner concept.
So I think a new class would make more sense, geared towards modeling the runtime state of a task. Maybe RuntimeTaskState. And that one shouldn't be persisted into the database.
Please write javadocs that make the distinction clear, too.
There was a problem hiding this comment.
If I make another RuntimeTaskState do you suggest that to contain all the states : running pending waiting complete success failed or just subclass of running i.e. running waiting pending. The reason I ask is TaskStatusPlus object contains a single property called statusCode, which is the TaskState and UI uses this property to fill status. Should we create two separate status codes or combine everything in one. I prefer latter, i.e. return one status code.
There was a problem hiding this comment.
I was thinking the runtime task status code should be its own thing (not a subclass of task state).
There was a problem hiding this comment.
okay, i added RunnerTaskState and now TaskStatusPlus will return 2 status codes for each task statusCode and runnerStatusCode.
| SUCCESS, | ||
| FAILED; | ||
| FAILED, | ||
| UNKNOWN; |
There was a problem hiding this comment.
What does UNKNOWN state mean?
There was a problem hiding this comment.
So instead of passing null for status, I added this UNKNOWN state, when we don't know the state of a task.
| if (state != null) { | ||
| switch (StringUtils.toUpperCase(state)) { | ||
| case "PENDING": | ||
| return workItemsResponse( |
There was a problem hiding this comment.
There are already endpoints to get pending, waiting, running, or complete tasks. Please share the code path rather than duplicating.
There was a problem hiding this comment.
Yes, the idea is to potentially call this tasks api with appropriate state query param from the existing endpoints to get pending, waiting, running, complete tasks. I did not do that change yet as this api is still being tested in nightly, will do that once this api passes those tests.
| } | ||
|
|
||
| private static class WaitingTask extends TaskRunnerWorkItem | ||
| private static class AnyTaskBuilder extends TaskRunnerWorkItem |
There was a problem hiding this comment.
This class is not a builder. Please change to a more appropriate name.
There was a problem hiding this comment.
yeah, i was hesitant to name it this way, i added one method withTaskState which is kind of builder.
There was a problem hiding this comment.
It's only a builder if it has a build() method, or something like that, that creates an object of a different type. This is more like a normal non-builder class.
IMO, given that this is not really a builder class, the withTaskState method is misleadingly named. Generally, on a non-builder class, you'd expect withX to make a copy with X changed. This is more of a setter, since it mutates the object. So a better name would be setTaskState. (It can still return this, if you want.)
Or, you could design the class to be immutable. In that case, make the field final, call the method withTaskState, and have it return a copy with a different state.
My personal preference would be to do an immutable class, because immutable classes are easier to reason about (you know they won't mutate on you after you get one).
| task.getType(), | ||
| SettableFuture.create(), | ||
| task.getDataSource(), | ||
| TaskState.PENDING |
There was a problem hiding this comment.
Why is the state fixed to pending?
There was a problem hiding this comment.
Initially, the state is set to PENDING as a placeholder for all active tasks, later in the method, we filter out the RUNNING and WAITING tasks from this list, so the remaining tasks would be actually in PENDING state.
| final List<Task> allActiveTasks = taskStorageQueryAdapter.getActiveTasks(); | ||
| final List<TaskRunnerWorkItem> allTasks = Lists.newArrayList(); | ||
| for (final Task task : allActiveTasks) { | ||
| allTasks.add(new AnyTaskBuilder( |
There was a problem hiding this comment.
Druid's convention is
allTasks.add(
new AnyTaskBuilder(
task.getId(),
task.getType(),
SettableFuture.create(),
task.getDataSource(),
TaskState.PENDING
)
);| ) | ||
| { | ||
|
|
||
| if (state != null) { |
There was a problem hiding this comment.
Is it possible to structure this instead like the following? It would make things clearer, I think, and eliminate the extra code for the "null state" case.
List<Something> tasksIAmGoingToReturn;
if (state == null || "pending".equals("state")) {
thePendingTasks = someCodeThatGetsPendingTasks();
tasksIAmGoingToReturn.addAll(thePendingTasks);
}
if (state == null || "waiting".equals("state")) {
theWaitingTasks = someCodeThatGetsWaitingTasks();
tasksIAmGoingToReturn.addAll(theWaitingTasks);
}
if (state == null || "running".equals("state")) {
theRunningTasks = someCodeThatGetsRunningTasks();
tasksIAmGoingToReturn.addAll(theRunningTasks);
}
if (state == null || "complete".equals("state")) {
theCompleteTasks = someCodeThatGetsCompleteTasks();
tasksIAmGoingToReturn.addAll(theCompleteTasks);
}
return tasksIAmGoingToReturn;
In this world, presumably the existing /runningTasks endpoint could be implemented like:
// call the same function as above
theRunningTasks = someCodeThatGetsRunningTasks();
return theRunningTasks;
| public enum TaskState | ||
| { | ||
| RUNNING, | ||
| WAITING, |
There was a problem hiding this comment.
IMO, TaskState is not the right place to add these things. That's because TaskState's purpose is to model task status as stored in the database (see TaskStorage.insert and TaskStorage.setStatus), where there is no concept of "waiting" or "pending" (those are both stored as "running"). Splitting "running" into waiting/pending/really-really-running is more of a task runner concept.
So I think a new class would make more sense, geared towards modeling the runtime state of a task. Maybe RuntimeTaskState. And that one shouldn't be persisted into the database.
Please write javadocs that make the distinction clear, too.
| } | ||
|
|
||
| private static class WaitingTask extends TaskRunnerWorkItem | ||
| private static class AnyTaskBuilder extends TaskRunnerWorkItem |
There was a problem hiding this comment.
It's only a builder if it has a build() method, or something like that, that creates an object of a different type. This is more like a normal non-builder class.
IMO, given that this is not really a builder class, the withTaskState method is misleadingly named. Generally, on a non-builder class, you'd expect withX to make a copy with X changed. This is more of a setter, since it mutates the object. So a better name would be setTaskState. (It can still return this, if you want.)
Or, you could design the class to be immutable. In that case, make the field final, call the method withTaskState, and have it return a copy with a different state.
My personal preference would be to do an immutable class, because immutable classes are easier to reason about (you know they won't mutate on you after you get one).
| { | ||
| Function<TaskStatusPlus, Iterable<ResourceAction>> raGenerator = taskStatusPlus -> { | ||
| final String taskId = taskStatusPlus.getId(); | ||
| final Optional<Task> optionalTask = taskStorageQueryAdapter.getTask(taskId); |
There was a problem hiding this comment.
This is still doing taskStorageQueryAdapter.getTask(taskId) for every single task. Let's try to structure the code to avoid this somehow. Possibly by retrieving dataSource info upfront for all relevant tasks, in one SQL call, and then caching it.
| final List<TaskStatusPlus> completeTasks = Lists.newArrayList(Iterables.transform( | ||
| recentlyFinishedTasks, | ||
| status -> { | ||
| final Pair<DateTime, String> pair = taskStorageQueryAdapter.getCreatedDateAndDataSource(status.getId()); |
There was a problem hiding this comment.
This is still doing taskStorageQueryAdapter.getCreatedDateAndDataSource for every single complete task. Let's try to structure the code to avoid this somehow. Possibly by adding a method to taskStorageQueryAdapter that returns this info for all complete tasks at once.
Added new class to capture task info for caching Other refactoring
so it can be accessed within server And other related classes like TaskState and TaskStatusPlus are in api
This is to mainly capture the task type from payload
This reverts commit 2a527a0.
* Add RunnerTaskState
* Fix the createdTime and queueInsertionTime for running/active tasks * Clean up tests
| WAITING, | ||
| PENDING, | ||
| RUNNING, | ||
| NONE; // is used for a completed task |
There was a problem hiding this comment.
Might "NONE" also be used for tasks that the runner doesn't know about yet? This could be the case during bootstrapping, i.e. in the time between when a overlord becomes the new leader and when its runner has actually been fed all the tasks that need to run.
There was a problem hiding this comment.
The tasks which are not in TaskRunner.getKnownTasks and not COMPLETE are marked in WAITING state. I am using this NONE state for completed tasks to keep common datatype for all returned tasks.
| */ | ||
| public class TaskInfo | ||
| { | ||
| private String id; |
There was a problem hiding this comment.
These fields should be final, because in general, any fields that don't need to be modified should be final. It helps improve code readability (no need to wonder whether the field values change over time) and reduces bugs (avoids accidentally assuming fields are constant when they might actually be modified).
If all fields are final, then instances of the class become immutable, which are nice for a bunch of reasons: https://hackernoon.com/5-benefits-of-immutable-objects-worth-considering-for-your-next-project-f98e7e85b6ac
| this.dataSource = datasource; | ||
| } | ||
|
|
||
| private TaskInfo() |
There was a problem hiding this comment.
This constructor could be removed if you implement the other suggestions in the review.
| return dataSource; | ||
| } | ||
|
|
||
| public static class TaskInfoBuilder |
There was a problem hiding this comment.
IMO, this builder isn't necessary, since TaskInfo instances seem to always be created with all four parameters provided up front. Builders are usually worth it when you have a lot of optional parameters, since they avoid the need to have constructors with huge numbers of nulls in them. But that is not the case here, so I'd suggest removing this builder.
As a side note, if this builder did continue to exist then I'd have a couple of comments about it:
- The
build()method should use the 4-arg TaskInfo constructor, which would allow TaskInfo to be an immutable class. - The
build(String id, DateTime createdTime, TaskState state, String datasource)method shouldn't exist: it isn't really builder-like. It's more of a static factory method (the third major way to create objects; the three being constructors, builders, and static factory methods).
| List<StatusType> getInactiveStatusesSince(DateTime timestamp, @Nullable Integer maxNumStatuses); | ||
|
|
||
| /** | ||
| * Return up to {@code maxNumStatuses} {@link TaskInfo} objects for all inactive entries created on or later than the given timestamp |
There was a problem hiding this comment.
This line is a bit too long (we generally use 120 chars max).
| Function<TaskStatusPlus, Iterable<ResourceAction>> raGenerator = taskStatusPlus -> { | ||
| final String taskId = taskStatusPlus.getId(); | ||
| //final Optional<Task> optionalTask = taskStorageQueryAdapter.getTask(taskId); | ||
| TaskInfo taskInfo = taskInfoCache.get(taskId); |
There was a problem hiding this comment.
Based on the comments above for taskInfoCache, you'll probably want to rethink how this works.
Maybe it would make sense to pass in a map of task id -> datasource (presumably freshly retrieved) to this securedTaskStatusPlus method?
| } | ||
| TaskRunner runner = taskRunnerOpt.get(); | ||
| Collection<? extends TaskRunnerWorkItem> runnersKnownTasks = runner.getKnownTasks(); | ||
| List<String> runnerKnownTaskIds = runnersKnownTasks.stream() |
There was a problem hiding this comment.
Since you're calling contains on this a lot, it should be a Set. contains on Lists is generally O(n), whereas contains on Sets is generally O(log n) or O(1).
|
|
||
| if (RunnerTaskState.PENDING.equals(state)) { | ||
| Collection<? extends TaskRunnerWorkItem> knownPendingTasks = runner.getPendingTasks(); | ||
| List<String> pendingTaskIds = knownPendingTasks |
|
|
||
| if (RunnerTaskState.RUNNING.equals(state)) { | ||
| Collection<? extends TaskRunnerWorkItem> knownRunningTasks = runner.getRunningTasks(); | ||
| List<String> runningTaskIds = knownRunningTasks |
| ); | ||
| } | ||
| TaskRunner runner = taskRunnerOpt.get(); | ||
| Collection<? extends TaskRunnerWorkItem> runnersKnownTasks = runner.getKnownTasks(); |
There was a problem hiding this comment.
It looks like these variables are only needed when the state is RunnerTaskState.WAITING, and could be moved inside that block.
Fix a bug in securedTaskStatusPlus
| ); | ||
| } | ||
|
|
||
| class TaskInfoMapper implements ResultSetMapper |
There was a problem hiding this comment.
ResultSetMapper can take a type parameter, let's change this to:
class TaskInfoMapper implements ResultSetMapper<TaskInfo<EntryType>>
| @Override | ||
| public TaskInfo<EntryType> map(int index, ResultSet resultSet, StatementContext context) throws SQLException | ||
| { | ||
| final TaskInfo taskInfo; |
There was a problem hiding this comment.
Can you add the EntryType type parameter to TaskInfo here?
| final TaskInfo taskInfo; | ||
| try { | ||
| TaskStatus status = getJsonMapper().readValue(resultSet.getBytes("status_payload"), getStatusType()); | ||
| taskInfo = new TaskInfo( |
There was a problem hiding this comment.
this can be taskInfo = new TaskInfo<>( after adding the EntryType parameter to TaskInfo above
| return storage.getActiveTaskInfo(); | ||
| } | ||
|
|
||
| public List<TaskStatus> getRecentlyFinishedTaskStatuses( |
There was a problem hiding this comment.
It looks like getRecentlyFinishedTaskStatuses could be removed, it's only called in tests and getRecentlyCompletedTaskInfo can provide a superset of that information
There was a problem hiding this comment.
i thought i am missing some unused code, yes removed it and tests don't need it really, so removed from tests as well.
| ); | ||
| }; | ||
|
|
||
| List<TaskStatusPlus> dataSourceFilter = collectionToFilter; |
There was a problem hiding this comment.
For the case where the caller provides a datasource, I think getTasks could be optimized in a couple of ways:
- We could offload filtering of task by datasource to the metadata store, this would cut down on the amount of data being transferred to and processed on the overlord and also allow users to set up indices for filtering by datasource on the task table if needed
- If we do that, we could additionally reduce the number of authorization checks needed from (number of tasks) to one (just check that that user has access to the datasource being filtered on once)
There was a problem hiding this comment.
Do you mean passing in the datasource as argument to getActiveTaskInfo or getRecentlyCompletedTaskInfo and then filtering during the query ?
There was a problem hiding this comment.
yes, we could add datasource as an argument to those methods to put in a "WHERE datasource = x" clause to the SQL queries being made
There was a problem hiding this comment.
ok, added WHERE datasource = :ds clause to all sql implementations
|
@surekhasaharan Left some initial comments, still reviewing |
Change TaskInfoMapper signature to add generic type
| if (!optionalTask.isPresent()) { | ||
| Function<TaskStatusPlus, Iterable<ResourceAction>> raGenerator = taskStatusPlus -> { | ||
| final String taskId = taskStatusPlus.getId(); | ||
| String taskDatasource = taskInfoCache.get(taskId); |
There was a problem hiding this comment.
I think we can remove the taskInfoCache, since it's purely local to getTasks now and it's only read from here (we can get the datasource from taskStatusPlus as well)
There was a problem hiding this comment.
right, it can be removed.
| " id, " | ||
| + " status_payload " | ||
| + " status_payload, " | ||
| + " created_date, " |
There was a problem hiding this comment.
nit: spacing on the new fields like created_date should align with status_payload
| } | ||
|
|
||
| private List<AnyTask> filterActiveTasks( | ||
| HttpServletRequest req, |
There was a problem hiding this comment.
req is unused here, let's remove it
| List<TaskInfo<Task>> allActiveTaskInfo | ||
| ) | ||
| { | ||
| final List<AnyTask> allTasks = Lists.newArrayList(); |
There was a problem hiding this comment.
Let's move allTasks and the for loop below to getTasks, so we only create this list once
There was a problem hiding this comment.
good point, moved inside getTasks
* Passed datasource as argument to be used in sql query * Other minor fixes
jon-wei
left a comment
There was a problem hiding this comment.
left some minor comments
| maxTaskStatuses | ||
| ) | ||
| DateTimes.nowUtc().minus(duration == null ? config.getRecentlyFinishedThreshold() : duration), | ||
| maxTaskStatuses, datasource |
There was a problem hiding this comment.
nit: let's put datasource on a separate line
| for (TaskInfo taskInfo : taskInfoList) { | ||
| taskInfoCache.putIfAbsent(taskInfo.getId(), taskInfo.getDataSource()); | ||
| } | ||
| maxCompletedTasks, duration, dataSource); |
There was a problem hiding this comment.
nit: can you move ); to a separate line
| + " %s " | ||
| + "WHERE " | ||
| + " active = FALSE AND created_date >= :start " | ||
| + whereStmt |
There was a problem hiding this comment.
nit: whereStmt can be replaced with geWhereClause(datasource), we can get rid of the extra variable there
| return query; | ||
| } | ||
|
|
||
| private String getWhereClause(@Nullable String datasource) |
There was a problem hiding this comment.
Suggest renaming getWhereClause to getWhereClauseForInactiveStatusesSinceQuery so it's clear what it's used for
*Some minor changes, rename method, spacing changes
| }; | ||
|
|
||
| List<TaskStatusPlus> dataSourceFilter = collectionToFilter; | ||
| if (dataSource != null) { |
There was a problem hiding this comment.
this block can be removed now since the datasource filtering is handled by the metadata store now
you can also have a more optimized authorization check path for the datasource != null case now, where you don't need to run the auth check on all the tasks returned (since they'll all have the same datasource), and just check that the user has access to the datasource once
* Add max limit to getRecentlyFinishedTaskInfo in HeapMemoryTaskStorage * Add TaskLocation to Anytask object
|
|
||
| //check for valid state | ||
| if (state != null) { | ||
| List validStates = ImmutableList.of("pending", "waiting", "running", "complete"); |
There was a problem hiding this comment.
It'd be more hygienic if this were a constant, like
public static final API_TASK_STATES = ImmutableList.of("pending", "waiting", "running", "complete")| List validStates = ImmutableList.of("pending", "waiting", "running", "complete"); | ||
| if (!validStates.contains(StringUtils.toLowerCase(state))) { | ||
| return Response.status(Status.BAD_REQUEST) | ||
| .entity("Bad state query param passed : " + state).entity(ImmutableList.of()).build(); |
There was a problem hiding this comment.
I don't think you can do entity twice. I think the last one wins if you do this, so you should get rid of the second entity. And while doing that, you might as well make the message a little nicer, something like:
String.format("Invalid 'state', valid values are: " + API_TASK_STATES)There was a problem hiding this comment.
yeah, missed that there was entity twice here.
| new Resource(dataSource, ResourceType.DATASOURCE), | ||
| Action.READ | ||
| ); | ||
| Access authResult = AuthorizationUtils.authorizeResourceAction( |
There was a problem hiding this comment.
It's a little weird to mix final and non-final variables here. Either one is ok (we don't have a strong style guide for whether variables like this should be final or not) but you should at least be consistent within a method.
If it was me writing it then I would do final for both :)
| Optional<TaskRunner> taskRunnerOpt = taskMaster.getTaskRunner(); | ||
| if (!taskRunnerOpt.isPresent()) { | ||
| throw new WebApplicationException( | ||
| Response.serverError().entity("No task runner found ").build() |
|
@surekhasaharan from the test, might this be a real error? |
* Add the new tasks api in overlordResource It takes 4 optional query params * state(pending/running/waiting/compelte) * dataSource * interval (applies to completed tasks) * maxCompletedTasks (applies to completed tasks) If all params are null, the api returns all the tasks * Add the state to each task returned by tasks endpoint * divide active tasks into waiting, pending or running * Add more unit tests * Add UNKNOWN state to TaskState * Fix the authorization calls * WIP: PR comments Added new class to capture task info for caching Other refactoring * Refactoring : move TaskStatus class to druid-api so it can be accessed within server And other related classes like TaskState and TaskStatusPlus are in api * Remove unused class and apis accessing it * Add a separate cache for recently completed tasks This is to mainly capture the task type from payload * Ignore a test * Add a RuntimeTaskState to encompass all states a task can be in * Revert "Add a RuntimeTaskState to encompass all states a task can be in" This reverts commit 2a527a0. * Fix wrong api call * Fix and unignore tests * Remove waiting,pending state from TaskState * Add RunnerTaskState * Missed the annotation runnerStatusCode * Fix the creationTime * Fix the createdTime and queueInsertionTime for running/active tasks * Clean up tests * Add javadocs * Potentially fix the teamcity build * Address PR comments *Get rid of TaskInfoBuilder *Make TaskInfoMapper static nested class *Other changes * fix import in MaterializedViewSupervisor after merge * Address PR comments on * Replace global cache with local map * combine multiple queries into one * Removed unused code * Fix unit tests Fix a bug in securedTaskStatusPlus * Remove getRecentlyFinishedTaskStatuses method Change TaskInfoMapper signature to add generic type * Address PR comments * Passed datasource as argument to be used in sql query * Other minor fixes * Address PR comments *Some minor changes, rename method, spacing changes * Add early auth check if datasource is not null * Fix test case * Add max limit to getRecentlyFinishedTaskInfo in HeapMemoryTaskStorage * Add TaskLocation to Anytask object * Address PR comments * Fix a bug in test case causing ClassCastException
* Add the new tasks api in overlordResource It takes 4 optional query params * state(pending/running/waiting/compelte) * dataSource * interval (applies to completed tasks) * maxCompletedTasks (applies to completed tasks) If all params are null, the api returns all the tasks * Add the state to each task returned by tasks endpoint * divide active tasks into waiting, pending or running * Add more unit tests * Add UNKNOWN state to TaskState * Fix the authorization calls * WIP: PR comments Added new class to capture task info for caching Other refactoring * Refactoring : move TaskStatus class to druid-api so it can be accessed within server And other related classes like TaskState and TaskStatusPlus are in api * Remove unused class and apis accessing it * Add a separate cache for recently completed tasks This is to mainly capture the task type from payload * Ignore a test * Add a RuntimeTaskState to encompass all states a task can be in * Revert "Add a RuntimeTaskState to encompass all states a task can be in" This reverts commit 2a527a0. * Fix wrong api call * Fix and unignore tests * Remove waiting,pending state from TaskState * Add RunnerTaskState * Missed the annotation runnerStatusCode * Fix the creationTime * Fix the createdTime and queueInsertionTime for running/active tasks * Clean up tests * Add javadocs * Potentially fix the teamcity build * Address PR comments *Get rid of TaskInfoBuilder *Make TaskInfoMapper static nested class *Other changes * fix import in MaterializedViewSupervisor after merge * Address PR comments on * Replace global cache with local map * combine multiple queries into one * Removed unused code * Fix unit tests Fix a bug in securedTaskStatusPlus * Remove getRecentlyFinishedTaskStatuses method Change TaskInfoMapper signature to add generic type * Address PR comments * Passed datasource as argument to be used in sql query * Other minor fixes * Address PR comments *Some minor changes, rename method, spacing changes * Add early auth check if datasource is not null * Fix test case * Add max limit to getRecentlyFinishedTaskInfo in HeapMemoryTaskStorage * Add TaskLocation to Anytask object * Address PR comments * Fix a bug in test case causing ClassCastException
Add the new
tasksapi in overlordResource