initial commit

.gitignore

@@ -0,0 +1,3 @@
+migrate
+bin/
+pkg/

Makefile

@@ -0,0 +1,14 @@
+.PHONY: all test build linux
+
+all: clean test build
+clean:
+	@rm -rf bin/*
+
+build: clean
+	@gb build
+
+test:
+	@gb test
+
+linux: clean
+	@GOOS=linux gb build

README.md

@@ -0,0 +1,46 @@
+migr8
+---
+
+Redis Migration Utility written in Go
+
+## Build
+migr8 uses [gb](http://getgb.io) to vendor dependencies.
+
+To install it run, `go get github.com/constabulary/gb/...`
+
+Tests require that `redis-server` is somewhere in your $PATH.
+
+`make` To run tests and create a binary
+
+## Usage
+```
+NAME:
+   migr8 - It's time to move some redis
+
+USAGE:
+   migr8 [global options] command [command options] [arguments...]
+
+VERSION:
+   0.0.0
+
+COMMANDS:
+   migrate	Migrate one redis to a new redis
+   delete	Delete all keys with the given prefix
+   help, h	Shows a list of commands or help for one command
+
+GLOBAL OPTIONS:
+   --dry-run, --dr			Run in dry-run mode
+   --source, -s "127.0.0.1:6379"	The redis server to pull data from
+   --dest, -d "127.0.0.1:6379"		The destination redis server
+   --workers, -w "2"			The count of workers to spin up
+   --batch, -b "10"			The batch size
+   --prefix, -p 			The key prefix to act on
+   --clear-dest, -c			Clear the destination of all it's keys and values
+   --help, -h				show help
+   --version, -v			print the version
+```
+
+#### Cross Compile for Linux:
+*Note:* You will need the Go cross compile tools. If you're using homebrew: `brew install go --cross-compile-common`
+
+`make linux` will build a linux binary in bin/

src/migr8/delete.go

@@ -0,0 +1,28 @@
+package main
+
+import (
+	"log"
+	"sync"
+
+	"github.com/garyburd/redigo/redis"
+)
+
+func deleteKeys(queue chan Task, wg *sync.WaitGroup) {
+	sourceConn := sourceConnection(config.Source)
+	for task := range queue {
+		for _, key := range task.list {
+			if config.DryRun {
+				log.Printf("Would have deleted %s", key)
+				continue
+			}
+
+			if _, err := redis.String(sourceConn.Do("del", key)); err != nil {
+				log.Printf("Deleted %s \n", key)
+			} else {
+				log.Printf("Could not deleted %s: %s\n", key, err)
+			}
+		}
+	}
+
+	wg.Done()
+}

src/migr8/delete_test.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/garyburd/redigo/redis"
+)
+
+func Test_DeleteAllKeysWithPrefix(t *testing.T) {
+	ClearRedis()
+
+	config = Config{
+		Source:  sourceServer.url,
+		Workers: 1,
+		Batch:   10,
+		Prefix:  "bar",
+	}
+
+	for i := 0; i < 100; i++ {
+		key := fmt.Sprintf("bar:%d", i)
+		sourceServer.conn.Do("SET", key, i)
+	}
+
+	sourceServer.conn.Do("SET", "baz:foo", "yolo")
+
+	RunAction(deleteKeys)
+
+	for i := 0; i < 100; i++ {
+		key := fmt.Sprintf("bar:%d", i)
+		exists, _ := redis.Bool(sourceServer.conn.Do("EXISTS", key))
+
+		if exists {
+			t.Errorf("Found a key %d that should have been deleted", key)
+		}
+	}
+
+	exists, _ := redis.Bool(sourceServer.conn.Do("EXISTS", "baz:foo"))
+
+	if !exists {
+		t.Errorf("Deleted a key %s that should not have been deleted", "baz:foo")
+	}
+}
+
+func Test_DoesNothingInDryRunModeForDelete(t *testing.T) {
+	ClearRedis()
+
+	config = Config{
+		Source:  sourceServer.url,
+		Workers: 1,
+		Batch:   10,
+		Prefix:  "bar",
+		DryRun:  true,
+	}
+
+	for i := 0; i < 100; i++ {
+		key := fmt.Sprintf("bar:%d", i)
+		sourceServer.conn.Do("SET", key, i)
+	}
+
+	RunAction(deleteKeys)
+
+	for i := 0; i < 100; i++ {
+		key := fmt.Sprintf("bar:%d", i)
+		exists, _ := redis.Bool(sourceServer.conn.Do("EXISTS", key))
+
+		if !exists {
+			t.Errorf("In DryRun mode, but found a key %d that was actually deleted", key)
+		}
+	}
+}

src/migr8/main.go

@@ -0,0 +1,151 @@
+package main
+
+import (
+	"log"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/codegangsta/cli"
+	"github.com/garyburd/redigo/redis"
+)
+
+type Task struct {
+	list []string
+}
+
+type Worker func(queue chan Task, wg *sync.WaitGroup)
+
+type Config struct {
+	Dest      string
+	Source    string
+	Workers   int
+	Batch     int
+	Prefix    string
+	ClearDest bool
+	DryRun    bool
+}
+
+var config Config
+
+func main() {
+	app := cli.NewApp()
+	app.Name = "migr8"
+	app.Usage = "It's time to move some redis"
+	app.Commands = []cli.Command{
+		{
+			Name:   "migrate",
+			Usage:  "Migrate one redis to a new redis",
+			Action: Migrate,
+		},
+		{
+			Name:   "delete",
+			Usage:  "Delete all keys with the given prefix",
+			Action: Delete,
+		},
+	}
+	app.Flags = []cli.Flag{
+		cli.BoolFlag{
+			Name:  "dry-run, n",
+			Usage: "Run in dry-run mode",
+		},
+		cli.StringFlag{
+			Name:  "source, s",
+			Usage: "The redis server to pull data from",
+			Value: "127.0.0.1:6379",
+		},
+		cli.StringFlag{
+			Name:  "dest, d",
+			Usage: "The destination redis server",
+			Value: "127.0.0.1:6379",
+		},
+		cli.IntFlag{
+			Name:  "workers, w",
+			Usage: "The count of workers to spin up",
+			Value: 2,
+		},
+		cli.IntFlag{
+			Name:  "batch, b",
+			Usage: "The batch size",
+			Value: 10,
+		},
+		cli.StringFlag{
+			Name:  "prefix, p",
+			Usage: "The key prefix to act on",
+		},
+		cli.BoolFlag{
+			Name:  "clear-dest, c",
+			Usage: "Clear the destination of all it's keys and values",
+		},
+	}
+
+	app.Run(os.Args)
+}
+
+func ParseConfig(c *cli.Context) {
+	config = Config{
+		Source:    c.GlobalString("source"),
+		Dest:      c.GlobalString("dest"),
+		Workers:   c.GlobalInt("workers"),
+		Batch:     c.GlobalInt("batch"),
+		Prefix:    c.GlobalString("prefix"),
+		ClearDest: c.GlobalBool("clear-dest"),
+		DryRun:    c.GlobalBool("dry-run"),
+	}
+}
+
+func sourceConnection(source string) redis.Conn {
+	// attempt to connect to source server
+	sourceConn, err := redis.Dial("tcp", source)
+	if err != nil {
+		panic(err)
+	}
+
+	return sourceConn
+}
+
+func destConnection(dest string) redis.Conn {
+	// attempt to connect to source server
+	destConn, err := redis.Dial("tcp", dest)
+	if err != nil {
+		panic(err)
+	}
+
+	return destConn
+}
+
+func RunAction(action Worker) {
+	wg := &sync.WaitGroup{}
+	workQueue := make(chan Task, config.Workers)
+	startedAt = time.Now()
+
+	wg.Add(1)
+	go scanKeys(workQueue, wg)
+
+	for i := 0; i <= config.Workers; i++ {
+		wg.Add(1)
+		go action(workQueue, wg)
+	}
+
+	wg.Wait()
+}
+
+func Migrate(c *cli.Context) {
+	ParseConfig(c)
+	log.Printf("Running migrate with config: %+v\n", config)
+	log.SetPrefix("migrate - ")
+
+	if config.ClearDest {
+		clearDestination(c.String("dest"))
+	}
+
+	RunAction(migrateKeys)
+}
+
+func Delete(c *cli.Context) {
+	ParseConfig(c)
+	log.Printf("Running delete with config: %+v\n", config)
+	log.SetPrefix("delete - ")
+
+	RunAction(deleteKeys)
+}

src/migr8/main_test.go

@@ -0,0 +1,79 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/garyburd/redigo/redis"
+)
+
+var sourceServer *RedisTestServer
+var destServer *RedisTestServer
+
+func ClearRedis() {
+	sourceServer.conn.Do("flushdb")
+	destServer.conn.Do("flushdb")
+}
+
+func StartTestServers() {
+	fmt.Println("Starting redis...")
+	sourceServer = NewRedisTestServer("6377")
+	destServer = NewRedisTestServer("6277")
+}
+
+func StopTestServers() {
+	fmt.Println("Stopping redis...")
+	sourceServer.Stop()
+	destServer.Stop()
+}
+
+func NewRedisTestServer(port string) *RedisTestServer {
+	srv := &RedisTestServer{
+		port: port,
+		url:  fmt.Sprintf("127.0.0.1:%s", port),
+	}
+
+	srv.Start()
+
+	return srv
+}
+
+type RedisTestServer struct {
+	cmd  *exec.Cmd
+	port string
+	url  string
+	conn redis.Conn
+}
+
+func (s *RedisTestServer) Start() {
+	args := fmt.Sprintf("--port %s", s.port)
+	s.cmd = exec.Command("redis-server", args)
+
+	err := s.cmd.Start()
+	time.Sleep(2 * time.Second)
+
+	conn, err := redis.Dial("tcp", s.url)
+	s.conn = conn
+
+	if err != nil {
+		panic("Could not start redis")
+	}
+}
+
+func (s *RedisTestServer) Stop() {
+	s.cmd.Process.Signal(syscall.SIGTERM)
+	s.cmd.Process.Wait()
+}
+
+func TestMain(m *testing.M) {
+	StartTestServers()
+
+	result := m.Run()
+
+	StopTestServers()
+	os.Exit(result)
+}