destination-s3 use instanceprofile if credentials are not provided

airbyte-integrations/connectors/destination-s3/README.md

@@ -8,6 +8,7 @@ As a community contributor, you will need access to AWS to run the integration t
 
 - Create an S3 bucket for testing.
 - Get your `access_key_id` and `secret_access_key` that can read and write to the above bucket.
+- if you leave `access_key_id` and `secret_access_key` in blank, the authentication will rely on the instance profile authentication
 - Paste the bucket and key information into the config files under [`./sample_secrets`](./sample_secrets).
 - Rename the directory from `sample_secrets` to `secrets`.
 - Feel free to modify the config files with different settings in the acceptance test file (e.g. `S3CsvDestinationAcceptanceTest.java`, method `getFormatConfig`), as long as they follow the schema defined in [spec.json](src/main/resources/spec.json).

airbyte-integrations/connectors/destination-s3/src/main/java/io/airbyte/integrations/destination/s3/S3DestinationConfig.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.destination.s3;
 
+import com.amazonaws.auth.InstanceProfileCredentialsProvider;
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
@@ -87,8 +88,8 @@ public static S3DestinationConfig getS3DestinationConfig(final JsonNode config)
         config.get("s3_bucket_name").asText(),
         bucketPath,
         config.get("s3_bucket_region").asText(),
-        config.get("access_key_id").asText(),
-        config.get("secret_access_key").asText(),
+        config.get("access_key_id") == null ? "" : config.get("access_key_id").asText(),
+        config.get("secret_access_key") == null ? "" : config.get("secret_access_key").asText(),
         partSize,
         format);
   }
@@ -128,7 +129,18 @@ public S3FormatConfig getFormatConfig() {
   public AmazonS3 getS3Client() {
     final AWSCredentials awsCreds = new BasicAWSCredentials(accessKeyId, secretAccessKey);
 
-    if (endpoint == null || endpoint.isEmpty()) {
+    if (accessKeyId.isEmpty() && !secretAccessKey.isEmpty() 
+    || !accessKeyId.isEmpty() && secretAccessKey.isEmpty()) {
+      throw new RuntimeException("Either both accessKeyId and secretAccessKey should be provided, or neither");
+    }
+
+    if (accessKeyId.isEmpty() && secretAccessKey.isEmpty()) {
+      return AmazonS3ClientBuilder.standard()
+      .withCredentials(new InstanceProfileCredentialsProvider(false))
+      .build();
+    } 
+
+    else if (endpoint == null || endpoint.isEmpty()) {
       return AmazonS3ClientBuilder.standard()
           .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
           .withRegion(bucketRegion)

airbyte-integrations/connectors/destination-s3/src/main/resources/spec.json

@@ -12,8 +12,6 @@
       "s3_bucket_name",
       "s3_bucket_path",
       "s3_bucket_region",
-      "access_key_id",
-      "secret_access_key",
       "format"
     ],
     "additionalProperties": false,
@@ -72,14 +70,14 @@
       },
       "access_key_id": {
         "type": "string",
-        "description": "The access key id to access the S3 bucket. Airbyte requires Read and Write permissions to the given bucket.",
+        "description": "The access key id to access the S3 bucket. Airbyte requires Read and Write permissions to the given bucket, if not set, Airbyte will rely on Instance Profile.",
         "title": "S3 Key Id",
         "airbyte_secret": true,
         "examples": ["A012345678910EXAMPLE"]
       },
       "secret_access_key": {
         "type": "string",
-        "description": "The corresponding secret to the access key id.",
+        "description": "The corresponding secret to the access key id, if S3 Key Id is set, then S3 Access Key must also be provided",
         "title": "S3 Access Key",
         "airbyte_secret": true,
         "examples": ["a012345678910ABCDEFGH/AbCdEfGhEXAMPLEKEY"]

docs/integrations/destinations/s3.md

@@ -199,7 +199,7 @@ Under the hood, an Airbyte data stream in Json schema is first converted to an A
 #### Requirements
 
 1. Allow connections from Airbyte server to your AWS S3/ Minio S3 cluster \(if they exist in separate VPCs\).
-2. An S3 bucket with credentials.
+2. An S3 bucket with credentials or an instanceprofile with read/write permissions configured for the host (ec2, eks).
 
 #### Setup Guide
 
@@ -211,18 +211,22 @@ Under the hood, an Airbyte data stream in Json schema is first converted to an A
   * **S3 Bucket Region**
   * **Access Key Id**
     * See [this](https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys) on how to generate an access key.
+    * See [this](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_use_switch-role-ec2_instance-profiles.html) on how to create a instanceprofile.
     * We recommend creating an Airbyte-specific user. This user will require [read and write permissions](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_examples_s3_rw-bucket.html) to objects in the staging bucket.
+    * If the Access Key and Secret Access Key are not provided, the authentication will rely on the instanceprofile.
   * **Secret Access Key**
     * Corresponding key to the above key id.
 * Make sure your S3 bucket is accessible from the machine running Airbyte.
   * This depends on your networking setup.
   * You can check AWS S3 documentation with a tutorial on how to properly configure your S3's access [here](https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-control-overview.html).
+  * If you will use instance profile authentication, make sure the role has permission to read/write on the bucket.
   * The easiest way to verify if Airbyte is able to connect to your S3 bucket is via the check connection tool in the UI.
 
 ## CHANGELOG
 
 | Version | Date | Pull Request | Subject |
 | :--- | :--- | :--- | :--- |
+| 0.2.4 | 2022-01-10 | [\#9399](https://github.com/airbytehq/airbyte/pull/9399) | Use instance profile authentication if credentials are not provided |
 | 0.2.3 | 2022-01-11 | [\#9367](https://github.com/airbytehq/airbyte/pull/9367) | Avro & Parquet: support array field with unknown item type; default any improperly typed field to string. |
 | 0.2.2 | 2021-12-21 | [\#8574](https://github.com/airbytehq/airbyte/pull/8574) | Added namespace to Avro and Parquet record types |
 | 0.2.1 | 2021-12-20 | [\#8974](https://github.com/airbytehq/airbyte/pull/8974) | Release a new version to ensure there is no excessive logging. |