def prepare_binlogs_as_relay_logs_for_pitr()

in mysqloperator/controller/backup/meb/restore_main.py [0:0]

• 34 lines of code
• 8 McCabe index (conditional complexity)

def prepare_binlogs_as_relay_logs_for_pitr(datadir: str,
                                           cluster_name: str,
                                           mebinfo: dict,
                                           logger: logging.Logger):
    storage = _get_storage(mebinfo)
    mebspec = mebinfo['spec']

    options = []
    if "extraOptions" in mebspec:
        options = mebspec["extraOptions"]

    if not "pitr" in mebspec or not "backupFile" in mebspec["pitr"]:
        return

    logger.info("Preparing relay logs for PITR")

    tmpdir = "/tmp/backup-tmp"

    m = meb.MySQLEnterpriseBackup(
        storage,
        options,
        tmpdir
    )

    try:
        m.extract(mebspec['pitr']['backupFile'])

        binlog_base = mebspec["pitr"]["binlogName"] if len(mebspec["pitr"]["binlogName"]) else "binlog"

        pattern = re.compile(rf'^{re.escape(binlog_base)}\.\d{{6}}$')
        binlogs = [l for l in os.listdir(f"{tmpdir}/datadir") if pattern.match(l)]
        binlogs.sort()

        logger.info("Trying to prepare binlogs from backup as relay logs")

        i = 0
        with open(f"{datadir}/{cluster_name}-0-relay-bin-pitr.index", "wt") as relay_index:
            for logfile in binlogs:
                i += 1
                logger.info(f"Preparing for PITR: COPY {tmpdir}/datadir/{logfile.strip()} TO {datadir}/{cluster_name}-0-relay-bin.{i:06}")
                shutil.copy(f"{tmpdir}/datadir/{logfile.strip()}", f"{datadir}/{cluster_name}-0-relay-bin-pitr.{i:06}")
                relay_index.write(f"./{cluster_name}-0-relay-bin-pitr.{i:06}\n")
    finally:
        m.cleanup()