(WIP) CDC reader in Apache Cassandra Sidecar - CASSANDRASC-27

build.gradle

@@ -179,6 +179,7 @@ dependencies {
 
     compile project(":common")
     compile project(":cassandra40")
+    compile 'org.apache.cassandra:cassandra-all:4.0-beta2'
 }
 
 task copyCodeStyle(type: Copy) {

cassandra-integration-tests/src/test/java/org/apache/cassandra/sidecar/common/testing/CassandraPod.java

@@ -193,7 +193,8 @@ public void start() throws ApiException, InterruptedException, CassandraPodExcep
             }
 
             started = namespacedPod.getStatus().getContainerStatuses().get(0).getStarted();
-            if (namespacedPod.getStatus().getContainerStatuses().get(0).getReady() && started) {
+            if (namespacedPod.getStatus().getContainerStatuses().get(0).getReady() && started)
+            {
                 logger.info("Pod startup OK");
                 break;
             }

src/main/dist/conf/sidecar.yaml

@@ -24,3 +24,6 @@ sidecar:
 
 healthcheck:
   - poll_freq_millis: 30000
+
+cdc:
+  - configPath: file:////etc/cassandra/conf/cassandra.yaml

src/main/java/org/apache/cassandra/sidecar/CassandraSidecarDaemon.java

@@ -27,8 +27,8 @@
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import io.vertx.core.http.HttpServer;
+import org.apache.cassandra.sidecar.cdc.CDCReaderService;
 import org.apache.cassandra.sidecar.utils.SslUtils;
-
 /**
  * Main class for initiating the Cassandra sidecar
  */
@@ -38,19 +38,22 @@ public class CassandraSidecarDaemon
     private static final Logger logger = LoggerFactory.getLogger(CassandraSidecarDaemon.class);
     private final HttpServer server;
     private final Configuration config;
+    private final CDCReaderService cdcReaderService;
 
     @Inject
-    public CassandraSidecarDaemon(HttpServer server, Configuration config)
+    public CassandraSidecarDaemon(HttpServer server, Configuration config, CDCReaderService cdcReaderService)
     {
         this.server = server;
         this.config = config;
+        this.cdcReaderService = cdcReaderService;
     }
 
     public void start()
     {
         banner(System.out);
         validate();
         logger.info("Starting Cassandra Sidecar on {}:{}", config.getHost(), config.getPort());
+        cdcReaderService.start();
         server.listen(config.getPort(), config.getHost());
     }
 

src/main/java/org/apache/cassandra/sidecar/Configuration.java

@@ -55,12 +55,17 @@ public class Configuration
 
     private final boolean isSslEnabled;
 
+    /* Cassandra server conf path */
+    @Nullable
+    private String cassandraConfigPath;
+
     public Configuration(String cassandraHost, Integer cassandraPort, String host, Integer port,
                          Integer healthCheckFrequencyMillis, boolean isSslEnabled,
                          @Nullable String keyStorePath,
                          @Nullable String keyStorePassword,
                          @Nullable String trustStorePath,
-                         @Nullable String trustStorePassword)
+                         @Nullable String trustStorePassword,
+                         @Nullable String cassandraConfigPath)
     {
         this.cassandraHost = cassandraHost;
         this.cassandraPort = cassandraPort;
@@ -73,6 +78,8 @@ public Configuration(String cassandraHost, Integer cassandraPort, String host, I
         this.trustStorePath = trustStorePath;
         this.trustStorePassword = trustStorePassword;
         this.isSslEnabled = isSslEnabled;
+
+        this.cassandraConfigPath = cassandraConfigPath;
     }
 
     /**
@@ -179,6 +186,15 @@ public String getTruststorePassword()
         return trustStorePassword;
     }
 
+    /**
+     * Get path of the Cassandra configuration file
+     */
+    @Nullable
+    public String getCassandraConfigPath()
+    {
+        return cassandraConfigPath;
+    }
+
     /**
      * Configuration Builder
      */
@@ -194,6 +210,7 @@ public static class Builder
         private String trustStorePath;
         private String trustStorePassword;
         private boolean isSslEnabled;
+        private String cassandraConfigPath;
 
         public Builder setCassandraHost(String host)
         {
@@ -255,10 +272,17 @@ public Builder setSslEnabled(boolean enabled)
             return this;
         }
 
+        public Builder setCassandraConfigPath(String configPath)
+        {
+            this.cassandraConfigPath = configPath;
+            return  this;
+        }
+
         public Configuration build()
         {
             return new Configuration(cassandraHost, cassandraPort, host, port, healthCheckFrequencyMillis, isSslEnabled,
-                                     keyStorePath, keyStorePassword, trustStorePath, trustStorePassword);
+                                     keyStorePath, keyStorePassword, trustStorePath, trustStorePassword,
+                                     cassandraConfigPath);
         }
     }
 }

