CASSSIDECAR-220: Add CdcRawDirectorySpaceCleaner for tracking and cleaning up the cdc_…

server/src/main/java/org/apache/cassandra/sidecar/config/CdcConfiguration.java

@@ -28,4 +28,41 @@ public interface CdcConfiguration
      * @return segment hard link cache expiration time used in {@link org.apache.cassandra.sidecar.cdc.CdcLogCache}
      */
     SecondBoundConfiguration segmentHardLinkCacheExpiry();
+
+    /* CdcRawDirectorySpaceCleaner Configuration */
+
+    /**
+     * @return the cadence at which the CdcRawDirectorySpaceCleaner period task should run to check and clean-up old `cdc_raw` log segments.
+     */
+    SecondBoundConfiguration cdcRawDirectorySpaceCleanerFrequency();
+
+    /**
+     * @return `true` if CdcRawDirectorySpaceCleaner should monitor the `cdc_raw` directory and clean up the oldest commit log segments.
+     */
+    boolean enableCdcRawDirectoryRoutineCleanUp();
+
+    /**
+     * @return fallback value for maximum directory size in bytes for the `cdc_raw` directory when can't be read from `system_views.settings` table.
+     */
+    long fallbackCdcRawDirectoryMaxSizeBytes();
+
+    /**
+     * @return max percent usage of the cdc_raw directory before CdcRawDirectorySpaceCleaner starts removing the oldest segments.
+     */
+    float cdcRawDirectoryMaxPercentUsage();
+
+    /**
+     * @return the critical time period in seconds that indicates the `cdc_raw` directory is not large enough to buffer this time-window of mutations.
+     */
+    SecondBoundConfiguration cdcRawDirectoryCriticalBufferWindow();
+
+    /**
+     * @return the low time period in seconds that indicates the `cdc_raw` directory is not large enough to buffer this time-window of mutations.
+     */
+    SecondBoundConfiguration cdcRawDirectoryLowBufferWindow();
+
+    /**
+     * @return the time period which the CdcRawDirectorySpaceCleaner should cache the cdc_total_space before refreshing.
+     */
+    SecondBoundConfiguration cacheMaxUsage();
 }

server/src/main/java/org/apache/cassandra/sidecar/config/yaml/CdcConfigurationImpl.java

@@ -36,16 +36,81 @@ public class CdcConfigurationImpl implements CdcConfiguration
     public static final SecondBoundConfiguration DEFAULT_SEGMENT_HARD_LINK_CACHE_EXPIRY =
     SecondBoundConfiguration.parse("5m");
 
+    public static final String CDC_RAW_CLEANER_FREQUENCY_PROPERTY = "cdc_raw_cleaner_frequency";
+    public static final SecondBoundConfiguration DEFAULT_CDC_RAW_CLEANER_FREQUENCY =
+    SecondBoundConfiguration.parse("1m");
+
+    public static final String ENABLE_CDC_RAW_CLEANER_PROPERTY = "enable_cdc_raw_cleaner";
+    public static final boolean DEFAULT_ENABLE_CDC_RAW_CLEANER_PROPERTY = true;
+
+    public static final String FALLBACK_CDC_RAW_MAX_DIRECTORY_SIZE_BYTES = "fallback_cdc_raw_max_directory_size_bytes";
+    public static final long DEFAULT_FALLBACK_CDC_RAW_MAX_DIRECTORY_SIZE_BYTES = 1L << 31; // 2 GiB
+
+    public static final String CDC_RAW_MAX_DIRECTORY_MAX_PERCENT = "cdc_raw_max_directory_max_percent";
+    public static final float DEFAULT_CDC_RAW_MAX_DIRECTORY_MAX_PERCENT = 1.0f;
+
+    public static final String CDC_RAW_MAX_CRITICAL_BUFFER_WINDOW = "cdc_raw_critical_buffer_window";
+    public static final SecondBoundConfiguration DEFAULT_CDC_RAW_MAX_CRITICAL_BUFFER_WINDOW = SecondBoundConfiguration.parse("15m");
+
+    public static final String CDC_RAW_MAX_LOW_BUFFER_WINDOW = "cdc_raw_low_buffer_window";
+    public static final SecondBoundConfiguration DEFAULT_CDC_RAW_MAX_LOW_BUFFER_WINDOW = SecondBoundConfiguration.parse("60m");
+
+    public static final String CDC_CACHE_MAX_USAGE_DURATION = "cdc_raw_cache_max_usage_duration";
+    public static final SecondBoundConfiguration DEFAULT_CDC_CACHE_MAX_USAGE_DURATION = SecondBoundConfiguration.parse("15m");
+
+
     protected SecondBoundConfiguration segmentHardLinkCacheExpiry;
