Merge pull request #20 from retail-ai-inc/feature/redis-support

feat: add support for Redis integration

Author: tanvir-retailai

.github/workflows/codecov.yml

@@ -72,7 +72,8 @@ jobs:
             echo "Waiting for MongoDB..."
             sleep 1
           done
-
+          
+          sleep 5
           echo "All services are up!"
         
       - name: Run tests
@@ -87,4 +88,4 @@ jobs:
       - name: Tear down Docker Compose
         if: always()
         run: |
-          docker-compose -f docker/docker-compose.yml down
+          docker-compose -f docker/docker-compose.yml down

README.md

@@ -3,7 +3,7 @@
 Synchronize Production NOSQL and SQL data to Standalone instances for Data scientists or other purposes. A **Go-based** tool to synchronize MongoDB or SQL data from a **MongoDB replica set** or **sharded cluster** or production SQL instance to a **standalone instance**, supports initial and incremental synchronization with change stream monitoring.
 
 > [!NOTE]
-> Sync is now supporting MongoDB, MySQl, PostgreSQL and MariaDB. Next `Sync` will support Redis and Elasticsearch.
+> Sync is now supporting MongoDB, MySQL, PostgreSQL, MariaDB, and Redis. Next, `Sync` will support Elasticsearch.
 
 ## What is the problem
 Let’s assume you have a mid to big-size SaaS platform or service with multiple tech teams and stakeholders. Different teams have different requirements for analyzing the production data independently. However, the tech team doesn’t want to allow all these stakeholders direct access to the production databases due to security and stability issues.
@@ -16,7 +16,8 @@ Create standalone databases outside of your production database servers with the
 - MongoDB (Sharded clusters, Replica sets)
 - MySQL
 - MariaDB 
-- PostgreSQL (PostgreSQL version 10+、Enable logical replication)
+- PostgreSQL (PostgreSQL version 10+ with logical replication enabled)
+- Redis (Standalone, Sentinel; does not support cluster mode)
 
 ## High Level Design Diagram
 
