cmd_migrate.go

package server

import (
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/ledisdb/ledisdb/ledis"
	"github.com/siddontang/go/hack"
	"github.com/siddontang/go/log"
	"github.com/siddontang/goredis"
)

func dumpCommand(c *client) error {
	if len(c.args) != 1 {
		return ErrCmdParams
	}

	key := c.args[0]
	data, err := c.db.Dump(key)
	if err != nil {
		return err
	}
	c.resp.writeBulk(data)
	return nil
}

func ldumpCommand(c *client) error {
	if len(c.args) != 1 {
		return ErrCmdParams
	}

	key := c.args[0]
	data, err := c.db.LDump(key)
	if err != nil {
		return err
	}
	c.resp.writeBulk(data)
	return nil
}

func hdumpCommand(c *client) error {
	if len(c.args) != 1 {
		return ErrCmdParams
	}

	key := c.args[0]
	data, err := c.db.HDump(key)
	if err != nil {
		return err
	}
	c.resp.writeBulk(data)
	return nil
}

func sdumpCommand(c *client) error {
	if len(c.args) != 1 {
		return ErrCmdParams
	}

	key := c.args[0]
	data, err := c.db.SDump(key)
	if err != nil {
		return err
	}
	c.resp.writeBulk(data)
	return nil
}

func zdumpCommand(c *client) error {
	if len(c.args) != 1 {
		return ErrCmdParams
	}

	key := c.args[0]
	data, err := c.db.ZDump(key)
	if err != nil {
		return err
	}
	c.resp.writeBulk(data)
	return nil
}

// unlike redis, restore will try to delete old key first
func restoreCommand(c *client) error {
	args := c.args
	if len(args) != 3 {
		return ErrCmdParams
	}

	key := args[0]
	ttl, err := ledis.StrInt64(args[1], nil)
	if err != nil {
		return err
	}
	data := args[2]

	if err = c.db.Restore(key, ttl, data); err != nil {
		return err
	}
	c.resp.writeStatus(OK)
	return nil
}

// maybe only used in xcodis for redis data port
func xrestoreCommand(c *client) error {
	args := c.args
	if len(args) != 4 {
		return ErrCmdParams
	}

	//	tp := strings.ToUpper(string(args[2]))
	key := args[1]
	ttl, err := ledis.StrInt64(args[2], nil)
	if err != nil {
		return err
	}
	data := args[3]

	if err = c.db.Restore(key, ttl, data); err != nil {
		return err
	}
	c.resp.writeStatus(OK)
	return nil
}

func xdump(db *ledis.DB, tp string, key []byte) ([]byte, error) {
	var err error
	var data []byte
	switch strings.ToUpper(tp) {
	case KVName:
		data, err = db.Dump(key)
	case HashName:
		data, err = db.HDump(key)
	case ListName:
		data, err = db.LDump(key)
	case SetName:
		data, err = db.SDump(key)
	case ZSetName:
		data, err = db.ZDump(key)
	default:
		err = fmt.Errorf("invalid key type %s", tp)
	}
	return data, err
}

func xdel(db *ledis.DB, tp string, key []byte) error {
	var err error
	switch strings.ToUpper(tp) {
	case KVName:
		_, err = db.Del(key)
	case HashName:
		_, err = db.HClear(key)
	case ListName:
		_, err = db.LClear(key)
	case SetName:
		_, err = db.SClear(key)
	case ZSetName:
		_, err = db.ZClear(key)
	default:
		err = fmt.Errorf("invalid key type %s", tp)
	}
	return err
}

func xttl(db *ledis.DB, tp string, key []byte) (int64, error) {
	switch strings.ToUpper(tp) {
	case KVName:
		return db.TTL(key)
	case HashName:
		return db.HTTL(key)
	case ListName:
		return db.LTTL(key)
	case SetName:
		return db.STTL(key)
	case ZSetName:
		return db.ZTTL(key)
	default:
		return 0, fmt.Errorf("invalid key type %s", tp)
	}
}

