honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use AmazonS3.doesObjectExist() method in S3BlobContainer (#27723) 

This pull request changes the S3BlobContainer.blobExists() method implementation 
to make it use the AmazonS3.doesObjectExist() method instead of 
AmazonS3.getObjectMetadata(). The AmazonS3 implementation takes care of 
catching any thrown AmazonS3Exception and compares its response code with 404, 
returning false (object does not exist) or lets the exception be propagated.
This commit is contained in:
Tanguy Leroux 2017-12-12 09:30:36 +01:00 committed by GitHub
parent 8188d9f7e5
commit f27cb96a64
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 58 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
core/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java
View File

@ -368,9 +368,9 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
writeIndexGen(updatedRepositoryData, repositoryStateId);
// delete the snapshot file
safeSnapshotBlobDelete(snapshot, snapshotId.getUUID());
deleteSnapshotBlobIgnoringErrors(snapshot, snapshotId.getUUID());
// delete the global metadata file
safeGlobalMetaDataBlobDelete(snapshot, snapshotId.getUUID());
deleteGlobalMetaDataBlobIgnoringErrors(snapshot, snapshotId.getUUID());
// Now delete all indices
for (String index : indices) {
@ -422,37 +422,27 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
}
}
private void safeSnapshotBlobDelete(final SnapshotInfo snapshotInfo, final String blobId) {
if (snapshotInfo != null) {
// we know the version the snapshot was created with
try {
snapshotFormat.delete(snapshotsBlobContainer, blobId);
} catch (IOException e) {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("[{}] Unable to delete snapshot file [{}]", snapshotInfo.snapshotId(), blobId), e);
}
} else {
try {
snapshotFormat.delete(snapshotsBlobContainer, blobId);
} catch (IOException e) {
// snapshot file could not be deleted, log the error
private void deleteSnapshotBlobIgnoringErrors(final SnapshotInfo snapshotInfo, final String blobId) {
try {
snapshotFormat.delete(snapshotsBlobContainer, blobId);
} catch (IOException e) {
if (snapshotInfo != null) {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("[{}] Unable to delete snapshot file [{}]",
snapshotInfo.snapshotId(), blobId), e);
} else {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("Unable to delete snapshot file [{}]", blobId), e);
}
}
}
private void safeGlobalMetaDataBlobDelete(final SnapshotInfo snapshotInfo, final String blobId) {
if (snapshotInfo != null) {
// we know the version the snapshot was created with
try {
globalMetaDataFormat.delete(snapshotsBlobContainer, blobId);
} catch (IOException e) {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("[{}] Unable to delete global metadata file [{}]", snapshotInfo.snapshotId(), blobId), e);
}
} else {
try {
globalMetaDataFormat.delete(snapshotsBlobContainer, blobId);
} catch (IOException e) {
// global metadata file could not be deleted, log the error
private void deleteGlobalMetaDataBlobIgnoringErrors(final SnapshotInfo snapshotInfo, final String blobId) {
try {
globalMetaDataFormat.delete(snapshotsBlobContainer, blobId);
} catch (IOException e) {
if (snapshotInfo != null) {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("[{}] Unable to delete global metadata file [{}]",
snapshotInfo.snapshotId(), blobId), e);
} else {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("Unable to delete global metadata file [{}]", blobId), e);
}
}
@ -512,9 +502,7 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
// When we delete corrupted snapshots we might not know which version we are dealing with
// We can try detecting the version based on the metadata file format
assert ignoreIndexErrors;
if (globalMetaDataFormat.exists(snapshotsBlobContainer, snapshotId.getUUID())) {
snapshotVersion = Version.CURRENT;
} else {
if (globalMetaDataFormat.exists(snapshotsBlobContainer, snapshotId.getUUID()) == false) {
throw new SnapshotMissingException(metadata.name(), snapshotId);
}
}

1
core/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java
View File

@ -581,7 +581,6 @@ public class SnapshotShardsService extends AbstractLifecycleComponent implements
entries.add(updatedEntry);
// Finalize snapshot in the repository
snapshotsService.endSnapshot(updatedEntry);
logger.info("snapshot [{}] is done", updatedEntry.snapshot());
}
} else {
entries.add(entry);

56
core/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java
View File

@ -263,7 +263,7 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
null);
snapshots = new SnapshotsInProgress(newSnapshot);
} else {
throw new ConcurrentSnapshotExecutionException(repositoryName, snapshotName, "a snapshot is already running");
throw new ConcurrentSnapshotExecutionException(repositoryName, snapshotName, " a snapshot is already running");
}
return ClusterState.builder(currentState).putCustom(SnapshotsInProgress.TYPE, snapshots).build();
}
@ -363,6 +363,8 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
repository.initializeSnapshot(snapshot.snapshot().getSnapshotId(), snapshot.indices(), metaData);
snapshotCreated = true;
logger.info("snapshot [{}] started", snapshot.snapshot());
if (snapshot.indices().isEmpty()) {
// No indices in this snapshot - we are done
userCreateSnapshotListener.onResponse();
@ -947,35 +949,33 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
* @param failure failure reason or null if snapshot was successful
*/
private void endSnapshot(final SnapshotsInProgress.Entry entry, final String failure) {
threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(new Runnable() {
@Override
public void run() {
final Snapshot snapshot = entry.snapshot();
try {
final Repository repository = repositoriesService.repository(snapshot.getRepository());
logger.trace("[{}] finalizing snapshot in repository, state: [{}], failure[{}]", snapshot, entry.state(), failure);
ArrayList<SnapshotShardFailure> shardFailures = new ArrayList<>();
for (ObjectObjectCursor<ShardId, ShardSnapshotStatus> shardStatus : entry.shards()) {
ShardId shardId = shardStatus.key;
ShardSnapshotStatus status = shardStatus.value;
if (status.state().failed()) {
shardFailures.add(new SnapshotShardFailure(status.nodeId(), shardId, status.reason()));
}
threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(() -> {
final Snapshot snapshot = entry.snapshot();
try {
final Repository repository = repositoriesService.repository(snapshot.getRepository());
logger.trace("[{}] finalizing snapshot in repository, state: [{}], failure[{}]", snapshot, entry.state(), failure);
ArrayList<SnapshotShardFailure> shardFailures = new ArrayList<>();
for (ObjectObjectCursor<ShardId, ShardSnapshotStatus> shardStatus : entry.shards()) {
ShardId shardId = shardStatus.key;
ShardSnapshotStatus status = shardStatus.value;
if (status.state().failed()) {
shardFailures.add(new SnapshotShardFailure(status.nodeId(), shardId, status.reason()));
}
SnapshotInfo snapshotInfo = repository.finalizeSnapshot(
snapshot.getSnapshotId(),
entry.indices(),
entry.startTime(),
failure,
entry.shards().size(),
Collections.unmodifiableList(shardFailures),
entry.getRepositoryStateId(),
entry.includeGlobalState());
removeSnapshotFromClusterState(snapshot, snapshotInfo, null);
} catch (Exception e) {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("[{}] failed to finalize snapshot", snapshot), e);
removeSnapshotFromClusterState(snapshot, null, e);
}
SnapshotInfo snapshotInfo = repository.finalizeSnapshot(
snapshot.getSnapshotId(),
entry.indices(),
entry.startTime(),
failure,
entry.shards().size(),
Collections.unmodifiableList(shardFailures),
entry.getRepositoryStateId(),
entry.includeGlobalState());
removeSnapshotFromClusterState(snapshot, snapshotInfo, null);
logger.info("snapshot [{}] completed with state [{}]", snapshot, snapshotInfo.state());
} catch (Exception e) {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("[{}] failed to finalize snapshot", snapshot), e);
removeSnapshotFromClusterState(snapshot, null, e);
}
});
}

13
plugins/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobContainer.java
View File

@ -73,14 +73,9 @@ class S3BlobContainer extends AbstractBlobContainer {
@Override
public boolean blobExists(String blobName) {
try {
return SocketAccess.doPrivileged(() -> {
blobStore.client().getObjectMetadata(blobStore.bucket(), buildKey(blobName));
return true;
});
} catch (AmazonS3Exception e) {
return false;
return SocketAccess.doPrivileged(() -> blobStore.client().doesObjectExist(blobStore.bucket(), buildKey(blobName)));
} catch (Exception e) {
throw new BlobStoreException("failed to check if blob exists", e);
throw new BlobStoreException("Failed to check if blob [" + blobName +"] exists", e);
}
}
@ -102,7 +97,7 @@ class S3BlobContainer extends AbstractBlobContainer {
@Override
public void writeBlob(String blobName, InputStream inputStream, long blobSize) throws IOException {
if (blobExists(blobName)) {
throw new FileAlreadyExistsException("blob [" + blobName + "] already exists, cannot overwrite");
throw new FileAlreadyExistsException("Blob [" + blobName + "] already exists, cannot overwrite");
}
SocketAccess.doPrivilegedIOException(() -> {
@ -117,7 +112,7 @@ class S3BlobContainer extends AbstractBlobContainer {
@Override
public void deleteBlob(String blobName) throws IOException {
if (!blobExists(blobName)) {
if (blobExists(blobName) == false) {
throw new NoSuchFileException("Blob [" + blobName + "] does not exist");
}

9
plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/MockAmazonS3.java
View File

@ -21,6 +21,7 @@ package org.elasticsearch.repositories.s3;
import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.SdkClientException;
import com.amazonaws.services.s3.AbstractAmazonS3;
import com.amazonaws.services.s3.model.AmazonS3Exception;
import com.amazonaws.services.s3.model.CopyObjectRequest;
@ -36,14 +37,12 @@ import com.amazonaws.services.s3.model.PutObjectResult;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectInputStream;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.amazonaws.util.Base64;
import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.net.InetAddress;
import java.net.Socket;
import java.security.DigestInputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@ -88,6 +87,12 @@ class MockAmazonS3 extends AbstractAmazonS3 {
return true;
}
@Override
public boolean doesObjectExist(String bucketName, String objectName) throws AmazonServiceException, SdkClientException {
simulateS3SocketConnection();
return blobs.containsKey(objectName);
}
@Override
public ObjectMetadata getObjectMetadata(
GetObjectMetadataRequest getObjectMetadataRequest)