Verify: migrate from Redis to KiviDB

Compare key counts first:

redis-cli -h redis.internal INFO keyspace
redis-cli -h kividb.internal -p 6380 INFO keyspace

keys= should match for every database; expires= should match too, give or take keys that expired during the copy. Then spot-check values. This script compares N random keys (or every key, with 0) between the two servers: the type, the whole value, the TTL, the expiry of each hash field, and each stream's entries and consumer groups.

#!/usr/bin/env python3
"""Spot-check a migration: compare keys between a source and KiviDB.

    pip install "redis>=5"
    python3 verify.py [--src-cluster] [--dst-cluster] SOURCE_URL TARGET_URL [N] [DB ...]

N random keys per database (default 1000); N=0 checks every key, by SCAN.
--src-cluster / --dst-cluster: that side is a cluster (database 0 only).
URLs look like redis://:password@host:6379 or rediss://user:password@host:6380.
Exits 1 if any key differs.
"""
import sys
import redis

flags = {a for a in sys.argv[1:] if a.startswith("--")}
args = [a for a in sys.argv[1:] if not a.startswith("--")]
src_url, dst_url = args[0], args[1]
n = int(args[2]) if len(args) > 2 else 1000
dbs = [int(d) for d in args[3:]] or [0]
TTL_SLACK_MS = 5000  # TTLs keep counting down while the check runs


def value(r, key, kind):
    """The whole value of a key, in a form two servers can be compared by."""
    if kind == "string":
        v = r.get(key)
        if v and v.startswith(b"HYLL"):  # a HyperLogLog: compare what it counts
            return "hyperloglog", r.pfcount(key)
        return v
    if kind == "list":
        return r.lrange(key, 0, -1)
    if kind == "set":
        return sorted(r.smembers(key))
    if kind == "zset":
        return r.zrange(key, 0, -1, withscores=True)
    if kind == "hash":
        h = r.hgetall(key)
        fields = sorted(h)
        try:  # per-field expiry as a unix time in ms (Redis 7.4+); -1 = none
            when = r.execute_command("HPEXPIRETIME", key, "FIELDS", len(fields), *fields)
        except redis.ResponseError:
            when = None  # the source has no field expiry
        return [(f, h[f]) for f in fields], when
    if kind == "stream":
        entries = r.xrange(key)
        groups = []
        for g in r.xinfo_groups(key):
            pending = r.xpending(key, g["name"])
            groups.append((g["name"], g["last-delivered-id"], g["consumers"], pending["pending"]))
        return entries, sorted(groups)
    return ("unchecked type", kind)


def ttl_close(a, b):
    if a < 0 or b < 0:
        return a == b  # -1 no expiry, -2 gone
    return abs(a - b) <= TTL_SLACK_MS


def same(a, b, kind):
    if kind != "hash":
        return a == b
    (fa, wa), (fb, wb) = a, b
    if fa != fb:
        return False
    if wa is None or wb is None:  # one side has no field expiry: the other must have none set
        return all(x == -1 for x in (wa or []) + (wb or []))
    return all(ttl_close(x, y) for x, y in zip(wa, wb))


def dbsize(r):
    if isinstance(r, redis.RedisCluster):
        return r.dbsize(target_nodes=redis.RedisCluster.PRIMARIES)  # summed over the shards
    return r.dbsize()


bad = checked = 0
for db in dbs:
    src = (redis.RedisCluster.from_url(src_url) if "--src-cluster" in flags
           else redis.Redis.from_url(src_url, db=db))
    dst = (redis.RedisCluster.from_url(dst_url) if "--dst-cluster" in flags
           else redis.Redis.from_url(dst_url, db=db))
    print(f"db{db}: DBSIZE source={dbsize(src)} target={dbsize(dst)}")
    keys = src.scan_iter(count=1000) if n == 0 else (src.randomkey() for _ in range(n))
    for key in keys:
        if key is None:
            break
        kind = src.type(key).decode()
        problems = []
        if dst.type(key).decode() != kind:
            problems.append(f"type {kind} vs {dst.type(key).decode()}")
        else:
            if not same(value(src, key, kind), value(dst, key, kind), kind):
                problems.append("value differs")
            if not ttl_close(src.pttl(key), dst.pttl(key)):
                problems.append(f"pttl {src.pttl(key)} vs {dst.pttl(key)}")
        checked += 1
        if problems:
            bad += 1
            print(f"  db{db} {key!r} ({kind}): {'; '.join(problems)}")

print(f"checked {checked} keys, {bad} differ")
sys.exit(1 if bad else 0)

Run it for each database you use:

python3 verify.py redis://:$REDIS_PASSWORD@redis.internal:6379 \
  rediss://kividbapp:$KIVIDB_PASSWORD@your-name.cloud.kividb.io:6380 1000 0 1
# db0: DBSIZE source=4123 target=4123
# db1: DBSIZE source=101 target=101
# checked 2000 keys, 0 differ

For a rediss:// URL with KiviDB Cloud's private CA, add ?ssl_ca_certs=kividb-ca.pem to the URL. The script only reads. While writes are still flowing, a key that changed between the two reads can show as different; run it again after writes stop. Keys of a module type are reported as unchecked type.