func xscan(db *ledis.DB, tp string, count int) ([][]byte, error) {
	switch strings.ToUpper(tp) {
	case KVName:
		return db.Scan(KV, nil, count, false, "")
	case HashName:
		return db.Scan(HASH, nil, count, false, "")
	case ListName:
		return db.Scan(LIST, nil, count, false, "")
	case SetName:
		return db.Scan(SET, nil, count, false, "")
	case ZSetName:
		return db.Scan(ZSET, nil, count, false, "")
	default:
		return nil, fmt.Errorf("invalid key type %s", tp)
	}
}

func xdumpCommand(c *client) error {
	args := c.args
	if len(args) != 2 {
		return ErrCmdParams
	}

	tp := strings.ToUpper(string(args[0]))
	key := args[1]

	data, err := xdump(c.db, tp, key)
	if err != nil {
		return err
	}
	c.resp.writeBulk(data)
	return nil
}

func (app *App) getMigrateClient(addr string) *goredis.Client {
	app.migrateM.Lock()

	mc, ok := app.migrateClients[addr]
	if !ok {
		mc = goredis.NewClient(addr, "")
		app.migrateClients[addr] = mc
	}

	app.migrateM.Unlock()

	return mc
}

type migrateKeyLocker struct {
	m sync.Mutex

	locks map[string]struct{}
}

func (m *migrateKeyLocker) Lock(key []byte) bool {
	m.m.Lock()
	defer m.m.Unlock()

	k := hack.String(key)
	_, ok := m.locks[k]
	if ok {
		return false
	}
	m.locks[k] = struct{}{}
	return true
}

func (m *migrateKeyLocker) Unlock(key []byte) {
	m.m.Lock()
	defer m.m.Unlock()

	delete(m.locks, hack.String(key))
}

func newMigrateKeyLocker() *migrateKeyLocker {
	m := new(migrateKeyLocker)

	m.locks = make(map[string]struct{})

	return m
}

func (app *App) newMigrateKeyLockers() {
	app.migrateKeyLockers = make(map[string]*migrateKeyLocker)

	app.migrateKeyLockers[KVName] = newMigrateKeyLocker()
	app.migrateKeyLockers[HashName] = newMigrateKeyLocker()
	app.migrateKeyLockers[ListName] = newMigrateKeyLocker()
	app.migrateKeyLockers[SetName] = newMigrateKeyLocker()
	app.migrateKeyLockers[ZSetName] = newMigrateKeyLocker()
}

func (app *App) migrateKeyLock(tp string, key []byte) bool {
	l, ok := app.migrateKeyLockers[strings.ToUpper(tp)]
	if !ok {
		return false
	}

	return l.Lock(key)
}

func (app *App) migrateKeyUnlock(tp string, key []byte) {
	l, ok := app.migrateKeyLockers[strings.ToUpper(tp)]
	if !ok {
		return
	}

	l.Unlock(key)
}

//XMIGRATEDB host port tp count db timeout
//select count tp type keys and migrate
//will block any other write operations
//maybe only for xcodis
func xmigratedbCommand(c *client) error {
	args := c.args
	if len(args) != 6 {
		return ErrCmdParams
	}

	addr := fmt.Sprintf("%s:%s", string(args[0]), string(args[1]))
	if addr == c.app.cfg.Addr {
		//same server， can not migrate
		return fmt.Errorf("migrate in same server is not allowed")
	}

	tp := strings.ToUpper(string(args[2]))

	count, err := ledis.StrInt64(args[3], nil)
	if err != nil {
		return err
	} else if count <= 0 {
		count = 10
	}

	db, err := parseMigrateDB(c, args[4])
	if err != nil {
		return err
	}

	timeout, err := ledis.StrInt64(args[5], nil)
	if err != nil {
		return err
	} else if timeout < 0 {
		return fmt.Errorf("invalid timeout %d", timeout)
	}

	keys, err := xscan(c.db, tp, int(count))
	if err != nil {
		return err
	} else if len(keys) == 0 {
		c.resp.writeInteger(0)
		return nil
	}

	conn, err := getMigrateDBConn(c, addr, db)
	if err != nil {
		return err
	}
	defer conn.Close()

	migrateNum := int64(0)
	for _, key := range keys {
		err = migrateKey(c, conn, tp, key, timeout)
		if err != nil {
			if err == errNoKey || err == errKeyInMigrating {
				continue
			} else {
				return err
			}
		}

		migrateNum++
	}

	c.resp.writeInteger(migrateNum)

	return nil
}

