HADOOP-15426 Make S3guard client resilient to DDB throttle events and network failures (Contributed by Steve Loughran)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/d7c0a08a Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/d7c0a08a Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/d7c0a08a Branch: refs/heads/trunk Commit: d7c0a08a1c077752918a8cf1b4f1900ce2721899 Parents: d32a8d5 Author: Steve Loughran <[email protected]> Authored: Wed Sep 12 21:04:49 2018 -0700 Committer: Aaron Fabbri <[email protected]> Committed: Wed Sep 12 21:04:49 2018 -0700 ---------------------------------------------------------------------- .../src/main/resources/core-default.xml | 19 +- hadoop-tools/hadoop-aws/pom.xml | 4 + .../org/apache/hadoop/fs/s3a/Constants.java | 12 +- .../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 15 +- .../hadoop/fs/s3a/S3AInstrumentation.java | 5 +- .../apache/hadoop/fs/s3a/S3ARetryPolicy.java | 24 +- .../java/org/apache/hadoop/fs/s3a/S3AUtils.java | 11 +- .../fs/s3a/s3guard/DynamoDBMetadataStore.java | 377 +++++++++--- .../s3guard/S3GuardDataAccessRetryPolicy.java | 47 ++ .../site/markdown/tools/hadoop-aws/s3guard.md | 150 ++++- .../site/markdown/tools/hadoop-aws/testing.md | 95 ++- .../fs/s3a/ITestS3AFileSystemContract.java | 5 + .../s3a/s3guard/ITestDynamoDBMetadataStore.java | 12 +- .../ITestDynamoDBMetadataStoreScale.java | 595 ++++++++++++++++--- .../s3a/s3guard/ITestS3GuardToolDynamoDB.java | 34 +- .../AbstractITestS3AMetadataStoreScale.java | 24 +- .../hadoop-aws/src/test/resources/core-site.xml | 10 + 17 files changed, 1181 insertions(+), 258 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml ---------------------------------------------------------------------- diff --git a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml index fdd7a87..81502dc 100644 --- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml +++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml @@ -1447,19 +1447,28 @@ <name>fs.s3a.s3guard.ddb.max.retries</name> <value>9</value> <description> - Max retries on batched DynamoDB operations before giving up and - throwing an IOException. Each retry is delayed with an exponential + Max retries on throttled/incompleted DynamoDB operations + before giving up and throwing an IOException. + Each retry is delayed with an exponential backoff timer which starts at 100 milliseconds and approximately doubles each time. The minimum wait before throwing an exception is sum(100, 200, 400, 800, .. 100*2^N-1 ) == 100 * ((2^N)-1) - So N = 9 yields at least 51.1 seconds (51,100) milliseconds of blocking - before throwing an IOException. + </description> +</property> + +<property> + <name>fs.s3a.s3guard.ddb.throttle.retry.interval</name> + <value>100ms</value> + <description> + Initial interval to retry after a request is throttled events; + the back-off policy is exponential until the number of retries of + fs.s3a.s3guard.ddb.max.retries is reached. </description> </property> <property> <name>fs.s3a.s3guard.ddb.background.sleep</name> - <value>25</value> + <value>25ms</value> <description> Length (in milliseconds) of pause between each batch of deletes when pruning metadata. Prevents prune operations (which can typically be low http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/pom.xml ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/pom.xml b/hadoop-tools/hadoop-aws/pom.xml index c6dddb0..78a3d5e 100644 --- a/hadoop-tools/hadoop-aws/pom.xml +++ b/hadoop-tools/hadoop-aws/pom.xml @@ -184,6 +184,8 @@ <exclude>**/ITestS3AFileContextStatistics.java</exclude> <exclude>**/ITestS3AEncryptionSSEC*.java</exclude> <exclude>**/ITestS3AHuge*.java</exclude> + <!-- this sets out to overlaod DynamoDB, so must be run standalone --> + <exclude>**/ITestDynamoDBMetadataStoreScale.java</exclude> </excludes> </configuration> </execution> @@ -216,6 +218,8 @@ <include>**/ITestS3AFileContextStatistics.java</include> <include>**/ITestS3AHuge*.java</include> <include>**/ITestS3AEncryptionSSEC*.java</include> + <!-- this sets out to overlaod DynamoDB, so must be run standalone --> + <include>**/ITestDynamoDBMetadataStoreScale.java</include> </includes> </configuration> </execution> http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java index 5c9d3cd..3fc25da 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java @@ -458,12 +458,20 @@ public final class Constants { @InterfaceStability.Unstable public static final String S3GUARD_DDB_MAX_RETRIES = "fs.s3a.s3guard.ddb.max.retries"; + /** - * Max retries on batched DynamoDB operations before giving up and + * Max retries on batched/throttled DynamoDB operations before giving up and * throwing an IOException. Default is {@value}. See core-default.xml for * more detail. */ - public static final int S3GUARD_DDB_MAX_RETRIES_DEFAULT = 9; + 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 = + "100ms"; /** * Period of time (in milliseconds) to sleep between batches of writes. http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java index fcbcbcc..e817f0d 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java @@ -1131,6 +1131,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities { /** * Increment a statistic by 1. + * This increments both the instrumentation and storage statistics. * @param statistic The operation to increment */ protected void incrementStatistic(Statistic statistic) { @@ -1139,6 +1140,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities { /** * Increment a statistic by a specific value. + * This increments both the instrumentation and storage statistics. * @param statistic The operation to increment * @param count the count to increment */ @@ -1175,8 +1177,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities { Statistic stat = isThrottleException(ex) ? STORE_IO_THROTTLED : IGNORED_ERRORS; - instrumentation.incrementCounter(stat, 1); - storageStatistics.incrementCounter(stat, 1); + incrementStatistic(stat); } /** @@ -1197,6 +1198,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities { /** * Callback from {@link Invoker} when an operation against a metastore * is retried. + * Always increments the {@link Statistic#S3GUARD_METADATASTORE_RETRY} + * statistic/counter; + * if it is a throttling exception will update the associated + * throttled metrics/statistics. + * * @param ex exception * @param retries number of retries * @param idempotent is the method idempotent @@ -1205,6 +1211,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities { int retries, boolean idempotent) { operationRetried(ex); + incrementStatistic(S3GUARD_METADATASTORE_RETRY); + if (isThrottleException(ex)) { + incrementStatistic(S3GUARD_METADATASTORE_THROTTLED); + instrumentation.addValueToQuantiles(S3GUARD_METADATASTORE_THROTTLE_RATE, 1); + } } /** http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java index 26ecefd..84f9c9f 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java @@ -1032,15 +1032,14 @@ public class S3AInstrumentation implements Closeable, MetricsSource { * Throttled request. */ public void throttled() { - incrementCounter(S3GUARD_METADATASTORE_THROTTLED, 1); - addValueToQuantiles(S3GUARD_METADATASTORE_THROTTLE_RATE, 1); + // counters are incremented by owner. } /** * S3Guard is retrying after a (retryable) failure. */ public void retrying() { - incrementCounter(S3GUARD_METADATASTORE_RETRY, 1); + // counters are incremented by owner. } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java index e6e7895..1e475e1 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java @@ -124,12 +124,7 @@ public class S3ARetryPolicy implements RetryPolicy { // and a separate policy for throttle requests, which are considered // repeatable, even for non-idempotent calls, as the service // rejected the call entirely - throttlePolicy = exponentialBackoffRetry( - conf.getInt(RETRY_THROTTLE_LIMIT, RETRY_THROTTLE_LIMIT_DEFAULT), - conf.getTimeDuration(RETRY_THROTTLE_INTERVAL, - RETRY_THROTTLE_INTERVAL_DEFAULT, - TimeUnit.MILLISECONDS), - TimeUnit.MILLISECONDS); + throttlePolicy = createThrottleRetryPolicy(conf); // client connectivity: fixed retries without care for idempotency connectivityFailure = fixedRetries; @@ -140,6 +135,22 @@ public class S3ARetryPolicy implements RetryPolicy { } /** + * Create the throttling policy. + * This will be called from the S3ARetryPolicy constructor, so + * subclasses must assume they are not initialized. + * @param conf configuration to use. + * @return the retry policy for throttling events. + */ + protected RetryPolicy createThrottleRetryPolicy(final Configuration conf) { + return exponentialBackoffRetry( + conf.getInt(RETRY_THROTTLE_LIMIT, RETRY_THROTTLE_LIMIT_DEFAULT), + conf.getTimeDuration(RETRY_THROTTLE_INTERVAL, + RETRY_THROTTLE_INTERVAL_DEFAULT, + TimeUnit.MILLISECONDS), + TimeUnit.MILLISECONDS); + } + + /** * Subclasses can override this like a constructor to change behavior: call * superclass method, then modify it as needed, and return it. * @return Map from exception type to RetryPolicy @@ -206,6 +217,7 @@ public class S3ARetryPolicy implements RetryPolicy { int retries, int failovers, boolean idempotent) throws Exception { + Preconditions.checkArgument(exception != null, "Null exception"); Exception ex = exception; if (exception instanceof AmazonClientException) { // uprate the amazon client exception for the purpose of exception http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java index c3f1e93..9318a5a 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java @@ -27,6 +27,7 @@ import com.amazonaws.SdkBaseException; import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.auth.EnvironmentVariableCredentialsProvider; import com.amazonaws.auth.InstanceProfileCredentialsProvider; +import com.amazonaws.retry.RetryUtils; import com.amazonaws.services.dynamodbv2.model.AmazonDynamoDBException; import com.amazonaws.services.dynamodbv2.model.LimitExceededException; import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughputExceededException; @@ -358,8 +359,10 @@ public final class S3AUtils { /** * Is the exception an instance of a throttling exception. That * is an AmazonServiceException with a 503 response, any - * exception from DynamoDB for limits exceeded, or an - * {@link AWSServiceThrottledException}. + * exception from DynamoDB for limits exceeded, an + * {@link AWSServiceThrottledException}, + * or anything which the AWS SDK's RetryUtils considers to be + * a throttling exception. * @param ex exception to examine * @return true if it is considered a throttling exception */ @@ -368,7 +371,9 @@ public final class S3AUtils { || ex instanceof ProvisionedThroughputExceededException || ex instanceof LimitExceededException || (ex instanceof AmazonServiceException - && 503 == ((AmazonServiceException)ex).getStatusCode()); + && 503 == ((AmazonServiceException)ex).getStatusCode()) + || (ex instanceof SdkBaseException + && RetryUtils.isThrottlingException((SdkBaseException) ex)); } /** http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java index 784a291..7c826c1 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java @@ -36,10 +36,12 @@ import java.util.Set; import java.util.TreeMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import com.amazonaws.AmazonClientException; +import com.amazonaws.AmazonServiceException; import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.services.dynamodbv2.AmazonDynamoDB; import com.amazonaws.services.dynamodbv2.document.BatchWriteItemOutcome; @@ -77,12 +79,12 @@ import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.s3a.AWSCredentialProviderList; +import org.apache.hadoop.fs.s3a.AWSServiceThrottledException; import org.apache.hadoop.fs.s3a.Constants; import org.apache.hadoop.fs.s3a.Invoker; import org.apache.hadoop.fs.s3a.Retries; import org.apache.hadoop.fs.s3a.S3AFileSystem; import org.apache.hadoop.fs.s3a.S3AInstrumentation; -import org.apache.hadoop.fs.s3a.S3ARetryPolicy; import org.apache.hadoop.fs.s3a.S3AUtils; import org.apache.hadoop.fs.s3a.Tristate; import org.apache.hadoop.fs.s3a.auth.RolePolicies; @@ -198,10 +200,6 @@ public class DynamoDBMetadataStore implements MetadataStore { public static final String E_INCOMPATIBLE_VERSION = "Database table is from an incompatible S3Guard version."; - /** Initial delay for retries when batched operations get throttled by - * DynamoDB. Value is {@value} msec. */ - public static final long MIN_RETRY_SLEEP_MSEC = 100; - @VisibleForTesting static final String DESCRIPTION = "S3Guard metadata store in DynamoDB"; @@ -214,6 +212,13 @@ public class DynamoDBMetadataStore implements MetadataStore { @VisibleForTesting static final String TABLE = "table"; + @VisibleForTesting + static final String HINT_DDB_IOPS_TOO_LOW + = " This may be because the write threshold of DynamoDB is set too low."; + + @VisibleForTesting + static final String THROTTLING = "Throttling"; + private static ValueMap deleteTrackingValueMap = new ValueMap().withBoolean(":false", false); @@ -226,7 +231,14 @@ public class DynamoDBMetadataStore implements MetadataStore { private Configuration conf; private String username; - private RetryPolicy dataAccessRetryPolicy; + /** + * This policy is mostly for batched writes, not for processing + * exceptions in invoke() calls. + * It also has a role purpose in {@link #getVersionMarkerItem()}; + * look at that method for the details. + */ + private RetryPolicy batchWriteRetryPolicy; + private S3AInstrumentation.S3GuardInstrumentation instrumentation; /** Owner FS: only valid if configured with an owner FS. */ @@ -237,8 +249,15 @@ public class DynamoDBMetadataStore implements MetadataStore { Invoker.NO_OP ); - /** Data access can have its own policies. */ - private Invoker dataAccess; + /** Invoker for read operations. */ + private Invoker readOp; + + /** Invoker for write operations. */ + private Invoker writeOp; + + private final AtomicLong readThrottleEvents = new AtomicLong(0); + private final AtomicLong writeThrottleEvents = new AtomicLong(0); + private final AtomicLong batchWriteCapacityExceededEvents = new AtomicLong(0); /** * Total limit on the number of throttle events after which @@ -292,10 +311,8 @@ public class DynamoDBMetadataStore implements MetadataStore { Preconditions.checkNotNull(fs, "Null filesystem"); Preconditions.checkArgument(fs instanceof S3AFileSystem, "DynamoDBMetadataStore only supports S3A filesystem."); - owner = (S3AFileSystem) fs; - instrumentation = owner.getInstrumentation().getS3GuardInstrumentation(); + bindToOwnerFilesystem((S3AFileSystem) fs); final String bucket = owner.getBucket(); - conf = owner.getConf(); String confRegion = conf.getTrimmed(S3GUARD_DDB_REGION_KEY); if (!StringUtils.isEmpty(confRegion)) { region = confRegion; @@ -316,7 +333,6 @@ public class DynamoDBMetadataStore implements MetadataStore { } LOG.debug("Inferring DynamoDB region from S3 bucket: {}", region); } - username = owner.getUsername(); credentials = owner.shareCredentials("s3guard"); dynamoDB = createDynamoDB(conf, region, bucket, credentials); @@ -325,7 +341,7 @@ public class DynamoDBMetadataStore implements MetadataStore { initDataAccessRetries(conf); // set up a full retry policy - invoker = new Invoker(new S3ARetryPolicy(conf), + invoker = new Invoker(new S3GuardDataAccessRetryPolicy(conf), this::retryEvent ); @@ -335,6 +351,20 @@ public class DynamoDBMetadataStore implements MetadataStore { } /** + * Declare that this table is owned by the specific S3A FS instance. + * This will bind some fields to the values provided by the owner, + * including wiring up the instrumentation. + * @param fs owner filesystem + */ + @VisibleForTesting + void bindToOwnerFilesystem(final S3AFileSystem fs) { + owner = fs; + conf = owner.getConf(); + instrumentation = owner.getInstrumentation().getS3GuardInstrumentation(); + username = owner.getUsername(); + } + + /** * Performs one-time initialization of the metadata store via configuration. * * This initialization depends on the configuration object to get AWS @@ -382,16 +412,23 @@ public class DynamoDBMetadataStore implements MetadataStore { /** * Set retry policy. This is driven by the value of * {@link Constants#S3GUARD_DDB_MAX_RETRIES} with an exponential backoff - * between each attempt of {@link #MIN_RETRY_SLEEP_MSEC} milliseconds. + * between each attempt of {@link Constants#S3GUARD_DDB_THROTTLE_RETRY_INTERVAL} + * milliseconds. * @param config configuration for data access */ private void initDataAccessRetries(Configuration config) { - int maxRetries = config.getInt(S3GUARD_DDB_MAX_RETRIES, - S3GUARD_DDB_MAX_RETRIES_DEFAULT); - dataAccessRetryPolicy = RetryPolicies - .exponentialBackoffRetry(maxRetries, MIN_RETRY_SLEEP_MSEC, + batchWriteRetryPolicy = RetryPolicies + .exponentialBackoffRetry( + config.getInt(S3GUARD_DDB_MAX_RETRIES, + S3GUARD_DDB_MAX_RETRIES_DEFAULT), + conf.getTimeDuration(S3GUARD_DDB_THROTTLE_RETRY_INTERVAL, + S3GUARD_DDB_THROTTLE_RETRY_INTERVAL_DEFAULT, + TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); - dataAccess = new Invoker(dataAccessRetryPolicy, this::retryEvent); + final RetryPolicy throttledRetryRetryPolicy + = new S3GuardDataAccessRetryPolicy(config); + readOp = new Invoker(throttledRetryRetryPolicy, this::readRetryEvent); + writeOp = new Invoker(throttledRetryRetryPolicy, this::writeRetryEvent); } @Override @@ -432,11 +469,17 @@ public class DynamoDBMetadataStore implements MetadataStore { if (tombstone) { Item item = PathMetadataDynamoDBTranslation.pathMetadataToItem( new DDBPathMetadata(PathMetadata.tombstone(path))); - invoker.retry("Put tombstone", path.toString(), idempotent, + writeOp.retry( + "Put tombstone", + path.toString(), + idempotent, () -> table.putItem(item)); } else { PrimaryKey key = pathToKey(path); - invoker.retry("Delete key", path.toString(), idempotent, + writeOp.retry( + "Delete key", + path.toString(), + idempotent, () -> table.deleteItem(key)); } } @@ -460,28 +503,38 @@ public class DynamoDBMetadataStore implements MetadataStore { } } - @Retries.OnceRaw - private Item getConsistentItem(PrimaryKey key) { + /** + * Get a consistent view of an item. + * @param path path to look up in the database + * @param path entry + * @return the result + * @throws IOException failure + */ + @Retries.RetryTranslated + private Item getConsistentItem(final Path path) throws IOException { + PrimaryKey key = pathToKey(path); final GetItemSpec spec = new GetItemSpec() .withPrimaryKey(key) .withConsistentRead(true); // strictly consistent read - return table.getItem(spec); + return readOp.retry("get", + path.toString(), + true, + () -> table.getItem(spec)); } @Override - @Retries.OnceTranslated + @Retries.RetryTranslated public DDBPathMetadata get(Path path) throws IOException { return get(path, false); } @Override - @Retries.OnceTranslated + @Retries.RetryTranslated public DDBPathMetadata get(Path path, boolean wantEmptyDirectoryFlag) throws IOException { checkPath(path); LOG.debug("Get from table {} in region {}: {}", tableName, region, path); - return Invoker.once("get", path.toString(), - () -> innerGet(path, wantEmptyDirectoryFlag)); + return innerGet(path, wantEmptyDirectoryFlag); } /** @@ -491,9 +544,8 @@ public class DynamoDBMetadataStore implements MetadataStore { * MetadataStore that it should try to compute the empty directory flag. * @return metadata for {@code path}, {@code null} if not found * @throws IOException IO problem - * @throws AmazonClientException dynamo DB level problem */ - @Retries.OnceRaw + @Retries.RetryTranslated private DDBPathMetadata innerGet(Path path, boolean wantEmptyDirectoryFlag) throws IOException { final DDBPathMetadata meta; @@ -502,7 +554,7 @@ public class DynamoDBMetadataStore implements MetadataStore { meta = new DDBPathMetadata(makeDirStatus(username, path)); } else { - final Item item = getConsistentItem(pathToKey(path)); + final Item item = getConsistentItem(path); meta = itemToPathMetadata(item, username); LOG.debug("Get from table {} in region {} returning for {}: {}", tableName, region, path, meta); @@ -517,8 +569,10 @@ public class DynamoDBMetadataStore implements MetadataStore { .withConsistentRead(true) .withFilterExpression(IS_DELETED + " = :false") .withValueMap(deleteTrackingValueMap); - final ItemCollection<QueryOutcome> items = table.query(spec); - boolean hasChildren = items.iterator().hasNext(); + boolean hasChildren = readOp.retry("get/hasChildren", + path.toString(), + true, + () -> table.query(spec).iterator().hasNext()); // When this class has support for authoritative // (fully-cached) directory listings, we may also be able to answer // TRUE here. Until then, we don't know if we have full listing or @@ -545,13 +599,16 @@ public class DynamoDBMetadataStore implements MetadataStore { } @Override - @Retries.OnceTranslated + @Retries.RetryTranslated public DirListingMetadata listChildren(final Path path) throws IOException { checkPath(path); LOG.debug("Listing table {} in region {}: {}", tableName, region, path); // find the children in the table - return Invoker.once("listChildren", path.toString(), + return readOp.retry( + "listChildren", + path.toString(), + true, () -> { final QuerySpec spec = new QuerySpec() .withHashKey(pathToParentKeyAttribute(path)) @@ -610,7 +667,7 @@ public class DynamoDBMetadataStore implements MetadataStore { } @Override - @Retries.OnceTranslated + @Retries.RetryTranslated public void move(Collection<Path> pathsToDelete, Collection<PathMetadata> pathsToCreate) throws IOException { if (pathsToDelete == null && pathsToCreate == null) { @@ -639,25 +696,25 @@ public class DynamoDBMetadataStore implements MetadataStore { } } - Invoker.once("move", tableName, - () -> processBatchWriteRequest(null, pathMetadataToItem(newItems))); + processBatchWriteRequest(null, pathMetadataToItem(newItems)); } /** * Helper method to issue a batch write request to DynamoDB. * - * The retry logic here is limited to repeating the write operations - * until all items have been written; there is no other attempt - * at recovery/retry. Throttling is handled internally. + * As well as retrying on the operation invocation, incomplete + * batches are retried until all have been deleted. * @param keysToDelete primary keys to be deleted; can be null * @param itemsToPut new items to be put; can be null + * @return the number of iterations needed to complete the call. */ - @Retries.OnceRaw("Outstanding batch items are updated with backoff") - private void processBatchWriteRequest(PrimaryKey[] keysToDelete, + @Retries.RetryTranslated("Outstanding batch items are updated with backoff") + private int processBatchWriteRequest(PrimaryKey[] keysToDelete, Item[] itemsToPut) throws IOException { final int totalToDelete = (keysToDelete == null ? 0 : keysToDelete.length); final int totalToPut = (itemsToPut == null ? 0 : itemsToPut.length); int count = 0; + int batches = 0; while (count < totalToDelete + totalToPut) { final TableWriteItems writeItems = new TableWriteItems(tableName); int numToDelete = 0; @@ -682,35 +739,66 @@ public class DynamoDBMetadataStore implements MetadataStore { count += numToPut; } - BatchWriteItemOutcome res = dynamoDB.batchWriteItem(writeItems); + // if there's a retry and another process updates things then it's not + // quite idempotent, but this was the case anyway + batches++; + BatchWriteItemOutcome res = writeOp.retry( + "batch write", + "", + true, + () -> dynamoDB.batchWriteItem(writeItems)); // Check for unprocessed keys in case of exceeding provisioned throughput Map<String, List<WriteRequest>> unprocessed = res.getUnprocessedItems(); int retryCount = 0; while (!unprocessed.isEmpty()) { - retryBackoff(retryCount++); - res = dynamoDB.batchWriteItemUnprocessed(unprocessed); + batchWriteCapacityExceededEvents.incrementAndGet(); + batches++; + retryBackoffOnBatchWrite(retryCount++); + // use a different reference to keep the compiler quiet + final Map<String, List<WriteRequest>> upx = unprocessed; + res = writeOp.retry( + "batch write", + "", + true, + () -> dynamoDB.batchWriteItemUnprocessed(upx)); unprocessed = res.getUnprocessedItems(); } } + return batches; } /** * Put the current thread to sleep to implement exponential backoff * depending on retryCount. If max retries are exceeded, throws an * exception instead. + * * @param retryCount number of retries so far * @throws IOException when max retryCount is exceeded. */ - private void retryBackoff(int retryCount) throws IOException { + private void retryBackoffOnBatchWrite(int retryCount) throws IOException { try { // Our RetryPolicy ignores everything but retryCount here. - RetryPolicy.RetryAction action = dataAccessRetryPolicy.shouldRetry(null, + RetryPolicy.RetryAction action = batchWriteRetryPolicy.shouldRetry( + null, retryCount, 0, true); if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) { - throw new IOException( - String.format("Max retries exceeded (%d) for DynamoDB. This may be" - + " because write threshold of DynamoDB is set too low.", - retryCount)); + // Create an AWSServiceThrottledException, with a fake inner cause + // which we fill in to look like a real exception so + // error messages look sensible + AmazonServiceException cause = new AmazonServiceException( + "Throttling"); + cause.setServiceName("S3Guard"); + cause.setStatusCode(AWSServiceThrottledException.STATUS_CODE); + cause.setErrorCode(THROTTLING); // used in real AWS errors + cause.setErrorType(AmazonServiceException.ErrorType.Service); + cause.setErrorMessage(THROTTLING); + cause.setRequestId("n/a"); + throw new AWSServiceThrottledException( + String.format("Max retries during batch write exceeded" + + " (%d) for DynamoDB." + + HINT_DDB_IOPS_TOO_LOW, + retryCount), + cause); } else { LOG.debug("Sleeping {} msec before next retry", action.delayMillis); Thread.sleep(action.delayMillis); @@ -720,12 +808,12 @@ public class DynamoDBMetadataStore implements MetadataStore { } catch (IOException e) { throw e; } catch (Exception e) { - throw new IOException("Unexpected exception", e); + throw new IOException("Unexpected exception " + e, e); } } @Override - @Retries.OnceRaw + @Retries.RetryTranslated public void put(PathMetadata meta) throws IOException { // For a deeply nested path, this method will automatically create the full // ancestry and save respective item in DynamoDB table. @@ -741,7 +829,7 @@ public class DynamoDBMetadataStore implements MetadataStore { } @Override - @Retries.OnceRaw + @Retries.RetryTranslated public void put(Collection<PathMetadata> metas) throws IOException { innerPut(pathMetaToDDBPathMeta(metas)); } @@ -757,8 +845,9 @@ public class DynamoDBMetadataStore implements MetadataStore { /** * Helper method to get full path of ancestors that are nonexistent in table. */ - @Retries.OnceRaw - private Collection<DDBPathMetadata> fullPathsToPut(DDBPathMetadata meta) + @VisibleForTesting + @Retries.RetryTranslated + Collection<DDBPathMetadata> fullPathsToPut(DDBPathMetadata meta) throws IOException { checkPathMetadata(meta); final Collection<DDBPathMetadata> metasToPut = new ArrayList<>(); @@ -771,7 +860,7 @@ public class DynamoDBMetadataStore implements MetadataStore { // first existent ancestor Path path = meta.getFileStatus().getPath().getParent(); while (path != null && !path.isRoot()) { - final Item item = getConsistentItem(pathToKey(path)); + final Item item = getConsistentItem(path); if (!itemExists(item)) { final FileStatus status = makeDirStatus(path, username); metasToPut.add(new DDBPathMetadata(status, Tristate.FALSE, false, @@ -810,7 +899,7 @@ public class DynamoDBMetadataStore implements MetadataStore { * @throws IOException IO problem */ @Override - @Retries.OnceTranslated("retry(listFullPaths); once(batchWrite)") + @Retries.RetryTranslated public void put(DirListingMetadata meta) throws IOException { LOG.debug("Saving to table {} in region {}: {}", tableName, region, meta); @@ -821,15 +910,12 @@ public class DynamoDBMetadataStore implements MetadataStore { false, meta.isAuthoritative()); // First add any missing ancestors... - final Collection<DDBPathMetadata> metasToPut = invoker.retry( - "paths to put", path.toString(), true, - () -> fullPathsToPut(ddbPathMeta)); + final Collection<DDBPathMetadata> metasToPut = fullPathsToPut(ddbPathMeta); // next add all children of the directory metasToPut.addAll(pathMetaToDDBPathMeta(meta.getListing())); - Invoker.once("put", path.toString(), - () -> processBatchWriteRequest(null, pathMetadataToItem(metasToPut))); + processBatchWriteRequest(null, pathMetadataToItem(metasToPut)); } @Override @@ -847,10 +933,10 @@ public class DynamoDBMetadataStore implements MetadataStore { closeAutocloseables(LOG, credentials); credentials = null; } -} + } @Override - @Retries.OnceTranslated + @Retries.RetryTranslated public void destroy() throws IOException { if (table == null) { LOG.info("In destroy(): no table to delete"); @@ -859,10 +945,11 @@ public class DynamoDBMetadataStore implements MetadataStore { LOG.info("Deleting DynamoDB table {} in region {}", tableName, region); Preconditions.checkNotNull(dynamoDB, "Not connected to DynamoDB"); try { - table.delete(); + invoker.retry("delete", null, true, + () -> table.delete()); table.waitForDelete(); - } catch (ResourceNotFoundException rnfe) { - LOG.info("ResourceNotFoundException while deleting DynamoDB table {} in " + } catch (FileNotFoundException rnfe) { + LOG.info("FileNotFoundException while deleting DynamoDB table {} in " + "region {}. This may indicate that the table does not exist, " + "or has been deleted by another concurrent thread or process.", tableName, region); @@ -872,39 +959,50 @@ public class DynamoDBMetadataStore implements MetadataStore { tableName, ie); throw new InterruptedIOException("Table " + tableName + " in region " + region + " has not been deleted"); - } catch (AmazonClientException e) { - throw translateException("destroy", tableName, e); } } - @Retries.OnceRaw + @Retries.RetryTranslated private ItemCollection<ScanOutcome> expiredFiles(long modTime, - String keyPrefix) { + String keyPrefix) throws IOException { String filterExpression = "mod_time < :mod_time and begins_with(parent, :parent)"; String projectionExpression = "parent,child"; ValueMap map = new ValueMap() .withLong(":mod_time", modTime) .withString(":parent", keyPrefix); - return table.scan(filterExpression, projectionExpression, null, map); + return readOp.retry( + "scan", + keyPrefix, + true, + () -> table.scan(filterExpression, projectionExpression, null, map)); } @Override - @Retries.OnceRaw("once(batchWrite)") + @Retries.RetryTranslated public void prune(long modTime) throws IOException { prune(modTime, "/"); } + /** + * Prune files, in batches. There's a sleep between each batch. + * @param modTime Oldest modification time to allow + * @param keyPrefix The prefix for the keys that should be removed + * @throws IOException Any IO/DDB failure. + * @throws InterruptedIOException if the prune was interrupted + */ @Override - @Retries.OnceRaw("once(batchWrite)") + @Retries.RetryTranslated public void prune(long modTime, String keyPrefix) throws IOException { int itemCount = 0; try { Collection<Path> deletionBatch = new ArrayList<>(S3GUARD_DDB_BATCH_WRITE_REQUEST_LIMIT); - int delay = conf.getInt(S3GUARD_DDB_BACKGROUND_SLEEP_MSEC_KEY, - S3GUARD_DDB_BACKGROUND_SLEEP_MSEC_DEFAULT); - Set<Path> parentPathSet = new HashSet<>(); + long delay = conf.getTimeDuration( + S3GUARD_DDB_BACKGROUND_SLEEP_MSEC_KEY, + S3GUARD_DDB_BACKGROUND_SLEEP_MSEC_DEFAULT, + TimeUnit.MILLISECONDS); + Set<Path> parentPathSet = new HashSet<>(); for (Item item : expiredFiles(modTime, keyPrefix)) { DDBPathMetadata md = PathMetadataDynamoDBTranslation .itemToPathMetadata(item, username); @@ -929,7 +1027,8 @@ public class DynamoDBMetadataStore implements MetadataStore { deletionBatch.clear(); } } - if (deletionBatch.size() > 0) { + // final batch of deletes + if (!deletionBatch.isEmpty()) { Thread.sleep(delay); processBatchWriteRequest(pathToKey(deletionBatch), null); @@ -1093,19 +1192,34 @@ public class DynamoDBMetadataStore implements MetadataStore { * Get the version mark item in the existing DynamoDB table. * * As the version marker item may be created by another concurrent thread or - * process, we sleep and retry a limited times before we fail to get it. - * This does not include handling any failure other than "item not found", - * so this method is tagged as "OnceRaw" + * process, we sleep and retry a limited number times if the lookup returns + * with a null value. + * DDB throttling is always retried. */ - @Retries.OnceRaw - private Item getVersionMarkerItem() throws IOException { + @VisibleForTesting + @Retries.RetryTranslated + Item getVersionMarkerItem() throws IOException { final PrimaryKey versionMarkerKey = createVersionMarkerPrimaryKey(VERSION_MARKER); int retryCount = 0; - Item versionMarker = table.getItem(versionMarkerKey); + // look for a version marker, with usual throttling/failure retries. + Item versionMarker = queryVersionMarker(versionMarkerKey); while (versionMarker == null) { + // The marker was null. + // Two possibilities + // 1. This isn't a S3Guard table. + // 2. This is a S3Guard table in construction; another thread/process + // is about to write/actively writing the version marker. + // So that state #2 is handled, batchWriteRetryPolicy is used to manage + // retries. + // This will mean that if the cause is actually #1, failure will not + // be immediate. As this will ultimately result in a failure to + // init S3Guard and the S3A FS, this isn't going to be a performance + // bottleneck -simply a slightly slower failure report than would otherwise + // be seen. + // "if your settings are broken, performance is not your main issue" try { - RetryPolicy.RetryAction action = dataAccessRetryPolicy.shouldRetry(null, + RetryPolicy.RetryAction action = batchWriteRetryPolicy.shouldRetry(null, retryCount, 0, true); if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) { break; @@ -1114,15 +1228,30 @@ public class DynamoDBMetadataStore implements MetadataStore { Thread.sleep(action.delayMillis); } } catch (Exception e) { - throw new IOException("initTable: Unexpected exception", e); + throw new IOException("initTable: Unexpected exception " + e, e); } retryCount++; - versionMarker = table.getItem(versionMarkerKey); + versionMarker = queryVersionMarker(versionMarkerKey); } return versionMarker; } /** + * Issue the query to get the version marker, with throttling for overloaded + * DDB tables. + * @param versionMarkerKey key to look up + * @return the marker + * @throws IOException failure + */ + @Retries.RetryTranslated + private Item queryVersionMarker(final PrimaryKey versionMarkerKey) + throws IOException { + return readOp.retry("getVersionMarkerItem", + VERSION_MARKER, true, + () -> table.getItem(versionMarkerKey)); + } + + /** * Verify that a table version is compatible with this S3Guard client. * @param tableName name of the table (for error messages) * @param versionMarker the version marker retrieved from the table @@ -1207,7 +1336,7 @@ public class DynamoDBMetadataStore implements MetadataStore { * @return the outcome. */ @Retries.OnceRaw - PutItemOutcome putItem(Item item) { + private PutItemOutcome putItem(Item item) { LOG.debug("Putting item {}", item); return table.putItem(item); } @@ -1255,6 +1384,11 @@ public class DynamoDBMetadataStore implements MetadataStore { } @VisibleForTesting + public String getTableName() { + return tableName; + } + + @VisibleForTesting DynamoDB getDynamoDB() { return dynamoDB; } @@ -1312,8 +1446,8 @@ public class DynamoDBMetadataStore implements MetadataStore { } map.put("description", DESCRIPTION); map.put("region", region); - if (dataAccessRetryPolicy != null) { - map.put("retryPolicy", dataAccessRetryPolicy.toString()); + if (batchWriteRetryPolicy != null) { + map.put("retryPolicy", batchWriteRetryPolicy.toString()); } return map; } @@ -1369,6 +1503,38 @@ public class DynamoDBMetadataStore implements MetadataStore { } /** + * Callback on a read operation retried. + * @param text text of the operation + * @param ex exception + * @param attempts number of attempts + * @param idempotent is the method idempotent (this is assumed to be true) + */ + void readRetryEvent( + String text, + IOException ex, + int attempts, + boolean idempotent) { + readThrottleEvents.incrementAndGet(); + retryEvent(text, ex, attempts, true); + } + + /** + * Callback on a write operation retried. + * @param text text of the operation + * @param ex exception + * @param attempts number of attempts + * @param idempotent is the method idempotent (this is assumed to be true) + */ + void writeRetryEvent( + String text, + IOException ex, + int attempts, + boolean idempotent) { + writeThrottleEvents.incrementAndGet(); + retryEvent(text, ex, attempts, idempotent); + } + + /** * Callback from {@link Invoker} when an operation is retried. * @param text text of the operation * @param ex exception @@ -1410,4 +1576,31 @@ public class DynamoDBMetadataStore implements MetadataStore { } } + /** + * Get the count of read throttle events. + * @return the current count of read throttle events. + */ + @VisibleForTesting + public long getReadThrottleEventCount() { + return readThrottleEvents.get(); + } + + /** + * Get the count of write throttle events. + * @return the current count of write throttle events. + */ + @VisibleForTesting + public long getWriteThrottleEventCount() { + return writeThrottleEvents.get(); + } + + @VisibleForTesting + public long getBatchWriteCapacityExceededCount() { + return batchWriteCapacityExceededEvents.get(); + } + + @VisibleForTesting + public Invoker getInvoker() { + return invoker; + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardDataAccessRetryPolicy.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardDataAccessRetryPolicy.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardDataAccessRetryPolicy.java new file mode 100644 index 0000000..915b94a --- /dev/null +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardDataAccessRetryPolicy.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.fs.s3a.s3guard; + +import java.util.concurrent.TimeUnit; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.s3a.S3ARetryPolicy; +import org.apache.hadoop.io.retry.RetryPolicy; + +import static org.apache.hadoop.fs.s3a.Constants.*; +import static org.apache.hadoop.io.retry.RetryPolicies.exponentialBackoffRetry; + +/** + * A Retry policy whose throttling comes from the S3Guard config options. + */ +public class S3GuardDataAccessRetryPolicy extends S3ARetryPolicy { + + public S3GuardDataAccessRetryPolicy(final Configuration conf) { + super(conf); + } + + protected RetryPolicy createThrottleRetryPolicy(final Configuration conf) { + return exponentialBackoffRetry( + conf.getInt(S3GUARD_DDB_MAX_RETRIES, S3GUARD_DDB_MAX_RETRIES_DEFAULT), + conf.getTimeDuration(S3GUARD_DDB_THROTTLE_RETRY_INTERVAL, + S3GUARD_DDB_THROTTLE_RETRY_INTERVAL_DEFAULT, + TimeUnit.MILLISECONDS), + TimeUnit.MILLISECONDS); + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md index 6dd840a..a8c8d6c 100644 --- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md +++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md @@ -298,8 +298,9 @@ rates. </property> ``` -Attempting to perform more IO than the capacity requested simply throttles the -IO; small capacity numbers are recommended when initially experimenting +Attempting to perform more IO than the capacity requested throttles the +IO, and may result in operations failing. Larger IO capacities cost more. +We recommending using small read and write capacities when initially experimenting with S3Guard. ## Authenticating with S3Guard @@ -327,7 +328,7 @@ to the options `fs.s3a.KEY` *for that bucket only*. As an example, here is a configuration to use different metadata stores and tables for different buckets -First, we define shortcuts for the metadata store classnames +First, we define shortcuts for the metadata store classnames: ```xml @@ -343,7 +344,7 @@ First, we define shortcuts for the metadata store classnames ``` Next, Amazon's public landsat database is configured with no -metadata store +metadata store: ```xml <property> @@ -355,7 +356,7 @@ metadata store ``` Next the `ireland-2` and `ireland-offline` buckets are configured with -DynamoDB as the store, and a shared table `production-table` +DynamoDB as the store, and a shared table `production-table`: ```xml @@ -716,10 +717,10 @@ Metadata Store Diagnostics: ``` After the update, the table status changes to `UPDATING`; this is a sign that -the capacity has been changed +the capacity has been changed. Repeating the same command will not change the capacity, as both read and -write values match that already in use +write values match that already in use. ``` 2017-08-30 16:24:35,337 [main] INFO s3guard.DynamoDBMetadataStore (DynamoDBMetadataStore.java:updateParameters(1090)) - Table capacity unchanged at read: 20, write: 20 @@ -736,6 +737,9 @@ Metadata Store Diagnostics: write-capacity=20 ``` +*Note*: There is a limit to how many times in a 24 hour period the capacity +of a bucket can be changed, either through this command or the AWS console. + ## Debugging and Error Handling If you run into network connectivity issues, or have a machine failure in the @@ -817,6 +821,97 @@ are only made after successful file creation, deletion and rename, the store is *unlikely* to get out of sync, it is still something which merits more testing before it could be considered reliable. +## Managing DynamoDB IO Capacity + +DynamoDB is not only billed on use (data and IO requests), it is billed +on allocated IO Capacity. + +When an application makes more requests than +the allocated capacity permits, the request is rejected; it is up to +the calling application to detect when it is being so throttled and +react. S3Guard does this, but as a result: when the client is being +throttled, operations are slower. This capacity throttling is averaged +over a few minutes: a briefly overloaded table will not be throttled, +but the rate cannot be sustained. + +The load on a table isvisible in the AWS console: go to the +DynamoDB page for the table and select the "metrics" tab. +If the graphs of throttled read or write +requests show that a lot of throttling has taken place, then there is not +enough allocated capacity for the applications making use of the table. + +Similarly, if the capacity graphs show that the read or write loads are +low compared to the allocated capacities, then the table *may* be overprovisioned +for the current workload. + +The S3Guard connector to DynamoDB can be configured to make +multiple attempts to repeat a throttled request, with an exponential +backoff between them. + +The relevant settings for managing retries in the connector are: + +```xml + +<property> + <name>fs.s3a.s3guard.ddb.max.retries</name> + <value>9</value> + <description> + Max retries on throttled/incompleted DynamoDB operations + before giving up and throwing an IOException. + Each retry is delayed with an exponential + backoff timer which starts at 100 milliseconds and approximately + doubles each time. The minimum wait before throwing an exception is + sum(100, 200, 400, 800, .. 100*2^N-1 ) == 100 * ((2^N)-1) + </description> +</property> + +<property> + <name>fs.s3a.s3guard.ddb.throttle.retry.interval</name> + <value>100ms</value> + <description> + Initial interval to retry after a request is throttled events; + the back-off policy is exponential until the number of retries of + fs.s3a.s3guard.ddb.max.retries is reached. + </description> +</property> + +<property> + <name>fs.s3a.s3guard.ddb.background.sleep</name> + <value>25ms</value> + <description> + Length (in milliseconds) of pause between each batch of deletes when + pruning metadata. Prevents prune operations (which can typically be low + priority background operations) from overly interfering with other I/O + operations. + </description> +</property> +``` + +Having a large value for `fs.s3a.s3guard.ddb.max.retries` will ensure +that clients of an overloaded table will not fail immediately. However +queries may be unexpectedly slow. + +If operations, especially directory operations, are slow, check the AWS +console. It is also possible to set up AWS alerts for capacity limits +being exceeded. + +[DynamoDB Auto Scaling](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/AutoScaling.html) +can automatically increase and decrease the allocated capacity. +This is good for keeping capacity high when needed, but avoiding large +bills when it is not. + +Experiments with S3Guard and DynamoDB Auto Scaling have shown that any Auto Scaling +operation will only take place after callers have been throttled for a period of +time. The clients will still need to be configured to retry when overloaded +until any extra capacity is allocated. Furthermore, as this retrying will +block the threads from performing other operations -including more IO, the +the autoscale may not scale fast enough. + +We recommend experimenting with this, based on usage information collected +from previous days, and and choosing a combination of +retry counts and an interval which allow for the clients to cope with +some throttling, but not to time out other applications. + ## Troubleshooting ### Error: `S3Guard table lacks version marker.` @@ -857,12 +952,49 @@ or the configuration is preventing S3Guard from finding the table. region as the bucket being used. 1. Create the table if necessary. + ### Error `"The level of configured provisioned throughput for the table was exceeded"` +``` +org.apache.hadoop.fs.s3a.AWSServiceThrottledException: listFiles on s3a://bucket/10/d1/d2/d3: +com.amazonaws.services.dynamodbv2.model.ProvisionedThroughputExceededException: +The level of configured provisioned throughput for the table was exceeded. +Consider increasing your provisioning level with the UpdateTable API. +(Service: AmazonDynamoDBv2; Status Code: 400; +Error Code: ProvisionedThroughputExceededException; +``` The IO load of clients of the (shared) DynamoDB table was exceeded. -Currently S3Guard doesn't do any throttling and retries here; the way to address -this is to increase capacity via the AWS console or the `set-capacity` command. +1. Increase the capacity of the DynamoDB table. +1. Increase the retry count and/or sleep time of S3Guard on throttle events. +1. Enable capacity autoscaling for the table in the AWS console. + +### Error `Max retries exceeded` + +The I/O load of clients of the (shared) DynamoDB table was exceeded, and +the number of attempts to retry the operation exceeded the configured amount. + +1. Increase the capacity of the DynamoDB table. +1. Increase the retry count and/or sleep time of S3Guard on throttle events. +1. Enable capacity autoscaling for the table in the AWS console. + + +### Error when running `set-capacity`: `org.apache.hadoop.fs.s3a.AWSServiceThrottledException: ProvisionTable` + +``` +org.apache.hadoop.fs.s3a.AWSServiceThrottledException: ProvisionTable on s3guard-example: +com.amazonaws.services.dynamodbv2.model.LimitExceededException: +Subscriber limit exceeded: Provisioned throughput decreases are limited within a given UTC day. +After the first 4 decreases, each subsequent decrease in the same UTC day can be performed at most once every 3600 seconds. +Number of decreases today: 6. +Last decrease at Wednesday, July 25, 2018 8:48:14 PM UTC. +Next decrease can be made at Wednesday, July 25, 2018 9:48:14 PM UTC +``` + +There's are limit on how often you can change the capacity of an DynamoDB table; +if you call set-capacity too often, it fails. Wait until the after the time indicated +and try again. + ## Other Topics http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/testing.md ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/testing.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/testing.md index 280636b..31d3a5f 100644 --- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/testing.md +++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/testing.md @@ -742,6 +742,54 @@ sequential one afterwards. The IO heavy ones must also be subclasses of This is invaluable for debugging test failures. +### Keeping AWS Costs down + +Most of the base S3 tests are designed to use public AWS data +(the landsat-pds bucket) for read IO, so you don't have to pay for bytes +downloaded or long term storage costs. The scale tests do work with more data +so will cost more as well as generally take more time to execute. + +You are however billed for + +1. Data left in S3 after test runs. +2. DynamoDB capacity reserved by S3Guard tables. +3. HTTP operations on files (HEAD, LIST, GET). +4. In-progress multipart uploads from bulk IO or S3A committer tests. +5. Encryption/decryption using AWS KMS keys. + +The GET/decrypt costs are incurred on each partial read of a file, +so random IO can cost more than sequential IO; the speedup of queries with +columnar data usually justifies this. + +The DynamoDB costs come from the number of entries stores and the allocated capacity. + +How to keep costs down + +* Don't run the scale tests with large datasets; keep `fs.s3a.scale.test.huge.filesize` unset, or a few MB (minimum: 5). +* Remove all files in the filesystem. The root tests usually do this, but +it can be manually done: + + hadoop fs -rm -r -f -skipTrash s3a://test-bucket/\* +* Abort all outstanding uploads: + + hadoop s3guard uploads -abort -force s3a://test-bucket/ +* If you don't need it, destroy the S3Guard DDB table. + + hadoop s3guard destroy s3a://hwdev-steve-ireland-new/ + +The S3Guard tests will automatically create the Dynamo DB table in runs with +`-Ds3guard -Ddynamodb` set; default capacity of these buckets +tests is very small; it keeps costs down at the expense of IO performance +and, for test runs in or near the S3/DDB stores, throttling events. + +If you want to manage capacity, use `s3guard set-capacity` to increase it +(performance) or decrease it (costs). +For remote `hadoop-aws` test runs, the read/write capacities of "10" each should suffice; +increase it if parallel test run logs warn of throttling. + +Tip: for agility, use DynamoDB autoscaling, setting the minimum to something very low (e.g 5 units), the maximum to the largest amount you are willing to pay. +This will automatically reduce capacity when you are not running tests against +the bucket, slowly increase it over multiple test runs, if the load justifies it. ## <a name="tips"></a> Tips @@ -985,9 +1033,13 @@ are included in the scale tests executed when `-Dscale` is passed to the maven command line. The two S3Guard scale tests are `ITestDynamoDBMetadataStoreScale` and -`ITestLocalMetadataStoreScale`. To run the DynamoDB test, you will need to -define your table name and region in your test configuration. For example, -the following settings allow us to run `ITestDynamoDBMetadataStoreScale` with +`ITestLocalMetadataStoreScale`. + +To run these tests, your DynamoDB table needs to be of limited capacity; +the values in `ITestDynamoDBMetadataStoreScale` currently require a read capacity +of 10 or less. a write capacity of 15 or more. + +The following settings allow us to run `ITestDynamoDBMetadataStoreScale` with artificially low read and write capacity provisioned, so we can judge the effects of being throttled by the DynamoDB service: @@ -1009,23 +1061,48 @@ effects of being throttled by the DynamoDB service: <value>my-scale-test</value> </property> <property> - <name>fs.s3a.s3guard.ddb.region</name> - <value>us-west-2</value> -</property> -<property> <name>fs.s3a.s3guard.ddb.table.create</name> <value>true</value> </property> <property> <name>fs.s3a.s3guard.ddb.table.capacity.read</name> - <value>10</value> + <value>5</value> </property> <property> <name>fs.s3a.s3guard.ddb.table.capacity.write</name> - <value>10</value> + <value>5</value> </property> ``` +These tests verify that the invoked operations can trigger retries in the +S3Guard code, rather than just in the AWS SDK level, so showing that if +SDK operations fail, they get retried. They also verify that the filesystem +statistics are updated to record that throttling took place. + +*Do not panic if these tests fail to detect throttling!* + +These tests are unreliable as they need certain conditions to be met +to repeatedly fail: + +1. You must have a low-enough latency connection to the DynamoDB store that, +for the capacity allocated, you can overload it. +1. The AWS Console can give you a view of what is happening here. +1. Running a single test on its own is less likely to trigger an overload +than trying to run the whole test suite. +1. And running the test suite more than once, back-to-back, can also help +overload the cluster. +1. Stepping through with a debugger will reduce load, so may not trigger +failures. + +If the tests fail, it *probably* just means you aren't putting enough load +on the table. + +These tests do not verify that the entire set of DynamoDB calls made +during the use of a S3Guarded S3A filesystem are wrapped by retry logic. + +*The best way to verify resilience is to run the entire `hadoop-aws` test suite, +or even a real application, with throttling enabled. + ### Testing only: Local Metadata Store There is an in-memory Metadata Store for testing. http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java index 27af23a..46d6ffc 100644 --- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java +++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java @@ -54,6 +54,11 @@ public class ITestS3AFileSystemContract extends FileSystemContractBaseTest { Thread.currentThread().setName("JUnit-" + methodName.getMethodName()); } + @Override + protected int getGlobalTimeout() { + return S3ATestConstants.S3A_TEST_TIMEOUT; + } + @Before public void setUp() throws Exception { nameThread(); http://git-wip-us.apache.org/repos/asf/hadoop/blob/d7c0a08a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java index 00db04a..a8425bf 100644 --- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java +++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java @@ -95,7 +95,7 @@ public class ITestDynamoDBMetadataStore extends MetadataStoreTestBase { private static DynamoDBMetadataStore ddbmsStatic; - private static String TEST_DYNAMODB_TABLE_NAME; + private static String testDynamoDBTableName; /** * Create a path under the test path provided by @@ -113,8 +113,8 @@ public class ITestDynamoDBMetadataStore extends MetadataStoreTestBase { Configuration conf = prepareTestConfiguration(new Configuration()); assertThatDynamoMetadataStoreImpl(conf); Assume.assumeTrue("Test DynamoDB table name should be set to run " - + "integration tests.", TEST_DYNAMODB_TABLE_NAME != null); - conf.set(S3GUARD_DDB_TABLE_NAME_KEY, TEST_DYNAMODB_TABLE_NAME); + + "integration tests.", testDynamoDBTableName != null); + conf.set(S3GUARD_DDB_TABLE_NAME_KEY, testDynamoDBTableName); s3AContract = new S3AContract(conf); s3AContract.init(); @@ -140,10 +140,10 @@ public class ITestDynamoDBMetadataStore extends MetadataStoreTestBase { public static void beforeClassSetup() throws IOException { Configuration conf = prepareTestConfiguration(new Configuration()); assertThatDynamoMetadataStoreImpl(conf); - TEST_DYNAMODB_TABLE_NAME = conf.get(S3GUARD_DDB_TEST_TABLE_NAME_KEY); + testDynamoDBTableName = conf.get(S3GUARD_DDB_TEST_TABLE_NAME_KEY); Assume.assumeTrue("Test DynamoDB table name should be set to run " - + "integration tests.", TEST_DYNAMODB_TABLE_NAME != null); - conf.set(S3GUARD_DDB_TABLE_NAME_KEY, TEST_DYNAMODB_TABLE_NAME); + + "integration tests.", testDynamoDBTableName != null); + conf.set(S3GUARD_DDB_TABLE_NAME_KEY, testDynamoDBTableName); LOG.debug("Creating static ddbms which will be shared between tests."); ddbmsStatic = new DynamoDBMetadataStore(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