src/main/java/org/apache/cassandra/sidecar/MainModule.java

@@ -41,9 +41,13 @@
 import io.vertx.ext.web.handler.LoggerHandler;
 import io.vertx.ext.web.handler.StaticHandler;
 import org.apache.cassandra.sidecar.cassandra40.Cassandra40Factory;
+import org.apache.cassandra.sidecar.cdc.output.ConsoleOutput;
+import org.apache.cassandra.sidecar.cdc.output.Output;
 import org.apache.cassandra.sidecar.common.CQLSession;
 import org.apache.cassandra.sidecar.common.CassandraAdapterDelegate;
 import org.apache.cassandra.sidecar.common.CassandraVersionProvider;
+import org.apache.cassandra.sidecar.metrics.cdc.CDCReaderMonitor;
+import org.apache.cassandra.sidecar.metrics.cdc.CDCReaderMonitorLogger;
 import org.apache.cassandra.sidecar.routes.HealthService;
 import org.apache.cassandra.sidecar.routes.SwaggerOpenApiResource;
 import org.jboss.resteasy.plugins.server.vertx.VertxRegistry;
@@ -126,6 +130,14 @@ public Router vertxRouter(Vertx vertx)
         return router;
     }
 
+    @Override
+    protected void  configure()
+    {
+        // TODO: Make the output type configurable
+        bind(CDCReaderMonitor.class).to(CDCReaderMonitorLogger.class);
+        bind(Output.class).to(ConsoleOutput.class);
+    }
+
     @Provides
     @Singleton
     public Configuration configuration() throws ConfigurationException, IOException
@@ -151,6 +163,7 @@ public Configuration configuration() throws ConfigurationException, IOException
                     .setTrustStorePath(yamlConf.get(String.class, "sidecar.ssl.truststore.path", null))
                     .setTrustStorePassword(yamlConf.get(String.class, "sidecar.ssl.truststore.password", null))
                     .setSslEnabled(yamlConf.get(Boolean.class, "sidecar.ssl.enabled", false))
+                    .setCassandraConfigPath(yamlConf.get(String.class, "cdc.configPath"))
                     .build();
         }
         catch (MalformedURLException e)

src/main/java/org/apache/cassandra/sidecar/cdc/CDCBookmark.java