+    protected SecondBoundConfiguration cdcRawCleanerFrequency;
+    protected boolean enableCdcRawCleaner;
+    protected long fallbackCdcRawMaxDirectorySize;
+    protected float cdcRawMaxPercent;
+    protected SecondBoundConfiguration cdcRawCriticalBufferWindow;
+    protected SecondBoundConfiguration cdcRawLowBufferWindow;
+    protected SecondBoundConfiguration cacheMaxUsage;
 
     public CdcConfigurationImpl()
     {
         this.segmentHardLinkCacheExpiry = DEFAULT_SEGMENT_HARD_LINK_CACHE_EXPIRY;
+        this.cdcRawCleanerFrequency = DEFAULT_CDC_RAW_CLEANER_FREQUENCY;
+        this.enableCdcRawCleaner = DEFAULT_ENABLE_CDC_RAW_CLEANER_PROPERTY;
+        this.fallbackCdcRawMaxDirectorySize = DEFAULT_FALLBACK_CDC_RAW_MAX_DIRECTORY_SIZE_BYTES;
+        this.cdcRawMaxPercent = DEFAULT_CDC_RAW_MAX_DIRECTORY_MAX_PERCENT;
+        this.cdcRawCriticalBufferWindow = DEFAULT_CDC_RAW_MAX_CRITICAL_BUFFER_WINDOW;
+        this.cdcRawLowBufferWindow = DEFAULT_CDC_RAW_MAX_LOW_BUFFER_WINDOW;
+        this.cacheMaxUsage = DEFAULT_CDC_CACHE_MAX_USAGE_DURATION;
     }
 
     public CdcConfigurationImpl(SecondBoundConfiguration segmentHardLinkCacheExpiry)
+    {
+        this(
+        segmentHardLinkCacheExpiry,
+        DEFAULT_CDC_RAW_CLEANER_FREQUENCY,
+        DEFAULT_ENABLE_CDC_RAW_CLEANER_PROPERTY,
+        DEFAULT_FALLBACK_CDC_RAW_MAX_DIRECTORY_SIZE_BYTES,
+        DEFAULT_CDC_RAW_MAX_DIRECTORY_MAX_PERCENT,
+        DEFAULT_CDC_RAW_MAX_CRITICAL_BUFFER_WINDOW,
+        DEFAULT_CDC_RAW_MAX_LOW_BUFFER_WINDOW,
+        DEFAULT_CDC_CACHE_MAX_USAGE_DURATION
+        );
+    }
+
+    public CdcConfigurationImpl(SecondBoundConfiguration segmentHardLinkCacheExpiry,
+                                SecondBoundConfiguration cdcRawCleanerFrequency,
+                                boolean enableCdcRawCleaner,
+                                long cdcRawMaxDirectorySize,
+                                float cdcRawMaxPercent,
+                                SecondBoundConfiguration cdcRawCriticalBufferWindow,
+                                SecondBoundConfiguration cdcRawLowBufferWindow,
+                                SecondBoundConfiguration cacheMaxUsage)
     {
         this.segmentHardLinkCacheExpiry = segmentHardLinkCacheExpiry;
+        this.cdcRawCleanerFrequency = cdcRawCleanerFrequency;
+        this.enableCdcRawCleaner = enableCdcRawCleaner;
+        this.fallbackCdcRawMaxDirectorySize = cdcRawMaxDirectorySize;
+        this.cdcRawMaxPercent = cdcRawMaxPercent;
+        this.cdcRawCriticalBufferWindow = cdcRawCriticalBufferWindow;
+        this.cdcRawLowBufferWindow = cdcRawLowBufferWindow;
+        this.cacheMaxUsage = cacheMaxUsage;
     }
 
     @Override
