flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmetz...@apache.org
Subject flink git commit: [FLINK-4222] Support automatic AWS Credentials discovery.
Date Mon, 22 Aug 2016 12:59:11 GMT
Repository: flink
Updated Branches:
  refs/heads/master 1c3666a7b -> a497ebc80


[FLINK-4222] Support automatic AWS Credentials discovery.

When called without credentials, the AmazonKinesisClient tries to configure itself automatically,
searching for credentials from environment variables, java system properties, and finally
from instance profile credentials delivered through the Amazon EC2 metadata service.

Add the AWSConfigConstant "AUTO", which supports creating an AmazonKinesisClient without any
AWSCredentials, which allows for this auto-discovery mechanism to take place and supports
getting kinesis credentials from the AWS EC2 metadata service.

This closes #2260


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a497ebc8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a497ebc8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a497ebc8

Branch: refs/heads/master
Commit: a497ebc80081b0d240e118456873d8b918b3cdc8
Parents: 1c3666a
Author: Nick Chadwick <chadnickbok@gmail.com>
Authored: Fri Jul 15 16:24:19 2016 -0700
Committer: Robert Metzger <rmetzger@apache.org>
Committed: Mon Aug 22 14:49:51 2016 +0200

----------------------------------------------------------------------
 docs/apis/streaming/connectors/kinesis.md              |  2 +-
 .../connectors/kinesis/config/AWSConfigConstants.java  |  5 ++++-
 .../streaming/connectors/kinesis/util/AWSUtil.java     | 13 +++++++++++--
 3 files changed, 16 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/a497ebc8/docs/apis/streaming/connectors/kinesis.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming/connectors/kinesis.md b/docs/apis/streaming/connectors/kinesis.md
index 63683e4..54a75db 100644
--- a/docs/apis/streaming/connectors/kinesis.md
+++ b/docs/apis/streaming/connectors/kinesis.md
@@ -104,7 +104,7 @@ The above is a simple example of using the consumer. Configuration for
the consu
 instance, the configuration keys for which can be found in `ConsumerConfigConstants`. The
example
 demonstrates consuming a single Kinesis stream in the AWS region "us-east-1". The AWS credentials
are supplied using the basic method in which
 the AWS access key ID and secret access key are directly supplied in the configuration (other
options are setting
-`ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER` to `ENV_VAR`, `SYS_PROP`, and `PROFILE`).
Also, data is being consumed
+`ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER` to `ENV_VAR`, `SYS_PROP`, `PROFILE`, and
`AUTO`). Also, data is being consumed
 from the newest position in the Kinesis stream (the other option will be setting `ConsumerConfigConstants.STREAM_INITIAL_POSITION`
 to `TRIM_HORIZON`, which lets the consumer start reading the Kinesis stream from the earliest
record possible).
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a497ebc8/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/config/AWSConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/config/AWSConfigConstants.java
b/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/config/AWSConfigConstants.java
index cd4eda3..01d4f00 100644
--- a/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/config/AWSConfigConstants.java
+++ b/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/config/AWSConfigConstants.java
@@ -40,7 +40,10 @@ public class AWSConfigConstants {
 		PROFILE,
 
 		/** Simply create AWS credentials by supplying the AWS access key ID and AWS secret key
in the configuration properties */
-		BASIC
+		BASIC,
+
+		/** A credentials provider chain will be used that searches for credentials in this order:
ENV_VARS, SYS_PROPS, PROFILE in the AWS instance metadata **/
+		AUTO,
 	}
 
 	/** The AWS region of the Kinesis streams to be pulled ("us-east-1" is used if not set)
*/

http://git-wip-us.apache.org/repos/asf/flink/blob/a497ebc8/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java
b/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java
index 72fbdcb..06f967b 100644
--- a/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java
+++ b/flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java
@@ -50,8 +50,14 @@ public class AWSUtil {
 		awsClientConfig.setUserAgent("Apache Flink " + EnvironmentInformation.getVersion() +
 			" (" + EnvironmentInformation.getRevisionInformation().commitId + ") Kinesis Connector");
 
-		AmazonKinesisClient client =
-			new AmazonKinesisClient(AWSUtil.getCredentialsProvider(configProps).getCredentials(),
awsClientConfig);
+		AmazonKinesisClient client;
+		if (AWSUtil.getCredentialsProvider(configProps) != null) {
+			client = new AmazonKinesisClient(
+				AWSUtil.getCredentialsProvider(configProps).getCredentials(), awsClientConfig);
+		} else {
+			client = new AmazonKinesisClient(awsClientConfig);
+		}
+
 		client.setRegion(Region.getRegion(Regions.fromName(configProps.getProperty(AWSConfigConstants.AWS_REGION))));
 		if (configProps.containsKey(AWSConfigConstants.AWS_ENDPOINT)) {
 			client.setEndpoint(configProps.getProperty(AWSConfigConstants.AWS_ENDPOINT));
@@ -87,6 +93,9 @@ public class AWSUtil {
 					? new ProfileCredentialsProvider(profileName)
 					: new ProfileCredentialsProvider(profileConfigPath, profileName);
 				break;
+			case AUTO:
+				credentialsProvider = null;
+				break;
 			default:
 			case BASIC:
 				credentialsProvider = new AWSCredentialsProvider() {


Mime
View raw message