Redact passwords from tasks fetched from the TaskQueue

indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskMaster.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.indexing.overlord;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Optional;
 import com.google.inject.Inject;
 import org.apache.druid.client.indexing.IndexingService;
@@ -94,7 +95,8 @@ public TaskMaster(
       final SupervisorManager supervisorManager,
       final OverlordDutyExecutor overlordDutyExecutor,
       @IndexingService final DruidLeaderSelector overlordLeaderSelector,
-      final SegmentAllocationQueue segmentAllocationQueue
+      final SegmentAllocationQueue segmentAllocationQueue,
+      final ObjectMapper mapper
   )
   {
     this.supervisorManager = supervisorManager;
@@ -125,7 +127,8 @@ public void becomeLeader()
               taskRunner,
               taskActionClientFactory,
               taskLockbox,
-              emitter
+              emitter,
+              mapper
           );
 
           // Sensible order to start stuff:

indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java

@@ -19,6 +19,8 @@
 
 package org.apache.druid.indexing.overlord;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
@@ -56,6 +58,8 @@
 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.PasswordProvider;
+import org.apache.druid.metadata.PasswordProviderRedactionMixIn;
 import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
 import org.apache.druid.utils.CollectionUtils;
 
@@ -117,6 +121,7 @@ public class TaskQueue
   private final TaskActionClientFactory taskActionClientFactory;
   private final TaskLockbox taskLockbox;
   private final ServiceEmitter emitter;
+  private final ObjectMapper passwordRedactingMapper;
 
   private final ReentrantLock giant = new ReentrantLock(true);
   @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
@@ -160,7 +165,8 @@ public TaskQueue(
       TaskRunner taskRunner,
       TaskActionClientFactory taskActionClientFactory,
       TaskLockbox taskLockbox,
-      ServiceEmitter emitter
+      ServiceEmitter emitter,
+      ObjectMapper mapper
   )
   {
     this.lockConfig = Preconditions.checkNotNull(lockConfig, "lockConfig");
@@ -175,6 +181,8 @@ public TaskQueue(
         config.getTaskCompleteHandlerNumThreads(),
         "TaskQueue-OnComplete-%d"
     );
+    this.passwordRedactingMapper = mapper.copy()
+                                         .addMixIn(PasswordProvider.class, PasswordProviderRedactionMixIn.class);
   }
 
   @VisibleForTesting
@@ -970,15 +978,34 @@ public CoordinatorRunStats getQueueStats()
     return stats;
   }
 
+  /**
+   * Returns an optional containing the task payload after successfully redacting credentials.
+   * Returns an absent optional if there is no task payload corresponding to the taskId in memory.
+   * Throws DruidException if password could not be redacted due to serialization / deserialization failure
+   */
   public Optional<Task> getActiveTask(String id)
   {
+    Task task;
     giant.lock();
     try {
-      return Optional.fromNullable(tasks.get(id));
+      task = tasks.get(id);
     }
     finally {
       giant.unlock();
     }
+    if (task != null) {
+      try {
+        // Write and read the value using a mapper with password redaction mixin.
+        task = passwordRedactingMapper.readValue(passwordRedactingMapper.writeValueAsString(task), Task.class);
+      }
+      catch (JsonProcessingException e) {
+        log.error(e, "Failed to serialize or deserialize task with id [%s].", task.getId());
+        throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+                            .ofCategory(DruidException.Category.RUNTIME_FAILURE)
+                            .build(e, "Failed to serialize or deserialize task[%s].", task.getId());
+      }
+    }
+    return Optional.fromNullable(task);
   }
 
   @VisibleForTesting

indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java

@@ -142,7 +142,8 @@ public void setup()
         taskRunner,
         taskActionClientFactory,
         getLockbox(),
-        new NoopServiceEmitter()
+        new NoopServiceEmitter(),
+        getObjectMapper()
     );
     runningTasks.clear();
     taskQueue.start();

indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java

@@ -696,7 +696,7 @@ private TaskQueue setUpTaskQueue(TaskStorage ts, TaskRunner tr) throws Exception
         TaskQueueConfig.class
     );
 
-    return new TaskQueue(lockConfig, tqc, new DefaultTaskConfig(), ts, tr, tac, taskLockbox, emitter);
+    return new TaskQueue(lockConfig, tqc, new DefaultTaskConfig(), ts, tr, tac, taskLockbox, emitter, mapper);
   }
 
   @After

indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLockConfigTest.java

@@ -30,6 +30,7 @@
 import org.apache.druid.indexing.overlord.config.TaskLockConfig;
 import org.apache.druid.indexing.overlord.config.TaskQueueConfig;
 import org.apache.druid.indexing.test.TestIndexerMetadataStorageCoordinator;
+import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.server.metrics.NoopServiceEmitter;
 import org.easymock.EasyMock;
@@ -120,7 +121,8 @@ public boolean isForceTimeChunkLock()
         taskRunner,
         actionClientFactory,
         lockbox,
-        emitter
+        emitter,
+        new DefaultObjectMapper()
     );
   }
 }

indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java

@@ -122,7 +122,8 @@ public <RetType> RetType submit(TaskAction<RetType> taskAction)
         taskRunner,
         unsupportedTaskActionFactory, // Not used for anything serious
         new TaskLockbox(taskStorage, storageCoordinator),
-        new NoopServiceEmitter()
+        new NoopServiceEmitter(),
+        jsonMapper
     );
 
     taskQueue.start();