From 602b75ed08d73bf1ed02b98f3eb940cf3ad99ac0 Mon Sep 17 00:00:00 2001 From: Lukasz Antoniak Date: Thu, 20 Aug 2026 13:33:49 +0200 Subject: [PATCH] CASSANALYTICS-185: Support custom authentication mechanisms in Sidecar client --- .../sidecar/client/RequestExecutor.java | 53 ++++--- .../sidecar/client/SidecarClient.java | 134 +++++++----------- .../sidecar/client/SidecarClientConfig.java | 11 ++ .../client/SidecarClientConfigImpl.java | 53 +++++-- .../client/SidecarIdentityProvider.java | 62 ++++++++ .../client/SidecarClientConfigTest.java | 22 +++ .../sidecar/client/SidecarClientTest.java | 7 +- .../client/request/RequestExecutorTest.java | 33 ++++- .../SidecarClientVertxRequestExecutor.java | 4 +- .../sidecar/client/VertxRequestExecutor.java | 4 +- .../client/VertxSidecarClientTest.java | 2 +- .../client/request/BaseVertxRequestTest.java | 2 +- .../cdc/sidecar/SidecarCdcClient.java | 37 ++++- .../cassandra/spark/utils/MapUtils.java | 13 ++ .../cassandra/spark/utils/Properties.java | 3 + .../clients/AnalyticsSidecarClient.java | 1 + .../spark/bulkwriter/BulkSparkConf.java | 32 +++++ .../spark/bulkwriter/WriterOptions.java | 2 + .../CloudStorageDataTransferApiImpl.java | 3 +- ...oordinatedCloudStorageDataTransferApi.java | 4 +- .../spark/data/CassandraDataLayer.java | 12 +- .../clients/SidecarClientConfigTest.java | 12 ++ .../spark/bulkwriter/BulkSparkConfTest.java | 22 +++ .../CassandraDataLayerValidationTest.java | 2 +- .../build.gradle | 2 + .../SharedClusterIntegrationTestBase.java | 16 +++ .../TestAuthenticationHandlerFactory.java | 95 +++++++++++++ .../testing/TestSidecarIdentityProvider.java | 32 +++++ .../build.gradle | 10 +- .../SidecarIdentityProviderTest.java | 87 ++++++++++++ .../org/apache/cassandra/clients/Sidecar.java | 66 ++++++++- .../apache/cassandra/clients/SidecarTest.java | 49 +++++++ docs/src/user.adoc | 8 ++ gradle.properties | 3 + 34 files changed, 751 insertions(+), 147 deletions(-) create mode 100644 analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarIdentityProvider.java create mode 100644 cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestAuthenticationHandlerFactory.java create mode 100644 cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestSidecarIdentityProvider.java create mode 100644 cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SidecarIdentityProviderTest.java diff --git a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java index eb535ceff..fb16ff090 100644 --- a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java +++ b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java @@ -46,10 +46,12 @@ public class RequestExecutor implements AutoCloseable protected final HttpClient httpClient; protected final ScheduledExecutorService singleThreadExecutorService; + protected final SidecarIdentityProvider identityProvider; - public RequestExecutor(HttpClient httpClient) + public RequestExecutor(HttpClient httpClient, SidecarIdentityProvider identityProvider) { this.httpClient = requireNonNull(httpClient, "The httpClient is required"); + this.identityProvider = identityProvider; this.singleThreadExecutorService = Executors.newSingleThreadScheduledExecutor(); } @@ -57,49 +59,51 @@ public RequestExecutor(HttpClient httpClient) * Executes the request and waits if necessary for at most the configured time in the * {@link HttpClientConfig#timeoutMillis()} for this future to complete, and then returns its result, if available. * - * @param context the request context - * @param the expected type for the instance + * @param requestBuilder the request context builder + * @param the expected type for the instance * @return the result value * @throws CancellationException if this future was cancelled * @throws ExecutionException if this future completed exceptionally * @throws InterruptedException if the current thread was interrupted while waiting * @throws TimeoutException if the wait timed out */ - public T executeRequest(RequestContext context) + public T executeRequest(RequestContext.Builder requestBuilder) throws ExecutionException, InterruptedException, TimeoutException { - return executeRequest(context, httpClient.config().timeoutMillis(), TimeUnit.MILLISECONDS); + return executeRequest(requestBuilder, httpClient.config().timeoutMillis(), TimeUnit.MILLISECONDS); } /** * Executes the request and waits if necessary for at most the provided {@code timeout} with units {@code unit} * for this future to complete, and then returns its result, if available. * - * @param context the request context - * @param timeout the maximum time to wait - * @param unit the time unit of the timeout argument - * @param the expected type for the instance + * @param requestBuilder the request context builder + * @param timeout the maximum time to wait + * @param unit the time unit of the timeout argument + * @param the expected type for the instance * @return the result value * @throws CancellationException if this future was cancelled * @throws ExecutionException if this future completed exceptionally * @throws InterruptedException if the current thread was interrupted while waiting * @throws TimeoutException if the wait timed out */ - public T executeRequest(RequestContext context, long timeout, TimeUnit unit) + public T executeRequest(RequestContext.Builder requestBuilder, long timeout, TimeUnit unit) throws ExecutionException, InterruptedException, TimeoutException { - return this.executeRequestAsync(context).get(timeout, unit); + return this.executeRequestAsync(requestBuilder).get(timeout, unit); } /** * Returns the expected instance of type {@code } after executing the {@code request} and processing it. * - * @param context the request context - * @param the expected type for the instance + * @param requestBuilder the request context builder + * @param the expected type for the instance * @return the expected instance of type {@code } after executing the {@code request} and processing it */ - public CompletableFuture executeRequestAsync(RequestContext context) + public CompletableFuture executeRequestAsync(RequestContext.Builder requestBuilder) { + injectCredentials(requestBuilder); + RequestContext context = requestBuilder.build(); Iterator iterator = context.instanceSelectionPolicy().iterator(); CompletableFuture resultFuture = new CompletableFuture<>(); if (!iterator.hasNext()) @@ -116,7 +120,7 @@ public CompletableFuture executeRequestAsync(RequestContext context) executeWithRetries(responseFuture, iterator, instance, context, 1); responseFuture.whenComplete((response, retryThrowable) -> - processResponse(resultFuture, context.request(), response, retryThrowable)); + processResponse(resultFuture, context, response, retryThrowable)); return resultFuture; } @@ -124,12 +128,14 @@ public CompletableFuture executeRequestAsync(RequestContext context) /** * Streams the request from the context to the {@code streamConsumer}. * - * @param context the request context + * @param requestBuilder the request context builder * @param streamConsumer the object that consumes the stream */ - public void streamRequest(RequestContext context, StreamConsumer streamConsumer) + public void streamRequest(RequestContext.Builder requestBuilder, StreamConsumer streamConsumer) { Objects.requireNonNull(streamConsumer, "streamConsumer must be non-null"); + injectCredentials(requestBuilder); + RequestContext context = requestBuilder.build(); Iterator iterator = context.instanceSelectionPolicy().iterator(); if (!iterator.hasNext()) { @@ -331,17 +337,18 @@ private void applyRetryPolicy(CompletableFuture future, * future when an error occurred during processing. * * @param future the future for the request - * @param request the request + * @param context the request context * @param response the {@link HttpResponse} received from the server * @param throwable the error encountered during the request, or null if no error was encountered * @param the type expected by the requester */ @SuppressWarnings("unchecked") private void processResponse(CompletableFuture future, - Request request, + RequestContext context, HttpResponse response, Throwable throwable) { + Request request = context.request(); if (throwable != null) { logger.error("Failed to process request={}, response={}", request, response, throwable); @@ -381,4 +388,12 @@ protected void schedule(long delayMillis, Runnable runnable) } runnable.run(); } + + private void injectCredentials(RequestContext.Builder requestBuilder) + { + if (identityProvider != null) + { + identityProvider.injectCredentials(requestBuilder); + } + } } diff --git a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java index 406a427cd..98aa5ed31 100644 --- a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java +++ b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java @@ -117,8 +117,7 @@ public CompletableFuture sidecarHealth() { return executor.executeRequestAsync(requestBuilder() .sidecarHealthRequest() - .retryPolicy(oncePerInstanceRetryPolicy) - .build()); + .retryPolicy(oncePerInstanceRetryPolicy)); } /** @@ -132,8 +131,7 @@ public CompletableFuture sidecarHealth(SidecarInstance instance) return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) .retryPolicy(oncePerInstanceRetryPolicy) - .sidecarHealthRequest() - .build()); + .sidecarHealthRequest()); } /** @@ -147,8 +145,7 @@ public CompletableFuture cassandraHealth() { return executor.executeRequestAsync(requestBuilder() .cassandraHealthRequest() - .retryPolicy(oncePerInstanceRetryPolicy) - .build()); + .retryPolicy(oncePerInstanceRetryPolicy)); } /** @@ -160,8 +157,7 @@ public CompletableFuture cassandraNativeHealth() { return executor.executeRequestAsync(requestBuilder() .cassandraNativeHealthRequest() - .retryPolicy(new OncePerInstanceRetryPolicy()) - .build()); + .retryPolicy(new OncePerInstanceRetryPolicy())); } /** @@ -173,8 +169,7 @@ public CompletableFuture cassandraJmxHealth() { return executor.executeRequestAsync(requestBuilder() .cassandraJmxHealthRequest() - .retryPolicy(new OncePerInstanceRetryPolicy()) - .build()); + .retryPolicy(new OncePerInstanceRetryPolicy())); } /** @@ -184,7 +179,7 @@ public CompletableFuture cassandraJmxHealth() */ public CompletableFuture fullSchema() { - return executor.executeRequestAsync(requestBuilder().schemaRequest().build()); + return executor.executeRequestAsync(requestBuilder().schemaRequest()); } /** @@ -196,7 +191,7 @@ public CompletableFuture fullSchema() */ public CompletableFuture schema(String keyspace) { - return executor.executeRequestAsync(requestBuilder().schemaRequest(keyspace).build()); + return executor.executeRequestAsync(requestBuilder().schemaRequest(keyspace)); } /** @@ -208,7 +203,7 @@ public CompletableFuture schema(String keyspace) */ public CompletableFuture ring(String keyspace) { - return executor.executeRequestAsync(requestBuilder().ringRequest(keyspace).build()); + return executor.executeRequestAsync(requestBuilder().ringRequest(keyspace)); } /** @@ -218,7 +213,7 @@ public CompletableFuture ring(String keyspace) */ public CompletableFuture nodeSettings() { - return executor.executeRequestAsync(requestBuilder().nodeSettingsRequest().build()); + return executor.executeRequestAsync(requestBuilder().nodeSettingsRequest()); } /** @@ -230,8 +225,7 @@ public CompletableFuture nodeSettings() public CompletableFuture nodeSettings(SidecarInstance instance) { return executor.executeRequestAsync(requestBuilder().singleInstanceSelectionPolicy(instance) - .nodeSettingsRequest() - .build()); + .nodeSettingsRequest()); } /** @@ -241,7 +235,7 @@ public CompletableFuture nodeSettings(SidecarInstance instance) */ public CompletableFuture gossipInfo() { - return executor.executeRequestAsync(requestBuilder().gossipInfoRequest().build()); + return executor.executeRequestAsync(requestBuilder().gossipInfoRequest()); } /** @@ -253,8 +247,7 @@ public CompletableFuture gossipInfo() public CompletableFuture gossipInfo(SidecarInstance instance) { return executor.executeRequestAsync(requestBuilder().singleInstanceSelectionPolicy(instance) - .gossipInfoRequest() - .build()); + .gossipInfoRequest()); } /** @@ -266,8 +259,7 @@ public CompletableFuture gossipHealth(SidecarInstance instance) { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) - .gossipHealthRequest() - .build()); + .gossipHealthRequest()); } /** @@ -277,7 +269,7 @@ public CompletableFuture gossipHealth(SidecarInstance instance) */ public CompletableFuture timeSkew() { - return executor.executeRequestAsync(requestBuilder().timeSkewRequest().build()); + return executor.executeRequestAsync(requestBuilder().timeSkewRequest()); } /** @@ -293,8 +285,7 @@ public CompletableFuture timeSkew(List tokenRangeReplicas(List listSnapshotFiles(SidecarIns .listSnapshotFilesRequest(keyspace, table, snapshotName, - includeSecondaryIndexFiles) - .build()); + includeSecondaryIndexFiles)); } /** @@ -373,8 +362,7 @@ public CompletableFuture clearSnapshot(SidecarInstance instance, String snapshotName) { return executor.executeRequestAsync(requestBuilder().singleInstanceSelectionPolicy(instance) - .clearSnapshotRequest(keyspace, table, snapshotName) - .build()); + .clearSnapshotRequest(keyspace, table, snapshotName)); } /** @@ -416,8 +404,7 @@ public CompletableFuture createSnapshot(SidecarInstance instance, return executor.executeRequestAsync(requestBuilder().retryPolicy(ignoreConflictRetryPolicy) .singleInstanceSelectionPolicy(instance) .createSnapshotRequest(keyspace, table, - snapshotName, snapshotTTL) - .build()); + snapshotName, snapshotTTL)); } /** @@ -446,8 +433,8 @@ public void streamSSTableComponent(SidecarInstance instance, { executor.streamRequest(requestBuilder() .singleInstanceSelectionPolicy(instance) - .ssTableComponentRequest(keyspace, table, snapshotName, componentName, range) - .build(), streamConsumer); + .ssTableComponentRequest(keyspace, table, snapshotName, componentName, range), + streamConsumer); } /** @@ -467,8 +454,8 @@ public void streamSSTableComponent(SidecarInstance instance, { executor.streamRequest(requestBuilder() .singleInstanceSelectionPolicy(instance) - .ssTableComponentRequest(fileInfo, range) - .build(), streamConsumer); + .ssTableComponentRequest(fileInfo, range), + streamConsumer); } /** @@ -497,8 +484,7 @@ public CompletableFuture uploadSSTableRequest(SidecarInstance instance, uploadId, componentName, digest, - filename) - .build()); + filename)); } /** @@ -526,8 +512,7 @@ public CompletableFuture importSSTableRequest(SidecarInst 10); return executor.executeRequestAsync(requestBuilder().singleInstanceSelectionPolicy(instance) .retryPolicy(retryPolicy) - .importSSTableRequest(keyspace, table, uploadId, options) - .build()); + .importSSTableRequest(keyspace, table, uploadId, options)); } /** @@ -540,8 +525,7 @@ public CompletableFuture importSSTableRequest(SidecarInst public CompletableFuture cleanUploadSession(SidecarInstance instance, String uploadId) { return executor.executeRequestAsync(requestBuilder().singleInstanceSelectionPolicy(instance) - .cleanSSTableUploadSessionRequest(uploadId) - .build()); + .cleanSSTableUploadSessionRequest(uploadId)); } /** @@ -553,8 +537,7 @@ public CompletableFuture listCdcSegments(SidecarInstanc { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(sidecarInstance) - .request(new ListCdcSegmentsRequest()) - .build()); + .request(new ListCdcSegmentsRequest())); } /** @@ -575,8 +558,8 @@ public void streamCdcSegments(SidecarInstance sidecarInstance, { executor.streamRequest(requestBuilder() .singleInstanceSelectionPolicy(sidecarInstance) - .request(new StreamCdcSegmentRequest(segment, range)) - .build(), streamConsumer); + .request(new StreamCdcSegmentRequest(segment, range)), + streamConsumer); } /** @@ -592,8 +575,7 @@ public CompletableFuture reportSchema(SidecarInstance instance) return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) .reportSchemaRequest() - .noRetryPolicy() // {@link NoRetryPolicy} is the preferred behavior here - .build()); + .noRetryPolicy()); // {@link NoRetryPolicy} is the preferred behavior here } /** @@ -605,8 +587,7 @@ public CompletableFuture reportSchema(SidecarInstance instance) public CompletableFuture allServicesConfig() { return executor.executeRequestAsync(requestBuilder() - .request(new AllServicesConfigRequest()) - .build()); + .request(new AllServicesConfigRequest())); } /** @@ -619,8 +600,7 @@ public CompletableFuture allServicesConfig() public CompletableFuture updateCdcServiceConfig(Service service, Map config) { return executor.executeRequestAsync(requestBuilder() - .request(new UpdateServiceConfigRequest(service, new UpdateCdcServiceConfigPayload(config))) - .build()); + .request(new UpdateServiceConfigRequest(service, new UpdateCdcServiceConfigPayload(config)))); } /** @@ -631,8 +611,7 @@ public CompletableFuture updateCdcServiceConfig(S public CompletableFuture deleteCdcServiceConfig(Service service) { return executor.executeRequestAsync(requestBuilder() - .request(new DeleteServiceConfigRequest(service)) - .build()); + .request(new DeleteServiceConfigRequest(service))); } /** @@ -646,8 +625,7 @@ public CompletableFuture createRestoreJob(Strin Objects.requireNonNull(payload, "payload cannot be null"); return executor.executeRequestAsync(requestBuilder() .retryPolicy(new CreateRestoreJobRetryPolicy(defaultRetryPolicy)) - .request(new CreateRestoreJobRequest(keyspace, table, payload)) - .build()); + .request(new CreateRestoreJobRequest(keyspace, table, payload))); } /** @@ -660,8 +638,7 @@ public CompletableFuture updateRestoreJob(String keyspace, UpdateRestoreJobRequestPayload payload) { return executor.executeRequestAsync(requestBuilder() - .request(new UpdateRestoreJobRequest(keyspace, table, jobId, payload)) - .build()); + .request(new UpdateRestoreJobRequest(keyspace, table, jobId, payload))); } /** @@ -672,8 +649,7 @@ public CompletableFuture abortRestoreJob(String keyspace, String table, UU AbortRestoreJobRequestPayload payload) { return executor.executeRequestAsync(requestBuilder() - .request(new AbortRestoreJobRequest(keyspace, table, jobId, payload)) - .build()); + .request(new AbortRestoreJobRequest(keyspace, table, jobId, payload))); } /** @@ -685,8 +661,7 @@ public CompletableFuture restoreJobSummary(Str UUID jobId) { return executor.executeRequestAsync(requestBuilder() - .request(new RestoreJobSummaryRequest(keyspace, table, jobId)) - .build()); + .request(new RestoreJobSummaryRequest(keyspace, table, jobId))); } /** @@ -701,8 +676,7 @@ public CompletableFuture createRestoreJobSlice(SidecarInstance instance, { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) - .request(new CreateRestoreJobSliceRequest(keyspace, table, jobId, payload)) - .build()); + .request(new CreateRestoreJobSliceRequest(keyspace, table, jobId, payload))); } /** @@ -712,8 +686,7 @@ public CompletableFuture createRestoreJobSlice(SidecarInstance instance, public CompletableFuture createRestoreJobSlice(String keyspace, String table, UUID jobId, CreateSliceRequestPayload payload) { return executor.executeRequestAsync(requestBuilder() - .request(new CreateRestoreJobSliceRequest(keyspace, table, jobId, payload)) - .build()); + .request(new CreateRestoreJobSliceRequest(keyspace, table, jobId, payload))); } /** @@ -723,8 +696,7 @@ public CompletableFuture createRestoreJobSlice(String keyspace, String tab public CompletableFuture restoreJobProgress(RestoreJobProgressRequestParams params) { return executor.executeRequestAsync(requestBuilder() - .request(new RestoreJobProgressRequest(params)) - .build()); + .request(new RestoreJobProgressRequest(params))); } /** @@ -737,8 +709,7 @@ public CompletableFuture connectedClientStats(Side { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) - .connectedClientStatsRequest() - .build()); + .connectedClientStatsRequest()); } /** @@ -755,8 +726,7 @@ public CompletableFuture tableStats(SidecarInstance instance { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) - .tableStatsRequest(keyspace, table) - .build()); + .tableStatsRequest(keyspace, table)); } /** @@ -770,8 +740,7 @@ public CompletableFuture operationalJobs(SidecarInstance { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) - .operationalJobRequest(jobId) - .build()); + .operationalJobRequest(jobId)); } /** @@ -784,8 +753,7 @@ public CompletableFuture listOperationalJobs(Sideca { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) - .listOperationalJobsRequest() - .build()); + .listOperationalJobsRequest()); } /** @@ -797,8 +765,7 @@ public CompletableFuture streamsStats(SidecarInstance insta { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) - .streamsStatsRequest() - .build()); + .streamsStatsRequest()); } /** @@ -810,8 +777,7 @@ public CompletableFuture nodeDecommission(SidecarInstanc { return executor.executeRequestAsync(requestBuilder() .singleInstanceSelectionPolicy(instance) - .nodeDecommissionRequest() - .build()); + .nodeDecommissionRequest()); } /** @@ -850,14 +816,14 @@ public RetryPolicy defaultRetryPolicy() * Returns a future with the expected instance of type {@code } after executing the {@code request} and * processing it. * - * @param context the request context - * @param the expected type for the instance + * @param requestBuilder the request context builder + * @param the expected type for the instance * @return a future with the expected instance of type {@code } after executing the {@code request} and * processing it */ - public CompletableFuture executeRequestAsync(RequestContext context) + public CompletableFuture executeRequestAsync(RequestContext.Builder requestBuilder) { - return executor.executeRequestAsync(context); + return executor.executeRequestAsync(requestBuilder); } /** diff --git a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientConfig.java b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientConfig.java index a3d9fc9f8..1136aedb2 100644 --- a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientConfig.java +++ b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientConfig.java @@ -20,6 +20,9 @@ package org.apache.cassandra.sidecar.client; import java.time.Duration; +import java.util.Map; + +import org.jetbrains.annotations.Nullable; /** * Encapsulates configurations for the {@link SidecarClient} @@ -50,4 +53,12 @@ public interface SidecarClientConfig * @return the maximum amount of time to wait before retrying a failed health check */ Duration maximumHealthRetryDelay(); + + /** + * @return implementation class of custom identity provider + */ + @Nullable + String identityProviderClass(); + + Map identityProviderParameters(); } diff --git a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientConfigImpl.java b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientConfigImpl.java index 5c4a6ce52..45f3df133 100644 --- a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientConfigImpl.java +++ b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientConfigImpl.java @@ -20,6 +20,8 @@ package org.apache.cassandra.sidecar.client; import java.time.Duration; +import java.util.Collections; +import java.util.Map; import org.apache.cassandra.sidecar.common.DataObjectBuilder; @@ -33,12 +35,16 @@ public class SidecarClientConfigImpl implements SidecarClientConfig public static final long DEFAULT_MAX_RETRY_DELAY_MILLIS = 60_000L; public static final Duration DEFAULT_MINIMUM_HEALTH_RETRY_DELAY = Duration.ofSeconds(1L); public static final Duration DEFAULT_MAXIMUM_HEALTH_RETRY_DELAY = Duration.ofSeconds(5L); + public static final String DEFAULT_IDENTITY_PROVIDER = null; + public static final Map DEFAULT_IDENTITY_PROVIDER_PARAMETERS = Collections.emptyMap(); protected final int maxRetries; protected final long retryDelayMillis; protected final long maxRetryDelayMillis; protected final Duration minimumHealthRetryDelay; protected final Duration maximumHealthRetryDelay; + protected final String identityProviderClass; + protected final Map identityProviderParameters; private SidecarClientConfigImpl(Builder builder) { @@ -47,6 +53,8 @@ private SidecarClientConfigImpl(Builder builder) maxRetryDelayMillis = builder.maxRetryDelayMillis; minimumHealthRetryDelay = builder.minimumHealthRetryDelay; maximumHealthRetryDelay = builder.maximumHealthRetryDelay; + identityProviderClass = builder.identityProviderClass; + identityProviderParameters = builder.identityProviderParameters; } /** @@ -79,20 +87,32 @@ public long maxRetryDelayMillis() /** * @return the minimum amount of time to wait before retrying a failed health check */ - @Override - public Duration minimumHealthRetryDelay() - { + @Override + public Duration minimumHealthRetryDelay() + { return minimumHealthRetryDelay; - } + } /** * @return the maximum amount of time to wait before retrying a failed health check */ - @Override - public Duration maximumHealthRetryDelay() - { + @Override + public Duration maximumHealthRetryDelay() + { return maximumHealthRetryDelay; - } + } + + @Override + public String identityProviderClass() + { + return identityProviderClass; + } + + @Override + public Map identityProviderParameters() + { + return identityProviderParameters; + } public static Builder builder() { @@ -109,6 +129,8 @@ public static class Builder implements DataObjectBuilder identityProviderParameters = DEFAULT_IDENTITY_PROVIDER_PARAMETERS; protected Builder() { @@ -175,6 +197,21 @@ public Builder maximumHealthRetryDelay(Duration maximumHealthRetryDelay) return update(builder -> builder.maximumHealthRetryDelay = maximumHealthRetryDelay); } + /** + * Sets the {@code identityProvider} and returns a reference to this Builder enabling method chaining + * + * @param implClass the {@code identityProviderClass} to set + * @param parameters the {@code identityProviderParameters} to set + * @return a reference to this Builder + */ + public Builder identityProvider(String implClass, Map parameters) + { + return update(builder -> { + builder.identityProviderClass = implClass; + builder.identityProviderParameters = parameters; + }); + } + /** * Returns a {@code SidecarConfig} built from the parameters previously set. * diff --git a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarIdentityProvider.java b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarIdentityProvider.java new file mode 100644 index 000000000..67a237f04 --- /dev/null +++ b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarIdentityProvider.java @@ -0,0 +1,62 @@ +/* + * 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.cassandra.sidecar.client; + +import java.util.Map; + +import org.jetbrains.annotations.ApiStatus; + +/** + * An extension point for a custom identity provider for Sidecar communication. + * + *

An identity provider is registered by specifying its fully qualified class name + * in the {@code sidecar_identity_provider_class} property. Implementations must + * provide a no-argument constructor. Initialization code should be placed in the + * {@link #initialize(Map, HttpClient)} method. + * + *

Implementations must be thread-safe, as methods may be invoked concurrently from + * multiple worker threads. Avoid performing long-running I/O interactions inside the + * {@link #injectCredentials(RequestContext.Builder)} callback. + */ +@ApiStatus.Experimental +public interface SidecarIdentityProvider +{ + SidecarIdentityProvider NOOP = requestBuilder -> {}; + + /** + * Initializes the identity provider. This method can be invoked from both the Spark driver and executors. + * + * @param options the identity provider's options passed as part of the Spark configuration, + * prefixed by {@code sidecar_identity_provider_parameter.}. + * For example, configuring {@code "sidecar_identity_provider_parameter.param1" = "value1"} + * will result in a map entry of {@code "param1" = "value1"}. + * @param httpClient the HTTP client used for communication + */ + default void initialize(Map options, HttpClient httpClient) + { + } + + /** + * Callback executed before a request is built and sent to Sidecar. + * A typical implementation will inject custom HTTP headers or credentials into the request. + * + * @param requestBuilder the request builder used to construct the upcoming request + */ + void injectCredentials(RequestContext.Builder requestBuilder); +} diff --git a/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/SidecarClientConfigTest.java b/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/SidecarClientConfigTest.java index a0c7c0bfc..59ab943a2 100644 --- a/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/SidecarClientConfigTest.java +++ b/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/SidecarClientConfigTest.java @@ -19,6 +19,9 @@ package org.apache.cassandra.sidecar.client; +import java.util.Collections; +import java.util.Map; + import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; @@ -35,6 +38,8 @@ void testDefaults() assertThat(config.maxRetries()).isEqualTo(3); assertThat(config.retryDelayMillis()).isEqualTo(500L); assertThat(config.maxRetryDelayMillis()).isEqualTo(60_000L); + assertThat(config.identityProviderClass()).isNull(); + assertThat(config.identityProviderParameters()).isEmpty(); } @Test @@ -58,16 +63,33 @@ void testMaxRetryDelayMillis() assertThat(config.maxRetryDelayMillis()).isEqualTo(5_100L); } + @Test + void testIdentityProvider() + { + String providerClass = "org.apache.cassandra.sidecar.client.TestIdentityProvider"; + Map providerParams = Collections.singletonMap("key1", "value1"); + SidecarClientConfig config = SidecarClientConfigImpl.builder() + .identityProvider(providerClass, providerParams) + .build(); + assertThat(config.identityProviderClass()).isEqualTo(providerClass); + assertThat(config.identityProviderParameters()).isEqualTo(providerParams); + } + @Test void testAllOptions() { + String providerClass = "org.apache.cassandra.sidecar.client.TestIdentityProvider"; + Map providerParams = Collections.singletonMap("key1", "value1"); SidecarClientConfig config = SidecarClientConfigImpl.builder() .maxRetries(10) .retryDelayMillis(100) .maxRetryDelayMillis(5_100) + .identityProvider(providerClass, providerParams) .build(); assertThat(config.maxRetries()).isEqualTo(10); assertThat(config.retryDelayMillis()).isEqualTo(100L); assertThat(config.maxRetryDelayMillis()).isEqualTo(5_100L); + assertThat(config.identityProviderClass()).isEqualTo(providerClass); + assertThat(config.identityProviderParameters()).isEqualTo(providerParams); } } diff --git a/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java b/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java index 3d6a1a1f5..7d6261be5 100644 --- a/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java +++ b/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java @@ -1404,7 +1404,7 @@ void testProvidingCustomRetryPolicy() throws ExecutionException, InterruptedExce MockResponse response = new MockResponse().setResponseCode(ACCEPTED.code()).setBody(nodeSettingsAsString); enqueue(response); - RequestContext requestContext = + RequestContext.Builder requestContextBuilder = client.requestBuilder() .request(new NodeSettingsRequest()) .retryPolicy(new RetryPolicy() @@ -1433,9 +1433,8 @@ public void onResponse(CompletableFuture responseFuture, retryAction); } } - }) - .build(); - NodeSettings result = client.executeRequestAsync(requestContext).get(30, TimeUnit.SECONDS); + }); + NodeSettings result = client.executeRequestAsync(requestContextBuilder).get(30, TimeUnit.SECONDS); assertThat(result).isNotNull(); assertThat(result.partitioner()).isEqualTo("test-partitioner"); assertThat(result.releaseVersion()).isEqualTo("4.0.0"); diff --git a/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/request/RequestExecutorTest.java b/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/request/RequestExecutorTest.java index 3a084236b..f3bd7b2de 100644 --- a/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/request/RequestExecutorTest.java +++ b/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/request/RequestExecutorTest.java @@ -25,6 +25,7 @@ import java.util.stream.Stream; import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.extension.ExtensionContext; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; @@ -36,6 +37,7 @@ import okhttp3.mockwebserver.RecordedRequest; import org.apache.cassandra.sidecar.client.HttpClientConfig; import org.apache.cassandra.sidecar.client.RequestContext; +import org.apache.cassandra.sidecar.client.SidecarIdentityProvider; import org.apache.cassandra.sidecar.client.exception.RetriesExhaustedException; import org.apache.cassandra.sidecar.client.selection.InstanceSelectionPolicy; @@ -43,6 +45,10 @@ import static io.netty.handler.codec.http.HttpResponseStatus.OK; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatException; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; /** * Provides base functionality for testing the RequestExecutor @@ -51,6 +57,14 @@ */ public abstract class RequestExecutorTest extends BaseRequestTest { + SidecarIdentityProvider mockIdentityProvider; + + @BeforeEach + void setUp() + { + mockIdentityProvider = mock(SidecarIdentityProvider.class); + } + @AfterEach void cleanup() throws Exception { @@ -73,6 +87,7 @@ public void testHappyPath(RequestTestParameters parameters, boolean async) th RecordedRequest request = server.takeRequest(); assertThat(request.getPath()).isEqualTo(parameters.expectedEndpointPath()); assertThat(request.getHeader("User-Agent")).isEqualTo("sidecar-client-test/1.0.0"); + verify(mockIdentityProvider, times(1)).injectCredentials(any()); } } @@ -104,6 +119,8 @@ public void testRetriesOnTimeoutWithSlowServer(RequestTestParameters paramete RecordedRequest requestToFastServer = fastServer.takeRequest(); assertThat(requestToFastServer.getPath()).isEqualTo(parameters.expectedEndpointPath()); assertThat(requestToFastServer.getHeader("User-Agent")).isEqualTo("sidecar-client-test/1.0.0"); + + verify(mockIdentityProvider, times(1)).injectCredentials(any()); } } @@ -134,6 +151,8 @@ public void testRetriesWithServerReturningServerError(RequestTestParameters p RecordedRequest requestToNormalServer = normalOperatingServer.takeRequest(); assertThat(requestToNormalServer.getPath()).isEqualTo(parameters.expectedEndpointPath()); assertThat(requestToNormalServer.getHeader("User-Agent")).isEqualTo("sidecar-client-test/1.0.0"); + + verify(mockIdentityProvider, times(1)).injectCredentials(any()); } } @@ -159,6 +178,8 @@ public void testWithSingleServerAndRetries(RequestTestParameters parameters, RecordedRequest request = server.takeRequest(); assertThat(request.getPath()).isEqualTo(parameters.expectedEndpointPath()); assertThat(request.getHeader("User-Agent")).isEqualTo("sidecar-client-test/1.0.0"); + + verify(mockIdentityProvider, times(1)).injectCredentials(any()); } } @@ -192,6 +213,7 @@ public void testWithSingleServerExhaustsRetries(RequestTestParameters paramet RecordedRequest request = server.takeRequest(); assertThat(request.getPath()).isEqualTo(parameters.expectedEndpointPath()); assertThat(request.getHeader("User-Agent")).isEqualTo("sidecar-client-test/1.0.0"); + verify(mockIdentityProvider, times(1)).injectCredentials(any()); } } @@ -204,21 +226,26 @@ protected HttpClientConfig.Builder httpClientConfigBuilder() .idleTimeoutMillis(100); } + protected SidecarIdentityProvider sidecarIdentityProvider() + { + return mockIdentityProvider; + } + private void runTestScenario(RequestTestParameters parameters, InstanceSelectionPolicy policy, boolean async) throws Exception { - RequestContext requestContext = parameters.specificRequest(builder(policy)).build(); + RequestContext.Builder requestContextBuilder = parameters.specificRequest(builder(policy)); T responseObject; if (async) { - CompletableFuture future = sidecarClient().executeRequestAsync(requestContext); + CompletableFuture future = sidecarClient().executeRequestAsync(requestContextBuilder); responseObject = future.join(); assertThat(future).isDone(); } else { - responseObject = sidecarClient().executeRequest(requestContext, 20, TimeUnit.SECONDS); + responseObject = sidecarClient().executeRequest(requestContextBuilder, 20, TimeUnit.SECONDS); } assertThat(responseObject).isNotNull(); parameters.validateResponse(responseObject); diff --git a/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientVertxRequestExecutor.java b/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientVertxRequestExecutor.java index 704492b3a..2181d1e65 100644 --- a/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientVertxRequestExecutor.java +++ b/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClientVertxRequestExecutor.java @@ -29,9 +29,9 @@ public class SidecarClientVertxRequestExecutor extends VertxRequestExecutor { private final Vertx vertx; - public SidecarClientVertxRequestExecutor(VertxHttpClient httpClient) + public SidecarClientVertxRequestExecutor(VertxHttpClient httpClient, SidecarIdentityProvider identityProvider) { - super(httpClient); + super(httpClient, identityProvider); this.vertx = requireNonNull(httpClient.vertx(), "The vertx instance is required"); } diff --git a/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxRequestExecutor.java b/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxRequestExecutor.java index afdbcded2..9dde4dacf 100644 --- a/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxRequestExecutor.java +++ b/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxRequestExecutor.java @@ -33,9 +33,9 @@ public class VertxRequestExecutor extends RequestExecutor { private final Vertx vertx; - public VertxRequestExecutor(VertxHttpClient httpClient) + public VertxRequestExecutor(VertxHttpClient httpClient, SidecarIdentityProvider identityProvider) { - super(httpClient); + super(httpClient, identityProvider); this.vertx = requireNonNull(httpClient.vertx(), "The vertx instance is required"); } diff --git a/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxSidecarClientTest.java b/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxSidecarClientTest.java index f7c1e0347..d40cf500f 100644 --- a/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxSidecarClientTest.java +++ b/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxSidecarClientTest.java @@ -51,7 +51,7 @@ protected SidecarClient initialize(List instances) sidecarClientConfig.maxRetryDelayMillis()); VertxHttpClient vertxHttpClient = new VertxHttpClient(vertx, httpClientConfig); - VertxRequestExecutor requestExecutor = new VertxRequestExecutor(vertxHttpClient); + VertxRequestExecutor requestExecutor = new VertxRequestExecutor(vertxHttpClient, null); SimpleSidecarInstancesProvider instancesProvider = new SimpleSidecarInstancesProvider(instances); return new SidecarClient(instancesProvider, requestExecutor, sidecarClientConfig, defaultRetryPolicy); } diff --git a/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/request/BaseVertxRequestTest.java b/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/request/BaseVertxRequestTest.java index 49486367c..191c71981 100644 --- a/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/request/BaseVertxRequestTest.java +++ b/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/request/BaseVertxRequestTest.java @@ -38,7 +38,7 @@ public class BaseVertxRequestTest extends RequestExecutorTest @Override protected RequestExecutor sidecarClient() { - return new VertxRequestExecutor(this.httpClient()); + return new VertxRequestExecutor(this.httpClient(), this.sidecarIdentityProvider()); } @Override diff --git a/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcClient.java b/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcClient.java index 922c06e85..28f49f6f9 100644 --- a/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcClient.java +++ b/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcClient.java @@ -49,6 +49,8 @@ import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import static org.apache.cassandra.spark.utils.Properties.DEFAULT_SIDECAR_IDENTITY_PROVIDER_CLASS; +import static org.apache.cassandra.spark.utils.Properties.DEFAULT_SIDECAR_IDENTITY_PROVIDER_PARAMETERS; import static org.apache.cassandra.spark.utils.Properties.DEFAULT_MAX_BUFFER_OVERRIDE; import static org.apache.cassandra.spark.utils.Properties.DEFAULT_MAX_BUFFER_SIZE; import static org.apache.cassandra.spark.utils.Properties.DEFAULT_MAX_MILLIS_TO_SLEEP; @@ -235,6 +237,8 @@ public static final class ClientConfig private final String cassandraRole; private final Map maxBufferOverride; private final Map chunkBufferOverride; + private final String identityProviderClass; + private final Map identityProviderParameters; // CHECKSTYLE IGNORE: Constructor with many parameters private ClientConfig(int userProvidedPort, @@ -247,7 +251,9 @@ private ClientConfig(int userProvidedPort, int timeoutSeconds, String cassandraRole, Map maxBufferOverride, - Map chunkBufferOverride) + Map chunkBufferOverride, + String identityProviderClass, + Map identityProviderParameters) { this.userProvidedPort = userProvidedPort; this.maxRetries = maxRetries; @@ -260,6 +266,8 @@ private ClientConfig(int userProvidedPort, this.cassandraRole = cassandraRole; this.maxBufferOverride = maxBufferOverride; this.chunkBufferOverride = chunkBufferOverride; + this.identityProviderClass = identityProviderClass; + this.identityProviderParameters = identityProviderParameters; } public int userProvidedPort() @@ -327,6 +335,17 @@ public int timeoutSeconds() return timeoutSeconds; } + @Nullable + public String identityProviderClass() + { + return identityProviderClass; + } + + public Map identityProviderParameters() + { + return identityProviderParameters; + } + @Nullable public String cassandraRole() { @@ -358,7 +377,9 @@ public static ClientConfig create(int userProvidedPort, int maxRetries, long mil DEFAULT_TIMEOUT_SECONDS, DEFAULT_CASSANDRA_ROLE, DEFAULT_MAX_BUFFER_OVERRIDE, - chunkOverride); + chunkOverride, + DEFAULT_SIDECAR_IDENTITY_PROVIDER_CLASS, + DEFAULT_SIDECAR_IDENTITY_PROVIDER_PARAMETERS); } public static Map buildMaxBufferOverride(Map options, @@ -398,7 +419,9 @@ public static ClientConfig create(int userProvidedPort, int timeoutSeconds, String cassandraRole, Map maxBufferOverride, - Map chunkBufferOverride) + Map chunkBufferOverride, + String identityProviderClass, + Map identityProviderParameters) { return new ClientConfig(userProvidedPort, maxRetries, @@ -410,7 +433,9 @@ public static ClientConfig create(int userProvidedPort, timeoutSeconds, cassandraRole, maxBufferOverride, - chunkBufferOverride); + chunkBufferOverride, + identityProviderClass, + identityProviderParameters); } public Sidecar.ClientConfig toGenericSidecarConfig() @@ -425,7 +450,9 @@ public Sidecar.ClientConfig toGenericSidecarConfig() this.timeoutSeconds, this.cassandraRole, this.maxBufferOverride, - this.chunkBufferOverride); + this.chunkBufferOverride, + this.identityProviderClass, + this.identityProviderParameters); } } } diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/MapUtils.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/MapUtils.java index 81c5e2673..4e9874242 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/MapUtils.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/MapUtils.java @@ -25,6 +25,7 @@ import java.util.Optional; import java.util.function.Function; import java.util.function.Supplier; +import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -249,6 +250,18 @@ public static String getOrDefault(Map options, String key, Strin return options.getOrDefault(lowerCaseKey(key), defaultValue); } + public static Map getKeysWithPrefix(Map options, String keyPrefix, + boolean truncatePrefix, Map defaultValue) + { + int prefixLength = keyPrefix.length(); + String lowerCasePrefix = lowerCaseKey(keyPrefix); + Map subMap = options.entrySet().stream() + .filter(entry -> lowerCaseKey(entry.getKey()).startsWith(lowerCasePrefix)) + .collect(Collectors.toMap(k -> truncatePrefix ? k.getKey().substring(prefixLength) : k.getKey(), + Map.Entry::getValue)); + return subMap.isEmpty() ? defaultValue : subMap; + } + /** * Method to check if key is present in {@code options} map. * diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/Properties.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/Properties.java index 346d76c67..e59ad7199 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/Properties.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/Properties.java @@ -20,6 +20,7 @@ package org.apache.cassandra.spark.utils; import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; import java.util.Map; import java.util.Set; @@ -61,6 +62,8 @@ public final class Properties entry(FileType.PARTITIONS_INDEX, 4 * KIBI_BYTES), entry(FileType.ROWS_INDEX, 4 * KIBI_BYTES)); public static final int DEFAULT_TIMEOUT_SECONDS = (int) TimeUnit.MINUTES.toSeconds(10); + public static final String DEFAULT_SIDECAR_IDENTITY_PROVIDER_CLASS = null; + public static final Map DEFAULT_SIDECAR_IDENTITY_PROVIDER_PARAMETERS = Collections.emptyMap(); // Expansion and Shrink public static final Set NODE_STATUS_NOT_CONSIDERED = new HashSet<>(Arrays.asList("JOINING", "DOWN")); diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java index 2096a60a9..986c1f6bd 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java @@ -69,6 +69,7 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid .maxRetries(conf.getSidecarRequestRetries()) .retryDelayMillis(conf.getSidecarRequestRetryDelayMillis()) .maxRetryDelayMillis(conf.getSidecarRequestMaxRetryDelayMillis()) + .identityProvider(conf.getSidecarIdentityProviderClass(), conf.getSidecarIdentityProviderParameters()) .build(); return Sidecar.buildClient(sidecarConfig, vertx, httpClientConfig, sidecarInstancesProvider); diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java index 6e64f145a..8d002bf80 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java @@ -49,6 +49,7 @@ import org.apache.cassandra.spark.common.SidecarInstanceFactory; import org.apache.cassandra.spark.utils.BuildInfo; import org.apache.cassandra.spark.utils.MapUtils; +import org.apache.cassandra.spark.utils.Properties; import org.apache.spark.SparkConf; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -174,6 +175,8 @@ public class BulkSparkConf implements Serializable // create sidecarInstances from sidecarContactPointsValue and effectiveSidecarPort private final String sidecarContactPointsValue; // It takes comma separated values private transient Set sidecarContactPoints; // not serialized + private String sidecarIdentityProviderClass; + private Map sidecarIdentityProviderParameters; protected final String coordinatedWriteConfJson; private transient CoordinatedWriteConf coordinatedWriteConf; // it is transient; deserialized from coordinatedWriteConfJson in executors @@ -247,6 +250,7 @@ public BulkSparkConf(SparkConf conf, Map options, @Nullable Logg storageClientEndpointOverride, nioHttpClientConnectionAcquisitionTimeoutSeconds, nioHttpClientMaxConcurrency); + parseSidecarIdentityProvider(options); DataTransport dataTransport = MapUtils.getEnumOption(options, WriterOptions.DATA_TRANSPORT.name(), DataTransport.DIRECT, "Data Transport"); long maxSizePerSSTableBundleInBytesS3Transport = MapUtils.getLong(options, WriterOptions.MAX_SIZE_PER_SSTABLE_BUNDLE_IN_BYTES_S3_TRANSPORT.name(), DEFAULT_MAX_SIZE_PER_SSTABLE_BUNDLE_IN_BYTES_S3_TRANSPORT); @@ -392,6 +396,24 @@ protected CoordinatedWriteConf buildCoordinatedWriteConf(DataTransport dataTrans return CoordinatedWriteConf.create(coordinatedWriteConfJson, consistencyLevel, SimpleClusterConf.class); } + protected void parseSidecarIdentityProvider(Map options) + { + String providerClazz = MapUtils.getOrDefault(options, WriterOptions.SIDECAR_IDENTITY_PROVIDER_CLASS.name(), null); + if (StringUtils.isEmpty(providerClazz)) + { + sidecarIdentityProviderClass = null; + sidecarIdentityProviderParameters = Properties.DEFAULT_SIDECAR_IDENTITY_PROVIDER_PARAMETERS; + } + else + { + sidecarIdentityProviderClass = providerClazz; + sidecarIdentityProviderParameters = MapUtils.getKeysWithPrefix(options, + WriterOptions.SIDECAR_IDENTITY_PROVIDER_PARAMETER.name() + ".", + true, + Properties.DEFAULT_SIDECAR_IDENTITY_PROVIDER_PARAMETERS); + } + } + protected void validateEnvironment() throws RuntimeException { Preconditions.checkNotNull(keyspace); @@ -582,6 +604,16 @@ public int getSidecarRequestTimeoutSeconds() return getInt(SIDECAR_REQUEST_TIMEOUT_SECONDS, DEFAULT_SIDECAR_REQUEST_TIMEOUT_SECONDS); } + public String getSidecarIdentityProviderClass() + { + return sidecarIdentityProviderClass; + } + + public Map getSidecarIdentityProviderParameters() + { + return sidecarIdentityProviderParameters; + } + public int getHttpConnectionTimeoutMs() { return getInt(HTTP_CONNECTION_TIMEOUT, DEFAULT_HTTP_CONNECTION_TIMEOUT); diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java index 5440016ac..6bb67d98f 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java @@ -50,6 +50,8 @@ public enum WriterOptions implements WriterOption KEYSTORE_PATH, KEYSTORE_BASE64_ENCODED, KEYSTORE_TYPE, + SIDECAR_IDENTITY_PROVIDER_CLASS, + SIDECAR_IDENTITY_PROVIDER_PARAMETER, /** * Option that specifies cassandra role used for role based access control in Sidecar. *

diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageDataTransferApiImpl.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageDataTransferApiImpl.java index be218365f..beda75535 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageDataTransferApiImpl.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageDataTransferApiImpl.java @@ -214,8 +214,7 @@ private CompletableFuture createRestoreSliceWithCustomRetry(SidecarInstanc return sidecarClient.executeRequestAsync(sidecarClient.requestBuilder() .retryPolicy(retryPolicy) .singleInstanceSelectionPolicy(sidecarInstance) - .request(request) - .build()); + .request(request)); } /** diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCloudStorageDataTransferApi.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCloudStorageDataTransferApi.java index 4978a1eb1..c7f02dad9 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCloudStorageDataTransferApi.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCloudStorageDataTransferApi.java @@ -174,10 +174,10 @@ private void createRestoreSliceInternal(String clusterId, restoreJobId, createSliceRequestPayload); RetryPolicy retryPolicy = new CloudStorageDataTransferApiImpl.ExecutorCreateSliceRetryPolicy(sidecarClient); - RequestContext requestContext = sidecarClient.requestBuilder().retryPolicy(retryPolicy).request(request).build(); + RequestContext.Builder requestBuilder = sidecarClient.requestBuilder().retryPolicy(retryPolicy).request(request); try { - sidecarClient.executeRequestAsync(requestContext).get(); + sidecarClient.executeRequestAsync(requestBuilder).get(); } catch (Exception exception) { diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java index ff7e0b67d..f366eb857 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java @@ -834,7 +834,9 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE in.readInt(), readNullable(in), (Map) in.readObject(), - (Map) in.readObject()); + (Map) in.readObject(), + readNullable(in), + (Map) in.readObject()); this.sslConfig = (SslConfig) in.readObject(); this.cqlTable = bridge.javaDeserialize(in, CqlTable.class); // Delegate (de-)serialization of version-specific objects to the Cassandra Bridge @@ -890,6 +892,8 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou writeNullable(out, this.sidecarClientConfig.cassandraRole()); out.writeObject(this.sidecarClientConfig.maxBufferOverride()); out.writeObject(this.sidecarClientConfig.chunkBufferOverride()); + writeNullable(out, this.sidecarClientConfig.identityProviderClass()); + out.writeObject(this.sidecarClientConfig.identityProviderParameters()); out.writeObject(this.sslConfig); bridge.javaSerialize(out, this.cqlTable); // Delegate (de-)serialization of version-specific objects to the Cassandra Bridge out.writeObject(this.tokenPartitioner); @@ -1022,6 +1026,8 @@ public void write(Kryo kryo, Output out, CassandraDataLayer dataLayer) out.writeString(dataLayer.sidecarClientConfig.cassandraRole()); kryo.writeObject(out, dataLayer.sidecarClientConfig.maxBufferOverride()); kryo.writeObject(out, dataLayer.sidecarClientConfig.chunkBufferOverride()); + kryo.writeObjectOrNull(out, dataLayer.sidecarClientConfig.identityProviderClass(), String.class); + kryo.writeObject(out, dataLayer.sidecarClientConfig.identityProviderParameters()); kryo.writeObjectOrNull(out, dataLayer.sslConfig, SslConfig.class); kryo.writeObject(out, dataLayer.cqlTable); kryo.writeObject(out, dataLayer.tokenPartitioner); @@ -1074,7 +1080,9 @@ public CassandraDataLayer read(Kryo kryo, Input in, Class ty in.readInt(), in.readString(), (Map) kryo.readObject(in, HashMap.class), - (Map) kryo.readObject(in, HashMap.class)), + (Map) kryo.readObject(in, HashMap.class), + in.readString(), + (Map) kryo.readObject(in, HashMap.class)), kryo.readObjectOrNull(in, SslConfig.class), kryo.readObject(in, CqlTable.class), kryo.readObject(in, TokenPartitioner.class), diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/SidecarClientConfigTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/SidecarClientConfigTest.java index 71a835d65..6bc6b695a 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/SidecarClientConfigTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/SidecarClientConfigTest.java @@ -45,6 +45,8 @@ public void testDefaults() assertThat(clientConfig.chunkBufferSize()).isEqualTo(4L * 1024L * 1024L); assertThat(clientConfig.maxPoolSize()).isEqualTo(64); assertThat(clientConfig.timeoutSeconds()).isEqualTo(600); + assertThat(clientConfig.identityProviderClass()).isNull(); + assertThat(clientConfig.identityProviderParameters()).isEmpty(); } @Test @@ -113,4 +115,14 @@ public void testTransportModeBasedWriterUserAgent() userAgentStr = AnalyticsSidecarClient.transportModeBasedWriterUserAgent(DataTransport.S3_COMPAT); assertThat(userAgentStr.endsWith(" writer-s3")).isTrue(); } + + @Test + public void testCustomSidecarIdentityProvider() + { + String providerClass = "org.apache.cassandra.sidecar.client.TestIdentityProvider"; + Sidecar.ClientConfig clientConfig = Sidecar.ClientConfig.create(ImmutableMap.of("sidecar_identity_provider_class", providerClass, + "sidecar_identity_provider_parameter.param1", "value1")); + assertThat(clientConfig.identityProviderClass()).isEqualTo(providerClass); + assertThat(clientConfig.identityProviderParameters()).isEqualTo(ImmutableMap.of("param1", "value1")); + } } diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java index 5fbca3504..15da6c370 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java @@ -24,6 +24,7 @@ import java.util.TreeMap; import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -426,6 +427,27 @@ void testSSTableVersionBasedBridgeEnabledExplicitly() .isFalse(); } + @Test + void testCustomSidecarIdentityProvider() + { + String providerClass = "org.apache.cassandra.sidecar.client.TestIdentityProvider"; + Map options = copyDefaultOptions(); + options.put(WriterOptions.SIDECAR_IDENTITY_PROVIDER_CLASS.name(), providerClass); + options.put(WriterOptions.SIDECAR_IDENTITY_PROVIDER_PARAMETER.name() + ".param1", "value1"); + BulkSparkConf bulkSparkConf = new BulkSparkConf(sparkConf, options, null); + assertThat(bulkSparkConf.getSidecarIdentityProviderClass()).isEqualTo(providerClass); + assertThat(bulkSparkConf.getSidecarIdentityProviderParameters()).isEqualTo(ImmutableMap.of("param1", "value1")); + } + + @Test + void testEmptySidecarIdentityProvider() + { + Map options = copyDefaultOptions(); + BulkSparkConf bulkSparkConf = new BulkSparkConf(sparkConf, options, null); + assertThat(bulkSparkConf.getSidecarIdentityProviderClass()).isNull(); + assertThat(bulkSparkConf.getSidecarIdentityProviderParameters()).isEmpty(); + } + private Map copyDefaultOptions() { TreeMap map = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerValidationTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerValidationTest.java index 68f7b9b1b..7cc11fbd3 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerValidationTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerValidationTest.java @@ -267,7 +267,7 @@ private static class TestCassandraDataLayer extends CassandraDataLayer false, // quoteIdentifiers "", // snapshotName null, // datacenter - Sidecar.ClientConfig.create(), // sidecarClientConfig + Sidecar.ClientConfig.create(), // sidecarClientConfig null, // sslConfig mock(CqlTable.class), // cqlTable mock(TokenPartitioner.class), // tokenPartitioner diff --git a/cassandra-analytics-integration-framework/build.gradle b/cassandra-analytics-integration-framework/build.gradle index 7c97cb3bf..ea85a3492 100644 --- a/cassandra-analytics-integration-framework/build.gradle +++ b/cassandra-analytics-integration-framework/build.gradle @@ -91,4 +91,6 @@ dependencies { exclude(group: 'io.confluent') } api(group: 'org.apache.cassandra', name: 'sidecar-server-common', version: "${sidecarVersion}") + + api(project(":cassandra-analytics-sidecar-client")) } diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java index 0577e4290..a2f76d2f8 100644 --- a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java +++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java @@ -66,6 +66,9 @@ import org.apache.cassandra.distributed.api.IInstance; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.distributed.shared.JMXUtil; +import org.apache.cassandra.sidecar.acl.authentication.AuthenticationHandlerFactoryRegistry; +import org.apache.cassandra.sidecar.acl.authentication.JwtAuthenticationHandlerFactory; +import org.apache.cassandra.sidecar.acl.authentication.MutualTlsAuthenticationHandlerFactory; import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate; import org.apache.cassandra.sidecar.cluster.InstancesMetadata; import org.apache.cassandra.sidecar.cluster.InstancesMetadataImpl; @@ -706,6 +709,19 @@ private List buildContactPoints() .collect(Collectors.toList()); } + @Provides + @Singleton + public AuthenticationHandlerFactoryRegistry authNHandlerFactoryRegistry(MutualTlsAuthenticationHandlerFactory mTLSAuthHandlerFactory, + JwtAuthenticationHandlerFactory jwtAuthHandlerFactory, + TestAuthenticationHandlerFactory testAuthenticationHandlerFactory) + { + AuthenticationHandlerFactoryRegistry registry = new AuthenticationHandlerFactoryRegistry(); + registry.register(mTLSAuthHandlerFactory); + registry.register(jwtAuthHandlerFactory); + registry.register(testAuthenticationHandlerFactory); + return registry; + } + private S3ProxyConfiguration buildTestS3ProxyConfig() { return new S3ProxyConfiguration() diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestAuthenticationHandlerFactory.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestAuthenticationHandlerFactory.java new file mode 100644 index 000000000..29bc56c1f --- /dev/null +++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestAuthenticationHandlerFactory.java @@ -0,0 +1,95 @@ +/* + * 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.cassandra.sidecar.testing; + +import java.util.Map; + +import com.google.inject.Singleton; +import io.vertx.core.AsyncResult; +import io.vertx.core.Future; +import io.vertx.core.Handler; +import io.vertx.core.MultiMap; +import io.vertx.core.Vertx; +import io.vertx.core.json.JsonObject; +import io.vertx.ext.auth.User; +import io.vertx.ext.auth.authentication.AuthenticationProvider; +import io.vertx.ext.auth.authentication.Credentials; +import io.vertx.ext.web.RoutingContext; +import io.vertx.ext.web.handler.impl.AuthenticationHandlerImpl; +import io.vertx.ext.web.handler.impl.AuthenticationHandlerInternal; +import org.apache.cassandra.sidecar.acl.authentication.AuthenticationHandlerFactory; +import org.apache.cassandra.sidecar.config.AccessControlConfiguration; +import org.apache.cassandra.sidecar.exceptions.ConfigurationException; +import org.apache.cassandra.sidecar.metrics.server.AuthMetrics; + +@Singleton +public class TestAuthenticationHandlerFactory implements AuthenticationHandlerFactory +{ + @Override + public AuthenticationHandlerInternal create(Vertx vertx, + AccessControlConfiguration accessControlConfiguration, + Map parameters, + AuthMetrics metrics) throws ConfigurationException + { + return new TestAuthHandler(new NoOpAuthentication(), "cassandra", "secret"); + } + + public static class TestAuthHandler extends AuthenticationHandlerImpl + { + static final String USERNAME_HTTP_HEADER = "AUTH_USER"; + static final String PASSWORD_HTTP_HEADER = "AUTH_PASS"; + + private final String authUser; + private final String authPass; + + public TestAuthHandler(NoOpAuthentication authProvider, String authUser, String authPass) + { + super(authProvider); + this.authUser = authUser; + this.authPass = authPass; + } + + public void authenticate(RoutingContext routingContext, Handler> handler) + { + MultiMap requestHeaders = routingContext.request().headers(); + if (authUser.equals(requestHeaders.get(USERNAME_HTTP_HEADER)) && authPass.equals(requestHeaders.get(PASSWORD_HTTP_HEADER))) + { + handler.handle(Future.succeededFuture(User.fromName("dummy"))); + return; + } + handler.handle(Future.failedFuture(new RuntimeException("dummy"))); + } + } + + public static class NoOpAuthentication implements AuthenticationProvider + { + @Override + public Future authenticate(Credentials credentials) + { + return Future.succeededFuture(User.fromName("dummy")); + } + + @Override + @Deprecated + public void authenticate(JsonObject credentials, Handler> resultHandler) + { + throw new UnsupportedOperationException(); + } + } +} diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestSidecarIdentityProvider.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestSidecarIdentityProvider.java new file mode 100644 index 000000000..596b17233 --- /dev/null +++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestSidecarIdentityProvider.java @@ -0,0 +1,32 @@ +/* + * 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.cassandra.sidecar.testing; + +import o.a.c.sidecar.client.shaded.client.RequestContext; +import o.a.c.sidecar.client.shaded.client.SidecarIdentityProvider; + +public class TestSidecarIdentityProvider implements SidecarIdentityProvider +{ + @Override + public void injectCredentials(RequestContext.Builder builder) + { + builder.addCustomHeader(TestAuthenticationHandlerFactory.TestAuthHandler.USERNAME_HTTP_HEADER, "cassandra"); + builder.addCustomHeader(TestAuthenticationHandlerFactory.TestAuthHandler.PASSWORD_HTTP_HEADER, "secret"); + } +} diff --git a/cassandra-analytics-integration-tests/build.gradle b/cassandra-analytics-integration-tests/build.gradle index 38cff0681..68c990787 100644 --- a/cassandra-analytics-integration-tests/build.gradle +++ b/cassandra-analytics-integration-tests/build.gradle @@ -49,11 +49,11 @@ configurations { configureEach { resolutionStrategy { // Force set the vertx versions that do not depend on a newer jackson version - force("io.vertx:vertx-web:${vertxVersion}") - force("io.vertx:vertx-core:${vertxVersion}") - force("io.vertx:vertx-auth-common:${vertxVersion}") - force("io.vertx:vertx-dropwizard-metrics:${vertxVersion}") - force("io.vertx:vertx-web-client:${vertxVersion}") + force("io.vertx:vertx-web:${vertxTestVersion}") + force("io.vertx:vertx-core:${vertxTestVersion}") + force("io.vertx:vertx-auth-common:${vertxTestVersion}") + force("io.vertx:vertx-dropwizard-metrics:${vertxTestVersion}") + force("io.vertx:vertx-web-client:${vertxTestVersion}") } } } diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SidecarIdentityProviderTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SidecarIdentityProviderTest.java new file mode 100644 index 000000000..bcea73829 --- /dev/null +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SidecarIdentityProviderTest.java @@ -0,0 +1,87 @@ +/* + * 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.cassandra.analytics; + +import java.util.Map; +import java.util.function.Function; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.sidecar.acl.authorization.AllowAllAuthorizationProvider; +import org.apache.cassandra.sidecar.config.yaml.AccessControlConfigurationImpl; +import org.apache.cassandra.sidecar.config.yaml.ParameterizedClassConfigurationImpl; +import org.apache.cassandra.sidecar.config.yaml.SidecarConfigurationImpl; +import org.apache.cassandra.sidecar.testing.QualifiedName; +import org.apache.cassandra.sidecar.testing.TestAuthenticationHandlerFactory; +import org.apache.cassandra.sidecar.testing.TestSidecarIdentityProvider; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; + +import static org.apache.cassandra.testing.TestUtils.CREATE_TEST_TABLE_STATEMENT; +import static org.apache.cassandra.testing.TestUtils.DC1_RF1; +import static org.apache.cassandra.testing.TestUtils.ROW_COUNT; +import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE; +import static org.assertj.core.api.Assertions.assertThat; + +class SidecarIdentityProviderTest extends SharedClusterSparkIntegrationTestBase +{ + private static final QualifiedName TABLE_NAME = new QualifiedName(TEST_KEYSPACE, SidecarIdentityProviderTest.class.getSimpleName().toLowerCase()); + + @Override + protected void initializeSchemaForTest() + { + createTestKeyspace(TEST_KEYSPACE, DC1_RF1); + createTestTable(TABLE_NAME, CREATE_TEST_TABLE_STATEMENT); + } + + @Override + protected Function configurationOverrides() + { + return builder -> { + ParameterizedClassConfigurationImpl authenticator = new ParameterizedClassConfigurationImpl(TestAuthenticationHandlerFactory.class.getName(), null); + ParameterizedClassConfigurationImpl authorizer = new ParameterizedClassConfigurationImpl(AllowAllAuthorizationProvider.class.getName(), null); + builder.accessControlConfiguration(AccessControlConfigurationImpl.builder() + .enabled(true) + .authenticatorsConfiguration(ImmutableList.of(authenticator)) + .authorizerConfiguration(authorizer) + .build()); + return builder; + }; + } + + @Test + void testCustomIdentityProvider() + { + Map additionalOptions = ImmutableMap.of("sidecar_identity_provider_class", + TestSidecarIdentityProvider.class.getName()); + + SparkSession spark = getOrCreateSparkSession(); + Dataset df = DataGenerationUtils.generateCourseData(spark, ROW_COUNT); + bulkWriterDataFrameWriter(df, TABLE_NAME, additionalOptions).save(); + + sparkTestUtils.validateWrites(df.collectAsList(), queryAllData(TABLE_NAME)); + + Dataset read = bulkReaderDataFrame(TABLE_NAME, additionalOptions).load(); + assertThat(read.count()).isEqualTo(ROW_COUNT); + } +} diff --git a/cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java b/cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java index 78798416a..50d209907 100644 --- a/cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java +++ b/cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java @@ -20,6 +20,7 @@ package org.apache.cassandra.clients; import java.io.IOException; +import java.lang.reflect.InvocationTargetException; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -31,6 +32,7 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import o.a.c.sidecar.client.shaded.client.SidecarIdentityProvider; import o.a.c.sidecar.client.shaded.common.response.GossipInfoResponse; import org.slf4j.Logger; @@ -63,6 +65,8 @@ import static org.apache.cassandra.spark.utils.Properties.DEFAULT_CHUNK_BUFFER_OVERRIDE; import static org.apache.cassandra.spark.utils.Properties.DEFAULT_CHUNK_BUFFER_SIZE; +import static org.apache.cassandra.spark.utils.Properties.DEFAULT_SIDECAR_IDENTITY_PROVIDER_CLASS; +import static org.apache.cassandra.spark.utils.Properties.DEFAULT_SIDECAR_IDENTITY_PROVIDER_PARAMETERS; import static org.apache.cassandra.spark.utils.Properties.DEFAULT_MAX_BUFFER_OVERRIDE; import static org.apache.cassandra.spark.utils.Properties.DEFAULT_MAX_BUFFER_SIZE; import static org.apache.cassandra.spark.utils.Properties.DEFAULT_MAX_MILLIS_TO_SLEEP; @@ -121,6 +125,7 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid .maxRetries(config.maxRetries()) .retryDelayMillis(config.millisToSleep()) .maxRetryDelayMillis(config.maxMillisToSleep()) + .identityProvider(config.identityProviderClass(), config.identityProviderParameters()) .build(); return buildClient(sidecarConfig, vertx, httpClientConfig, sidecarInstancesProvider); @@ -136,10 +141,32 @@ public static SidecarClient buildClient(SidecarClientConfig sidecarConfig, sidecarConfig.maxRetryDelayMillis()); VertxHttpClient vertxHttpClient = new VertxHttpClient(vertx, httpClientConfig); - VertxRequestExecutor requestExecutor = new VertxRequestExecutor(vertxHttpClient); + SidecarIdentityProvider identityProvider = buildIdentityProvider(sidecarConfig, vertxHttpClient); + VertxRequestExecutor requestExecutor = new VertxRequestExecutor(vertxHttpClient, identityProvider); return new SidecarClient(clusterConfig, requestExecutor, sidecarConfig, defaultRetryPolicy); } + public static SidecarIdentityProvider buildIdentityProvider(SidecarClientConfig sidecarConfig, VertxHttpClient vertxHttpClient) + { + if (sidecarConfig.identityProviderClass() == null) + { + return SidecarIdentityProvider.NOOP; + } + try + { + // instantiate identity provider only once we have created the HTTP client + SidecarIdentityProvider instance = (SidecarIdentityProvider) Class.forName(sidecarConfig.identityProviderClass()) + .getDeclaredConstructor().newInstance(); + instance.initialize(sidecarConfig.identityProviderParameters(), vertxHttpClient); + return instance; + } + catch (ClassNotFoundException | ClassCastException | InvocationTargetException | InstantiationException + | IllegalAccessException | NoSuchMethodException e) + { + throw new RuntimeException("Failed to instantiate identity provider: " + sidecarConfig.identityProviderClass(), e); + } + } + public static List> allNodeSettings(SidecarClient client, Set instances) { @@ -250,6 +277,8 @@ public static final class ClientConfig public static final String TIMEOUT_SECONDS_KEY = "timeoutSeconds"; public static final String CASSANDRA_ROLE_KEY = "cassandra_role"; public static final String DEFAULT_CASSANDRA_ROLE = null; + public static final String SIDECAR_IDENTITY_PROVIDER_CLASS = "sidecar_identity_provider_class"; + public static final String SIDECAR_IDENTITY_PROVIDER_PARAMETER_PREFIX = "sidecar_identity_provider_parameter."; private final int userProvidedPort; private final int maxRetries; @@ -262,6 +291,8 @@ public static final class ClientConfig private final String cassandraRole; private final Map maxBufferOverride; private final Map chunkBufferOverride; + private final String identityProviderClass; + private final Map identityProviderParameters; // CHECKSTYLE IGNORE: Constructor with many parameters private ClientConfig(int userProvidedPort, @@ -274,7 +305,9 @@ private ClientConfig(int userProvidedPort, int timeoutSeconds, String cassandraRole, Map maxBufferOverride, - Map chunkBufferOverride) + Map chunkBufferOverride, + String identityProviderClass, + Map identityProviderParameters) { this.userProvidedPort = userProvidedPort; this.maxRetries = maxRetries; @@ -287,6 +320,8 @@ private ClientConfig(int userProvidedPort, this.cassandraRole = cassandraRole; this.maxBufferOverride = maxBufferOverride; this.chunkBufferOverride = chunkBufferOverride; + this.identityProviderClass = identityProviderClass; + this.identityProviderParameters = identityProviderParameters; } public int userProvidedPort() @@ -354,6 +389,17 @@ public int timeoutSeconds() return timeoutSeconds; } + @Nullable + public String identityProviderClass() + { + return identityProviderClass; + } + + public Map identityProviderParameters() + { + return identityProviderParameters; + } + @Nullable public String cassandraRole() { @@ -382,7 +428,9 @@ public static ClientConfig create(int userProvidedPort, int maxRetries, long mil DEFAULT_TIMEOUT_SECONDS, DEFAULT_CASSANDRA_ROLE, DEFAULT_MAX_BUFFER_OVERRIDE, - DEFAULT_CHUNK_BUFFER_OVERRIDE); + DEFAULT_CHUNK_BUFFER_OVERRIDE, + DEFAULT_SIDECAR_IDENTITY_PROVIDER_CLASS, + DEFAULT_SIDECAR_IDENTITY_PROVIDER_PARAMETERS); } public static ClientConfig create(Map options) @@ -398,7 +446,9 @@ public static ClientConfig create(Map options) MapUtils.getInt(options, TIMEOUT_SECONDS_KEY, DEFAULT_TIMEOUT_SECONDS), MapUtils.getOrDefault(options, CASSANDRA_ROLE_KEY, DEFAULT_CASSANDRA_ROLE), buildMaxBufferOverride(options, DEFAULT_MAX_BUFFER_OVERRIDE), - buildChunkBufferOverride(options, DEFAULT_CHUNK_BUFFER_OVERRIDE) + buildChunkBufferOverride(options, DEFAULT_CHUNK_BUFFER_OVERRIDE), + MapUtils.getOrDefault(options, SIDECAR_IDENTITY_PROVIDER_CLASS, DEFAULT_SIDECAR_IDENTITY_PROVIDER_CLASS), + MapUtils.getKeysWithPrefix(options, SIDECAR_IDENTITY_PROVIDER_PARAMETER_PREFIX, true, DEFAULT_SIDECAR_IDENTITY_PROVIDER_PARAMETERS) ); } @@ -439,7 +489,9 @@ public static ClientConfig create(int userProvidedPort, int timeoutSeconds, String cassandraRole, Map maxBufferOverride, - Map chunkBufferOverride) + Map chunkBufferOverride, + String identityProviderClass, + Map identityProviderParameters) { return new ClientConfig(userProvidedPort, maxRetries, @@ -451,7 +503,9 @@ public static ClientConfig create(int userProvidedPort, timeoutSeconds, cassandraRole, maxBufferOverride, - chunkBufferOverride); + chunkBufferOverride, + identityProviderClass, + identityProviderParameters); } } } diff --git a/cassandra-analytics-sidecar-client/src/test/java/org/apache/cassandra/clients/SidecarTest.java b/cassandra-analytics-sidecar-client/src/test/java/org/apache/cassandra/clients/SidecarTest.java index a434a5db1..3ad3a93ac 100644 --- a/cassandra-analytics-sidecar-client/src/test/java/org/apache/cassandra/clients/SidecarTest.java +++ b/cassandra-analytics-sidecar-client/src/test/java/org/apache/cassandra/clients/SidecarTest.java @@ -23,14 +23,20 @@ import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; import org.junit.jupiter.api.Test; +import o.a.c.sidecar.client.shaded.client.HttpClient; +import o.a.c.sidecar.client.shaded.client.RequestContext; import o.a.c.sidecar.client.shaded.client.SidecarClient; +import o.a.c.sidecar.client.shaded.client.SidecarClientConfig; +import o.a.c.sidecar.client.shaded.client.SidecarIdentityProvider; import o.a.c.sidecar.client.shaded.client.SidecarInstance; import o.a.c.sidecar.client.shaded.client.SidecarInstanceImpl; +import o.a.c.sidecar.client.shaded.client.VertxHttpClient; import o.a.c.sidecar.client.shaded.common.response.GossipInfoResponse; import static org.assertj.core.api.Assertions.assertThat; @@ -176,6 +182,33 @@ void testGetSSTableVersionsFromClusterWithEmptyInstancesSet() .isEmpty(); } + @Test + void testNoOpIdentityProvider() + { + SidecarClientConfig clientConfig = mock(SidecarClientConfig.class); + VertxHttpClient vertxHttpClient = mock(VertxHttpClient.class); + + SidecarIdentityProvider identityProvider = Sidecar.buildIdentityProvider(clientConfig, vertxHttpClient); + + assertThat(identityProvider).isEqualTo(SidecarIdentityProvider.NOOP); + } + + @Test + void testCustomIdentityProvider() + { + SidecarClientConfig clientConfig = mock(SidecarClientConfig.class); + Map providerParams = Collections.singletonMap("key1", "value1"); + when(clientConfig.identityProviderClass()).thenReturn(MockIdentityProvider.class.getName()); + when(clientConfig.identityProviderParameters()).thenReturn(providerParams); + VertxHttpClient vertxHttpClient = mock(VertxHttpClient.class); + + SidecarIdentityProvider identityProvider = Sidecar.buildIdentityProvider(clientConfig, vertxHttpClient); + + assertThat(identityProvider).isInstanceOf(MockIdentityProvider.class); + MockIdentityProvider testIdentityProvider = (MockIdentityProvider) identityProvider; + assertThat(testIdentityProvider.options).isEqualTo(providerParams); + } + private GossipInfoResponse.GossipInfo createGossipInfo(List sstableVersions) { GossipInfoResponse.GossipInfo info = new GossipInfoResponse.GossipInfo(); @@ -190,4 +223,20 @@ private GossipInfoResponse.GossipInfo createGossipInfo(List sstableVersi } return info; } + + public static class MockIdentityProvider implements SidecarIdentityProvider + { + Map options; + + public void initialize(Map options, HttpClient httpClient) + { + this.options = options; + } + + @Override + public void injectCredentials(RequestContext.Builder builder) + { + // no-op + } + } } diff --git a/docs/src/user.adoc b/docs/src/user.adoc index 80ae10ab3..fc6bfc279 100644 --- a/docs/src/user.adoc +++ b/docs/src/user.adoc @@ -68,6 +68,14 @@ with an optional port number (e.g. `localhost1,localhost2`, `127.0.0.1,127.0.0.2 |Specific role that Sidecar shall use to authorize the request. For further details consult Sidecar documentation for `cassandra-auth-role` HTTP header +|_sidecar_identity_provider_class_ +|no +| +|The fully qualified class name that implements `org.apache.cassandra.sidecar.client.SidecarIdentityProvider`. A custom +identity provider can modify HTTP requests sent to Cassandra Sidecar to inject authentication details. All configuration +properties prefixed with `sidecar_identity_provider_parameter.` are automatically passed to the provider's +initialization method. + |=== == Bulk Reader diff --git a/gradle.properties b/gradle.properties index eaa11965a..ae96a99eb 100644 --- a/gradle.properties +++ b/gradle.properties @@ -38,6 +38,9 @@ guavaVersion=16.0.1 # force version 4.5.1 of vertx to prevent issues initializing io.vertx.core.json.jackson.JacksonCodec, # which requires a newer version of jackson, which is not available in spark 2 vertxVersion=4.5.1 +# use later vertx version for integration tests, not to experience CASSSIDECAR-207 +# upgrading to more recent version (e.g. 4.5.23) can break integration tests with older C* versions +vertxTestVersion=4.5.12 aswSdkVersion=2.26.12 commonsLang3Version=3.12.0 cassandraDriverVersion=3.11.3