func parseMigrateDB(c *client, arg []byte) (uint64, error) {
	db, err := ledis.StrUint64(arg, nil)
	if err != nil {
		return 0, err
	} else if db >= uint64(c.app.cfg.Databases) {
		return 0, fmt.Errorf("invalid db index %d, must < %d", db, c.app.cfg.Databases)
	}
	return db, nil
}

//XMIGRATE host port type key destination-db timeout
//will block any other write operations
//maybe only for xcodis
func xmigrateCommand(c *client) error {
	args := c.args

	if len(args) != 6 {
		return ErrCmdParams
	}

	addr := fmt.Sprintf("%s:%s", string(args[0]), string(args[1]))
	if addr == c.app.cfg.Addr {
		//same server， can not migrate
		return fmt.Errorf("migrate in same server is not allowed")
	}

	tp := strings.ToUpper(string(args[2]))
	key := args[3]
	db, err := parseMigrateDB(c, args[4])
	if err != nil {
		return err
	}

	timeout, err := ledis.StrInt64(args[5], nil)
	if err != nil {
		return err
	} else if timeout < 0 {
		return fmt.Errorf("invalid timeout %d", timeout)
	}

	conn, err := getMigrateDBConn(c, addr, db)
	if err != nil {
		return err
	}
	defer conn.Close()

	// if key is in migrating, we will wait 500ms and retry again
	for i := 0; i < 10; i++ {
		if tp == "ALL" {
			// if tp is ALL, we will migrate the key in all types
			// this feature is useful for xcodis RESTORE or other commands that we don't know the data type exactly
			err = migrateAllTypeKeys(c, conn, key, timeout)
		} else {
			err = migrateKey(c, conn, tp, key, timeout)
		}

		if err != errKeyInMigrating {
			break
		} else {
			log.Infof("%s key %s is in migrating, wait 500ms and retry", tp, key)
			time.Sleep(500 * time.Millisecond)
		}
	}

	if err != nil {
		if err == errNoKey {
			c.resp.writeStatus(NOKEY)
			return nil
		}
		return err
	}

	c.resp.writeStatus(OK)
	return nil
}

func getMigrateDBConn(c *client, addr string, db uint64) (*goredis.PoolConn, error) {
	mc := c.app.getMigrateClient(addr)

	conn, err := mc.Get()
	if err != nil {
		return nil, err
	}

	if _, err = conn.Do("select", db); err != nil {
		conn.Close()
		return nil, err
	}

	return conn, nil
}

var (
	errNoKey          = errors.New("migrate key is not exists")
	errKeyInMigrating = errors.New("key is in migrating yet")
)

func migrateKey(c *client, conn *goredis.PoolConn, tp string, key []byte, timeout int64) error {
	if !c.app.migrateKeyLock(tp, key) {
		// other may also migrate this key, skip it
		return errKeyInMigrating
	}

	defer c.app.migrateKeyUnlock(tp, key)

	data, err := xdump(c.db, tp, key)
	if err != nil {
		return err
	} else if data == nil {
		return errNoKey
	}

	ttl, err := xttl(c.db, tp, key)
	if err != nil {
		return err
	}

	//timeout is milliseconds
	t := time.Duration(timeout) * time.Millisecond

	conn.SetReadDeadline(time.Now().Add(t))

	//ttl is second, but restore need millisecond
	if _, err = conn.Do("restore", key, ttl*1e3, data); err != nil {
		return err
	}

	if err = xdel(c.db, tp, key); err != nil {
		return err
	}

	return nil
}

func migrateAllTypeKeys(c *client, conn *goredis.PoolConn, key []byte, timeout int64) error {
	for _, tp := range TypeNames {
		err := migrateKey(c, conn, tp, key, timeout)
		if err != nil {
			if err == errNoKey {
				continue
			} else {
				return err
			}
		}
	}

	return nil
}

func init() {
	register("dump", dumpCommand)
	register("ldump", ldumpCommand)
	register("hdump", hdumpCommand)
	register("sdump", sdumpCommand)
	register("zdump", zdumpCommand)
	register("restore", restoreCommand)
	register("xrestore", xrestoreCommand)
	register("xdump", xdumpCommand)
	register("xmigrate", xmigrateCommand)
	register("xmigratedb", xmigratedbCommand)
}