| /* |
| * 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.indexing.overlord; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Optional; |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.Sets; |
| import com.google.common.util.concurrent.FutureCallback; |
| import com.google.common.util.concurrent.Futures; |
| import com.google.common.util.concurrent.ListenableFuture; |
| import com.google.common.util.concurrent.ThreadFactoryBuilder; |
| import com.google.errorprone.annotations.concurrent.GuardedBy; |
| import org.apache.druid.annotations.SuppressFBWarnings; |
| import org.apache.druid.indexer.TaskLocation; |
| import org.apache.druid.indexer.TaskStatus; |
| import org.apache.druid.indexing.common.Counters; |
| import org.apache.druid.indexing.common.actions.TaskActionClientFactory; |
| import org.apache.druid.indexing.common.task.IndexTaskUtils; |
| import org.apache.druid.indexing.common.task.Task; |
| import org.apache.druid.indexing.common.task.Tasks; |
| import org.apache.druid.indexing.common.task.batch.MaxAllowedLocksExceededException; |
| import org.apache.druid.indexing.common.task.batch.parallel.SinglePhaseParallelIndexTaskRunner; |
| import org.apache.druid.indexing.overlord.config.DefaultTaskConfig; |
| import org.apache.druid.indexing.overlord.config.TaskLockConfig; |
| import org.apache.druid.indexing.overlord.config.TaskQueueConfig; |
| import org.apache.druid.java.util.common.ISE; |
| import org.apache.druid.java.util.common.StringUtils; |
| import org.apache.druid.java.util.common.concurrent.ScheduledExecutors; |
| import org.apache.druid.java.util.common.lifecycle.LifecycleStart; |
| import org.apache.druid.java.util.common.lifecycle.LifecycleStop; |
| import org.apache.druid.java.util.emitter.EmittingLogger; |
| import org.apache.druid.java.util.emitter.service.ServiceEmitter; |
| import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; |
| import org.apache.druid.metadata.EntryExistsException; |
| import org.apache.druid.utils.CollectionUtils; |
| |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.ArrayBlockingQueue; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.locks.ReentrantLock; |
| import java.util.stream.Collectors; |
| |
| /** |
| * Interface between task producers and the task runner. |
| * <p/> |
| * This object accepts tasks from producers using {@link #add} and manages delivery of these tasks to a |
| * {@link TaskRunner}. Tasks will run in a mostly-FIFO order, with deviations when the natural next task is not ready |
| * in time (based on its {@link Task#isReady} method). |
| * <p/> |
| * For persistence, we save all new tasks and task status changes using a {@link TaskStorage} object. |
| */ |
| public class TaskQueue |
| { |
| private static final long MANAGEMENT_WAIT_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(60); |
| private static final long MIN_WAIT_TIME_MS = 100; |
| |
| // Task ID -> Task, for tasks that are active in some way (submitted, running, or finished and to-be-cleaned-up). |
| @GuardedBy("giant") |
| private final LinkedHashMap<String, Task> tasks = new LinkedHashMap<>(); |
| |
| // Task ID -> Future from the TaskRunner |
| @GuardedBy("giant") |
| private final Map<String, ListenableFuture<TaskStatus>> taskFutures = new HashMap<>(); |
| |
| // Tasks that are in the process of being cleaned up by notifyStatus. Prevents manageInternal from re-launching them. |
| @GuardedBy("giant") |
| private final Set<String> recentlyCompletedTasks = new HashSet<>(); |
| |
| private final TaskLockConfig lockConfig; |
| private final TaskQueueConfig config; |
| private final DefaultTaskConfig defaultTaskConfig; |
| private final TaskStorage taskStorage; |
| private final TaskRunner taskRunner; |
| private final TaskActionClientFactory taskActionClientFactory; |
| private final TaskLockbox taskLockbox; |
| private final ServiceEmitter emitter; |
| |
| private final ReentrantLock giant = new ReentrantLock(true); |
| @SuppressWarnings("MismatchedQueryAndUpdateOfCollection") |
| private final BlockingQueue<Object> managementMayBeNecessary = new ArrayBlockingQueue<>(8); |
| private final ExecutorService managerExec = Executors.newSingleThreadExecutor( |
| new ThreadFactoryBuilder() |
| .setDaemon(false) |
| .setNameFormat("TaskQueue-Manager").build() |
| ); |
| private final ScheduledExecutorService storageSyncExec = Executors.newSingleThreadScheduledExecutor( |
| new ThreadFactoryBuilder() |
| .setDaemon(false) |
| .setNameFormat("TaskQueue-StorageSync").build() |
| ); |
| |
| private volatile boolean active = false; |
| |
| private static final EmittingLogger log = new EmittingLogger(TaskQueue.class); |
| |
| private final ConcurrentHashMap<String, AtomicLong> totalSuccessfulTaskCount = new ConcurrentHashMap<>(); |
| private final ConcurrentHashMap<String, AtomicLong> totalFailedTaskCount = new ConcurrentHashMap<>(); |
| @GuardedBy("totalSuccessfulTaskCount") |
| private Map<String, Long> prevTotalSuccessfulTaskCount = new HashMap<>(); |
| @GuardedBy("totalFailedTaskCount") |
| private Map<String, Long> prevTotalFailedTaskCount = new HashMap<>(); |
| |
| public TaskQueue( |
| TaskLockConfig lockConfig, |
| TaskQueueConfig config, |
| DefaultTaskConfig defaultTaskConfig, |
| TaskStorage taskStorage, |
| TaskRunner taskRunner, |
| TaskActionClientFactory taskActionClientFactory, |
| TaskLockbox taskLockbox, |
| ServiceEmitter emitter |
| ) |
| { |
| this.lockConfig = Preconditions.checkNotNull(lockConfig, "lockConfig"); |
| this.config = Preconditions.checkNotNull(config, "config"); |
| this.defaultTaskConfig = Preconditions.checkNotNull(defaultTaskConfig, "defaultTaskContextConfig"); |
| this.taskStorage = Preconditions.checkNotNull(taskStorage, "taskStorage"); |
| this.taskRunner = Preconditions.checkNotNull(taskRunner, "taskRunner"); |
| this.taskActionClientFactory = Preconditions.checkNotNull(taskActionClientFactory, "taskActionClientFactory"); |
| this.taskLockbox = Preconditions.checkNotNull(taskLockbox, "taskLockbox"); |
| this.emitter = Preconditions.checkNotNull(emitter, "emitter"); |
| } |
| |
| @VisibleForTesting |
| void setActive(boolean active) |
| { |
| this.active = active; |
| } |
| |
| /** |
| * Starts this task queue. Allows {@link #add(Task)} to accept new tasks. |
| */ |
| @LifecycleStart |
| public void start() |
| { |
| giant.lock(); |
| |
| try { |
| Preconditions.checkState(!active, "queue must be stopped"); |
| active = true; |
| syncFromStorage(); |
| managerExec.submit( |
| new Runnable() |
| { |
| @Override |
| public void run() |
| { |
| while (true) { |
| try { |
| manage(); |
| break; |
| } |
| catch (InterruptedException e) { |
| log.info("Interrupted, exiting!"); |
| break; |
| } |
| catch (Exception e) { |
| final long restartDelay = config.getRestartDelay().getMillis(); |
| log.makeAlert(e, "Failed to manage").addData("restartDelay", restartDelay).emit(); |
| try { |
| Thread.sleep(restartDelay); |
| } |
| catch (InterruptedException e2) { |
| log.info("Interrupted, exiting!"); |
| break; |
| } |
| } |
| } |
| } |
| } |
| ); |
| ScheduledExecutors.scheduleAtFixedRate( |
| storageSyncExec, |
| config.getStorageSyncRate(), |
| new Callable<ScheduledExecutors.Signal>() |
| { |
| @Override |
| public ScheduledExecutors.Signal call() |
| { |
| try { |
| syncFromStorage(); |
| } |
| catch (Exception e) { |
| if (active) { |
| log.makeAlert(e, "Failed to sync with storage").emit(); |
| } |
| } |
| if (active) { |
| return ScheduledExecutors.Signal.REPEAT; |
| } else { |
| return ScheduledExecutors.Signal.STOP; |
| } |
| } |
| } |
| ); |
| requestManagement(); |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| /** |
| * Shuts down the queue. |
| */ |
| @LifecycleStop |
| public void stop() |
| { |
| giant.lock(); |
| |
| try { |
| tasks.clear(); |
| taskFutures.clear(); |
| active = false; |
| managerExec.shutdownNow(); |
| storageSyncExec.shutdownNow(); |
| requestManagement(); |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| public boolean isActive() |
| { |
| return active; |
| } |
| |
| /** |
| * Request management from the management thread. Non-blocking. |
| * |
| * Other callers (such as notifyStatus) should trigger activity on the |
| * TaskQueue thread by requesting management here. |
| */ |
| void requestManagement() |
| { |
| // use a BlockingQueue since the offer/poll/wait behaviour is simple |
| // and very easy to reason about |
| |
| // the request has to be offer (non blocking), since someone might request |
| // while already holding giant lock |
| |
| // do not care if the item fits into the queue: |
| // if the queue is already full, request has been triggered anyway |
| managementMayBeNecessary.offer(this); |
| } |
| |
| /** |
| * Await for an event to manage. |
| * |
| * This should only be called from the management thread to wait for activity. |
| * |
| * @param nanos |
| * @throws InterruptedException |
| */ |
| @SuppressFBWarnings(value = "RV_RETURN_VALUE_IGNORED", justification = "using queue as notification mechanism, result has no value") |
| void awaitManagementNanos(long nanos) throws InterruptedException |
| { |
| // mitigate a busy loop, it can get pretty busy when there are a lot of start/stops |
| try { |
| Thread.sleep(MIN_WAIT_TIME_MS); |
| } |
| catch (InterruptedException e) { |
| throw new RuntimeException(e); |
| } |
| |
| // wait for an item, if an item arrives (or is already available), complete immediately |
| // (does not actually matter what the item is) |
| managementMayBeNecessary.poll(nanos - (TimeUnit.MILLISECONDS.toNanos(MIN_WAIT_TIME_MS)), TimeUnit.NANOSECONDS); |
| |
| // there may have been multiple requests, clear them all |
| managementMayBeNecessary.clear(); |
| } |
| |
| /** |
| * Main task runner management loop. Meant to run forever, or, at least until we're stopped. |
| */ |
| private void manage() throws InterruptedException |
| { |
| log.info("Beginning management in %s.", config.getStartDelay()); |
| Thread.sleep(config.getStartDelay().getMillis()); |
| |
| // Ignore return value- we'll get the IDs and futures from getKnownTasks later. |
| taskRunner.restore(); |
| |
| while (active) { |
| manageInternal(); |
| |
| // awaitNanos because management may become necessary without this condition signalling, |
| // due to e.g. tasks becoming ready when other folks mess with the TaskLockbox. |
| awaitManagementNanos(MANAGEMENT_WAIT_TIMEOUT_NANOS); |
| } |
| } |
| |
| @VisibleForTesting |
| void manageInternal() |
| { |
| Set<String> knownTaskIds = new HashSet<>(); |
| Map<String, ListenableFuture<TaskStatus>> runnerTaskFutures = new HashMap<>(); |
| |
| giant.lock(); |
| |
| try { |
| manageInternalCritical(knownTaskIds, runnerTaskFutures); |
| } |
| finally { |
| giant.unlock(); |
| } |
| |
| manageInternalPostCritical(knownTaskIds, runnerTaskFutures); |
| } |
| |
| |
| /** |
| * Management loop critical section tasks. |
| * |
| * @param knownTaskIds will be modified - filled with known task IDs |
| * @param runnerTaskFutures will be modified - filled with futures related to getting the running tasks |
| */ |
| @GuardedBy("giant") |
| private void manageInternalCritical( |
| final Set<String> knownTaskIds, |
| final Map<String, ListenableFuture<TaskStatus>> runnerTaskFutures |
| ) |
| { |
| // Task futures available from the taskRunner |
| for (final TaskRunnerWorkItem workItem : taskRunner.getKnownTasks()) { |
| if (!recentlyCompletedTasks.contains(workItem.getTaskId())) { |
| // Don't do anything with tasks that have recently finished; notifyStatus will handle it. |
| runnerTaskFutures.put(workItem.getTaskId(), workItem.getResult()); |
| } |
| } |
| // Attain futures for all active tasks (assuming they are ready to run). |
| // Copy tasks list, as notifyStatus may modify it. |
| for (final Task task : ImmutableList.copyOf(tasks.values())) { |
| if (recentlyCompletedTasks.contains(task.getId())) { |
| // Don't do anything with tasks that have recently finished; notifyStatus will handle it. |
| continue; |
| } |
| |
| knownTaskIds.add(task.getId()); |
| |
| if (!taskFutures.containsKey(task.getId())) { |
| final ListenableFuture<TaskStatus> runnerTaskFuture; |
| if (runnerTaskFutures.containsKey(task.getId())) { |
| runnerTaskFuture = runnerTaskFutures.get(task.getId()); |
| } else { |
| // Task should be running, so run it. |
| final boolean taskIsReady; |
| try { |
| taskIsReady = task.isReady(taskActionClientFactory.create(task)); |
| } |
| catch (Exception e) { |
| log.warn(e, "Exception thrown during isReady for task: %s", task.getId()); |
| final String errorMessage; |
| if (e instanceof MaxAllowedLocksExceededException) { |
| errorMessage = e.getMessage(); |
| } else { |
| errorMessage = "Failed while waiting for the task to be ready to run. " |
| + "See overlord logs for more details."; |
| } |
| notifyStatus(task, TaskStatus.failure(task.getId(), errorMessage), errorMessage); |
| continue; |
| } |
| if (taskIsReady) { |
| log.info("Asking taskRunner to run: %s", task.getId()); |
| runnerTaskFuture = taskRunner.run(task); |
| } else { |
| // Task.isReady() can internally lock intervals or segments. |
| // We should release them if the task is not ready. |
| taskLockbox.unlockAll(task); |
| continue; |
| } |
| } |
| taskFutures.put(task.getId(), attachCallbacks(task, runnerTaskFuture)); |
| } else if (isTaskPending(task)) { |
| // if the taskFutures contain this task and this task is pending, also let the taskRunner |
| // to run it to guarantee it will be assigned to run |
| // see https://github.com/apache/druid/pull/6991 |
| taskRunner.run(task); |
| } |
| } |
| } |
| |
| @VisibleForTesting |
| private void manageInternalPostCritical( |
| final Set<String> knownTaskIds, |
| final Map<String, ListenableFuture<TaskStatus>> runnerTaskFutures |
| ) |
| { |
| // Kill tasks that shouldn't be running |
| final Set<String> tasksToKill = Sets.difference(runnerTaskFutures.keySet(), knownTaskIds); |
| if (!tasksToKill.isEmpty()) { |
| log.info("Asking taskRunner to clean up %,d tasks.", tasksToKill.size()); |
| |
| // On large installations running several thousands of tasks, |
| // concatenating the list of known task ids can be compupationally expensive. |
| final boolean logKnownTaskIds = log.isDebugEnabled(); |
| final String reason = logKnownTaskIds |
| ? StringUtils.format("Task is not in knownTaskIds[%s]", knownTaskIds) |
| : "Task is not in knownTaskIds"; |
| |
| for (final String taskId : tasksToKill) { |
| try { |
| taskRunner.shutdown(taskId, reason); |
| } |
| catch (Exception e) { |
| log.warn(e, "TaskRunner failed to clean up task: %s", taskId); |
| } |
| } |
| } |
| } |
| |
| private boolean isTaskPending(Task task) |
| { |
| return taskRunner.getPendingTasks() |
| .stream() |
| .anyMatch(workItem -> workItem.getTaskId().equals(task.getId())); |
| } |
| |
| /** |
| * Adds some work to the queue and the underlying task storage facility with a generic "running" status. |
| * |
| * @param task task to add |
| * |
| * @return true |
| * |
| * @throws EntryExistsException if the task already exists |
| */ |
| public boolean add(final Task task) throws EntryExistsException |
| { |
| if (taskStorage.getTask(task.getId()).isPresent()) { |
| throw new EntryExistsException(StringUtils.format("Task %s already exists", task.getId())); |
| } |
| |
| // Set forceTimeChunkLock before adding task spec to taskStorage, so that we can see always consistent task spec. |
| task.addToContextIfAbsent(Tasks.FORCE_TIME_CHUNK_LOCK_KEY, lockConfig.isForceTimeChunkLock()); |
| defaultTaskConfig.getContext().forEach(task::addToContextIfAbsent); |
| // Every task shuold use the lineage-based segment allocation protocol unless it is explicitly set to |
| // using the legacy protocol. |
| task.addToContextIfAbsent( |
| SinglePhaseParallelIndexTaskRunner.CTX_USE_LINEAGE_BASED_SEGMENT_ALLOCATION_KEY, |
| SinglePhaseParallelIndexTaskRunner.DEFAULT_USE_LINEAGE_BASED_SEGMENT_ALLOCATION |
| ); |
| |
| giant.lock(); |
| |
| try { |
| Preconditions.checkState(active, "Queue is not active!"); |
| Preconditions.checkNotNull(task, "task"); |
| Preconditions.checkState(tasks.size() < config.getMaxSize(), "Too many tasks (max = %,d)", config.getMaxSize()); |
| |
| // If this throws with any sort of exception, including TaskExistsException, we don't want to |
| // insert the task into our queue. So don't catch it. |
| taskStorage.insert(task, TaskStatus.running(task.getId())); |
| addTaskInternal(task); |
| requestManagement(); |
| return true; |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| @GuardedBy("giant") |
| private void addTaskInternal(final Task task) |
| { |
| final Task existingTask = tasks.putIfAbsent(task.getId(), task); |
| |
| if (existingTask == null) { |
| taskLockbox.add(task); |
| } else if (!existingTask.equals(task)) { |
| throw new ISE("Cannot add task ID [%s] with same ID as task that has already been added", task.getId()); |
| } |
| } |
| |
| /** |
| * Removes a task from {@link #tasks} and {@link #taskLockbox}, if it exists. Returns whether the task was |
| * removed or not. |
| */ |
| @GuardedBy("giant") |
| private boolean removeTaskInternal(final String taskId) |
| { |
| final Task task = tasks.remove(taskId); |
| if (task != null) { |
| taskLockbox.remove(task); |
| return true; |
| } else { |
| return false; |
| } |
| } |
| |
| /** |
| * Shuts down a task if it has not yet finished. |
| * |
| * @param taskId task to kill |
| * @param reasonFormat A format string indicating the shutdown reason |
| * @param args arguments for reasonFormat |
| */ |
| public void shutdown(final String taskId, String reasonFormat, Object... args) |
| { |
| giant.lock(); |
| |
| try { |
| final Task task = tasks.get(Preconditions.checkNotNull(taskId, "taskId")); |
| if (task != null) { |
| notifyStatus(task, TaskStatus.failure(taskId, StringUtils.format(reasonFormat, args)), reasonFormat, args); |
| } |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| /** |
| * Shuts down a task, but records the task status as a success, unike {@link #shutdown(String, String, Object...)} |
| * |
| * @param taskId task to shutdown |
| * @param reasonFormat A format string indicating the shutdown reason |
| * @param args arguments for reasonFormat |
| */ |
| public void shutdownWithSuccess(final String taskId, String reasonFormat, Object... args) |
| { |
| giant.lock(); |
| |
| try { |
| final Task task = tasks.get(Preconditions.checkNotNull(taskId, "taskId")); |
| if (task != null) { |
| notifyStatus(task, TaskStatus.success(taskId), reasonFormat, args); |
| } |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| /** |
| * Notify this queue that some task has an updated status. If this update is valid, the status will be persisted in |
| * the task storage facility. If the status is a completed status, the task will be unlocked and no further |
| * updates will be accepted. |
| * |
| * @param task task to update |
| * @param taskStatus new task status |
| * |
| * @throws NullPointerException if task or status is null |
| * @throws IllegalArgumentException if the task ID does not match the status ID |
| * @throws IllegalStateException if this queue is currently shut down |
| */ |
| private void notifyStatus(final Task task, final TaskStatus taskStatus, String reasonFormat, Object... args) |
| { |
| Preconditions.checkNotNull(task, "task"); |
| Preconditions.checkNotNull(taskStatus, "status"); |
| Preconditions.checkState(active, "Queue is not active!"); |
| Preconditions.checkArgument( |
| task.getId().equals(taskStatus.getId()), |
| "Mismatching task ids[%s/%s]", |
| task.getId(), |
| taskStatus.getId() |
| ); |
| |
| if (!taskStatus.isComplete()) { |
| // Nothing to do for incomplete statuses. |
| return; |
| } |
| |
| // Critical section: add this task to recentlyCompletedTasks, so it isn't managed while being cleaned up. |
| giant.lock(); |
| try { |
| recentlyCompletedTasks.add(task.getId()); |
| } |
| finally { |
| giant.unlock(); |
| } |
| |
| final TaskLocation taskLocation = taskRunner.getTaskLocation(task.getId()); |
| |
| // Save status to metadata store first, so if we crash while doing the rest of the shutdown, our successor |
| // remembers that this task has completed. |
| try { |
| final Optional<TaskStatus> previousStatus = taskStorage.getStatus(task.getId()); |
| if (!previousStatus.isPresent() || !previousStatus.get().isRunnable()) { |
| log.makeAlert("Ignoring notification for already-complete task").addData("task", task.getId()).emit(); |
| } else { |
| taskStorage.setStatus(taskStatus.withLocation(taskLocation)); |
| } |
| } |
| catch (Throwable e) { |
| // If persist fails, even after the retries performed in taskStorage, then metadata store and actual cluster |
| // state have diverged. Send out an alert and continue with the task shutdown routine. |
| log.makeAlert(e, "Failed to persist status for task") |
| .addData("task", task.getId()) |
| .addData("statusCode", taskStatus.getStatusCode()) |
| .emit(); |
| } |
| |
| // Inform taskRunner that this task can be shut down. |
| try { |
| taskRunner.shutdown(task.getId(), reasonFormat, args); |
| } |
| catch (Throwable e) { |
| // If task runner shutdown fails, continue with the task shutdown routine. We'll come back and try to |
| // shut it down again later in manageInternalPostCritical, once it's removed from the "tasks" map. |
| log.warn(e, "TaskRunner failed to cleanup task after completion: %s", task.getId()); |
| } |
| |
| // Critical section: remove this task from all of our tracking data structures. |
| giant.lock(); |
| try { |
| if (removeTaskInternal(task.getId())) { |
| taskFutures.remove(task.getId()); |
| } else { |
| log.warn("Unknown task completed: %s", task.getId()); |
| } |
| |
| recentlyCompletedTasks.remove(task.getId()); |
| requestManagement(); |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| /** |
| * Attach success and failure handlers to a task status future, such that when it completes, we perform the |
| * appropriate updates. |
| * |
| * @param statusFuture a task status future |
| * |
| * @return the same future, for convenience |
| */ |
| private ListenableFuture<TaskStatus> attachCallbacks(final Task task, final ListenableFuture<TaskStatus> statusFuture) |
| { |
| final ServiceMetricEvent.Builder metricBuilder = new ServiceMetricEvent.Builder(); |
| IndexTaskUtils.setTaskDimensions(metricBuilder, task); |
| |
| Futures.addCallback( |
| statusFuture, |
| new FutureCallback<TaskStatus>() |
| { |
| @Override |
| public void onSuccess(final TaskStatus status) |
| { |
| log.info("Received %s status for task: %s", status.getStatusCode(), status.getId()); |
| handleStatus(status); |
| } |
| |
| @Override |
| public void onFailure(final Throwable t) |
| { |
| log.makeAlert(t, "Failed to run task") |
| .addData("task", task.getId()) |
| .addData("type", task.getType()) |
| .addData("dataSource", task.getDataSource()) |
| .emit(); |
| handleStatus( |
| TaskStatus.failure(task.getId(), "Failed to run this task. See overlord logs for more details.") |
| ); |
| } |
| |
| private void handleStatus(final TaskStatus status) |
| { |
| try { |
| // If we're not supposed to be running anymore, don't do anything. Somewhat racey if the flag gets set |
| // after we check and before we commit the database transaction, but better than nothing. |
| if (!active) { |
| log.info("Abandoning task due to shutdown: %s", task.getId()); |
| return; |
| } |
| |
| notifyStatus(task, status, "notified status change from task"); |
| |
| // Emit event and log, if the task is done |
| if (status.isComplete()) { |
| IndexTaskUtils.setTaskStatusDimensions(metricBuilder, status); |
| emitter.emit(metricBuilder.build("task/run/time", status.getDuration())); |
| |
| log.info( |
| "Task %s: %s (%d run duration)", |
| status.getStatusCode(), |
| task.getId(), |
| status.getDuration() |
| ); |
| |
| if (status.isSuccess()) { |
| Counters.incrementAndGetLong(totalSuccessfulTaskCount, task.getDataSource()); |
| } else { |
| Counters.incrementAndGetLong(totalFailedTaskCount, task.getDataSource()); |
| } |
| } |
| } |
| catch (Exception e) { |
| log.makeAlert(e, "Failed to handle task status") |
| .addData("task", task.getId()) |
| .addData("statusCode", status.getStatusCode()) |
| .emit(); |
| } |
| } |
| } |
| ); |
| return statusFuture; |
| } |
| |
| /** |
| * Resync the contents of this task queue with our storage facility. Useful to make sure our in-memory state |
| * corresponds to the storage facility even if the latter is manually modified. |
| */ |
| private void syncFromStorage() |
| { |
| giant.lock(); |
| |
| try { |
| if (active) { |
| final Map<String, Task> newTasks = toTaskIDMap(taskStorage.getActiveTasks()); |
| final int tasksSynced = newTasks.size(); |
| final Map<String, Task> oldTasks = new HashMap<>(tasks); |
| |
| // Calculate differences on IDs instead of Task Objects. |
| Set<String> commonIds = Sets.newHashSet(Sets.intersection(newTasks.keySet(), oldTasks.keySet())); |
| for (String taskID : commonIds) { |
| newTasks.remove(taskID); |
| oldTasks.remove(taskID); |
| } |
| Collection<Task> addedTasks = newTasks.values(); |
| Collection<Task> removedTasks = oldTasks.values(); |
| |
| // Clean up removed Tasks |
| for (Task task : removedTasks) { |
| removeTaskInternal(task.getId()); |
| } |
| |
| // Add newly Added tasks to the queue |
| for (Task task : addedTasks) { |
| addTaskInternal(task); |
| } |
| |
| log.info( |
| "Synced %d tasks from storage (%d tasks added, %d tasks removed).", |
| tasksSynced, |
| addedTasks.size(), |
| removedTasks.size() |
| ); |
| requestManagement(); |
| } else { |
| log.info("Not active. Skipping storage sync."); |
| } |
| } |
| catch (Exception e) { |
| log.warn(e, "Failed to sync tasks from storage!"); |
| throw new RuntimeException(e); |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| private static Map<String, Task> toTaskIDMap(List<Task> taskList) |
| { |
| Map<String, Task> rv = new HashMap<>(); |
| for (Task task : taskList) { |
| rv.put(task.getId(), task); |
| } |
| return rv; |
| } |
| |
| private Map<String, Long> getDeltaValues(Map<String, Long> total, Map<String, Long> prev) |
| { |
| return total.entrySet() |
| .stream() |
| .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() - prev.getOrDefault(e.getKey(), 0L))); |
| } |
| |
| public Map<String, Long> getSuccessfulTaskCount() |
| { |
| Map<String, Long> total = CollectionUtils.mapValues(totalSuccessfulTaskCount, AtomicLong::get); |
| synchronized (totalSuccessfulTaskCount) { |
| Map<String, Long> delta = getDeltaValues(total, prevTotalSuccessfulTaskCount); |
| prevTotalSuccessfulTaskCount = total; |
| return delta; |
| } |
| } |
| |
| public Map<String, Long> getFailedTaskCount() |
| { |
| Map<String, Long> total = CollectionUtils.mapValues(totalFailedTaskCount, AtomicLong::get); |
| synchronized (totalFailedTaskCount) { |
| Map<String, Long> delta = getDeltaValues(total, prevTotalFailedTaskCount); |
| prevTotalFailedTaskCount = total; |
| return delta; |
| } |
| } |
| |
| Map<String, String> getCurrentTaskDatasources() |
| { |
| giant.lock(); |
| try { |
| return tasks.values().stream().collect(Collectors.toMap(Task::getId, Task::getDataSource)); |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| public Map<String, Long> getRunningTaskCount() |
| { |
| Map<String, String> taskDatasources = getCurrentTaskDatasources(); |
| return taskRunner.getRunningTasks() |
| .stream() |
| .collect(Collectors.toMap( |
| e -> taskDatasources.getOrDefault(e.getTaskId(), ""), |
| e -> 1L, |
| Long::sum |
| )); |
| } |
| |
| public Map<String, Long> getPendingTaskCount() |
| { |
| Map<String, String> taskDatasources = getCurrentTaskDatasources(); |
| return taskRunner.getPendingTasks() |
| .stream() |
| .collect(Collectors.toMap( |
| e -> taskDatasources.getOrDefault(e.getTaskId(), ""), |
| e -> 1L, |
| Long::sum |
| )); |
| } |
| |
| public Map<String, Long> getWaitingTaskCount() |
| { |
| Set<String> runnerKnownTaskIds = taskRunner.getKnownTasks() |
| .stream() |
| .map(TaskRunnerWorkItem::getTaskId) |
| .collect(Collectors.toSet()); |
| |
| giant.lock(); |
| try { |
| return tasks.values().stream().filter(task -> !runnerKnownTaskIds.contains(task.getId())) |
| .collect(Collectors.toMap(Task::getDataSource, task -> 1L, Long::sum)); |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| |
| @VisibleForTesting |
| List<Task> getTasks() |
| { |
| giant.lock(); |
| try { |
| return new ArrayList<>(tasks.values()); |
| } |
| finally { |
| giant.unlock(); |
| } |
| } |
| } |
