HADOOP-15563 S3Guard to create on-demand DDB tables

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -1581,23 +1581,27 @@
 
 <property>
   <name>fs.s3a.s3guard.ddb.table.capacity.read</name>
-  <value>500</value>
+  <value>0</value>
   <description>
     Provisioned throughput requirements for read operations in terms of capacity
-    units for the DynamoDB table.  This config value will only be used when
-    creating a new DynamoDB table, though later you can manually provision by
-    increasing or decreasing read capacity as needed for existing tables.
-    See DynamoDB documents for more information.
+    units for the DynamoDB table. This config value will only be used when
+    creating a new DynamoDB table.
+    If set to 0 (the default), new tables are created with "per-request" capacity.
+    If a positive integer is provided for this and the write capacity, then
+    a table with "provisioned capacity" will be created.
+    You can change the capacity of an existing provisioned-capacity table
+    through the "s3guard set-capacity" command.
   </description>
 </property>
 
 <property>
   <name>fs.s3a.s3guard.ddb.table.capacity.write</name>
-  <value>100</value>
+  <value>0</value>
   <description>
     Provisioned throughput requirements for write operations in terms of
-    capacity units for the DynamoDB table.  Refer to related config
-    fs.s3a.s3guard.ddb.table.capacity.read before usage.
+    capacity units for the DynamoDB table.
+    If set to 0 (the default), new tables are created with "per-request" capacity.
+    Refer to related configuration option fs.s3a.s3guard.ddb.table.capacity.read
   </description>
 </property>
 

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

