Add parameterize vreplgen fully, add some docs.

vreplgen/README.md

@@ -0,0 +1,43 @@
+A golang CLI utility to generate vtctlclient commands to add vreplication
+rules:
+
+```
+Usage: vreplgen [-on_ddl (ignore|stop|exec|exec_ignore)] <tablet_id> <src_keyspace> <src_shard> <dest_keyspace> <dest_table1> 'filter1' [<dest_table2> 'filter2']...
+```
+
+E.g.:
+
+```
+./vreplgen cell-0000000001 main -80 main_copy transactionhistory 'select * from transactionhistory where in_keyrange(merchant_id, "hash", "80-")'
+```
+
+The utility also supports multiple table filters, which allows multiple tables
+to be specified in a single vreplication stream (good for if you have
+a lot of tables you want to process via vreplication).  E.g.:
+
+```
+./vreplgen cell-0000000001 main -80 main_copy transactionhistory 'select * from transactionhistory where in_keyrange(merchant_id, "hash", "80-")' transactionhistory2 'select * from transactionhistory2 where in_keyrange(merchant_id, "hash", "-80")'
+```
+
+An important thing to note is that a single vreplication stream cannot use 
+the same source table in the same stream.  The utility will not prevent
+you from doing this, however.
+
+`vreplgen` assumes you are running vtctld on localhost port 15999.  If not,
+you can set your VTCTLCLIENT environment variable to the vtctlclient command
+you want `vreplgen` to generate, e.g.:
+
+```
+export VTCTLCLIENT="vtctlclient -server vtctld:15999"
+```
+
+Lastly, you can control the on_ddl flag using vreplgen. The default if you
+do not specify the `-on_ddl` option is `ignore`, but you can specify:
+
+  * `-on_ddl ignore`
+  * `-on_ddl stop`
+  * `-on_ddl exec`
+  * `-on_ddl exec_ignore`
+
+depending on how you want your DDL to be handled in your replication streams.
+See the main vreplication documentation for more details.

vreplgen/vreplgen.go

@@ -23,35 +23,79 @@ package main
 
 import (
 	"bytes"
+	"flag"
 	"fmt"
+	"os"
 	"strings"
 
 	"vitess.io/vitess/go/sqltypes"
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
+var onDdl string
+
+func init() {
+	flag.StringVar(&onDdl, "on_ddl", "ignore", "Set on_ddl value for replication stream - ignore, stop, exec, exec_ignore")
+	flag.Parse()
+}
+
 func main() {
-	vtctl := "./lvtctl.sh"
-	tabletID := "test-400"
-	dbName := "vt_merchant"
+	argOffset := 0
+	if (len(os.Args) > 1 && strings.HasPrefix(os.Args[1], "-")) {
+		argOffset = 2
+	}
+
+	if (len(os.Args) < (7+argOffset)) {
+		fmt.Println("Usage: vreplgen [-on_ddl (ignore|stop|exec|exec_ignore)] <tablet_id> <src_keyspace> <src_shard> <dest_keyspace> <dest_table1> 'filter1' [<dest_table2> 'filter2']...")
+		os.Exit(1)
+	}
+
+	vtctl := os.Getenv("VTCTLCLIENT")
+	if (vtctl == "") {
+	  vtctl = "vtctlclient -server localhost:15999"
+	}
+	tabletID := os.Args[1+argOffset]
+	sourceKeyspace := os.Args[2+argOffset]
+	sourceShard := os.Args[3+argOffset]
+	destKeyspace := os.Args[4+argOffset]
+	destDbName := "vt_" + destKeyspace
+	var rules []*binlogdatapb.Rule
+	for i := 5+argOffset; i < len(os.Args); i = i+2 {
+		destTable := os.Args[i]
+		destFilter := os.Args[i+1]
+		rule := new(binlogdatapb.Rule)
+		rule.Match = destTable
+		rule.Filter = destFilter
+		rules = append(rules, rule)
+	}
 	filter := &binlogdatapb.Filter{
-		Rules: []*binlogdatapb.Rule{{
-			Match:  "morder",
-			Filter: "select * from uorder where in_keyrange(mname, 'unicode_loose_md5', '-80')",
-		}},
+		Rules: rules,
 	}
+
+	var onDdlAction binlogdatapb.OnDDLAction
+	switch onDdl {
+	case "ignore":
+		onDdlAction = binlogdatapb.OnDDLAction_IGNORE
+	case "stop":
+		onDdlAction = binlogdatapb.OnDDLAction_STOP
+	case "exec":
+		onDdlAction = binlogdatapb.OnDDLAction_EXEC
+	case "exec_ignore":
+		onDdlAction = binlogdatapb.OnDDLAction_EXEC_IGNORE
+	}
+
 	bls := &binlogdatapb.BinlogSource{
-		Keyspace: "user",
-		Shard:    "-80",
+		Keyspace: sourceKeyspace,
+		Shard:    sourceShard,
 		Filter:   filter,
-		OnDdl:    binlogdatapb.OnDDLAction_IGNORE,
+		OnDdl:    onDdlAction,
 	}
 	val := sqltypes.NewVarBinary(fmt.Sprintf("%v", bls))
 	var sqlEscaped bytes.Buffer
 	val.EncodeSQL(&sqlEscaped)
 	query := fmt.Sprintf("insert into _vt.vreplication "+
 		"(db_name, source, pos, max_tps, max_replication_lag, tablet_types, time_updated, transaction_timestamp, state) values"+
-		"('%s', %s, '', 9999, 9999, 'master', 0, 0, 'Running')", dbName, sqlEscaped.String())
+		"('%s', %s, '', 9999, 9999, 'master', 0, 0, 'Running')", destDbName, sqlEscaped.String())
 
 	fmt.Printf("%s VReplicationExec %s '%s'\n", vtctl, tabletID, strings.Replace(query, "'", "'\"'\"'", -1))
 }