@@ -74,4 +139,117 @@ public void setSegmentHardLinkCacheExpiryInSecs(long segmentHardlinkCacheExpiryI
         LOGGER.warn("'segment_hardlink_cache_expiry_in_secs' is deprecated, use 'segment_hardlink_cache_expiry' instead");
         setSegmentHardLinkCacheExpiry(new SecondBoundConfiguration(segmentHardlinkCacheExpiryInSecs, TimeUnit.SECONDS));
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonProperty(value = CDC_RAW_CLEANER_FREQUENCY_PROPERTY)
+    public SecondBoundConfiguration cdcRawDirectorySpaceCleanerFrequency()
+    {
+        return cdcRawCleanerFrequency;
+    }
+
+    @JsonProperty(value = CDC_RAW_CLEANER_FREQUENCY_PROPERTY)
+    public void setCdcRawDirectorySpaceCleanerFrequency(SecondBoundConfiguration cdcRawCleanerFrequency)
+    {
+        this.cdcRawCleanerFrequency = cdcRawCleanerFrequency;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonProperty(value = ENABLE_CDC_RAW_CLEANER_PROPERTY)
+    public boolean enableCdcRawDirectoryRoutineCleanUp()
+    {
+        return enableCdcRawCleaner;
+    }
+
+    @JsonProperty(value = ENABLE_CDC_RAW_CLEANER_PROPERTY)
+    public void setEnableCdcRawCleaner(boolean enableCdcRawCleaner)
+    {
+        this.enableCdcRawCleaner = enableCdcRawCleaner;
+    }
+
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonProperty(value = FALLBACK_CDC_RAW_MAX_DIRECTORY_SIZE_BYTES)
+    public long fallbackCdcRawDirectoryMaxSizeBytes()
+    {
+        return fallbackCdcRawMaxDirectorySize;
+    }
+
+    @JsonProperty(value = FALLBACK_CDC_RAW_MAX_DIRECTORY_SIZE_BYTES)
+    public void setFallbackCdcRawDirectoryMaxSizeBytes(long fallbackCdcRawMaxDirectorySize)
+    {
+        this.fallbackCdcRawMaxDirectorySize = fallbackCdcRawMaxDirectorySize;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonProperty(value = CDC_RAW_MAX_DIRECTORY_MAX_PERCENT)
+    public float cdcRawDirectoryMaxPercentUsage()
+    {
+        return cdcRawMaxPercent;
+    }
+
+    @JsonProperty(value = CDC_RAW_MAX_DIRECTORY_MAX_PERCENT)
+    public void setCdcRawDirectoryMaxPercentUsage(long cdcRawMaxPercent)
+    {
+        this.cdcRawMaxPercent = cdcRawMaxPercent;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonProperty(value = CDC_RAW_MAX_CRITICAL_BUFFER_WINDOW)
+    public SecondBoundConfiguration cdcRawDirectoryCriticalBufferWindow()
+    {
+        return cdcRawCriticalBufferWindow;
+    }
+
+    @JsonProperty(value = CDC_RAW_MAX_CRITICAL_BUFFER_WINDOW)
+    public void setCdcRawDirectoryCriticalBufferWindow(SecondBoundConfiguration cdcRawCriticalBufferWindow)
+    {
+        this.cdcRawCriticalBufferWindow = cdcRawCriticalBufferWindow;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonProperty(value = CDC_RAW_MAX_LOW_BUFFER_WINDOW)
+    public SecondBoundConfiguration cdcRawDirectoryLowBufferWindow()
+    {
+        return cdcRawLowBufferWindow;
+    }
+
+    @JsonProperty(value = CDC_RAW_MAX_LOW_BUFFER_WINDOW)
+    public void setCdcRawDirectoryLowBufferWindow(SecondBoundConfiguration cdcRawLowBufferWindow)
+    {
+        this.cdcRawLowBufferWindow = cdcRawLowBufferWindow;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @JsonProperty(value = CDC_CACHE_MAX_USAGE_DURATION)
+    @Override
+    public SecondBoundConfiguration cacheMaxUsage()
+    {
+        return cacheMaxUsage;
+    }
+
+    @JsonProperty(value = CDC_CACHE_MAX_USAGE_DURATION)
+    public void setCacheMaxUsage(SecondBoundConfiguration cacheMaxUsage)
+    {
+        this.cacheMaxUsage = cacheMaxUsage;
+    }
 }

server/src/main/java/org/apache/cassandra/sidecar/db/SystemViewsDatabaseAccessor.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.db;
+
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
+import org.apache.cassandra.sidecar.db.schema.SystemViewsSchema;
+import org.apache.cassandra.sidecar.exceptions.SchemaUnavailableException;
+import org.apache.cassandra.sidecar.utils.FileUtils;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Database Accessor that queries cassandra to get information maintained under system_auth keyspace.
+ */
+@Singleton
+public class SystemViewsDatabaseAccessor extends DatabaseAccessor<SystemViewsSchema>
+{
+    static final String YAML_PROP_PREVIOUS = "cdc_total_space_in_mb";
+    static final String YAML_PROP_CURRENT = "cdc_total_space";
+
+    @Inject
+    public SystemViewsDatabaseAccessor(SystemViewsSchema systemViewsSchema,
+                                       CQLSessionProvider sessionProvider)
+    {
+        super(systemViewsSchema, sessionProvider);
+    }
+
+    @Nullable
+    public Long getCdcTotalSpaceSetting() throws SchemaUnavailableException
+    {
+        // attempt to parse Cassandra v4.0 'cdc_total_space_in_mb' yaml prop
+        String cdcTotalSpaceInMb = getSetting(YAML_PROP_PREVIOUS);
+        if (cdcTotalSpaceInMb != null)
+        {
+            return FileUtils.mbStringToBytes(cdcTotalSpaceInMb);
+        }
+
+        // otherwise parse current (v5.0+) 'cdc_total_space' yaml prop
+        String storageStringToBytes = getSetting(YAML_PROP_CURRENT);
+        if (storageStringToBytes != null)
+        {
+            return FileUtils.storageStringToBytes(storageStringToBytes);
+        }
+
+        return null;
+    }
+
+    /**
+     * Load a setting value from the `system_views.settings` table.
+     *
+     * @param name name of setting
+     * @return setting value for a given `name` loaded from the `system_views.settings` table.
+     */
+    @Nullable
+    public String getSetting(String name) throws SchemaUnavailableException
+    {
+        BoundStatement statement = tableSchema.selectSettings().bind(name);
+        ResultSet result = execute(statement);
+        Row row = result.one();
+        return row != null && !row.isNull(0) ? row.getString(0) : null;
+    }
+}

server/src/main/java/org/apache/cassandra/sidecar/db/schema/SystemViewsSchema.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.db.schema;
+
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.exceptions.SchemaUnavailableException;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Schema for getting information stored in system_views keyspace.
+ */
+@Singleton
+public class SystemViewsSchema extends CassandraSystemTableSchema
+{
+    protected static final String SYSTEM_VIEWS_KEYSPACE_NAME = "system_views";
+    protected static final String SYSTEM_VIEWS_SETTINGS_TABLE_NAME = "settings";
+    private PreparedStatement selectSettings;
+
+    @Override
+    protected String keyspaceName()
+    {
+        return SYSTEM_VIEWS_KEYSPACE_NAME;
+    }
+
+    @Override
+    protected String tableName()
+    {
+        return SYSTEM_VIEWS_SETTINGS_TABLE_NAME;
+    }
+
+    public PreparedStatement selectSettings() throws SchemaUnavailableException
+    {
+        ensureSchemaAvailable();
+        return selectSettings;
+    }
+
+    @Override
+    protected void prepareStatements(@NotNull Session session)
+    {
+        this.selectSettings = session.prepare("SELECT value FROM system_views.settings WHERE name = ?");
+    }
+
+    protected void ensureSchemaAvailable() throws SchemaUnavailableException
+    {
+        if (selectSettings == null)
+        {
+            throw new SchemaUnavailableException(keyspaceName(), tableName());
+        }
+    }
+}