@@ -31,7 +32,7 @@ Create standalone databases outside of your production database servers with the
 
 ![image](https://github.com/user-attachments/assets/65b23a4c-56db-4833-89a1-0f802af878bd)
 
-### Grafana Integration(temporary transition phase)
+### Grafana Integration (temporary transition phase)
 ![image](https://github.com/user-attachments/assets/cdc8e57b-8aa4-4386-8aa8-de5028698fd0)
 
 
@@ -41,14 +42,17 @@ Create standalone databases outside of your production database servers with the
   - MongoDB: Bulk synchronization of data from the MongoDB cluster or MongoDB replica set to the standalone MongoDB instance.
   - MySQL/MariaDB: Initial synchronization using batch inserts (default batch size: 100 rows) from the source to the target if the target table is empty.
   - PostgreSQL: Initial synchronization using batch inserts (default batch size: 100 rows) from the source to the target using logical replication slots and the pgoutput plugin.
-- **Change Stream & Binlog Monitoring**:
+  - Redis: Supports full data synchronization for standalone Redis and Sentinel setups using Redis Streams and Keyspace Notifications.
+- **Change Stream & Incremental Updates**:
   - MongoDB: Watches for real-time changes (insert, update, replace, delete) in the cluster's collections and reflects them in the standalone instance.
   - MySQL/MariaDB: Uses binlog replication events to capture and apply incremental changes to the target.
   - PostgreSQL: Uses WAL (Write-Ahead Log) with the pgoutput plugin to capture and apply incremental changes to the target.
+  - Redis: Uses Redis Streams and Keyspace Notifications to capture and sync incremental changes in real-time.
 - **Batch Processing & Concurrency**:  
   Handles synchronization in batches for optimized performance and supports parallel synchronization for multiple collections/tables.
 - **Restart Resilience**: 
-  Stores MongoDB resume tokens, MySQL binlog positions, and PostgreSQL replication positions in configurable state files, allowing the tool to resume synchronization from the last known position after a restart.
+  Stores MongoDB resume tokens, MySQL binlog positions, PostgreSQL replication positions, and Redis stream offsets in configurable state files, allowing the tool to resume synchronization from the last known position after a restart.
+  - **Note for Redis**: Redis does not support resuming from the last state after a sync interruption. If `Sync` is interrupted or crashes, it will restart the synchronization process by executing the initial sync method to retrieve all keys and sync them to the target database. This is due to limitations in Redis Streams and Keyspace Notifications, which do not provide a built-in mechanism to persist and resume stream offsets across restarts. As a result, the tool cannot accurately determine the last synced state and must perform a full resync to ensure data consistency.
 - **Grafana Integration**:
   - For data visualization, this tool integrates with **Grafana** using data from **GCP Logging** and **GCP BigQuery**.
   - When **`enable_table_row_count_monitoring`** is enabled, the tool records data changes, including table row counts, in **GCP Logging**.
@@ -65,6 +69,10 @@ Create standalone databases outside of your production database servers with the
 - For PostgreSQL sources:
   - A PostgreSQL instance with logical replication enabled and a replication slot created.
   - A target PostgreSQL instance with write permissions.
+- For Redis sources:
+  - Redis standalone or Sentinel setup with Redis version >= 5.0.
+  - Redis Streams and Keyspace Notifications enabled.
+  - A target Redis instance with write permissions.
 
 ## Quick start
 
@@ -169,21 +177,36 @@ sync_configs:
         tables:
           - source_table: "users"
             target_table: "users"
+
+  - type: "redis"
+    enable: true
+    source_connection: "redis://localhost:6379/0"
+    target_connection: "redis://localhost:6379/1"
+    redis_position_path: "/tmp/state/redis_position"
+    mappings:
+      - source_database: "db0"
+        target_database: "db1"
+        tables:
+          - source_table: "source_stream"  # Redis Stream Name
+            target_table: "" 
 ```
 
 ## Real-Time Synchronization
 
 - MongoDB: Uses Change Streams from replica sets or sharded clusters for incremental updates.
 - MySQL/MariaDB: Uses binlog replication to apply incremental changes to the target.
 - PostgreSQL: Uses WAL (Write-Ahead Log) with the pgoutput plugin to apply incremental changes to the target.
+- Redis: Uses Redis Streams and Keyspace Notifications to sync changes in real-time.
+  - **Note for Redis**: If `Sync` is interrupted, Redis will restart the synchronization process with an initial sync of all keys to the target. This ensures data consistency but may increase synchronization time after interruptions.
 
 On the restart, the tool resumes from the stored state (resume token for MongoDB, binlog position for MySQL/MariaDB, replication slot for PostgreSQL).
 
 ## Availability  
 
 - MongoDB: MongoDB Change Streams require a replica set or sharded cluster. See [Convert Standalone to Replica Set](https://www.mongodb.com/docs/manual/tutorial/convert-standalone-to-replica-set/).
 - MySQL/MariaDB: MySQL/MariaDB binlog-based incremental sync requires ROW or MIXED binlog format for proper event capturing.
-- PostgreSQL incremental sync requires logical replication enabled with a replication slot.
+- PostgreSQL: PostgreSQL incremental sync requires logical replication enabled with a replication slot.
+- Redis: Redis sync supports standalone and Sentinel setups but does not support Redis Cluster mode. Redis does not support resuming from the last synced state after a crash or interruption.
 
 ## Contributing
 

cmd/sync/main.go

@@ -2,11 +2,8 @@ package main
 
 import (
 	"context"
-	"database/sql"
-	"fmt"
 	"os"
 	"os/signal"
-	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -16,9 +13,7 @@ import (
 	"github.com/retail-ai-inc/sync/pkg/config"
 	"github.com/retail-ai-inc/sync/pkg/logger"
 	"github.com/retail-ai-inc/sync/pkg/syncer"
-	"github.com/sirupsen/logrus"
-	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
+	"github.com/retail-ai-inc/sync/pkg/utils"
 )
 
 // Interval for row count monitoring every minute
@@ -74,6 +69,12 @@ func main() {
 				syncer := syncer.NewPostgreSQLSyncer(syncCfg, log)
 				syncer.Start(ctx)
 			}(syncCfg)
+		case "redis":
+			go func(syncCfg config.SyncConfig) {
+				defer wg.Done()
+				syncer := syncer.NewRedisSyncer(syncCfg, log)
+				syncer.Start(ctx)
+			}(syncCfg)
 		default:
 			log.Errorf("Unknown sync type: %s", syncCfg.Type)
 			wg.Done()
@@ -82,24 +83,7 @@ func main() {
 
 	// Start monitoring goroutine: output row counts every minute for each mapped table (source/target)
 	if cfg.EnableTableRowCountMonitoring {
-		go func() {
-			ticker := time.NewTicker(monitorInterval)
-			defer ticker.Stop()
-			for {
-				select {
-				case <-ctx.Done():
-					return
-				case <-ticker.C:
-					// For each sync config, gather row counts and log
-					for _, sc := range cfg.SyncConfigs {
-						if !sc.Enable {
-							continue
-						}
-						countAndLogTables(ctx, sc)
-					}
-				}
-			}
-		}()
+		utils.StartRowCountMonitoring(ctx, cfg, log, monitorInterval)
 	}
 
 	// Wait for all sync to complete
@@ -109,178 +93,4 @@ func main() {
 	// Wait for program to end
 	<-ctx.Done()
 	logger.Log.Info("Program has exited")
-}
-
-// countAndLogTables logs row counts for a single database connection
-func countAndLogTables(ctx context.Context, sc config.SyncConfig) {
-	switch strings.ToLower(sc.Type) {
-	case "mysql", "mariadb":
-		countAndLogMySQLOrMariaDB(ctx, sc)
-	case "postgresql":
-		countAndLogPostgreSQL(ctx, sc)
-	case "mongodb":
-		countAndLogMongoDB(ctx, sc)
-	default:
-		// Unsupported or not needed
-	}
-}
-
-// countAndLogMySQLOrMariaDB obtains row counts for MySQL / MariaDB tables
-func countAndLogMySQLOrMariaDB(ctx context.Context, sc config.SyncConfig) {
-	db, err := sql.Open("mysql", sc.SourceConnection)
-	if err != nil {
-		logger.Log.WithError(err).WithField("db_type", sc.Type).
-			Error("[Monitor] Fail connect to source")
-		return
-	}
-	defer db.Close()
-
-	db2, err := sql.Open("mysql", sc.TargetConnection)
-	if err != nil {
-		logger.Log.WithError(err).WithField("db_type", sc.Type).
-			Error("[Monitor] Fail connect to target")
-		return
-	}
-	defer db2.Close()
-
-	dbType := strings.ToUpper(sc.Type) // "MYSQL" or "MARIADB"
-	for _, mapping := range sc.Mappings {
-		srcDBName := mapping.SourceDatabase
-		tgtDBName := mapping.TargetDatabase
-		for _, tblMap := range mapping.Tables {
-			srcName := tblMap.SourceTable
-			tgtName := tblMap.TargetTable
-
-			// Source table
-			srcCount := getRowCount(db, fmt.Sprintf("%s.%s", srcDBName, srcName))
-			// Target table
-			tgtCount := getRowCount(db2, fmt.Sprintf("%s.%s", tgtDBName, tgtName))
-
-			logger.Log.WithFields(logrus.Fields{
-				"db_type":        dbType,
-				"src_db":         srcDBName,
-				"src_table":      srcName,
-				"src_row_count":  srcCount,
-				"tgt_db":         tgtDBName,
-				"tgt_table":      tgtName,
-				"tgt_row_count":  tgtCount,
-				"monitor_action": "row_count_minutely",
-			}).Info("row_count_minutely")
-		}
-	}
-}
-
-// countAndLogPostgreSQL obtains row counts for PostgreSQL tables
-func countAndLogPostgreSQL(ctx context.Context, sc config.SyncConfig) {
-	db, err := sql.Open("postgres", sc.SourceConnection)
-	if err != nil {
-		logger.Log.WithError(err).WithField("db_type", "POSTGRESQL").
-			Error("[Monitor] Fail connect to source")
-		return
-	}
-	defer db.Close()
-
-	db2, err := sql.Open("postgres", sc.TargetConnection)
-	if err != nil {
-		logger.Log.WithError(err).WithField("db_type", "POSTGRESQL").
-			Error("[Monitor] Fail connect to target")
-		return
-	}
-	defer db2.Close()
-
-	for _, mapping := range sc.Mappings {
-		srcDBName := mapping.SourceDatabase
-		tgtDBName := mapping.TargetDatabase
-		srcSchema := mapping.SourceSchema
-		if srcSchema == "" {
-			srcSchema = "public"
-		}
-		tgtSchema := mapping.TargetSchema
-		if tgtSchema == "" {
-			tgtSchema = "public"
-		}
-
-		for _, tblMap := range mapping.Tables {
-			srcName := tblMap.SourceTable
-			tgtName := tblMap.TargetTable
-
-			fullSrc := fmt.Sprintf("%s.%s", srcSchema, srcName)
-			fullTgt := fmt.Sprintf("%s.%s", tgtSchema, tgtName)
-
-			srcCount := getRowCount(db, fullSrc)
-			tgtCount := getRowCount(db2, fullTgt)
-
-			logger.Log.WithFields(logrus.Fields{
-				"db_type":        "POSTGRESQL",
-				"src_schema":     srcSchema,
-				"src_table":      srcName,
-				"src_db":         srcDBName,
-				"src_row_count":  srcCount,
-				"tgt_schema":     tgtSchema,
-				"tgt_table":      tgtName,
-				"tgt_db":         tgtDBName,
-				"tgt_row_count":  tgtCount,
-				"monitor_action": "row_count_minutely",
-			}).Info("row_count_minutely")
-		}
-	}
-}
-
-// countAndLogMongoDB obtains document counts for MongoDB collections
-func countAndLogMongoDB(ctx context.Context, sc config.SyncConfig) {
-	srcClient, err := mongo.Connect(ctx, options.Client().ApplyURI(sc.SourceConnection))
-	if err != nil {
-		logger.Log.WithError(err).WithField("db_type", "MONGODB").
-			Error("[Monitor] Fail connect to source")
-		return
-	}
-	defer func() {
-		_ = srcClient.Disconnect(ctx)
-	}()
-
-	tgtClient, err := mongo.Connect(ctx, options.Client().ApplyURI(sc.TargetConnection))
-	if err != nil {
-		logger.Log.WithError(err).WithField("db_type", "MONGODB").
-			Error("[Monitor] Fail connect to target")
-		return
-	}
-	defer func() {
-		_ = tgtClient.Disconnect(ctx)
-	}()
-
-	for _, mapping := range sc.Mappings {
-		srcDBName := mapping.SourceDatabase
-		tgtDBName := mapping.TargetDatabase
-		for _, tblMap := range mapping.Tables {
-			srcName := tblMap.SourceTable
-			tgtName := tblMap.TargetTable
-
-			srcColl := srcClient.Database(srcDBName).Collection(srcName)
-			tgtColl := tgtClient.Database(tgtDBName).Collection(tgtName)
-
-			srcCount, _ := srcColl.EstimatedDocumentCount(ctx)
-			tgtCount, _ := tgtColl.EstimatedDocumentCount(ctx)
-
-			logger.Log.WithFields(logrus.Fields{
-				"db_type":        "MONGODB",
-				"src_db":         srcDBName,
-				"src_coll":       srcName,
-				"src_row_count":  srcCount,
-				"tgt_db":         tgtDBName,
-				"tgt_coll":       tgtName,
-				"tgt_row_count":  tgtCount,
-				"monitor_action": "row_count_minutely",
-			}).Info("row_count_minutely")
-		}
-	}
-}
-
-// getRowCount is a general SQL count function
-func getRowCount(db *sql.DB, table string) int64 {
-	query := fmt.Sprintf("SELECT COUNT(*) FROM %s", table)
-	var cnt int64
-	if err := db.QueryRow(query).Scan(&cnt); err != nil {
-		return -1
-	}
-	return cnt
-}
+}

configs/config.yaml

@@ -53,4 +53,16 @@ sync_configs:
         target_schema: "public"
         tables:
           - source_table: "users"
-            target_table: "users"
+            target_table: "users"
+
+  - type: "redis"
+    enable: true
+    source_connection: "redis://localhost:6379/0"
+    target_connection: "redis://localhost:6379/1"
+    redis_position_path: "/tmp/state/redis_position"
+    mappings:
+      - source_database: "db0"
+        target_database: "db1"
+        tables:
+          - source_table: "source_stream"  # Redis Stream Name
+            target_table: ""