diff --git a/CHANGES.txt b/CHANGES.txt index d9caa34bb..bc4eee4fe 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Fix the spurious oldest segment age in CdcRawDirectorySpaceCleaner (CASSSIDECAR-484) * Wire CDC configs in configs table to SidecarCdcOptions/SidecarStatePersister (CASSSIDECAR-483) * Implement durable operational job tracker (CASSSIDECAR-374) * Remove filesystem path from Http response (CASSSIDECAR-477) diff --git a/server/src/main/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleaner.java b/server/src/main/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleaner.java index 03ed40a14..efe3f618b 100644 --- a/server/src/main/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleaner.java +++ b/server/src/main/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleaner.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.attribute.BasicFileAttributes; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -232,8 +233,13 @@ protected void cleanUpCdcRawDirectory(File cdcRawDirectory) Collections.sort(segmentFiles); long nowInMillis = timeProvider.currentTimeMillis(); - // track the age of the oldest commit log segment to give indication of the time-window buffer available - cdcMetrics.oldestSegmentAge.metric.setValue((int) MILLISECONDS.toSeconds(nowInMillis - segmentFiles.get(0).lastModified())); + // track the age of the oldest commit log segment to give indication of the time-window buffer available. + // Skip emission if lastModified is 0 (file was deleted before we could snapshot its lastModified time). + long oldestLastModified = segmentFiles.get(0).lastModified(); + if (oldestLastModified > 0) + { + cdcMetrics.oldestSegmentAge.metric.setValue((int) MILLISECONDS.toSeconds(nowInMillis - oldestLastModified)); + } LOGGER.debug("Cdc data cleaner directorySizeBytes={} maxedUsageBytes={} upperLimitBytes={}", directorySizeBytes, maxUsageBytes, upperLimitBytes); @@ -254,31 +260,47 @@ protected void cleanUpCdcRawDirectory(File cdcRawDirectory) while (i < segmentFiles.size() - 1 && directorySizeBytes > upperLimitBytes) { CdcRawSegmentFile segment = segmentFiles.get(i); - long ageMillis = nowInMillis - segment.lastModified(); - - if (ageMillis < criticalMillis) - { - LOGGER.error("Insufficient Cdc buffer size to maintain {}-minute window segment={} maxSize={} ageMinutes={}", - MILLISECONDS.toMinutes(criticalMillis), segment, upperLimitBytes, - MILLISECONDS.toMinutes(ageMillis)); - cdcMetrics.criticalCdcRawSpace.metric.update(1); - } - else if (ageMillis < lowMillis) - { - LOGGER.warn("Insufficient Cdc buffer size to maintain {}-minute window segment={} maxSize={} ageMinutes={}", - MILLISECONDS.toMinutes(lowMillis), segment, upperLimitBytes, - MILLISECONDS.toMinutes(ageMillis)); - cdcMetrics.lowCdcRawSpace.metric.update(1); - } + long segmentLastModified = segment.lastModified(); long length = 0; - try + // When lastModified is 0 the segment was already reclaimed between the directory + // Skip the buffer-window alerts and the metrics update but still discount the cached size + // from the local budget so the outer loop stops at the right point instead of over-cleaning + // subsequent live segments. + if (segmentLastModified > 0) { - length = deleteSegment(segment); - cdcMetrics.deletedSegment.metric.update(length); + long ageMillis = nowInMillis - segmentLastModified; + + if (ageMillis < criticalMillis) + { + LOGGER.error("Insufficient Cdc buffer size to maintain {}-minute window segment={} maxSize={} ageMinutes={}", + MILLISECONDS.toMinutes(criticalMillis), segment, upperLimitBytes, + MILLISECONDS.toMinutes(ageMillis)); + cdcMetrics.criticalCdcRawSpace.metric.update(1); + } + else if (ageMillis < lowMillis) + { + LOGGER.warn("Insufficient Cdc buffer size to maintain {}-minute window segment={} maxSize={} ageMinutes={}", + MILLISECONDS.toMinutes(lowMillis), segment, upperLimitBytes, + MILLISECONDS.toMinutes(ageMillis)); + cdcMetrics.lowCdcRawSpace.metric.update(1); + } + + try + { + length = deleteSegment(segment); + cdcMetrics.deletedSegment.metric.update(length); + } + catch (IOException e) + { + LOGGER.warn("Failed to delete cdc segment", e); + } } - catch (IOException e) + else { - LOGGER.warn("Failed to delete cdc segment", e); + LOGGER.warn("Skipping delete for already-reclaimed cdc segment {}; a concurrent reclaim " + + "(e.g. Cassandra reclaiming cdc_raw) occurred between the directory scan and " + + "the deletion pass", segment); + length = segment.length(); } directorySizeBytes -= length; i++; @@ -398,14 +420,33 @@ protected static class CdcRawSegmentFile implements Comparable 0 + // with lastModified == 0). A later delete before the cleanup loop is still handled by + // the lastModified > 0 guards. + private final long lastModified; CdcRawSegmentFile(File logFile) { this.file = logFile; final String name = logFile.getName(); this.segmentId = parseSegmentId(name); - this.len = logFile.length(); this.indexFile = CdcUtil.getIdxFile(logFile); + long size = 0L; + long mtime = 0L; + try + { + BasicFileAttributes attrs = Files.readAttributes(logFile.toPath(), BasicFileAttributes.class); + size = attrs.size(); + mtime = attrs.lastModifiedTime().toMillis(); + } + catch (IOException e) + { + // File removed or unreadable between listFiles and construction; len/mtime + // stay 0 and the lastModified > 0 guards skip metric emission and deletion. + } + this.len = size; + this.lastModified = mtime; } public boolean exists() @@ -430,7 +471,7 @@ public long indexLength() public long lastModified() { - return file.lastModified(); + return lastModified; } public Path path() diff --git a/server/src/test/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleanerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleanerTest.java index 6c0dc7345..ff2f7c54c 100644 --- a/server/src/test/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleanerTest.java +++ b/server/src/test/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleanerTest.java @@ -54,6 +54,7 @@ import org.mockito.stubbing.Answer; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.assumeTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -140,6 +141,127 @@ void testCdcRawDirectorySpaceCleaner(@TempDir Path tempDir) throws IOException cleaner.routineCleanUp(); // it should run fine. } + @Test + void testOldestSegmentAgeNotPoisonedWhenTimestampUnavailable(@TempDir Path tempDir) throws IOException + { + // Large cdc_total_space so the cleaner does not delete anything; we only exercise the age metric. + CdcSystemViewsDatabaseAccessor systemViewsDatabaseAccessor = mock(CdcSystemViewsDatabaseAccessor.class); + when(systemViewsDatabaseAccessor.cdcTotalSpaceBytesSetting()).thenReturn(1L << 30); // 1 GiB + CdcConfiguration cdcConfiguration = new CdcConfigurationImpl(); + ServiceConfiguration serviceConfiguration = mock(ServiceConfiguration.class); + when(serviceConfiguration.cdcConfiguration()).thenReturn(cdcConfiguration); + + // Two small, completed (index-backed) segments so the cleaner proceeds past the < 2 files guard. + InstanceMetadata instanceMetadata = mock(InstanceMetadata.class); + File cdcDir = Files.createDirectory(tempDir.resolve(CdcRawDirectorySpaceCleaner.CDC_DIR_NAME)).toFile(); + writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_1, 1024, true); + writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_2, 1024, true); + when(instanceMetadata.dataDirs()).thenReturn(List.of(cdcDir.getParent() + "/data")); + when(instanceMetadata.cdcDir()).thenReturn(cdcDir.getParent() + "/cdc_raw"); + InstancesMetadata instancesMetadata = new InstancesMetadataImpl(instanceMetadata, DnsResolvers.DEFAULT); + + // Pin "now" so the emitted age is deterministic: file mtime is exactly 120s before the cleaner's clock. + FakeTimeProvider timeProvider = new FakeTimeProvider(); + long nowMillis = (System.currentTimeMillis() / 1000) * 1000; // second-aligned to dodge fs mtime rounding + timeProvider.advance(nowMillis, TimeUnit.MILLISECONDS); + + // FILE_1 has the smallest segmentId, so it is the oldest and drives the gauge. Set it to 120s ago. + File oldest = new File(cdcDir, TEST_SEGMENT_FILE_NAME_1); + long twoMinutesAgoMillis = nowMillis - TimeUnit.SECONDS.toMillis(120); + assumeTrue(oldest.setLastModified(twoMinutesAgoMillis), "Filesystem does not support setting mtime"); + + CdcRawDirectorySpaceCleaner cleaner = new CdcRawDirectorySpaceCleaner( + timeProvider, + systemViewsDatabaseAccessor, + serviceConfiguration, + instancesMetadata, + mockSidecarMetrics + ); + + // First run: the oldest segment was last modified 120s before "now", so the gauge must report ~120s. + // Allow +/-1s to tolerate filesystems that store mtime at second granularity. + cleaner.routineCleanUp(); + int ageAfterValid = cdcMetrics.oldestSegmentAge.metric.getValue(); + assertThat(ageAfterValid).isBetween(119, 121); + + // Simulate the oldest segment's timestamp being unavailable (concurrent delete/rotation makes + // File.lastModified() return 0). Skip on filesystems that do not support a 0 mtime. + assumeTrue(oldest.setLastModified(0L) && oldest.lastModified() == 0L, + "Filesystem does not support a 0 mtime; skipping guard assertion"); + + // Second run: the guard must skip emitting, leaving the previous value intact instead of + // publishing the ~1.8e9s (~57 year) bogus value. + cleaner.routineCleanUp(); + int ageAfterUnavailable = cdcMetrics.oldestSegmentAge.metric.getValue(); + assertThat(ageAfterUnavailable) + .as("oldestSegmentAge must not be poisoned to a decades-large value when the timestamp is unavailable") + .isEqualTo(ageAfterValid) + .isLessThan(1_000_000); + } + + @Test + void testDeletionLoopSkipsSegmentWhenTimestampUnavailable(@TempDir Path tempDir) throws IOException + { + // Small cdc_total_space so the deletion loop enters and must decide per-segment whether + // to call deleteSegment(). Three ~4 KiB segments against a 4 KiB budget → the loop + // processes seg 1 and seg 2 (seg 3 is retained as the last active segment). + CdcSystemViewsDatabaseAccessor systemViewsDatabaseAccessor = mock(CdcSystemViewsDatabaseAccessor.class); + when(systemViewsDatabaseAccessor.cdcTotalSpaceBytesSetting()).thenReturn(4L * 1024L); + CdcConfiguration cdcConfiguration = new CdcConfigurationImpl(); + ServiceConfiguration serviceConfiguration = mock(ServiceConfiguration.class); + when(serviceConfiguration.cdcConfiguration()).thenReturn(cdcConfiguration); + + InstanceMetadata instanceMetadata = mock(InstanceMetadata.class); + File cdcDir = Files.createDirectory(tempDir.resolve(CdcRawDirectorySpaceCleaner.CDC_DIR_NAME)).toFile(); + writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_1, 4096, true); + writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_2, 4096, true); + writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_3, 4096, true); + when(instanceMetadata.dataDirs()).thenReturn(List.of(cdcDir.getParent() + "/data")); + when(instanceMetadata.cdcDir()).thenReturn(cdcDir.getParent() + "/cdc_raw"); + InstancesMetadata instancesMetadata = new InstancesMetadataImpl(instanceMetadata, DnsResolvers.DEFAULT); + + FakeTimeProvider timeProvider = new FakeTimeProvider(); + long nowMillis = (System.currentTimeMillis() / 1000) * 1000; + timeProvider.advance(nowMillis, TimeUnit.MILLISECONDS); + + // Set the oldest segment's mtime to 0 to simulate a concurrent reclamation/rotation whose + // timestamp is unavailable by the time the sidecar reads it in CdcRawSegmentFile's constructor. + File oldest = new File(cdcDir, TEST_SEGMENT_FILE_NAME_1); + File oldestIdx = new File(cdcDir, CdcUtil.getIdxFileName(TEST_SEGMENT_FILE_NAME_1)); + assumeTrue(oldest.setLastModified(0L) && oldest.lastModified() == 0L, + "Filesystem does not support a 0 mtime; skipping guard assertion"); + + CdcRawDirectorySpaceCleaner cleaner = new CdcRawDirectorySpaceCleaner( + timeProvider, + systemViewsDatabaseAccessor, + serviceConfiguration, + instancesMetadata, + mockSidecarMetrics + ); + + // Drain any residual accumulation the DeltaGauges may carry from a prior test that + // shared the static MetricRegistry — getValue() atomically reads and resets to 0. + cdcMetrics.deletedSegment.metric.getValue(); + + cleaner.routineCleanUp(); + + // Guard must have caused the deletion loop to skip calling deleteSegment() on the phantom. + // Pre-fix, (now - 0) collapsed to ~10^12 ms which is neither < critical nor < low, so no + // alert fired, but deleteSegment() still ran and Files.deleteIfExists removed both files. + assertThat(oldest) + .as("phantom segment's log file must not be deleted by the cleaner when its timestamp is unavailable") + .exists(); + assertThat(oldestIdx) + .as("phantom segment's idx file must not be deleted by the cleaner when its timestamp is unavailable") + .exists(); + + // deletedSegment must count only bytes the cleaner actually deleted (seg 2). Pre-fix it + // would have included seg 1's bytes as well, roughly doubling the value. + assertThat(cdcMetrics.deletedSegment.metric.getValue()) + .as("deletedSegment metric must count only actually-deleted segments, not the phantom") + .isLessThan(2L * 4096L); + } + @Test void testMaxUsageBytes() {