This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new e4bd0d2a3a9 Refactor: Move auth utility method to AuthorizationUtils
(#17886)
e4bd0d2a3a9 is described below
commit e4bd0d2a3a9dbc93344610c9731744d9922798c8
Author: Kashif Faraz <[email protected]>
AuthorDate: Mon Apr 7 17:09:50 2025 +0530
Refactor: Move auth utility method to AuthorizationUtils (#17886)
Changes
---------
- Replace method `IndexTaskUtils.datasourceAuthorizationCheck` with
`AuthorizationUtils.verifyUnrestrictedDatasourceAccess`
- Move the logic in `IndexTaskUtils.getSavedParseExceptionReports` to
`ParseExceptionReportHandler`
---
.../indexing/common/task/HadoopIndexTask.java | 7 ++-
.../druid/indexing/common/task/IndexTask.java | 24 ++++------
.../druid/indexing/common/task/IndexTaskUtils.java | 54 ----------------------
.../parallel/ParallelIndexSupervisorTask.java | 25 +++++-----
.../batch/parallel/PartialSegmentGenerateTask.java | 18 ++------
.../task/batch/parallel/SinglePhaseSubTask.java | 28 ++++-------
.../SeekableStreamIndexTaskRunner.java | 50 ++++++++------------
.../druid/indexing/common/task/IndexTaskTest.java | 1 +
.../ParallelIndexSupervisorTaskResourceTest.java | 1 +
.../SeekableStreamIndexTaskRunnerAuthTest.java | 11 +++--
.../segment/incremental/ParseExceptionHandler.java | 15 +++++-
.../incremental/ParseExceptionHandlerTest.java | 4 +-
.../druid/server/security/AuthorizationUtils.java | 44 ++++++++++--------
13 files changed, 108 insertions(+), 174 deletions(-)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java
index 7361759ed22..e894080a569 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java
@@ -63,6 +63,7 @@ import
org.apache.druid.segment.incremental.RowIngestionMeters;
import org.apache.druid.segment.realtime.ChatHandler;
import org.apache.druid.segment.realtime.ChatHandlerProvider;
import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthorizationUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.server.security.Resource;
import org.apache.druid.server.security.ResourceAction;
@@ -660,8 +661,7 @@ public class HadoopIndexTask extends HadoopTask implements
ChatHandler
@QueryParam("windows") List<Integer> windows
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
- Map<String, Object> returnMap = new HashMap<>();
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
Map<String, Object> totalsMap = new HashMap<>();
if (determinePartitionsStatsGetter != null) {
@@ -672,8 +672,7 @@ public class HadoopIndexTask extends HadoopTask implements
ChatHandler
totalsMap.put(RowIngestionMeters.BUILD_SEGMENTS,
buildSegmentsStatsGetter.getTotalMetrics());
}
- returnMap.put("totals", totalsMap);
- return Response.ok(returnMap).build();
+ return Response.ok(Map.of("totals", totalsMap)).build();
}
private TaskReport.ReportMap getTaskCompletionReports()
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
index 4ea7e9d3dee..cacf1c7f4f1 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
@@ -94,6 +94,7 @@ import
org.apache.druid.segment.realtime.appenderator.SegmentsAndCommitMetadata;
import
org.apache.druid.segment.realtime.appenderator.TransactionalSegmentPublisher;
import org.apache.druid.segment.writeout.SegmentWriteOutMediumFactory;
import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthorizationUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.server.security.Resource;
import org.apache.druid.server.security.ResourceAction;
@@ -101,7 +102,6 @@ import org.apache.druid.server.security.ResourceType;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.partition.HashBasedNumberedShardSpec;
import org.apache.druid.timeline.partition.NumberedShardSpec;
-import org.apache.druid.utils.CircularBuffer;
import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
import org.joda.time.Interval;
import org.joda.time.Period;
@@ -331,7 +331,7 @@ public class IndexTask extends AbstractBatchIndexTask
implements ChatHandler, Pe
@QueryParam("full") String full
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
return Response.ok(doGetUnparseableEvents(full != null)).build();
}
@@ -342,18 +342,14 @@ public class IndexTask extends AbstractBatchIndexTask
implements ChatHandler, Pe
if (addDeterminePartitionStatsToReport(isFullReport, ingestionState)) {
events.put(
RowIngestionMeters.DETERMINE_PARTITIONS,
- IndexTaskUtils.getReportListFromSavedParseExceptions(
-
determinePartitionsParseExceptionHandler.getSavedParseExceptionReports()
- )
+
determinePartitionsParseExceptionHandler.getSavedParseExceptionReports()
);
}
if (addBuildSegmentStatsToReport(isFullReport, ingestionState)) {
events.put(
RowIngestionMeters.BUILD_SEGMENTS,
- IndexTaskUtils.getReportListFromSavedParseExceptions(
-
buildSegmentsParseExceptionHandler.getSavedParseExceptionReports()
- )
+ buildSegmentsParseExceptionHandler.getSavedParseExceptionReports()
);
}
return events;
@@ -400,7 +396,7 @@ public class IndexTask extends AbstractBatchIndexTask
implements ChatHandler, Pe
@QueryParam("full") String full
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
return Response.ok(doGetRowStats(full != null)).build();
}
@@ -412,7 +408,7 @@ public class IndexTask extends AbstractBatchIndexTask
implements ChatHandler, Pe
@QueryParam("full") String full
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final TaskReport.ReportMap liveReports = buildLiveIngestionStatsReport(
ingestionState,
@@ -547,19 +543,19 @@ public class IndexTask extends AbstractBatchIndexTask
implements ChatHandler, Pe
protected Map<String, Object> getTaskCompletionUnparseableEvents()
{
Map<String, Object> unparseableEventsMap = new HashMap<>();
- CircularBuffer<ParseExceptionReport>
determinePartitionsParseExceptionReports =
+ List<ParseExceptionReport> determinePartitionsParseExceptionReports =
determinePartitionsParseExceptionHandler.getSavedParseExceptionReports();
- CircularBuffer<ParseExceptionReport> buildSegmentsParseExceptionReports =
+ List<ParseExceptionReport> buildSegmentsParseExceptionReports =
buildSegmentsParseExceptionHandler.getSavedParseExceptionReports();
if (determinePartitionsParseExceptionReports != null ||
buildSegmentsParseExceptionReports != null) {
unparseableEventsMap.put(
RowIngestionMeters.DETERMINE_PARTITIONS,
-
IndexTaskUtils.getReportListFromSavedParseExceptions(determinePartitionsParseExceptionReports)
+ determinePartitionsParseExceptionReports
);
unparseableEventsMap.put(
RowIngestionMeters.BUILD_SEGMENTS,
-
IndexTaskUtils.getReportListFromSavedParseExceptions(buildSegmentsParseExceptionReports)
+ buildSegmentsParseExceptionReports
);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
index 1d0f2622ce3..3108c31ce83 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
@@ -26,66 +26,12 @@ import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.emitter.service.SegmentMetadataEvent;
import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
import org.apache.druid.query.DruidMetrics;
-import org.apache.druid.segment.incremental.ParseExceptionReport;
-import org.apache.druid.server.security.Action;
-import org.apache.druid.server.security.AuthorizationResult;
-import org.apache.druid.server.security.AuthorizationUtils;
-import org.apache.druid.server.security.AuthorizerMapper;
-import org.apache.druid.server.security.ForbiddenException;
-import org.apache.druid.server.security.Resource;
-import org.apache.druid.server.security.ResourceAction;
-import org.apache.druid.server.security.ResourceType;
import org.apache.druid.timeline.DataSegment;
-import org.apache.druid.utils.CircularBuffer;
-import javax.annotation.Nullable;
-import javax.servlet.http.HttpServletRequest;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
public class IndexTaskUtils
{
- @Nullable
- public static List<ParseExceptionReport>
getReportListFromSavedParseExceptions(
- CircularBuffer<ParseExceptionReport> savedParseExceptionReports
- )
- {
- if (savedParseExceptionReports == null) {
- return null;
- }
- List<ParseExceptionReport> reports = new ArrayList<>();
- for (int i = 0; i < savedParseExceptionReports.size(); i++) {
- reports.add(savedParseExceptionReports.getLatest(i));
- }
-
- return reports;
- }
-
- /**
- * Authorizes action to be performed on a task's datasource
- *
- * @return authorization result
- */
- public static AuthorizationResult datasourceAuthorizationCheck(
- final HttpServletRequest req,
- Action action,
- String datasource,
- AuthorizerMapper authorizerMapper
- )
- {
- ResourceAction resourceAction = new ResourceAction(
- new Resource(datasource, ResourceType.DATASOURCE),
- action
- );
-
- AuthorizationResult authResult =
AuthorizationUtils.authorizeResourceAction(req, resourceAction,
authorizerMapper);
- if (!authResult.allowAccessWithNoRestriction()) {
- throw new ForbiddenException(authResult.getErrorMessage());
- }
- return authResult;
- }
-
public static void setTaskDimensions(final ServiceMetricEvent.Builder
metricBuilder, final Task task)
{
metricBuilder.setDimension(DruidMetrics.TASK_ID, task.getId());
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
index 34cc71a3a12..50378db7d69 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
@@ -80,6 +80,7 @@ import org.apache.druid.segment.realtime.ChatHandlers;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
import
org.apache.druid.segment.realtime.appenderator.TransactionalSegmentPublisher;
import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthorizationUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.server.security.Resource;
import org.apache.druid.server.security.ResourceAction;
@@ -1441,7 +1442,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getMode(@Context final HttpServletRequest req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
return Response.ok(isParallelMode() ? "parallel" : "sequential").build();
}
@@ -1450,7 +1451,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getPhaseName(@Context final HttpServletRequest req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
if (isParallelMode()) {
final ParallelIndexTaskRunner runner = getCurrentRunner();
if (runner == null) {
@@ -1468,7 +1469,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getProgress(@Context final HttpServletRequest req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final ParallelIndexTaskRunner currentRunner = getCurrentRunner();
if (currentRunner == null) {
return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("task
is not running yet").build();
@@ -1482,7 +1483,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getRunningTasks(@Context final HttpServletRequest req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final ParallelIndexTaskRunner currentRunner = getCurrentRunner();
if (currentRunner == null) {
return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("task
is not running yet").build();
@@ -1496,7 +1497,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getSubTaskSpecs(@Context final HttpServletRequest req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final ParallelIndexTaskRunner currentRunner = getCurrentRunner();
if (currentRunner == null) {
return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("task
is not running yet").build();
@@ -1510,7 +1511,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getRunningSubTaskSpecs(@Context final HttpServletRequest req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final ParallelIndexTaskRunner currentRunner = getCurrentRunner();
if (currentRunner == null) {
return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("task
is not running yet").build();
@@ -1524,7 +1525,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getCompleteSubTaskSpecs(@Context final HttpServletRequest
req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final ParallelIndexTaskRunner currentRunner = getCurrentRunner();
if (currentRunner == null) {
return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("task
is not running yet").build();
@@ -1538,7 +1539,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getSubTaskSpec(@PathParam("id") String id, @Context final
HttpServletRequest req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final ParallelIndexTaskRunner currentRunner = getCurrentRunner();
if (currentRunner == null) {
@@ -1558,7 +1559,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Produces(MediaType.APPLICATION_JSON)
public Response getSubTaskState(@PathParam("id") String id, @Context final
HttpServletRequest req)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final ParallelIndexTaskRunner currentRunner = getCurrentRunner();
if (currentRunner == null) {
return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("task
is not running yet").build();
@@ -1580,7 +1581,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@Context final HttpServletRequest req
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
final ParallelIndexTaskRunner currentRunner = getCurrentRunner();
if (currentRunner == null) {
return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("task
is not running yet").build();
@@ -1826,7 +1827,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@QueryParam("full") String full
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
return Response.ok(doGetRowStatsAndUnparseableEvents(full != null,
false).lhs).build();
}
@@ -1862,7 +1863,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
@QueryParam("full") String full
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
return Response.ok(doGetLiveReports(full != null)).build();
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentGenerateTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentGenerateTask.java
index 620d593467a..c27c112071f 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentGenerateTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentGenerateTask.java
@@ -20,7 +20,6 @@
package org.apache.druid.indexing.common.task.batch.parallel;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
import org.apache.druid.data.input.InputSource;
import org.apache.druid.indexer.IngestionState;
import org.apache.druid.indexer.TaskStatus;
@@ -30,7 +29,6 @@ import
org.apache.druid.indexing.common.TaskRealtimeMetricsMonitorBuilder;
import org.apache.druid.indexing.common.TaskToolbox;
import org.apache.druid.indexing.common.stats.TaskRealtimeMetricsMonitor;
import org.apache.druid.indexing.common.task.BatchAppenderators;
-import org.apache.druid.indexing.common.task.IndexTaskUtils;
import org.apache.druid.indexing.common.task.InputSourceProcessor;
import org.apache.druid.indexing.common.task.SegmentAllocatorForBatch;
import org.apache.druid.indexing.common.task.SequenceNameFunction;
@@ -57,9 +55,9 @@ import
org.checkerframework.checker.nullness.qual.MonotonicNonNull;
import java.io.File;
import java.io.IOException;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
@@ -253,18 +251,12 @@ abstract class PartialSegmentGenerateTask<T extends
GeneratedPartitionsReport> e
@Override
protected Map<String, Object> getTaskCompletionUnparseableEvents()
{
- Map<String, Object> unparseableEventsMap = new HashMap<>();
- List<ParseExceptionReport> parseExceptionMessages =
IndexTaskUtils.getReportListFromSavedParseExceptions(
- parseExceptionHandler.getSavedParseExceptionReports()
+ List<ParseExceptionReport> parseExceptionMessages =
Objects.requireNonNullElse(
+ parseExceptionHandler.getSavedParseExceptionReports(),
+ List.of()
);
- if (parseExceptionMessages != null) {
- unparseableEventsMap.put(RowIngestionMeters.BUILD_SEGMENTS,
parseExceptionMessages);
- } else {
- unparseableEventsMap.put(RowIngestionMeters.BUILD_SEGMENTS,
ImmutableList.of());
- }
-
- return unparseableEventsMap;
+ return Map.of(RowIngestionMeters.BUILD_SEGMENTS, parseExceptionMessages);
}
@Override
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
index 0b459913de4..8a0a8d3e6fc 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
@@ -25,7 +25,6 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.FluentIterable;
-import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.apache.druid.data.input.InputRow;
import org.apache.druid.data.input.InputSource;
@@ -44,7 +43,6 @@ import
org.apache.druid.indexing.common.task.AbstractBatchIndexTask;
import org.apache.druid.indexing.common.task.AbstractTask;
import org.apache.druid.indexing.common.task.BatchAppenderators;
import org.apache.druid.indexing.common.task.IndexTask;
-import org.apache.druid.indexing.common.task.IndexTaskUtils;
import org.apache.druid.indexing.common.task.SegmentAllocatorForBatch;
import org.apache.druid.indexing.common.task.SegmentAllocators;
import org.apache.druid.indexing.common.task.TaskResource;
@@ -68,6 +66,7 @@ import
org.apache.druid.segment.realtime.appenderator.BaseAppenderatorDriver;
import org.apache.druid.segment.realtime.appenderator.BatchAppenderatorDriver;
import
org.apache.druid.segment.realtime.appenderator.SegmentsAndCommitMetadata;
import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthorizationUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.server.security.Resource;
import org.apache.druid.server.security.ResourceAction;
@@ -96,6 +95,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
@@ -488,15 +488,13 @@ public class SinglePhaseSubTask extends
AbstractBatchSubtask implements ChatHand
@QueryParam("full") String full
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
Map<String, List<ParseExceptionReport>> events = new HashMap<>();
if (addBuildSegmentStatsToReport(full != null, ingestionState)) {
events.put(
RowIngestionMeters.BUILD_SEGMENTS,
- IndexTaskUtils.getReportListFromSavedParseExceptions(
- parseExceptionHandler.getSavedParseExceptionReports()
- )
+ parseExceptionHandler.getSavedParseExceptionReports()
);
}
@@ -533,7 +531,7 @@ public class SinglePhaseSubTask extends
AbstractBatchSubtask implements ChatHand
@QueryParam("full") String full
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
return Response.ok(doGetRowStats(full != null)).build();
}
@@ -554,7 +552,7 @@ public class SinglePhaseSubTask extends
AbstractBatchSubtask implements ChatHand
@QueryParam("full") String full
)
{
- IndexTaskUtils.datasourceAuthorizationCheck(req, Action.READ,
getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(req,
getDataSource(), authorizerMapper);
return Response.ok(doGetLiveReports(full != null)).build();
}
@@ -578,17 +576,11 @@ public class SinglePhaseSubTask extends
AbstractBatchSubtask implements ChatHand
@Override
protected Map<String, Object> getTaskCompletionUnparseableEvents()
{
- Map<String, Object> unparseableEventsMap = new HashMap<>();
- List<ParseExceptionReport> parseExceptionMessages =
IndexTaskUtils.getReportListFromSavedParseExceptions(
- parseExceptionHandler.getSavedParseExceptionReports()
+ List<ParseExceptionReport> parseExceptionMessages =
Objects.requireNonNullElse(
+ parseExceptionHandler.getSavedParseExceptionReports(),
+ List.of()
);
- if (parseExceptionMessages != null) {
- unparseableEventsMap.put(RowIngestionMeters.BUILD_SEGMENTS,
parseExceptionMessages);
- } else {
- unparseableEventsMap.put(RowIngestionMeters.BUILD_SEGMENTS,
ImmutableList.of());
- }
-
- return unparseableEventsMap;
+ return Map.of(RowIngestionMeters.BUILD_SEGMENTS, parseExceptionMessages);
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
index a30d5f2368e..cb5214eb6d9 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
@@ -67,7 +67,6 @@ import
org.apache.druid.indexing.common.actions.SegmentLockAcquireAction;
import org.apache.druid.indexing.common.actions.TaskLocks;
import org.apache.druid.indexing.common.actions.TimeChunkLockAcquireAction;
import org.apache.druid.indexing.common.stats.TaskRealtimeMetricsMonitor;
-import org.apache.druid.indexing.common.task.IndexTaskUtils;
import org.apache.druid.indexing.input.InputRowSchemas;
import
org.apache.druid.indexing.seekablestream.common.OrderedPartitionableRecord;
import org.apache.druid.indexing.seekablestream.common.OrderedSequenceNumber;
@@ -90,8 +89,7 @@ import
org.apache.druid.segment.realtime.appenderator.AppenderatorDriverAddResul
import
org.apache.druid.segment.realtime.appenderator.SegmentsAndCommitMetadata;
import org.apache.druid.segment.realtime.appenderator.StreamAppenderator;
import org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriver;
-import org.apache.druid.server.security.Action;
-import org.apache.druid.server.security.AuthorizationResult;
+import org.apache.druid.server.security.AuthorizationUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.utils.CollectionUtils;
@@ -1177,9 +1175,8 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
private Map<String, Object> getTaskCompletionUnparseableEvents()
{
Map<String, Object> unparseableEventsMap = new HashMap<>();
- List<ParseExceptionReport> buildSegmentsParseExceptionMessages =
IndexTaskUtils.getReportListFromSavedParseExceptions(
- parseExceptionHandler.getSavedParseExceptionReports()
- );
+ List<ParseExceptionReport> buildSegmentsParseExceptionMessages =
+ parseExceptionHandler.getSavedParseExceptionReports();
if (buildSegmentsParseExceptionMessages != null) {
unparseableEventsMap.put(RowIngestionMeters.BUILD_SEGMENTS,
buildSegmentsParseExceptionMessages);
}
@@ -1452,12 +1449,10 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
/**
* Authorizes action to be performed on this task's datasource
- *
- * @return authorization result
*/
- private AuthorizationResult authorizationCheck(final HttpServletRequest req,
Action action)
+ private void authorizationCheck(final HttpServletRequest request)
{
- return IndexTaskUtils.datasourceAuthorizationCheck(req, action,
task.getDataSource(), authorizerMapper);
+ AuthorizationUtils.verifyUnrestrictedAccessToDatasource(request,
task.getDataSource(), authorizerMapper);
}
public Appenderator getAppenderator()
@@ -1537,7 +1532,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Path("/stop")
public Response stop(@Context final HttpServletRequest req)
{
- authorizationCheck(req, Action.WRITE);
+ authorizationCheck(req);
stopGracefully();
return Response.status(Response.Status.OK).build();
}
@@ -1547,7 +1542,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Produces(MediaType.APPLICATION_JSON)
public Status getStatusHTTP(@Context final HttpServletRequest req)
{
- authorizationCheck(req, Action.READ);
+ authorizationCheck(req);
return status;
}
@@ -1562,7 +1557,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Produces(MediaType.APPLICATION_JSON)
public Map<PartitionIdType, SequenceOffsetType> getCurrentOffsets(@Context
final HttpServletRequest req)
{
- authorizationCheck(req, Action.READ);
+ authorizationCheck(req);
return getCurrentOffsets();
}
@@ -1576,7 +1571,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Produces(MediaType.APPLICATION_JSON)
public Map<PartitionIdType, SequenceOffsetType> getEndOffsetsHTTP(@Context
final HttpServletRequest req)
{
- authorizationCheck(req, Action.READ);
+ authorizationCheck(req);
return getEndOffsets();
}
@@ -1596,7 +1591,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Context final HttpServletRequest req
) throws InterruptedException
{
- authorizationCheck(req, Action.WRITE);
+ authorizationCheck(req);
return setEndOffsets(sequences, finish);
}
@@ -1610,7 +1605,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Context final HttpServletRequest req
)
{
- authorizationCheck(req, Action.WRITE);
+ authorizationCheck(req);
try {
((StreamAppenderator)
appenderator).registerUpgradedPendingSegment(upgradedPendingSegment);
return Response.ok().build();
@@ -1653,7 +1648,6 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
public Map<String, Object> doGetLiveReports()
{
- Map<String, Object> returnMap = new HashMap<>();
Map<String, Object> ingestionStatsAndErrors = new HashMap<>();
Map<String, Object> payload = new HashMap<>();
Map<String, Object> events = getTaskCompletionUnparseableEvents();
@@ -1666,8 +1660,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
ingestionStatsAndErrors.put("payload", payload);
ingestionStatsAndErrors.put("type", "ingestionStatsAndErrors");
- returnMap.put("ingestionStatsAndErrors", ingestionStatsAndErrors);
- return returnMap;
+ return Map.of("ingestionStatsAndErrors", ingestionStatsAndErrors);
}
@GET
@@ -1677,7 +1670,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Context final HttpServletRequest req
)
{
- authorizationCheck(req, Action.READ);
+ authorizationCheck(req);
return Response.ok(doGetRowStats()).build();
}
@@ -1688,7 +1681,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Context final HttpServletRequest req
)
{
- authorizationCheck(req, Action.READ);
+ authorizationCheck(req);
return Response.ok(doGetLiveReports()).build();
}
@@ -1700,11 +1693,8 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Context final HttpServletRequest req
)
{
- authorizationCheck(req, Action.READ);
- List<ParseExceptionReport> events =
IndexTaskUtils.getReportListFromSavedParseExceptions(
- parseExceptionHandler.getSavedParseExceptionReports()
- );
- return Response.ok(events).build();
+ authorizationCheck(req);
+ return
Response.ok(parseExceptionHandler.getSavedParseExceptionReports()).build();
}
@VisibleForTesting
@@ -1854,7 +1844,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Context final HttpServletRequest req
)
{
- authorizationCheck(req, Action.READ);
+ authorizationCheck(req);
return getCheckpoints();
}
@@ -1881,7 +1871,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Context final HttpServletRequest req
) throws InterruptedException
{
- authorizationCheck(req, Action.WRITE);
+ authorizationCheck(req);
return pause();
}
@@ -1940,7 +1930,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Path("/resume")
public Response resumeHTTP(@Context final HttpServletRequest req) throws
InterruptedException
{
- authorizationCheck(req, Action.WRITE);
+ authorizationCheck(req);
resume();
return Response.status(Response.Status.OK).build();
}
@@ -1973,7 +1963,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
@Produces(MediaType.APPLICATION_JSON)
public DateTime getStartTime(@Context final HttpServletRequest req)
{
- authorizationCheck(req, Action.WRITE);
+ authorizationCheck(req);
return startTime;
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java
index bef25dec434..53549d3c791 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java
@@ -2532,6 +2532,7 @@ public class IndexTaskTest extends IngestionTestBase
private <T> T invokeApi(Function<HttpServletRequest, Response> api)
{
final HttpServletRequest request = EasyMock.mock(HttpServletRequest.class);
+ EasyMock.expect(request.getMethod()).andReturn("GET").anyTimes();
EasyMock.expect(request.getAttribute(EasyMock.anyString()))
.andReturn("allow-all");
EasyMock.replay(request);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskResourceTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskResourceTest.java
index d7ec15deff7..1222679ced8 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskResourceTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskResourceTest.java
@@ -386,6 +386,7 @@ public class ParallelIndexSupervisorTaskResourceTest
extends AbstractParallelInd
private static HttpServletRequest newRequest()
{
final HttpServletRequest request =
EasyMock.niceMock(HttpServletRequest.class);
+ EasyMock.expect(request.getMethod()).andReturn("GET").anyTimes();
EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null);
EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT))
.andReturn(new AuthenticationResult("test", "test", "test",
Collections.emptyMap()));
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerAuthTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerAuthTest.java
index e30502254bb..89ee5e6ca35 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerAuthTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerAuthTest.java
@@ -197,12 +197,12 @@ public class SeekableStreamIndexTaskRunnerAuthTest
)
{
// Verify that datasource write user can access
- HttpServletRequest allowedRequest = createRequest(Users.DATASOURCE_WRITE);
+ HttpServletRequest allowedRequest = createRequest(Users.DATASOURCE_WRITE,
"POST");
replay(allowedRequest);
method.accept(allowedRequest);
// Verify that no other user can access
- HttpServletRequest blockedRequest = createRequest(Users.DATASOURCE_READ);
+ HttpServletRequest blockedRequest = createRequest(Users.DATASOURCE_READ,
"POST");
replay(blockedRequest);
expectedException.expect(ForbiddenException.class);
method.accept(blockedRequest);
@@ -213,20 +213,21 @@ public class SeekableStreamIndexTaskRunnerAuthTest
)
{
// Verify that datasource read user can access
- HttpServletRequest allowedRequest = createRequest(Users.DATASOURCE_READ);
+ HttpServletRequest allowedRequest = createRequest(Users.DATASOURCE_READ,
"GET");
replay(allowedRequest);
method.accept(allowedRequest);
// Verify that no other user can access
- HttpServletRequest blockedRequest = createRequest(Users.DATASOURCE_WRITE);
+ HttpServletRequest blockedRequest = createRequest(Users.DATASOURCE_WRITE,
"GET");
replay(blockedRequest);
expectedException.expect(ForbiddenException.class);
method.accept(blockedRequest);
}
- private HttpServletRequest createRequest(String username)
+ private HttpServletRequest createRequest(String username, String method)
{
HttpServletRequest request = mock(HttpServletRequest.class);
+ EasyMock.expect(request.getMethod()).andReturn(method);
AuthenticationResult authenticationResult = new
AuthenticationResult(username, "druid", null, null);
EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).anyTimes();
diff --git
a/processing/src/main/java/org/apache/druid/segment/incremental/ParseExceptionHandler.java
b/processing/src/main/java/org/apache/druid/segment/incremental/ParseExceptionHandler.java
index a67da717272..cddd5a5e7b6 100644
---
a/processing/src/main/java/org/apache/druid/segment/incremental/ParseExceptionHandler.java
+++
b/processing/src/main/java/org/apache/druid/segment/incremental/ParseExceptionHandler.java
@@ -29,6 +29,8 @@ import
org.apache.druid.java.util.common.parsers.UnparseableColumnsParseExceptio
import org.apache.druid.utils.CircularBuffer;
import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
/**
* A handler for {@link ParseException}s thrown during ingestion. Based on the
given configuration, this handler can
@@ -98,9 +100,18 @@ public class ParseExceptionHandler
}
@Nullable
- public CircularBuffer<ParseExceptionReport> getSavedParseExceptionReports()
+ public List<ParseExceptionReport> getSavedParseExceptionReports()
{
- return savedParseExceptionReports;
+ if (savedParseExceptionReports == null) {
+ return null;
+ }
+
+ final List<ParseExceptionReport> reports = new ArrayList<>();
+ for (int i = 0; i < savedParseExceptionReports.size(); i++) {
+ reports.add(savedParseExceptionReports.getLatest(i));
+ }
+
+ return reports;
}
@VisibleForTesting
diff --git
a/processing/src/test/java/org/apache/druid/segment/incremental/ParseExceptionHandlerTest.java
b/processing/src/test/java/org/apache/druid/segment/incremental/ParseExceptionHandlerTest.java
index 26c70c9cd1d..75cb8dbfd87 100644
---
a/processing/src/test/java/org/apache/druid/segment/incremental/ParseExceptionHandlerTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/incremental/ParseExceptionHandlerTest.java
@@ -137,7 +137,7 @@ public class ParseExceptionHandlerTest
for (int i = 0; i < maxSavedParseExceptions; i++) {
Assert.assertEquals(
StringUtils.format("test %d", i),
-
parseExceptionHandler.getSavedParseExceptionReports().get(i).getDetails().get(0)
+
parseExceptionHandler.getSavedParseExceptionReports().get(maxSavedParseExceptions
- i - 1).getDetails().get(0)
);
}
for (; exceptionCounter < 5; exceptionCounter++) {
@@ -149,7 +149,7 @@ public class ParseExceptionHandlerTest
for (int i = 0; i < maxSavedParseExceptions; i++) {
Assert.assertEquals(
StringUtils.format("test %d", i + 2),
-
parseExceptionHandler.getSavedParseExceptionReports().get(i).getDetails().get(0)
+
parseExceptionHandler.getSavedParseExceptionReports().get(maxSavedParseExceptions
- i - 1).getDetails().get(0)
);
}
}
diff --git
a/server/src/main/java/org/apache/druid/server/security/AuthorizationUtils.java
b/server/src/main/java/org/apache/druid/server/security/AuthorizationUtils.java
index 741a5f1ae08..94ae93e3adb 100644
---
a/server/src/main/java/org/apache/druid/server/security/AuthorizationUtils.java
+++
b/server/src/main/java/org/apache/druid/server/security/AuthorizationUtils.java
@@ -30,6 +30,7 @@ import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.query.policy.Policy;
+import javax.annotation.Nullable;
import javax.servlet.http.HttpServletRequest;
import java.util.ArrayList;
import java.util.Collections;
@@ -73,6 +74,26 @@ public class AuthorizationUtils
);
}
+ /**
+ * Verifies that the user has unrestricted access to perform the required
+ * action on the given datasource.
+ *
+ * @throws ForbiddenException if the user does not have unrestricted access
to
+ * perform the required action on the given datasource.
+ */
+ public static void verifyUnrestrictedAccessToDatasource(
+ final HttpServletRequest req,
+ String datasource,
+ AuthorizerMapper authorizerMapper
+ )
+ {
+ ResourceAction resourceAction = createDatasourceResourceAction(datasource,
req);
+ AuthorizationResult authResult = authorizeResourceAction(req,
resourceAction, authorizerMapper);
+ if (!authResult.allowAccessWithNoRestriction()) {
+ throw new ForbiddenException(authResult.getErrorMessage());
+ }
+ }
+
/**
* Returns the authentication information for a request.
*
@@ -97,6 +118,7 @@ public class AuthorizationUtils
* Extracts the identity from the authentication result if set as an
atrribute
* of this request.
*/
+ @Nullable
public static String getAuthenticatedIdentity(HttpServletRequest request)
{
final AuthenticationResult authenticationResult = (AuthenticationResult)
request.getAttribute(
@@ -332,7 +354,7 @@ public class AuthorizationUtils
}
final Map<ResourceAction, Access> resultCache = new HashMap<>();
- final Iterable<ResType> filteredResources = Iterables.filter(
+ return Iterables.filter(
resources,
resource -> {
final Iterable<ResourceAction> resourceActions =
resourceActionGenerator.apply(resource);
@@ -358,8 +380,6 @@ public class AuthorizationUtils
return true;
}
);
-
- return filteredResources;
}
/**
@@ -414,7 +434,7 @@ public class AuthorizationUtils
)
);
- if (filteredList.size() > 0) {
+ if (!filteredList.isEmpty()) {
filteredResources.put(
entry.getKey(),
filteredList
@@ -508,20 +528,4 @@ public class AuthorizationUtils
Action.WRITE
);
- /**
- * Function for the common pattern of generating a resource-action for
reading from a view, using the
- * view name.
- */
- public static final Function<String, ResourceAction> VIEW_READ_RA_GENERATOR
= input -> new ResourceAction(
- new Resource(input, ResourceType.VIEW),
- Action.READ
- );
-
- /**
- * Function for the pattern of generating a {@link ResourceAction} for
reading from a given {@link Resource}
- */
- public static final Function<Resource, ResourceAction>
RESOURCE_READ_RA_GENERATOR = input -> new ResourceAction(
- input,
- Action.READ
- );
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
