NIFI-4199: Added ProxyConfigurationService

- Added ProxyConfigurationService to manage centralized proxy
configurations
- Adopt ProxyConfigurationService at FTP and HTTP processors

NIFI-4175 - Add HTTP proxy support to *SFTP processors

This closes #2018.

Signed-off-by: Koji Kawamura <ijokarumawak@apache.org>

NIFI-4199: Add ProxyConfigurationService to SFTP processors

- Fixed check style issue
- Use the same proxy related PropertyDescriptors from FTPTransfer and
SFTPTransfer
- Dropped FlowFile EL evaluation support to make it align with other
processors spec, Now it supports VARIABLE_REGISTRY
- Added ProxyConfigurationService to SFTP processors
- Added SOCKS proxy support to SFTP processors

NIFI-4199: Added ProxyConfigurationService to ElasticsearchHttp processors

- ElasticsearchHttp processors now support SOCKS proxy, too
- Added proxy support to PutElasticsearchHttpRecord
- Moved more common property descriptors to
AbstractElasticsearchHttpProcessor and just return static unmodifiable
property descriptor list at each implementation processors

NIFI-4196 - Expose AWS proxy authentication settings

NIFI-4196 - Fix jUnit errors

This closes #2016.

Signed-off-by: Koji Kawamura <ijokarumawak@apache.org>

NIFI-4199: Add ProxyConfigService to AWS processors

- Applied ProxyConfigService to S3 processors
- Added proxy support to following processors:
  - PutKinesisFirehose, PutKinesisStream
  - PutDynamoDB, DeleteDynamoDB, GetDynamoDB
  - PutKinesisStream
- All AWS processors support HTTP proxy now

NIFI-4199: Add ProxyConfigService to Azure processors

NIFI-4199: More explicit validation and docs for Proxy spec

- Each processor has different supporting Proxy specs
- Show supported spec to ProxyConfigurationService property doc
- Validate not only Proxy type, but also with Authentication

NIFI-4199: Incorporated review comments

- Fixed TestListS3 property descriptor check
- Separate name and displayName

This closes #2016
This closes #2018
This closes #2704

Signed-off-by: Mike Thomsen <mikerthomsen@gmail.com>

Author: ijokarumawak

nifi-assembly/pom.xml

@@ -703,6 +703,12 @@ language governing permissions and limitations under the License. -->
             <version>1.7.0-SNAPSHOT</version>
             <type>nar</type>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-proxy-configuration-nar</artifactId>
+            <version>1.7.0-SNAPSHOT</version>
+            <type>nar</type>
+        </dependency>
     </dependencies>
     <profiles>
         <profile>

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/pom.xml

@@ -76,6 +76,10 @@
             <artifactId>nifi-utils</artifactId>
             <version>1.7.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-proxy-configuration-api</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java

@@ -28,6 +28,7 @@
 import com.amazonaws.regions.Regions;
 import java.io.File;
 import java.io.IOException;