@@ -439,7 +439,6 @@ private Constants() {
    * This config has no default value. If the user does not set this, the
    * S3Guard will operate table in the associated S3 bucket region.
    */
-  @InterfaceStability.Unstable
   public static final String S3GUARD_DDB_REGION_KEY =
       "fs.s3a.s3guard.ddb.region";
 
@@ -449,7 +448,6 @@ private Constants() {
    * This config has no default value. If the user does not set this, the
    * S3Guard implementation will use the respective S3 bucket name.
    */
-  @InterfaceStability.Unstable
   public static final String S3GUARD_DDB_TABLE_NAME_KEY =
       "fs.s3a.s3guard.ddb.table";
 
@@ -459,36 +457,45 @@ private Constants() {
    * For example:
    * fs.s3a.s3guard.ddb.table.tag.mytag
    */
-  @InterfaceStability.Unstable
   public static final String S3GUARD_DDB_TABLE_TAG =
       "fs.s3a.s3guard.ddb.table.tag.";
 
-  /**
-   * Test table name to use during DynamoDB integration test.
-   *
-   * The table will be modified, and deleted in the end of the tests.
-   * If this value is not set, the integration tests that would be destructive
-   * won't run.
-   */
-  @InterfaceStability.Unstable
-  public static final String S3GUARD_DDB_TEST_TABLE_NAME_KEY =
-      "fs.s3a.s3guard.ddb.test.table";
-
   /**
    * Whether to create the DynamoDB table if the table does not exist.
+   * Value: {@value}.
    */
-  @InterfaceStability.Unstable
   public static final String S3GUARD_DDB_TABLE_CREATE_KEY =
       "fs.s3a.s3guard.ddb.table.create";
 
-  @InterfaceStability.Unstable
+  /**
+   * Read capacity when creating a table.
+   * When it and the write capacity are both "0", a per-request table is
+   * created.
+   * Value: {@value}.
+   */
   public static final String S3GUARD_DDB_TABLE_CAPACITY_READ_KEY =
       "fs.s3a.s3guard.ddb.table.capacity.read";
-  public static final long S3GUARD_DDB_TABLE_CAPACITY_READ_DEFAULT = 500;
-  @InterfaceStability.Unstable
+
+  /**
+   * Default read capacity when creating a table.
+   * Value: {@value}.
+   */
+  public static final long S3GUARD_DDB_TABLE_CAPACITY_READ_DEFAULT = 0;
+
+  /**
+   * Write capacity when creating a table.
+   * When it and the read capacity are both "0", a per-request table is
+   * created.
+   * Value: {@value}.
+   */
   public static final String S3GUARD_DDB_TABLE_CAPACITY_WRITE_KEY =
       "fs.s3a.s3guard.ddb.table.capacity.write";
-  public static final long S3GUARD_DDB_TABLE_CAPACITY_WRITE_DEFAULT = 100;
+
+  /**
+   * Default write capacity when creating a table.
+   * Value: {@value}.
+   */
+  public static final long S3GUARD_DDB_TABLE_CAPACITY_WRITE_DEFAULT = 0;
 
   /**
    * The maximum put or delete requests per BatchWriteItem request.
@@ -497,7 +504,6 @@ private Constants() {
    */
   public static final int S3GUARD_DDB_BATCH_WRITE_REQUEST_LIMIT = 25;
 
-  @InterfaceStability.Unstable
   public static final String S3GUARD_DDB_MAX_RETRIES =
       "fs.s3a.s3guard.ddb.max.retries";
 
@@ -509,7 +515,6 @@ private Constants() {
   public static final int S3GUARD_DDB_MAX_RETRIES_DEFAULT =
       DEFAULT_MAX_ERROR_RETRIES;
 
-  @InterfaceStability.Unstable
   public static final String S3GUARD_DDB_THROTTLE_RETRY_INTERVAL =
       "fs.s3a.s3guard.ddb.throttle.retry.interval";
   public static final String S3GUARD_DDB_THROTTLE_RETRY_INTERVAL_DEFAULT =
@@ -528,7 +533,6 @@ private Constants() {
   /**
    * The default "Null" metadata store: {@value}.
    */
-  @InterfaceStability.Unstable
   public static final String S3GUARD_METASTORE_NULL
       = "org.apache.hadoop.fs.s3a.s3guard.NullMetadataStore";
 
@@ -561,7 +565,6 @@ private Constants() {
   /**
    * Use DynamoDB for the metadata: {@value}.
    */
-  @InterfaceStability.Unstable
   public static final String S3GUARD_METASTORE_DYNAMO
       = "org.apache.hadoop.fs.s3a.s3guard.DynamoDBMetadataStore";
 

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

@@ -59,6 +59,7 @@
 import com.amazonaws.services.dynamodbv2.document.spec.GetItemSpec;
 import com.amazonaws.services.dynamodbv2.document.spec.QuerySpec;
 import com.amazonaws.services.dynamodbv2.document.utils.ValueMap;
+import com.amazonaws.services.dynamodbv2.model.BillingMode;
 import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
 import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
 import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughputDescription;
@@ -1259,11 +1260,26 @@ void initTable() throws IOException {
             tableName, region, (created != null) ? new Date(created) : null);
       } catch (ResourceNotFoundException rnfe) {
         if (conf.getBoolean(S3GUARD_DDB_TABLE_CREATE_KEY, false)) {
-          final ProvisionedThroughput capacity = new ProvisionedThroughput(
-              conf.getLong(S3GUARD_DDB_TABLE_CAPACITY_READ_KEY,
-                  S3GUARD_DDB_TABLE_CAPACITY_READ_DEFAULT),
-              conf.getLong(S3GUARD_DDB_TABLE_CAPACITY_WRITE_KEY,
-                  S3GUARD_DDB_TABLE_CAPACITY_WRITE_DEFAULT));
+          long readCapacity = conf.getLong(S3GUARD_DDB_TABLE_CAPACITY_READ_KEY,
+              S3GUARD_DDB_TABLE_CAPACITY_READ_DEFAULT);
+          long writeCapacity = conf.getLong(
+              S3GUARD_DDB_TABLE_CAPACITY_WRITE_KEY,
+              S3GUARD_DDB_TABLE_CAPACITY_WRITE_DEFAULT);
+          ProvisionedThroughput capacity;
+          if (readCapacity > 0 && writeCapacity > 0) {
+            capacity = new ProvisionedThroughput(
+                readCapacity,
+                writeCapacity);
+          } else {
+            // at least one capacity value is <= 0
+            // verify they are both exactly zero
+            Preconditions.checkArgument(
+                readCapacity == 0 && writeCapacity == 0,
+                "S3Guard table read capacity %d and and write capacity %d"
+                    + " are inconsistent", readCapacity, writeCapacity);
+            // and set the capacity to null for per-request billing.
+            capacity = null;
+          }
 
           createTable(capacity);
         } else {
@@ -1403,20 +1419,31 @@ private void waitForTableActive(Table t) throws IOException {
    * marker.
    * Creating an setting up the table isn't wrapped by any retry operations;
    * the wait for a table to become available is RetryTranslated.
-   * @param capacity capacity to provision
+   * @param capacity capacity to provision. If null: create a per-request
+   * table.
    * @throws IOException on any failure.
    * @throws InterruptedIOException if the wait was interrupted
    */
   @Retries.OnceRaw
   private void createTable(ProvisionedThroughput capacity) throws IOException {
     try {
-      LOG.info("Creating non-existent DynamoDB table {} in region {}",
-          tableName, region);
-      table = dynamoDB.createTable(new CreateTableRequest()
+      String mode;
+      CreateTableRequest request = new CreateTableRequest()
           .withTableName(tableName)
           .withKeySchema(keySchema())
-          .withAttributeDefinitions(attributeDefinitions())
-          .withProvisionedThroughput(capacity));
+          .withAttributeDefinitions(attributeDefinitions());
+      if (capacity != null) {
+        mode = String.format("with provisioned read capacity %d and"
+                + " write capacity %s",
+            capacity.getReadCapacityUnits(), capacity.getWriteCapacityUnits());
+        request.withProvisionedThroughput(capacity);
+      } else {
+        mode = "with pay-per-request billing";
+        request.withBillingMode(BillingMode.PAY_PER_REQUEST);
+      }
+      LOG.info("Creating non-existent DynamoDB table {} in region {} {}",
+          tableName, region, mode);
+      table = dynamoDB.createTable(request);
       LOG.debug("Awaiting table becoming active");
     } catch (ResourceInUseException e) {
       LOG.warn("ResourceInUseException while creating DynamoDB table {} "
@@ -1446,13 +1473,21 @@ private PutItemOutcome putItem(Item item) {
    * Provision the table with given read and write capacity units.
    * Call will fail if the table is busy, or the new values match the current
    * ones.
-   * @param readCapacity read units
-   * @param writeCapacity write units
+   * <p>
+   * Until the AWS SDK lets us switch a table to on-demand, an attempt to
+   * set the I/O capacity to zero will fail.
+   * @param readCapacity read units: must be greater than zero
+   * @param writeCapacity write units: must be greater than zero
    * @throws IOException on a failure
    */
   @Retries.RetryTranslated
   void provisionTable(Long readCapacity, Long writeCapacity)
       throws IOException {
+
+    if (readCapacity == 0 || writeCapacity == 0) {
+      // table is pay on demand
+      throw new IOException(E_ON_DEMAND_NO_SET_CAPACITY);
+    }
     final ProvisionedThroughput toProvision = new ProvisionedThroughput()
         .withReadCapacityUnits(readCapacity)
         .withWriteCapacityUnits(writeCapacity);

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

@@ -434,7 +434,9 @@ static class Init extends S3GuardTool {
         "\n" +
         "  URLs for Amazon DynamoDB are of the form dynamodb://TABLE_NAME.\n" +
         "  Specifying both the -" + REGION_FLAG + " option and an S3A path\n" +
-        "  is not supported.";
+        "  is not supported.\n"
+        + "To create a table with per-request billing, set the read and write\n"
+        + "capacities to 0";
 
     Init(Configuration conf) {
       super(conf);