You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2020/07/25 21:44:30 UTC
[GitHub] [nifi] stijn192 opened a new pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
stijn192 opened a new pull request #4430:
URL: https://github.com/apache/nifi/pull/4430
Thank you for submitting a contribution to Apache NiFi.
Please provide a short description of the PR here:
#### Description of PR
_Enables usage of "Azure Identity" library by upgrading `azure-storage` (8.4.0) to the latest `azure-storage-blob` (12.7.0) and `azure-storage-queue` (12.5.2)_ This upgrade deserves a seperate issue, since it needs rewriting of the current blob and queue client builder code, as well as proxy configuration.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [x] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [x] Does your PR title start with **NIFI-XXXX** where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [x] Has your PR been rebased against the latest commit within the target branch (typically `main`)?
- [ ] Is your initial contribution a single, squashed commit? _Additional commits in response to PR reviewer feedback should be made on this branch and pushed to allow change tracking. Do not `squash` or use `--force` when pushing to allow for clean monitoring of changes._
### For code changes:
- [x] Have you ensured that the full suite of tests is executed via `mvn -Pcontrib-check clean install` at the root `nifi` folder?
- [x] Have you written or updated unit tests to verify your changes?
- [x] Have you verified that the full build is successful on JDK 8?
- [x] Have you verified that the full build is successful on JDK 11?
- [x] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the `LICENSE` file, including the main `LICENSE` file under `nifi-assembly`?
- [ ] If applicable, have you updated the `NOTICE` file, including the main `NOTICE` file found under `nifi-assembly`?
- [ ] If adding new Properties, have you added `.displayName` in addition to .name (programmatic access) for each of the new properties?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI for build issues and submit an update to your PR as soon as possible.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] jfrazee commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
jfrazee commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463197857
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
Review comment:
This will break things for any downstream `AzureStorageUtils` users.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463234889
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
final AzureStorageCredentialsDetails storageCredentialsDetails = getStorageCredentialsDetails(context, flowFile);
- final CloudStorageAccount cloudStorageAccount = new CloudStorageAccount(
- storageCredentialsDetails.getStorageCredentials(),
- true,
- storageCredentialsDetails.getStorageSuffix(),
- storageCredentialsDetails.getStorageAccountName());
- final CloudBlobClient cloudBlobClient = cloudStorageAccount.createCloudBlobClient();
-
- return cloudBlobClient;
+
+ final String storageSuffix = StringUtils.isNotBlank(storageCredentialsDetails.getStorageSuffix())
+ ? storageCredentialsDetails.getStorageSuffix()
+ : "blob.core.windows.net";
Review comment:
@turcsanyip good question, actually now (to limit the size of this already quite large PR) I've chosen to stick to the current functionality, where there is and was no working custom endpoint yet in the Azure Queue Storage processors. (read: the value could be set on the processor, but nothing happened when configuring it). Hence I've chosen for a more hardcoded endpoint which is visible over at line 72 on the [AbstractAzureQueueStorage.java](https://github.com/stijn192/nifi/blob/2cfadeedc9e6546a58cf903074cf9913c9f11f66/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/queue/AbstractAzureQueueStorage.java#L72)
We could implement the custom endpoint now, or make it a seperate feature, where we will end up having to make a separate credential class (such as the `ADLSCredentialsDetails` class), as far as I can see it. Tested sending to the `blob.core.windows.net` endpoint will actually fail and will route message to fail output from NiFi processor.
---
@jfrazee - from v12 api's onwards, endpoint need to be full url's and will give an `malformed endpoint` error when not setting the full url. See also here:
[QueueServiceClientBuilder Microsoft API Docs](https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/queue/QueueServiceClientBuilder.html)
[BlobServiceClientBuilder Microsoft API Docs](https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/blob/BlobServiceClientBuilder.html)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] jfrazee commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
jfrazee commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463246197
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
Review comment:
I made an overall comment on the PR. We wouldn't necessarily have to have two sets of processors, but let's say we say we're just going to break it, I think now would be a good time to improve things by wrapping it in something we control.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] MuazmaZ commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
MuazmaZ commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463327284
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
##########
@@ -114,14 +112,14 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
}
try {
- blob.upload(in, length, null, null, operationContext);
+ blob.upload(in, length);
BlobProperties properties = blob.getProperties();
attributes.put("azure.container", containerName);
- attributes.put("azure.primaryUri", blob.getSnapshotQualifiedUri().toString());
- attributes.put("azure.etag", properties.getEtag());
+ attributes.put("azure.primaryUri", blob.getBlobUrl());
Review comment:
Got response: This is by design.
A lot of our APIs that accept blob URLs expect URL encoded URLs, so when we return the URL, it's probably best that it is URL encoded.
When you say it was different earlier, could you provide the version number between which it changed? @turcsanyip
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r475589378
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
Review comment:
@jfrazee thank you for your comments the other day. Since Microsoft/Azure SDK's ServiceClient are not implements any service, I've created an custom AzureServiceClient abstract class that will be extended by AzureBlob / AzureQueue ServiceClient abstract classes, in which we wrap the functions for this service client. Although I feel this is only a small level of control - from here onwards we could include required functions from these wrapper objects.
What do you think about this solution?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
turcsanyip commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r462824772
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
##########
@@ -114,14 +112,14 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
}
try {
- blob.upload(in, length, null, null, operationContext);
+ blob.upload(in, length);
BlobProperties properties = blob.getProperties();
attributes.put("azure.container", containerName);
- attributes.put("azure.primaryUri", blob.getSnapshotQualifiedUri().toString());
- attributes.put("azure.etag", properties.getEtag());
+ attributes.put("azure.primaryUri", blob.getBlobUrl());
Review comment:
@MuazmaZ Do you have any info if it is a bug or a feature in the azure-storage-blob client lib?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r475399575
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/services/azure/storage/AzureStorageCredentialsControllerService.java
##########
@@ -77,16 +78,37 @@
final String accountKey = validationContext.getProperty(AzureStorageUtils.ACCOUNT_KEY).getValue();
final String sasToken = validationContext.getProperty(AzureStorageUtils.PROP_SAS_TOKEN).getValue();
- if (StringUtils.isBlank(accountKey) && StringUtils.isBlank(sasToken)) {
+ final Boolean accountKeyIsSet = validationContext.getProperty(AzureStorageUtils.ACCOUNT_KEY).isSet();
+ final Boolean sasTokenIsSet = validationContext.getProperty(AzureStorageUtils.PROP_SAS_TOKEN).isSet();
+
+ if (StringUtils.isAllBlank(accountKey, sasToken)) {
results.add(new ValidationResult.Builder().subject("AzureStorageCredentialsControllerService")
.valid(false)
.explanation("either " + AzureStorageUtils.ACCOUNT_KEY.getDisplayName() + " or " + AzureStorageUtils.PROP_SAS_TOKEN.getDisplayName() + " is required")
.build());
- } else if (StringUtils.isNotBlank(accountKey) && StringUtils.isNotBlank(sasToken)) {
- results.add(new ValidationResult.Builder().subject("AzureStorageCredentialsControllerService")
+ }
+
+ if (BooleanUtils.xor(new Boolean[] { accountKeyIsSet, sasTokenIsSet })) {
+ if (accountKeyIsSet) {
+ if (StringUtils.isBlank(accountKey)) {
+ results.add(new ValidationResult.Builder().subject("AzureStorageCredentialsControllerService")
.valid(false)
- .explanation("cannot set both " + AzureStorageUtils.ACCOUNT_KEY.getDisplayName() + " and " + AzureStorageUtils.PROP_SAS_TOKEN.getDisplayName())
+ .explanation(AzureStorageUtils.ACCOUNT_KEY.getDisplayName() + " must be set when using Account Key authentication.")
.build());
+ }
+ } else if (sasTokenIsSet) {
+ if (StringUtils.isBlank(sasToken)) {
+ results.add(new ValidationResult.Builder().subject("AzureStorageCredentialsControllerService")
+ .valid(false)
+ .explanation(AzureStorageUtils.PROP_SAS_TOKEN.getDisplayName() + " must be set when using SAS token authentication.")
+ .build());
+ }
Review comment:
Also added an validator on the parameter.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463237030
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/DeleteAzureBlobStorage.java
##########
@@ -48,20 +48,17 @@
@InputRequirement(Requirement.INPUT_REQUIRED)
public class DeleteAzureBlobStorage extends AbstractAzureBlobProcessor {
- private static final AllowableValue DELETE_SNAPSHOTS_NONE = new AllowableValue(DeleteSnapshotsOption.NONE.name(), "None", "Delete the blob only.");
-
- private static final AllowableValue DELETE_SNAPSHOTS_ALSO = new AllowableValue(DeleteSnapshotsOption.INCLUDE_SNAPSHOTS.name(), "Include Snapshots", "Delete the blob and its snapshots.");
+ private static final AllowableValue DELETE_SNAPSHOTS_ALSO = new AllowableValue(DeleteSnapshotsOptionType.INCLUDE.name(), "Include Snapshots", "Delete the blob and its snapshots.");
- private static final AllowableValue DELETE_SNAPSHOTS_ONLY = new AllowableValue(DeleteSnapshotsOption.DELETE_SNAPSHOTS_ONLY.name(), "Delete Snapshots Only", "Delete only the blob's snapshots.");
+ private static final AllowableValue DELETE_SNAPSHOTS_ONLY = new AllowableValue(DeleteSnapshotsOptionType.ONLY.name(), "Delete Snapshots Only", "Delete only the blob's snapshots.");
Review comment:
@turcsanyip Thank you for this remark! Maybe also due to my inexperience with the NiFi codebase, but my go to solution would be to implement a solution such as this:
```
// translation enum for backwards compatability from DeleteSnapshotsOption (v8 api) -> DeleteSnapshotsOptionType (v12 api)
public enum DeleteSnapshotsOption {
// don't delete snapshots
NONE(null),
// include snapshots on blob deletion
INCLUDE_SNAPSHOTS(DeleteSnapshotsOptionType.INCLUDE),
// only delete snapshots of blob and don't delete blob itself
DELETE_SNAPSHOTS_ONLY(DeleteSnapshotsOptionType.ONLY);
private final DeleteSnapshotsOptionType deleteSnapshotsOptionType;
DeleteSnapshotsOption(DeleteSnapshotsOptionType type) {
this.deleteSnapshotsOptionType = type;
}
public DeleteSnapshotsOptionType getValue() {
return this.deleteSnapshotsOptionType;
}
}
```
would this suffice to achieve backwards compatability?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r475586661
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -37,17 +28,22 @@
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.proxy.ProxyConfiguration;
-import org.apache.nifi.proxy.ProxySpec;
import org.apache.nifi.services.azure.storage.AzureStorageCredentialsDetails;
import org.apache.nifi.services.azure.storage.AzureStorageCredentialsService;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+
public final class AzureStorageUtils {
+
public static final String BLOCK = "Block";
public static final String PAGE = "Page";
+ public static final String APPEND = "Append";
Review comment:
These were used for setting blob type, but this is no longer necessary since new API.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -37,17 +28,22 @@
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.proxy.ProxyConfiguration;
-import org.apache.nifi.proxy.ProxySpec;
import org.apache.nifi.services.azure.storage.AzureStorageCredentialsDetails;
import org.apache.nifi.services.azure.storage.AzureStorageCredentialsService;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+
public final class AzureStorageUtils {
+
public static final String BLOCK = "Block";
public static final String PAGE = "Page";
+ public static final String APPEND = "Append";
Review comment:
These were used for setting blob type, but this is no longer necessary since new sdk.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463239345
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
##########
@@ -114,14 +112,14 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
}
try {
- blob.upload(in, length, null, null, operationContext);
+ blob.upload(in, length);
BlobProperties properties = blob.getProperties();
attributes.put("azure.container", containerName);
- attributes.put("azure.primaryUri", blob.getSnapshotQualifiedUri().toString());
- attributes.put("azure.etag", properties.getEtag());
+ attributes.put("azure.primaryUri", blob.getBlobUrl());
Review comment:
thank you, looking forward to hear more about 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] jfrazee commented on pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
jfrazee commented on pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#issuecomment-666653028
I have a very general concern about this change. The return type of `createBlobServiceClient()` changes from `CloudBlobClient` to `BlobServiceClient` in `AzureStorageUtils`, and `createQueueServiceClient()` changes from `CloudQueueClient` to `QueueServiceClient`, all of which are types from the Microsoft/Azure SDKs.
Arguably, being public, it would have been better to not do this in the first place, but that's a benefit of hindsight. I see a few obvious choices with this:
* Break it and don't worry about it since it's not necessarily supposed to be used elsewhere.
* Break it but future proof it a little bit and wrap it in another class we control.
* Leave the old ones in place for a little while and add new ones that may/may not be wrapped.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 closed pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 closed pull request #4430:
URL: https://github.com/apache/nifi/pull/4430
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463234889
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
final AzureStorageCredentialsDetails storageCredentialsDetails = getStorageCredentialsDetails(context, flowFile);
- final CloudStorageAccount cloudStorageAccount = new CloudStorageAccount(
- storageCredentialsDetails.getStorageCredentials(),
- true,
- storageCredentialsDetails.getStorageSuffix(),
- storageCredentialsDetails.getStorageAccountName());
- final CloudBlobClient cloudBlobClient = cloudStorageAccount.createCloudBlobClient();
-
- return cloudBlobClient;
+
+ final String storageSuffix = StringUtils.isNotBlank(storageCredentialsDetails.getStorageSuffix())
+ ? storageCredentialsDetails.getStorageSuffix()
+ : "blob.core.windows.net";
Review comment:
@turcsanyip good question, actually now (to limit the size of this already quite large PR) I've chosen to stick to the current functionality, where there is and was no working custom endpoint yet in the Azure Queue Storage processors. (read: the value could be set on the processor, but nothing happened when configuring it). Hence I've chosen for a more hardcoded endpoint which is visible over at line 72 on the [AbstractAzureQueueStorage.java](https://github.com/stijn192/nifi/blob/2cfadeedc9e6546a58cf903074cf9913c9f11f66/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/queue/AbstractAzureQueueStorage.java#L72)
We could implement the custom endpoint now, or make it a seperate feature, where we willend up having to make a separate credential class, as far as I can see it. Tested sending to the `blob.core.windows.net` endpoint will actually fail and will give the following error:
```
java.lang.AssertionError: Expected all Transferred FlowFiles to go to success but 1 were routed to failure
```
---
@jfrazee - from v12 api's onwards, endpoint need to be full url's and will give an `malformed endpoint` error when not setting the full url. See also here:
[QueueServiceClientBuilder Microsoft API Docs](https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/queue/QueueServiceClientBuilder.html)
[BlobServiceClientBuilder Microsoft API Docs](https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/blob/BlobServiceClientBuilder.html)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463234889
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
final AzureStorageCredentialsDetails storageCredentialsDetails = getStorageCredentialsDetails(context, flowFile);
- final CloudStorageAccount cloudStorageAccount = new CloudStorageAccount(
- storageCredentialsDetails.getStorageCredentials(),
- true,
- storageCredentialsDetails.getStorageSuffix(),
- storageCredentialsDetails.getStorageAccountName());
- final CloudBlobClient cloudBlobClient = cloudStorageAccount.createCloudBlobClient();
-
- return cloudBlobClient;
+
+ final String storageSuffix = StringUtils.isNotBlank(storageCredentialsDetails.getStorageSuffix())
+ ? storageCredentialsDetails.getStorageSuffix()
+ : "blob.core.windows.net";
Review comment:
@turcsanyip good question, actually now (to limit the size of this already quite large PR) I've chosen to stick to the current functionality, where there is and was no working custom endpoint yet in the Azure Queue Storage processors. (read: the value could be set on the processor, but nothing happened when configuring it). Hence I've chosen for a more hardcoded endpoint which is visible over at line 72 on the [AbstractAzureQueueStorage.java](https://github.com/stijn192/nifi/blob/2cfadeedc9e6546a58cf903074cf9913c9f11f66/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/queue/AbstractAzureQueueStorage.java#L72)
We could implement the custom endpoint now, or make it a seperate feature, where we willend up having to make a separate credential class, as far as I can see it. Tested sending to the `blob.core.windows.net` endpoint will actually fail and will route message to fail output from NiFi processor.
---
@jfrazee - from v12 api's onwards, endpoint need to be full url's and will give an `malformed endpoint` error when not setting the full url. See also here:
[QueueServiceClientBuilder Microsoft API Docs](https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/queue/QueueServiceClientBuilder.html)
[BlobServiceClientBuilder Microsoft API Docs](https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/blob/BlobServiceClientBuilder.html)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
turcsanyip commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r462330562
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/services/azure/storage/AzureStorageCredentialsControllerService.java
##########
@@ -77,16 +78,37 @@
final String accountKey = validationContext.getProperty(AzureStorageUtils.ACCOUNT_KEY).getValue();
final String sasToken = validationContext.getProperty(AzureStorageUtils.PROP_SAS_TOKEN).getValue();
- if (StringUtils.isBlank(accountKey) && StringUtils.isBlank(sasToken)) {
+ final Boolean accountKeyIsSet = validationContext.getProperty(AzureStorageUtils.ACCOUNT_KEY).isSet();
+ final Boolean sasTokenIsSet = validationContext.getProperty(AzureStorageUtils.PROP_SAS_TOKEN).isSet();
+
+ if (StringUtils.isAllBlank(accountKey, sasToken)) {
results.add(new ValidationResult.Builder().subject("AzureStorageCredentialsControllerService")
.valid(false)
.explanation("either " + AzureStorageUtils.ACCOUNT_KEY.getDisplayName() + " or " + AzureStorageUtils.PROP_SAS_TOKEN.getDisplayName() + " is required")
.build());
- } else if (StringUtils.isNotBlank(accountKey) && StringUtils.isNotBlank(sasToken)) {
- results.add(new ValidationResult.Builder().subject("AzureStorageCredentialsControllerService")
+ }
+
+ if (BooleanUtils.xor(new Boolean[] { accountKeyIsSet, sasTokenIsSet })) {
+ if (accountKeyIsSet) {
+ if (StringUtils.isBlank(accountKey)) {
+ results.add(new ValidationResult.Builder().subject("AzureStorageCredentialsControllerService")
.valid(false)
- .explanation("cannot set both " + AzureStorageUtils.ACCOUNT_KEY.getDisplayName() + " and " + AzureStorageUtils.PROP_SAS_TOKEN.getDisplayName())
+ .explanation(AzureStorageUtils.ACCOUNT_KEY.getDisplayName() + " must be set when using Account Key authentication.")
.build());
+ }
+ } else if (sasTokenIsSet) {
+ if (StringUtils.isBlank(sasToken)) {
+ results.add(new ValidationResult.Builder().subject("AzureStorageCredentialsControllerService")
+ .valid(false)
+ .explanation(AzureStorageUtils.PROP_SAS_TOKEN.getDisplayName() + " must be set when using SAS token authentication.")
+ .build());
+ }
Review comment:
I believe these two cases should be handled via `StandardValidators.NON_BLANK_VALIDATOR` directly on the properties which would be less complicated / more straightforward.
Nevertheless, it is a good point to check blank values and the current `NON_EMPTY_VALIDATOR` is not the best option.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/utils/TestAzureStorageUtilsGetStorageCredentialsDetails.java
##########
@@ -147,18 +154,40 @@ private void configureControllerService(String accountName, String accountKey, S
processContext.setProperty(AzureStorageUtils.STORAGE_CREDENTIALS_SERVICE, CREDENTIALS_SERVICE_VALUE);
}
- private void assertStorageCredentialsDetailsAccountNameAndAccountKey(AzureStorageCredentialsDetails storageCredentialsDetails) {
+ private void assertStorageCredentialsDetailsAccountNameAndAccountKey(
+ AzureStorageCredentialsDetails storageCredentialsDetails) {
+
assertEquals(ACCOUNT_NAME_VALUE, storageCredentialsDetails.getStorageAccountName());
- assertTrue(storageCredentialsDetails.getStorageCredentials() instanceof StorageCredentialsAccountAndKey);
- StorageCredentialsAccountAndKey storageCredentials = (StorageCredentialsAccountAndKey) storageCredentialsDetails.getStorageCredentials();
+ assertTrue(storageCredentialsDetails.getCredentialType() == AzureStorageCredentialsDetails.CredentialType.STORAGE_ACCOUNT_KEY);
+ assertTrue(storageCredentialsDetails.getStorageSharedKeyCredential() instanceof StorageSharedKeyCredential);
+
+ // test credential object
+ StorageSharedKeyCredential storageCredentials = (StorageSharedKeyCredential) storageCredentialsDetails
Review comment:
Unnecessary type cast.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java
##########
@@ -176,45 +174,38 @@ protected String getDefaultTimePrecision() {
protected List<BlobInfo> performListing(final ProcessContext context, final Long minTimestamp) throws IOException {
String containerName = context.getProperty(AzureStorageUtils.CONTAINER).evaluateAttributeExpressions().getValue();
String prefix = context.getProperty(PROP_PREFIX).evaluateAttributeExpressions().getValue();
- if (prefix == null) {
- prefix = "";
- }
+
final List<BlobInfo> listing = new ArrayList<>();
try {
- CloudBlobClient blobClient = AzureStorageUtils.createCloudBlobClient(context, getLogger(), null);
- CloudBlobContainer container = blobClient.getContainerReference(containerName);
+ BlobServiceClient blobServiceClient = AzureStorageUtils.createBlobServiceClient(context, null);
+ BlobContainerClient blobContainerClient = blobServiceClient.getBlobContainerClient(containerName);
- final OperationContext operationContext = new OperationContext();
- AzureStorageUtils.setProxy(operationContext, context);
+ final ListBlobsOptions listBlobsOptions = new ListBlobsOptions()
+ .setPrefix(prefix)
+ .setDetails(new BlobListDetails()
+ .setRetrieveMetadata(true));
- for (ListBlobItem blob : container.listBlobs(prefix, true, EnumSet.of(BlobListingDetails.METADATA), null, operationContext)) {
- if (blob instanceof CloudBlob) {
- CloudBlob cloudBlob = (CloudBlob) blob;
- BlobProperties properties = cloudBlob.getProperties();
- StorageUri uri = cloudBlob.getSnapshotQualifiedStorageUri();
+ blobContainerClient.listBlobs().forEach(blob -> {
+ if (blob instanceof BlobItem) {
+ BlobItem blobItem = (BlobItem) blob;
Review comment:
Unnecessary type check / type cast. `listBlobs()` returns `PagedIterable<BlobItem>`.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureProxyUtils.java
##########
@@ -0,0 +1,115 @@
+
+/*
+ * 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.nifi.processors.azure.storage.utils;
+
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.util.Collection;
+
+import com.azure.core.http.HttpClient;
+import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.netty.NettyAsyncHttpClientBuilder;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxySpec;
+import org.apache.nifi.util.StringUtils;
+
+public class AzureProxyUtils {
+ private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP, ProxySpec.SOCKS};
+
+ private static ProxyOptions.Type getProxyOptionsTypeFromProxyType(final Proxy.Type proxyType) {
+ for (final ProxyOptions.Type item : ProxyOptions.Type.values()) {
+ if (item.toProxyType() == proxyType) {
+ return item;
+ }
+ }
+ return null;
+ }
+
+ public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE = ProxyConfiguration
+ .createProxyConfigPropertyDescriptor(false, PROXY_SPECS);
+
+ public static HttpClient createHttpClient(final PropertyContext propertyContext) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(propertyContext);
+ final ProxyOptions proxyOptions = getProxyOptions(proxyConfig);
+
+ final HttpClient client = new NettyAsyncHttpClientBuilder()
+ .proxy(proxyOptions)
+ .build();
+
+ return client;
+ }
+
+ public static void validateProxySpec(final ValidationContext context, final Collection<ValidationResult> results) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context);
+
+ final String proxyServerHost = proxyConfig.getProxyServerHost();
+ final Integer proxyServerPort = proxyConfig.getProxyServerPort();
+ final String proxyServerUser = proxyConfig.getProxyUserName();
+ final String proxyServerPassword = proxyConfig.getProxyUserPassword();
+
+ if ((StringUtils.isNotBlank(proxyServerHost) && proxyServerPort == null)
+ || (StringUtils.isBlank(proxyServerHost) && proxyServerPort != null)) {
+ results.add(new ValidationResult.Builder().subject("AzureProxyUtils Details").valid(false)
+ .explanation(
+ "When specifying address information, both `host` and `port` information must be provided.")
Review comment:
We use "normal" apostrophes (') in UI messages instead of (`).
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -180,21 +198,27 @@ public static AzureStorageCredentialsDetails createStorageCredentialsDetails(Pro
final String accountKey = context.getProperty(ACCOUNT_KEY).evaluateAttributeExpressions(attributes).getValue();
final String sasToken = context.getProperty(PROP_SAS_TOKEN).evaluateAttributeExpressions(attributes).getValue();
+ AzureStorageCredentialsDetails azureStorageCredentialsDetails;
+
if (StringUtils.isBlank(accountName)) {
throw new IllegalArgumentException(String.format("'%s' must not be empty.", ACCOUNT_NAME.getDisplayName()));
}
- StorageCredentials storageCredentials;
+ if (StringUtils.isAllBlank(accountKey, sasToken)) {
+ throw new IllegalArgumentException(String.format("Either '%s' or '%s' must be defined.", ACCOUNT_KEY.getDisplayName(),
+ PROP_SAS_TOKEN.getDisplayName()));
Review comment:
It is checked in the `else` branch below. Seems to be duplicated here.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -180,21 +198,27 @@ public static AzureStorageCredentialsDetails createStorageCredentialsDetails(Pro
final String accountKey = context.getProperty(ACCOUNT_KEY).evaluateAttributeExpressions(attributes).getValue();
final String sasToken = context.getProperty(PROP_SAS_TOKEN).evaluateAttributeExpressions(attributes).getValue();
+ AzureStorageCredentialsDetails azureStorageCredentialsDetails;
+
if (StringUtils.isBlank(accountName)) {
throw new IllegalArgumentException(String.format("'%s' must not be empty.", ACCOUNT_NAME.getDisplayName()));
}
- StorageCredentials storageCredentials;
+ if (StringUtils.isAllBlank(accountKey, sasToken)) {
+ throw new IllegalArgumentException(String.format("Either '%s' or '%s' must be defined.", ACCOUNT_KEY.getDisplayName(),
+ PROP_SAS_TOKEN.getDisplayName()));
+ }
- if (StringUtils.isNotBlank(accountKey)) {
- storageCredentials = new StorageCredentialsAccountAndKey(accountName, accountKey);
+ if(StringUtils.isNotBlank(accountKey)) {
Review comment:
Minor: `if (`
Please use a formatter (default settings of IntelliJ / Eclipse should be fine.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/pom.xml
##########
@@ -77,8 +76,19 @@
<version>${azure-eventhubs-eph.version}</version>
</dependency>
<dependency>
- <groupId>com.microsoft.azure</groupId>
- <artifactId>azure-storage</artifactId>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-queue</artifactId>
+ <version>${azure-storage-queue.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ </exclusions>
Review comment:
Could you please define it in the parent pom's `dependencyManagement` section? (similar to the blob dependency)
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
final AzureStorageCredentialsDetails storageCredentialsDetails = getStorageCredentialsDetails(context, flowFile);
- final CloudStorageAccount cloudStorageAccount = new CloudStorageAccount(
- storageCredentialsDetails.getStorageCredentials(),
- true,
- storageCredentialsDetails.getStorageSuffix(),
- storageCredentialsDetails.getStorageAccountName());
- final CloudBlobClient cloudBlobClient = cloudStorageAccount.createCloudBlobClient();
-
- return cloudBlobClient;
+
+ final String storageSuffix = StringUtils.isNotBlank(storageCredentialsDetails.getStorageSuffix())
+ ? storageCredentialsDetails.getStorageSuffix()
+ : "blob.core.windows.net";
+ final String endpoint = String.format("https://%s.%s", storageCredentialsDetails.getStorageAccountName(),
+ storageSuffix);
+
+ // use HttpClient object to allow proxy setting
+ final HttpClient httpClient = AzureProxyUtils.createHttpClient(context);
+ final BlobServiceClientBuilder blobServiceClientBuilder = new BlobServiceClientBuilder()
+ .endpoint(endpoint)
+ .httpClient(httpClient);
+ BlobServiceClient blobServiceClient;
+
+ switch(storageCredentialsDetails.getCredentialType()) {
+ case SAS_TOKEN:
+ blobServiceClient = blobServiceClientBuilder.sasToken(storageCredentialsDetails.getSasToken())
+ .buildClient();
+ break;
+ case STORAGE_ACCOUNT_KEY:
+ blobServiceClient = blobServiceClientBuilder.credential(storageCredentialsDetails.getStorageSharedKeyCredential())
+ .buildClient();
+ break;
+ default:
+ throw new IllegalArgumentException(String.format("Invalid credential type '%s'!", storageCredentialsDetails.getCredentialType().toString()));
+ }
+
+ return blobServiceClient;
}
public static AzureStorageCredentialsDetails getStorageCredentialsDetails(PropertyContext context, FlowFile flowFile) {
- final Map<String, String> attributes = flowFile != null ? flowFile.getAttributes() : Collections.emptyMap();
+ final Map<String, String> attributes = flowFile != null
+ ? flowFile.getAttributes()
+ : Collections.emptyMap();
Review comment:
Please try to avoid reformatting existing code (supposed it was formatted properly).
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
final AzureStorageCredentialsDetails storageCredentialsDetails = getStorageCredentialsDetails(context, flowFile);
- final CloudStorageAccount cloudStorageAccount = new CloudStorageAccount(
- storageCredentialsDetails.getStorageCredentials(),
- true,
- storageCredentialsDetails.getStorageSuffix(),
- storageCredentialsDetails.getStorageAccountName());
- final CloudBlobClient cloudBlobClient = cloudStorageAccount.createCloudBlobClient();
-
- return cloudBlobClient;
+
+ final String storageSuffix = StringUtils.isNotBlank(storageCredentialsDetails.getStorageSuffix())
+ ? storageCredentialsDetails.getStorageSuffix()
+ : "blob.core.windows.net";
+ final String endpoint = String.format("https://%s.%s", storageCredentialsDetails.getStorageAccountName(),
+ storageSuffix);
+
+ // use HttpClient object to allow proxy setting
+ final HttpClient httpClient = AzureProxyUtils.createHttpClient(context);
+ final BlobServiceClientBuilder blobServiceClientBuilder = new BlobServiceClientBuilder()
+ .endpoint(endpoint)
+ .httpClient(httpClient);
+ BlobServiceClient blobServiceClient;
+
+ switch(storageCredentialsDetails.getCredentialType()) {
Review comment:
Minor: `switch (`
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-services-api/pom.xml
##########
@@ -25,13 +25,12 @@
<dependencies>
<dependency>
- <groupId>com.microsoft.azure</groupId>
- <artifactId>azure-storage</artifactId>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob</artifactId>
Review comment:
`azure-storage-common` dependency would be sufficient in the services api module.
Now it is coming as the transitive dependency of `azure-storage-blob` but the blob itself is not needed here.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureProxyUtils.java
##########
@@ -0,0 +1,115 @@
+
+/*
+ * 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.nifi.processors.azure.storage.utils;
+
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.util.Collection;
+
+import com.azure.core.http.HttpClient;
+import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.netty.NettyAsyncHttpClientBuilder;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxySpec;
+import org.apache.nifi.util.StringUtils;
+
+public class AzureProxyUtils {
+ private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP, ProxySpec.SOCKS};
+
+ private static ProxyOptions.Type getProxyOptionsTypeFromProxyType(final Proxy.Type proxyType) {
+ for (final ProxyOptions.Type item : ProxyOptions.Type.values()) {
+ if (item.toProxyType() == proxyType) {
+ return item;
+ }
+ }
+ return null;
+ }
+
+ public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE = ProxyConfiguration
+ .createProxyConfigPropertyDescriptor(false, PROXY_SPECS);
+
+ public static HttpClient createHttpClient(final PropertyContext propertyContext) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(propertyContext);
+ final ProxyOptions proxyOptions = getProxyOptions(proxyConfig);
+
+ final HttpClient client = new NettyAsyncHttpClientBuilder()
+ .proxy(proxyOptions)
+ .build();
+
+ return client;
+ }
+
+ public static void validateProxySpec(final ValidationContext context, final Collection<ValidationResult> results) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context);
+
+ final String proxyServerHost = proxyConfig.getProxyServerHost();
+ final Integer proxyServerPort = proxyConfig.getProxyServerPort();
+ final String proxyServerUser = proxyConfig.getProxyUserName();
+ final String proxyServerPassword = proxyConfig.getProxyUserPassword();
+
+ if ((StringUtils.isNotBlank(proxyServerHost) && proxyServerPort == null)
+ || (StringUtils.isBlank(proxyServerHost) && proxyServerPort != null)) {
+ results.add(new ValidationResult.Builder().subject("AzureProxyUtils Details").valid(false)
+ .explanation(
+ "When specifying address information, both `host` and `port` information must be provided.")
+ .build());
+ }
+
+ if ((StringUtils.isBlank(proxyServerUser) && StringUtils.isNotBlank(proxyServerPassword))
+ || (StringUtils.isNotBlank(proxyServerUser) && StringUtils.isBlank(proxyServerPassword))) {
+ results.add(new ValidationResult.Builder().subject("AzureProxyUtils Details").valid(false)
+ .explanation(
+ "When specifying credentials, both `user` and `password` must be provided.")
+ .build());
+ }
+
+ ProxyConfiguration.validateProxySpec(context, results, PROXY_SPECS);
+ }
+
+ public static ProxyOptions getProxyOptions(final ProxyConfiguration proxyConfig) {
+ final String proxyServerHost = proxyConfig.getProxyServerHost();
+ final Integer proxyServerPort = proxyConfig.getProxyServerPort();
+ final String proxyServerUser = proxyConfig.getProxyUserName();
+ final String proxyServerPassword = proxyConfig.getProxyUserPassword();
+
+ final Boolean proxyServerProvided = StringUtils.isNotBlank(proxyServerHost) && proxyServerPort != null;
+ final Boolean proxyCredentialsProvided = StringUtils.isNotBlank(proxyServerUser) && StringUtils.isNotBlank(proxyServerPassword);
Review comment:
Variables can be primitive `boolean`.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java
##########
@@ -176,45 +174,38 @@ protected String getDefaultTimePrecision() {
protected List<BlobInfo> performListing(final ProcessContext context, final Long minTimestamp) throws IOException {
String containerName = context.getProperty(AzureStorageUtils.CONTAINER).evaluateAttributeExpressions().getValue();
String prefix = context.getProperty(PROP_PREFIX).evaluateAttributeExpressions().getValue();
- if (prefix == null) {
- prefix = "";
- }
+
final List<BlobInfo> listing = new ArrayList<>();
try {
- CloudBlobClient blobClient = AzureStorageUtils.createCloudBlobClient(context, getLogger(), null);
- CloudBlobContainer container = blobClient.getContainerReference(containerName);
+ BlobServiceClient blobServiceClient = AzureStorageUtils.createBlobServiceClient(context, null);
+ BlobContainerClient blobContainerClient = blobServiceClient.getBlobContainerClient(containerName);
- final OperationContext operationContext = new OperationContext();
- AzureStorageUtils.setProxy(operationContext, context);
+ final ListBlobsOptions listBlobsOptions = new ListBlobsOptions()
+ .setPrefix(prefix)
+ .setDetails(new BlobListDetails()
+ .setRetrieveMetadata(true));
- for (ListBlobItem blob : container.listBlobs(prefix, true, EnumSet.of(BlobListingDetails.METADATA), null, operationContext)) {
- if (blob instanceof CloudBlob) {
- CloudBlob cloudBlob = (CloudBlob) blob;
- BlobProperties properties = cloudBlob.getProperties();
- StorageUri uri = cloudBlob.getSnapshotQualifiedStorageUri();
+ blobContainerClient.listBlobs().forEach(blob -> {
+ if (blob instanceof BlobItem) {
+ BlobItem blobItem = (BlobItem) blob;
+ BlobItemProperties properties = blobItem.getProperties();
+ BlobClient blobClient = blobContainerClient.getBlobClient(blobItem.getName());
+ String uri = blobClient.getBlobUrl();
Builder builder = new BlobInfo.Builder()
- .primaryUri(uri.getPrimaryUri().toString())
- .blobName(cloudBlob.getName())
- .containerName(containerName)
+ .primaryUri(uri)
+ .blobName(blobItem.getName())
+ .blobType(properties.getBlobType().toString())
Review comment:
Minor backward incompatibility issue: `Block` (orig) vs `BlockBlob` (new)
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java
##########
@@ -176,45 +174,38 @@ protected String getDefaultTimePrecision() {
protected List<BlobInfo> performListing(final ProcessContext context, final Long minTimestamp) throws IOException {
String containerName = context.getProperty(AzureStorageUtils.CONTAINER).evaluateAttributeExpressions().getValue();
String prefix = context.getProperty(PROP_PREFIX).evaluateAttributeExpressions().getValue();
- if (prefix == null) {
- prefix = "";
- }
+
final List<BlobInfo> listing = new ArrayList<>();
try {
- CloudBlobClient blobClient = AzureStorageUtils.createCloudBlobClient(context, getLogger(), null);
- CloudBlobContainer container = blobClient.getContainerReference(containerName);
+ BlobServiceClient blobServiceClient = AzureStorageUtils.createBlobServiceClient(context, null);
+ BlobContainerClient blobContainerClient = blobServiceClient.getBlobContainerClient(containerName);
- final OperationContext operationContext = new OperationContext();
- AzureStorageUtils.setProxy(operationContext, context);
+ final ListBlobsOptions listBlobsOptions = new ListBlobsOptions()
+ .setPrefix(prefix)
+ .setDetails(new BlobListDetails()
+ .setRetrieveMetadata(true));
- for (ListBlobItem blob : container.listBlobs(prefix, true, EnumSet.of(BlobListingDetails.METADATA), null, operationContext)) {
- if (blob instanceof CloudBlob) {
- CloudBlob cloudBlob = (CloudBlob) blob;
- BlobProperties properties = cloudBlob.getProperties();
- StorageUri uri = cloudBlob.getSnapshotQualifiedStorageUri();
+ blobContainerClient.listBlobs().forEach(blob -> {
+ if (blob instanceof BlobItem) {
+ BlobItem blobItem = (BlobItem) blob;
+ BlobItemProperties properties = blobItem.getProperties();
+ BlobClient blobClient = blobContainerClient.getBlobClient(blobItem.getName());
+ String uri = blobClient.getBlobUrl();
Builder builder = new BlobInfo.Builder()
- .primaryUri(uri.getPrimaryUri().toString())
- .blobName(cloudBlob.getName())
- .containerName(containerName)
+ .primaryUri(uri)
Review comment:
The same url encoding issue as in case of the Put processor.
It also affects a `filename` attribute of the outgoing FlowFile which would be calculated as:
`primaryUri.substring(primaryUri.lastIndexOf('/') + 1);`
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -37,17 +28,22 @@
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.proxy.ProxyConfiguration;
-import org.apache.nifi.proxy.ProxySpec;
import org.apache.nifi.services.azure.storage.AzureStorageCredentialsDetails;
import org.apache.nifi.services.azure.storage.AzureStorageCredentialsService;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+
public final class AzureStorageUtils {
+
public static final String BLOCK = "Block";
public static final String PAGE = "Page";
+ public static final String APPEND = "Append";
Review comment:
What is the point in these constants?
The existing ones are not used either (as far as I can see).
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
final AzureStorageCredentialsDetails storageCredentialsDetails = getStorageCredentialsDetails(context, flowFile);
- final CloudStorageAccount cloudStorageAccount = new CloudStorageAccount(
- storageCredentialsDetails.getStorageCredentials(),
- true,
- storageCredentialsDetails.getStorageSuffix(),
- storageCredentialsDetails.getStorageAccountName());
- final CloudBlobClient cloudBlobClient = cloudStorageAccount.createCloudBlobClient();
-
- return cloudBlobClient;
+
+ final String storageSuffix = StringUtils.isNotBlank(storageCredentialsDetails.getStorageSuffix())
+ ? storageCredentialsDetails.getStorageSuffix()
+ : "blob.core.windows.net";
Review comment:
Does it work for Queue Storage as well?
The "official" suffix would be `queue.core.windows.net` for queues.
With my storage account I get back different IPs with nslookup but they might point to the same place behind the scenes...
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/utils/TestAzureStorageUtilsGetStorageCredentialsDetails.java
##########
@@ -147,18 +154,40 @@ private void configureControllerService(String accountName, String accountKey, S
processContext.setProperty(AzureStorageUtils.STORAGE_CREDENTIALS_SERVICE, CREDENTIALS_SERVICE_VALUE);
}
- private void assertStorageCredentialsDetailsAccountNameAndAccountKey(AzureStorageCredentialsDetails storageCredentialsDetails) {
+ private void assertStorageCredentialsDetailsAccountNameAndAccountKey(
+ AzureStorageCredentialsDetails storageCredentialsDetails) {
+
assertEquals(ACCOUNT_NAME_VALUE, storageCredentialsDetails.getStorageAccountName());
- assertTrue(storageCredentialsDetails.getStorageCredentials() instanceof StorageCredentialsAccountAndKey);
- StorageCredentialsAccountAndKey storageCredentials = (StorageCredentialsAccountAndKey) storageCredentialsDetails.getStorageCredentials();
+ assertTrue(storageCredentialsDetails.getCredentialType() == AzureStorageCredentialsDetails.CredentialType.STORAGE_ACCOUNT_KEY);
+ assertTrue(storageCredentialsDetails.getStorageSharedKeyCredential() instanceof StorageSharedKeyCredential);
Review comment:
It is always true, the assert has no real effect.
Also in `assertStorageCredentialsDetailsAccountNameAndSasToken`.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/DeleteAzureBlobStorage.java
##########
@@ -48,20 +48,17 @@
@InputRequirement(Requirement.INPUT_REQUIRED)
public class DeleteAzureBlobStorage extends AbstractAzureBlobProcessor {
- private static final AllowableValue DELETE_SNAPSHOTS_NONE = new AllowableValue(DeleteSnapshotsOption.NONE.name(), "None", "Delete the blob only.");
-
- private static final AllowableValue DELETE_SNAPSHOTS_ALSO = new AllowableValue(DeleteSnapshotsOption.INCLUDE_SNAPSHOTS.name(), "Include Snapshots", "Delete the blob and its snapshots.");
+ private static final AllowableValue DELETE_SNAPSHOTS_ALSO = new AllowableValue(DeleteSnapshotsOptionType.INCLUDE.name(), "Include Snapshots", "Delete the blob and its snapshots.");
- private static final AllowableValue DELETE_SNAPSHOTS_ONLY = new AllowableValue(DeleteSnapshotsOption.DELETE_SNAPSHOTS_ONLY.name(), "Delete Snapshots Only", "Delete only the blob's snapshots.");
+ private static final AllowableValue DELETE_SNAPSHOTS_ONLY = new AllowableValue(DeleteSnapshotsOptionType.ONLY.name(), "Delete Snapshots Only", "Delete only the blob's snapshots.");
Review comment:
This is a backward incompatible change in terms of existing flows.
We need to keep the `NONE` option.
We also need to keep the original names for `AllowableValue`-s and translate them to the new enum values in `onTrigger()`.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java
##########
@@ -176,45 +174,38 @@ protected String getDefaultTimePrecision() {
protected List<BlobInfo> performListing(final ProcessContext context, final Long minTimestamp) throws IOException {
String containerName = context.getProperty(AzureStorageUtils.CONTAINER).evaluateAttributeExpressions().getValue();
String prefix = context.getProperty(PROP_PREFIX).evaluateAttributeExpressions().getValue();
- if (prefix == null) {
- prefix = "";
- }
+
final List<BlobInfo> listing = new ArrayList<>();
try {
- CloudBlobClient blobClient = AzureStorageUtils.createCloudBlobClient(context, getLogger(), null);
- CloudBlobContainer container = blobClient.getContainerReference(containerName);
+ BlobServiceClient blobServiceClient = AzureStorageUtils.createBlobServiceClient(context, null);
+ BlobContainerClient blobContainerClient = blobServiceClient.getBlobContainerClient(containerName);
- final OperationContext operationContext = new OperationContext();
- AzureStorageUtils.setProxy(operationContext, context);
+ final ListBlobsOptions listBlobsOptions = new ListBlobsOptions()
+ .setPrefix(prefix)
+ .setDetails(new BlobListDetails()
+ .setRetrieveMetadata(true));
- for (ListBlobItem blob : container.listBlobs(prefix, true, EnumSet.of(BlobListingDetails.METADATA), null, operationContext)) {
- if (blob instanceof CloudBlob) {
- CloudBlob cloudBlob = (CloudBlob) blob;
- BlobProperties properties = cloudBlob.getProperties();
- StorageUri uri = cloudBlob.getSnapshotQualifiedStorageUri();
+ blobContainerClient.listBlobs().forEach(blob -> {
+ if (blob instanceof BlobItem) {
+ BlobItem blobItem = (BlobItem) blob;
+ BlobItemProperties properties = blobItem.getProperties();
+ BlobClient blobClient = blobContainerClient.getBlobClient(blobItem.getName());
+ String uri = blobClient.getBlobUrl();
Builder builder = new BlobInfo.Builder()
- .primaryUri(uri.getPrimaryUri().toString())
- .blobName(cloudBlob.getName())
- .containerName(containerName)
+ .primaryUri(uri)
+ .blobName(blobItem.getName())
+ .blobType(properties.getBlobType().toString())
+ .containerName(blobClient.getContainerName())
.contentType(properties.getContentType())
.contentLanguage(properties.getContentLanguage())
- .etag(properties.getEtag())
- .lastModifiedTime(properties.getLastModified().getTime())
- .length(properties.getLength());
-
- if (uri.getSecondaryUri() != null) {
- builder.secondaryUri(uri.getSecondaryUri().toString());
- }
-
- if (blob instanceof CloudBlockBlob) {
- builder.blobType(AzureStorageUtils.BLOCK);
- } else {
- builder.blobType(AzureStorageUtils.PAGE);
- }
+ .etag(properties.getETag())
+ .lastModifiedTime(properties.getLastModified().toEpochSecond())
+ .length(properties.getContentLength());
+
Review comment:
Minor backward incompatibility: secondaryUri is missing.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
##########
@@ -114,14 +112,14 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
}
try {
- blob.upload(in, length, null, null, operationContext);
+ blob.upload(in, length);
BlobProperties properties = blob.getProperties();
attributes.put("azure.container", containerName);
- attributes.put("azure.primaryUri", blob.getSnapshotQualifiedUri().toString());
- attributes.put("azure.etag", properties.getEtag());
+ attributes.put("azure.primaryUri", blob.getBlobUrl());
Review comment:
`BlobClient.getBlobUrl()` returns the slashes (/) in the blob name in url encoded form.
Eg.: `https://mystorageaccount.blob.core.windows.net/my-blob-container/dir1%2Ffile1`
It was earlier (and I think it still should be): `https://mystorageaccount.blob.core.windows.net/my-blob-container/dir1/file1`
It seems to me a bug in the new client lib.
It affects the List processor too.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/utils/TestAzureStorageUtilsGetStorageCredentialsDetails.java
##########
@@ -147,18 +154,40 @@ private void configureControllerService(String accountName, String accountKey, S
processContext.setProperty(AzureStorageUtils.STORAGE_CREDENTIALS_SERVICE, CREDENTIALS_SERVICE_VALUE);
}
- private void assertStorageCredentialsDetailsAccountNameAndAccountKey(AzureStorageCredentialsDetails storageCredentialsDetails) {
+ private void assertStorageCredentialsDetailsAccountNameAndAccountKey(
+ AzureStorageCredentialsDetails storageCredentialsDetails) {
+
assertEquals(ACCOUNT_NAME_VALUE, storageCredentialsDetails.getStorageAccountName());
- assertTrue(storageCredentialsDetails.getStorageCredentials() instanceof StorageCredentialsAccountAndKey);
- StorageCredentialsAccountAndKey storageCredentials = (StorageCredentialsAccountAndKey) storageCredentialsDetails.getStorageCredentials();
+ assertTrue(storageCredentialsDetails.getCredentialType() == AzureStorageCredentialsDetails.CredentialType.STORAGE_ACCOUNT_KEY);
Review comment:
`assertSame` should be used
Also in `assertStorageCredentialsDetailsAccountNameAndSasToken`
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/utils/TestAzureProxyUtilsGetProxyOptions.java
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.nifi.processors.azure.storage.utils;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+
+import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.ProxyOptions.Type;
+
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processors.azure.storage.ListAzureBlobStorage;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxyConfigurationService;
+import org.apache.nifi.util.MockProcessContext;
+import org.apache.nifi.util.StringUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestAzureProxyUtilsGetProxyOptions {
+
+ private MockProcessContext processContext;
+ private ProxyConfiguration proxyConfig;
+
+ private static final String PROXY_CONFIG_SERVICE_VALUE = "ProxyConfigurationService";
+ private static final String PROXY_HOST = "localhost";
+ private static final Integer PROXY_PORT = 9000;
+ private static final String PROXY_USER = "microsoft";
+ private static final String PROXY_PASSWORD = "azure";
+
+ @Before
+ public void setUp() {
+ final Processor processor = new ListAzureBlobStorage();
+ processContext = new MockProcessContext(processor);
+ }
+
+ private void configureMockedHTTPProxyService(String proxyHost, Integer proxyPort, String proxyUser, String proxyUserPassword) {
+ proxyConfig = new ProxyConfiguration();
+ proxyConfig.setProxyType(Proxy.Type.HTTP);
+
+ if(StringUtils.isNotBlank(proxyHost)) {
+ proxyConfig.setProxyServerHost(proxyHost);
+ }
+ if(proxyPort != null) {
+ proxyConfig.setProxyServerPort(proxyPort);
+ }
+ if(StringUtils.isNotBlank(proxyUser)) {
+ proxyConfig.setProxyUserName(proxyUser);
+ }
+ if(StringUtils.isNotBlank(proxyUserPassword)) {
+ proxyConfig.setProxyUserPassword(proxyUserPassword);
+ }
+
+ MockProxyConfigurationService mockProxyConfigurationService = new MockProxyConfigurationService(proxyConfig);
+ // set mocked proxy service
+ processContext.addControllerService(mockProxyConfigurationService, PROXY_CONFIG_SERVICE_VALUE);
+ processContext.setProperty(AzureProxyUtils.PROXY_CONFIGURATION_SERVICE, PROXY_CONFIG_SERVICE_VALUE);
+ }
+
+ @Test
+ public void testHTTPProxy() {
+ configureMockedHTTPProxyService(PROXY_HOST, PROXY_PORT, null, null);
+ final MockProxyConfigurationService mockProxyConfigurationService = processContext.getProperty(
+ AzureProxyUtils.PROXY_CONFIGURATION_SERVICE).asControllerService(MockProxyConfigurationService.class);
+
+ final ProxyOptions proxyOptions = AzureProxyUtils.getProxyOptions(mockProxyConfigurationService.getConfiguration());
+ final InetSocketAddress socketAddress = new InetSocketAddress(PROXY_HOST, PROXY_PORT);
+
+ assertEquals(proxyOptions.getAddress(), socketAddress);
+ assertEquals(proxyOptions.getType(), Type.HTTP);
Review comment:
The expected value should be the 1st argument.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/utils/TestAzureStorageUtilsGetStorageCredentialsDetails.java
##########
@@ -54,25 +57,27 @@ public void setUp() {
public void testAccountNameAndAccountKeyConfiguredOnProcessor() {
configureProcessorProperties(ACCOUNT_NAME_VALUE, ACCOUNT_KEY_VALUE, null);
- AzureStorageCredentialsDetails storageCredentialsDetails = AzureStorageUtils.getStorageCredentialsDetails(processContext, null);
+ AzureStorageCredentialsDetails storageCredentialsDetails = AzureStorageUtils
+ .getStorageCredentialsDetails(processContext, null);
assertStorageCredentialsDetailsAccountNameAndAccountKey(storageCredentialsDetails);
}
@Test
public void testAccountNameAndSasTokenConfiguredOnProcessor() {
configureProcessorProperties(ACCOUNT_NAME_VALUE, null, SAS_TOKEN_VALUE);
-
- AzureStorageCredentialsDetails storageCredentialsDetails = AzureStorageUtils.getStorageCredentialsDetails(processContext, null);
+ AzureStorageCredentialsDetails storageCredentialsDetails = AzureStorageUtils
+ .getStorageCredentialsDetails(processContext, null);
assertStorageCredentialsDetailsAccountNameAndSasToken(storageCredentialsDetails);
}
@Test
public void testAccountNameAndAccountKeyConfiguredOnControllerService() {
- configureControllerService(ACCOUNT_NAME_VALUE, ACCOUNT_KEY_VALUE, null);
- AzureStorageCredentialsDetails storageCredentialsDetails = AzureStorageUtils.getStorageCredentialsDetails(processContext, null);
+ configureControllerService(ACCOUNT_NAME_VALUE, ACCOUNT_KEY_VALUE, null);
+ AzureStorageCredentialsDetails storageCredentialsDetails = AzureStorageUtils
Review comment:
Please do not reformat the existing code.
The max. line length is 200 in checkstyle config.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/utils/TestAzureProxyUtilsGetProxyOptions.java
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.nifi.processors.azure.storage.utils;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+
+import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.ProxyOptions.Type;
+
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processors.azure.storage.ListAzureBlobStorage;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxyConfigurationService;
+import org.apache.nifi.util.MockProcessContext;
+import org.apache.nifi.util.StringUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestAzureProxyUtilsGetProxyOptions {
+
+ private MockProcessContext processContext;
+ private ProxyConfiguration proxyConfig;
+
+ private static final String PROXY_CONFIG_SERVICE_VALUE = "ProxyConfigurationService";
+ private static final String PROXY_HOST = "localhost";
+ private static final Integer PROXY_PORT = 9000;
+ private static final String PROXY_USER = "microsoft";
+ private static final String PROXY_PASSWORD = "azure";
+
+ @Before
+ public void setUp() {
+ final Processor processor = new ListAzureBlobStorage();
+ processContext = new MockProcessContext(processor);
+ }
+
+ private void configureMockedHTTPProxyService(String proxyHost, Integer proxyPort, String proxyUser, String proxyUserPassword) {
+ proxyConfig = new ProxyConfiguration();
+ proxyConfig.setProxyType(Proxy.Type.HTTP);
+
+ if(StringUtils.isNotBlank(proxyHost)) {
+ proxyConfig.setProxyServerHost(proxyHost);
+ }
+ if(proxyPort != null) {
+ proxyConfig.setProxyServerPort(proxyPort);
+ }
+ if(StringUtils.isNotBlank(proxyUser)) {
+ proxyConfig.setProxyUserName(proxyUser);
+ }
+ if(StringUtils.isNotBlank(proxyUserPassword)) {
+ proxyConfig.setProxyUserPassword(proxyUserPassword);
+ }
+
+ MockProxyConfigurationService mockProxyConfigurationService = new MockProxyConfigurationService(proxyConfig);
+ // set mocked proxy service
+ processContext.addControllerService(mockProxyConfigurationService, PROXY_CONFIG_SERVICE_VALUE);
+ processContext.setProperty(AzureProxyUtils.PROXY_CONFIGURATION_SERVICE, PROXY_CONFIG_SERVICE_VALUE);
+ }
+
+ @Test
+ public void testHTTPProxy() {
+ configureMockedHTTPProxyService(PROXY_HOST, PROXY_PORT, null, null);
+ final MockProxyConfigurationService mockProxyConfigurationService = processContext.getProperty(
+ AzureProxyUtils.PROXY_CONFIGURATION_SERVICE).asControllerService(MockProxyConfigurationService.class);
+
+ final ProxyOptions proxyOptions = AzureProxyUtils.getProxyOptions(mockProxyConfigurationService.getConfiguration());
+ final InetSocketAddress socketAddress = new InetSocketAddress(PROXY_HOST, PROXY_PORT);
+
+ assertEquals(proxyOptions.getAddress(), socketAddress);
+ assertEquals(proxyOptions.getType(), Type.HTTP);
+ // null asserts
+ assertNull(proxyOptions.getUsername());
+ assertNull(proxyOptions.getPassword());
+ }
+
+ @Test
+ public void testHTTPProxyWithAuth() {
+ configureMockedHTTPProxyService(PROXY_HOST, PROXY_PORT, PROXY_USER, PROXY_PASSWORD);
+ final MockProxyConfigurationService mockProxyConfigurationService = processContext.getProperty(
+ AzureProxyUtils.PROXY_CONFIGURATION_SERVICE).asControllerService(MockProxyConfigurationService.class);
+
+ final ProxyOptions proxyOptions = AzureProxyUtils.getProxyOptions(mockProxyConfigurationService.getConfiguration());
+ final InetSocketAddress socketAddress = new InetSocketAddress(PROXY_HOST, PROXY_PORT);
+
+ assertEquals(proxyOptions.getAddress(), socketAddress);
+ assertEquals(proxyOptions.getType(), Type.HTTP);
+ assertEquals(proxyOptions.getUsername(), PROXY_USER);
+ assertEquals(proxyOptions.getPassword(), PROXY_PASSWORD);
+ }
+
+ @Test
+ public void testHTTPProxyWithOnlyUser() {
+ configureMockedHTTPProxyService(PROXY_HOST, PROXY_PORT, PROXY_USER, null);
+ final MockProxyConfigurationService mockProxyConfigurationService = processContext.getProperty(
+ AzureProxyUtils.PROXY_CONFIGURATION_SERVICE).asControllerService(MockProxyConfigurationService.class);
+
+ final ProxyOptions proxyOptions = AzureProxyUtils.getProxyOptions(mockProxyConfigurationService.getConfiguration());
+ final InetSocketAddress socketAddress = new InetSocketAddress(PROXY_HOST, PROXY_PORT);
+
+ assertEquals(proxyOptions.getAddress(), socketAddress);
+ assertEquals(proxyOptions.getType(), Type.HTTP);
+ // null asserts
+ assertNull(proxyOptions.getUsername());
+ assertNull(proxyOptions.getPassword());
+ }
+
+ @Test
+ public void testHTTPProxyWithOnlyProxyPort() {
+ configureMockedHTTPProxyService(null, PROXY_PORT, null, null);
+ final MockProxyConfigurationService mockProxyConfigurationService = processContext.getProperty(
+ AzureProxyUtils.PROXY_CONFIGURATION_SERVICE).asControllerService(MockProxyConfigurationService.class);
+
+ final ProxyOptions proxyOptions = AzureProxyUtils.getProxyOptions(mockProxyConfigurationService.getConfiguration());
+
+ // null asserts
+ assertNull(proxyOptions);
+ }
+
+ @Test
+ public void testHTTPProxyWithoutInput() {
+ configureMockedHTTPProxyService(null, null, null, null);
+ final MockProxyConfigurationService mockProxyConfigurationService = processContext.getProperty(
+ AzureProxyUtils.PROXY_CONFIGURATION_SERVICE).asControllerService(MockProxyConfigurationService.class);
+
+ final ProxyOptions proxyOptions = AzureProxyUtils.getProxyOptions(mockProxyConfigurationService.getConfiguration());
+
+ // null asserts
+ assertNull(proxyOptions);
+ }
+
+ private class MockProxyConfigurationService extends AbstractControllerService implements ProxyConfigurationService {
Review comment:
The nested class could be static.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/DeleteAzureBlobStorage.java
##########
@@ -82,21 +79,23 @@ public void onTrigger(ProcessContext context, ProcessSession session) throws Pro
final long startNanos = System.nanoTime();
final String containerName = context.getProperty(AzureStorageUtils.CONTAINER).evaluateAttributeExpressions(flowFile).getValue();
final String blobPath = context.getProperty(BLOB).evaluateAttributeExpressions(flowFile).getValue();
- final String deleteSnapshotOptions = context.getProperty(DELETE_SNAPSHOTS_OPTION).getValue();
+ final String deleteSnapshotOption = context.getProperty(DELETE_SNAPSHOTS_OPTION).isSet()
Review comment:
It seems `deleteSnapshotsOption` would be the right name.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureProxyUtils.java
##########
@@ -0,0 +1,115 @@
+
+/*
+ * 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.nifi.processors.azure.storage.utils;
+
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.util.Collection;
+
+import com.azure.core.http.HttpClient;
+import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.netty.NettyAsyncHttpClientBuilder;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxySpec;
+import org.apache.nifi.util.StringUtils;
+
+public class AzureProxyUtils {
+ private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP, ProxySpec.SOCKS};
+
+ private static ProxyOptions.Type getProxyOptionsTypeFromProxyType(final Proxy.Type proxyType) {
+ for (final ProxyOptions.Type item : ProxyOptions.Type.values()) {
+ if (item.toProxyType() == proxyType) {
+ return item;
+ }
+ }
+ return null;
+ }
+
+ public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE = ProxyConfiguration
+ .createProxyConfigPropertyDescriptor(false, PROXY_SPECS);
+
+ public static HttpClient createHttpClient(final PropertyContext propertyContext) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(propertyContext);
+ final ProxyOptions proxyOptions = getProxyOptions(proxyConfig);
+
+ final HttpClient client = new NettyAsyncHttpClientBuilder()
+ .proxy(proxyOptions)
+ .build();
+
+ return client;
+ }
+
+ public static void validateProxySpec(final ValidationContext context, final Collection<ValidationResult> results) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context);
+
+ final String proxyServerHost = proxyConfig.getProxyServerHost();
+ final Integer proxyServerPort = proxyConfig.getProxyServerPort();
+ final String proxyServerUser = proxyConfig.getProxyUserName();
+ final String proxyServerPassword = proxyConfig.getProxyUserPassword();
+
+ if ((StringUtils.isNotBlank(proxyServerHost) && proxyServerPort == null)
+ || (StringUtils.isBlank(proxyServerHost) && proxyServerPort != null)) {
+ results.add(new ValidationResult.Builder().subject("AzureProxyUtils Details").valid(false)
Review comment:
"Proxy Configuration" would be a more understandable subject for the end user I think.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureProxyUtils.java
##########
@@ -0,0 +1,115 @@
+
+/*
+ * 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.nifi.processors.azure.storage.utils;
+
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.util.Collection;
+
+import com.azure.core.http.HttpClient;
+import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.netty.NettyAsyncHttpClientBuilder;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxySpec;
+import org.apache.nifi.util.StringUtils;
+
+public class AzureProxyUtils {
+ private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP, ProxySpec.SOCKS};
+
+ private static ProxyOptions.Type getProxyOptionsTypeFromProxyType(final Proxy.Type proxyType) {
+ for (final ProxyOptions.Type item : ProxyOptions.Type.values()) {
+ if (item.toProxyType() == proxyType) {
+ return item;
+ }
+ }
+ return null;
+ }
+
+ public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE = ProxyConfiguration
+ .createProxyConfigPropertyDescriptor(false, PROXY_SPECS);
+
+ public static HttpClient createHttpClient(final PropertyContext propertyContext) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(propertyContext);
+ final ProxyOptions proxyOptions = getProxyOptions(proxyConfig);
+
+ final HttpClient client = new NettyAsyncHttpClientBuilder()
+ .proxy(proxyOptions)
+ .build();
+
+ return client;
+ }
+
+ public static void validateProxySpec(final ValidationContext context, final Collection<ValidationResult> results) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context);
+
+ final String proxyServerHost = proxyConfig.getProxyServerHost();
+ final Integer proxyServerPort = proxyConfig.getProxyServerPort();
+ final String proxyServerUser = proxyConfig.getProxyUserName();
+ final String proxyServerPassword = proxyConfig.getProxyUserPassword();
+
+ if ((StringUtils.isNotBlank(proxyServerHost) && proxyServerPort == null)
Review comment:
These checks are relevant in case of `HTTP` and `SOCKS` proxy but not for `DIRECT`.
In case of `DIRECT` (that is no proxy), it could be checked that no other properties are specified.
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureProxyUtils.java
##########
@@ -0,0 +1,115 @@
+
+/*
+ * 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.nifi.processors.azure.storage.utils;
+
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.util.Collection;
+
+import com.azure.core.http.HttpClient;
+import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.netty.NettyAsyncHttpClientBuilder;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxySpec;
+import org.apache.nifi.util.StringUtils;
+
+public class AzureProxyUtils {
+ private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP, ProxySpec.SOCKS};
+
+ private static ProxyOptions.Type getProxyOptionsTypeFromProxyType(final Proxy.Type proxyType) {
+ for (final ProxyOptions.Type item : ProxyOptions.Type.values()) {
+ if (item.toProxyType() == proxyType) {
+ return item;
+ }
+ }
+ return null;
+ }
+
+ public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE = ProxyConfiguration
+ .createProxyConfigPropertyDescriptor(false, PROXY_SPECS);
+
+ public static HttpClient createHttpClient(final PropertyContext propertyContext) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(propertyContext);
+ final ProxyOptions proxyOptions = getProxyOptions(proxyConfig);
+
+ final HttpClient client = new NettyAsyncHttpClientBuilder()
+ .proxy(proxyOptions)
+ .build();
+
+ return client;
+ }
+
+ public static void validateProxySpec(final ValidationContext context, final Collection<ValidationResult> results) {
+ final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context);
+
+ final String proxyServerHost = proxyConfig.getProxyServerHost();
+ final Integer proxyServerPort = proxyConfig.getProxyServerPort();
+ final String proxyServerUser = proxyConfig.getProxyUserName();
+ final String proxyServerPassword = proxyConfig.getProxyUserPassword();
+
+ if ((StringUtils.isNotBlank(proxyServerHost) && proxyServerPort == null)
+ || (StringUtils.isBlank(proxyServerHost) && proxyServerPort != null)) {
+ results.add(new ValidationResult.Builder().subject("AzureProxyUtils Details").valid(false)
+ .explanation(
+ "When specifying address information, both `host` and `port` information must be provided.")
+ .build());
+ }
+
+ if ((StringUtils.isBlank(proxyServerUser) && StringUtils.isNotBlank(proxyServerPassword))
+ || (StringUtils.isNotBlank(proxyServerUser) && StringUtils.isBlank(proxyServerPassword))) {
+ results.add(new ValidationResult.Builder().subject("AzureProxyUtils Details").valid(false)
+ .explanation(
+ "When specifying credentials, both `user` and `password` must be provided.")
+ .build());
+ }
+
+ ProxyConfiguration.validateProxySpec(context, results, PROXY_SPECS);
+ }
+
+ public static ProxyOptions getProxyOptions(final ProxyConfiguration proxyConfig) {
+ final String proxyServerHost = proxyConfig.getProxyServerHost();
+ final Integer proxyServerPort = proxyConfig.getProxyServerPort();
+ final String proxyServerUser = proxyConfig.getProxyUserName();
+ final String proxyServerPassword = proxyConfig.getProxyUserPassword();
+
+ final Boolean proxyServerProvided = StringUtils.isNotBlank(proxyServerHost) && proxyServerPort != null;
+ final Boolean proxyCredentialsProvided = StringUtils.isNotBlank(proxyServerUser) && StringUtils.isNotBlank(proxyServerPassword);
+
+ // if no endpoint is provided, return zero
+ if (!proxyServerProvided) {
+ return null;
+ }
+
+ // translate Proxy.Type to ProxyOptions.Type
+ final ProxyOptions.Type proxyType = getProxyOptionsTypeFromProxyType(proxyConfig.getProxyType());
+ final InetSocketAddress socketAddress = new InetSocketAddress(proxyServerHost, proxyServerPort);
+
+ final ProxyOptions proxyOptions = new ProxyOptions(proxyType, socketAddress);
+
+ if (proxyCredentialsProvided) {
+ return proxyOptions.setCredentials(proxyServerUser, proxyServerPassword);
+ } else {
+ return proxyOptions;
+ }
Review comment:
```
if (proxyCredentialsProvided) {
proxyOptions.setCredentials(proxyServerUser, proxyServerPassword);
}
return proxyOptions;
```
would be more readable.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] MuazmaZ commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
MuazmaZ commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463176963
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
##########
@@ -114,14 +112,14 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
}
try {
- blob.upload(in, length, null, null, operationContext);
+ blob.upload(in, length);
BlobProperties properties = blob.getProperties();
attributes.put("azure.container", containerName);
- attributes.put("azure.primaryUri", blob.getSnapshotQualifiedUri().toString());
- attributes.put("azure.etag", properties.getEtag());
+ attributes.put("azure.primaryUri", blob.getBlobUrl());
Review comment:
@turcsanyip tracking this, I will report with findings sometime today.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r475587981
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java
##########
@@ -176,45 +174,38 @@ protected String getDefaultTimePrecision() {
protected List<BlobInfo> performListing(final ProcessContext context, final Long minTimestamp) throws IOException {
String containerName = context.getProperty(AzureStorageUtils.CONTAINER).evaluateAttributeExpressions().getValue();
String prefix = context.getProperty(PROP_PREFIX).evaluateAttributeExpressions().getValue();
- if (prefix == null) {
- prefix = "";
- }
+
final List<BlobInfo> listing = new ArrayList<>();
try {
- CloudBlobClient blobClient = AzureStorageUtils.createCloudBlobClient(context, getLogger(), null);
- CloudBlobContainer container = blobClient.getContainerReference(containerName);
+ BlobServiceClient blobServiceClient = AzureStorageUtils.createBlobServiceClient(context, null);
+ BlobContainerClient blobContainerClient = blobServiceClient.getBlobContainerClient(containerName);
- final OperationContext operationContext = new OperationContext();
- AzureStorageUtils.setProxy(operationContext, context);
+ final ListBlobsOptions listBlobsOptions = new ListBlobsOptions()
+ .setPrefix(prefix)
+ .setDetails(new BlobListDetails()
+ .setRetrieveMetadata(true));
- for (ListBlobItem blob : container.listBlobs(prefix, true, EnumSet.of(BlobListingDetails.METADATA), null, operationContext)) {
- if (blob instanceof CloudBlob) {
- CloudBlob cloudBlob = (CloudBlob) blob;
- BlobProperties properties = cloudBlob.getProperties();
- StorageUri uri = cloudBlob.getSnapshotQualifiedStorageUri();
+ blobContainerClient.listBlobs().forEach(blob -> {
+ if (blob instanceof BlobItem) {
+ BlobItem blobItem = (BlobItem) blob;
+ BlobItemProperties properties = blobItem.getProperties();
+ BlobClient blobClient = blobContainerClient.getBlobClient(blobItem.getName());
+ String uri = blobClient.getBlobUrl();
Builder builder = new BlobInfo.Builder()
- .primaryUri(uri.getPrimaryUri().toString())
- .blobName(cloudBlob.getName())
- .containerName(containerName)
+ .primaryUri(uri)
Review comment:
As mentioned in the conversation on `attributes.put("azure.primaryUri", blob.getBlobUrl());` - I'm unable to reproduce this error and filename is deployed correctly when running the pipeline.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463237030
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/DeleteAzureBlobStorage.java
##########
@@ -48,20 +48,17 @@
@InputRequirement(Requirement.INPUT_REQUIRED)
public class DeleteAzureBlobStorage extends AbstractAzureBlobProcessor {
- private static final AllowableValue DELETE_SNAPSHOTS_NONE = new AllowableValue(DeleteSnapshotsOption.NONE.name(), "None", "Delete the blob only.");
-
- private static final AllowableValue DELETE_SNAPSHOTS_ALSO = new AllowableValue(DeleteSnapshotsOption.INCLUDE_SNAPSHOTS.name(), "Include Snapshots", "Delete the blob and its snapshots.");
+ private static final AllowableValue DELETE_SNAPSHOTS_ALSO = new AllowableValue(DeleteSnapshotsOptionType.INCLUDE.name(), "Include Snapshots", "Delete the blob and its snapshots.");
- private static final AllowableValue DELETE_SNAPSHOTS_ONLY = new AllowableValue(DeleteSnapshotsOption.DELETE_SNAPSHOTS_ONLY.name(), "Delete Snapshots Only", "Delete only the blob's snapshots.");
+ private static final AllowableValue DELETE_SNAPSHOTS_ONLY = new AllowableValue(DeleteSnapshotsOptionType.ONLY.name(), "Delete Snapshots Only", "Delete only the blob's snapshots.");
Review comment:
@turcsanyip Thank you for this remark! Maybe also due to my inexperience with the NiFi codebase, but my go to solution would be to implement a solution such as this:
```java
// translation enum for backwards compatability from DeleteSnapshotsOption (v8 api) -> DeleteSnapshotsOptionType (v12 api)
public enum DeleteSnapshotsOption {
// don't delete snapshots
NONE(null),
// include snapshots on blob deletion
INCLUDE_SNAPSHOTS(DeleteSnapshotsOptionType.INCLUDE),
// only delete snapshots of blob and don't delete blob itself
DELETE_SNAPSHOTS_ONLY(DeleteSnapshotsOptionType.ONLY);
private final DeleteSnapshotsOptionType deleteSnapshotsOptionType;
DeleteSnapshotsOption(DeleteSnapshotsOptionType type) {
this.deleteSnapshotsOptionType = type;
}
public DeleteSnapshotsOptionType getValue() {
return this.deleteSnapshotsOptionType;
}
}
```
would this suffice to achieve backwards compatability?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] github-actions[bot] commented on pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#issuecomment-826170727
We're marking this PR as stale due to lack of updates in the past few months. If after another couple of weeks the stale label has not been removed this PR will be closed. This stale marker and eventual auto close does not indicate a judgement of the PR just lack of reviewer bandwidth and helps us keep the PR queue more manageable. If you would like this PR re-opened you can do so and a committer can remove the stale tag. Or you can open a new PR. Try to help review other PRs to increase PR review bandwidth which in turn helps yours.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] jfrazee commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
jfrazee commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463196544
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
final AzureStorageCredentialsDetails storageCredentialsDetails = getStorageCredentialsDetails(context, flowFile);
- final CloudStorageAccount cloudStorageAccount = new CloudStorageAccount(
- storageCredentialsDetails.getStorageCredentials(),
- true,
- storageCredentialsDetails.getStorageSuffix(),
- storageCredentialsDetails.getStorageAccountName());
- final CloudBlobClient cloudBlobClient = cloudStorageAccount.createCloudBlobClient();
-
- return cloudBlobClient;
+
+ final String storageSuffix = StringUtils.isNotBlank(storageCredentialsDetails.getStorageSuffix())
+ ? storageCredentialsDetails.getStorageSuffix()
+ : "blob.core.windows.net";
Review comment:
I think the API does the right thing if this is null. So defaulting it might be unnecessary anyway.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] github-actions[bot] commented on pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#issuecomment-826170727
We're marking this PR as stale due to lack of updates in the past few months. If after another couple of weeks the stale label has not been removed this PR will be closed. This stale marker and eventual auto close does not indicate a judgement of the PR just lack of reviewer bandwidth and helps us keep the PR queue more manageable. If you would like this PR re-opened you can do so and a committer can remove the stale tag. Or you can open a new PR. Try to help review other PRs to increase PR review bandwidth which in turn helps yours.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] github-actions[bot] closed pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #4430:
URL: https://github.com/apache/nifi/pull/4430
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r463238291
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
##########
@@ -146,24 +142,46 @@ private AzureStorageUtils() {
}
/**
- * Create CloudBlobClient instance.
+ * Create BlobServiceClient instance.
* @param flowFile An incoming FlowFile can be used for NiFi Expression Language evaluation to derive
* Account Name, Account Key or SAS Token. This can be null if not available.
*/
- public static CloudBlobClient createCloudBlobClient(ProcessContext context, ComponentLog logger, FlowFile flowFile) throws URISyntaxException {
+ public static BlobServiceClient createBlobServiceClient(PropertyContext context, FlowFile flowFile) {
Review comment:
@jfrazee I don't think we can achieve this type of downstream compatability without having two versions of the azure-storage processors available, one for the legacy sdk and one for the v12 sdk.
What would your suggestion be?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r475587319
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java
##########
@@ -176,45 +174,38 @@ protected String getDefaultTimePrecision() {
protected List<BlobInfo> performListing(final ProcessContext context, final Long minTimestamp) throws IOException {
String containerName = context.getProperty(AzureStorageUtils.CONTAINER).evaluateAttributeExpressions().getValue();
String prefix = context.getProperty(PROP_PREFIX).evaluateAttributeExpressions().getValue();
- if (prefix == null) {
- prefix = "";
- }
+
final List<BlobInfo> listing = new ArrayList<>();
try {
- CloudBlobClient blobClient = AzureStorageUtils.createCloudBlobClient(context, getLogger(), null);
- CloudBlobContainer container = blobClient.getContainerReference(containerName);
+ BlobServiceClient blobServiceClient = AzureStorageUtils.createBlobServiceClient(context, null);
+ BlobContainerClient blobContainerClient = blobServiceClient.getBlobContainerClient(containerName);
- final OperationContext operationContext = new OperationContext();
- AzureStorageUtils.setProxy(operationContext, context);
+ final ListBlobsOptions listBlobsOptions = new ListBlobsOptions()
+ .setPrefix(prefix)
+ .setDetails(new BlobListDetails()
+ .setRetrieveMetadata(true));
- for (ListBlobItem blob : container.listBlobs(prefix, true, EnumSet.of(BlobListingDetails.METADATA), null, operationContext)) {
- if (blob instanceof CloudBlob) {
- CloudBlob cloudBlob = (CloudBlob) blob;
- BlobProperties properties = cloudBlob.getProperties();
- StorageUri uri = cloudBlob.getSnapshotQualifiedStorageUri();
+ blobContainerClient.listBlobs().forEach(blob -> {
+ if (blob instanceof BlobItem) {
+ BlobItem blobItem = (BlobItem) blob;
+ BlobItemProperties properties = blobItem.getProperties();
+ BlobClient blobClient = blobContainerClient.getBlobClient(blobItem.getName());
+ String uri = blobClient.getBlobUrl();
Builder builder = new BlobInfo.Builder()
- .primaryUri(uri.getPrimaryUri().toString())
- .blobName(cloudBlob.getName())
- .containerName(containerName)
+ .primaryUri(uri)
+ .blobName(blobItem.getName())
+ .blobType(properties.getBlobType().toString())
+ .containerName(blobClient.getContainerName())
.contentType(properties.getContentType())
.contentLanguage(properties.getContentLanguage())
- .etag(properties.getEtag())
- .lastModifiedTime(properties.getLastModified().getTime())
- .length(properties.getLength());
-
- if (uri.getSecondaryUri() != null) {
- builder.secondaryUri(uri.getSecondaryUri().toString());
- }
-
- if (blob instanceof CloudBlockBlob) {
- builder.blobType(AzureStorageUtils.BLOCK);
- } else {
- builder.blobType(AzureStorageUtils.PAGE);
- }
+ .etag(properties.getETag())
+ .lastModifiedTime(properties.getLastModified().toEpochSecond())
+ .length(properties.getContentLength());
+
Review comment:
Secondary URI is no longer receivable from v12 onwards. Would you suggest we keep it in here as a backwards compatible output (but it will be empty anyways...) or just remove it from BlobInfo 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] stijn192 commented on a change in pull request #4430: NIFI-7677 - Upgrade Azure Storage API to latest (v12) for enabling identity authentication
Posted by GitBox <gi...@apache.org>.
stijn192 commented on a change in pull request #4430:
URL: https://github.com/apache/nifi/pull/4430#discussion_r475399169
##########
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
##########
@@ -114,14 +112,14 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
}
try {
- blob.upload(in, length, null, null, operationContext);
+ blob.upload(in, length);
BlobProperties properties = blob.getProperties();
attributes.put("azure.container", containerName);
- attributes.put("azure.primaryUri", blob.getSnapshotQualifiedUri().toString());
- attributes.put("azure.etag", properties.getEtag());
+ attributes.put("azure.primaryUri", blob.getBlobUrl());
Review comment:
Hi @turcsanyip , I've tried running this through debug mode directly from the azuresdk (with .getBlobUrl() on the client object) as well as running it manually through NiFi and the URL's and filenames do resolve correctly. Under which circumstances were you running into this issue?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org