HADOOP-16653. S3Guard DDB overreacts to no tag access (#1660). Contributed by Gabor Bota.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStoreTableManager.java

@@ -21,6 +21,7 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.nio.file.AccessDeniedException;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
@@ -68,6 +69,7 @@
 import static org.apache.hadoop.fs.s3a.Constants.S3GUARD_DDB_TABLE_CAPACITY_WRITE_KEY;
 import static org.apache.hadoop.fs.s3a.Constants.S3GUARD_DDB_TABLE_CREATE_KEY;
 import static org.apache.hadoop.fs.s3a.Constants.S3GUARD_DDB_TABLE_TAG;
+import static org.apache.hadoop.fs.s3a.S3AUtils.translateDynamoDBException;
 import static org.apache.hadoop.fs.s3a.S3AUtils.translateException;
 import static org.apache.hadoop.fs.s3a.s3guard.DynamoDBMetadataStore.E_ON_DEMAND_NO_SET_CAPACITY;
 import static org.apache.hadoop.fs.s3a.s3guard.DynamoDBMetadataStore.VERSION;
@@ -228,19 +230,19 @@ Table initTable() throws IOException {
     return table;
   }
 
-  protected void tagTableWithVersionMarker() {
+  protected void tagTableWithVersionMarker() throws AmazonDynamoDBException {
     try {
       TagResourceRequest tagResourceRequest = new TagResourceRequest()
           .withResourceArn(table.getDescription().getTableArn())
           .withTags(newVersionMarkerTag());
       amazonDynamoDB.tagResource(tagResourceRequest);
     } catch (AmazonDynamoDBException e) {
-      LOG.warn("Exception during tagging table: {}", e.getMessage());
+      LOG.debug("Exception during tagging table: {}", e.getMessage(), e);
     }
   }
 
   protected static Item getVersionMarkerFromTags(Table table,
-      AmazonDynamoDB addb) {
+      AmazonDynamoDB addb) throws IOException {
     List<Tag> tags = null;
     try {
       final TableDescription description = table.describe();
@@ -252,8 +254,10 @@ protected static Item getVersionMarkerFromTags(Table table,
       LOG.error("Table: {} not found.", table.getTableName());
       throw e;
     } catch (AmazonDynamoDBException e) {
-      LOG.warn("Exception while getting tags from the dynamo table: {}",
-          e.getMessage());
+      LOG.debug("Exception while getting tags from the dynamo table: {}",
+          e.getMessage(), e);
+      throw translateDynamoDBException(table.getTableName(),
+          "Retrieving tags.", e);
     }
 
     if (tags == null) {
@@ -374,8 +378,15 @@ private static Tag newVersionMarkerTag() {
   @VisibleForTesting
   protected void verifyVersionCompatibility() throws IOException {
     final Item versionMarkerItem = getVersionMarkerItem();
-    final Item versionMarkerFromTag =
-        getVersionMarkerFromTags(table, amazonDynamoDB);
+    Item versionMarkerFromTag = null;
+    boolean canReadDdbTags = true;
+
+    try {
+      versionMarkerFromTag = getVersionMarkerFromTags(table, amazonDynamoDB);
+    } catch (AccessDeniedException e) {
+      LOG.debug("Can not read tags of table.");
+      canReadDdbTags = false;
+    }
 
     LOG.debug("versionMarkerItem: {};  versionMarkerFromTag: {}",
         versionMarkerItem, versionMarkerFromTag);
@@ -387,12 +398,19 @@ protected void verifyVersionCompatibility() throws IOException {
             + " Table: " + tableName);
       }
 
-      LOG.info("Table {} contains no version marker item or tag. " +
-              "The table is empty, so the version marker will be added " +
-              "as TAG and ITEM.", tableName);
+      if (canReadDdbTags) {
+        LOG.info("Table {} contains no version marker item and tag. " +
+            "The table is empty, so the version marker will be added " +
+            "as TAG and ITEM.", tableName);
+        putVersionMarkerItemToTable();
+        tagTableWithVersionMarker();
+      }
 
-      tagTableWithVersionMarker();
-      putVersionMarkerItemToTable();
+      if (!canReadDdbTags) {
+        LOG.info("Table {} contains no version marker item and the tags are not readable. " +
+            "The table is empty, so the ITEM version marker will be added .", tableName);
+        putVersionMarkerItemToTable();
+      }
     }
 
     if (versionMarkerItem == null && versionMarkerFromTag != null) {
@@ -408,7 +426,8 @@ protected void verifyVersionCompatibility() throws IOException {
       putVersionMarkerItemToTable();
     }
 
-    if (versionMarkerItem != null && versionMarkerFromTag == null) {
+    if (versionMarkerItem != null && versionMarkerFromTag == null
+        && canReadDdbTags) {
       final int itemVersionMarker =
           extractVersionFromMarker(versionMarkerItem);
       throwExceptionOnVersionMismatch(itemVersionMarker, tableName,

hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md

@@ -977,10 +977,7 @@ the check throws IOException
 
 *Note*: If the user does not have sufficient rights to tag the table the 
 initialization of S3Guard will not fail, but there will be no version marker tag
-on the dynamo table and the following message will be logged on WARN level:
-```
-Exception during tagging table: {AmazonDynamoDBException exception message}
-```
+on the dynamo table.
 
 *Versioning policy*
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumeRole.java

@@ -756,4 +756,28 @@ public void testBucketLocationForbidden() throws Throwable {
     Assertions.assertThat(info)
         .contains(S3GuardTool.BucketInfo.LOCATION_UNKNOWN);
   }
+  /**
+   * Turn off access to dynamo DB Tags and see how DDB table init copes.
+   * There's no testing of the codepath other than checking the logs
+   * - this test does make sure that no regression stops the tag permission
+   * failures from halting the client
+   */
+  @Test
+  public void testRestrictDDBTagAccess() throws Throwable {
+
+    describe("extra policies in assumed roles need;"
+        + " all required policies stated");
+    Configuration conf = createAssumedRoleConfig();
+
+    bindRolePolicyStatements(conf,
+        STATEMENT_S3GUARD_CLIENT,
+        STATEMENT_ALLOW_SSE_KMS_RW,
+        STATEMENT_ALL_S3,
+        new Statement(Effects.Deny)
+            .addActions(S3_PATH_RW_OPERATIONS)
+            .addResources(ALL_DDB_TABLES));
+    Path path = path("testRestrictDDBTagAccess");
+
+    roleFS = (S3AFileSystem) path.getFileSystem(conf);
+  }
 }