@@ -0,0 +1,231 @@
+/*
+ * 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.cdc;
+
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.commitlog.CommitLogDescriptor;
+import org.apache.cassandra.db.commitlog.CommitLogPosition;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.sidecar.Configuration;
+
+/**
+ * Manages the CDC reader bookmark. This tracks the last successfully processed offset
+ * of a commit log.
+ */
+@Singleton
+public class CDCBookmark extends TimerTask
+{
+    // Tracks last disk sync'd commit log position.
+    private CommitLogPosition lastSyncedPosition;
+    // Tracks last successfully processed commit log position by the CDC reader.
+    private CommitLogPosition lastProcessedPosition;
+    private final Timer timer;
+    private static final String BOOKMARK = "CdcReader.bookmark";
+    private static final Logger logger = LoggerFactory.getLogger(CDCBookmark.class);
+    private final ReentrantLock bookmarkLock = new ReentrantLock();
+    private final Configuration conf;
+
+    @Inject
+    CDCBookmark(Configuration conf)
+    {
+        this.lastSyncedPosition = null;
+        this.lastProcessedPosition = null;
+        this.conf = conf;
+        this.timer = new Timer();
+    }
+
+    /**
+     * Persists last successfully processed commit log offset to the disk.
+     */
+    public void syncBookmark()
+    {
+        CommitLogPosition lastPosition = this.getLastProcessedPosition();
+
+        if (lastPosition == null)
+        {
+            return;
+        }
+        logger.debug("Last processed bookmark {}", this.lastProcessedPosition.toString());
+        try
+        {
+            if (lastPosition.equals(this.lastSyncedPosition))
+            {
+                return;
+            }
+
+            CommitLogPosition.CommitLogPositionSerializer serializer =
+                    new CommitLogPosition.CommitLogPositionSerializer();
+
+            // TODO: JSON ser-de and write-rename instead of writing directly to the bookmark
+            try (FileOutputStream fileOutputStream = new FileOutputStream(
+                    new File(this.getBookmarkPath())))
+            {
+                DataOutputPlus outBuffer = new DataOutputBuffer();
+                serializer.serialize(lastPosition, outBuffer);
+                fileOutputStream.write(((DataOutputBuffer) outBuffer).getData());
+                fileOutputStream.flush();
+                this.lastSyncedPosition = lastPosition;
+                logger.info("Successfully synced bookmark {} to the file {}", this.lastSyncedPosition.toString(),
+                        this.getBookmarkPath());
+            }
+            catch (IOException e)
+            {
+                logger.error("Error when writing bookmark {} to the file {}", lastPosition.toString(),
+                        this.getBookmarkPath());
+            }
+        }
+        catch (Exception ex)
+        {
+            logger.error("Sync exception {}", ex.getMessage());
+        }
+    }
+
+    /**
+     * Gets the path to the CDC reader bookmark.
+     *
+     * @return complete path to the bookmark file.
+     */
+    public String getBookmarkPath()
+    {
+        return String.format("%s/%s", DatabaseDescriptor.getCDCLogLocation(),
+                BOOKMARK);
+    }
+
+    @Override
+    public void run()
+    {
+        this.syncBookmark();
+    }
+
+    /**
+     * Gets the last successfully processed commit log offset.
+     * This method is thread safe.
+     *
+     * @return last successfully processed commit log offset.
+     */
+    public CommitLogPosition getLastProcessedPosition()
+    {
+        CommitLogPosition lastPosition = null;
+        try
+        {
+            bookmarkLock.lock();
+            if (this.lastProcessedPosition != null)
+            {
+                lastPosition = new CommitLogPosition(this.lastProcessedPosition.segmentId,
+                        this.lastProcessedPosition.position);
+
+            }
+        }
+        finally
+        {
+            bookmarkLock.unlock();
+        }
+        return lastPosition;
+    }
+
+    /**
+     * Sets the last successfully processed commit log offset.
+     * This method is thread safe.
+     *
+     */
+    public void setLastProcessedPosition(CommitLogPosition processedPosition)
+    {
+        try
+        {
+            bookmarkLock.lock();
+            this.lastProcessedPosition = processedPosition;
+        }
+        finally
+        {
+            bookmarkLock.unlock();
+        }
+    }
+
+    /**
+     * Starts the background thread to write processed commit log positions to the disk.
+     * */
+    public void startBookmarkSync()
+    {
+        timer.schedule(this, 0, DatabaseDescriptor.getCommitLogSyncPeriod());
+    }
+
+    /**
+     * Gets the persisted commit log offset from the bookmark on the disk.
+     *
+     * @return persisted commit log offset.
+     */
+    public CommitLogPosition getPersistedBookmark()
+    {
+        CommitLogPosition.CommitLogPositionSerializer serializer =
+                new CommitLogPosition.CommitLogPositionSerializer();
+        try (FileInputStream fileInputStream =
+                     new FileInputStream(new File(this.getBookmarkPath())))
+        {
+            DataInputPlus inputBuffer = new DataInputPlus.DataInputStreamPlus(fileInputStream);
+            return serializer.deserialize(inputBuffer);
+        }
+        catch (IOException ex)
+        {
+            logger.error("Error when reading the saved bookmark {}", this.getBookmarkPath());
+            return null;
+        }
+    }
+
+    /**
+     * Checks whether there's a valid persisted bookmark.
+     *
+     * @return whether there's a valid bookmark.
+     */
+    public boolean isValidBookmark()
+    {
+        CommitLogPosition bookmark = getPersistedBookmark();
+        if (bookmark == null)
+        {
+            return false;
+        }
+        // It's fine for compression to be null as we are not using this CommitLogDescriptor to write commit logs.
+        CommitLogDescriptor commitLogDescriptor = new CommitLogDescriptor(bookmark.segmentId, null,
+                null);
+
+        if (commitLogDescriptor == null ||
+                !Paths.get(DatabaseDescriptor.getCDCLogLocation(),
+                        commitLogDescriptor.cdcIndexFileName()).toFile().exists() ||
+                !Paths.get(DatabaseDescriptor.getCDCLogLocation(),
+                        commitLogDescriptor.fileName()).toFile().exists())
+        {
+            return false;
+        }
+        return true;
+    }
+}