Optionally make S3 writes unconditional #137185
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,138 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the "Elastic License | ||
| * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side | ||
| * Public License v 1"; you may not use this file except in compliance with, at | ||
| * your election, the "Elastic License 2.0", the "GNU Affero General Public | ||
| * License v3.0 only", or the "Server Side Public License, v 1". | ||
| */ | ||
|
|
||
| package org.elasticsearch.repositories.s3; | ||
|
|
||
| import fixture.aws.DynamicRegionSupplier; | ||
| import fixture.s3.S3HttpFixture; | ||
| import fixture.s3.S3HttpHandler; | ||
|
|
||
| import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters; | ||
| import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope; | ||
| import com.sun.net.httpserver.HttpHandler; | ||
|
|
||
| import org.elasticsearch.ExceptionsHelper; | ||
| import org.elasticsearch.common.ReferenceDocs; | ||
| import org.elasticsearch.common.Strings; | ||
| import org.elasticsearch.common.io.Streams; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.test.cluster.ElasticsearchCluster; | ||
| import org.elasticsearch.test.cluster.LogType; | ||
| import org.elasticsearch.test.fixtures.testcontainers.TestContainersThreadFilter; | ||
| import org.junit.ClassRule; | ||
| import org.junit.rules.RuleChain; | ||
| import org.junit.rules.TestRule; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.function.Supplier; | ||
| import java.util.function.UnaryOperator; | ||
|
|
||
| import static fixture.aws.AwsCredentialsUtils.fixedAccessKey; | ||
| import static org.hamcrest.Matchers.allOf; | ||
| import static org.hamcrest.Matchers.containsString; | ||
| import static org.hamcrest.Matchers.hasItem; | ||
|
|
||
| @ThreadLeakFilters(filters = { TestContainersThreadFilter.class }) | ||
| @ThreadLeakScope(ThreadLeakScope.Scope.NONE) // https://github.com/elastic/elasticsearch/issues/102482 | ||
| public class RepositoryS3ConditionalWritesUnsupportedRestIT extends AbstractRepositoryS3RestTestCase { | ||
|
|
||
| private static final String PREFIX = getIdentifierPrefix("RepositoryS3BasicCredentialsRestIT"); | ||
| private static final String BUCKET = PREFIX + "bucket"; | ||
| private static final String BASE_PATH = PREFIX + "base_path"; | ||
| private static final String ACCESS_KEY = PREFIX + "access-key"; | ||
| private static final String SECRET_KEY = PREFIX + "secret-key"; | ||
| private static final String CLIENT = "no_conditional_writes_client"; | ||
|
|
||
| private static final Supplier<String> regionSupplier = new DynamicRegionSupplier(); | ||
| private static final S3HttpFixture s3Fixture = new S3HttpFixture( | ||
| true, | ||
| BUCKET, | ||
| BASE_PATH, | ||
| fixedAccessKey(ACCESS_KEY, regionSupplier, "s3") | ||
| ) { | ||
| @Override | ||
| protected HttpHandler createHandler() { | ||
| final var delegateHandler = asInstanceOf(S3HttpHandler.class, super.createHandler()); | ||
| return exchange -> { | ||
| if (exchange.getRequestHeaders().containsKey("if-match") || exchange.getRequestHeaders().containsKey("if-none-match")) { | ||
| final var exception = new AssertionError( | ||
| Strings.format( | ||
| "unsupported conditional write: [%s] with headers [%s]", | ||
| delegateHandler.parseRequest(exchange), | ||
| exchange.getRequestHeaders() | ||
| ) | ||
| ); | ||
| ExceptionsHelper.maybeDieOnAnotherThread(exception); | ||
| throw exception; | ||
| } | ||
| delegateHandler.handle(exchange); | ||
| }; | ||
| } | ||
| }; | ||
|
|
||
| @Override | ||
| protected Settings extraRepositorySettings() { | ||
| return Settings.builder() | ||
| .put(super.extraRepositorySettings()) | ||
| .put(S3Repository.UNSAFELY_INCOMPATIBLE_WITH_S3_WRITES.getKey(), true) | ||
| .build(); | ||
| } | ||
|
|
||
| public static ElasticsearchCluster cluster = ElasticsearchCluster.local() | ||
| .module("repository-s3") | ||
| .systemProperty("aws.region", regionSupplier) | ||
| .keystore("s3.client." + CLIENT + ".access_key", ACCESS_KEY) | ||
| .keystore("s3.client." + CLIENT + ".secret_key", SECRET_KEY) | ||
| .setting("s3.client." + CLIENT + ".endpoint", s3Fixture::getAddress) | ||
| .build(); | ||
|
|
||
| @ClassRule | ||
| public static TestRule ruleChain = RuleChain.outerRule(s3Fixture).around(cluster); | ||
|
|
||
| @Override | ||
| protected String getTestRestCluster() { | ||
| return cluster.getHttpAddresses(); | ||
| } | ||
|
|
||
| @Override | ||
| protected String getBucketName() { | ||
| return BUCKET; | ||
| } | ||
|
|
||
| @Override | ||
| protected String getBasePath() { | ||
| return BASE_PATH; | ||
| } | ||
|
|
||
| @Override | ||
| protected String getClientName() { | ||
| return CLIENT; | ||
| } | ||
|
|
||
| public void testWarningLog() throws IOException { | ||
| final var repoName = randomIdentifier(); | ||
| final var testRepository = new TestRepository(repoName, getClientName(), getBucketName(), getBasePath(), extraRepositorySettings()); | ||
| try (var ignored = testRepository.register(UnaryOperator.identity()); var logStream = cluster.getNodeLog(0, LogType.SERVER)) { | ||
| assertThat( | ||
| Streams.readAllLines(logStream), | ||
| hasItem( | ||
| allOf( | ||
| containsString("WARN"), | ||
| containsString(repoName), | ||
| containsString(""" | ||
| is configured to unsafely avoid conditional writes which may lead to repository corruption; to resolve this \ | ||
| warning, upgrade your storage to a system that is fully compatible with AWS S3 and then remove the \ | ||
| [unsafely_incompatible_with_s3_conditional_writes] repository setting"""), | ||
| containsString(ReferenceDocs.S3_COMPATIBLE_REPOSITORIES.toString()) | ||
| ) | ||
| ) | ||
| ); | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -249,6 +249,11 @@ class S3Repository extends MeteredBlobStoreRepository { | |
| Setting.Property.Dynamic | ||
| ); | ||
|
|
||
| static final Setting<Boolean> UNSAFELY_INCOMPATIBLE_WITH_S3_WRITES = Setting.boolSetting( | ||
| "unsafely_incompatible_with_s3_conditional_writes", | ||
|
Contributor
There was a problem hiding this comment. Do you have other names for the setting in mind? I assume with conditional ETag there would be another parameter, since it's a different feature from existence check, and s3-like-storage might have one of those. Maybe
Contributor
Author
There was a problem hiding this comment. I wasn't planning on distinguishing conditional-on-object-exists from conditional-on-etag writes. Do you see a need to do that? My thinking is that "conditional writes" covers both cases, and although these things were announced by AWS as separate features they're now documented together.
Contributor
There was a problem hiding this comment. The only argument I have is that these features came at different time and use different implementation that s3-like-repos might have only one of them. But the feature parity is blurry, and how much we should put effort into featuring these knobs. I think it's s3-like-repo responsibility provide own s3-like-client rather than using not fully compatible aws client.
Member
There was a problem hiding this comment. I think it might be worth to name this setting more precisely. If a storage does support etag but not the other, it is not obvious whether it should set this setting to I also wonder whether it is better named more positively, e.g.
Contributor
Author
There was a problem hiding this comment.
I think it's clear in that case that the storage does not properly support conditional writes so the user needs to disable them until they can upgrade their storage. I don't think it means much to the user to distinguish the different kinds of conditional writes.
Yeah I'd prefer that but it's also important to get the |
||
| false | ||
| ); | ||
|
|
||
| private final S3Service service; | ||
|
|
||
| private final String bucket; | ||
|
|
@@ -271,6 +276,13 @@ class S3Repository extends MeteredBlobStoreRepository { | |
| */ | ||
| private final TimeValue coolDown; | ||
|
|
||
| /** | ||
| * Some storage claims S3-compatibility despite failing to support the {@code If-Match} and {@code If-None-Match} functionality | ||
| * properly. We allow to disable the use of this functionality, making all writes unconditional, using the | ||
| * {@link #UNSAFELY_INCOMPATIBLE_WITH_S3_WRITES} setting. | ||
| */ | ||
| private final boolean supportsConditionalWrites; | ||
|
|
||
| private final Executor snapshotExecutor; | ||
|
|
||
| private final S3RepositoriesMetrics s3RepositoriesMetrics; | ||
|
|
@@ -347,6 +359,19 @@ class S3Repository extends MeteredBlobStoreRepository { | |
| } | ||
|
|
||
| coolDown = COOLDOWN_PERIOD.get(metadata.settings()); | ||
| supportsConditionalWrites = UNSAFELY_INCOMPATIBLE_WITH_S3_WRITES.get(metadata.settings()) == Boolean.FALSE; | ||
|
|
||
| if (supportsConditionalWrites == false) { | ||
| logger.warn( | ||
| """ | ||
| repository [{}] is configured to unsafely avoid conditional writes which may lead to repository corruption; to resolve \ | ||
| this warning, upgrade your storage to a system that is fully compatible with AWS S3 and then remove the [{}] \ | ||
| repository setting; for more information, see [{}]""", | ||
| metadata.name(), | ||
| UNSAFELY_INCOMPATIBLE_WITH_S3_WRITES.getKey(), | ||
| ReferenceDocs.S3_COMPATIBLE_REPOSITORIES | ||
| ); | ||
| } | ||
|
|
||
| logger.debug( | ||
| "using bucket [{}], chunk_size [{}], server_side_encryption [{}], buffer_size [{}], " | ||
|
|
@@ -485,6 +510,7 @@ protected S3BlobStore createBlobStore() { | |
| maxCopySizeBeforeMultipart, | ||
| cannedACL, | ||
| storageClass, | ||
| supportsConditionalWrites, | ||
| metadata, | ||
| bigArrays, | ||
| threadPool, | ||
|
|
||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
NB no tests for the
REPOSITORY_ANALYSIScondition here, they'll be added when we resolve #134632.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does this mean repo analysis will always fail for storage service not supporting conditional writes? I assume that's intentional even if this means existing compatible ones could start failing? IIUC, this work has something to do with AWS's MPU guarantee. Do we consider a storage passing the analysis if it supports the MPU guarantees but not the conditional writes?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes that's right, this is covered in the docs:
This doesn't yet matter for the non-linearizable MPUs but when we move to using conditional writes for CAS operations we will only do so if the storage supports conditional writes. If the user indicates that their storage doesn't support conditional writes yet then we'll fall back to today's MPU-based CAS implementation.