diff --git a/ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java b/ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java index 4459409be4..af6fe4c8a8 100644 --- a/ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java +++ b/ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java @@ -119,6 +119,16 @@ default CompletableFuture put(NamedBlobRecord record) { */ CompletableFuture pullStaleBlobs(Container container, String blobName, int maxResults); + /** + * Returns the first (lowest) blob name in the container at or after {@code blobNameFrom}, or {@code null} if the + * container has no such blob. A cheap index-only point lookup used by the cleanup runner to advance its cursor past a + * blob name whose stale-version scan repeatedly crosses the database long-transaction limit, so the rest of the + * container can still be cleaned. + * @param container the container to look in. + * @param blobNameFrom the inclusive lower-bound blob name. + */ + CompletableFuture getFirstBlobName(Container container, String blobNameFrom); + /** * Cleanup the stale blobs records */ diff --git a/ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java b/ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java index 050015091c..9dec7bf508 100644 --- a/ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java +++ b/ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java @@ -227,6 +227,23 @@ public CompletableFuture pullStaleBlobs(Container return pullStaleBlobs(container, latestBlob); } + @Override + public CompletableFuture getFirstBlobName(Container container, String blobNameFrom) { + String containerName = container.getName(); + String first = null; + for (String accountName : allRecords.keySet()) { + TreeMap> rowsPerContainer = allRecords.get(accountName).get(containerName); + if (rowsPerContainer == null) { + continue; + } + String candidate = rowsPerContainer.ceilingKey(blobNameFrom); + if (candidate != null && (first == null || candidate.compareTo(first) < 0)) { + first = candidate; + } + } + return CompletableFuture.completedFuture(first); + } + @Override public CompletableFuture pullStaleBlobs(Container container, String latestBlob) { CompletableFuture future = new CompletableFuture<>(); diff --git a/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobsCleanupRunner.java b/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobsCleanupRunner.java index 184f94bc74..b1ecd3216e 100644 --- a/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobsCleanupRunner.java +++ b/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobsCleanupRunner.java @@ -78,6 +78,11 @@ public class NamedBlobsCleanupRunner implements Runnable { private final Counter containerCleanupFailedCount; /** Increments each time a stale-blob page scan is killed by the DB long-transaction limit (full-size or shrunk). */ private final Counter pageScanKilledCount; + /** + * Increments each time the runner skips past a single blob name whose stale-version scan cannot complete under the + * database time limit even at the shrunk page size, so the rest of the container can still be cleaned. + */ + private final Counter blobNameSkippedCount; /** * Maximum number of attempts for a single stale-blob page scan before the container is deferred to the next @@ -94,6 +99,14 @@ public class NamedBlobsCleanupRunner implements Runnable { * database time limit, and the cursor keeps moving forward through the container. */ private static final int SHRUNK_PAGE_SIZE = 50; + /** + * Maximum number of blob names the runner skips past in one container in a single run when even the shrunk page scan + * is killed. Bounds the database load (each skip follows a killed scan) while still letting the rest of the container + * make progress; any remaining skips happen on later runs, which resume from the saved cursor. + */ + private static final int MAX_BLOB_NAME_SKIPS_PER_RUN = 3; + /** Client-side wait for the cheap indexed lookup that finds the next blob name to skip to. */ + private static final long GET_FIRST_BLOB_NAME_TIMEOUT_SECONDS = 30L; public NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountService accountService) { this(router, namedBlobDb, accountService, 0); @@ -141,6 +154,8 @@ public NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountSe metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "ContainerFailedCount")); this.pageScanKilledCount = metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "PageScanKilledCount")); + this.blobNameSkippedCount = + metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "BlobNameSkippedCount")); String cursorMapSizeGauge = MetricRegistry.name(NamedBlobsCleanupRunner.class, "CursorMapSize"); metricRegistry.remove(cursorMapSizeGauge); metricRegistry.register(cursorMapSizeGauge, (Gauge) containerCleanupCursors::size); @@ -228,22 +243,43 @@ private void cleanupContainer(Container container) throws Exception { logger.info("Resuming cleanup of container {} from a saved cursor after a previous deferral", container.getId()); } int pageSize = 0; // 0 means the database default (full-size) page. + int blobNameSkips = 0; NamedBlobDb.StaleBlobsWithLatestBlobName staleBlobsWithLatestBlobName; do { try { staleBlobsWithLatestBlobName = pullStaleBlobsResilient(container, blobName, pageSize); } catch (PageKilledException e) { - if (pageSize != 0) { - // Already at the shrunk page size and still killed: cannot make progress on this container right now. - // Defer it; the next scheduled run resumes from this same cursor. + if (pageSize == 0) { + // First kill at full size: shrink and retry the same cursor so the scan reads fewer rows and the cursor can + // still move forward, rather than re-running the same heavy query (which only adds load). + pageSize = SHRUNK_PAGE_SIZE; + logger.warn("Stale-blob scan for container {} was killed at its cursor; shrinking the page to {} to make " + + "progress instead of retrying the same query", container.getId(), pageSize); + continue; + } + // Even the shrunk page was killed: the blob name at this cursor has too many live versions to scan under the + // database time limit. Skip past it with a cheap indexed lookup so the rest of the container is still cleaned; + // the skipped blob name is left to a targeted reap / TTL / version cap. Bound the skips per run so a broadly + // bloated container cannot spin on many killed scans in a single run. + if (blobNameSkips >= MAX_BLOB_NAME_SKIPS_PER_RUN) { + throw new IllegalStateException("Stale-blob scan for container " + container.getId() + + " was killed even at the shrunk page size; deferring after " + blobNameSkips + " skips this run", e); + } + String stuckBlobName = + namedBlobDb.getFirstBlobName(container, blobName).get(GET_FIRST_BLOB_NAME_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (stuckBlobName == null) { + // Nothing at or after the cursor to skip to: defer rather than drop the cursor, so the next run resumes here. throw new IllegalStateException( "Stale-blob scan for container " + container.getId() + " was killed even at the shrunk page size", e); } - // Shrink and retry the same cursor so the scan reads fewer rows and the cursor can still move forward, - // rather than re-running the same heavy query (which only adds load) or sitting on the page across runs. - pageSize = SHRUNK_PAGE_SIZE; - logger.warn("Stale-blob scan for container {} was killed at its cursor; shrinking the page to {} to make " - + "progress instead of retrying the same query", container.getId(), pageSize); + blobName = stuckBlobName + smallestASCII; // advance strictly past the stuck blob name + blobNameSkips++; + blobNameSkippedCount.inc(); + pageSize = 0; // resume full-size scanning for the region after the skipped blob name + containerCleanupCursors.put(cursorKey, blobName); + logger.warn("Stale-blob scan for container {} was killed even at the shrunk page size; skipping blob name {} " + + "(skip {} of {}) and resuming after it", container.getId(), stuckBlobName, blobNameSkips, + MAX_BLOB_NAME_SKIPS_PER_RUN); continue; } List batchStaleBlobs = staleBlobsWithLatestBlobName.getStaleBlobs(); diff --git a/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobsCleanupRunnerTest.java b/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobsCleanupRunnerTest.java index eef8720f47..aa47ead3ba 100644 --- a/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobsCleanupRunnerTest.java +++ b/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobsCleanupRunnerTest.java @@ -155,6 +155,34 @@ public void testKilledScanDefersContainerWithoutAbortingRunOrScheduler() { verify(namedBlobDb, times(1)).pullStaleBlobs(goodContainer, FIRST_BLOB_NAME); } + @Test + public void testKilledShrunkScanSkipsBloatedBlobNameAndContinues() { + // When even the shrunk page is killed, the runner looks up the offending blob name and skips past it, resuming + // after it so the rest of the container is still cleaned instead of making no progress at all. + Container container = mockContainer((short) 1, Container.NamedBlobMode.OPTIONAL, (short) 100, "container-a"); + AccountService accountService = mock(AccountService.class); + when(accountService.getContainersByStatus(Container.ContainerStatus.ACTIVE)).thenReturn( + Collections.singleton(container)); + when(accountService.getContainersByStatus(Container.ContainerStatus.INACTIVE)).thenReturn(Collections.emptySet()); + + NamedBlobDb namedBlobDb = mock(NamedBlobDb.class); + CompletableFuture killedScan = new CompletableFuture<>(); + killedScan.completeExceptionally(new SQLException("Query execution was interrupted", "70100", 1317)); + // The blob name at the start of the container is fatally bloated: both the full and shrunk scans are killed. + when(namedBlobDb.pullStaleBlobs(eq(container), eq(FIRST_BLOB_NAME))).thenReturn(killedScan); + when(namedBlobDb.pullStaleBlobs(eq(container), eq(FIRST_BLOB_NAME), anyInt())).thenReturn(killedScan); + when(namedBlobDb.getFirstBlobName(eq(container), eq(FIRST_BLOB_NAME))).thenReturn( + CompletableFuture.completedFuture("bloated")); + // Everything after the skipped blob name scans cleanly and the container finishes. + when(namedBlobDb.pullStaleBlobs(eq(container), eq("bloated\u0000"))).thenReturn(CompletableFuture.completedFuture( + new NamedBlobDb.StaleBlobsWithLatestBlobName(Collections.emptyList(), null))); + + new NamedBlobsCleanupRunner(mock(Router.class), namedBlobDb, accountService, 0, new MockTime()).run(); + + verify(namedBlobDb).getFirstBlobName(container, FIRST_BLOB_NAME); + verify(namedBlobDb).pullStaleBlobs(container, "bloated\u0000"); + } + @Test public void testResumesFromSavedCursorAfterKill() { // On the first run the container's first page succeeds (cursor advances to "cursor1"), then the page at "cursor1" diff --git a/ambry-named-mysql/src/main/java/com/github/ambry/named/MySqlNamedBlobDb.java b/ambry-named-mysql/src/main/java/com/github/ambry/named/MySqlNamedBlobDb.java index ea45d7f2f8..0f20113989 100644 --- a/ambry-named-mysql/src/main/java/com/github/ambry/named/MySqlNamedBlobDb.java +++ b/ambry-named-mysql/src/main/java/com/github/ambry/named/MySqlNamedBlobDb.java @@ -167,6 +167,10 @@ public class MySqlNamedBlobDb implements NamedBlobDb { + "ORDER BY %s ASC, %s DESC " + "LIMIT ?", ACCOUNT_ID, CONTAINER_ID, BLOB_NAME, BLOB_ID, VERSION, BLOB_STATE, MODIFIED_TS, DELETED_TS, NAMED_BLOBS_V2, BLOB_NAME, VERSION); + private static final String GET_FIRST_BLOB_NAME = String.format( + "SELECT %s FROM %s WHERE %s = ? AND %s = ? AND %s >= ? ORDER BY %s ASC, %s ASC LIMIT 1", BLOB_NAME, + NAMED_BLOBS_V2, ACCOUNT_ID, CONTAINER_ID, BLOB_NAME, BLOB_NAME, VERSION); + private final AccountService accountService; private final String localDatacenter; private final List remoteDatacenters; @@ -532,6 +536,25 @@ public CompletableFuture pullStaleBlobs(Container }, transactionStateTracker); } + @Override + public CompletableFuture getFirstBlobName(Container container, String blobNameFrom) { + TransactionStateTracker transactionStateTracker = + new GetTransactionStateTracker(remoteDatacenters, localDatacenter); + return executeGenericTransactionAsync(true, (connection) -> { + try (PreparedStatement statement = connection.prepareStatement(GET_FIRST_BLOB_NAME)) { + statement.setInt(1, container.getParentAccountId()); + statement.setInt(2, container.getId()); + statement.setString(3, blobNameFrom); + try (ResultSet resultSet = statement.executeQuery()) { + if (resultSet.next()) { + return resultSet.getString(1); + } + return null; + } + } + }, transactionStateTracker); + } + @Override public CompletableFuture cleanupStaleData(List staleRecords) { return executeGenericTransactionAsync(true, (connection) -> {