From 0cb3df506b382b9c5edb5b180c4635eea091ba38 Mon Sep 17 00:00:00 2001 From: Amatya Date: Thu, 21 Mar 2024 17:18:46 +0530 Subject: [PATCH 1/4] Redact passwords from tasks fetched from the TaskQueue --- .../druid/indexing/overlord/TaskQueue.java | 33 +++++++++++++++++-- .../overlord/TaskStorageQueryAdapter.java | 12 +++++-- 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java index 6dbf6e707989..b50c6952fd48 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java @@ -19,6 +19,9 @@ package org.apache.druid.indexing.overlord; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.json.JsonMapper; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Optional; import com.google.common.base.Preconditions; @@ -56,6 +59,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; @@ -152,6 +157,8 @@ public class TaskQueue private final AtomicInteger statusUpdatesInQueue = new AtomicInteger(); private final AtomicInteger handledStatusUpdates = new AtomicInteger(); + private final ObjectMapper jsonMapper; + public TaskQueue( TaskLockConfig lockConfig, TaskQueueConfig config, @@ -175,6 +182,11 @@ public TaskQueue( config.getTaskCompleteHandlerNumThreads(), "TaskQueue-OnComplete-%d" ); + // Add mixin to redact passwords. + this.jsonMapper = (new JsonMapper()).addMixIn( + PasswordProvider.class, + PasswordProviderRedactionMixIn.class + ); } @VisibleForTesting @@ -970,15 +982,32 @@ public CoordinatorRunStats getQueueStats() return stats; } - public Optional getActiveTask(String id) + /** + * 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 JsonProcessingException if password could not be redacted due to serialization / deserialization failure + */ + public Optional getActiveTask(String id) throws JsonProcessingException { + 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 = jsonMapper.readValue(jsonMapper.writeValueAsString(task), Task.class); + } + catch (JsonProcessingException e) { + log.error(e, "Failed to serialize or deserialize task with id [%s].", task.getId()); + throw e; + } + } + return Optional.fromNullable(task); } @VisibleForTesting diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java index ba2ca3c7066a..1378c0c6eb39 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java @@ -19,6 +19,7 @@ package org.apache.druid.indexing.overlord; +import com.fasterxml.jackson.core.JsonProcessingException; import com.google.common.base.Optional; import com.google.inject.Inject; import org.apache.druid.indexer.TaskInfo; @@ -107,9 +108,14 @@ public List getTaskStatusPlusList( public Optional getTask(final String taskid) { if (taskQueue.isPresent()) { - Optional activeTask = taskQueue.get().getActiveTask(taskid); - if (activeTask.isPresent()) { - return activeTask; + try { + Optional activeTask = taskQueue.get().getActiveTask(taskid); + if (activeTask.isPresent()) { + return activeTask; + } + } + catch (JsonProcessingException e) { + // do nothing } } return storage.getTask(taskid); From 9c50006aa8c7c378eeb66600e3b2e91cccb2163c Mon Sep 17 00:00:00 2001 From: Amatya Date: Fri, 22 Mar 2024 17:09:38 +0530 Subject: [PATCH 2/4] Inject mapper and add test --- .../druid/indexing/overlord/TaskMaster.java | 7 +- .../druid/indexing/overlord/TaskQueue.java | 23 ++- .../overlord/TaskStorageQueryAdapter.java | 4 +- .../ConcurrentReplaceAndAppendTest.java | 3 +- .../indexing/overlord/TaskLifecycleTest.java | 2 +- .../indexing/overlord/TaskLockConfigTest.java | 4 +- .../indexing/overlord/TaskQueueScaleTest.java | 3 +- .../indexing/overlord/TaskQueueTest.java | 142 ++++++++++++++++-- .../indexing/overlord/http/OverlordTest.java | 4 +- 9 files changed, 160 insertions(+), 32 deletions(-) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskMaster.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskMaster.java index 4798513aafd4..9829e1ffa19f 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskMaster.java +++ b/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: diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java index b50c6952fd48..dd09e8484cd4 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java @@ -21,7 +21,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.json.JsonMapper; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Optional; import com.google.common.base.Preconditions; @@ -122,6 +121,7 @@ public class TaskQueue private final TaskActionClientFactory taskActionClientFactory; private final TaskLockbox taskLockbox; private final ServiceEmitter emitter; + private final ObjectMapper PASSWORD_REDACTING_MAPPER; private final ReentrantLock giant = new ReentrantLock(true); @SuppressWarnings("MismatchedQueryAndUpdateOfCollection") @@ -157,7 +157,6 @@ public class TaskQueue private final AtomicInteger statusUpdatesInQueue = new AtomicInteger(); private final AtomicInteger handledStatusUpdates = new AtomicInteger(); - private final ObjectMapper jsonMapper; public TaskQueue( TaskLockConfig lockConfig, @@ -167,7 +166,8 @@ public TaskQueue( TaskRunner taskRunner, TaskActionClientFactory taskActionClientFactory, TaskLockbox taskLockbox, - ServiceEmitter emitter + ServiceEmitter emitter, + ObjectMapper mapper ) { this.lockConfig = Preconditions.checkNotNull(lockConfig, "lockConfig"); @@ -182,11 +182,8 @@ public TaskQueue( config.getTaskCompleteHandlerNumThreads(), "TaskQueue-OnComplete-%d" ); - // Add mixin to redact passwords. - this.jsonMapper = (new JsonMapper()).addMixIn( - PasswordProvider.class, - PasswordProviderRedactionMixIn.class - ); + this.PASSWORD_REDACTING_MAPPER = mapper.copy() + .addMixIn(PasswordProvider.class, PasswordProviderRedactionMixIn.class); } @VisibleForTesting @@ -987,7 +984,7 @@ public CoordinatorRunStats getQueueStats() * Returns an absent optional if there is no task payload corresponding to the taskId in memory. * Throws JsonProcessingException if password could not be redacted due to serialization / deserialization failure */ - public Optional getActiveTask(String id) throws JsonProcessingException + public Optional getActiveTask(String id) { Task task; giant.lock(); @@ -1000,11 +997,13 @@ public Optional getActiveTask(String id) throws JsonProcessingException if (task != null) { try { // Write and read the value using a mapper with password redaction mixin. - task = jsonMapper.readValue(jsonMapper.writeValueAsString(task), Task.class); + task = PASSWORD_REDACTING_MAPPER.readValue(PASSWORD_REDACTING_MAPPER.writeValueAsString(task), Task.class); } catch (JsonProcessingException e) { - log.error(e, "Failed to serialize or deserialize task with id [%s].", task.getId()); - throw e; + log.error(e, "Failed to serialize or deserialize task with id [%s].", task.getId()); + throw DruidException.forPersona(DruidException.Persona.USER) + .ofCategory(DruidException.Category.RUNTIME_FAILURE) + .build(e, "Failed to serialize or deserialize task."); } } return Optional.fromNullable(task); diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java index 1378c0c6eb39..f5627ff8c492 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java @@ -19,9 +19,9 @@ package org.apache.druid.indexing.overlord; -import com.fasterxml.jackson.core.JsonProcessingException; import com.google.common.base.Optional; import com.google.inject.Inject; +import org.apache.druid.error.DruidException; import org.apache.druid.indexer.TaskInfo; import org.apache.druid.indexer.TaskStatus; import org.apache.druid.indexer.TaskStatusPlus; @@ -114,7 +114,7 @@ public Optional getTask(final String taskid) return activeTask; } } - catch (JsonProcessingException e) { + catch (DruidException e) { // do nothing } } diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java index 91c7d6a71755..864d84dbfa17 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java +++ b/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(); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java index 852ea0df02dc..b60bc10a8de3 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java +++ b/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 diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLockConfigTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLockConfigTest.java index 7d2fbcd79230..1c90802f4801 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLockConfigTest.java +++ b/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() ); } } diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java index ade1314884e7..1d478d9cd149 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java @@ -122,7 +122,8 @@ public RetType submit(TaskAction taskAction) taskRunner, unsupportedTaskActionFactory, // Not used for anything serious new TaskLockbox(taskStorage, storageCoordinator), - new NoopServiceEmitter() + new NoopServiceEmitter(), + jsonMapper ); taskQueue.start(); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java index 8ee341c5db16..886c9bee8f44 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java @@ -19,6 +19,9 @@ package org.apache.druid.indexing.overlord; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.InjectableValues; +import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.base.Optional; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; @@ -26,6 +29,11 @@ import com.google.common.util.concurrent.ListenableFuture; import org.apache.curator.framework.CuratorFramework; import org.apache.druid.common.guava.DSuppliers; +import org.apache.druid.data.input.impl.DimensionsSpec; +import org.apache.druid.data.input.impl.HttpInputSource; +import org.apache.druid.data.input.impl.HttpInputSourceConfig; +import org.apache.druid.data.input.impl.NoopInputFormat; +import org.apache.druid.data.input.impl.TimestampSpec; import org.apache.druid.discovery.DruidNodeDiscoveryProvider; import org.apache.druid.discovery.WorkerNodeService; import org.apache.druid.error.DruidException; @@ -38,10 +46,14 @@ import org.apache.druid.indexing.common.TaskToolbox; import org.apache.druid.indexing.common.actions.TaskActionClient; import org.apache.druid.indexing.common.actions.TaskActionClientFactory; +import org.apache.druid.indexing.common.config.TaskStorageConfig; import org.apache.druid.indexing.common.task.AbstractBatchIndexTask; import org.apache.druid.indexing.common.task.IngestionTestBase; import org.apache.druid.indexing.common.task.Task; import org.apache.druid.indexing.common.task.Tasks; +import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexIOConfig; +import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexIngestionSpec; +import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexSupervisorTask; import org.apache.druid.indexing.common.task.batch.parallel.SinglePhaseParallelIndexTaskRunner; import org.apache.druid.indexing.overlord.autoscaling.NoopProvisioningStrategy; import org.apache.druid.indexing.overlord.autoscaling.ScalingStats; @@ -53,9 +65,11 @@ import org.apache.druid.indexing.overlord.hrtr.HttpRemoteTaskRunnerTest; import org.apache.druid.indexing.overlord.hrtr.WorkerHolder; import org.apache.druid.indexing.overlord.setup.DefaultWorkerBehaviorConfig; +import org.apache.druid.indexing.test.TestIndexerMetadataStorageCoordinator; import org.apache.druid.indexing.worker.TaskAnnouncement; import org.apache.druid.indexing.worker.Worker; import org.apache.druid.indexing.worker.config.WorkerConfig; +import org.apache.druid.jackson.DefaultObjectMapper; import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.Pair; import org.apache.druid.java.util.common.StringUtils; @@ -64,7 +78,12 @@ import org.apache.druid.java.util.common.granularity.Granularity; import org.apache.druid.java.util.http.client.HttpClient; import org.apache.druid.java.util.metrics.StubServiceEmitter; +import org.apache.druid.metadata.DefaultPasswordProvider; +import org.apache.druid.metadata.DerbyMetadataStorageActionHandlerFactory; +import org.apache.druid.metadata.SQLMetadataConnector; import org.apache.druid.segment.TestHelper; +import org.apache.druid.segment.indexing.DataSchema; +import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec; import org.apache.druid.server.coordinator.stats.CoordinatorRunStats; import org.apache.druid.server.initialization.IndexerZkConfig; import org.apache.druid.server.initialization.ZkPathsConfig; @@ -77,6 +96,7 @@ import org.mockito.Mockito; import javax.annotation.Nullable; +import java.net.URI; import java.util.Collection; import java.util.Collections; import java.util.List; @@ -109,7 +129,8 @@ public void testManageInternalReleaseLockWhenTaskIsNotReady() throws Exception new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); // task1 emulates a case when there is a task that was issued before task2 and acquired locks conflicting @@ -154,7 +175,8 @@ public void testShutdownReleasesTaskLock() throws Exception new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); @@ -194,7 +216,8 @@ public void testTaskErrorWhenExceptionIsThrownDueToQueueSize() new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); @@ -219,7 +242,8 @@ public void testSetUseLineageBasedSegmentAllocationByDefault() new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); final Task task = new TestTask("t1", Intervals.of("2021-01-01/P1D")); @@ -254,7 +278,8 @@ public Map getContext() new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); final Task task = new TestTask("t1", Intervals.of("2021-01-01/P1D")); @@ -279,7 +304,8 @@ public void testUserProvidedTaskContextOverrideDefaultLineageBasedSegmentAllocat new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); final Task task = new TestTask( @@ -321,7 +347,8 @@ public Map getContext() new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); final Task task = new TestTask("t1", Intervals.of("2021-01-01/P1D")); @@ -344,7 +371,8 @@ public void testUserProvidedContextOverrideLockConfig() new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); final Task task = new TestTask( @@ -374,7 +402,8 @@ public void testTaskStatusWhenExceptionIsThrownInIsReady() new SimpleTaskRunner(actionClientFactory), actionClientFactory, getLockbox(), - new NoopServiceEmitter() + new NoopServiceEmitter(), + getObjectMapper() ); taskQueue.setActive(true); final Task task = new TestTask("t1", Intervals.of("2021-01-01/P1D")) @@ -421,7 +450,8 @@ public void testKilledTasksEmitRuntimeMetricWithHttpRemote() throws InterruptedE taskRunner, actionClientFactory, getLockbox(), - metricsVerifier + metricsVerifier, + getObjectMapper() ); taskQueue.setActive(true); final Task task = new TestTask( @@ -507,7 +537,8 @@ public void testGetTaskStatus() taskRunner, createActionClientFactory(), getLockbox(), - new StubServiceEmitter("druid/overlord", "testHost") + new StubServiceEmitter("druid/overlord", "testHost"), + getObjectMapper() ); taskQueue.setActive(true); @@ -519,6 +550,95 @@ public void testGetTaskStatus() Assert.assertEquals(TaskStatus.failure(failedTask, failedTask), taskQueue.getTaskStatus(failedTask).get()); } + @Test + public void testGetActiveTaskRedactsPassword() throws JsonProcessingException + { + final String password = "AbCd_1234"; + final ObjectMapper mapper = getObjectMapper(); + + final HttpInputSourceConfig httpInputSourceConfig = new HttpInputSourceConfig(Collections.singleton("http")); + mapper.setInjectableValues(new InjectableValues.Std() + .addValue(HttpInputSourceConfig.class, httpInputSourceConfig) + .addValue(ObjectMapper.class, new DefaultObjectMapper()) + ); + + final SQLMetadataConnector derbyConnector = derbyConnectorRule.getConnector(); + final TaskStorage taskStorage = new MetadataTaskStorage( + derbyConnector, + new TaskStorageConfig(null), + new DerbyMetadataStorageActionHandlerFactory( + derbyConnector, + derbyConnectorRule.metadataTablesConfigSupplier().get(), + mapper + ) + ); + + final TaskQueue taskQueue = new TaskQueue( + new TaskLockConfig(), + new TaskQueueConfig(null, null, null, null, null), + new DefaultTaskConfig(), + taskStorage, + EasyMock.createMock(HttpRemoteTaskRunner.class), + createActionClientFactory(), + new TaskLockbox(taskStorage, new TestIndexerMetadataStorageCoordinator()), + new StubServiceEmitter("druid/overlord", "testHost"), + mapper + ); + + final DataSchema dataSchema = new DataSchema( + "DS", + new TimestampSpec(null, null, null), + new DimensionsSpec(null), + null, + new UniformGranularitySpec(Granularities.YEAR, Granularities.DAY, null), + null + ); + final ParallelIndexIOConfig ioConfig = new ParallelIndexIOConfig( + null, + new HttpInputSource(Collections.singletonList(URI.create("http://host.org")), + "user", + new DefaultPasswordProvider(password), + null, + httpInputSourceConfig), + new NoopInputFormat(), + null, + null + ); + final ParallelIndexSupervisorTask taskWithPassword = new ParallelIndexSupervisorTask( + "taskWithPassword", + "taskWithPassword", + null, + new ParallelIndexIngestionSpec( + dataSchema, + ioConfig, + null + ), + null, + null, + false + ); + Assert.assertTrue(mapper.writeValueAsString(taskWithPassword).contains(password)); + + taskQueue.start(); + taskQueue.add(taskWithPassword); + + final Optional taskFromTaskStorage = taskStorage.getTask(taskWithPassword.getId()); + Assert.assertTrue(taskFromTaskStorage.isPresent()); + Assert.assertNotNull(taskFromTaskStorage.get()); + Assert.assertFalse(mapper.writeValueAsString(taskFromTaskStorage.get()).contains(password)); + + + final Optional taskFromTaskQueue = taskQueue.getActiveTask(taskWithPassword.getId()); + Assert.assertTrue(taskFromTaskQueue.isPresent()); + Assert.assertNotNull(taskFromTaskQueue.get()); + Assert.assertFalse(mapper.writeValueAsString(taskFromTaskQueue.get()).contains(password)); + + Assert.assertEquals( + mapper.writeValueAsString(taskFromTaskStorage.get()), + mapper.writeValueAsString(taskFromTaskQueue.get()) + ); + } + private HttpRemoteTaskRunner createHttpRemoteTaskRunner(List runningTasks) { HttpRemoteTaskRunnerTest.TestDruidNodeDiscovery druidNodeDiscovery = new HttpRemoteTaskRunnerTest.TestDruidNodeDiscovery(); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java index b0df15ce4ef5..066e4c6c58a0 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java @@ -65,6 +65,7 @@ import org.apache.druid.indexing.overlord.duty.OverlordDutyExecutor; import org.apache.druid.indexing.overlord.supervisor.SupervisorManager; import org.apache.druid.indexing.test.TestIndexerMetadataStorageCoordinator; +import org.apache.druid.jackson.DefaultObjectMapper; import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.Pair; import org.apache.druid.java.util.common.concurrent.Execs; @@ -247,7 +248,8 @@ public MockTaskRunner get() supervisorManager, EasyMock.createNiceMock(OverlordDutyExecutor.class), new TestDruidLeaderSelector(), - EasyMock.createNiceMock(SegmentAllocationQueue.class) + EasyMock.createNiceMock(SegmentAllocationQueue.class), + new DefaultObjectMapper() ); EmittingLogger.registerEmitter(serviceEmitter); } From 7dec19965bfce623608f5b8e932d29d8f9982977 Mon Sep 17 00:00:00 2001 From: Amatya Date: Fri, 22 Mar 2024 18:20:44 +0530 Subject: [PATCH 3/4] Address feedback --- .../druid/indexing/overlord/TaskQueue.java | 12 ++++++------ .../druid/indexing/overlord/TaskQueueTest.java | 18 ++++++++---------- 2 files changed, 14 insertions(+), 16 deletions(-) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java index dd09e8484cd4..df05a72df908 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java @@ -121,7 +121,7 @@ public class TaskQueue private final TaskActionClientFactory taskActionClientFactory; private final TaskLockbox taskLockbox; private final ServiceEmitter emitter; - private final ObjectMapper PASSWORD_REDACTING_MAPPER; + private final ObjectMapper passwordRedactingMapper; private final ReentrantLock giant = new ReentrantLock(true); @SuppressWarnings("MismatchedQueryAndUpdateOfCollection") @@ -182,8 +182,8 @@ public TaskQueue( config.getTaskCompleteHandlerNumThreads(), "TaskQueue-OnComplete-%d" ); - this.PASSWORD_REDACTING_MAPPER = mapper.copy() - .addMixIn(PasswordProvider.class, PasswordProviderRedactionMixIn.class); + this.passwordRedactingMapper = mapper.copy() + .addMixIn(PasswordProvider.class, PasswordProviderRedactionMixIn.class); } @VisibleForTesting @@ -997,13 +997,13 @@ public Optional getActiveTask(String id) if (task != null) { try { // Write and read the value using a mapper with password redaction mixin. - task = PASSWORD_REDACTING_MAPPER.readValue(PASSWORD_REDACTING_MAPPER.writeValueAsString(task), Task.class); + 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.USER) + throw DruidException.forPersona(DruidException.Persona.OPERATOR) .ofCategory(DruidException.Category.RUNTIME_FAILURE) - .build(e, "Failed to serialize or deserialize task."); + .build(e, "Failed to serialize or deserialize task[%s].", task.getId()); } } return Optional.fromNullable(task); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java index 886c9bee8f44..544de4c6cae0 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java @@ -622,20 +622,18 @@ public void testGetActiveTaskRedactsPassword() throws JsonProcessingException taskQueue.start(); taskQueue.add(taskWithPassword); - final Optional taskFromTaskStorage = taskStorage.getTask(taskWithPassword.getId()); - Assert.assertTrue(taskFromTaskStorage.isPresent()); - Assert.assertNotNull(taskFromTaskStorage.get()); - Assert.assertFalse(mapper.writeValueAsString(taskFromTaskStorage.get()).contains(password)); + final Optional taskInStorage = taskStorage.getTask(taskWithPassword.getId()); + Assert.assertTrue(taskInStorage.isPresent()); + Assert.assertFalse(mapper.writeValueAsString(taskInStorage.get()).contains(password)); - final Optional taskFromTaskQueue = taskQueue.getActiveTask(taskWithPassword.getId()); - Assert.assertTrue(taskFromTaskQueue.isPresent()); - Assert.assertNotNull(taskFromTaskQueue.get()); - Assert.assertFalse(mapper.writeValueAsString(taskFromTaskQueue.get()).contains(password)); + final Optional taskInQueue = taskQueue.getActiveTask(taskWithPassword.getId()); + Assert.assertTrue(taskInQueue.isPresent()); + Assert.assertFalse(mapper.writeValueAsString(taskInQueue.get()).contains(password)); Assert.assertEquals( - mapper.writeValueAsString(taskFromTaskStorage.get()), - mapper.writeValueAsString(taskFromTaskQueue.get()) + mapper.writeValueAsString(taskInStorage.get()), + mapper.writeValueAsString(taskInQueue.get()) ); } From b90d61726adbed03260c44ef2f7152c49d0023cd Mon Sep 17 00:00:00 2001 From: Amatya Date: Sat, 23 Mar 2024 11:17:20 +0530 Subject: [PATCH 4/4] Address feedback --- .../apache/druid/indexing/overlord/TaskQueue.java | 3 +-- .../indexing/overlord/TaskStorageQueryAdapter.java | 12 +++--------- .../druid/indexing/overlord/TaskQueueTest.java | 12 +++++------- 3 files changed, 9 insertions(+), 18 deletions(-) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java index df05a72df908..d3ca29d3abc4 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java @@ -157,7 +157,6 @@ public class TaskQueue private final AtomicInteger statusUpdatesInQueue = new AtomicInteger(); private final AtomicInteger handledStatusUpdates = new AtomicInteger(); - public TaskQueue( TaskLockConfig lockConfig, TaskQueueConfig config, @@ -982,7 +981,7 @@ public CoordinatorRunStats getQueueStats() /** * 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 JsonProcessingException if password could not be redacted due to serialization / deserialization failure + * Throws DruidException if password could not be redacted due to serialization / deserialization failure */ public Optional getActiveTask(String id) { diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java index f5627ff8c492..ba2ca3c7066a 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java @@ -21,7 +21,6 @@ import com.google.common.base.Optional; import com.google.inject.Inject; -import org.apache.druid.error.DruidException; import org.apache.druid.indexer.TaskInfo; import org.apache.druid.indexer.TaskStatus; import org.apache.druid.indexer.TaskStatusPlus; @@ -108,14 +107,9 @@ public List getTaskStatusPlusList( public Optional getTask(final String taskid) { if (taskQueue.isPresent()) { - try { - Optional activeTask = taskQueue.get().getActiveTask(taskid); - if (activeTask.isPresent()) { - return activeTask; - } - } - catch (DruidException e) { - // do nothing + Optional activeTask = taskQueue.get().getActiveTask(taskid); + if (activeTask.isPresent()) { + return activeTask; } } return storage.getTask(taskid); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java index 544de4c6cae0..6a45cbb4c79c 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java @@ -624,17 +624,15 @@ public void testGetActiveTaskRedactsPassword() throws JsonProcessingException final Optional taskInStorage = taskStorage.getTask(taskWithPassword.getId()); Assert.assertTrue(taskInStorage.isPresent()); - Assert.assertFalse(mapper.writeValueAsString(taskInStorage.get()).contains(password)); - + final String taskInStorageAsString = mapper.writeValueAsString(taskInStorage.get()); + Assert.assertFalse(taskInStorageAsString.contains(password)); final Optional taskInQueue = taskQueue.getActiveTask(taskWithPassword.getId()); Assert.assertTrue(taskInQueue.isPresent()); - Assert.assertFalse(mapper.writeValueAsString(taskInQueue.get()).contains(password)); + final String taskInQueueAsString = mapper.writeValueAsString(taskInQueue.get()); + Assert.assertFalse(taskInQueueAsString.contains(password)); - Assert.assertEquals( - mapper.writeValueAsString(taskInStorage.get()), - mapper.writeValueAsString(taskInQueue.get()) - ); + Assert.assertEquals(taskInStorageAsString, taskInQueueAsString); } private HttpRemoteTaskRunner createHttpRemoteTaskRunner(List runningTasks)