in s3-artifact-storage-server/src/main/java/jetbrains/buildServer/artifacts/s3/cleanup/S3CleanupExtension.java [158:257]
private void doClean(@NotNull BuildCleanupContext cleanupContext, @NotNull SFinishedBuild build, @NotNull String pathPrefix, @NotNull List<String> pathsToDelete,
@NotNull Map<String, String> storageSettings) throws IOException, InvalidSettingsException {
String bucketName = S3Util.getBucketName(storageSettings);
assert bucketName != null;
SProject project = findProjectToGetConnection(build);
if (project == null) {
String errMsg = String.format("Failed to cleanup S3 objects from %s bucket, project is not specified to get correct Connection", bucketName);
CLEANUP.warn(errMsg);
cleanupContext.onBuildCleanupError(this, build, errMsg);
return;
}
String projectId = project.getProjectId();
Map<String, String> settings = new HashMap<>(storageSettings);
ParamUtil.putSslValues(myServerPaths, settings);
String suffix = " from S3 bucket [" + bucketName + "]" + " from path [" + pathPrefix + "]";
AtomicInteger succeededNum = new AtomicInteger();
AtomicInteger errorNum = new AtomicInteger();
AtomicReference<List<String>> currentPart = new AtomicReference<>(Collections.emptyList());
List<String> pathsFailedToDelete = new ArrayList<>();
try {
myAmazonS3Provider.withCorrectingRegionAndAcceleration(settings, projectId, client -> {
int batchSize = TeamCityProperties.getInteger(S3Constants.S3_CLEANUP_BATCH_SIZE, 1000);
List<List<String>> partitions = Lists.partition(pathsToDelete, batchSize);
AtomicInteger processedChunksNum = new AtomicInteger();
for (List<String> part : partitions) {
currentPart.set(part);
Disposable threadName = NamedThreadFactory.patchThreadName(progressMessage(build, pathsToDelete, succeededNum, processedChunksNum, partitions.size(), part.size()));
try {
// recreate retrier every time to ensure we can quickly adjust its parameters in runtime
DeleteObjectsResponse response = createRetrier(project).execute(() -> deleteChunk(pathPrefix, bucketName, client, part));
List<DeletedObject> deleted = response.deleted();
succeededNum.addAndGet(deleted.size());
List<S3Error> errors = response.errors();
errors
.forEach(error -> {
String key = error.key();
if (key.startsWith(pathPrefix)) {
CLEANUP.info(() -> "Failed to remove " + key + " from S3 bucket " + bucketName + ": " + error.message());
pathsFailedToDelete.add(key.substring(pathPrefix.length()));
myCleanupListeners.forEach(listener ->
listener.onError(
S3Exception.builder()
.message(error.message())
.build(),
false)
);
}
});
errorNum.addAndGet(errors.size());
deleted
.stream()
.map(DeletedObject::key)
.forEach(key -> myCleanupListeners.forEach(listener -> listener.onSuccess(key)));
} finally {
threadName.dispose();
}
}
return null;
}, true);
} catch (SdkClientException e) {
Throwable innerException = e.getCause();
if (innerException instanceof UnknownHostException) {
CLEANUP.warnAndDebugDetails("Could not establish connection to AWS server", innerException);
} else if (innerException != null) {
CLEANUP.warnAndDebugDetails(EXCEPTION_MESSAGE + currentPart.get(), innerException);
} else {
CLEANUP.warnAndDebugDetails(EXCEPTION_MESSAGE + currentPart.get(), e);
}
errorNum.addAndGet(currentPart.get().size());
myCleanupListeners.forEach(listener -> listener.onError(e, false));
} catch (Exception e) {
CLEANUP.warnAndDebugDetails(EXCEPTION_MESSAGE + currentPart.get(), e);
errorNum.addAndGet(currentPart.get().size());
myCleanupListeners.forEach(listener -> listener.onError(e, false));
}
if (errorNum.get() > 0) {
CLEANUP.warn("Failed to remove [" + errorNum + "] S3 " + StringUtil.pluralize("object", errorNum.get()) + suffix);
cleanupContext.onBuildCleanupError(this, build, "Failed to remove some S3 objects.");
}
CLEANUP.info(() -> "Removed [" + succeededNum + "] S3 " + StringUtil.pluralize("object", succeededNum.get()) + suffix);
if (!pathsFailedToDelete.isEmpty())
pathsToDelete.removeAll(pathsFailedToDelete);
myHelper.removeFromArtifactList(build, pathsToDelete);
}