You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "georgew5656 (via GitHub)" <gi...@apache.org> on 2023/10/31 15:49:56 UTC
[PR] Azure client upgrade to allow identity options (druid)
georgew5656 opened a new pull request, #15287:
URL: https://github.com/apache/druid/pull/15287
### Description
Druid does not support various forms of auth that Azure Storage Accounts has added support for, like Service Client auth, Workload Identity Auth (e.g. for AKS) or managed identity Auth. Adding the AzureDefaultCredential chain to the azure clients druid uses allows support for configuring these things. See: https://learn.microsoft.com/en-us/java/api/overview/azure/identity-readme?view=azure-java-stable
In order to support this new identity library, I had to upgrade the druid azure client libraries for blob storage (they were very old anyways).
#### Fixed the bug ...
#### Renamed the class ...
#### Added a forbidden-apis entry ...
- Many of the changes from the client upgrade were pretty straightforward (I used this .NET docs a lot: https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/storage/Azure.Storage.Blobs/AzureStorageNetMigrationV12.md#class-conversion-reference)
- Most of these changes are in AzureStorage
Less straightforward changes
- The client supports retrying requests by default, so I removed instances of RetryUtils in favor of using the client directly. There are several places where druid uses a different retry configuration for different azure operations, so I had to add AzureClientFactory to support creating additional clients on the fly (because the library doesn't let you override the retry policy on existing clients). I also updated AzureStorage to reuse these per-container overrided clients, the new client is stateless so it is technically fine to just recreate a client each time but i figured it would be better to reuse them as a performance optimization. A couple of the tests were removed because they were testing this retry functionality
- I had to change the AzureCloudBlobIterator because the new client supports pagination by default (it only fetches additional pages if you request additional items)
#### Release note
- Upgrade druid azure client libraries
- Support AzureDefaultCredential for druid auth to azure services
- I removed some of the mocks from the tests and added a test implementation (TestPagedResponse). This let me remove the ListBlobItemHolder class that was mostly there for testing.
##### Key changed/added classes in this PR
* AzureClientFactory
* AzureAccountConfig
* AzureCloudBlobIterator
* AzureStorage
* AzureStorageDruidModule
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [X] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [X] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [X] been tested in a test Druid cluster.
I have tested the following scenarios for auth
SAS token auth
Key auth
Workload identity auth (using a k8s serviceAccount and managed identity to mount credentials for druid running on a pod in AKS)
Environment variable auth (using a App registration client id/secret)
Managed identity auth (I technically have not tested this directly on a VM, but I am pretty sure this will work on a VM with a managed identity by setting the managedIdentityClientId property https://github.com/Azure/azure-sdk-for-java/wiki/Azure-Identity-Examples#authenticating-in-azure-with-managed-identity)
Technically the AzureDefaultCredential supports more types of auth but I did not test any of them.
I also tested the following migration scenarios
Cluster running old azure extension using a sas token -> new azure extension using a sas token
Sas token (new client) -> workload credential auth (new client)
For actual functionality I tested all the basic druid functionality (ingesting data from azure, creating and loading segments, managing task logs, killing segments), but I didn't test any of the retry logic that the new clients promise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1439719466
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,137 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
- }
- public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
- {
- return listDir(containerName, virtualDirPath, null);
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(Utility.urlEncode(blobPath)).exists();
}
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
+ @VisibleForTesting
+ BlobServiceClient getRetriableBlobServiceClient(@Nonnull Integer maxAttempts)
{
- if (maxAttempts == null) {
- return null;
+ // To avoid keeping a ton of clients in memory, if maxAttempts is specified use a client with at least that many retries configured.
Review Comment:
this change is mainly to balance backwards compatibility with performance. the retry logic in the extension is such that some requests don't specify any retries (the default behavior in the previous client was to not retry at all), and some requests specify retries (these can come from different configs). the new azure client doesn't let you override the retry configuration for a client, so in the interest of not either 1. creating a client every time we make a request, 2. keeping a bunch of different clients cached in memory, i added the "at least retry attempts" logic, this way we can keep 2 clients around
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,137 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
- }
- public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
- {
- return listDir(containerName, virtualDirPath, null);
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(Utility.urlEncode(blobPath)).exists();
}
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
+ @VisibleForTesting
+ BlobServiceClient getRetriableBlobServiceClient(@Nonnull Integer maxAttempts)
{
- if (maxAttempts == null) {
- return null;
+ // To avoid keeping a ton of clients in memory, if maxAttempts is specified use a client with at least that many retries configured.
Review Comment:
this change is mainly to balance backwards compatibility with performance. the retry logic in the extension is such that some requests don't specify any retries (the default behavior in the previous client was to not retry at all), and some requests specify retries (these can come from different configs). the new azure client doesn't let you override the retry configuration for a client, so in the interest of not either 1. creating a client every time we make a request, 2. keeping a bunch of different clients cached in memory, i added the "at least retry attempts" logic, this way we can keep 2 clients around only
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1394330175
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,136 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(blobPath).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(blobPath).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(blobPath).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
+
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return listDir(containerName, virtualDirPath, null);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
- }
-
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
- {
- if (maxAttempts == null) {
- return null;
- }
- BlobRequestOptions requestOptions = new BlobRequestOptions();
- requestOptions.setRetryPolicyFactory(new RetryExponentialRetry(DELTA_BACKOFF_MS, maxAttempts));
- return requestOptions;
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(blobPath).exists();
}
@VisibleForTesting
- CloudBlobClient getCloudBlobClient()
+ BlobServiceClient getBlobServiceClient()
{
- return this.cloudBlobClient.get();
+ return this.blobServiceClient.get();
}
@VisibleForTesting
- ResultSegment<ListBlobItem> listBlobsWithPrefixInContainerSegmented(
+ PagedIterable<BlobItem> listBlobsWithPrefixInContainerSegmented(
final String containerName,
final String prefix,
- ResultContinuation continuationToken,
int maxResults
- ) throws StorageException, URISyntaxException
+ ) throws BlobStorageException
{
- CloudBlobContainer cloudBlobContainer = cloudBlobClient.get().getContainerReference(containerName);
- return cloudBlobContainer
- .listBlobsSegmented(
- prefix,
- /* Use flat blob listing here so that we get only blob types and not directories.*/
- USE_FLAT_BLOB_LISTING,
- EnumSet
- .noneOf(BlobListingDetails.class),
- maxResults,
- continuationToken,
- null,
- null
- );
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName);
+ return blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(prefix).setMaxResultsPerPage(maxResults),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
}
- private CloudBlobContainer getOrCreateCloudBlobContainer(final String containerName)
- throws StorageException, URISyntaxException
+ private BlobContainerClient getOrCreateBlobContainerClient(final String containerName)
{
- CloudBlobContainer cloudBlobContainer = cloudBlobClient.get().getContainerReference(containerName);
- cloudBlobContainer.createIfNotExists();
+ return blobServiceClient.get().createBlobContainerIfNotExists(containerName);
Review Comment:
i guess we probably should if we're caching the other ones as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "317brian (via GitHub)" <gi...@apache.org>.
317brian commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1394765309
##########
docs/development/extensions-core/azure.md:
##########
@@ -33,8 +33,10 @@ To use this Apache Druid extension, [include](../../configuration/extensions.md#
|--------|---------------|-----------|-------|
|`druid.storage.type`|azure||Must be set.|
|`druid.azure.account`||Azure Storage account name.|Must be set.|
-|`druid.azure.key`||Azure Storage account key.|Optional. Either set key or sharedAccessStorageToken but not both.|
-|`druid.azure.sharedAccessStorageToken`||Azure Shared Storage access token|Optional. Either set key or sharedAccessStorageToken but not both.|
+|`druid.azure.key`||Azure Storage account key.|Optional. Set one of key, sharedAccessStorageToken or useAzureCredentialsChain.|
+|`druid.azure.sharedAccessStorageToken`||Azure Shared Storage access token|Optional. Set one of key, sharedAccessStorageToken or useAzureCredentialsChain..|
+|`druid.azure.useAzureCredentialsChain`|Use [DefaultAzureCredential](https://learn.microsoft.com/en-us/java/api/overview/azure/identity-readme?view=azure-java-stable) for authentication|Optional. Set one of key, sharedAccessStorageToken or useAzureCredentialsChain.|False|
+|`druid.azure.managedIdentityClientId`|If you want to used managed identity auth in the DefaultAzureCredential. useAzureCredentialsChain must be true.||Optional.|
Review Comment:
Minor nit
```suggestion
|`druid.azure.managedIdentityClientId`|If you want to use managed identity authentication in the `DefaultAzureCredential`, `useAzureCredentialsChain` must be true.||Optional.|
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1439624904
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,137 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
- }
- public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
- {
- return listDir(containerName, virtualDirPath, null);
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(Utility.urlEncode(blobPath)).exists();
}
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
+ @VisibleForTesting
+ BlobServiceClient getRetriableBlobServiceClient(@Nonnull Integer maxAttempts)
Review Comment:
```suggestion
BlobServiceClient getRetriableBlobServiceClient(int maxAttempts)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1394331964
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,136 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(blobPath).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(blobPath).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(blobPath).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
+
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return listDir(containerName, virtualDirPath, null);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
- }
-
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
- {
- if (maxAttempts == null) {
- return null;
- }
- BlobRequestOptions requestOptions = new BlobRequestOptions();
- requestOptions.setRetryPolicyFactory(new RetryExponentialRetry(DELTA_BACKOFF_MS, maxAttempts));
- return requestOptions;
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(blobPath).exists();
}
@VisibleForTesting
- CloudBlobClient getCloudBlobClient()
+ BlobServiceClient getBlobServiceClient()
{
- return this.cloudBlobClient.get();
+ return this.blobServiceClient.get();
}
@VisibleForTesting
- ResultSegment<ListBlobItem> listBlobsWithPrefixInContainerSegmented(
+ PagedIterable<BlobItem> listBlobsWithPrefixInContainerSegmented(
final String containerName,
final String prefix,
- ResultContinuation continuationToken,
int maxResults
- ) throws StorageException, URISyntaxException
+ ) throws BlobStorageException
{
- CloudBlobContainer cloudBlobContainer = cloudBlobClient.get().getContainerReference(containerName);
- return cloudBlobContainer
- .listBlobsSegmented(
- prefix,
- /* Use flat blob listing here so that we get only blob types and not directories.*/
- USE_FLAT_BLOB_LISTING,
- EnumSet
- .noneOf(BlobListingDetails.class),
- maxResults,
- continuationToken,
- null,
- null
- );
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName);
+ return blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(prefix).setMaxResultsPerPage(maxResults),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
}
- private CloudBlobContainer getOrCreateCloudBlobContainer(final String containerName)
- throws StorageException, URISyntaxException
+ private BlobContainerClient getOrCreateBlobContainerClient(final String containerName)
{
- CloudBlobContainer cloudBlobContainer = cloudBlobClient.get().getContainerReference(containerName);
- cloudBlobContainer.createIfNotExists();
+ return blobServiceClient.get().createBlobContainerIfNotExists(containerName);
Review Comment:
realizing we probably need to check the maxRetries field as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1430565962
##########
extensions-core/azure-extensions/pom.xml:
##########
@@ -40,29 +40,30 @@
<version>${project.parent.version}</version>
<scope>provided</scope>
</dependency>
-
<dependency>
- <groupId>com.microsoft.azure</groupId>
- <artifactId>azure-storage</artifactId>
- <version>8.6.0</version>
- <exclusions>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-identity</artifactId>
+ <version>1.10.1</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob</artifactId>
+ <version>12.24.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob-batch</artifactId>
+ <version>12.20.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-common</artifactId>
+ <version>12.23.0</version>
+ </dependency>
Review Comment:
unfortunately azure does not lineup these versions
##########
extensions-core/azure-extensions/pom.xml:
##########
@@ -40,29 +40,30 @@
<version>${project.parent.version}</version>
<scope>provided</scope>
</dependency>
-
<dependency>
- <groupId>com.microsoft.azure</groupId>
- <artifactId>azure-storage</artifactId>
- <version>8.6.0</version>
- <exclusions>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-identity</artifactId>
+ <version>1.10.1</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob</artifactId>
+ <version>12.24.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob-batch</artifactId>
+ <version>12.20.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-common</artifactId>
+ <version>12.23.0</version>
+ </dependency>
Review Comment:
unfortunately azure does not sync these versions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1394241215
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,136 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(blobPath).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(blobPath).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(blobPath).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
+
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return listDir(containerName, virtualDirPath, null);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
- }
-
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
- {
- if (maxAttempts == null) {
- return null;
- }
- BlobRequestOptions requestOptions = new BlobRequestOptions();
- requestOptions.setRetryPolicyFactory(new RetryExponentialRetry(DELTA_BACKOFF_MS, maxAttempts));
- return requestOptions;
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(blobPath).exists();
}
@VisibleForTesting
- CloudBlobClient getCloudBlobClient()
+ BlobServiceClient getBlobServiceClient()
{
- return this.cloudBlobClient.get();
+ return this.blobServiceClient.get();
}
@VisibleForTesting
- ResultSegment<ListBlobItem> listBlobsWithPrefixInContainerSegmented(
+ PagedIterable<BlobItem> listBlobsWithPrefixInContainerSegmented(
final String containerName,
final String prefix,
- ResultContinuation continuationToken,
int maxResults
- ) throws StorageException, URISyntaxException
+ ) throws BlobStorageException
{
- CloudBlobContainer cloudBlobContainer = cloudBlobClient.get().getContainerReference(containerName);
- return cloudBlobContainer
- .listBlobsSegmented(
- prefix,
- /* Use flat blob listing here so that we get only blob types and not directories.*/
- USE_FLAT_BLOB_LISTING,
- EnumSet
- .noneOf(BlobListingDetails.class),
- maxResults,
- continuationToken,
- null,
- null
- );
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName);
+ return blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(prefix).setMaxResultsPerPage(maxResults),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
}
- private CloudBlobContainer getOrCreateCloudBlobContainer(final String containerName)
- throws StorageException, URISyntaxException
+ private BlobContainerClient getOrCreateBlobContainerClient(final String containerName)
{
- CloudBlobContainer cloudBlobContainer = cloudBlobClient.get().getContainerReference(containerName);
- cloudBlobContainer.createIfNotExists();
+ return blobServiceClient.get().createBlobContainerIfNotExists(containerName);
Review Comment:
do you not want to cache this client?
##########
docs/development/extensions-core/azure.md:
##########
@@ -33,8 +33,10 @@ To use this Apache Druid extension, [include](../../configuration/extensions.md#
|--------|---------------|-----------|-------|
|`druid.storage.type`|azure||Must be set.|
|`druid.azure.account`||Azure Storage account name.|Must be set.|
-|`druid.azure.key`||Azure Storage account key.|Optional. Either set key or sharedAccessStorageToken but not both.|
-|`druid.azure.sharedAccessStorageToken`||Azure Shared Storage access token|Optional. Either set key or sharedAccessStorageToken but not both.|
+|`druid.azure.key`||Azure Storage account key.|Optional. Set one of key, sharedAccessStorageToken or useAzureCredentialsChain.|
+|`druid.azure.sharedAccessStorageToken`||Azure Shared Storage access token|Optional. Set one of key, sharedAccessStorageToken or useAzureCredentialsChain..|
+|`druid.azure.useAzureCredentialsChain`|Use [DefaultAzureCredential](https://learn.microsoft.com/en-us/java/api/overview/azure/identity-readme?view=azure-java-stable) for authentication|Optional. Set one of key, sharedAccessStorageToken or useAzureCredentialsChain.|False|
+|`druid.azure.managedIdentityClientId`||If you want to used managed identity auth in the DefaultAzureCredential. useAzureCredentialsChain must be true.|Optional.|
Review Comment:
```suggestion
|`druid.azure.managedIdentityClientId`|If you want to used managed identity auth in the DefaultAzureCredential. useAzureCredentialsChain must be true.||Optional.|
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1409922286
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -69,42 +73,41 @@ public class AzureStorage
*
* See OmniDataSegmentKiller for how DataSegmentKillers are initialized.
*/
- private final Supplier<CloudBlobClient> cloudBlobClient;
+ private final Supplier<BlobServiceClient> blobServiceClient;
+ private final AzureClientFactory azureClientFactory;
+ private final ConcurrentHashMap<String, Pair<Integer, BlobContainerClient>> blobContainerClients = new ConcurrentHashMap<>();
public AzureStorage(
- Supplier<CloudBlobClient> cloudBlobClient
+ Supplier<BlobServiceClient> blobServiceClient,
Review Comment:
i think i can definitely just remove this supplier and lazily call factory.create instead, but i would prefer to make that as a followup change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1420500841
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureTaskLogsConfig.java:
##########
@@ -37,19 +36,14 @@ public class AzureTaskLogsConfig
@NotNull
private String prefix = null;
- @JsonProperty
- @Min(1)
- private int maxTries = 3;
Review Comment:
i don't really see this config documented anywhere and its not supported with S3TaskLogs or GoogleTaskLogs so i just removed it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1394623751
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,136 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(blobPath).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(blobPath).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(blobPath).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
+
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return listDir(containerName, virtualDirPath, null);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
- }
-
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
- {
- if (maxAttempts == null) {
- return null;
- }
- BlobRequestOptions requestOptions = new BlobRequestOptions();
- requestOptions.setRetryPolicyFactory(new RetryExponentialRetry(DELTA_BACKOFF_MS, maxAttempts));
- return requestOptions;
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(blobPath).exists();
}
@VisibleForTesting
- CloudBlobClient getCloudBlobClient()
+ BlobServiceClient getBlobServiceClient()
{
- return this.cloudBlobClient.get();
+ return this.blobServiceClient.get();
}
@VisibleForTesting
- ResultSegment<ListBlobItem> listBlobsWithPrefixInContainerSegmented(
+ PagedIterable<BlobItem> listBlobsWithPrefixInContainerSegmented(
final String containerName,
final String prefix,
- ResultContinuation continuationToken,
int maxResults
- ) throws StorageException, URISyntaxException
+ ) throws BlobStorageException
{
- CloudBlobContainer cloudBlobContainer = cloudBlobClient.get().getContainerReference(containerName);
- return cloudBlobContainer
- .listBlobsSegmented(
- prefix,
- /* Use flat blob listing here so that we get only blob types and not directories.*/
- USE_FLAT_BLOB_LISTING,
- EnumSet
- .noneOf(BlobListingDetails.class),
- maxResults,
- continuationToken,
- null,
- null
- );
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName);
+ return blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(prefix).setMaxResultsPerPage(maxResults),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
}
- private CloudBlobContainer getOrCreateCloudBlobContainer(final String containerName)
- throws StorageException, URISyntaxException
+ private BlobContainerClient getOrCreateBlobContainerClient(final String containerName)
{
- CloudBlobContainer cloudBlobContainer = cloudBlobClient.get().getContainerReference(containerName);
- cloudBlobContainer.createIfNotExists();
+ return blobServiceClient.get().createBlobContainerIfNotExists(containerName);
Review Comment:
updated this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1439612932
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureAccountConfig.java:
##########
@@ -46,6 +46,12 @@ public class AzureAccountConfig
@JsonProperty
private String sharedAccessStorageToken;
+ @JsonProperty
+ private String managedIdentityClientId;
Review Comment:
Can we add validation to error out if the user sets this without setting `useAzureCredentialsChain = true` as called out in the docs
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureClientFactory.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.druid.storage.azure;
+
+import com.azure.core.http.policy.ExponentialBackoffOptions;
+import com.azure.core.http.policy.RetryOptions;
+import com.azure.identity.DefaultAzureCredentialBuilder;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.common.StorageSharedKeyCredential;
+
+import javax.annotation.Nonnull;
+import java.time.Duration;
+
+/**
+ * Factory class for generating BlobServiceClient objects.
+ */
+public class AzureClientFactory
+{
+
+ private final AzureAccountConfig config;
+
+ public AzureClientFactory(AzureAccountConfig config)
+ {
+ this.config = config;
+ }
+
+ public BlobServiceClient getBlobServiceClient()
+ {
+ return getAuthenticatedBlobServiceClientBuilder().buildClient();
+ }
+
+ /**
+ * Azure doesn't let us override retryConfigs on BlobServiceClient so we need a second instance.
+ * @param retryCount number of retries
+ * @return BlobServiceClient with a custom retryCount
+ */
+ public BlobServiceClient getRetriableBlobServiceClient(@Nonnull Integer retryCount)
Review Comment:
Is there a reason for this factory to return both a retriable and non-retriable client?
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,137 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
- }
- public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
- {
- return listDir(containerName, virtualDirPath, null);
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(Utility.urlEncode(blobPath)).exists();
}
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
+ @VisibleForTesting
+ BlobServiceClient getRetriableBlobServiceClient(@Nonnull Integer maxAttempts)
Review Comment:
```suggestion
BlobServiceClient getRetriableBlobServiceClient(intmaxAttempts)
```
##########
extensions-core/azure-extensions/pom.xml:
##########
@@ -40,29 +40,30 @@
<version>${project.parent.version}</version>
<scope>provided</scope>
</dependency>
-
<dependency>
- <groupId>com.microsoft.azure</groupId>
- <artifactId>azure-storage</artifactId>
- <version>8.6.0</version>
- <exclusions>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-identity</artifactId>
+ <version>1.10.1</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob</artifactId>
+ <version>12.24.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob-batch</artifactId>
+ <version>12.20.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-common</artifactId>
+ <version>12.23.0</version>
+ </dependency>
Review Comment:
https://learn.microsoft.com/en-us/azure/developer/java/sdk/get-started-maven#use-the-azure-sdk-for-java-build-tool - have you looked into adding this to see if the azure libraries are added correctly?
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,137 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference = container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null, getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference = getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return getOrCreateBlobContainerClient(containerName).getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName, blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath, null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null, getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).openInputStream(new BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation, getRequestOptionsWithRetry(maxAttempts), null);
- }
- public List<String> listDir(final String containerName, final String virtualDirPath)
- throws URISyntaxException, StorageException
- {
- return listDir(containerName, virtualDirPath, null);
+ BlobBatchClient blobBatchClient = new BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName, maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths), DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient = getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null, getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob -> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null, getRequestOptionsWithRetry(maxAttempts), null);
+ return getOrCreateBlobContainerClient(container, maxAttempts).getBlobClient(Utility.urlEncode(blobPath)).exists();
}
- /**
- * If maxAttempts is provided, this method returns request options with retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
+ @VisibleForTesting
+ BlobServiceClient getRetriableBlobServiceClient(@Nonnull Integer maxAttempts)
{
- if (maxAttempts == null) {
- return null;
+ // To avoid keeping a ton of clients in memory, if maxAttempts is specified use a client with at least that many retries configured.
Review Comment:
This seems a little confusing - why would we want max attempts to mean "at least retry attempts".
Also, if this is the desired behavior, we can push this logic into the factory instead of having it in the storage class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1439719943
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureClientFactory.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.druid.storage.azure;
+
+import com.azure.core.http.policy.ExponentialBackoffOptions;
+import com.azure.core.http.policy.RetryOptions;
+import com.azure.identity.DefaultAzureCredentialBuilder;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.common.StorageSharedKeyCredential;
+
+import javax.annotation.Nonnull;
+import java.time.Duration;
+
+/**
+ * Factory class for generating BlobServiceClient objects.
+ */
+public class AzureClientFactory
+{
+
+ private final AzureAccountConfig config;
+
+ public AzureClientFactory(AzureAccountConfig config)
+ {
+ this.config = config;
+ }
+
+ public BlobServiceClient getBlobServiceClient()
+ {
+ return getAuthenticatedBlobServiceClientBuilder().buildClient();
+ }
+
+ /**
+ * Azure doesn't let us override retryConfigs on BlobServiceClient so we need a second instance.
+ * @param retryCount number of retries
+ * @return BlobServiceClient with a custom retryCount
+ */
+ public BlobServiceClient getRetriableBlobServiceClient(@Nonnull Integer retryCount)
Review Comment:
explained this in a below comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "YongGang (via GitHub)" <gi...@apache.org>.
YongGang commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1421969529
##########
extensions-core/azure-extensions/pom.xml:
##########
@@ -40,29 +40,30 @@
<version>${project.parent.version}</version>
<scope>provided</scope>
</dependency>
-
<dependency>
- <groupId>com.microsoft.azure</groupId>
- <artifactId>azure-storage</artifactId>
- <version>8.6.0</version>
- <exclusions>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-identity</artifactId>
+ <version>1.10.1</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob</artifactId>
+ <version>12.24.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob-batch</artifactId>
+ <version>12.20.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-common</artifactId>
+ <version>12.23.0</version>
+ </dependency>
Review Comment:
Should these dependencies be the same version?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1394502069
##########
extensions-core/azure-extensions/src/test/java/org/apache/druid/storage/azure/AzureStorageTest.java:
##########
@@ -19,53 +19,91 @@
package org.apache.druid.storage.azure;
+import com.azure.core.http.rest.PagedIterable;
+import com.azure.core.http.rest.PagedResponse;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.models.BlobItem;
+import com.azure.storage.blob.models.BlobItemProperties;
+import com.azure.storage.blob.models.BlobStorageException;
import com.google.common.collect.ImmutableList;
-import com.microsoft.azure.storage.StorageException;
-import com.microsoft.azure.storage.blob.CloudBlobClient;
-import com.microsoft.azure.storage.blob.CloudBlobContainer;
-import com.microsoft.azure.storage.blob.CloudBlockBlob;
-import com.microsoft.azure.storage.blob.ListBlobItem;
+import org.apache.druid.common.guava.SettableSupplier;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRunner;
+import org.easymock.EasyMockSupport;
+import org.easymock.Mock;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.junit.runner.RunWith;
import org.mockito.ArgumentMatchers;
import org.mockito.Mockito;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.List;
-public class AzureStorageTest
+@RunWith(EasyMockRunner.class)
+public class AzureStorageTest extends EasyMockSupport
{
AzureStorage azureStorage;
- CloudBlobClient cloudBlobClient = Mockito.mock(CloudBlobClient.class);
- CloudBlobContainer cloudBlobContainer = Mockito.mock(CloudBlobContainer.class);
+ BlobServiceClient blobServiceClient = Mockito.mock(BlobServiceClient.class);
+ BlobContainerClient blobContainerClient = Mockito.mock(BlobContainerClient.class);
+
+ @Mock
+ AzureClientFactory azureClientFactory;
@Before
- public void setup() throws URISyntaxException, StorageException
+ public void setup() throws BlobStorageException
{
- Mockito.doReturn(cloudBlobContainer).when(cloudBlobClient).getContainerReference(ArgumentMatchers.anyString());
- azureStorage = new AzureStorage(() -> cloudBlobClient);
+ Mockito.doReturn(blobContainerClient).when(blobServiceClient).createBlobContainerIfNotExists(ArgumentMatchers.anyString());
+ azureStorage = new AzureStorage(() -> blobServiceClient, azureClientFactory);
}
@Test
- public void testListDir() throws URISyntaxException, StorageException
+ public void testListDir() throws BlobStorageException
{
- List<ListBlobItem> listBlobItems = ImmutableList.of(
- new CloudBlockBlob(new URI("azure://dummy.com/container/blobName"))
- );
-
- Mockito.doReturn(listBlobItems).when(cloudBlobContainer).listBlobs(
- ArgumentMatchers.anyString(),
- ArgumentMatchers.anyBoolean(),
- ArgumentMatchers.any(),
+ BlobItem blobItem = new BlobItem().setName("blobName").setProperties(new BlobItemProperties().setContentLength(10L));
+ SettableSupplier<PagedResponse<BlobItem>> supplier = new SettableSupplier<>();
+ supplier.set(new TestPagedResponse<>(ImmutableList.of(blobItem)));
+ PagedIterable<BlobItem> pagedIterable = new PagedIterable<>(supplier);
+ Mockito.doReturn(pagedIterable).when(blobContainerClient).listBlobs(
ArgumentMatchers.any(),
ArgumentMatchers.any()
);
+ EasyMock.expect(azureClientFactory.getBlobContainerClient("test", null)).andReturn(blobContainerClient).times(1);
+
+ replayAll();
Assert.assertEquals(ImmutableList.of("blobName"), azureStorage.listDir("test", ""));
+ verifyAll();
+ }
+
+ @Test
+ public void testListDir_withMaxAttempts_factoryCreatesNewContainerClient() throws BlobStorageException
+ {
+ Integer maxAttempts = 5;
Review Comment:
## Boxed variable is never null
The variable 'maxAttempts' is only assigned values of primitive type and is never 'null', but it is declared with the boxed type 'Integer'.
[Show more details](https://github.com/apache/druid/security/code-scanning/5979)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Azure client upgrade to allow identity options (druid)
Posted by "georgew5656 (via GitHub)" <gi...@apache.org>.
georgew5656 merged PR #15287:
URL: https://github.com/apache/druid/pull/15287
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org