[BC] Add options to specify source/destination dbNames, and also source collection names to watch

Author: JT

cmd/root.go

@@ -58,19 +58,26 @@ func init() {
 
 	RootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default is $HOME/.go-sync-mongo.yaml)")
 	RootCmd.PersistentFlags().String("src", "", "mongodb://host1:27017")
+	RootCmd.PersistentFlags().String("src-db", "", "source database name")
+	RootCmd.PersistentFlags().String("src-collections", "", "source collections separated by ,")
 	RootCmd.PersistentFlags().String("src-username", "", "source database username")
 	RootCmd.PersistentFlags().String("src-password", "", "source database password")
 	RootCmd.PersistentFlags().Bool("src-ssl", false, "source ssl enabled (true)")
 	RootCmd.PersistentFlags().String("dst", "", "mongodb://host1:27017,host2:27017")
+	RootCmd.PersistentFlags().String("dst-db", "", "target database name")
+	RootCmd.PersistentFlags().String("dst-collections", "", "dst collections separated by ,")
 	RootCmd.PersistentFlags().String("dst-username", "", "destination database username")
 	RootCmd.PersistentFlags().String("dst-password", "", "destiantion database password")
 	RootCmd.PersistentFlags().Bool("dst-ssl", false, "destination ssl enabled (true)")
 
 	viper.BindPFlag("src", RootCmd.PersistentFlags().Lookup("src"))
+	viper.BindPFlag("src-db", RootCmd.PersistentFlags().Lookup("src-db"))
+	viper.BindPFlag("src-collections", RootCmd.PersistentFlags().Lookup("src-collections"))
 	viper.BindPFlag("src-username", RootCmd.PersistentFlags().Lookup("src-username"))
 	viper.BindPFlag("src-password", RootCmd.PersistentFlags().Lookup("src-password"))
 	viper.BindPFlag("src-ssl", RootCmd.PersistentFlags().Lookup("src-ssl"))
 	viper.BindPFlag("dst", RootCmd.PersistentFlags().Lookup("dst"))
+	viper.BindPFlag("dst-db", RootCmd.PersistentFlags().Lookup("dst-db"))
 	viper.BindPFlag("dst-username", RootCmd.PersistentFlags().Lookup("dst-username"))
 	viper.BindPFlag("dst-password", RootCmd.PersistentFlags().Lookup("dst-password"))
 	viper.BindPFlag("dst-ssl", RootCmd.PersistentFlags().Lookup("dst-ssl"))

cmd/sync.go

@@ -2,8 +2,9 @@ package cmd
 
 import (
 	"fmt"
+	s "strings"
 
-	db "github.com/checkr/go-sync-mongo/db"
+	db "../db"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	mgo "gopkg.in/mgo.v2"
@@ -21,6 +22,8 @@ var syncCmd = &cobra.Command{
 				Username: viper.GetString("src-username"),
 				Password: viper.GetString("src-password"),
 			},
+			Database:    viper.GetString("src-db"),
+			Collections: s.Split(viper.GetString("src-collections"), ","),
 		}
 		src, err := db.NewConnection(srcConfig)
 		if err != nil {
@@ -34,6 +37,8 @@ var syncCmd = &cobra.Command{
 				Username: viper.GetString("dst-username"),
 				Password: viper.GetString("dst-password"),
 			},
+			Database:    viper.GetString("dst-db"),
+			Collections: make([]string, 0),
 		}
 		dst, err := db.NewConnection(dstConfig)
 		if err != nil {

db/config.go

@@ -3,9 +3,11 @@ package db
 import mgo "gopkg.in/mgo.v2"
 
 type Config struct {
-	URI   string
-	SSL   bool
-	Creds mgo.Credential
+	URI         string
+	SSL         bool
+	Creds       mgo.Credential
+	Database    string
+	Collections []string
 }
 
 func (p *Config) Load() error {

db/db.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"net"
+	s "strings"
 	"sync"
 	"time"
 
@@ -88,7 +89,7 @@ func (c *Connection) Databases() ([]string, error) {
 	var slice []string
 
 	for _, dbname := range dbnames {
-		if dbname != "local" && dbname != "admin" {
+		if dbname == c.config.Database {
 			slice = append(slice, dbname)
 		}
 	}
@@ -104,7 +105,7 @@ func (c *Connection) databaseRegExs() ([]bson.RegEx, error) {
 	var slice []bson.RegEx
 
 	for _, dbname := range dbnames {
-		if dbname != "local" && dbname != "admin" {
+		if dbname == c.config.Database {
 			slice = append(slice, bson.RegEx{Pattern: dbname + ".*"})
 		}
 	}
@@ -179,6 +180,7 @@ func (c *Connection) SyncOplog(dst *Connection) error {
 			// apply the operation
 			opsToApply := []Oplog{oplogEntry}
 			err := dst.Session.Run(bson.M{"applyOps": opsToApply}, &applyOpsResponse)
+
 			if err != nil {
 				return fmt.Errorf("error applying ops: %v", err)
 			}
@@ -198,7 +200,8 @@ func (c *Connection) SyncOplog(dst *Connection) error {
 		}
 	}
 
-	fmt.Println("Tailing...")
+	fmt.Println("Tailing.....")
+
 	iter = oplog.Find(tail_query).Tail(1 * time.Second)
 	for {
 		for iter.Next(&oplogEntry) {
@@ -207,11 +210,46 @@ func (c *Connection) SyncOplog(dst *Connection) error {
 				log.Printf("skipping no-op for namespace `%v`", oplogEntry.Namespace)
 				continue
 			}
-			opCount++
+
+			if !s.Contains(oplogEntry.Namespace, c.config.Database+".") {
+				log.Printf("skipping namespace `%v`", oplogEntry.Namespace)
+				continue
+			}
+
+			// check collection against config
+			collection := s.Split(oplogEntry.Namespace, ".")[1]
+
+			isCollectionMatch := false
+			for _, permittedCollection := range c.config.Collections {
+				if collection == permittedCollection {
+					isCollectionMatch = true
+				}
+			}
+
+			if !isCollectionMatch {
+				log.Printf("skipping collection `%v`", oplogEntry.Namespace)
+				continue
+			}
+
+			oplogEntry.Namespace = dst.config.Database + "." + collection
+
+			if false {
+				fmt.Println("\n")
+				fmt.Println("****************************** %v", oplogEntry.HistoryID)
+				fmt.Println("****************************** %v", oplogEntry.Namespace)
+				fmt.Println("****************************** %v", oplogEntry.Object)
+				fmt.Println("****************************** %v", oplogEntry.Operation)
+				fmt.Println("****************************** %v", oplogEntry.Query)
+				fmt.Println("****************************** %v", oplogEntry.Timestamp)
+				fmt.Println("****************************** %v", oplogEntry.Version)
+				fmt.Println("%v", oplogEntry.Namespace)
+			}
 
 			// apply the operation
+			opCount++
 			opsToApply := []Oplog{oplogEntry}
 			err := dst.Session.Run(bson.M{"applyOps": opsToApply}, &applyOpsResponse)
+
 			if err != nil {
 				return fmt.Errorf("error applying ops: %v", err)
 			}

go-sync-mongo

@@ -14,7 +14,7 @@
 
 package main
 
-import "github.com/checkr/go-sync-mongo/cmd"
+import "./cmd"
 
 func main() {
 	cmd.Execute()