Skip to content

fix(inkless:storage): drain object-store deletes monotonically under throttling #715

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jeqo wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(inkless:storage): drain object-store deletes monotonically under throttling #715

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 39 additions & 24 deletions storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,28 +105,28 @@ public void run() {
final Duration sleepDuration = Duration.ofMillis(sleepMillis);
LOGGER.info("No files to delete, sleeping for {}", sleepDuration);
time.sleep(sleepMillis);
} else {
LOGGER.info("Running file cleaner: deleting {} of {} marked files", objectKeyPaths.size(), filesToDelete.size());
metrics.recordFileCleanerStart();
// 1-element holder to carry the duration out of the (synchronous, same-thread) callback
// for the log line below; a plain local cannot be assigned from the lambda.
final long[] durationMs = {0};
TimeUtils.measureDurationMs(time, () -> {
try {
cleanFiles(objectKeyPaths);
} catch (StorageBackendException e) {
LOGGER.error("Error while cleaning files", e);
throw new RuntimeException(e);
}
}, duration -> {
durationMs[0] = duration;
metrics.recordFileCleanerTotalTime(duration);
});
metrics.recordFileCleanerCompleted(objectKeyPaths.size());
LOGGER.info("File cleaner deleted {} files in {} ms", objectKeyPaths.size(), durationMs[0]);
attempts.set(0);
return;
}

attempts.set(0);
LOGGER.info("Running file cleaner: deleting {} of {} marked files", objectKeyPaths.size(), filesToDelete.size());
metrics.recordFileCleanerStart();
final int deletedCount = TimeUtils.measureDurationMs(time,
() -> cleanFiles(objectKeyPaths),
metrics::recordFileCleanerTotalTime);

if (deletedCount == 0) {
// There was work but nothing drained: the backend is likely throttling (S3.delete does
// not throw for that) or hitting hard errors. Back off before the next cycle so we do
// not keep retrying at a fixed cadence while under pressure. Request-rate backoff within
// a cycle is handled by the S3 client's adaptive retry strategy.
final long backoff = errorBackoff.backoff(attempts.incrementAndGet());
LOGGER.warn("File cleaner drained no files this cycle, backing off for {}",
Duration.ofMillis(backoff));
time.sleep(backoff);
} else {
attempts.set(0);
}
} catch (final Exception e) {
metrics.recordFileCleanerError();
final long backoff = errorBackoff.backoff(attempts.incrementAndGet());
Expand All @@ -135,15 +135,30 @@ public void run() {
}
}