+import java.net.Proxy;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -53,6 +54,8 @@
 import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.processors.aws.credentials.provider.factory.CredentialPropertyDescriptors;
 import org.apache.nifi.processors.aws.regions.AWSRegions;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxySpec;
 import org.apache.nifi.ssl.SSLContextService;
 
 /**
@@ -93,6 +96,25 @@ public abstract class AbstractAWSProcessor<ClientType extends AmazonWebServiceCl
             .addValidator(StandardValidators.PORT_VALIDATOR)
             .build();
 
+    public static final PropertyDescriptor PROXY_USERNAME = new PropertyDescriptor.Builder()
+            .name("proxy-user-name")
+            .displayName("Proxy Username")
+            .description("Proxy username")
+            .expressionLanguageSupported(true)
+            .required(false)
+            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+            .build();
+
+    public static final PropertyDescriptor PROXY_PASSWORD = new PropertyDescriptor.Builder()
+            .name("proxy-user-password")
+            .displayName("Proxy Password")
+            .description("Proxy password")
+            .expressionLanguageSupported(true)
+            .required(false)
+            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+            .sensitive(true)
+            .build();
+
     public static final PropertyDescriptor REGION = new PropertyDescriptor.Builder()
             .name("Region")
             .required(true)
@@ -131,6 +153,9 @@ public abstract class AbstractAWSProcessor<ClientType extends AmazonWebServiceCl
     protected static final Protocol DEFAULT_PROTOCOL = Protocol.HTTPS;
     protected static final String DEFAULT_USER_AGENT = "NiFi";
 
+    private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP_AUTH};
+    public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE = ProxyConfiguration.createProxyConfigPropertyDescriptor(true, PROXY_SPECS);
+
     private static AllowableValue createAllowableValue(final Regions region) {
         return new AllowableValue(region.getName(), AWSRegions.getRegionDisplayName(region.getName()));
     }
@@ -169,6 +194,8 @@ protected Collection<ValidationResult> customValidate(final ValidationContext va
             problems.add(new ValidationResult.Builder().input("Proxy Host Port").valid(false).explanation("Both proxy host and port must be set").build());
         }
 
+        ProxyConfiguration.validateProxySpec(validationContext, problems, PROXY_SPECS);
+
         return problems;
     }
 
@@ -193,11 +220,31 @@ protected ClientConfiguration createConfiguration(final ProcessContext context)
             }
         }
 
-        if (context.getProperty(PROXY_HOST).isSet()) {
-            String proxyHost = context.getProperty(PROXY_HOST).evaluateAttributeExpressions().getValue();
-            config.setProxyHost(proxyHost);
-            Integer proxyPort = context.getProperty(PROXY_HOST_PORT).evaluateAttributeExpressions().asInteger();
-            config.setProxyPort(proxyPort);
+        final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context, () -> {
+            if (context.getProperty(PROXY_HOST).isSet()) {
+                final ProxyConfiguration componentProxyConfig = new ProxyConfiguration();
+                String proxyHost = context.getProperty(PROXY_HOST).evaluateAttributeExpressions().getValue();
+                Integer proxyPort = context.getProperty(PROXY_HOST_PORT).evaluateAttributeExpressions().asInteger();
+                String proxyUsername = context.getProperty(PROXY_USERNAME).evaluateAttributeExpressions().getValue();
+                String proxyPassword = context.getProperty(PROXY_PASSWORD).evaluateAttributeExpressions().getValue();
+                componentProxyConfig.setProxyType(Proxy.Type.HTTP);
+                componentProxyConfig.setProxyServerHost(proxyHost);
+                componentProxyConfig.setProxyServerPort(proxyPort);
+                componentProxyConfig.setProxyUserName(proxyUsername);
+                componentProxyConfig.setProxyUserPassword(proxyPassword);
+                return componentProxyConfig;
+            }
+            return ProxyConfiguration.DIRECT_CONFIGURATION;
+        });
+
+        if (Proxy.Type.HTTP.equals(proxyConfig.getProxyType())) {
+            config.setProxyHost(proxyConfig.getProxyServerHost());
+            config.setProxyPort(proxyConfig.getProxyServerPort());
+
+            if (proxyConfig.hasCredential()) {
+                config.setProxyUsername(proxyConfig.getProxyUserName());
+                config.setProxyPassword(proxyConfig.getProxyUserPassword());
+            }
         }
 
         return config;

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/cloudwatch/PutCloudWatchMetric.java

@@ -174,7 +174,7 @@ public ValidationResult validate(String subject, String input, ValidationContext
             Collections.unmodifiableList(
                     Arrays.asList(NAMESPACE, METRIC_NAME, VALUE, MAXIMUM, MINIMUM, SAMPLECOUNT, SUM, TIMESTAMP,
                             UNIT, REGION, ACCESS_KEY, SECRET_KEY, CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE,
-                            TIMEOUT, SSL_CONTEXT_SERVICE, ENDPOINT_OVERRIDE, PROXY_HOST, PROXY_HOST_PORT)
+                            TIMEOUT, SSL_CONTEXT_SERVICE, ENDPOINT_OVERRIDE, PROXY_HOST, PROXY_HOST_PORT, PROXY_USERNAME, PROXY_PASSWORD)
             );
 
     private volatile Set<String> dynamicPropertyNames = new HashSet<>();

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/dynamodb/DeleteDynamoDB.java

@@ -74,7 +74,8 @@ public class DeleteDynamoDB extends AbstractWriteDynamoDBProcessor {
     public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(
             Arrays.asList(TABLE, HASH_KEY_NAME, RANGE_KEY_NAME, HASH_KEY_VALUE, RANGE_KEY_VALUE,
                 HASH_KEY_VALUE_TYPE, RANGE_KEY_VALUE_TYPE, BATCH_SIZE, REGION, ACCESS_KEY, SECRET_KEY,
-                CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT, SSL_CONTEXT_SERVICE));
+                CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT, SSL_CONTEXT_SERVICE,
+                PROXY_CONFIGURATION_SERVICE, PROXY_HOST, PROXY_HOST_PORT, PROXY_USERNAME, PROXY_PASSWORD));
 
     @Override
     protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/dynamodb/GetDynamoDB.java

@@ -80,7 +80,8 @@ public class GetDynamoDB extends AbstractDynamoDBProcessor {
     public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(
             Arrays.asList(TABLE, HASH_KEY_NAME, RANGE_KEY_NAME, HASH_KEY_VALUE, RANGE_KEY_VALUE,
                 HASH_KEY_VALUE_TYPE, RANGE_KEY_VALUE_TYPE, JSON_DOCUMENT, BATCH_SIZE, REGION, ACCESS_KEY, SECRET_KEY,
-                CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT, SSL_CONTEXT_SERVICE));
+                CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT, SSL_CONTEXT_SERVICE,
+                PROXY_CONFIGURATION_SERVICE, PROXY_HOST, PROXY_HOST_PORT, PROXY_USERNAME, PROXY_PASSWORD));
 
     public static final Relationship REL_NOT_FOUND = new Relationship.Builder().name("not found")
             .description("FlowFiles are routed to not found relationship if key not found in the table").build();

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/dynamodb/PutDynamoDB.java

@@ -84,7 +84,8 @@ public class PutDynamoDB extends AbstractWriteDynamoDBProcessor {
     public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(
         Arrays.asList(TABLE, HASH_KEY_NAME, RANGE_KEY_NAME, HASH_KEY_VALUE, RANGE_KEY_VALUE,
             HASH_KEY_VALUE_TYPE, RANGE_KEY_VALUE_TYPE, JSON_DOCUMENT, DOCUMENT_CHARSET, BATCH_SIZE,
-            REGION, ACCESS_KEY, SECRET_KEY, CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT, SSL_CONTEXT_SERVICE));
+            REGION, ACCESS_KEY, SECRET_KEY, CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT, SSL_CONTEXT_SERVICE,
+            PROXY_CONFIGURATION_SERVICE, PROXY_HOST, PROXY_HOST_PORT, PROXY_USERNAME, PROXY_PASSWORD));
 
     /**
      * Dyamodb max item size limit 400 kb

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/firehose/PutKinesisFirehose.java

@@ -72,7 +72,7 @@ public class PutKinesisFirehose extends AbstractKinesisFirehoseProcessor {
 
     public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(
             Arrays.asList(KINESIS_FIREHOSE_DELIVERY_STREAM_NAME, BATCH_SIZE, MAX_MESSAGE_BUFFER_SIZE_MB, REGION, ACCESS_KEY, SECRET_KEY, CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT,
-                  PROXY_HOST, PROXY_HOST_PORT, ENDPOINT_OVERRIDE));
+                    PROXY_CONFIGURATION_SERVICE, PROXY_HOST, PROXY_HOST_PORT, PROXY_USERNAME, PROXY_PASSWORD, ENDPOINT_OVERRIDE));
 
     /**
      * Max buffer size 1 MB

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/PutKinesisStream.java

@@ -80,7 +80,7 @@ public class PutKinesisStream extends AbstractKinesisStreamProcessor {
 
     public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(
             Arrays.asList(KINESIS_STREAM_NAME, KINESIS_PARTITION_KEY, BATCH_SIZE, MAX_MESSAGE_BUFFER_SIZE_MB, REGION, ACCESS_KEY, SECRET_KEY, CREDENTIALS_FILE,
-                AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT, PROXY_HOST, PROXY_HOST_PORT, ENDPOINT_OVERRIDE));
+                AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT, PROXY_CONFIGURATION_SERVICE, PROXY_HOST, PROXY_HOST_PORT, PROXY_USERNAME, PROXY_PASSWORD, ENDPOINT_OVERRIDE));
 
     /** A random number generator for cases where partition key is not available */
     protected Random randomParitionKeyGenerator = new Random();

nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/lambda/PutLambda.java

@@ -127,8 +127,8 @@ public class PutLambda extends AbstractAWSLambdaProcessor {
     public static final long MAX_REQUEST_SIZE = 6 * 1000 * 1000;
 
     public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(
-            Arrays.asList(AWS_LAMBDA_FUNCTION_NAME, AWS_LAMBDA_FUNCTION_QUALIFIER, REGION, ACCESS_KEY, SECRET_KEY, CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT
-            ));
+            Arrays.asList(AWS_LAMBDA_FUNCTION_NAME, AWS_LAMBDA_FUNCTION_QUALIFIER, REGION, ACCESS_KEY, SECRET_KEY, CREDENTIALS_FILE, AWS_CREDENTIALS_PROVIDER_SERVICE, TIMEOUT,
+                    PROXY_CONFIGURATION_SERVICE, PROXY_HOST, PROXY_HOST_PORT, PROXY_USERNAME, PROXY_PASSWORD));
 
     @Override
     protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {