Get the data out: migrate from Memcached to KiviDB

Memcached is a cache: everything in it can be rebuilt from your database. That gives you two choices.

1. Warm the new cache (usually right)

Point the application at KiviDB and let it fill on misses, as it filled Memcached. Nothing is copied. To soften the first minutes:

• Switch a slice of traffic first (one service, one region), and widen it as the hit rate climbs. INFO stats on KiviDB shows keyspace_hits and keyspace_misses.
• Or, for a while, read from KiviDB and fall back to Memcached on a miss, writing what you find into KiviDB. Remove the fallback when the misses drop.

2. Copy the items

When a cold cache would overload the database behind it, copy the items. Memcached has no command to list every key, but two debug commands come close:

Command	Gives	Notes
lru_crawler metadump hash	One line per item: key, exp (Unix time, -1 for none), la, cas, fetch, cls, size, flags	Walks the hash table, so each item is seen once. Keys are URI-encoded.
lru_crawler metadump all	The same, walking each slab class's LRU	Items moved during the walk can be missed.
lru_crawler mgdump hash	mg <key> lines, ready to send back	Binary keys come base64-encoded with the b flag.
stats cachedump <class> <limit>	Up to about 1 MB of keys per slab class	Old and incomplete; don't use it for a copy.

All of them are refused if Memcached was started with -X (--disable-dumping). The dump is a snapshot: items written after it are not in it, and items can expire or be evicted before you read them.

The program below lists the keys with metadump, reads each item with the meta command mg <key> k f t v (key, client flags, remaining TTL, value), and writes it to KiviDB with its remaining TTL. It needs Python 3 and redis-py (pip install redis).

"""Copy every item in a memcached server into KiviDB.

  python3 copy_memcached.py MEMCACHED_HOST MEMCACHED_PORT KIVIDB_URL
"""
import socket, sys
from urllib.parse import unquote
import redis

BATCH = 500


class Meta:
    def __init__(self, host, port):
        self.s = socket.create_connection((host, port))
        self.buf = b""

    def send(self, data):
        self.s.sendall(data)

    def line(self):
        while b"\r\n" not in self.buf:
            chunk = self.s.recv(1 << 20)
            if not chunk:
                raise ConnectionError("memcached closed the connection")
            self.buf += chunk
        line, self.buf = self.buf.split(b"\r\n", 1)
        return line

    def exactly(self, n):
        while len(self.buf) < n + 2:
            self.buf += self.s.recv(1 << 20)
        data, self.buf = self.buf[:n], self.buf[n + 2:]
        return data


def list_keys(host, port):
    # Item lines end in "\n", not "\r\n"; the dump ends with "END\r\n".
    s = socket.create_connection((host, port))
    s.sendall(b"lru_crawler metadump hash\r\n")
    buf = b""
    while not (buf.endswith(b"\nEND\r\n") or buf == b"END\r\n"):
        chunk = s.recv(1 << 20)
        if not chunk:
            raise ConnectionError("memcached closed the connection")
        buf += chunk
        if buf.startswith((b"BUSY", b"BADCLASS", b"ERROR", b"CLIENT_ERROR", b"SERVER_ERROR")) and buf.endswith(b"\r\n"):
            raise RuntimeError(buf.decode().strip())
    s.close()
    keys = []
    for line in buf.split(b"\n"):
        line = line.strip()
        if not line.startswith(b"key="):    # "OK", "END" and blank lines
            continue
        fields = dict(p.split(b"=", 1) for p in line.split(b" ") if b"=" in p)
        keys.append(unquote(fields[b"key"].decode()))
    return keys


def read_items(m, keys):
    """Pipelined meta gets. Yields (key, value, flags, ttl); ttl is -1 for no expiry."""
    m.send(b"".join(f"mg {k} k f t v\r\n".encode() for k in keys) + b"mn\r\n")
    while True:
        line = m.line()
        if line == b"MN":
            return
        if line == b"EN":           # expired or evicted since the dump
            continue
        parts = line.split(b" ")
        if parts[0] != b"VA":
            raise RuntimeError(line.decode())
        size = int(parts[1])
        tok = {p[:1]: p[1:] for p in parts[2:]}
        yield tok[b"k"].decode(), m.exactly(size), int(tok[b"f"]), int(tok[b"t"])


def main():
    host, port, url = sys.argv[1], int(sys.argv[2]), sys.argv[3]
    kv = redis.Redis.from_url(url)
    keys = list_keys(host, port)
    m = Meta(host, port)
    copied = with_ttl = flagged = 0
    for i in range(0, len(keys), BATCH):
        pipe = kv.pipeline(transaction=False)
        for key, value, flags, ttl in read_items(m, keys[i:i + BATCH]):
            if ttl > 0:
                pipe.set(key, value, ex=ttl)
                with_ttl += 1
            else:
                pipe.set(key, value)
            if flags:
                flagged += 1
            copied += 1
        pipe.execute()
    print(f"listed {len(keys)} keys, copied {copied}, {with_ttl} with a TTL, {flagged} with non-zero client flags")


if __name__ == "__main__":
    main()

Run it once per Memcached server:

python3 copy_memcached.py memcached-1.internal 11211 redis://:$KIVIDB_PASSWORD@kividb.internal:6380

The meta commands need Memcached 1.6 or later. The program copies the bytes as they are. If its last line reports items with non-zero client flags, read Client flags below before you switch.