private void cleanFiles(Set<String> objectKeyPaths) throws StorageBackendException {
private int cleanFiles(Set<String> objectKeyPaths) throws StorageBackendException {
final Set<ObjectKey> objectKeys = objectKeyPaths.stream()
.map(objectKeyCreator::from)
.collect(Collectors.toSet());
// delete files from storage backend
storage.delete(objectKeys);
// Delete files from the storage backend. Deletion may be partial (e.g. under S3 throttling):
// only the keys the backend confirmed deleted are dereferenced in the control plane, so the
// remaining keys stay marked for deletion and are retried on the next cycle instead of being
// re-attempted after already being deleted.
final Set<ObjectKey> deletedKeys = storage.delete(objectKeys);
if (deletedKeys.isEmpty()) {
LOGGER.warn("No files deleted from storage out of {} candidates; retrying next cycle",
objectKeyPaths.size());
return 0;
}
final Set<String> deletedPaths = deletedKeys.stream()
.map(ObjectKey::value)
.collect(Collectors.toSet());
// update control plane
final DeleteFilesRequest request = new DeleteFilesRequest(objectKeyPaths);
final DeleteFilesRequest request = new DeleteFilesRequest(deletedPaths);
controlPlane.deleteFiles(request);

metrics.recordFileCleanerCompleted(deletedPaths.size());
LOGGER.info("Deleted {} of {} files", deletedPaths.size(), objectKeyPaths.size());
return deletedPaths.size();
}

@Override
Expand Down
27 changes: 20 additions & 7 deletions storage/inkless/src/main/java/io/aiven/inkless/storage_backend/azure/AzureBlobStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,12 +32,16 @@
import com.azure.storage.common.StorageSharedKeyCredential;
import com.groupcdg.pitest.annotations.CoverageIgnore;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
Expand All @@ -52,6 +56,8 @@

@CoverageIgnore // tested on integration level
public final class AzureBlobStorage extends StorageBackend {
private static final Logger LOGGER = LoggerFactory.getLogger(AzureBlobStorage.class);

private AzureBlobStorageConfig config;
private BlobContainerClient blobContainerClient;
private MetricCollector.MetricsPolicy policy;
Expand Down Expand Up @@ -195,16 +201,23 @@ public void delete(final ObjectKey key) throws StorageBackendException {
}

@Override
public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
try {
for (ObjectKey key : keys) {
public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
// Deleting one blob at a time (there is no Azure batch-delete dependency here), so a failure
// on one key must not abandon the rest: accumulate the keys that were removed and leave the
// failed ones for the next FileCleaner cycle. deleteIfExists() returns true if the blob was
// deleted and false if it was already absent; both mean the key is gone (idempotent).
final Set<ObjectKey> deleted = new HashSet<>();
for (final ObjectKey key : keys) {
try {
blobContainerClient.getBlobClient(key.value()).deleteIfExists();
deleted.add(key);
} catch (final BlobStorageException e) {
LOGGER.warn("Failed to delete {}; leaving it for the next cycle", key, e);
} catch (final RuntimeException e) {
LOGGER.warn("Failed to delete {}; leaving it for the next cycle", key, Exceptions.unwrap(e));
}
} catch (final BlobStorageException e) {
throw new StorageBackendException("Failed to delete " + keys, e);
} catch (final RuntimeException e) {
throw unwrapReactorExceptions(e, "Failed to delete " + keys);
}
return deleted;
}

private StorageBackendException unwrapReactorExceptions(final RuntimeException e, final String message) {
Expand Down
9 changes: 8 additions & 1 deletion storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectDeleter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,13 @@ public interface ObjectDeleter extends Closeable {
* Delete objects from a set of keys.
*
* <p>If the object doesn't exist, the operation still succeeds as it is idempotent.
*
* <p>Deletion may be partial: implementations return the subset of {@code keys} that were
* confirmed deleted (which includes keys that were already absent). Keys omitted from the
* returned set were not deleted this round (e.g. throttled) and are safe to retry, since
* deletion is idempotent. Implementations may still throw for a total/unexpected failure.
*
* @return the subset of {@code keys} confirmed deleted.
*/
void delete(Set<ObjectKey> keys) throws StorageBackendException;
Set<ObjectKey> delete(Set<ObjectKey> keys) throws StorageBackendException;
}
8 changes: 7 additions & 1 deletion storage/inkless/src/main/java/io/aiven/inkless/storage_backend/gcs/GcsStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,13 +124,19 @@ public void delete(final ObjectKey key) throws StorageBackendException {
}

@Override
public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
try {
final Set<BlobId> ids = keys.stream()
.map(k -> BlobId.of(this.bucketName,k.value()))
.collect(Collectors.toSet());

// storage.delete returns a List<Boolean> of deleted-vs-already-absent, but a genuine
// failure surfaces as a thrown BaseServiceException rather than a per-blob flag, so we
// cannot extract a confirmed-deleted subset the way the S3 backend does. This stays
// all-or-nothing: on success every key is gone (idempotent), and on failure we delete
// nothing and let the FileCleaner cycle retry the whole set.
storage.delete(ids);
return Set.copyOf(keys);
} catch (final BaseServiceException e) {
throw new StorageBackendException("Failed to delete " + keys, e);
}
Expand Down
3 changes: 2 additions & 1 deletion ...age/inkless/src/main/java/io/aiven/inkless/storage_backend/in_memory/InMemoryStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,9 +110,10 @@ public void delete(final ObjectKey key) throws StorageBackendException {
}

@Override
public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
Objects.requireNonNull(keys, "keys cannot be null");
keys.forEach(storage::remove);
return Set.copyOf(keys);
}

@Override
Expand Down
100 changes: 73 additions & 27 deletions storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,11 +22,15 @@

import com.groupcdg.pitest.annotations.CoverageIgnore;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand Down Expand Up @@ -54,11 +58,20 @@
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.S3Error;

@CoverageIgnore // tested on integration level
public final class S3Storage extends StorageBackend {

private static final Logger LOGGER = LoggerFactory.getLogger(S3Storage.class);

public static final int MAX_DELETE_KEYS_LIMIT = 1000;

// Per-key S3 error codes that indicate throttling rather than a hard, non-transient failure. Used
// only to log throttling distinctly; both kinds are left for the next FileCleaner cycle to retry.
private static final Set<String> THROTTLE_ERROR_CODES =
Set.of("SlowDown", "ServiceUnavailable", "RequestLimitExceeded");

private S3Client s3Client;
private String bucketName;

Expand Down Expand Up @@ -155,37 +168,70 @@ public void delete(final ObjectKey key) throws StorageBackendException {
}

@Override
public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
final List<ObjectKey> objectKeys = new ArrayList<>(keys);
try {
for (int i = 0; i < objectKeys.size(); i += MAX_DELETE_KEYS_LIMIT) {
final var batch = objectKeys.subList(
i,
Math.min(i + MAX_DELETE_KEYS_LIMIT, objectKeys.size())
);

final Set<ObjectIdentifier> ids = batch.stream()
.map(k -> ObjectIdentifier.builder().key(k.value()).build())
.collect(Collectors.toSet());
final Delete delete = Delete.builder().objects(ids).build();
final DeleteObjectsRequest deleteObjectsRequest = DeleteObjectsRequest.builder()
.bucket(bucketName)
.delete(delete)
.build();
final DeleteObjectsResponse response = s3Client.deleteObjects(deleteObjectsRequest);

if (!response.errors().isEmpty()) {
final var errors = response.errors().stream()
.map(e -> String.format("Error %s: %s (%s)", e.key(), e.message(), e.code()))
.collect(Collectors.joining(", "));
throw new StorageBackendException("Failed to delete keys " + keys + ": " + errors);
final Set<ObjectKey> deleted = new HashSet<>();
for (int i = 0; i < objectKeys.size(); i += MAX_DELETE_KEYS_LIMIT) {
final Set<ObjectKey> batch = new HashSet<>(objectKeys.subList(
i,
Math.min(i + MAX_DELETE_KEYS_LIMIT, objectKeys.size())
));
final Map<String, ObjectKey> byValue = batch.stream()
.collect(Collectors.toMap(ObjectKey::value, k -> k, (a, b) -> a));
final DeleteObjectsResponse response;
try {
response = deleteObjectsOnce(batch);
} catch (final SdkException e) {
// Whole-request failure, including a 503 the SDK's adaptive retry already exhausted and
// timeouts. Stop this pass and leave the remaining keys for the next cleanup cycle;
// deletion is idempotent, so re-attempting them later is safe.
LOGGER.warn("DeleteObjects request failed; leaving {} keys for the next cycle",
objectKeys.size() - deleted.size(), e);
break;
}

for (final var deletedObject : response.deleted()) {
final ObjectKey key = byValue.get(deletedObject.key());
if (key != null) {
deleted.add(key);
}
}
} catch (final ApiCallTimeoutException | ApiCallAttemptTimeoutException e) {
throw new StorageBackendTimeoutException("Failed to delete keys " + keys, e);
} catch (final SdkException e) {
throw new StorageBackendException("Failed to delete keys " + keys, e);
logDeleteErrors(response.errors());
}
return deleted;
}

/**
* Logs per-key delete errors, distinguishing throttling (expected under load, aggregated) from
* hard errors (logged individually). No retry happens here: keys that were not deleted stay marked
* for deletion and are retried on the next FileCleaner cycle, while request-rate backoff is left to
* the S3 client's adaptive retry strategy.
*/
private void logDeleteErrors(final List<S3Error> errors) {
int throttled = 0;
for (final var error : errors) {
if (THROTTLE_ERROR_CODES.contains(error.code())) {
throttled++;
} else {
LOGGER.warn("Failed to delete {}: {} ({}); leaving it for the next cycle",
error.key(), error.message(), error.code());
}
}
if (throttled > 0) {
LOGGER.info("{} keys throttled by S3; leaving them for the next cycle", throttled);
}
}

private DeleteObjectsResponse deleteObjectsOnce(final Set<ObjectKey> keys) {
final Set<ObjectIdentifier> ids = keys.stream()
.map(k -> ObjectIdentifier.builder().key(k.value()).build())
.collect(Collectors.toSet());
final Delete delete = Delete.builder().objects(ids).build();
final DeleteObjectsRequest deleteObjectsRequest = DeleteObjectsRequest.builder()
.bucket(bucketName)
.delete(delete)
.build();
return s3Client.deleteObjects(deleteObjectsRequest);
}

@Override
Expand Down
3 changes: 2 additions & 1 deletion storage/inkless/src/test/java/io/aiven/inkless/config/ConfigTestStorageBackend.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,8 @@ public void delete(ObjectKey key) throws StorageBackendException {
}

@Override
public void delete(Set<ObjectKey> keys) throws StorageBackendException {
public Set<ObjectKey> delete(Set<ObjectKey> keys) throws StorageBackendException {
return Set.copyOf(keys);
}

@Override
Expand Down
Loading
Loading