import contextlib import logging import threading import time import pymysql import pytest from . import util def test_set_scheduled_metadata_backup(cli): path = '/v1/metadata-schedule-backup/metadata-scheduled-backup-manager' resp = cli.put(path, json={ "maxReservedTime": "168h", "schedule": "Saturday,Sunday,Friday,Monday 01:00" }) assert resp.status_code < 300 assert resp.json() == cli.get(path).json() def test_backup_metadata_manually(cli): resp = cli.post('/v1/metadata-backup', json={}) name = resp.json()['name'] first_check = [True] def check_backup_done(): backup = [b for b in cli.get('/v1/metadata-backup').json()['items'] if b['name'] == name] if not backup: if first_check[0]: return False raise RuntimeError('backup not found: ' + name) for b in backup: assert b['status'] != 'failed' if b['status'] != 'succeeded': return False assert b['startTime'] <= b['endTime'] assert b['size'] > 1024 assert b['backupTemplate']['operation'] == 'manual' return True try: util.wait_until(check_backup_done) finally: cli.delete('/v1/metadata-backup/'+name) def test_set_scheduled_instance_backup(cli): tc = None for item in cli.get('/v1/tidbclusters').json()['items']: if 'backup-test' in item['tags']: tc = item break if not tc: pytest.skip("no tidb cluster instance for backup test") path = '/v1/namespaces/{}/tidbcluster-schedule-backup' \ '/tidb-scheduled-backup-manager'.format(tc['metadata']['name']) resp = cli.put(path, json={ "maxReservedTime": "168h", "schedule": "Saturday,Sunday,Friday,Monday 00:00", "backupTemplate": {"cluster": "h77d6042c5a1e9ad", "type": "full", "method": "logic"} }) assert resp.status_code < 300 assert resp.json() == cli.get(path).json() def test_backup_instance(cli, k8s_cli): instance_util = util.InstanceUtil(cli) logging.info("create instances for backup") with instance_util.new_instance('test-backup-'+str(int(time.time()))) as tc1: logging.info("backup empty instance") path = '/v1/namespaces/{}/tidbcluster-backup'.format(tc1['metadata']['name']) resp = cli.post(path, json={ "backupTemplate": { "cluster": tc1['metadata']['name'], "type": "full", "method": "logic" } }) assert resp.status_code < 300 empty_backup = resp.json() def check_backup(target): def pool(): for backup in cli.get(path).json()['items']: if backup['metadata']['name'] == target['metadata']['name']: assert backup['status'].lower() != 'failed' if backup['status'].lower() == 'complete': return True return False return pool time.sleep(5) util.wait_until(check_backup(empty_backup)) dg = DataGen({ 'host': tc1['endpoints'][0], 'port': tc1['port'], 'user': 'root', 'password': util.InstanceUtil.default_passwd, }, 100, 10, 10) logging.info("setup data") dg.setup() logging.info("check data") dg.check() logging.info("backup instance") resp = cli.post(path, json={ "backupTemplate": { "cluster": tc1['metadata']['name'], "type": "full", "method": "logic" } }) assert resp.status_code < 300 data_backup = resp.json() time.sleep(5) util.wait_until(check_backup(data_backup), timeout=900) logging.info("create instance for restore") with instance_util.new_instance('test-restore-'+str(int(time.time()))) as tc2: logging.info("restore to new instance") resp = cli.post('/v1/namespaces/{}/tidbcluster-restore'.format(tc2['metadata']['name']), json={ 'backup': data_backup['metadata']['name'], 'backupNamespace': tc1['metadata']['name'], 'cluster': tc2['metadata']['name'], }) assert resp.status_code < 300 restore = resp.json() logging.info("wait for restore done") def poll(): api = k8s_cli.CustomObjectsApi() r = api.get_namespaced_custom_object( group='pingcap.com', version='v1alpha1', plural='restores', namespace=tc2['metadata']['name'], name=restore['metadata']['name'], ) return r["status"]["timeCompleted"] is not None util.wait_until(poll) logging.info("check restored data") dg.conn_info['host'] = tc2['endpoints'][0] dg.conn_info['port'] = tc2['port'] dg.check() class DataGen(object): def __init__(self, conn_info, n, m, c): self.conn_info = conn_info self.n = n self.m = m self.c = c def setup(self): with self.cursor() as cursor: cursor.execute('create database if not exists backup_data') cursor.execute('use backup_data') cursor.execute("drop table if exists t1, t2") cursor.execute("create table t1(a bigint, b bigint, c bigint)") cursor.execute("create table t2(a bigint, b bigint, c bigint, d bigint, e bigint, f bigint)") for i in range(self.n): cursor.execute("insert into t1 values (%s, %s, %s)", (i, i * 2, i * i)) def task(): with self.cursor('backup_data') as task_cursor: for _ in range(self.m): task_cursor.execute("insert into t2 (select * from t1 ta, t1 tb)") tasks = [threading.Thread(name='gen-data-{}'.format(i), target=task) for i in range(self.c)] for t in tasks: t.start() for t in tasks: t.join() def check(self): with self.cursor('backup_data') as cursor: logging.info("check t1") cursor.execute("select count(*) from t1") assert self.n == cursor.fetchone()[0] cursor.execute("select count(*) from t1 where a * 2 = b") assert self.n == cursor.fetchone()[0] cursor.execute("select count(*) from t1 where a * a = c") assert self.n == cursor.fetchone()[0] logging.info("check t2") cursor.execute("select count(*) from t2") assert self.n * self.n * self.m * self.c == cursor.fetchone()[0] cursor.execute("select avg(a), avg(d) from t2") row = cursor.fetchone() assert row[0] == row[1] @contextlib.contextmanager def cursor(self, db=None): conn = pymysql.connect(database=db, autocommit=True, **self.conn_info) cursor = conn.cursor() try: yield cursor finally: cursor.close() conn.close()