You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/08/15 23:57:12 UTC
(pinot) branch master updated: [server] Enforce access control on admin API routes (#19233)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new f379d2f165e [server] Enforce access control on admin API routes (#19233)
f379d2f165e is described below
commit f379d2f165e2f625e109b06e0f310efa89b005ba
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Sat Aug 15 16:57:05 2026 -0700
[server] Enforce access control on admin API routes (#19233)
---
.../BaseSingleStageBrokerRequestHandler.java | 7 +-
.../BaseSingleStageBrokerRequestHandlerTest.java | 77 ++++++--
.../apache/pinot/common/http/MultiHttpRequest.java | 26 ++-
.../pinot/common/swagger/SwaggerSetupUtils.java | 14 +-
.../common/utils/FileUploadDownloadClient.java | 21 ++
.../pinot/common/http/MultiHttpRequestTest.java | 84 ++++++++
.../common/utils/FileUploadDownloadClientTest.java | 56 ++++++
.../apache/pinot/controller/ControllerConf.java | 2 +
.../controller/api/resources/DebugResource.java | 6 +-
.../api/resources/PinotControllerLogger.java | 20 +-
.../api/resources/PinotTableInstances.java | 3 +-
.../api/resources/PinotTableRestletResource.java | 6 +-
.../api/resources/ServerTableSizeReader.java | 12 +-
.../helix/core/PinotHelixResourceManager.java | 12 ++
.../realtime/PinotLLCRealtimeSegmentManager.java | 12 +-
.../services/PinotTableReloadStatusReporter.java | 3 +-
.../controller/util/CompletionServiceHelper.java | 59 +++++-
.../util/ConsumingSegmentInfoReader.java | 3 +-
.../util/ServerCompressionStatsReader.java | 11 ++
.../util/ServerSegmentMetadataReader.java | 33 ++--
.../controller/util/ServerTableTierReader.java | 11 +-
.../pinot/controller/util/TableMetadataReader.java | 21 +-
.../pinot/controller/util/TableSizeReader.java | 6 +-
.../pinot/controller/util/TableTierReader.java | 3 +-
.../validation/ResourceUtilizationChecker.java | 3 +-
.../workload/WorkloadPropagationClient.java | 49 ++++-
.../PinotLLCRealtimeSegmentManagerTest.java | 55 ++++--
.../workload/WorkloadPropagationClientTest.java | 141 +++++++++++++
.../apache/pinot/core/auth/BasicAuthPrincipal.java | 5 +
.../apache/pinot/server/access/AccessControl.java | 18 ++
.../pinot/server/access/AllowAllAccessFactory.java | 7 +
.../server/access/BasicAuthAccessFactory.java | 52 +++--
.../pinot/server/access/HttpRequesterIdentity.java | 23 +++
.../server/access/ZkBasicAuthAccessFactory.java | 102 +++++++---
.../access/ZkBasicAuthAccessFactoryTest.java | 83 ++++++++
.../tests/BasicAuthBatchIntegrationTest.java | 16 ++
.../CertBasedTlsChannelAccessControlFactory.java | 8 +
pinot-server/README.md | 120 +++++++++++
.../pinot/server/api/AdminApiApplication.java | 9 +-
.../api/ServerAdminApiAccessControlFilter.java | 83 ++++++++
.../api/ServerAdminApiAccessControlHandler.java | 89 +++++++++
.../apache/pinot/server/api/ServerDataAccess.java | 34 ++--
.../pinot/server/api/ServerPublicAccess.java | 32 ++-
.../api/ServerRequestMethodCaptureFilter.java | 32 ++-
.../api/resources/DefaultExceptionMapper.java | 2 +-
.../server/api/resources/HealthCheckResource.java | 4 +
.../pinot/server/api/resources/TablesResource.java | 3 +
.../apache/pinot/server/api/AccessControlTest.java | 219 +++++++++++++++------
.../server/api/PinotServerAppConfigsTest.java | 3 +-
.../api/ServerAdminApiAccessControlFilterTest.java | 213 ++++++++++++++++++++
.../apache/pinot/spi/utils/CommonConstants.java | 2 +
51 files changed, 1656 insertions(+), 259 deletions(-)
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
index 94fa3709686..b61abf35f08 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
@@ -55,6 +55,7 @@ import org.apache.pinot.broker.api.AccessControl;
import org.apache.pinot.broker.broker.AccessControlFactory;
import org.apache.pinot.broker.querylog.QueryLogger;
import org.apache.pinot.broker.queryquota.QueryQuotaManager;
+import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.common.evaluator.GroovyFunctionEvaluator;
import org.apache.pinot.common.http.MultiHttpRequest;
@@ -107,6 +108,7 @@ import org.apache.pinot.materializedview.handler.MaterializedViewSplitExecutionC
import org.apache.pinot.materializedview.rewrite.MaterializedViewRewritePlan;
import org.apache.pinot.query.parser.utils.ParserUtils;
import org.apache.pinot.spi.accounting.ThreadAccountant;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.auth.AuthorizationResult;
import org.apache.pinot.spi.auth.TableAuthorizationResult;
import org.apache.pinot.spi.auth.TableRowColAccessResult;
@@ -170,6 +172,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ
protected final boolean _enableMultistageMigrationMetric;
protected final boolean _useMSEToFillEmptyResponseSchema;
protected final boolean _enableQueryFingerprinting;
+ protected final AuthProvider _serverAdminAuthProvider;
protected ExecutorService _multistageCompileExecutor;
protected BlockingQueue<Pair<String, String>> _multistageCompileQueryQueue;
protected ImplicitHybridTableRouteProvider _implicitHybridTableRouteProvider;
@@ -195,6 +198,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ
super(config, brokerId, requestIdGenerator, routingManager, accessControlFactory, queryQuotaManager, tableCache,
threadAccountant, multiClusterRoutingContext);
_materializedViewHandler = materializedViewHandler;
+ _serverAdminAuthProvider = AuthProviderUtils.extractAuthProvider(config, Broker.SERVER_ADMIN_AUTH_PREFIX);
_disableGroovy = _config.getProperty(Broker.DISABLE_GROOVY, Broker.DEFAULT_DISABLE_GROOVY);
_useApproximateFunction = _config.getProperty(Broker.USE_APPROXIMATE_FUNCTION, false);
_defaultHllLog2m = _config.getProperty(CommonConstants.Helix.DEFAULT_HYPERLOGLOG_LOG2M_KEY,
@@ -313,7 +317,8 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ
LOGGER.debug("Cancelling the query: {} via server urls: {}", _queryLogger.redactQuery(queryServers._query),
serverUrls);
CompletionService<MultiHttpRequestResponse> completionService =
- new MultiHttpRequest(executor, connMgr).execute(serverUrls, null, timeoutMs, "DELETE", HttpDelete::new);
+ new MultiHttpRequest(executor, connMgr).execute(serverUrls, null, _serverAdminAuthProvider, timeoutMs,
+ "DELETE", HttpDelete::new);
List<String> errMsgs = new ArrayList<>(serverUrls.size());
for (int i = 0; i < serverUrls.size(); i++) {
MultiHttpRequestResponse httpRequestResponse = null;
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandlerTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandlerTest.java
index 0d441a05038..ae661c0107a 100644
--- a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandlerTest.java
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandlerTest.java
@@ -18,6 +18,8 @@
*/
package org.apache.pinot.broker.requesthandler;
+import com.sun.net.httpserver.HttpServer;
+import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -25,7 +27,11 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.broker.api.AccessControl;
import org.apache.pinot.broker.broker.AccessControlFactory;
@@ -283,8 +289,19 @@ public class BaseSingleStageBrokerRequestHandlerTest {
}
@Test
- public void testCancelQuery() {
+ public void testCancelQuery()
+ throws Exception {
String tableName = "myTable_OFFLINE";
+ String serviceToken = "server-admin-token";
+ AtomicReference<String> receivedAuthorization = new AtomicReference<>();
+ HttpServer adminServer = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+ adminServer.createContext("/", exchange -> {
+ receivedAuthorization.set(exchange.getRequestHeaders().getFirst("Authorization"));
+ exchange.sendResponseHeaders(200, -1);
+ exchange.close();
+ });
+ adminServer.start();
+
// Mock pretty much everything until the query can be submitted.
TableCache tableCache = mock(TableCache.class);
TableConfig tableCfg = mock(TableConfig.class);
@@ -296,7 +313,12 @@ public class BaseSingleStageBrokerRequestHandlerTest {
when(routingManager.routingExists(tableName)).thenReturn(true);
when(routingManager.getQueryTimeoutMs(tableName)).thenReturn(10000L);
RoutingTable rt = mock(RoutingTable.class);
- when(rt.getServerInstanceToSegmentsMap()).thenReturn(Map.of(new ServerInstance(new InstanceConfig("server01_9000")),
+ int adminPort = adminServer.getAddress().getPort();
+ InstanceConfig serverConfig = new InstanceConfig("Server_localhost_9000");
+ serverConfig.setHostName("localhost");
+ serverConfig.setPort("9000");
+ serverConfig.getRecord().setIntField(CommonConstants.Helix.Instance.ADMIN_PORT_KEY, adminPort);
+ when(rt.getServerInstanceToSegmentsMap()).thenReturn(Map.of(new ServerInstance(serverConfig),
new SegmentsToQuery(List.of("segment01"), List.of())));
when(routingManager.getRoutingTable(any(), Mockito.anyLong())).thenReturn(rt);
QueryQuotaManager queryQuotaManager = mock(QueryQuotaManager.class);
@@ -307,6 +329,8 @@ public class BaseSingleStageBrokerRequestHandlerTest {
long[] testRequestId = {-1};
BrokerMetrics.register(mock(BrokerMetrics.class));
PinotConfiguration config = new PinotConfiguration();
+ config.setProperty(Broker.SERVER_ADMIN_AUTH_PREFIX + ".token", serviceToken);
+ config.setProperty(Broker.SERVER_ADMIN_AUTH_PREFIX + ".prefix", "Bearer");
BrokerQueryEventListenerFactory.init(config);
BaseSingleStageBrokerRequestHandler requestHandler =
new BaseSingleStageBrokerRequestHandler(config, "testBrokerId", new BrokerRequestIdGenerator(), routingManager,
@@ -327,7 +351,7 @@ public class BaseSingleStageBrokerRequestHandlerTest {
throws Exception {
testRequestId[0] = requestId;
latch.await();
- return null;
+ return BrokerResponseNative.empty();
}
@Override
@@ -339,25 +363,46 @@ public class BaseSingleStageBrokerRequestHandlerTest {
throw new UnsupportedOperationException("Not implemented in test");
}
};
- CompletableFuture.runAsync(() -> {
+ CompletableFuture<Void> queryFuture = CompletableFuture.runAsync(() -> {
try {
requestHandler.handleRequest(String.format("select * from %s limit 10", tableName));
} catch (Exception e) {
throw new RuntimeException(e);
}
});
- TestUtils.waitForCondition((aVoid) -> requestHandler.getRunningServers(testRequestId[0]).size() == 1, 500, 5000,
- "Failed to submit query");
- Map.Entry<Long, String> entry = requestHandler.getRunningQueries().entrySet().iterator().next();
- Assert.assertEquals(entry.getKey().longValue(), testRequestId[0]);
- Assert.assertTrue(entry.getValue().contains("select * from myTable_OFFLINE limit 10"));
- Set<ServerInstance> servers = requestHandler.getRunningServers(testRequestId[0]);
- Assert.assertEquals(servers.size(), 1);
- Assert.assertEquals(servers.iterator().next().getHostname(), "server01");
- Assert.assertEquals(servers.iterator().next().getPort(), 9000);
- Assert.assertEquals(servers.iterator().next().getInstanceId(), "server01_9000");
- Assert.assertEquals(servers.iterator().next().getAdminEndpoint(), "http://server01:8097");
- latch.countDown();
+ ExecutorService cancellationExecutor = Executors.newSingleThreadExecutor();
+ PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
+ try {
+ TestUtils.waitForCondition((aVoid) -> requestHandler.getRunningServers(testRequestId[0]).size() == 1, 500, 5000,
+ "Failed to submit query");
+ Map.Entry<Long, String> entry = requestHandler.getRunningQueries().entrySet().iterator().next();
+ Assert.assertEquals(entry.getKey().longValue(), testRequestId[0]);
+ Assert.assertTrue(entry.getValue().contains("select * from myTable_OFFLINE limit 10"));
+ Set<ServerInstance> servers = requestHandler.getRunningServers(testRequestId[0]);
+ Assert.assertEquals(servers.size(), 1);
+ Assert.assertEquals(servers.iterator().next().getHostname(), "localhost");
+ Assert.assertEquals(servers.iterator().next().getPort(), 9000);
+ Assert.assertEquals(servers.iterator().next().getInstanceId(), "Server_localhost_9000");
+ Assert.assertEquals(servers.iterator().next().getAdminEndpoint(), "http://localhost:" + adminPort);
+
+ Map<String, Integer> serverResponses = new HashMap<>();
+ Assert.assertTrue(requestHandler.cancelQuery(testRequestId[0], 3000, cancellationExecutor, connectionManager,
+ serverResponses));
+ Assert.assertEquals(receivedAuthorization.get(), "Bearer " + serviceToken);
+ Assert.assertEquals(serverResponses, Map.of("localhost:" + adminPort, 200));
+ } finally {
+ latch.countDown();
+ try {
+ queryFuture.get(5, TimeUnit.SECONDS);
+ } finally {
+ cancellationExecutor.shutdownNow();
+ try {
+ connectionManager.close();
+ } finally {
+ adminServer.stop(0);
+ }
+ }
+ }
}
@Test
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/http/MultiHttpRequest.java b/pinot-common/src/main/java/org/apache/pinot/common/http/MultiHttpRequest.java
index b297105af16..e292a71d861 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/http/MultiHttpRequest.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/http/MultiHttpRequest.java
@@ -44,6 +44,8 @@ import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import org.apache.hc.core5.http.io.entity.StringEntity;
import org.apache.hc.core5.util.Timeout;
+import org.apache.pinot.common.auth.AuthProviderUtils;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -117,17 +119,22 @@ public class MultiHttpRequest {
public <T extends HttpUriRequestBase> CompletionService<MultiHttpRequestResponse> execute(
List<Pair<String, String>> urlsAndRequestBodies, @Nullable Map<String, String> requestHeaders, int timeoutMs,
String httpMethodName, Function<String, T> httpRequestBaseSupplier) {
+ return execute(urlsAndRequestBodies, requestHeaders, null, timeoutMs, httpMethodName, httpRequestBaseSupplier);
+ }
+
+ /// Executes HTTP requests and resolves authentication headers from the provider separately for every request.
+ /// Provider headers take precedence over the caller-supplied functional headers.
+ public <T extends HttpUriRequestBase> CompletionService<MultiHttpRequestResponse> execute(
+ List<Pair<String, String>> urlsAndRequestBodies, @Nullable Map<String, String> requestHeaders,
+ @Nullable AuthProvider authProvider, int timeoutMs, String httpMethodName,
+ Function<String, T> httpRequestBaseSupplier) {
// Create global request configuration
Timeout timeout = Timeout.of(timeoutMs, TimeUnit.MILLISECONDS);
RequestConfig defaultRequestConfig =
RequestConfig.custom().setConnectionRequestTimeout(timeout).setResponseTimeout(timeout)
.build(); // setting the socket
- HttpClientBuilder httpClientBuilder =
- HttpClients.custom().setConnectionManager(_connectionManager).setDefaultRequestConfig(defaultRequestConfig);
-
- CompletionService<MultiHttpRequestResponse> completionService = new ExecutorCompletionService<>(_executor);
- CloseableHttpClient client = httpClientBuilder.build();
+ List<HttpUriRequestBase> requests = new ArrayList<>(urlsAndRequestBodies.size());
for (Pair<String, String> pair : urlsAndRequestBodies) {
String url = pair.getLeft();
String body = pair.getRight();
@@ -139,6 +146,15 @@ public class MultiHttpRequest {
if (requestHeaders != null) {
requestHeaders.forEach(httpMethod::setHeader);
}
+ AuthProviderUtils.makeAuthHeadersMap(authProvider).forEach(httpMethod::setHeader);
+ requests.add(httpMethod);
+ }
+
+ HttpClientBuilder httpClientBuilder =
+ HttpClients.custom().setConnectionManager(_connectionManager).setDefaultRequestConfig(defaultRequestConfig);
+ CompletionService<MultiHttpRequestResponse> completionService = new ExecutorCompletionService<>(_executor);
+ CloseableHttpClient client = httpClientBuilder.build();
+ for (HttpUriRequestBase httpMethod : requests) {
completionService.submit(() -> {
CloseableHttpResponse response = null;
boolean responseHandedOff = false;
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/swagger/SwaggerSetupUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/swagger/SwaggerSetupUtils.java
index 93f3088c0d5..e9dec9cb3cd 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/swagger/SwaggerSetupUtils.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/swagger/SwaggerSetupUtils.java
@@ -29,11 +29,13 @@ import java.net.URL;
import java.net.URLClassLoader;
import java.util.Objects;
import java.util.Properties;
+import java.util.function.UnaryOperator;
import javax.ws.rs.core.HttpHeaders;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.utils.PinotStaticHttpHandler;
import org.apache.pinot.spi.utils.CommonConstants;
import org.glassfish.grizzly.http.server.CLStaticHttpHandler;
+import org.glassfish.grizzly.http.server.HttpHandler;
import org.glassfish.grizzly.http.server.HttpServer;
@@ -43,19 +45,27 @@ public class SwaggerSetupUtils {
public static void setupSwagger(String componentType, String resourcePackage, boolean useHttps, String basePath,
HttpServer httpServer) {
+ setupSwagger(componentType, resourcePackage, useHttps, basePath, httpServer, UnaryOperator.identity());
+ }
+
+ /// Sets up the Swagger resources, applying the given wrapper to each directly registered Grizzly handler.
+ ///
+ /// @param httpHandlerWrapper wrapper for component-specific request handling such as access control
+ public static void setupSwagger(String componentType, String resourcePackage, boolean useHttps, String basePath,
+ HttpServer httpServer, UnaryOperator<HttpHandler> httpHandlerWrapper) {
BeanConfig beanConfig = buildSwaggerConfig(componentType, resourcePackage, useHttps, basePath);
beanConfig.setScan(true);
ClassLoader classLoader = SwaggerSetupUtils.class.getClassLoader();
CLStaticHttpHandler staticHttpHandler = new CLStaticHttpHandler(classLoader, "/api/");
// map both /api and /help to swagger docs. /api because it looks nice. /help for backward compatibility
- httpServer.getServerConfiguration().addHttpHandler(staticHttpHandler, "/api/", "/help/");
+ httpServer.getServerConfiguration().addHttpHandler(httpHandlerWrapper.apply(staticHttpHandler), "/api/", "/help/");
String swaggerVersion = findSwaggerVersion(classLoader);
URL swaggerDistLocation = classLoader.getResource(
CommonConstants.CONFIG_OF_SWAGGER_RESOURCES_PATH + swaggerVersion + "/");
CLStaticHttpHandler swaggerDist = new PinotStaticHttpHandler(new URLClassLoader(new URL[]{swaggerDistLocation}));
- httpServer.getServerConfiguration().addHttpHandler(swaggerDist, "/swaggerui-dist/");
+ httpServer.getServerConfiguration().addHttpHandler(httpHandlerWrapper.apply(swaggerDist), "/swaggerui-dist/");
}
@VisibleForTesting
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java
index d89d290a9ae..7565384133b 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java
@@ -794,7 +794,14 @@ public class FileUploadDownloadClient implements AutoCloseable {
/// TODO: migrate this method to another class
public String uploadToSegmentStore(String uri)
throws URISyntaxException, IOException, HttpErrorStatusException {
+ return uploadToSegmentStore(uri, null);
+ }
+
+ /// Authenticated variant of [#uploadToSegmentStore(String)].
+ public String uploadToSegmentStore(String uri, @Nullable AuthProvider authProvider)
+ throws URISyntaxException, IOException, HttpErrorStatusException {
ClassicRequestBuilder requestBuilder = ClassicRequestBuilder.post(new URI(uri)).setVersion(HttpVersion.HTTP_1_1);
+ AuthProviderUtils.toRequestHeaders(authProvider).forEach(requestBuilder::setHeader);
// sendRequest checks the response status code
SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
_httpClient.sendRequest(requestBuilder.build(), HttpClient.DEFAULT_SOCKET_TIMEOUT_MS));
@@ -816,7 +823,14 @@ public class FileUploadDownloadClient implements AutoCloseable {
/// @throws HttpErrorStatusException
public TableLLCSegmentUploadResponse uploadLLCToSegmentStore(String uri)
throws URISyntaxException, IOException, HttpErrorStatusException {
+ return uploadLLCToSegmentStore(uri, null);
+ }
+
+ /// Authenticated variant of [#uploadLLCToSegmentStore(String)].
+ public TableLLCSegmentUploadResponse uploadLLCToSegmentStore(String uri, @Nullable AuthProvider authProvider)
+ throws URISyntaxException, IOException, HttpErrorStatusException {
ClassicRequestBuilder requestBuilder = ClassicRequestBuilder.post(new URI(uri)).setVersion(HttpVersion.HTTP_1_1);
+ AuthProviderUtils.toRequestHeaders(authProvider).forEach(requestBuilder::setHeader);
// sendRequest checks the response status code
SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
_httpClient.sendRequest(requestBuilder.build(), HttpClient.DEFAULT_SOCKET_TIMEOUT_MS));
@@ -840,7 +854,14 @@ public class FileUploadDownloadClient implements AutoCloseable {
/// @throws HttpErrorStatusException
public SegmentZKMetadata uploadLLCToSegmentStoreWithZKMetadata(String uri)
throws URISyntaxException, IOException, HttpErrorStatusException {
+ return uploadLLCToSegmentStoreWithZKMetadata(uri, null);
+ }
+
+ /// Authenticated variant of [#uploadLLCToSegmentStoreWithZKMetadata(String)].
+ public SegmentZKMetadata uploadLLCToSegmentStoreWithZKMetadata(String uri, @Nullable AuthProvider authProvider)
+ throws URISyntaxException, IOException, HttpErrorStatusException {
ClassicRequestBuilder requestBuilder = ClassicRequestBuilder.post(new URI(uri)).setVersion(HttpVersion.HTTP_1_1);
+ AuthProviderUtils.toRequestHeaders(authProvider).forEach(requestBuilder::setHeader);
// sendRequest checks the response status code
SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
_httpClient.sendRequest(requestBuilder.build(), HttpClient.DEFAULT_SOCKET_TIMEOUT_MS));
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/http/MultiHttpRequestTest.java b/pinot-common/src/test/java/org/apache/pinot/common/http/MultiHttpRequestTest.java
index 5615ee4bac8..829bb8be087 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/http/MultiHttpRequestTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/http/MultiHttpRequestTest.java
@@ -28,7 +28,10 @@ import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
@@ -40,10 +43,12 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import org.apache.hc.core5.http.io.entity.StringEntity;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -96,6 +101,7 @@ public class MultiHttpRequestTest {
startServer(_portStart + 1, createHandler(ERROR_CODE, ERROR_MSG, 0));
startServer(_portStart + 2, createHandler(SUCCESS_CODE, TIMEOUT_MSG, TIMEOUT_MS));
startServer(_portStart + 3, createPostHandler(SUCCESS_CODE, SUCCESS_MSG, 0));
+ startServer(_portStart + 4, createHeaderEchoHandler());
}
@AfterTest
@@ -152,6 +158,16 @@ public class MultiHttpRequestTest {
};
}
+ private HttpHandler createHeaderEchoHandler() {
+ return httpExchange -> {
+ String response = httpExchange.getRequestHeaders().getFirst("Authorization");
+ httpExchange.sendResponseHeaders(SUCCESS_CODE, response.length());
+ try (OutputStream responseBody = httpExchange.getResponseBody()) {
+ responseBody.write(response.getBytes());
+ }
+ };
+ }
+
private void startServer(int port, HttpHandler handler)
throws IOException {
final HttpServer server = HttpServer.create(new InetSocketAddress(port), 0);
@@ -213,6 +229,74 @@ public class MultiHttpRequestTest {
Assert.assertEquals(result.getTimeouts(), 1);
}
+ @Test
+ public void testAuthProviderResolvedPerRequestAndOverridesFunctionalHeaders()
+ throws Exception {
+ String url = "http://localhost:" + (_portStart + 4) + URI_PATH;
+ List<Pair<String, String>> requests = List.of(Pair.of(url, null), Pair.of(url, null));
+ AtomicInteger tokenSequence = new AtomicInteger();
+ AuthProvider authProvider = new AuthProvider() {
+ @Override
+ public Map<String, Object> getRequestHeaders() {
+ return Map.of("Authorization", "Service-" + tokenSequence.incrementAndGet());
+ }
+
+ @Override
+ public String getTaskToken() {
+ return null;
+ }
+ };
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
+ try {
+ CompletionService<MultiHttpRequestResponse> completionService =
+ new MultiHttpRequest(executor, connectionManager).execute(requests, Map.of("authorization", "End-user"),
+ authProvider, TIMEOUT_MS, "GET", HttpGet::new);
+ Set<String> credentials = new HashSet<>();
+ for (int i = 0; i < requests.size(); i++) {
+ try (MultiHttpRequestResponse response = completionService.take().get()) {
+ credentials.add(EntityUtils.toString(response.getResponse().getEntity()));
+ }
+ }
+ Assert.assertEquals(credentials, Set.of("Service-1", "Service-2"));
+ } finally {
+ executor.shutdownNow();
+ connectionManager.close();
+ }
+ }
+
+ @Test
+ public void testAuthProviderFailureDoesNotSubmitPartialBatch()
+ throws Exception {
+ AtomicInteger headerResolutions = new AtomicInteger();
+ AuthProvider authProvider = new AuthProvider() {
+ @Override
+ public Map<String, Object> getRequestHeaders() {
+ if (headerResolutions.incrementAndGet() == 2) {
+ throw new IllegalStateException("credential refresh failed");
+ }
+ return Map.of("Authorization", "Service");
+ }
+
+ @Override
+ public String getTaskToken() {
+ return null;
+ }
+ };
+ AtomicInteger submittedTasks = new AtomicInteger();
+ Executor executor = command -> submittedTasks.incrementAndGet();
+ PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
+ try {
+ MultiHttpRequest multiHttpRequest = new MultiHttpRequest(executor, connectionManager);
+ Assert.expectThrows(IllegalStateException.class,
+ () -> multiHttpRequest.execute(List.of(Pair.of("http://localhost/foo", null),
+ Pair.of("http://localhost/bar", null)), null, authProvider, TIMEOUT_MS, "GET", HttpGet::new));
+ Assert.assertEquals(submittedTasks.get(), 0);
+ } finally {
+ connectionManager.close();
+ }
+ }
+
@Test
public void testBufferedExecutionSubmitsAndPollsRequests()
throws Exception {
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/utils/FileUploadDownloadClientTest.java b/pinot-common/src/test/java/org/apache/pinot/common/utils/FileUploadDownloadClientTest.java
index c96d17ced29..e0c1fea1fc7 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/utils/FileUploadDownloadClientTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/utils/FileUploadDownloadClientTest.java
@@ -31,11 +31,15 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;
import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.io.IOUtils;
import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.http.NameValuePair;
import org.apache.hc.core5.http.message.BasicHeader;
+import org.apache.pinot.common.auth.StaticTokenAuthProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.restlet.resources.TableLLCSegmentUploadResponse;
import org.apache.pinot.common.utils.FileUploadDownloadClient.FileUploadType;
import org.apache.pinot.common.utils.http.HttpClient;
import org.apache.pinot.spi.utils.CommonConstants;
@@ -52,13 +56,17 @@ public class FileUploadDownloadClientTest {
private static final int TEST_PORT = new Random().nextInt(10000) + 10000;
private static final String TEST_URI = "http://testhost/segments/testSegment";
private static final String TEST_CRYPTER = "testCrypter";
+ private static final String TEST_AUTH_TOKEN = "Bearer test-token";
+ private static final String TEST_DOWNLOAD_URL = "file:///tmp/testSegment";
private HttpServer _testServer;
+ private final AtomicInteger _authenticatedRequests = new AtomicInteger();
@BeforeClass
public void setUp()
throws Exception {
_testServer = HttpServer.create(new InetSocketAddress(TEST_PORT), 0);
_testServer.createContext("/v2/segments", new TestSegmentUploadHandler());
+ _testServer.createContext("/segmentStore", new AuthenticatedSegmentStoreHandler());
_testServer.setExecutor(null); // creates a default executor
_testServer.start();
}
@@ -98,6 +106,37 @@ public class FileUploadDownloadClientTest {
}
}
+ private class AuthenticatedSegmentStoreHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange httpExchange)
+ throws IOException {
+ Assert.assertEquals(httpExchange.getRequestHeaders().getFirst("Authorization"), TEST_AUTH_TOKEN);
+ _authenticatedRequests.incrementAndGet();
+
+ String response;
+ switch (httpExchange.getRequestURI().getPath()) {
+ case "/segmentStore/plain":
+ response = TEST_DOWNLOAD_URL;
+ break;
+ case "/segmentStore/llc":
+ response = JsonUtils.objectToString(
+ new TableLLCSegmentUploadResponse("testSegment", 123L, 456L, TEST_DOWNLOAD_URL));
+ break;
+ case "/segmentStore/metadata":
+ SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata("testSegment");
+ segmentZKMetadata.setDownloadUrl(TEST_DOWNLOAD_URL);
+ response = segmentZKMetadata.toJsonString();
+ break;
+ default:
+ throw new IllegalArgumentException("Unexpected test path: " + httpExchange.getRequestURI().getPath());
+ }
+ httpExchange.sendResponseHeaders(HttpStatus.SC_OK, response.length());
+ try (OutputStream outputStream = httpExchange.getResponseBody()) {
+ outputStream.write(response.getBytes());
+ }
+ }
+ }
+
@Test
public void testSendFileWithUriAndCrypter()
throws Exception {
@@ -138,6 +177,23 @@ public class FileUploadDownloadClientTest {
Assert.assertEquals(new URI("https://myhost:9443/segments/batchUpload"), uriWithEndpoint);
}
+ @Test
+ public void testAuthenticatedServerUploadRequests()
+ throws Exception {
+ StaticTokenAuthProvider authProvider = new StaticTokenAuthProvider(TEST_AUTH_TOKEN);
+ String baseUrl = "http://" + TEST_HOST + ":" + TEST_PORT + "/segmentStore";
+ try (FileUploadDownloadClient fileUploadDownloadClient = new FileUploadDownloadClient()) {
+ Assert.assertEquals(fileUploadDownloadClient.uploadToSegmentStore(baseUrl + "/plain", authProvider),
+ TEST_DOWNLOAD_URL);
+ Assert.assertEquals(fileUploadDownloadClient.uploadLLCToSegmentStore(baseUrl + "/llc", authProvider)
+ .getDownloadUrl(), TEST_DOWNLOAD_URL);
+ Assert.assertEquals(
+ fileUploadDownloadClient.uploadLLCToSegmentStoreWithZKMetadata(baseUrl + "/metadata", authProvider)
+ .getDownloadUrl(), TEST_DOWNLOAD_URL);
+ }
+ Assert.assertEquals(_authenticatedRequests.get(), 3);
+ }
+
@AfterClass
public void shutDown() {
_testServer.stop(0);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
index 415e96d7201..747068a72fd 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
@@ -60,6 +60,8 @@ public class ControllerConf extends PinotConfiguration {
public static final String CONTROLLER_BROKER_PORT_OVERRIDE = "controller.broker.port.override";
public static final String CONTROLLER_BROKER_TLS_PREFIX = "controller.broker.tls";
public static final String CONTROLLER_BROKER_AUTH_PREFIX = "controller.broker.auth";
+ /// Namespace for the service credentials used by the controller when invoking Server admin APIs.
+ public static final String CONTROLLER_SERVER_ADMIN_AUTH_PREFIX = "controller.server.admin.auth";
public static final String CONTROLLER_TLS_PREFIX = "controller.tls";
public static final String CONTROLLER_HOST = "controller.host";
public static final String CONTROLLER_PORT = "controller.port";
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/DebugResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/DebugResource.java
index 436e1478010..a8308f1e7f2 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/DebugResource.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/DebugResource.java
@@ -277,7 +277,8 @@ public class DebugResource {
}
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverUrls, tableNameWithType, false, serverRequestTimeoutMs);
@@ -476,7 +477,8 @@ public class DebugResource {
}
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverUrls, tableNameWithType, false, timeoutMs);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotControllerLogger.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotControllerLogger.java
index 52797cb69c5..922c33a1563 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotControllerLogger.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotControllerLogger.java
@@ -54,6 +54,7 @@ import org.apache.commons.collections4.MapUtils;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.core5.http.HttpVersion;
import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
+import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.utils.FileUploadDownloadClient;
import org.apache.pinot.common.utils.LoggerUtils;
import org.apache.pinot.common.utils.SimpleHttpResponse;
@@ -66,6 +67,7 @@ import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.core.auth.Actions;
import org.apache.pinot.core.auth.Authorize;
import org.apache.pinot.core.auth.TargetType;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
import static org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_KEY;
@@ -183,8 +185,9 @@ public class PinotControllerLogger {
@ApiParam(value = "Instance Name", required = true) @PathParam("instanceName") String instanceName) {
try {
URI uri = new URI(getInstanceBaseUri(instanceName) + "/loggers/files");
- Map<String, String> headers = new HashMap<>();
- if (authorization != null) {
+ Map<String, String> headers =
+ new HashMap<>(getServerAdminAuthHeaders(instanceName));
+ if (headers.isEmpty() && authorization != null) {
headers.put(HttpHeaders.AUTHORIZATION, authorization);
}
SimpleHttpResponse simpleHttpResponse = _fileUploadDownloadClient.getHttpClient().sendGetRequest(uri, headers);
@@ -214,14 +217,18 @@ public class PinotControllerLogger {
URI uri = UriBuilder.fromUri(getInstanceBaseUri(instanceName)).path("/loggers/download")
.queryParam("filePath", filePath).build();
ClassicRequestBuilder requestBuilder = ClassicRequestBuilder.get(uri).setVersion(HttpVersion.HTTP_1_1);
+ Map<String, String> serviceAuthHeaders = getServerAdminAuthHeaders(instanceName);
if (MapUtils.isNotEmpty(headers)) {
for (Map.Entry<String, String> header : headers.entrySet()) {
- requestBuilder.addHeader(header.getKey(), header.getValue());
+ if (serviceAuthHeaders.isEmpty() || !HttpHeaders.AUTHORIZATION.equalsIgnoreCase(header.getKey())) {
+ requestBuilder.addHeader(header.getKey(), header.getValue());
+ }
}
}
- if (authorization != null) {
+ if (serviceAuthHeaders.isEmpty() && authorization != null) {
requestBuilder.addHeader(HttpHeaders.AUTHORIZATION, authorization);
}
+ serviceAuthHeaders.forEach(requestBuilder::setHeader);
StreamingOutput streamingOutput = output -> {
try (CloseableHttpResponse response = _fileUploadDownloadClient.getHttpClient().execute(requestBuilder.build());
@@ -244,4 +251,9 @@ public class PinotControllerLogger {
private String getInstanceBaseUri(String instanceName) {
return InstanceUtils.getInstanceBaseUri(_pinotHelixResourceManager.getHelixInstanceConfig(instanceName));
}
+
+ private Map<String, String> getServerAdminAuthHeaders(String instanceName) {
+ return InstanceTypeUtils.isServer(instanceName)
+ ? AuthProviderUtils.makeAuthHeadersMap(_pinotHelixResourceManager.getServerAdminAuthProvider()) : Map.of();
+ }
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableInstances.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableInstances.java
index 51b9dd2e61b..905dda599aa 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableInstances.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableInstances.java
@@ -231,7 +231,8 @@ public class PinotTableInstances {
SimpleHttpResponse simpleHttpResponse;
try {
simpleHttpResponse =
- HttpClient.wrapAndThrowHttpException(HttpClient.getInstance().sendDeleteRequest(URI.create(fullUrl)));
+ HttpClient.wrapAndThrowHttpException(HttpClient.getInstance().sendDeleteRequest(URI.create(fullUrl), Map.of(),
+ _pinotHelixResourceManager.getServerAdminAuthProvider()));
} catch (Exception e) {
throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
index e69de9e181e..aff800e9b4c 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
@@ -1430,7 +1430,8 @@ public class PinotTableRestletResource {
BiMap<String, String> serverEndPoints =
_pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverToSegments.keySet());
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, serverEndPoints);
+ new CompletionServiceHelper(_executor, _connectionManager, serverEndPoints,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
List<String> serverUrls = new ArrayList<>();
BiMap<String, String> endpointsToServers = serverEndPoints.inverse();
@@ -1598,7 +1599,8 @@ public class PinotTableRestletResource {
BiMap<String, String> serverEndPoints =
_pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverToSegments.keySet());
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, serverEndPoints);
+ new CompletionServiceHelper(_executor, _connectionManager, serverEndPoints,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
List<String> serverUrls = new ArrayList<>();
BiMap<String, String> endpointsToServers = serverEndPoints.inverse();
for (String endpoint : endpointsToServers.keySet()) {
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/ServerTableSizeReader.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/ServerTableSizeReader.java
index 77ecb1bb829..73ca27b787b 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/ServerTableSizeReader.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/ServerTableSizeReader.java
@@ -25,10 +25,12 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
+import javax.annotation.Nullable;
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.pinot.common.restlet.resources.SegmentSizeInfo;
import org.apache.pinot.common.restlet.resources.TableSizeInfo;
import org.apache.pinot.controller.util.CompletionServiceHelper;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.utils.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,10 +43,18 @@ public class ServerTableSizeReader {
private final Executor _executor;
private final HttpClientConnectionManager _connectionManager;
+ @Nullable
+ private final AuthProvider _authProvider;
public ServerTableSizeReader(Executor executor, HttpClientConnectionManager connectionManager) {
+ this(executor, connectionManager, null);
+ }
+
+ public ServerTableSizeReader(Executor executor, HttpClientConnectionManager connectionManager,
+ @Nullable AuthProvider authProvider) {
_executor = executor;
_connectionManager = connectionManager;
+ _authProvider = authProvider;
}
/// Reads server segment sizes without compression statistics.
@@ -100,7 +110,7 @@ public class ServerTableSizeReader {
// Helper service to run a httpget call to the server
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers, _authProvider);
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverUrls, tableNameWithType, false, timeoutMs,
"get segment size info from servers");
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index 61c570ed917..70703680f7a 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -92,6 +92,7 @@ import org.apache.helix.zookeeper.impl.client.ZkClient;
import org.apache.pinot.common.assignment.InstanceAssignmentConfigUtils;
import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.assignment.InstancePartitionsUtils;
+import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.common.config.provider.ZkTableCache;
import org.apache.pinot.common.exception.InvalidConfigException;
@@ -171,6 +172,7 @@ import org.apache.pinot.materializedview.metadata.MaterializedViewDefinitionMeta
import org.apache.pinot.materializedview.metadata.MaterializedViewRuntimeMetadataUtils;
import org.apache.pinot.segment.local.utils.TableConfigUtils;
import org.apache.pinot.segment.spi.SegmentMetadata;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.config.DatabaseConfig;
import org.apache.pinot.spi.config.instance.Instance;
import org.apache.pinot.spi.config.instance.InstanceConfigValidatorRegistry;
@@ -244,6 +246,7 @@ public class PinotHelixResourceManager {
private final boolean _enableTieredSegmentAssignment;
@Nullable
private final ControllerConf _controllerConf;
+ private final AuthProvider _serverAdminAuthProvider;
private HelixManager _helixZkManager;
private HelixAdmin _helixAdmin;
@@ -274,6 +277,8 @@ public class PinotHelixResourceManager {
_deletedSegmentsRetentionInDays = deletedSegmentsRetentionInDays;
_enableTieredSegmentAssignment = enableTieredSegmentAssignment;
_controllerConf = controllerConf;
+ _serverAdminAuthProvider =
+ AuthProviderUtils.extractAuthProvider(controllerConf, ControllerConf.CONTROLLER_SERVER_ADMIN_AUTH_PREFIX);
_instanceAdminEndpointCache =
CacheBuilder.newBuilder().expireAfterWrite(CACHE_ENTRY_EXPIRE_TIME_HOURS, TimeUnit.HOURS)
.build(new CacheLoader<>() {
@@ -4264,6 +4269,13 @@ public class PinotHelixResourceManager {
return endpointToInstance;
}
+ /// Returns the service identity provider used for outbound Server admin API requests.
+ ///
+ /// Callers must resolve the headers for each request so providers that rotate credentials are honored.
+ public AuthProvider getServerAdminAuthProvider() {
+ return _serverAdminAuthProvider;
+ }
+
/// Helper method to return a list of tables that exists and matches the given table name and type, or throws
/// [ControllerApplicationException] if no table found.
///
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
index b48a41f7409..9303c6e5bcb 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
@@ -2264,7 +2264,8 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
LOGGER.info("Asking server to upload segment: {} by path: {}", segmentName, serverUploadRequestUrl);
try {
SegmentZKMetadata uploadedMetadata =
- _fileUploadDownloadClient.uploadLLCToSegmentStoreWithZKMetadata(serverUploadRequestUrl);
+ _fileUploadDownloadClient.uploadLLCToSegmentStoreWithZKMetadata(serverUploadRequestUrl,
+ _helixResourceManager.getServerAdminAuthProvider());
handleMetadataUpload(rawTableName, segmentName, segmentZKMetadata, uploadedMetadata, pinotFS);
return;
} catch (Exception e) {
@@ -2279,7 +2280,8 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
LOGGER.info("Asking server to upload segment: {} by path: {}", segmentName, serverUploadRequestUrl);
try {
TableLLCSegmentUploadResponse response =
- _fileUploadDownloadClient.uploadLLCToSegmentStore(serverUploadRequestUrl);
+ _fileUploadDownloadClient.uploadLLCToSegmentStore(serverUploadRequestUrl,
+ _helixResourceManager.getServerAdminAuthProvider());
handleLLCUpload(segmentName, rawTableName, segmentZKMetadata, response, pinotFS);
return;
} catch (Exception e) {
@@ -2290,7 +2292,8 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
serverUploadRequestUrl = getUploadUrl(uri, "upload");
LOGGER.info("Asking server to upload segment: {} by path: {}", segmentName, serverUploadRequestUrl);
try {
- String segmentLocation = _fileUploadDownloadClient.uploadToSegmentStore(serverUploadRequestUrl);
+ String segmentLocation = _fileUploadDownloadClient.uploadToSegmentStore(serverUploadRequestUrl,
+ _helixResourceManager.getServerAdminAuthProvider());
handleBasicUpload(rawTableName, segmentName, segmentZKMetadata, segmentLocation, pinotFS);
} catch (Exception e) {
throw new RuntimeException("Failed to ask server to upload segment: " + segmentName, e);
@@ -3099,7 +3102,8 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
URI reingestUri = FileUploadDownloadClient.getURI(scheme, serverHost, Integer.parseInt(serverPort),
REINGEST_SEGMENT_PATH + "/" + segmentName);
- HttpClient.wrapAndThrowHttpException(HttpClient.getInstance().sendJsonPostRequest(reingestUri, ""));
+ HttpClient.wrapAndThrowHttpException(HttpClient.getInstance().sendJsonPostRequest(reingestUri, "", Map.of(),
+ _helixResourceManager.getServerAdminAuthProvider()));
}
/// Picks one server among a set of servers that are supposed to host the segment,
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadStatusReporter.java b/pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadStatusReporter.java
index 5c975c86263..8e0019bfe91 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadStatusReporter.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadStatusReporter.java
@@ -166,7 +166,8 @@ public class PinotTableReloadStatusReporter {
final List<String> serverUrls = getServerUrls(serverEndPoints, reloadJobMetadata, serverToSegments);
final CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, serverEndPoints);
+ new CompletionServiceHelper(_executor, _connectionManager, serverEndPoints,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
final CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverUrls, null, true, 10000);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/CompletionServiceHelper.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/CompletionServiceHelper.java
index dd48863c29d..990fa8f4f3b 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/CompletionServiceHelper.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/CompletionServiceHelper.java
@@ -29,11 +29,14 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.hc.core5.http.io.entity.StringEntity;
+import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.http.MultiHttpRequest;
import org.apache.pinot.common.http.MultiHttpRequestBufferedResponse;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,12 +52,21 @@ public class CompletionServiceHelper {
private final Executor _executor;
private final HttpClientConnectionManager _httpConnectionManager;
private final BiMap<String, String> _endpointsToServers;
+ @Nullable
+ private final AuthProvider _authProvider;
public CompletionServiceHelper(Executor executor, HttpClientConnectionManager httpConnectionManager,
BiMap<String, String> endpointsToServers) {
+ this(executor, httpConnectionManager, endpointsToServers, null);
+ }
+
+ /// Creates a helper whose requests resolve service authentication headers from the provider at request time.
+ public CompletionServiceHelper(Executor executor, HttpClientConnectionManager httpConnectionManager,
+ BiMap<String, String> endpointsToServers, @Nullable AuthProvider authProvider) {
_executor = executor;
_httpConnectionManager = httpConnectionManager;
_endpointsToServers = endpointsToServers;
+ _authProvider = authProvider;
}
public CompletionServiceResponse doMultiGetRequest(List<String> serverURLs, String tableNameWithType,
@@ -78,9 +90,7 @@ public class CompletionServiceHelper {
public CompletionServiceResponse doMultiGetRequest(List<String> serverURLs, String tableNameWithType,
boolean multiRequestPerServer, @Nullable Map<String, String> requestHeaders, int timeoutMs,
@Nullable String useCase) {
- MultiHttpRequest.BufferedRequestExecution execution =
- new MultiHttpRequest(_executor, _httpConnectionManager).executeGetBuffered(serverURLs, requestHeaders,
- timeoutMs);
+ MultiHttpRequest.BufferedRequestExecution execution = executeGetBuffered(serverURLs, requestHeaders, timeoutMs);
return collectBufferedResponse(tableNameWithType, serverURLs.size(), execution, multiRequestPerServer, useCase);
}
@@ -88,8 +98,7 @@ public class CompletionServiceHelper {
/// [System#nanoTime()] clock is reached.
public CompletionServiceResponse doMultiGetRequestUntil(List<String> serverURLs, String tableNameWithType,
boolean multiRequestPerServer, int timeoutMs, long nanoTimeDeadline, @Nullable String useCase) {
- MultiHttpRequest.BufferedRequestExecution execution =
- new MultiHttpRequest(_executor, _httpConnectionManager).executeGetBuffered(serverURLs, null, timeoutMs);
+ MultiHttpRequest.BufferedRequestExecution execution = executeGetBuffered(serverURLs, null, timeoutMs);
return collectBufferedResponseUntil(tableNameWithType, serverURLs.size(), execution, multiRequestPerServer, useCase,
nanoTimeDeadline);
}
@@ -116,8 +125,9 @@ public class CompletionServiceHelper {
for (Pair<String, String> requestAndBody : serverURLsAndRequestBodies) {
HttpPost request = new HttpPost(requestAndBody.getLeft());
request.setEntity(new StringEntity(requestAndBody.getRight()));
- if (requestHeaders != null) {
- requestHeaders.forEach(request::setHeader);
+ Map<String, String> mergedRequestHeaders = getRequestHeaders(requestHeaders);
+ if (mergedRequestHeaders != null) {
+ mergedRequestHeaders.forEach(request::setHeader);
}
execution.submit(request);
}
@@ -129,6 +139,41 @@ public class CompletionServiceHelper {
multiRequestPerServer, useCase);
}
+ @Nullable
+ private Map<String, String> getRequestHeaders(@Nullable Map<String, String> requestHeaders) {
+ Map<String, String> authHeaders = AuthProviderUtils.makeAuthHeadersMap(_authProvider);
+ if (authHeaders.isEmpty()) {
+ return requestHeaders;
+ }
+ Map<String, String> mergedRequestHeaders =
+ requestHeaders != null ? new HashMap<>(requestHeaders) : new HashMap<>();
+ // A configured service identity takes precedence over a caller-supplied end-user credential.
+ mergedRequestHeaders.keySet()
+ .removeIf(header -> authHeaders.keySet().stream().anyMatch(header::equalsIgnoreCase));
+ mergedRequestHeaders.putAll(authHeaders);
+ return mergedRequestHeaders;
+ }
+
+ private MultiHttpRequest.BufferedRequestExecution executeGetBuffered(List<String> serverURLs,
+ @Nullable Map<String, String> requestHeaders, int timeoutMs) {
+ MultiHttpRequest.BufferedRequestExecution execution =
+ new MultiHttpRequest(_executor, _httpConnectionManager).createBufferedExecution(timeoutMs);
+ try {
+ for (String serverURL : serverURLs) {
+ HttpGet request = new HttpGet(serverURL);
+ Map<String, String> mergedRequestHeaders = getRequestHeaders(requestHeaders);
+ if (mergedRequestHeaders != null) {
+ mergedRequestHeaders.forEach(request::setHeader);
+ }
+ execution.submit(request);
+ }
+ return execution;
+ } catch (RuntimeException | Error e) {
+ execution.close();
+ throw e;
+ }
+ }
+
private CompletionServiceResponse collectBufferedResponse(String tableNameWithType, int size,
MultiHttpRequest.BufferedRequestExecution execution, boolean multiRequestPerServer, @Nullable String useCase) {
return collectBufferedResponse(tableNameWithType, size, execution, multiRequestPerServer, useCase, 0, false);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java
index 7cc6ebba270..2cd851fe348 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java
@@ -107,7 +107,8 @@ public class ConsumingSegmentInfoReader {
}
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverUrls, tableNameWithType, false, timeoutMs);
Map<String, List<SegmentConsumerInfo>> serverToConsumingSegmentInfoList = new HashMap<>();
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerCompressionStatsReader.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerCompressionStatsReader.java
index 4133cbde189..ab7bd2ddcfa 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerCompressionStatsReader.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerCompressionStatsReader.java
@@ -44,6 +44,7 @@ import org.apache.hc.client5.http.config.RequestConfig;
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.hc.core5.http.io.entity.StringEntity;
import org.apache.hc.core5.util.Timeout;
+import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.compression.ColumnCompressionStatsAccumulator;
import org.apache.pinot.common.http.MultiHttpRequest;
import org.apache.pinot.common.http.MultiHttpRequestBufferedResponse;
@@ -53,6 +54,7 @@ import org.apache.pinot.common.restlet.resources.CompressionStatsSummary;
import org.apache.pinot.common.restlet.resources.SegmentCompressionStatsContribution;
import org.apache.pinot.common.restlet.resources.ServerCompressionStatsResponse;
import org.apache.pinot.common.restlet.resources.TableSegments;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.UrlBuilderUtils;
import org.slf4j.Logger;
@@ -75,10 +77,18 @@ final class ServerCompressionStatsReader {
private final Executor _executor;
private final HttpClientConnectionManager _connectionManager;
+ @Nullable
+ private final AuthProvider _authProvider;
ServerCompressionStatsReader(Executor executor, HttpClientConnectionManager connectionManager) {
+ this(executor, connectionManager, null);
+ }
+
+ ServerCompressionStatsReader(Executor executor, HttpClientConnectionManager connectionManager,
+ @Nullable AuthProvider authProvider) {
_executor = executor;
_connectionManager = connectionManager;
+ _authProvider = authProvider;
}
CompressionStatsResult read(String tableNameWithType, Map<String, List<String>> serverToSegments,
@@ -217,6 +227,7 @@ final class ServerCompressionStatsReader {
batch._endpoint);
HttpPost request = new HttpPost(url);
request.setHeader("Content-Type", "application/json");
+ AuthProviderUtils.makeAuthHeadersMap(_authProvider).forEach(request::setHeader);
request.setConfig(RequestConfig.custom()
.setConnectionRequestTimeout(Timeout.of(timeoutMs, TimeUnit.MILLISECONDS))
.setResponseTimeout(Timeout.of(timeoutMs, TimeUnit.MILLISECONDS))
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java
index 324b2df9ee6..87e9a05f6c5 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java
@@ -38,10 +38,12 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Invocation;
import javax.ws.rs.core.MediaType;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
+import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.restlet.resources.ServerTableMetadataInfo;
import org.apache.pinot.common.restlet.resources.TableMetadataInfo;
import org.apache.pinot.common.restlet.resources.TableSegments;
@@ -49,6 +51,7 @@ import org.apache.pinot.common.restlet.resources.ValidDocIdsBitmapResponse;
import org.apache.pinot.common.restlet.resources.ValidDocIdsMetadataInfo;
import org.apache.pinot.controller.api.resources.TableStaleSegmentResponse;
import org.apache.pinot.segment.local.data.manager.StaleSegment;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.UrlBuilderUtils;
import org.glassfish.jersey.client.ClientConfig;
@@ -68,15 +71,23 @@ public class ServerSegmentMetadataReader {
private final Executor _executor;
private final HttpClientConnectionManager _connectionManager;
+ @Nullable
+ private final AuthProvider _authProvider;
public ServerSegmentMetadataReader() {
- _executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
- _connectionManager = new PoolingHttpClientConnectionManager();
+ this(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()),
+ new PoolingHttpClientConnectionManager(), null);
}
public ServerSegmentMetadataReader(Executor executor, HttpClientConnectionManager connectionManager) {
+ this(executor, connectionManager, null);
+ }
+
+ public ServerSegmentMetadataReader(Executor executor, HttpClientConnectionManager connectionManager,
+ @Nullable AuthProvider authProvider) {
_executor = executor;
_connectionManager = connectionManager;
+ _authProvider = authProvider;
}
/// This method is called when the API request is to fetch aggregated segment metadata for all segments of the table.
@@ -112,7 +123,7 @@ public class ServerSegmentMetadataReader {
// Helper service to run a http get call to the server
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers, _authProvider);
CompletionServiceHelper.CompletionServiceResponse serviceResponse = compressionStatsEnabled
? completionServiceHelper.doMultiGetRequestUntil(serverUrls, tableNameWithType, false, timeoutMs, deadlineNanos,
"aggregate segment metadata")
@@ -176,7 +187,7 @@ public class ServerSegmentMetadataReader {
totalNumRows /= numReplica;
ServerCompressionStatsReader.CompressionStatsResult compressionStats = compressionStatsEnabled
- ? new ServerCompressionStatsReader(_executor, _connectionManager).read(tableNameWithType,
+ ? new ServerCompressionStatsReader(_executor, _connectionManager, _authProvider).read(tableNameWithType,
serverToSegmentsMap, serverEndPoints, columns, includeColumnCompressionStats, deadlineNanos) : null;
TableMetadataInfo aggregateTableMetadataInfo =
@@ -210,7 +221,7 @@ public class ServerSegmentMetadataReader {
}
BiMap<String, String> endpointsToServers = endpoints.inverse();
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers, _authProvider);
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, true, timeoutMs);
List<String> segmentsMetadata = new ArrayList<>();
@@ -246,7 +257,7 @@ public class ServerSegmentMetadataReader {
}
BiMap<String, String> endpointsToServers = endpoints.inverse();
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers, _authProvider);
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, true, timeoutMs);
List<String> serversNeedReloadResponses = new ArrayList<>();
@@ -321,7 +332,7 @@ public class ServerSegmentMetadataReader {
// request the urls from the servers
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers, _authProvider);
Map<String, String> requestHeaders = Map.of("Content-Type", "application/json");
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
@@ -404,9 +415,9 @@ public class ServerSegmentMetadataReader {
clientConfig.property(ClientProperties.CONNECT_TIMEOUT, timeoutMs);
clientConfig.property(ClientProperties.READ_TIMEOUT, timeoutMs);
- ValidDocIdsBitmapResponse response =
- ClientBuilder.newClient(clientConfig).target(url).request(MediaType.APPLICATION_JSON)
- .get(ValidDocIdsBitmapResponse.class);
+ Invocation.Builder request = ClientBuilder.newClient(clientConfig).target(url).request(MediaType.APPLICATION_JSON);
+ AuthProviderUtils.makeAuthHeadersMap(_authProvider).forEach(request::header);
+ ValidDocIdsBitmapResponse response = request.get(ValidDocIdsBitmapResponse.class);
Preconditions.checkNotNull(response, "Unable to retrieve validDocIdsBitmap from %s", url);
return response;
}
@@ -420,7 +431,7 @@ public class ServerSegmentMetadataReader {
}
BiMap<String, String> endpointsToServers = endpoints.inverse();
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers, _authProvider);
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, false, timeoutMs);
Map<String, TableStaleSegmentResponse> serverResponses = new HashMap<>();
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerTableTierReader.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerTableTierReader.java
index 8552861be36..ca167b1820a 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerTableTierReader.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerTableTierReader.java
@@ -28,6 +28,7 @@ import java.util.concurrent.Executor;
import javax.annotation.Nullable;
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.pinot.common.restlet.resources.TableTierInfo;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.utils.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,10 +41,18 @@ public class ServerTableTierReader {
private final Executor _executor;
private final HttpClientConnectionManager _connectionManager;
+ @Nullable
+ private final AuthProvider _authProvider;
public ServerTableTierReader(Executor executor, HttpClientConnectionManager connectionManager) {
+ this(executor, connectionManager, null);
+ }
+
+ public ServerTableTierReader(Executor executor, HttpClientConnectionManager connectionManager,
+ @Nullable AuthProvider authProvider) {
_executor = executor;
_connectionManager = connectionManager;
+ _authProvider = authProvider;
}
public Map<String, TableTierInfo> getTableTierInfoFromServers(BiMap<String, String> serverEndPoints,
@@ -64,7 +73,7 @@ public class ServerTableTierReader {
}
LOGGER.debug("Getting table tier info with serverUrls: {}", serverUrls);
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers, _authProvider);
CompletionServiceHelper.CompletionServiceResponse serviceResponse =
completionServiceHelper.doMultiGetRequest(serverUrls, tableNameWithType, false, timeoutMs);
Map<String, TableTierInfo> serverToTableTierInfoMap = new HashMap<>();
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableMetadataReader.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableMetadataReader.java
index bb75b8c1bfc..7b6389862a4 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableMetadataReader.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableMetadataReader.java
@@ -108,7 +108,8 @@ public class TableMetadataReader {
int timeoutMs, Set<String> serverInstanceSet) throws InvalidConfigException {
BiMap<String, String> endpoints = _pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverInstanceSet);
ServerSegmentMetadataReader serverSegmentMetadataReader =
- new ServerSegmentMetadataReader(_executor, _connectionManager);
+ new ServerSegmentMetadataReader(_executor, _connectionManager,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
return serverSegmentMetadataReader.getCheckReloadSegmentsFromServer(tableNameWithType, serverInstanceSet, endpoints,
timeoutMs);
}
@@ -137,7 +138,8 @@ public class TableMetadataReader {
private JsonNode fetchAndAggregateMetadata(List<String> urls, BiMap<String, String> endpoints, boolean perSegmentJson,
String tableNameWithType, int timeoutMs)
throws InvalidConfigException, IOException {
- CompletionServiceHelper cs = new CompletionServiceHelper(_executor, _connectionManager, endpoints);
+ CompletionServiceHelper cs = new CompletionServiceHelper(_executor, _connectionManager, endpoints,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
CompletionServiceHelper.CompletionServiceResponse resp =
cs.doMultiGetRequest(urls, tableNameWithType, perSegmentJson, timeoutMs);
// all requests will fail if new server endpoint is not available
@@ -194,7 +196,8 @@ public class TableMetadataReader {
BiMap<String, String> endpoints =
_pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverToSegs.keySet());
ServerSegmentMetadataReader reader =
- new ServerSegmentMetadataReader(_executor, _connectionManager);
+ new ServerSegmentMetadataReader(_executor, _connectionManager,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
// try table level endpoint first
try {
@@ -225,7 +228,8 @@ public class TableMetadataReader {
BiMap<String, String> endpoints = _pinotHelixResourceManager.getDataInstanceAdminEndpoints(servers);
ServerSegmentMetadataReader serverSegmentMetadataReader =
- new ServerSegmentMetadataReader(_executor, _connectionManager);
+ new ServerSegmentMetadataReader(_executor, _connectionManager,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
List<String> segmentsMetadata =
serverSegmentMetadataReader.getSegmentMetadataFromServer(tableNameWithType, serverToSegments, endpoints,
@@ -260,7 +264,8 @@ public class TableMetadataReader {
BiMap<String, String> endpoints =
_pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverToSegments.keySet());
ServerSegmentMetadataReader serverSegmentMetadataReader =
- new ServerSegmentMetadataReader(_executor, _connectionManager);
+ new ServerSegmentMetadataReader(_executor, _connectionManager,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
TableMetadataInfo aggregateTableMetadataInfo =
serverSegmentMetadataReader.getAggregatedTableMetadataFromServer(tableNameWithType, endpoints, columns,
@@ -278,7 +283,8 @@ public class TableMetadataReader {
BiMap<String, String> endpoints =
_pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverToSegments.keySet());
ServerSegmentMetadataReader serverSegmentMetadataReader =
- new ServerSegmentMetadataReader(_executor, _connectionManager);
+ new ServerSegmentMetadataReader(_executor, _connectionManager,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
List<ValidDocIdsMetadataInfo> aggregateTableMetadataInfo =
serverSegmentMetadataReader.getValidDocIdsMetadataFromServer(tableNameWithType, serverToSegments, endpoints,
@@ -294,7 +300,8 @@ public class TableMetadataReader {
Set<String> serverInstanceSet = new HashSet<>(serverInstances);
BiMap<String, String> endpoints = _pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverInstanceSet);
ServerSegmentMetadataReader serverSegmentMetadataReader =
- new ServerSegmentMetadataReader(_executor, _connectionManager);
+ new ServerSegmentMetadataReader(_executor, _connectionManager,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
return serverSegmentMetadataReader.getStaleSegmentsFromServer(tableNameWithType, serverInstanceSet, endpoints,
timeoutMs);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableSizeReader.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableSizeReader.java
index 3f4ed17ae00..46b748c78a1 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableSizeReader.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableSizeReader.java
@@ -398,7 +398,8 @@ public class TableSizeReader {
long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMs);
Map<String, List<String>> serverToSegmentsMap = _helixResourceManager.getServerToSegmentsMap(tableNameWithType,
null, includeReplacedSegments);
- ServerTableSizeReader serverTableSizeReader = new ServerTableSizeReader(_executor, _connectionManager);
+ ServerTableSizeReader serverTableSizeReader = new ServerTableSizeReader(_executor, _connectionManager,
+ _helixResourceManager.getServerAdminAuthProvider());
BiMap<String, String> endpoints = _helixResourceManager.getDataInstanceAdminEndpoints(serverToSegmentsMap.keySet());
Map<String, TableSizeInfo> serverToTableSizeInfoMap =
serverTableSizeReader.getTableSizeInfoFromServers(endpoints, tableNameWithType, timeoutMs);
@@ -480,7 +481,8 @@ public class TableSizeReader {
if (compressionStatsEnabled) {
ServerCompressionStatsReader compressionStatsReader =
- new ServerCompressionStatsReader(_executor, _connectionManager);
+ new ServerCompressionStatsReader(_executor, _connectionManager,
+ _helixResourceManager.getServerAdminAuthProvider());
ServerCompressionStatsReader.CompressionStatsResult compressionStats = includeSelectedCompressionStats
? compressionStatsReader.readWithSelectedSegments(tableNameWithType, serverToSegmentsMap, endpoints, null,
includeColumnCompressionStats, deadlineNanos)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableTierReader.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableTierReader.java
index 32c3213e503..83721e79d53 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableTierReader.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableTierReader.java
@@ -80,7 +80,8 @@ public class TableTierReader {
}
}
BiMap<String, String> endpoints = _helixResourceManager.getDataInstanceAdminEndpoints(serverToSegmentsMap.keySet());
- ServerTableTierReader serverTableTierReader = new ServerTableTierReader(_executor, _connectionManager);
+ ServerTableTierReader serverTableTierReader = new ServerTableTierReader(_executor, _connectionManager,
+ _helixResourceManager.getServerAdminAuthProvider());
Map<String, TableTierInfo> serverToTableTierInfoMap =
serverTableTierReader.getTableTierInfoFromServers(endpoints, tableNameWithType, segmentName, timeoutMs);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/ResourceUtilizationChecker.java b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/ResourceUtilizationChecker.java
index a73679ecb7f..f8c2a032e14 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/ResourceUtilizationChecker.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/ResourceUtilizationChecker.java
@@ -75,7 +75,8 @@ public class ResourceUtilizationChecker extends BasePeriodicTask {
BiMap<String, String> instanceAdminEndpoints = _helixResourceManager.getDataInstanceAdminEndpoints(instances);
BiMap<String, String> endpointsToInstances = instanceAdminEndpoints.inverse();
CompletionServiceHelper completionServiceHelper =
- new CompletionServiceHelper(_executor, _connectionManager, endpointsToInstances);
+ new CompletionServiceHelper(_executor, _connectionManager, endpointsToInstances,
+ _helixResourceManager.getServerAdminAuthProvider());
for (UtilizationChecker utilizationChecker : _utilizationCheckers) {
LOGGER.debug("Computing resource utilization for checker: {}", utilizationChecker.getName());
utilizationChecker.computeResourceUtilization(endpointsToInstances, completionServiceHelper);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/workload/WorkloadPropagationClient.java b/pinot-controller/src/main/java/org/apache/pinot/controller/workload/WorkloadPropagationClient.java
index 33fb21b8c29..fbf8b9c9c45 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/workload/WorkloadPropagationClient.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/workload/WorkloadPropagationClient.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.workload;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.RateLimiter;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.IOException;
@@ -33,6 +34,7 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
@@ -51,6 +53,7 @@ import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.reactor.IOReactorConfig;
import org.apache.hc.core5.util.Timeout;
import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.exception.HttpErrorStatusException;
import org.apache.pinot.common.metrics.ControllerMeter;
import org.apache.pinot.common.metrics.ControllerMetrics;
@@ -171,17 +174,26 @@ public class WorkloadPropagationClient implements AutoCloseable {
_rateLimiter.acquire();
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.QUERY_WORKLOAD_MESSAGES_COUNT, 1);
// Send async request with retry logic (callbacks processed on _httpCallbackExecutor)
- SimpleHttpRequest httpRequest;
+ Supplier<SimpleHttpRequest> requestSupplier;
if (queryWorkloadRequest.isRefresh()) {
String requestBody = JsonUtils.objectToString(queryWorkloadRequest.getWorkloadToCostMap());
- httpRequest = SimpleRequestBuilder.post(url).setHeader(HttpHeaders.CONTENT_TYPE, HttpClient.JSON_CONTENT_TYPE)
- .setBody(requestBody, ContentType.APPLICATION_JSON).build();
+ requestSupplier = () -> {
+ SimpleRequestBuilder requestBuilder =
+ SimpleRequestBuilder.post(url).setHeader(HttpHeaders.CONTENT_TYPE, HttpClient.JSON_CONTENT_TYPE)
+ .setBody(requestBody, ContentType.APPLICATION_JSON);
+ setServerAdminAuthHeaders(instance, requestBuilder);
+ return requestBuilder.build();
+ };
} else {
String deleteUrl = url + "?workloadNames=" + String.join(",",
queryWorkloadRequest.getWorkloadToCostMap().keySet());
- httpRequest = SimpleRequestBuilder.delete(deleteUrl).build();
+ requestSupplier = () -> {
+ SimpleRequestBuilder requestBuilder = SimpleRequestBuilder.delete(deleteUrl);
+ setServerAdminAuthHeaders(instance, requestBuilder);
+ return requestBuilder.build();
+ };
}
- CompletableFuture<Boolean> future = sendWorkloadRequestWithRetry(httpRequest, instance);
+ CompletableFuture<Boolean> future = sendWorkloadRequestWithRetry(requestSupplier, instance);
futures.add(future);
} catch (Exception e) {
LOGGER.error("Error creating request for instance: {}", instance, e);
@@ -222,6 +234,13 @@ public class WorkloadPropagationClient implements AutoCloseable {
}
}
+ private void setServerAdminAuthHeaders(String instance, SimpleRequestBuilder requestBuilder) {
+ if (InstanceTypeUtils.isServer(instance)) {
+ AuthProviderUtils.makeAuthHeadersMap(_pinotHelixResourceManager.getServerAdminAuthProvider())
+ .forEach(requestBuilder::setHeader);
+ }
+ }
+
private String getServerUrl(String instanceName) {
if (_serverHttpSchemeAndPort == null) {
_serverHttpSchemeAndPort = discoverHttpSchemesAndPort(NodeConfig.Type.SERVER_NODE)
@@ -292,13 +311,23 @@ public class WorkloadPropagationClient implements AutoCloseable {
/// Sends a workload refresh request with automatic retries and exponential backoff.
/// Uses async HTTP client for non-blocking I/O with callbacks processed on dedicated executor.
- private CompletableFuture<Boolean> sendWorkloadRequestWithRetry(SimpleHttpRequest request, String instanceId) {
- return sendWorkloadRequestWithRetryInternal(request, instanceId, 0);
+ @VisibleForTesting
+ CompletableFuture<Boolean> sendWorkloadRequestWithRetry(Supplier<SimpleHttpRequest> requestSupplier,
+ String instanceId) {
+ return sendWorkloadRequestWithRetryInternal(requestSupplier, instanceId, 0);
}
- private CompletableFuture<Boolean> sendWorkloadRequestWithRetryInternal(SimpleHttpRequest request,
+ private CompletableFuture<Boolean> sendWorkloadRequestWithRetryInternal(Supplier<SimpleHttpRequest> requestSupplier,
String instanceId, int attemptNumber) {
- return sendWorkloadRequestAsync(request, instanceId)
+ CompletableFuture<Boolean> requestFuture;
+ try {
+ requestFuture = sendWorkloadRequestAsync(requestSupplier.get(), instanceId);
+ } catch (RuntimeException e) {
+ LOGGER.warn("Failed to create workload request for instance {} on attempt {}", instanceId, attemptNumber + 1,
+ e);
+ requestFuture = CompletableFuture.completedFuture(false);
+ }
+ return requestFuture
.thenComposeAsync(success -> {
if (success || attemptNumber >= WORKLOAD_PROPAGATION_MAX_RETRIES - 1) {
return CompletableFuture.completedFuture(success);
@@ -309,7 +338,7 @@ public class WorkloadPropagationClient implements AutoCloseable {
instanceId, delayMs, attemptNumber + 1, WORKLOAD_PROPAGATION_MAX_RETRIES);
CompletableFuture<Boolean> delayed = new CompletableFuture<>();
CompletableFuture.delayedExecutor(delayMs, TimeUnit.MILLISECONDS, _httpCallbackExecutor)
- .execute(() -> sendWorkloadRequestWithRetryInternal(request, instanceId, attemptNumber + 1)
+ .execute(() -> sendWorkloadRequestWithRetryInternal(requestSupplier, instanceId, attemptNumber + 1)
.whenComplete((result, error) -> {
if (error != null) {
delayed.completeExceptionally(error);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
index ce20061ce2f..039346f46d6 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
@@ -75,6 +75,7 @@ import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
import org.apache.pinot.segment.spi.creator.SegmentVersion;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
+import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
import org.apache.pinot.spi.config.table.DisasterRecoveryMode;
import org.apache.pinot.spi.config.table.PauseState;
@@ -1491,6 +1492,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
throws HttpErrorStatusException, IOException, URISyntaxException {
// mock the behavior for PinotHelixResourceManager
PinotHelixResourceManager pinotHelixResourceManager = mock(PinotHelixResourceManager.class);
+ AuthProvider serverAdminAuthProvider = mock(AuthProvider.class);
+ when(pinotHelixResourceManager.getServerAdminAuthProvider()).thenReturn(serverAdminAuthProvider);
HelixManager helixManager = mock(HelixManager.class);
HelixAdmin helixAdmin = mock(HelixAdmin.class);
ZkHelixPropertyStore<ZNRecord> zkHelixPropertyStore =
@@ -1547,8 +1550,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
// its final location. This is the expected segment location.
String expectedSegmentLocation =
segmentManager.createSegmentPath(RAW_TABLE_NAME, segmentsZKMetadata.get(0).getSegmentName()).toString();
- when(segmentManager._mockedFileUploadDownloadClient.uploadToSegmentStore(serverUploadRequestUrl0)).thenReturn(
- tempSegmentFileLocation.getPath());
+ when(segmentManager._mockedFileUploadDownloadClient.uploadToSegmentStore(serverUploadRequestUrl0,
+ serverAdminAuthProvider)).thenReturn(tempSegmentFileLocation.getPath());
// Change 2nd segment status to be DONE, but with default peer download url.
// Verify later the download url isn't fixed after upload failure.
@@ -1565,9 +1568,9 @@ public class PinotLLCRealtimeSegmentManagerTest {
String serverUploadRequestUrl1 =
String.format("http://%s:%d/segments/%s/%s/upload?uploadTimeoutMs=-1", instance1, adminPort,
REALTIME_TABLE_NAME, segmentsZKMetadata.get(1).getSegmentName());
- when(segmentManager._mockedFileUploadDownloadClient.uploadToSegmentStore(serverUploadRequestUrl1)).thenThrow(
- new HttpErrorStatusException("failed to upload segment",
- Response.Status.INTERNAL_SERVER_ERROR.getStatusCode()));
+ when(segmentManager._mockedFileUploadDownloadClient.uploadToSegmentStore(serverUploadRequestUrl1,
+ serverAdminAuthProvider)).thenThrow(new HttpErrorStatusException("failed to upload segment",
+ Response.Status.INTERNAL_SERVER_ERROR.getStatusCode()));
// Change 3rd segment status to be DONE, but with default peer download url.
// Verify later the download url isn't fixed because no ONLINE replica found in any server.
@@ -1608,6 +1611,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
assertEquals(segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentNames.get(3), null).getDownloadUrl(),
defaultDownloadUrl);
assertNull(segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentNames.get(4), null).getDownloadUrl());
+ verify(segmentManager._mockedFileUploadDownloadClient).uploadToSegmentStore(serverUploadRequestUrl0,
+ serverAdminAuthProvider);
+ verify(segmentManager._mockedFileUploadDownloadClient).uploadToSegmentStore(serverUploadRequestUrl1,
+ serverAdminAuthProvider);
}
/// Test cases for fixing LLC segment by uploading to segment store if missing
@@ -1616,6 +1623,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
throws HttpErrorStatusException, IOException, URISyntaxException {
// mock the behavior for PinotHelixResourceManager
PinotHelixResourceManager pinotHelixResourceManager = mock(PinotHelixResourceManager.class);
+ AuthProvider serverAdminAuthProvider = mock(AuthProvider.class);
+ when(pinotHelixResourceManager.getServerAdminAuthProvider()).thenReturn(serverAdminAuthProvider);
HelixManager helixManager = mock(HelixManager.class);
HelixAdmin helixAdmin = mock(HelixAdmin.class);
ZkHelixPropertyStore<ZNRecord> zkHelixPropertyStore =
@@ -1675,9 +1684,11 @@ public class PinotLLCRealtimeSegmentManagerTest {
SegmentZKMetadata segmentZKMetadataCopy =
new SegmentZKMetadata(new ZNRecord(segmentsZKMetadata.get(0).toZNRecord()));
- when(segmentManager._mockedFileUploadDownloadClient.uploadLLCToSegmentStore(serverUploadRequestUrl0)).thenReturn(
- new TableLLCSegmentUploadResponse(segmentsZKMetadata.get(0).getSegmentName(), 12345678L, 43210L,
- tempSegmentFileLocation.getPath()));
+ when(segmentManager._mockedFileUploadDownloadClient.uploadLLCToSegmentStore(serverUploadRequestUrl0,
+ serverAdminAuthProvider))
+ .thenReturn(
+ new TableLLCSegmentUploadResponse(segmentsZKMetadata.get(0).getSegmentName(), 12345678L, 43210L,
+ tempSegmentFileLocation.getPath()));
// Change 2nd segment status to be DONE, but with default peer download url.
// Verify later the download url isn't fixed after upload failure.
@@ -1694,9 +1705,11 @@ public class PinotLLCRealtimeSegmentManagerTest {
String serverUploadRequestUrl1 =
String.format("http://%s:%d/segments/%s/%s/uploadLLCSegment?uploadTimeoutMs=-1", instance1, adminPort,
REALTIME_TABLE_NAME, segmentsZKMetadata.get(1).getSegmentName());
- when(segmentManager._mockedFileUploadDownloadClient.uploadLLCToSegmentStore(serverUploadRequestUrl1)).thenThrow(
- new HttpErrorStatusException("failed to upload segment",
- Response.Status.INTERNAL_SERVER_ERROR.getStatusCode()));
+ when(segmentManager._mockedFileUploadDownloadClient.uploadLLCToSegmentStore(serverUploadRequestUrl1,
+ serverAdminAuthProvider))
+ .thenThrow(
+ new HttpErrorStatusException("failed to upload segment",
+ Response.Status.INTERNAL_SERVER_ERROR.getStatusCode()));
// Change 3rd segment status to be DONE, but with default peer download url.
// Verify later the download url isn't fixed because no ONLINE replica found in any server.
@@ -1737,6 +1750,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
assertEquals(segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentNames.get(3), null).getDownloadUrl(),
defaultDownloadUrl);
assertNull(segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentNames.get(4), null).getDownloadUrl());
+ verify(segmentManager._mockedFileUploadDownloadClient).uploadLLCToSegmentStore(serverUploadRequestUrl0,
+ serverAdminAuthProvider);
+ verify(segmentManager._mockedFileUploadDownloadClient).uploadLLCToSegmentStore(serverUploadRequestUrl1,
+ serverAdminAuthProvider);
}
@Test
@@ -1744,6 +1761,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
throws HttpErrorStatusException, IOException, URISyntaxException {
// mock the behavior for PinotHelixResourceManager
PinotHelixResourceManager pinotHelixResourceManager = mock(PinotHelixResourceManager.class);
+ AuthProvider serverAdminAuthProvider = mock(AuthProvider.class);
+ when(pinotHelixResourceManager.getServerAdminAuthProvider()).thenReturn(serverAdminAuthProvider);
HelixManager helixManager = mock(HelixManager.class);
HelixAdmin helixAdmin = mock(HelixAdmin.class);
ZkHelixPropertyStore<ZNRecord> zkHelixPropertyStore =
@@ -1816,7 +1835,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
uploadedCustomMap.put("segmentFileKey", "segmentFileValue");
segmentZKMetadataCopy.setCustomMap(uploadedCustomMap);
when(segmentManager._mockedFileUploadDownloadClient.uploadLLCToSegmentStoreWithZKMetadata(
- serverUploadRequestUrl0)).thenReturn(segmentZKMetadataCopy);
+ serverUploadRequestUrl0, serverAdminAuthProvider)).thenReturn(segmentZKMetadataCopy);
// Change 2nd segment status to be DONE, but with default peer download url.
// Verify later the download url isn't fixed after upload failure.
@@ -1833,9 +1852,11 @@ public class PinotLLCRealtimeSegmentManagerTest {
String serverUploadRequestUrl1 =
String.format("http://%s:%d/segments/%s/%s/uploadCommittedSegment?uploadTimeoutMs=-1", instance1, adminPort,
REALTIME_TABLE_NAME, segmentsZKMetadata.get(1).getSegmentName());
- when(segmentManager._mockedFileUploadDownloadClient.uploadLLCToSegmentStore(serverUploadRequestUrl1)).thenThrow(
- new HttpErrorStatusException("failed to upload segment",
- Response.Status.INTERNAL_SERVER_ERROR.getStatusCode()));
+ when(segmentManager._mockedFileUploadDownloadClient.uploadLLCToSegmentStoreWithZKMetadata(
+ serverUploadRequestUrl1, serverAdminAuthProvider))
+ .thenThrow(
+ new HttpErrorStatusException("failed to upload segment",
+ Response.Status.INTERNAL_SERVER_ERROR.getStatusCode()));
// Change 3rd segment status to be DONE, but with default peer download url.
// Verify later the download url isn't fixed because no ONLINE replica found in any server.
@@ -1881,6 +1902,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
assertEquals(segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentNames.get(3), null).getDownloadUrl(),
defaultDownloadUrl);
assertNull(segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentNames.get(4), null).getDownloadUrl());
+ verify(segmentManager._mockedFileUploadDownloadClient).uploadLLCToSegmentStoreWithZKMetadata(
+ serverUploadRequestUrl0, serverAdminAuthProvider);
+ verify(segmentManager._mockedFileUploadDownloadClient).uploadLLCToSegmentStoreWithZKMetadata(
+ serverUploadRequestUrl1, serverAdminAuthProvider);
}
@Test
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/workload/WorkloadPropagationClientTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/workload/WorkloadPropagationClientTest.java
new file mode 100644
index 00000000000..be8ad86e19e
--- /dev/null
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/workload/WorkloadPropagationClientTest.java
@@ -0,0 +1,141 @@
+/**
+ * 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.pinot.controller.workload;
+
+import com.sun.net.httpserver.HttpServer;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
+import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.spi.auth.AuthProvider;
+import org.apache.pinot.spi.config.workload.InstanceCost;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+/// Tests retry behavior for requests whose service credential is resolved while the request is built.
+public class WorkloadPropagationClientTest {
+ @Test
+ public void testServerRequestsResolveAdminAuthForEveryAttempt()
+ throws Exception {
+ List<String> receivedAuthHeaders = new CopyOnWriteArrayList<>();
+ List<String> receivedMethods = new CopyOnWriteArrayList<>();
+ AtomicInteger receivedRequests = new AtomicInteger();
+ HttpServer server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+ server.createContext("/queryWorkloadConfigs", exchange -> {
+ String authHeader = exchange.getRequestHeaders().getFirst("Authorization");
+ receivedAuthHeaders.add(authHeader != null ? authHeader : "<none>");
+ receivedMethods.add(exchange.getRequestMethod());
+ int status = receivedRequests.incrementAndGet() == 1 ? 500 : 200;
+ exchange.sendResponseHeaders(status, -1);
+ exchange.close();
+ });
+ server.start();
+
+ int port = server.getAddress().getPort();
+ String serverInstance = "Server_localhost_1234";
+ String brokerInstance = "Broker_localhost_1234";
+ PinotHelixResourceManager resourceManager = mock(PinotHelixResourceManager.class);
+ when(resourceManager.getAllInstances()).thenReturn(List.of(serverInstance, brokerInstance));
+ when(resourceManager.getHelixInstanceConfig(serverInstance)).thenReturn(instanceConfig(serverInstance, port));
+ when(resourceManager.getHelixInstanceConfig(brokerInstance)).thenReturn(instanceConfig(brokerInstance, port));
+ AtomicInteger authResolutions = new AtomicInteger();
+ when(resourceManager.getServerAdminAuthProvider()).thenReturn(new AuthProvider() {
+ @Override
+ public Map<String, Object> getRequestHeaders() {
+ return Map.of("Authorization", "token-" + authResolutions.incrementAndGet());
+ }
+
+ @Override
+ public String getTaskToken() {
+ return null;
+ }
+ });
+
+ try (WorkloadPropagationClient client = new WorkloadPropagationClient(resourceManager, new ControllerConf(),
+ mock(ControllerMetrics.class))) {
+ client.sendQueryWorkloadMessage(
+ Map.of(serverInstance, new QueryWorkloadRequest("refresh", new InstanceCost(1L, 2L))));
+ client.sendQueryWorkloadMessage(
+ Map.of(serverInstance, new QueryWorkloadRequest("delete", (InstanceCost) null)));
+ client.sendQueryWorkloadMessage(
+ Map.of(brokerInstance, new QueryWorkloadRequest("broker", new InstanceCost(3L, 4L))));
+
+ Assert.assertEquals(receivedMethods, List.of("POST", "POST", "DELETE", "POST"));
+ Assert.assertEquals(receivedAuthHeaders, List.of("token-1", "token-2", "token-3", "<none>"));
+ Assert.assertEquals(authResolutions.get(), 3);
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ public void testRequestConstructionFailureIsRetried()
+ throws Exception {
+ AtomicInteger receivedRequests = new AtomicInteger();
+ HttpServer server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+ server.createContext("/queryWorkloadConfigs", exchange -> {
+ receivedRequests.incrementAndGet();
+ exchange.sendResponseHeaders(200, -1);
+ exchange.close();
+ });
+ server.start();
+
+ PinotHelixResourceManager resourceManager = mock(PinotHelixResourceManager.class);
+ when(resourceManager.getAllInstances()).thenReturn(List.of());
+ AtomicInteger requestConstructions = new AtomicInteger();
+ Supplier<SimpleHttpRequest> requestSupplier = () -> {
+ if (requestConstructions.incrementAndGet() == 1) {
+ throw new IllegalStateException("credential refresh failed");
+ }
+ return SimpleRequestBuilder.get(
+ "http://localhost:" + server.getAddress().getPort() + "/queryWorkloadConfigs").build();
+ };
+
+ try (WorkloadPropagationClient client = new WorkloadPropagationClient(resourceManager, new ControllerConf(),
+ mock(ControllerMetrics.class))) {
+ Assert.assertTrue(client.sendWorkloadRequestWithRetry(requestSupplier, "Server_localhost_1234")
+ .get(10, TimeUnit.SECONDS));
+ Assert.assertEquals(requestConstructions.get(), 2);
+ Assert.assertEquals(receivedRequests.get(), 1);
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ private static InstanceConfig instanceConfig(String instanceName, int adminPort) {
+ InstanceConfig instanceConfig = new InstanceConfig(instanceName);
+ instanceConfig.setHostName("localhost");
+ instanceConfig.getRecord().setIntField(CommonConstants.Helix.Instance.ADMIN_PORT_KEY, adminPort);
+ return instanceConfig;
+ }
+}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/auth/BasicAuthPrincipal.java b/pinot-core/src/main/java/org/apache/pinot/core/auth/BasicAuthPrincipal.java
index b31f3c36879..80997af2d60 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/auth/BasicAuthPrincipal.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/auth/BasicAuthPrincipal.java
@@ -74,6 +74,11 @@ public class BasicAuthPrincipal {
return _permissions.isEmpty() || _permissions.contains(permission.toLowerCase());
}
+ /// Returns whether this principal was explicitly granted the given permission.
+ public boolean hasExplicitPermission(String permission) {
+ return _permissions.contains(permission.toLowerCase());
+ }
+
/// Gets the Row-Level Security (RLS) filter configured for the given table.
/// The RLS filter is applied only if the user has access to the table
/// (as determined by [#hasTable(String)]).
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/access/AccessControl.java b/pinot-core/src/main/java/org/apache/pinot/server/access/AccessControl.java
index 626c065db28..50f7d70240e 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/access/AccessControl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/server/access/AccessControl.java
@@ -21,6 +21,8 @@ package org.apache.pinot.server.access;
import io.netty.channel.ChannelHandlerContext;
import org.apache.pinot.spi.annotations.InterfaceAudience;
import org.apache.pinot.spi.annotations.InterfaceStability;
+import org.apache.pinot.spi.auth.AuthorizationResult;
+import org.apache.pinot.spi.auth.BasicAuthorizationResultImpl;
import org.apache.pinot.spi.auth.server.RequesterIdentity;
@@ -32,6 +34,22 @@ public interface AccessControl {
/// @return Whether the client has access to query server
boolean isAuthorizedChannel(ChannelHandlerContext channelHandlerContext);
+ /// Authorizes privileged access to non-table Server administrative operations.
+ ///
+ /// Implementations must throw {@link javax.ws.rs.NotAuthorizedException} when the request does not contain an
+ /// acceptable authenticated identity. An authenticated identity that lacks Server administrative authority must
+ /// return a result whose {@link AuthorizationResult#hasAccess()} is {@code false}.
+ ///
+ /// The default implementation denies access so that existing custom access-control implementations remain binary
+ /// compatible without silently granting access to newly protected administrative routes. Legacy implementations
+ /// therefore receive a forbidden response until they implement this administrative authorization contract.
+ ///
+ /// @param requesterIdentity request identity
+ /// @return authorization result for Server administrative access
+ default AuthorizationResult authorizeAdminAccess(RequesterIdentity requesterIdentity) {
+ return new BasicAuthorizationResultImpl(false);
+ }
+
/// Return whether the client has data access to the given table.
///
/// @param requesterIdentity Request identity
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java b/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
index e23a46fe83b..d47553e5ac6 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
+++ b/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
@@ -19,6 +19,8 @@
package org.apache.pinot.server.access;
import io.netty.channel.ChannelHandlerContext;
+import org.apache.pinot.spi.auth.AuthorizationResult;
+import org.apache.pinot.spi.auth.BasicAuthorizationResultImpl;
import org.apache.pinot.spi.auth.server.RequesterIdentity;
@@ -29,6 +31,11 @@ public class AllowAllAccessFactory implements AccessControlFactory {
return true;
}
+ @Override
+ public AuthorizationResult authorizeAdminAccess(RequesterIdentity requesterIdentity) {
+ return BasicAuthorizationResultImpl.success();
+ }
+
@Override
public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
return true;
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/access/BasicAuthAccessFactory.java b/pinot-core/src/main/java/org/apache/pinot/server/access/BasicAuthAccessFactory.java
index eec15d14312..1c4ab076feb 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/access/BasicAuthAccessFactory.java
+++ b/pinot-core/src/main/java/org/apache/pinot/server/access/BasicAuthAccessFactory.java
@@ -20,13 +20,17 @@ package org.apache.pinot.server.access;
import io.netty.channel.ChannelHandlerContext;
import java.util.Collection;
+import java.util.List;
import java.util.Map;
-import java.util.Objects;
+import java.util.Optional;
import java.util.stream.Collectors;
+import javax.ws.rs.NotAuthorizedException;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.auth.BasicAuthTokenUtils;
import org.apache.pinot.core.auth.BasicAuthPrincipal;
import org.apache.pinot.core.auth.BasicAuthPrincipalUtils;
+import org.apache.pinot.spi.auth.AuthorizationResult;
+import org.apache.pinot.spi.auth.BasicAuthorizationResultImpl;
import org.apache.pinot.spi.auth.server.RequesterIdentity;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
@@ -36,6 +40,7 @@ public class BasicAuthAccessFactory implements AccessControlFactory {
private static final String PREFIX = "principals";
private static final String AUTHORIZATION_KEY = "authorization";
+ private static final String ADMIN_PERMISSION = "admin";
private AccessControl _accessControl;
@@ -62,30 +67,51 @@ public class BasicAuthAccessFactory implements AccessControlFactory {
return true;
}
+ @Override
+ public AuthorizationResult authorizeAdminAccess(RequesterIdentity requesterIdentity) {
+ Optional<BasicAuthPrincipal> principal = getPrincipal(requesterIdentity);
+ if (!principal.isPresent()) {
+ throw new NotAuthorizedException("Basic");
+ }
+ return new BasicAuthorizationResultImpl(principal.get().hasExplicitPermission(ADMIN_PERMISSION));
+ }
+
@Override
public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
- Collection<String> tokens = getTokens(requesterIdentity);
- return tokens.stream()
- .map(BasicAuthTokenUtils::normalizeBase64Token)
- .map(_token2principal::get)
- .filter(Objects::nonNull)
- .findFirst()
- // existence of principal required to allow access
+ return getPrincipal(requesterIdentity)
.map(principal -> StringUtils.isEmpty(tableName) || principal.hasTable(
TableNameBuilder.extractRawTableName(tableName)))
.orElse(false);
}
+ private Optional<BasicAuthPrincipal> getPrincipal(RequesterIdentity requesterIdentity) {
+ for (String token : getTokens(requesterIdentity)) {
+ String normalizedToken;
+ try {
+ normalizedToken = BasicAuthTokenUtils.normalizeBase64Token(token);
+ } catch (IllegalArgumentException e) {
+ continue;
+ }
+ BasicAuthPrincipal principal = _token2principal.get(normalizedToken);
+ if (principal != null) {
+ return Optional.of(principal);
+ }
+ }
+ return Optional.empty();
+ }
+
private Collection<String> getTokens(RequesterIdentity requesterIdentity) {
+ Collection<String> tokens;
if (requesterIdentity instanceof GrpcRequesterIdentity) {
GrpcRequesterIdentity identity = (GrpcRequesterIdentity) requesterIdentity;
- return identity.getGrpcMetadata().get(AUTHORIZATION_KEY);
- }
- if (requesterIdentity instanceof HttpRequesterIdentity) {
+ tokens = identity.getGrpcMetadata().get(AUTHORIZATION_KEY);
+ } else if (requesterIdentity instanceof HttpRequesterIdentity) {
HttpRequesterIdentity identity = (HttpRequesterIdentity) requesterIdentity;
- return identity.getHttpHeaders().get(AUTHORIZATION_KEY);
+ tokens = identity.getHeaderValues(AUTHORIZATION_KEY);
+ } else {
+ return List.of();
}
- throw new UnsupportedOperationException("GrpcRequesterIdentity or HttpRequesterIdentity is required");
+ return tokens != null ? tokens : List.of();
}
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/access/HttpRequesterIdentity.java b/pinot-core/src/main/java/org/apache/pinot/server/access/HttpRequesterIdentity.java
index c916620c7d0..775682b5414 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/access/HttpRequesterIdentity.java
+++ b/pinot-core/src/main/java/org/apache/pinot/server/access/HttpRequesterIdentity.java
@@ -20,8 +20,12 @@ package org.apache.pinot.server.access;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
import javax.ws.rs.core.HttpHeaders;
import org.apache.pinot.spi.auth.server.RequesterIdentity;
+import org.glassfish.grizzly.http.server.Request;
/// Identity container for HTTP requests with (optional) authorization headers
@@ -34,6 +38,25 @@ public class HttpRequesterIdentity extends RequesterIdentity {
httpHeaders.getRequestHeaders().forEach(_httpHeaders::putAll);
}
+ public HttpRequesterIdentity(Request request) {
+ _httpHeaders = HashMultimap.create();
+ request.getHeaderNames().forEach(name -> request.getHeaders(name).forEach(value -> _httpHeaders.put(name, value)));
+ String requestUrl = request.getRequestURL().toString();
+ String queryString = request.getQueryString();
+ _endpointUrl = queryString != null ? requestUrl + "?" + queryString : requestUrl;
+ }
+
+ /// Returns all values for a header name, using HTTP's case-insensitive name semantics.
+ public Collection<String> getHeaderValues(String name) {
+ List<String> values = new ArrayList<>();
+ _httpHeaders.asMap().forEach((headerName, headerValues) -> {
+ if (headerName.equalsIgnoreCase(name)) {
+ values.addAll(headerValues);
+ }
+ });
+ return values;
+ }
+
public Multimap<String, String> getHttpHeaders() {
return _httpHeaders;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/access/ZkBasicAuthAccessFactory.java b/pinot-core/src/main/java/org/apache/pinot/server/access/ZkBasicAuthAccessFactory.java
index f66a2ca8852..9090e32571f 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/access/ZkBasicAuthAccessFactory.java
+++ b/pinot-core/src/main/java/org/apache/pinot/server/access/ZkBasicAuthAccessFactory.java
@@ -20,9 +20,13 @@ package org.apache.pinot.server.access;
import io.netty.channel.ChannelHandlerContext;
import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
-import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
+import javax.ws.rs.NotAuthorizedException;
import org.apache.commons.lang3.StringUtils;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.auth.BasicAuthTokenUtils;
@@ -30,7 +34,11 @@ import org.apache.pinot.common.config.provider.AccessControlUserCache;
import org.apache.pinot.common.utils.BcryptUtils;
import org.apache.pinot.core.auth.BasicAuthPrincipalUtils;
import org.apache.pinot.core.auth.ZkBasicAuthPrincipal;
+import org.apache.pinot.spi.auth.AuthorizationResult;
+import org.apache.pinot.spi.auth.BasicAuthorizationResultImpl;
import org.apache.pinot.spi.auth.server.RequesterIdentity;
+import org.apache.pinot.spi.config.user.ComponentType;
+import org.apache.pinot.spi.config.user.RoleType;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
@@ -45,7 +53,7 @@ public class ZkBasicAuthAccessFactory implements AccessControlFactory {
private static final String AUTHORIZATION_KEY = "authorization";
private HelixManager _helixManager;
- private AccessControl _accessControl;
+ private final AtomicReference<AccessControl> _accessControl = new AtomicReference<>();
@Override
public void init(PinotConfiguration configuration, HelixManager helixManager) {
@@ -54,23 +62,24 @@ public class ZkBasicAuthAccessFactory implements AccessControlFactory {
@Override
public AccessControl create() {
- if (_accessControl == null) {
- _accessControl = new BasicAuthAccessControl(_helixManager);
+ AccessControl accessControl = _accessControl.get();
+ if (accessControl == null) {
+ _accessControl.compareAndSet(null, new BasicAuthAccessControl(_helixManager));
+ accessControl = _accessControl.get();
}
- return _accessControl;
+ return accessControl;
}
/// Access Control using metadata-based basic grpc authentication
private static class BasicAuthAccessControl implements AccessControl {
- private Map<String, ZkBasicAuthPrincipal> _name2principal;
- private AccessControlUserCache _userCache;
- private HelixManager _innerHelixManager;
+ private volatile AccessControlUserCache _userCache;
+ private final HelixManager _innerHelixManager;
public BasicAuthAccessControl(HelixManager helixManager) {
_innerHelixManager = helixManager;
}
- public void initUserCache() {
+ public synchronized void initUserCache() {
if (_userCache == null) {
_userCache = new AccessControlUserCache(_innerHelixManager.getHelixPropertyStore());
}
@@ -81,38 +90,79 @@ public class ZkBasicAuthAccessFactory implements AccessControlFactory {
return true;
}
+ @Override
+ public AuthorizationResult authorizeAdminAccess(RequesterIdentity requesterIdentity) {
+ Optional<ZkBasicAuthPrincipal> principal = getPrincipal(requesterIdentity);
+ if (!principal.isPresent()) {
+ throw new NotAuthorizedException("Basic");
+ }
+ return new BasicAuthorizationResultImpl(
+ principal.get().hasPermission(RoleType.ADMIN, ComponentType.SERVER));
+ }
+
@Override
public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
+ return getPrincipal(requesterIdentity)
+ .map(principal -> StringUtils.isEmpty(tableName) || principal.hasTable(
+ TableNameBuilder.extractRawTableName(tableName)))
+ .orElse(false);
+ }
+
+ private Optional<ZkBasicAuthPrincipal> getPrincipal(RequesterIdentity requesterIdentity) {
+ Collection<String> tokens = getTokens(requesterIdentity);
+ if (tokens.isEmpty()) {
+ return Optional.empty();
+ }
+
+ Map<String, String> name2password = new LinkedHashMap<>();
+ for (String token : tokens) {
+ String decodedToken;
+ try {
+ decodedToken = BasicAuthTokenUtils.decodeBasicAuthToken(token);
+ } catch (IllegalArgumentException e) {
+ continue;
+ }
+ int separatorIndex = decodedToken != null ? decodedToken.indexOf(':') : -1;
+ if (separatorIndex <= 0 || separatorIndex == decodedToken.length() - 1) {
+ continue;
+ }
+ String username = decodedToken.substring(0, separatorIndex);
+ String password = decodedToken.substring(separatorIndex + 1);
+ name2password.put(username, password);
+ }
+ if (name2password.isEmpty()) {
+ return Optional.empty();
+ }
+
if (_userCache == null) {
initUserCache();
}
- Collection<String> tokens = getTokens(requesterIdentity);
- _name2principal =
+ Map<String, ZkBasicAuthPrincipal> name2principal =
BasicAuthPrincipalUtils.extractBasicAuthPrincipals(_userCache.getAllServerUserConfig()).stream()
.collect(Collectors.toMap(ZkBasicAuthPrincipal::getName, p -> p));
- Map<String, String> name2password = tokens.stream().collect(
- Collectors.toMap(BasicAuthTokenUtils::extractUsername, BasicAuthTokenUtils::extractPassword,
- (v1, v2) -> v2));
- Map<String, ZkBasicAuthPrincipal> password2principal =
- name2password.keySet().stream().collect(Collectors.toMap(name2password::get, _name2principal::get));
- return password2principal.entrySet().stream().filter(
- entry -> BcryptUtils.checkpwWithCache(entry.getKey(), entry.getValue().getPassword(),
- _userCache.getUserPasswordAuthCache())).map(u -> u.getValue()).filter(Objects::nonNull).findFirst().map(
- zkprincipal -> StringUtils.isEmpty(tableName) || zkprincipal.hasTable(
- TableNameBuilder.extractRawTableName(tableName))).orElse(false);
+ for (Map.Entry<String, String> entry : name2password.entrySet()) {
+ ZkBasicAuthPrincipal principal = name2principal.get(entry.getKey());
+ if (principal != null && BcryptUtils.checkpwWithCache(entry.getValue(), principal.getPassword(),
+ _userCache.getUserPasswordAuthCache())) {
+ return Optional.of(principal);
+ }
+ }
+ return Optional.empty();
}
private Collection<String> getTokens(RequesterIdentity requesterIdentity) {
+ Collection<String> tokens;
if (requesterIdentity instanceof GrpcRequesterIdentity) {
GrpcRequesterIdentity identity = (GrpcRequesterIdentity) requesterIdentity;
- return identity.getGrpcMetadata().get(AUTHORIZATION_KEY);
- }
- if (requesterIdentity instanceof HttpRequesterIdentity) {
+ tokens = identity.getGrpcMetadata().get(AUTHORIZATION_KEY);
+ } else if (requesterIdentity instanceof HttpRequesterIdentity) {
HttpRequesterIdentity identity = (HttpRequesterIdentity) requesterIdentity;
- return identity.getHttpHeaders().get(AUTHORIZATION_KEY);
+ tokens = identity.getHeaderValues(AUTHORIZATION_KEY);
+ } else {
+ return List.of();
}
- throw new UnsupportedOperationException("GrpcRequesterIdentity or HttpRequesterIdentity is required");
+ return tokens != null ? tokens : List.of();
}
}
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/server/access/ZkBasicAuthAccessFactoryTest.java b/pinot-core/src/test/java/org/apache/pinot/server/access/ZkBasicAuthAccessFactoryTest.java
new file mode 100644
index 00000000000..af7e713b9d4
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/server/access/ZkBasicAuthAccessFactoryTest.java
@@ -0,0 +1,83 @@
+/**
+ * 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.pinot.server.access;
+
+import java.util.List;
+import java.util.Map;
+import javax.ws.rs.NotAuthorizedException;
+import org.apache.helix.HelixManager;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.common.auth.BasicAuthTokenUtils;
+import org.apache.pinot.common.utils.BcryptUtils;
+import org.apache.pinot.common.utils.config.AccessControlUserConfigUtils;
+import org.apache.pinot.spi.config.user.ComponentType;
+import org.apache.pinot.spi.config.user.RoleType;
+import org.apache.pinot.spi.config.user.UserConfig;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+/// Tests the ZK-backed Server administrative authorization contract.
+public class ZkBasicAuthAccessFactoryTest {
+ @Test
+ public void testAdminAccessRequiresServerAdminIdentity()
+ throws Exception {
+ UserConfig admin = user("admin", "admin-secret", RoleType.ADMIN);
+ UserConfig reader = user("reader", "reader-secret", RoleType.USER);
+
+ @SuppressWarnings("unchecked")
+ ZkHelixPropertyStore<ZNRecord> propertyStore = mock(ZkHelixPropertyStore.class);
+ when(propertyStore.getChildNames(anyString(), anyInt())).thenReturn(List.of("admin_SERVER", "reader_SERVER"));
+ when(propertyStore.get(anyList(), isNull(), anyInt(), org.mockito.ArgumentMatchers.eq(false)))
+ .thenReturn(List.of(AccessControlUserConfigUtils.toZNRecord(admin),
+ AccessControlUserConfigUtils.toZNRecord(reader)));
+ HelixManager helixManager = mock(HelixManager.class);
+ when(helixManager.getHelixPropertyStore()).thenReturn(propertyStore);
+
+ ZkBasicAuthAccessFactory factory = new ZkBasicAuthAccessFactory();
+ factory.init(new PinotConfiguration(), helixManager);
+ AccessControl accessControl = factory.create();
+
+ Assert.assertTrue(accessControl.authorizeAdminAccess(identity("admin", "admin-secret")).hasAccess());
+ Assert.assertFalse(accessControl.authorizeAdminAccess(identity("reader", "reader-secret")).hasAccess());
+ Assert.expectThrows(NotAuthorizedException.class,
+ () -> accessControl.authorizeAdminAccess(identity("admin", "wrong-secret")));
+ Assert.expectThrows(NotAuthorizedException.class,
+ () -> accessControl.authorizeAdminAccess(new GrpcRequesterIdentity(Map.of())));
+ }
+
+ private static UserConfig user(String name, String password, RoleType roleType) {
+ return new UserConfig(name, BcryptUtils.encrypt(password), ComponentType.SERVER.name(), roleType.name(),
+ List.of("testTable"), List.of(), List.of());
+ }
+
+ private static GrpcRequesterIdentity identity(String name, String password) {
+ return new GrpcRequesterIdentity(
+ Map.of("authorization", BasicAuthTokenUtils.toBasicAuthToken(name, password)));
+ }
+}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasicAuthBatchIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasicAuthBatchIntegrationTest.java
index 55fb9e7f17b..f9a1156bad9 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasicAuthBatchIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasicAuthBatchIntegrationTest.java
@@ -30,6 +30,7 @@ import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.exception.HttpErrorStatusException;
+import org.apache.pinot.server.access.BasicAuthAccessFactory;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.tools.BootstrapTableTool;
@@ -78,16 +79,25 @@ public class BasicAuthBatchIntegrationTest extends ClusterTest {
@Override
protected void overrideControllerConf(Map<String, Object> properties) {
BasicAuthTestUtils.addControllerConfiguration(properties);
+ properties.put("controller.server.admin.auth.token", AUTH_TOKEN);
}
@Override
protected void overrideBrokerConf(PinotConfiguration brokerConf) {
BasicAuthTestUtils.addBrokerConfiguration(brokerConf);
+ brokerConf.setProperty("pinot.broker.server.admin.auth.token", AUTH_TOKEN);
}
@Override
protected void overrideServerConf(PinotConfiguration serverConf) {
BasicAuthTestUtils.addServerConfiguration(serverConf);
+ serverConf.setProperty("pinot.server.admin.access.control.factory.class", BasicAuthAccessFactory.class.getName());
+ serverConf.setProperty("pinot.server.admin.access.control.principals", "admin,user");
+ serverConf.setProperty("pinot.server.admin.access.control.principals.admin.password", "verysecret");
+ serverConf.setProperty("pinot.server.admin.access.control.principals.admin.permissions", "admin");
+ serverConf.setProperty("pinot.server.admin.access.control.principals.user.password", "secret");
+ serverConf.setProperty("pinot.server.admin.access.control.principals.user.tables", "userTableOnly");
+ serverConf.setProperty("pinot.server.admin.access.control.principals.user.permissions", "read");
}
@Override
@@ -175,6 +185,12 @@ public class BasicAuthBatchIntegrationTest extends ClusterTest {
"must return row count 97889");
Assert.assertTrue(response.get("exceptions").isEmpty(), "must not return exception");
+ // The Controller fans this request out to the protected Server admin API with its configured service identity.
+ JsonNode tableSizeResponse = JsonUtils.stringToJsonNode(
+ sendGetRequest("http://localhost:" + getControllerPort() + "/tables/baseballStats/size", AUTH_HEADER));
+ Assert.assertTrue(tableSizeResponse.get("reportedSizeInBytes").asLong() > 0,
+ "must return the size reported by the Server");
+
// user with valid auth but no table access - must return 403
try {
sendPostRequest("http://localhost:" + getRandomBrokerPort() + "/query/sql",
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/access/CertBasedTlsChannelAccessControlFactory.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/access/CertBasedTlsChannelAccessControlFactory.java
index 6efcb233369..8cc8fca95eb 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/access/CertBasedTlsChannelAccessControlFactory.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/access/CertBasedTlsChannelAccessControlFactory.java
@@ -31,6 +31,8 @@ import java.util.Set;
import javax.net.ssl.SSLPeerUnverifiedException;
import org.apache.pinot.server.access.AccessControl;
import org.apache.pinot.server.access.AccessControlFactory;
+import org.apache.pinot.spi.auth.AuthorizationResult;
+import org.apache.pinot.spi.auth.BasicAuthorizationResultImpl;
import org.apache.pinot.spi.auth.server.RequesterIdentity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -72,5 +74,11 @@ public class CertBasedTlsChannelAccessControlFactory implements AccessControlFac
public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
return true;
}
+
+ @Override
+ public AuthorizationResult authorizeAdminAccess(RequesterIdentity requesterIdentity) {
+ // The integration-test admin listener requires a trusted client certificate before Jersey handles the request.
+ return BasicAuthorizationResultImpl.success();
+ }
}
}
diff --git a/pinot-server/README.md b/pinot-server/README.md
new file mode 100644
index 00000000000..6883b22e67c
--- /dev/null
+++ b/pinot-server/README.md
@@ -0,0 +1,120 @@
+<!--
+
+ 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.
+
+-->
+
+# Pinot Server
+
+## Admin API access control
+
+The Server admin listener uses the Server `AccessControlFactory` as its inbound trust boundary. The route inventory at
+the time this contract was introduced has three authorization categories across JAX-RS declarations and static HTTP
+handler prefixes:
+
+| Category | Inventory | Families | Authorization |
+|---|---:|---|---|
+| Public health/readiness | 3 JAX-RS declarations | `/health`, `/health/liveness`, `/health/readiness` | No credentials |
+| Table data | 2 JAX-RS declarations | Immutable segment and valid-document bitmap downloads | `hasDataAccess` for the requested table |
+| Privileged administration | 48 JAX-RS declarations | Configuration and instance inspection, table and segment administration, reload and reingestion, query and worker administration, logging and diagnostics, and tier or workload operations | `authorizeAdminAccess` |
+| Privileged administration | 3 static handler prefixes | `/api/`, `/help/`, `/swaggerui-dist/` | `authorizeAdminAccess` before serving static content |
+
+The three health paths are exact GET exceptions (a query string does not change the path). Other methods on those
+paths, all other JAX-RS routes, and all static handlers on the admin listener are privileged unless they are one of the
+two built-in table-data downloads. The public and table-data markers are honored only on the exact built-in
+`HealthCheckResource` and `TablesResource` classes; custom resource classes remain privileged even if they reuse a
+marker. New routes therefore fail closed and require administrative authorization until the built-in inventory and
+enforcement code are deliberately extended together. This boundary does not replace the Server query-port channel
+checks.
+
+### Authorization contract
+
+`AccessControl.authorizeAdminAccess(RequesterIdentity)` is the contract for non-table administrative operations. An
+implementation should:
+
+- throw `NotAuthorizedException` when credentials are absent or invalid, producing HTTP 401;
+- return a denied `AuthorizationResult` for an authenticated identity without administrative authority, producing
+ HTTP 403; and
+- return an allowed result for an authorized administrator.
+
+The default implementation denies administrative access. Custom `AccessControl` implementations must override the
+method before enabling them on an upgraded Server; otherwise table-data authorization continues to work but
+administrative requests receive 403. This fail-closed default preserves source and binary compatibility without
+silently granting a new class of access.
+
+`BasicAuthAccessFactory` authorizes administrative operations with the `admin` permission. Configure it under the
+existing Server access-control namespace, for example:
+
+```properties
+pinot.server.admin.access.control.factory.class=org.apache.pinot.server.access.BasicAuthAccessFactory
+pinot.server.admin.access.control.principals=serverAdmin,segmentReader
+pinot.server.admin.access.control.principals.serverAdmin.password=<admin-password>
+pinot.server.admin.access.control.principals.serverAdmin.permissions=admin
+pinot.server.admin.access.control.principals.segmentReader.password=<reader-password>
+pinot.server.admin.access.control.principals.segmentReader.permissions=read
+pinot.server.admin.access.control.principals.segmentReader.tables=<allowed-tables>
+```
+
+The `admin` permission must be explicit. Although a Basic principal with no configured permissions retains its
+existing wildcard behavior for older permission checks, it does not pass the Server administrative check. This avoids
+turning an existing table-data identity into an administrator during upgrade. Use dedicated service identities instead
+of sharing an operator credential.
+
+`ZkBasicAuthAccessFactory` uses the Server user model instead: the authenticated user must have role `ADMIN` and
+component `SERVER`. Its password and table permissions remain independent inputs to authentication and table-data
+authorization.
+
+`AllowAllAccessFactory` remains the default when no Server access-control factory is configured. It allows the health,
+table-data, and administrative categories, preserving the unconfigured behavior. Operators who require protection
+must configure an access-control factory; upgrading alone does not enable authentication.
+
+### Internal callers and rolling migration
+
+Controllers and Brokers invoke privileged Server operations. Their outbound credentials must identify a Server
+administrator: an explicit `admin` permission for `BasicAuthAccessFactory`, or role `ADMIN` and component `SERVER` for
+`ZkBasicAuthAccessFactory`:
+
+- configure Controller-to-Server credentials under `controller.server.admin.auth.*`; and
+- configure Broker-to-Server credentials under `pinot.broker.server.admin.auth.*`.
+
+The two data-download families continue to use table authorization rather than administrative authorization. Segment
+archive downloads use segment-fetcher credentials under `controller.segment.fetcher.auth.*` and
+`pinot.server.segment.fetcher.auth.*`. Direct valid-document bitmap downloads are currently used by Minion task code
+and remain subject to the separate Minion limitation below. Administrative and table access remain separate decisions.
+
+For a rolling deployment:
+
+1. Add or update inbound Server principals. Grant `admin` to each Controller, Broker, automation, and operator identity
+ that needs privileged operations. Update custom `AccessControl` implementations at this stage.
+2. Deploy Controller and Broker support for the outbound admin credential prefixes and configure the credentials.
+ Older Servers ignore the additional authorization header.
+3. Upgrade Servers and enable the configured access-control factory. Verify Controller fan-out, Broker-to-Server
+ operations, segment downloads, and the three unauthenticated probes before tightening network policy.
+4. Migrate remaining direct admin-listener clients. They now receive 401 for missing or invalid credentials and 403
+ when authenticated without administrative authority.
+
+### Minion boundary
+
+Minion is intentionally outside this Server change. It has no equivalent inbound `AccessControlFactory` contract, so
+its administrative listener must remain restricted to a trusted internal network (or an independently enforced
+service-mesh or proxy boundary). Minion's outbound credentials do not provide inbound protection, and Minion should
+not be treated as a configured-BasicAuth exception for the Server listener. Minion task code, including task generators
+hosted by the Controller and workers that fetch valid-document bitmaps, does not consume the new Server service
+credential in this patch. Deployments that combine those task paths with a protected Server listener must inject a
+service identity at a trusted proxy or service-mesh boundary. No Minion task, worker, or listener behavior is changed by
+this patch.
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java b/pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java
index e80b2da117e..ba33d84aa47 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java
@@ -58,6 +58,7 @@ public class AdminApiApplication extends ResourceConfig {
private final AtomicBoolean _shutDownInProgress = new AtomicBoolean();
private final ServerInstance _serverInstance;
+ private final AccessControlFactory _accessControlFactory;
private HttpServer _httpServer;
private final String _adminApiResourcePackages;
@@ -66,6 +67,7 @@ public class AdminApiApplication extends ResourceConfig {
ServerReloadJobStatusCache reloadJobStatusCache,
PinotConfiguration serverConf) {
_serverInstance = instance;
+ _accessControlFactory = accessControlFactory;
_adminApiResourcePackages = serverConf.getProperty(CommonConstants.Server.CONFIG_OF_SERVER_RESOURCE_PACKAGES,
CommonConstants.Server.DEFAULT_SERVER_RESOURCE_PACKAGES);
@@ -80,7 +82,7 @@ public class AdminApiApplication extends ResourceConfig {
bind(_serverInstance).to(ServerInstance.class);
bind(_serverInstance.getHelixManager()).to(HelixManager.class);
bind(_serverInstance.getServerMetrics()).to(ServerMetrics.class);
- bind(accessControlFactory).to(AccessControlFactory.class);
+ bind(_accessControlFactory).to(AccessControlFactory.class);
bind(reloadJobStatusCache).to(ServerReloadJobStatusCache.class);
bind(serverConf.getProperty(CommonConstants.Server.CONFIG_OF_INSTANCE_ID)).named(SERVER_INSTANCE_ID);
@@ -94,6 +96,8 @@ public class AdminApiApplication extends ResourceConfig {
}
});
+ register(ServerRequestMethodCaptureFilter.class);
+ register(ServerAdminApiAccessControlFilter.class);
register(JacksonFeature.class);
register(SwaggerApiListingResource.class);
@@ -127,7 +131,8 @@ public class AdminApiApplication extends ResourceConfig {
boolean useHttps = Boolean.parseBoolean(
pinotConfiguration.getProperty(CommonConstants.Server.CONFIG_OF_SWAGGER_USE_HTTPS));
PinotReflectionUtils.runWithLock(() ->
- SwaggerSetupUtils.setupSwagger("Server", _adminApiResourcePackages, useHttps, "/", _httpServer));
+ SwaggerSetupUtils.setupSwagger("Server", _adminApiResourcePackages, useHttps, "/", _httpServer,
+ handler -> new ServerAdminApiAccessControlHandler(handler, _accessControlFactory)));
}
return true;
}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/ServerAdminApiAccessControlFilter.java b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerAdminApiAccessControlFilter.java
new file mode 100644
index 00000000000..1ac637e07f4
--- /dev/null
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerAdminApiAccessControlFilter.java
@@ -0,0 +1,83 @@
+/**
+ * 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.pinot.server.api;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import javax.annotation.Priority;
+import javax.inject.Inject;
+import javax.ws.rs.Priorities;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.container.ContainerRequestContext;
+import javax.ws.rs.container.ContainerRequestFilter;
+import javax.ws.rs.container.ResourceInfo;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.pinot.server.access.AccessControlFactory;
+import org.apache.pinot.server.access.HttpRequesterIdentity;
+import org.apache.pinot.server.api.resources.HealthCheckResource;
+import org.apache.pinot.server.api.resources.TablesResource;
+import org.apache.pinot.spi.auth.AuthorizationResult;
+
+
+/// Enforces the authorization boundary for the Server administrative API.
+///
+/// Only the Server health endpoints are public. Methods marked with [ServerDataAccess] retain their explicit table-data
+/// access checks. Every other Server route, including custom resource routes, requires administrative authorization
+/// from the configured Server access control.
+@Priority(Priorities.AUTHENTICATION)
+public class ServerAdminApiAccessControlFilter implements ContainerRequestFilter {
+ private static final String GET = "GET";
+
+ @Inject
+ private AccessControlFactory _accessControlFactory;
+
+ @Context
+ private ResourceInfo _resourceInfo;
+
+ @Context
+ private HttpHeaders _httpHeaders;
+
+ @Override
+ public void filter(ContainerRequestContext requestContext)
+ throws IOException {
+ Method endpointMethod = _resourceInfo.getResourceMethod();
+ Object originalRequestMethod = requestContext.getProperty(ServerRequestMethodCaptureFilter.REQUEST_METHOD_PROPERTY);
+ String requestMethod = originalRequestMethod instanceof String ? (String) originalRequestMethod
+ : requestContext.getMethod();
+ if (GET.equals(requestMethod) && HealthCheckResource.class.equals(_resourceInfo.getResourceClass())
+ && endpointMethod != null && endpointMethod.isAnnotationPresent(ServerPublicAccess.class)) {
+ return;
+ }
+ if (TablesResource.class.equals(_resourceInfo.getResourceClass()) && endpointMethod != null
+ && endpointMethod.isAnnotationPresent(ServerDataAccess.class)) {
+ return;
+ }
+
+ HttpRequesterIdentity requesterIdentity = new HttpRequesterIdentity(_httpHeaders);
+ requesterIdentity.setEndpointUrl(requestContext.getUriInfo().getRequestUri().toString());
+ AuthorizationResult authorizationResult = _accessControlFactory.create().authorizeAdminAccess(requesterIdentity);
+ if (!authorizationResult.hasAccess()) {
+ throw new WebApplicationException(Response.status(Response.Status.FORBIDDEN)
+ .type(MediaType.TEXT_PLAIN_TYPE).entity("Forbidden").build());
+ }
+ }
+}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/ServerAdminApiAccessControlHandler.java b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerAdminApiAccessControlHandler.java
new file mode 100644
index 00000000000..6e31a46cd4a
--- /dev/null
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerAdminApiAccessControlHandler.java
@@ -0,0 +1,89 @@
+/**
+ * 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.pinot.server.api;
+
+import java.util.List;
+import javax.ws.rs.NotAuthorizedException;
+import javax.ws.rs.core.HttpHeaders;
+import org.apache.pinot.server.access.AccessControlFactory;
+import org.apache.pinot.server.access.HttpRequesterIdentity;
+import org.apache.pinot.spi.auth.AuthorizationResult;
+import org.glassfish.grizzly.http.server.HttpHandler;
+import org.glassfish.grizzly.http.server.Request;
+import org.glassfish.grizzly.http.server.Response;
+import org.glassfish.grizzly.http.server.StaticHttpHandlerBase;
+import org.glassfish.grizzly.http.util.HttpStatus;
+
+
+/// Applies Server administrative authorization to a directly registered Grizzly HTTP handler.
+///
+/// The wrapped handler's lifecycle is delegated. File caching is disabled for static handlers because Grizzly serves
+/// cache hits before invoking this wrapper's authorization check. The wrapper stores no mutable request state and is
+/// safe for concurrent use when the delegate and access-control factory are safe for concurrent use.
+public class ServerAdminApiAccessControlHandler extends HttpHandler {
+ private static final String UNAUTHORIZED_MESSAGE = "Unauthorized";
+ private static final String FORBIDDEN_MESSAGE = "Forbidden";
+
+ private final HttpHandler _delegate;
+ private final AccessControlFactory _accessControlFactory;
+
+ public ServerAdminApiAccessControlHandler(HttpHandler delegate, AccessControlFactory accessControlFactory) {
+ super(delegate.getName());
+ _delegate = delegate;
+ _accessControlFactory = accessControlFactory;
+ if (delegate instanceof StaticHttpHandlerBase) {
+ // Grizzly serves file-cache hits before HttpHandler.service(), which would bypass this authorization wrapper.
+ ((StaticHttpHandlerBase) delegate).setFileCacheEnabled(false);
+ }
+ }
+
+ @Override
+ public void service(Request request, Response response)
+ throws Exception {
+ AuthorizationResult authorizationResult;
+ try {
+ authorizationResult =
+ _accessControlFactory.create().authorizeAdminAccess(new HttpRequesterIdentity(request));
+ } catch (NotAuthorizedException e) {
+ List<Object> challenges = e.getChallenges();
+ if (challenges != null) {
+ for (Object challenge : challenges) {
+ response.addHeader(HttpHeaders.WWW_AUTHENTICATE, challenge.toString());
+ }
+ }
+ response.sendError(HttpStatus.UNAUTHORIZED_401.getStatusCode(), UNAUTHORIZED_MESSAGE);
+ return;
+ }
+ if (!authorizationResult.hasAccess()) {
+ response.sendError(HttpStatus.FORBIDDEN_403.getStatusCode(), FORBIDDEN_MESSAGE);
+ return;
+ }
+ _delegate.service(request, response);
+ }
+
+ @Override
+ public void start() {
+ _delegate.start();
+ }
+
+ @Override
+ public void destroy() {
+ _delegate.destroy();
+ }
+}
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerDataAccess.java
similarity index 51%
copy from pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
copy to pinot-server/src/main/java/org/apache/pinot/server/api/ServerDataAccess.java
index e23a46fe83b..efffa15d419 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerDataAccess.java
@@ -16,27 +16,21 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pinot.server.access;
+package org.apache.pinot.server.api;
-import io.netty.channel.ChannelHandlerContext;
-import org.apache.pinot.spi.auth.server.RequesterIdentity;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
-public class AllowAllAccessFactory implements AccessControlFactory {
- private static final AccessControl ALLOW_ALL_ACCESS = new AccessControl() {
- @Override
- public boolean isAuthorizedChannel(ChannelHandlerContext channelHandlerContext) {
- return true;
- }
-
- @Override
- public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
- return true;
- }
- };
-
- @Override
- public AccessControl create() {
- return ALLOW_ALL_ACCESS;
- }
+/// Marks a built-in Server table-data method that performs authorization within the request handler.
+///
+/// The Server request filter honors this marker only on [org.apache.pinot.server.api.resources.TablesResource]. Those
+/// methods bypass centralized administrative authorization and must call `ServerResourceUtils.validateDataAccess(...)`
+/// before accessing table or segment data. Methods on custom resource classes remain privileged even if they carry
+/// this annotation.
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.METHOD)
+public @interface ServerDataAccess {
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerPublicAccess.java
similarity index 56%
copy from pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
copy to pinot-server/src/main/java/org/apache/pinot/server/api/ServerPublicAccess.java
index e23a46fe83b..68226e9ceb0 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerPublicAccess.java
@@ -16,27 +16,19 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pinot.server.access;
+package org.apache.pinot.server.api;
-import io.netty.channel.ChannelHandlerContext;
-import org.apache.pinot.spi.auth.server.RequesterIdentity;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
-public class AllowAllAccessFactory implements AccessControlFactory {
- private static final AccessControl ALLOW_ALL_ACCESS = new AccessControl() {
- @Override
- public boolean isAuthorizedChannel(ChannelHandlerContext channelHandlerContext) {
- return true;
- }
-
- @Override
- public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
- return true;
- }
- };
-
- @Override
- public AccessControl create() {
- return ALLOW_ALL_ACCESS;
- }
+/// Marks a built-in Server health or readiness method as intentionally unauthenticated.
+///
+/// The Server request filter honors this marker only on [org.apache.pinot.server.api.resources.HealthCheckResource].
+/// Methods on custom resource classes remain privileged even if they carry this annotation.
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface ServerPublicAccess {
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerRequestMethodCaptureFilter.java
similarity index 50%
copy from pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
copy to pinot-server/src/main/java/org/apache/pinot/server/api/ServerRequestMethodCaptureFilter.java
index e23a46fe83b..81f7a09688c 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/access/AllowAllAccessFactory.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/ServerRequestMethodCaptureFilter.java
@@ -16,27 +16,25 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pinot.server.access;
+package org.apache.pinot.server.api;
-import io.netty.channel.ChannelHandlerContext;
-import org.apache.pinot.spi.auth.server.RequesterIdentity;
+import java.io.IOException;
+import javax.annotation.Priority;
+import javax.ws.rs.Priorities;
+import javax.ws.rs.container.ContainerRequestContext;
+import javax.ws.rs.container.ContainerRequestFilter;
+import javax.ws.rs.container.PreMatching;
-public class AllowAllAccessFactory implements AccessControlFactory {
- private static final AccessControl ALLOW_ALL_ACCESS = new AccessControl() {
- @Override
- public boolean isAuthorizedChannel(ChannelHandlerContext channelHandlerContext) {
- return true;
- }
-
- @Override
- public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
- return true;
- }
- };
+/// Captures the incoming HTTP method before Jersey maps implicit HEAD requests onto GET resource methods.
+@PreMatching
+@Priority(Priorities.AUTHENTICATION - 1)
+public class ServerRequestMethodCaptureFilter implements ContainerRequestFilter {
+ static final String REQUEST_METHOD_PROPERTY = ServerRequestMethodCaptureFilter.class.getName() + ".requestMethod";
@Override
- public AccessControl create() {
- return ALLOW_ALL_ACCESS;
+ public void filter(ContainerRequestContext requestContext)
+ throws IOException {
+ requestContext.setProperty(REQUEST_METHOD_PROPERTY, requestContext.getMethod());
}
}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/DefaultExceptionMapper.java b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/DefaultExceptionMapper.java
index 7522c81fe01..05ab28d11c3 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/DefaultExceptionMapper.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/DefaultExceptionMapper.java
@@ -33,7 +33,7 @@ public class DefaultExceptionMapper implements ExceptionMapper<WebApplicationExc
@Override
public Response toResponse(final WebApplicationException exception) {
Response.ResponseBuilder builder =
- Response.status(exception.getResponse().getStatus()).entity(toJson(exception)).type(MediaType.APPLICATION_JSON);
+ Response.fromResponse(exception.getResponse()).entity(toJson(exception)).type(MediaType.APPLICATION_JSON);
return builder.build();
}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/HealthCheckResource.java b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/HealthCheckResource.java
index 9291070dd97..4ef0498518b 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/HealthCheckResource.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/HealthCheckResource.java
@@ -45,6 +45,7 @@ import org.apache.pinot.core.auth.Actions;
import org.apache.pinot.core.auth.Authorize;
import org.apache.pinot.core.auth.TargetType;
import org.apache.pinot.server.api.AdminApiApplication;
+import org.apache.pinot.server.api.ServerPublicAccess;
/// REST API to do health check through ServiceStatus.
@@ -68,6 +69,7 @@ public class HealthCheckResource {
@GET
@Path("/health")
+ @ServerPublicAccess
@Produces(MediaType.TEXT_PLAIN)
@ApiOperation(value = "Checking server health")
@ApiResponses(value = {
@@ -86,6 +88,7 @@ public class HealthCheckResource {
@GET
@Path("/health/liveness")
+ @ServerPublicAccess
@Produces(MediaType.TEXT_PLAIN)
@ApiOperation(value = "Checking server liveness status.")
@ApiResponses(value = {
@@ -99,6 +102,7 @@ public class HealthCheckResource {
@GET
@Path("/health/readiness")
+ @ServerPublicAccess
@Produces(MediaType.TEXT_PLAIN)
@ApiOperation(value = "Checking server readiness status")
@ApiResponses(value = {
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
index 67c98124611..d860f9c37df 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
@@ -109,6 +109,7 @@ import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.server.access.AccessControlFactory;
import org.apache.pinot.server.api.AdminApiApplication;
+import org.apache.pinot.server.api.ServerDataAccess;
import org.apache.pinot.server.starter.ServerInstance;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -562,6 +563,7 @@ public class TablesResource {
@GET
@Produces(MediaType.APPLICATION_OCTET_STREAM)
@Path("/segments/{tableNameWithType}/{segmentName}")
+ @ServerDataAccess
@Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType", action = Actions.Table.DOWNLOAD_SEGMENT)
@ApiOperation(value = "Download an immutable segment", notes = "Download an immutable segment in zipped tar format.")
public Response downloadSegment(
@@ -618,6 +620,7 @@ public class TablesResource {
@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("/segments/{tableNameWithType}/{segmentName}/validDocIdsBitmap")
+ @ServerDataAccess
@ApiOperation(value = "Download validDocIds bitmap for an REALTIME immutable segment", notes =
"Download validDocIds for " + "an immutable segment in bitmap format.")
public ValidDocIdsBitmapResponse downloadValidDocIdsBitmap(
diff --git a/pinot-server/src/test/java/org/apache/pinot/server/api/AccessControlTest.java b/pinot-server/src/test/java/org/apache/pinot/server/api/AccessControlTest.java
index 4c4231bab11..c87da3866ef 100644
--- a/pinot-server/src/test/java/org/apache/pinot/server/api/AccessControlTest.java
+++ b/pinot-server/src/test/java/org/apache/pinot/server/api/AccessControlTest.java
@@ -22,9 +22,10 @@ import io.netty.channel.ChannelHandlerContext;
import java.io.File;
import java.io.IOException;
import java.net.ServerSocket;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import javax.ws.rs.NotAuthorizedException;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.HttpHeaders;
@@ -34,11 +35,13 @@ import org.apache.helix.HelixManager;
import org.apache.pinot.common.auth.BasicAuthTokenUtils;
import org.apache.pinot.common.config.TlsConfig;
import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.core.data.manager.InstanceDataManager;
import org.apache.pinot.core.transport.HttpServerThreadPoolConfig;
import org.apache.pinot.core.transport.ListenerConfig;
+import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.segment.local.utils.ServerReloadJobStatusCache;
import org.apache.pinot.server.access.AccessControl;
-import org.apache.pinot.server.access.AccessControlFactory;
import org.apache.pinot.server.access.BasicAuthAccessFactory;
import org.apache.pinot.server.access.GrpcRequesterIdentity;
import org.apache.pinot.server.access.HttpRequesterIdentity;
@@ -60,10 +63,18 @@ import static org.mockito.Mockito.when;
public class AccessControlTest {
private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(), "AccessControlTest");
- protected static final String TABLE_NAME = "testTable";
+ private static final String TABLE_NAME = "testTable";
+ private static final String REALTIME_TABLE_NAME = TABLE_NAME + "_REALTIME";
+ private static final String ADMIN_TOKEN = BasicAuthTokenUtils.toBasicAuthToken("admin123", "verysecret");
+ private static final String ADMIN_TOKEN_WITHOUT_PADDING = ADMIN_TOKEN.replace("=", "");
+ private static final String DATA_TOKEN = BasicAuthTokenUtils.toBasicAuthToken("user456", "kindasecret");
+ private static final String OTHER_TABLE_TOKEN = BasicAuthTokenUtils.toBasicAuthToken("user789", "othersecret");
+ private static final String INVALID_TOKEN = BasicAuthTokenUtils.toBasicAuthToken("unknown", "wrongpassword");
private AdminApiApplication _adminApiApplication;
- protected WebTarget _webTarget;
+ private WebTarget _webTarget;
+ private AccessControl _accessControl;
+ private String _instanceId;
@BeforeClass
public void setUp()
@@ -71,11 +82,14 @@ public class AccessControlTest {
FileUtils.deleteQuietly(INDEX_DIR);
Assert.assertTrue(INDEX_DIR.mkdirs());
- // Mock the instance data manager
- // Mock the server instance
+ InstanceDataManager instanceDataManager = mock(InstanceDataManager.class);
+ TableDataManager tableDataManager = mock(TableDataManager.class);
+ when(instanceDataManager.getTableDataManager(REALTIME_TABLE_NAME)).thenReturn(tableDataManager);
+
ServerInstance serverInstance = mock(ServerInstance.class);
when(serverInstance.getServerMetrics()).thenReturn(mock(ServerMetrics.class));
when(serverInstance.getHelixManager()).thenReturn(mock(HelixManager.class));
+ when(serverInstance.getInstanceDataManager()).thenReturn(instanceDataManager);
PinotConfiguration serverConf = new PinotConfiguration();
String hostname = serverConf.getProperty(CommonConstants.Helix.KEY_OF_SERVER_NETTY_HOST,
@@ -83,9 +97,27 @@ public class AccessControlTest {
? NetUtils.getHostnameOrAddress() : NetUtils.getHostAddress());
int port = serverConf.getProperty(CommonConstants.Helix.KEY_OF_SERVER_NETTY_PORT,
CommonConstants.Helix.DEFAULT_SERVER_NETTY_PORT);
- serverConf.setProperty(CommonConstants.Server.CONFIG_OF_INSTANCE_ID,
- CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE + hostname + "_" + port);
- _adminApiApplication = new AdminApiApplication(serverInstance, new DenyAllAccessFactory(),
+ _instanceId = CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE + hostname + "_" + port;
+ serverConf.setProperty(CommonConstants.Server.CONFIG_OF_INSTANCE_ID, _instanceId);
+
+ BasicAuthAccessFactory basicAuthAccessFactory = new BasicAuthAccessFactory();
+ basicAuthAccessFactory.init(new PinotConfiguration(Map.of(
+ "principals", "admin123,user456,user789",
+ "principals.admin123.password", "verysecret",
+ "principals.admin123.permissions", "admin",
+ "principals.user456.password", "kindasecret",
+ "principals.user456.tables", TABLE_NAME,
+ "principals.user456.permissions", "read",
+ "principals.user789.password", "othersecret",
+ "principals.user789.tables", "otherTable")));
+ _accessControl = basicAuthAccessFactory.create();
+
+ ServiceStatus.ServiceStatusCallback serviceStatusCallback = mock(ServiceStatus.ServiceStatusCallback.class);
+ when(serviceStatusCallback.getServiceStatus()).thenReturn(ServiceStatus.Status.GOOD);
+ when(serviceStatusCallback.getStatusDescription()).thenReturn("Ready");
+ ServiceStatus.setServiceStatusCallback(_instanceId, serviceStatusCallback);
+
+ _adminApiApplication = new AdminApiApplication(serverInstance, basicAuthAccessFactory,
mock(ServerReloadJobStatusCache.class),
serverConf);
@@ -101,86 +133,145 @@ public class AccessControlTest {
@AfterClass
public void tearDown() {
_adminApiApplication.stop();
+ ServiceStatus.removeServiceStatusCallback(_instanceId);
FileUtils.deleteQuietly(INDEX_DIR);
}
@Test
- public void testAccessDenied() {
- String segmentPath = "/segments/testTable_REALTIME/segment_name";
- // Download any segment will be denied.
- Response response = _webTarget.path(segmentPath).request().get(Response.class);
- Assert.assertNotEquals(response.getStatus(), Response.Status.FORBIDDEN);
+ public void testAdministrativeHttpAccess() {
+ assertGetStatus("/appconfigs", null, Response.Status.UNAUTHORIZED);
+ assertGetStatus("/appconfigs", INVALID_TOKEN, Response.Status.UNAUTHORIZED);
+ assertGetStatus("/appconfigs", DATA_TOKEN, Response.Status.FORBIDDEN);
+ assertGetStatus("/appconfigs", OTHER_TABLE_TOKEN, Response.Status.FORBIDDEN);
+ assertGetStatus("/appconfigs", ADMIN_TOKEN, Response.Status.OK);
}
- public static class DenyAllAccessFactory implements AccessControlFactory {
- private static final AccessControl DENY_ALL_ACCESS = new AccessControl() {
- @Override
- public boolean isAuthorizedChannel(ChannelHandlerContext channelHandlerContext) {
- return false;
- }
+ @Test
+ public void testStaticSwaggerHttpAccess() {
+ for (String path : List.of("/api/index.html", "/swaggerui-dist/index.html")) {
+ assertGetStatus(path, null, Response.Status.UNAUTHORIZED);
+ assertGetStatus(path, INVALID_TOKEN, Response.Status.UNAUTHORIZED);
+ assertGetStatus(path, DATA_TOKEN, Response.Status.FORBIDDEN);
+ assertGetStatus(path, ADMIN_TOKEN, Response.Status.OK);
+ // An authorized response must not populate Grizzly's listener-wide file cache and bypass later checks.
+ assertGetStatus(path, null, Response.Status.UNAUTHORIZED);
+ }
+ }
- @Override
- public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
- return false;
- }
- };
+ @Test
+ public void testOnlyHealthAndReadinessRoutesArePublic() {
+ assertGetStatus("/health", null, Response.Status.OK);
+ assertGetStatus("/health/liveness", null, Response.Status.OK);
+ assertGetStatus("/health/readiness", null, Response.Status.OK);
+ assertHeadStatus("/health", null, Response.Status.UNAUTHORIZED);
- @Override
- public AccessControl create() {
- return DENY_ALL_ACCESS;
+ for (String path : List.of("/uptime", "/start-time")) {
+ assertGetStatus(path, null, Response.Status.UNAUTHORIZED);
+ assertGetStatus(path, DATA_TOKEN, Response.Status.FORBIDDEN);
+ assertGetStatus(path, ADMIN_TOKEN, Response.Status.OK);
+ }
+ }
+
+ @Test
+ public void testSegmentDataAccessStillUsesTablePermission() {
+ for (String path : List.of(
+ "/segments/testTable_REALTIME/segment_name",
+ "/segments/testTable_REALTIME/segment_name/validDocIdsBitmap")) {
+ // The authorized table principal reaches the resource and receives its missing-segment result.
+ assertGetStatus(path, DATA_TOKEN, Response.Status.NOT_FOUND);
+ // A valid principal without access to this table is rejected by the existing table-level check.
+ assertGetStatus(path, OTHER_TABLE_TOKEN, Response.Status.FORBIDDEN);
}
}
@Test
public void testGrpcBasicAuth() {
- testBasicAuth(new GrpcRequesterIdentity(
- Map.of("authorization", BasicAuthTokenUtils.toBasicAuthToken("admin123", "verysecret"))), true);
- testBasicAuth(new GrpcRequesterIdentity(
- Map.of("authorization", BasicAuthTokenUtils.toBasicAuthToken("user456", "kindasecret"))), false);
-
- testBasicAuth(new GrpcRequesterIdentity(
- Map.of("authorization", "Basic YWRtaW4xMjM6dmVyeXNlY3JldA")), true);
- testBasicAuth(new GrpcRequesterIdentity(
- Map.of("authorization", "Basic dXNlcjQ1NjpraW5kYXNlY3JldA==")), false);
+ testBasicAuth(new GrpcRequesterIdentity(Map.of("authorization", ADMIN_TOKEN)), true);
+ testBasicAuth(new GrpcRequesterIdentity(Map.of("authorization", DATA_TOKEN)), false);
+ testBasicAuth(new GrpcRequesterIdentity(Map.of("authorization", ADMIN_TOKEN_WITHOUT_PADDING)), true);
}
@Test
public void testHttpBasicAuth() {
HttpHeaders headers = new ContainerRequest(null, null, null, null, new MapPropertiesDelegate());
- headers.getRequestHeaders()
- .put("authorization", Arrays.asList(
- BasicAuthTokenUtils.toBasicAuthToken("admin123", "verysecret")));
+ headers.getRequestHeaders().put("authorization", List.of(ADMIN_TOKEN));
testBasicAuth(new HttpRequesterIdentity(headers), true);
- headers.getRequestHeaders()
- .put("authorization", Arrays.asList(BasicAuthTokenUtils.toBasicAuthToken("user456", "kindasecret")));
+ headers.getRequestHeaders().put("authorization", List.of(DATA_TOKEN));
testBasicAuth(new HttpRequesterIdentity(headers), false);
- headers.getRequestHeaders().put("authorization", Arrays.asList("Basic YWRtaW4xMjM6dmVyeXNlY3JldA"));
+ headers.getRequestHeaders().put("authorization", List.of(OTHER_TABLE_TOKEN));
+ HttpRequesterIdentity tableOnlyIdentity = new HttpRequesterIdentity(headers);
+ Assert.assertFalse(_accessControl.authorizeAdminAccess(tableOnlyIdentity).hasAccess());
+ Assert.assertTrue(_accessControl.hasDataAccess(tableOnlyIdentity, "otherTable"));
+ Assert.assertFalse(_accessControl.hasDataAccess(tableOnlyIdentity, TABLE_NAME));
+ headers.getRequestHeaders().put("authorization", List.of(ADMIN_TOKEN_WITHOUT_PADDING));
testBasicAuth(new HttpRequesterIdentity(headers), true);
- headers.getRequestHeaders().put("authorization", Arrays.asList("Basic dXNlcjQ1NjpraW5kYXNlY3JldA=="));
- testBasicAuth(new HttpRequesterIdentity(headers), false);
+
+ HttpHeaders missingHeaders = new ContainerRequest(null, null, null, null, new MapPropertiesDelegate());
+ Assert.expectThrows(NotAuthorizedException.class,
+ () -> _accessControl.authorizeAdminAccess(new HttpRequesterIdentity(missingHeaders)));
+ headers.getRequestHeaders().put("authorization", List.of(INVALID_TOKEN));
+ Assert.expectThrows(NotAuthorizedException.class,
+ () -> _accessControl.authorizeAdminAccess(new HttpRequesterIdentity(headers)));
+ }
+
+ @Test
+ public void testHttpIdentityPreservesHeadersAndUsesCaseInsensitiveLookup() {
+ HttpHeaders headers = new ContainerRequest(null, null, null, null, new MapPropertiesDelegate());
+ headers.getRequestHeaders().put(HttpHeaders.AUTHORIZATION, List.of(ADMIN_TOKEN));
+ headers.getRequestHeaders().put("X-Custom", List.of("value"));
+
+ HttpRequesterIdentity identity = new HttpRequesterIdentity(headers);
+
+ Assert.assertEquals(identity.getHttpHeaders().keySet(), Set.of(HttpHeaders.AUTHORIZATION, "X-Custom"));
+ Assert.assertEquals(identity.getHeaderValues("authorization"), List.of(ADMIN_TOKEN));
}
- public void testBasicAuth(RequesterIdentity requesterIdentity, boolean isAdmin) {
- final BasicAuthAccessFactory basicAuthAccessFactory = new BasicAuthAccessFactory();
- PinotConfiguration config = new PinotConfiguration(Map.of("principals", "admin123,user456",
- "principals.admin123.password", "verysecret", "principals.user456.password", "kindasecret",
- "principals.user456.tables", "stuff,lessImportantStuff"));
- basicAuthAccessFactory.init(config);
- final AccessControl accessControl = basicAuthAccessFactory.create();
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "stuff"));
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "stuff_OFFLINE"));
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "stuff_REALTIME"));
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "lessImportantStuff"));
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "lessImportantStuff_OFFLINE"));
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "lessImportantStuff_REALTIME"));
+ @Test
+ public void testLegacyAccessControlDefaultsToAdminDenied() {
+ AccessControl legacyAccessControl = new AccessControl() {
+ @Override
+ public boolean isAuthorizedChannel(ChannelHandlerContext channelHandlerContext) {
+ return true;
+ }
+
+ @Override
+ public boolean hasDataAccess(RequesterIdentity requesterIdentity, String tableName) {
+ return true;
+ }
+ };
+
+ Assert.assertFalse(legacyAccessControl.authorizeAdminAccess(new GrpcRequesterIdentity(Map.of())).hasAccess());
+ }
+
+ private void testBasicAuth(RequesterIdentity requesterIdentity, boolean isAdmin) {
+ Assert.assertTrue(_accessControl.hasDataAccess(requesterIdentity, TABLE_NAME));
+ Assert.assertTrue(_accessControl.hasDataAccess(requesterIdentity, TABLE_NAME + "_OFFLINE"));
+ Assert.assertTrue(_accessControl.hasDataAccess(requesterIdentity, REALTIME_TABLE_NAME));
if (isAdmin) {
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "myTable"));
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "myTable_OFFLINE"));
- Assert.assertTrue(accessControl.hasDataAccess(requesterIdentity, "myTable_REALTIME"));
+ Assert.assertTrue(_accessControl.authorizeAdminAccess(requesterIdentity).hasAccess());
+ Assert.assertTrue(_accessControl.hasDataAccess(requesterIdentity, "myTable"));
} else {
- Assert.assertFalse(accessControl.hasDataAccess(requesterIdentity, "myTable"));
- Assert.assertFalse(accessControl.hasDataAccess(requesterIdentity, "myTable_OFFLINE"));
- Assert.assertFalse(accessControl.hasDataAccess(requesterIdentity, "myTable_REALTIME"));
+ Assert.assertFalse(_accessControl.authorizeAdminAccess(requesterIdentity).hasAccess());
+ Assert.assertFalse(_accessControl.hasDataAccess(requesterIdentity, "myTable"));
+ }
+ }
+
+ private void assertGetStatus(String path, String authorization, Response.Status expectedStatus) {
+ try (Response response = authorization == null
+ ? _webTarget.path(path).request().get(Response.class)
+ : _webTarget.path(path).request().header(HttpHeaders.AUTHORIZATION, authorization).get(Response.class)) {
+ Assert.assertEquals(response.getStatus(), expectedStatus.getStatusCode(), path);
+ if (expectedStatus == Response.Status.UNAUTHORIZED) {
+ Assert.assertEquals(response.getHeaderString(HttpHeaders.WWW_AUTHENTICATE), "Basic", path);
+ }
+ }
+ }
+
+ private void assertHeadStatus(String path, String authorization, Response.Status expectedStatus) {
+ try (Response response = authorization == null
+ ? _webTarget.path(path).request().head()
+ : _webTarget.path(path).request().header(HttpHeaders.AUTHORIZATION, authorization).head()) {
+ Assert.assertEquals(response.getStatus(), expectedStatus.getStatusCode(), path);
}
}
diff --git a/pinot-server/src/test/java/org/apache/pinot/server/api/PinotServerAppConfigsTest.java b/pinot-server/src/test/java/org/apache/pinot/server/api/PinotServerAppConfigsTest.java
index 84668002ad6..c495778782b 100644
--- a/pinot-server/src/test/java/org/apache/pinot/server/api/PinotServerAppConfigsTest.java
+++ b/pinot-server/src/test/java/org/apache/pinot/server/api/PinotServerAppConfigsTest.java
@@ -39,7 +39,7 @@ public class PinotServerAppConfigsTest extends BaseResourceTest {
///
/// @throws JsonProcessingException In case an exception is encountered during JSON processing.
@Test
- public void testAppConfigs()
+ public void testAppConfigsWithAllowAllAccessControl()
throws JsonProcessingException, SocketException, UnknownHostException {
PinotConfiguration expectedServerConf = new PinotConfiguration();
String hostname = expectedServerConf.getProperty(CommonConstants.Helix.KEY_OF_SERVER_NETTY_HOST,
@@ -52,6 +52,7 @@ public class PinotServerAppConfigsTest extends BaseResourceTest {
PinotAppConfigs expected = new PinotAppConfigs(expectedServerConf);
Response response = _webTarget.path("/appconfigs").request().get(Response.class);
+ Assert.assertEquals(response.getStatus(), Response.Status.OK.getStatusCode());
String configsJson = response.readEntity(String.class);
PinotAppConfigs actual = JsonUtils.stringToObject(configsJson, PinotAppConfigs.class);
diff --git a/pinot-server/src/test/java/org/apache/pinot/server/api/ServerAdminApiAccessControlFilterTest.java b/pinot-server/src/test/java/org/apache/pinot/server/api/ServerAdminApiAccessControlFilterTest.java
new file mode 100644
index 00000000000..9a4361a7b9e
--- /dev/null
+++ b/pinot-server/src/test/java/org/apache/pinot/server/api/ServerAdminApiAccessControlFilterTest.java
@@ -0,0 +1,213 @@
+/**
+ * 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.pinot.server.api;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.net.URI;
+import java.util.List;
+import javax.ws.rs.NotAuthorizedException;
+import javax.ws.rs.container.ContainerRequestContext;
+import javax.ws.rs.container.ResourceInfo;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.MultivaluedHashMap;
+import javax.ws.rs.core.Response.Status;
+import javax.ws.rs.core.UriInfo;
+import org.apache.pinot.server.access.AccessControl;
+import org.apache.pinot.server.access.AccessControlFactory;
+import org.apache.pinot.server.access.HttpRequesterIdentity;
+import org.apache.pinot.server.api.resources.HealthCheckResource;
+import org.apache.pinot.spi.auth.BasicAuthorizationResultImpl;
+import org.apache.pinot.spi.auth.server.RequesterIdentity;
+import org.glassfish.grizzly.http.server.HttpHandler;
+import org.glassfish.grizzly.http.server.Request;
+import org.glassfish.grizzly.http.server.Response;
+import org.mockito.ArgumentCaptor;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+
+/// Tests the fail-closed classification and request identity supplied by the Server administrative filter.
+public class ServerAdminApiAccessControlFilterTest {
+ private static final URI HEALTH_URI = URI.create("http://localhost:8098/health");
+
+ @Test
+ public void testOnlyAnnotatedHealthGetIsPublic()
+ throws Exception {
+ Method healthMethod = HealthCheckResource.class.getMethod("checkHealth", String.class);
+
+ AccessControlFactory accessControlFactory = mock(AccessControlFactory.class);
+ ServerAdminApiAccessControlFilter filter = createFilter(accessControlFactory,
+ resourceInfo(HealthCheckResource.class, healthMethod));
+ filter.filter(request("GET", HEALTH_URI));
+ verifyNoInteractions(accessControlFactory);
+
+ assertAdministrativeAccess(HealthCheckResource.class, healthMethod, "HEAD");
+
+ Method customHealthMethod = CustomHealthResource.class.getMethod("checkHealth");
+ assertAdministrativeAccess(CustomHealthResource.class, customHealthMethod, "GET");
+
+ Method customDataMethod = CustomDataResource.class.getMethod("getData");
+ assertAdministrativeAccess(CustomDataResource.class, customDataMethod, "GET");
+ }
+
+ @Test
+ public void testStaticHandlerPreservesAllAuthenticationChallenges()
+ throws Exception {
+ HttpHandler delegate = mock(HttpHandler.class);
+ AccessControlFactory accessControlFactory = mock(AccessControlFactory.class);
+ AccessControl accessControl = mock(AccessControl.class);
+ when(accessControlFactory.create()).thenReturn(accessControl);
+ when(accessControl.authorizeAdminAccess(any()))
+ .thenThrow(new NotAuthorizedException((Object) "Basic", "Bearer"));
+
+ Request request = mock(Request.class);
+ when(request.getHeaderNames()).thenReturn(List.of());
+ when(request.getRequestURL()).thenReturn(new StringBuilder("http://localhost:8098/api/index.html"));
+ Response response = mock(Response.class);
+
+ new ServerAdminApiAccessControlHandler(delegate, accessControlFactory).service(request, response);
+
+ verify(response).addHeader(HttpHeaders.WWW_AUTHENTICATE, "Basic");
+ verify(response).addHeader(HttpHeaders.WWW_AUTHENTICATE, "Bearer");
+ verify(response).sendError(401, "Unauthorized");
+ verify(delegate, never()).service(any(), any());
+ }
+
+ @Test
+ public void testStaticHandlerSuppliesRequestIdentity()
+ throws Exception {
+ HttpHandler delegate = mock(HttpHandler.class);
+ AccessControlFactory accessControlFactory = mock(AccessControlFactory.class);
+ AccessControl accessControl = mock(AccessControl.class);
+ when(accessControlFactory.create()).thenReturn(accessControl);
+ when(accessControl.authorizeAdminAccess(any())).thenReturn(new BasicAuthorizationResultImpl(true));
+
+ Request request = mock(Request.class);
+ when(request.getHeaderNames()).thenReturn(List.of("X-Custom"));
+ when(request.getHeaders("X-Custom")).thenReturn(List.of("value"));
+ when(request.getRequestURL()).thenReturn(new StringBuilder("http://localhost:8098/api/index.html"));
+ when(request.getQueryString()).thenReturn("mode=test");
+ Response response = mock(Response.class);
+
+ new ServerAdminApiAccessControlHandler(delegate, accessControlFactory).service(request, response);
+
+ ArgumentCaptor<RequesterIdentity> identityCaptor = ArgumentCaptor.forClass(RequesterIdentity.class);
+ verify(accessControl).authorizeAdminAccess(identityCaptor.capture());
+ HttpRequesterIdentity identity = (HttpRequesterIdentity) identityCaptor.getValue();
+ Assert.assertEquals(identity.getEndpointUrl(), "http://localhost:8098/api/index.html?mode=test");
+ Assert.assertEquals(identity.getHeaderValues("x-custom"), List.of("value"));
+ verify(delegate).service(request, response);
+ }
+
+ @Test
+ public void testStaticHandlerAcceptsAuthenticationExceptionWithoutChallenge()
+ throws Exception {
+ HttpHandler delegate = mock(HttpHandler.class);
+ AccessControlFactory accessControlFactory = mock(AccessControlFactory.class);
+ AccessControl accessControl = mock(AccessControl.class);
+ when(accessControlFactory.create()).thenReturn(accessControl);
+ when(accessControl.authorizeAdminAccess(any()))
+ .thenThrow(new NotAuthorizedException(javax.ws.rs.core.Response.status(Status.UNAUTHORIZED).build()));
+
+ Request request = mock(Request.class);
+ when(request.getHeaderNames()).thenReturn(List.of());
+ when(request.getRequestURL()).thenReturn(new StringBuilder("http://localhost:8098/api/index.html"));
+ Response response = mock(Response.class);
+
+ new ServerAdminApiAccessControlHandler(delegate, accessControlFactory).service(request, response);
+
+ verify(response).sendError(401, "Unauthorized");
+ verify(response, never()).addHeader(any(String.class), any(String.class));
+ verify(delegate, never()).service(any(), any());
+ }
+
+ private static void assertAdministrativeAccess(Class<?> resourceClass, Method resourceMethod, String httpMethod)
+ throws Exception {
+ AccessControlFactory accessControlFactory = mock(AccessControlFactory.class);
+ AccessControl accessControl = mock(AccessControl.class);
+ when(accessControlFactory.create()).thenReturn(accessControl);
+ when(accessControl.authorizeAdminAccess(any())).thenReturn(new BasicAuthorizationResultImpl(true));
+ ServerAdminApiAccessControlFilter filter = createFilter(accessControlFactory,
+ resourceInfo(resourceClass, resourceMethod));
+
+ filter.filter(request(httpMethod, HEALTH_URI));
+
+ ArgumentCaptor<RequesterIdentity> identityCaptor = ArgumentCaptor.forClass(RequesterIdentity.class);
+ verify(accessControl).authorizeAdminAccess(identityCaptor.capture());
+ RequesterIdentity requesterIdentity = identityCaptor.getValue();
+ Assert.assertTrue(requesterIdentity instanceof HttpRequesterIdentity);
+ Assert.assertEquals(((HttpRequesterIdentity) requesterIdentity).getEndpointUrl(), HEALTH_URI.toString());
+ }
+
+ private static ServerAdminApiAccessControlFilter createFilter(AccessControlFactory accessControlFactory,
+ ResourceInfo resourceInfo)
+ throws Exception {
+ ServerAdminApiAccessControlFilter filter = new ServerAdminApiAccessControlFilter();
+ setField(filter, "_accessControlFactory", accessControlFactory);
+ setField(filter, "_resourceInfo", resourceInfo);
+ HttpHeaders httpHeaders = mock(HttpHeaders.class);
+ when(httpHeaders.getRequestHeaders()).thenReturn(new MultivaluedHashMap<>());
+ setField(filter, "_httpHeaders", httpHeaders);
+ return filter;
+ }
+
+ private static ResourceInfo resourceInfo(Class<?> resourceClass, Method resourceMethod) {
+ ResourceInfo resourceInfo = mock(ResourceInfo.class);
+ doReturn(resourceClass).when(resourceInfo).getResourceClass();
+ when(resourceInfo.getResourceMethod()).thenReturn(resourceMethod);
+ return resourceInfo;
+ }
+
+ private static ContainerRequestContext request(String method, URI uri) {
+ ContainerRequestContext requestContext = mock(ContainerRequestContext.class);
+ UriInfo uriInfo = mock(UriInfo.class);
+ when(uriInfo.getRequestUri()).thenReturn(uri);
+ when(requestContext.getMethod()).thenReturn(method);
+ when(requestContext.getUriInfo()).thenReturn(uriInfo);
+ return requestContext;
+ }
+
+ private static void setField(Object target, String fieldName, Object value)
+ throws Exception {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ private static class CustomHealthResource {
+ @ServerPublicAccess
+ public void checkHealth() {
+ }
+ }
+
+ private static class CustomDataResource {
+ @ServerDataAccess
+ public void getData() {
+ }
+ }
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index c117713b8c8..3b6a9c195af 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -341,6 +341,8 @@ public class CommonConstants {
public static class Broker {
public static final String ROUTING_TABLE_CONFIG_PREFIX = "pinot.broker.routing.table";
public static final String ACCESS_CONTROL_CONFIG_PREFIX = "pinot.broker.access.control";
+ /// Namespace for service credentials used by the broker when invoking Server admin APIs.
+ public static final String SERVER_ADMIN_AUTH_PREFIX = "pinot.broker.server.admin.auth";
/// Config prefix for the broker-side MaterializedViewHandler. Implementation class is
/// loaded from `pinot.broker.materialized.view.handler.class`; other settings sit
/// under the same prefix and are passed through to the handler's `init`. Default
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org