mysqloperator/sidecar_main.py

# Copyright (c) 2020, 2025, Oracle and/or its affiliates.
#
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
#

# Bootstrap sequence:

# 1 - initconf (initContainer)
# 1.1 - initializes MySQL configuration files
#
# 2 - initmysql (initContainer) - via docker container entrypoint script
# 2.1 - initializes MySQL datadir
# 2.2 - create default root account
# 2.3 - create localroot
#
# 3 - mysql - via container entrypoint:
# 3.1 - start mysqld
#
# 4 - sidecar:
# 4.1 - configureInstance()
# 4.2 - initialize db (clone, loadDump etc)
# 4.3 - restart (optional)
# 4.4 - mark as ready
# # TODO move some stuff to initdatadir?

# Initializes a MySQL Pod to be used in an InnoDBCluster managed by Kubernetes
#
# This performs the following tasks and runs instead of the regular
# initialization done by the Docker image entry point:
#  - configureInstance() and create InnoDB Cluster admin account
#  - if this is the seed pod:
#     - populate database from what's specified in initDB
#     - create remote root user
#
# This process will start and restart mysqld as needed, but mysqld will be
# stopped when it exits. It is meant to be called from a initContainer, if
# this script fails, the Pod creation fails.
#
# This script is called whenever the Pod is created, including when the Pod
# is deleted and recreated soon after by the StatefulSet. When a Pod is
# recreated, the volumes may still contain data from the previous run of the
# same Pod of the same StatefulSet (?).
#

import base64
from logging import Logger
from typing import Optional, TYPE_CHECKING, Tuple, cast
import mysqlsh
import sys
import os
import logging
import time
import asyncio
import argparse
import kopf
import json
from threading import Lock

from .controller import utils, mysqlutils, k8sobject, fqdn, config
from .controller.api_utils import Edition
from .controller.backup import meb
from .controller.innodbcluster import initdb
from .controller.innodbcluster.cluster_api import CloneInitDBSpec, DumpInitDBSpec, MebInitDBSpec, InnoDBCluster, MySQLPod, ClusterSetInitDBSpec
from .controller.kubeutils import api_core, client as api_client
from .controller.innodbcluster import router_objects
from .controller.plugins import install_enterprise_plugins, install_enterprise_encryption, install_keyring_udf

if TYPE_CHECKING:
    from mysqlsh.mysql import ClassicSession
    from mysqlsh import Dba

mysql = mysqlsh.mysql


k8sobject.g_component = "sidecar"
k8sobject.g_host = os.getenv("HOSTNAME")

g_cluster_name = None
g_pod_index = 0
g_pod_name = None
g_pod_namespace = None
g_tls_change_underway = False
g_ca_change_underway = False
g_ca_tls_change_underway_lock = Lock()

g_ready_gate = False
g_ready_gate_lock = Lock()

# The time it takes for mysqld to restart after a clone can be very long,
# because it has to apply redo logs. OTOH we monitor the error log to see
# if there's any activity happening, so the timeout is for activity to happen
# not the total time it takes for the server to start.
CLONE_RESTART_TIMEOUT = 60*10

# Path to a file created to indicate server bootstrap was done
BOOTSTRAP_DONE_FILE = "/var/run/mysql/bootstrap-done"


def create_local_accounts(session: 'ClassicSession', logger: Logger):
    """
    Creates:
    - a localroot@localhost account with auth_socket authentication.
    """
    sql = [
        "SET sql_log_bin=0;",
        "CREATE USER IF NOT EXISTS localroot@localhost IDENTIFIED WITH auth_socket AS 'mysql';",
        "GRANT ALL ON *.* TO 'localroot'@localhost WITH GRANT OPTION;",
        "GRANT PROXY ON ''@'' TO 'localroot'@localhost WITH GRANT OPTION;",
        "SET sql_log_bin=1;"
    ]
    logger.info("Creating local accounts")
    for s in sql:
        try:
            ret = session.run_sql(s)
            if ret is not None and ret.warnings != []:
                logger.warning(f"Warnings during execution of {s}: {ret.warnings}")
        except Exception as e:
            logger.error(f"Error executing {s}: {e}")
            raise Exception("Error creating local accounts")


def configure_for_innodb_cluster(dba:'Dba', logger: Logger):
    """
    Configure instance for InnoDB Cluster.
    """
    options = {}
    logger.info("Configuring instance for InnoDB Cluster")
    dba.configure_instance(None, options)
    logger.info("Instance configured")


def wipe_old_innodb_cluster(session: 'ClassicSession', logger: Logger):
    # drop innodb cluster accounts
    try:
        rows = session.run_sql(
            "select attributes->>'$.recoveryAccountUser', attributes->>'$.recoveryAccountHost' from mysql_innodb_cluster_metadata.v2_instances").fetch_all()
        for user, host in rows:
            if user and host:
                logger.info(f"Dropping user {user}@{host}")
                session.run_sql("drop user if exists ?@?", [user, host])
    except mysqlsh.Error as e:
        if e.code in (mysql.ErrorCode.ER_BAD_DB_ERROR, mysql.ErrorCode.ER_NO_SUCH_TABLE):
            pass
        else:
            logger.error(
                f"Could not query for old InnoDB Cluster accounts: {e}")
            raise

    # drop metadata schema if there's one
    logger.info("Dropping cloned mysql_innodb_cluster_metadata schema")
    session.run_sql("drop schema if exists mysql_innodb_cluster_metadata")


def populate_with_clone(datadir: str, session: 'ClassicSession', cluster: InnoDBCluster, clone_spec: CloneInitDBSpec, pod: MySQLPod, logger: Logger):
    """
    Initialize the DB using clone.
    Server may be restarted multiple times but will be back up on return.
    After the clone:
    - If the donor had the metadata schema, it will be dropped.
    - Local accounts will be created if missing.
    - root password will be reset if needed
    """
    logger.info(f"Initializing mysql via clone...")

    start_time = session.run_sql("select now(6)").fetch_one()[0]

    logger.info(f"Starting at {start_time}")
    # TODO monitor clone from a thread and dump progress
    # initdb.monitor_clone(session, start_time, logger)

    initdb.start_clone_seed_pod(
        session, cluster, pod, clone_spec, logger)

    logger.info("Waiting for mysqld to be restarted/shutdown by clone")

    logger.info("Restarting mysqld back up, this may take a while")

    # root credentials for the new instance are supposed to match in donor
    try:
        user, host, password = get_root_account_info(cluster)
    except Exception as e:
        pod.error(action="InitDB", reason="InvalidArgument", message=f"{e}")
        raise

    logger.info(f"Connecting as {user}@{host}")
    session = connect(user, password, logger)

    initdb.finish_clone_seed_pod(session, cluster, logger)

    # create local accounts again since the donor may not have them
    create_local_accounts(session, logger)

    create_admin_account(session, cluster, logger)

    # reset password of the IC admin account
    # this won't be done by create and may be needed if donor contains such an account
    admin_user, admin_pass = cluster.get_admin_account()
    logger.info(f"Resetting password for {admin_user}@%")
    session.run_sql("SET PASSWORD FOR ?@'%'=?", [admin_user, admin_pass])

    wipe_old_innodb_cluster(session, logger)
    # recreate metrics user if needed
    create_metrics_account(session, cluster, logger)

    create_admin_account(session, cluster, logger)

    return session


def populate_with_dump(datadir: str, session: 'ClassicSession', cluster: InnoDBCluster, init_spec: DumpInitDBSpec, pod: MySQLPod, logger: Logger):
    logger.info(f"Initializing mysql from a dump...")

    initdb.load_dump(session, cluster, pod, init_spec, logger)

    # create local accounts again since the donor may not have them
    create_local_accounts(session, logger)

    wipe_old_innodb_cluster(session, logger)

    wipe_old_innodb_cluster(session, logger)

    return session

def meb_do_pitr(datadir:str, session: 'ClassicSession', cluster: InnoDBCluster,
                init_spec: MebInitDBSpec, pod: MySQLPod, logger: Logger):
    """Do point-in-time-recovery (PITR)

    For doing PITR we have to do those things:

    In restore_main donme during init:

    1) Extract the backup image so we get access to the binlogs
    2) move them in as relay logs

    To do here:

    3) set GTIDs to skip, if any
    4) apply relay logs (binlogs from backup) up to target gtid
    5) hold further processing till done
    6) cleanup extracted backup file
    """


    import time

    def is_relay_log_fully_applied(session, logger):
        res = session.run_sql("SHOW REPLICA STATUS")
        replica_status = res.fetch_one_object()

        if not replica_status:
            raise Exception("This server is not configured as a MySQL replica.")

        io_running = replica_status["Replica_IO_Running"]
        sql_running = replica_status["Replica_SQL_Running"]
        sql_running_state = replica_status["Replica_SQL_Running_State"]

        # states by which we end may vary on sever version and exact reason for
        # application of binlogs to end and we are tolerant here. In most cases
        # with MySQL 9.0 and later we should have an empty string ("")
        end_states = [
            "",
            "Slave has read all relay log; waiting for more updates",
            "Replica has read all relay log; waiting for more updates",
            "Replica has stopped"
        ]

        logger.info((f"Applying PITR: IO Running: {io_running}, "
                     f"SQL Running: {sql_running}, SQL State: {sql_running_state}"))

        return io_running == "No" and sql_running == "No" and sql_running_state in end_states


    logger.info("Applying Binary Logs")

    session.run_sql(f"CHANGE REPLICATION SOURCE TO RELAY_LOG_FILE='{cluster.name}-0-relay-bin-pitr.000001', RELAY_LOG_POS=1, SOURCE_HOST='pitr' FOR CHANNEL 'pitr'")

    if (init_spec.pitr_end_term and init_spec.pitr_end_value):
        logger.info(f"START REPLICA SQL_THREAD UNTIL {init_spec.pitr_end_term} = ? FOR CHANNEL 'pitr'",
                        [init_spec.pitr_end_value])
        session.run_sql(f"START REPLICA SQL_THREAD UNTIL {init_spec.pitr_end_term} = ? FOR CHANNEL 'pitr'",
                        [init_spec.pitr_end_value])
    else:
        session.run_sql("START REPLICA SQL_THREAD FOR CHANNEL 'pitr'")

    def growing_sleep():
        """When only short log is to be applied (especially in testcases likely
        we don't want to wait long, however if there is a lot to do we don't
        want to query too quickly (and in consequence spam the log)
        The growth factor here is absolutely hand-wavy and not based on any
        science
        """
        s = 0
        while s < 5:
            s += 0.2
            yield time.sleep(s)

        while s < 15:
            s += 1
            yield time.sleep(s)

        while True:
            yield time.sleep(15)

    for _ in growing_sleep():
        if is_relay_log_fully_applied(session, logger):
            break

    session.run_sql("STOP REPLICA SQL_THREAD FOR CHANNEL 'pitr'").fetch_all()
    session.run_sql("RESET REPLICA ALL").fetch_all()


def populate_with_meb(datadir: str, session: 'ClassicSession',
                      cluster: InnoDBCluster, init_spec: MebInitDBSpec,
                      pod: MySQLPod, logger: Logger):
    if init_spec.pitr_backup_file and len(init_spec.pitr_backup_file):
        meb_do_pitr(datadir, session, cluster, init_spec, pod, logger)

    wipe_old_innodb_cluster(session, logger)

    return session


def populate_by_joining_cluster_set(_, session: 'ClassicSession',
                                    cluster: InnoDBCluster,
                                    init_spec: ClusterSetInitDBSpec,
                                    pod: MySQLPod, logger: Logger):

    # recreate our accounts

    #old_read_only = session.run_sql("SELECT @@super_read_only").fetch_one()[0]
    # we are in a replica cluster, thus should be in super_Read_only mode, but let's play safe
    #try:
    #    # this is somewhat duplicated with initialize() - probably wise to factor out
    #    session.run_sql("SET sql_log_bin=0")
    #    create_local_accounts(session, logger)
    #finally:
    #    session.run_sql("SET GLOBAL super_read_only=?", [old_read_only])
    #    session.run_sql("SET sql_log_bin=1")

    logger.info(f"Joining InnoDB ClusterSet at {init_spec.uri}")
    session = initdb.join_innodb_cluster_set(session, cluster, init_spec, pod,
                                             logger, create_admin_account)

    try:
        user, host, password = get_root_account_info(cluster)
    except Exception as e:
        pod.error(action="InitDB", reason="InvalidArgument", message=f"{e}")
        raise

    logger.info(f"Connecting as {user}@{host}")
    session = connect(user, password, logger)

    return session


def populate_db(datadir: str, session: 'ClassicSession', cluster: InnoDBCluster, pod, logger: Logger) -> 'ClassicSession':
    """
    Populate DB from source specified in the cluster spec.
    Also creates main root account specified by user.

    mysqld may get restarted by clone.
    """
    if cluster.parsed_spec.initDB:
        if cluster.parsed_spec.initDB.clone:
            logger.info("Populate with clone")
            return populate_with_clone(datadir, session, cluster, cluster.parsed_spec.initDB.clone, pod, logger)
        elif cluster.parsed_spec.initDB.dump:
            logger.info("Populate with dump")
            return populate_with_dump(datadir, session, cluster, cluster.parsed_spec.initDB.dump, pod, logger)
        elif cluster.parsed_spec.initDB.meb:
            logger.info("PITR and cleanup after MEB")
            return populate_with_meb(datadir, session, cluster, cluster.parsed_spec.initDB.meb, pod, logger)
        elif cluster.parsed_spec.initDB.cluster_set:
            logging.info("Joining InnoDB ClusterSet")
            return populate_by_joining_cluster_set(datadir, session, cluster, cluster.parsed_spec.initDB.cluster_set, pod, logger)
        else:
            logger.warning("spec.initDB ignored because no supported initialization parameters found")
    logger.info("InitDB not specified")

    return session


def get_root_account_info(cluster: InnoDBCluster) -> Tuple[str, str, str]:
    secrets = cluster.get_user_secrets()
    if secrets:
        user = secrets.data.get("rootUser")
        host = secrets.data.get("rootHost")
        password = secrets.data.get("rootPassword", None)
        if not password:
            raise Exception(
                f"rootPassword missing in secret {cluster.parsed_spec.secretName}")
        if user:
            user = utils.b64decode(user)
        else:
            user = "root"
        if host:
            host = utils.b64decode(host)
        else:
            host = "%"
        password = utils.b64decode(password)

        return user, host, password

    raise Exception(
        f"Could not get secret '{cluster.parsed_spec.secretName}' with for root account information for {cluster.namespace}/{cluster.name}")


def create_root_account(session: 'ClassicSession', pod: MySQLPod, cluster: InnoDBCluster, logger: Logger) -> None:
    """
    Create general purpose root account (owned by user) as specified by user.
    """
    logger.info("Creating root account")
    try:
        user, host, password = get_root_account_info(cluster)
    except Exception as e:
        pod.error(action="InitDB", reason="InvalidArgument", message=f"{e}")
        raise


    # DROP root@localhost which has random password in case create_root_account() is called from initialize
    # or drop the root@localhost from a initDB and set new password, as the one in the dump/clone might differ
    # The `root` user might not have username root thus we need to drop root@localhost created by the MySQL --initialize-only
    # It was created with random password we are not aware of but it doesn't matter. We create an user as specified in the secret
    ret = session.run_sql("DROP USER IF EXISTS root@localhost")
    logger.info(f"DROP USER root@localhost - Warnings {ret.warnings if ret is not None else []}")

    logger.info(f"Creating own root account {user}@{host}")

    ret = session.run_sql("CREATE USER IF NOT EXISTS ?@? IDENTIFIED BY ?", [user, host, password])
    logger.info(f"CREATE USER - Warnings {ret.warnings if ret is not None else []}")

    ret = session.run_sql("GRANT ALL ON *.* TO ?@? WITH GRANT OPTION", [user, host])
    logger.info(f"GRANT ALL - Warnings {ret.warnings if ret is not None else []}")

    ret = session.run_sql("GRANT PROXY ON ''@'' TO ?@? WITH GRANT OPTION", [user, host])
    logger.info(f"GRANT PROXY - Warnings {ret.warnings if ret is not None else []}")



def create_admin_account(session, cluster, logger: Logger):
    """
    Create a super-user to be used by the operator.
    """
    host = "%"
    user, password = cluster.get_admin_account()
    logger.info(f"Creating account {user}@{host}")
    # binlog has to be disabled for this, because we need to create the account
    # independently in all instances (so that we can run configure on them),
    # which would cause diverging GTID sets
    session.run_sql("DROP USER IF EXISTS ?@?", [user, host])
    session.run_sql("CREATE USER ?@? IDENTIFIED BY ?", [user, host, password])
    session.run_sql("GRANT ALL ON *.* TO ?@? WITH GRANT OPTION", [user, host])
    session.run_sql("GRANT PROXY ON ''@'' TO ?@? WITH GRANT OPTION", [user, host])
    logger.info("Admin account created")


def create_metrics_account(session: 'ClassicSession', cluster: InnoDBCluster, logger: Logger):
    """
    Create a user for metrics, if needed
    """
    if not cluster.parsed_spec.metrics or not cluster.parsed_spec.metrics.enable:
        return

    host = "localhost"
    user = cluster.parsed_spec.metrics.dbuser_name
    max_connections = cluster.parsed_spec.metrics.dbuser_max_connections
    grants = cluster.parsed_spec.metrics.dbuser_grants

    logger.info(f"Creating account {user}@{host} using {session.get_uri()}")
    # binlog has to be disabled for this, because we need to create the account
    # independently in all instances (so that metrics are available even on later config failure),
    # which would cause diverging GTID sets
    mysqlutils.setup_metrics_user(session, user, grants, max_connections)
    logger.info("Metrics account created")


def create_backup_account(session: 'ClassicSession', cluster: InnoDBCluster, logger: Logger):
    """
    Create a user for backup
    """
    user, password = cluster.get_backup_account()
    logger.debug(f"Creating backup account {user} using {session.get_uri()}")
    mysqlutils.setup_backup_account(session, user, password)
    logger.info("Backup account created")


def connect(user: str, password: str, logger: Logger, timeout: Optional[int] = 60) -> 'ClassicSession':
    shell = mysqlsh.globals.shell

    i = 0
    while timeout is None or i < timeout:
        try:
            shell.connect(
                {"user": user, "password": password, "scheme": "mysql"})
            logger.info(f"Connect attempt #{i} successful")
            break
        except mysqlsh.Error as e:
            if mysqlutils.is_client_error(e.code):
                logger.warning(f"Connect attempt #{i} failed: {e}")
                time.sleep(2)
            else:
                logger.critical(f"Unexpected MySQL error during connection: {e}")
                raise
        i += 1
    else:
        raise Exception("Could not connect to MySQL server after initialization")

    assert mysqlsh.globals.session

    return mysqlsh.globals.session


def initialize(session: 'ClassicSession', datadir: str, pod: MySQLPod, cluster: InnoDBCluster, logger: Logger) -> None:
    session.run_sql("SET sql_log_bin=0")
    create_root_account(session, pod, cluster, logger)
    create_admin_account(session, cluster, logger)
    session.run_sql("SET sql_log_bin=1")

    user, password = cluster.get_admin_account()
    session = connect(user, password, logger)

    configure_for_innodb_cluster(mysqlsh.globals.dba, logger)

    if pod.index == 0 and cluster.get_create_time() is None:
        # if this is the 1st pod of the cluster, then initialize it and create default accounts
        session = populate_db(datadir, session, cluster, pod, logger)

    # TODO: This will fail after restarting server due to cloning - wonder why
    #       I only noticed this with clusterSet not with plain clone
    #
    #        mysqlsh.DBError: MySQL Error (2013): ClassicSession.run_sql: Lost connection to MySQL server during query

    session.run_sql("SET sql_log_bin=0")
    old_read_only = session.run_sql("SELECT @@super_read_only").fetch_one()[0]
    session.run_sql("SET GLOBAL super_read_only=0")

    try:
        # These need to be created on every pod and not replicated. Thus under sql_log_bin=0
        # Not created earlier, as if there is initdb, for example from dump of clone, the
        # creation will be overwritten and we will need to create them again in the populate method
        create_metrics_account(session, cluster, logger)
        create_backup_account(session, cluster, logger)

        # Some commands like INSTALL [PLUGIN|COMPONENT] are not being
        # replicated we run them on any restart, those have to be idempotent

        # With enterprise edition activate enterprise plugins
        if cluster.parsed_spec.edition == Edition.enterprise:
            install_enterprise_plugins(cluster.parsed_spec.version, session, logger)

        # If a Keyring setup is requested install keyring UDFs
        if "keyring" in cluster.spec:
            print(f"KEYRING: {cluster.spec['keyring']}")
            install_keyring_udf(cluster.parsed_spec.version, session, logger)
    finally:
        session.run_sql("SET GLOBAL super_read_only=?", [old_read_only])
        session.run_sql("SET sql_log_bin=1")


def metadata_schema_version(session: 'ClassicSession', logger: Logger) -> Optional[str]:
    try:
        r = session.run_sql(
            "select * from mysql_innodb_cluster_metadata.schema_version").fetch_one()
        return r[0]
    except Exception as e:
        logger.info(f"Metadata check failed: {e}")
        return None


def bootstrap(pod: MySQLPod, datadir: str, logger: Logger) -> int:
    """
    Prepare MySQL instance for InnoDB cluster.

    This function must be idempotent, because the sidecar container can get
    restarted in many different scenarios. It's also possible that the
    whole pod gets deleted and recreated while its underlying PV is reused.
    In that case, the Pod will look brand new (so we can't rely on any data
    stored in the Pod object), but the instance will be already prepared and
    not be in the expected initial state with initial defaults.

    Returns 1 if bootstrapped, 0 if already configured and -1 on error
    """
    name = pod.name
    namespace = pod.namespace

    # Check if the Pod is already configured according to itself
    gate = pod.get_member_readiness_gate("configured")
    if gate:
        logger.info(f"MySQL server was already initialized configured={gate}")
        return 0

    cluster = pod.get_cluster()

    # In most cases we have a fresh datadir at this point, where only a user
    # localroot exists. A restore using MySQL Shell would only be done later.
    # However if a restore was done using MEB the datadir will be populated
    # with whatever data there was in the backup during init before this runs.
    # In that case we have to get in using credentials passed by the user.
    # We also must be careful about changs we do, as a PITR may follow and
    # changes may break application of binlogs.
    if pod.index == 0 and cluster.get_create_time() is None and cluster.parsed_spec.initDB and cluster.parsed_spec.initDB.meb:
        logger.info("Detected MEB restore, giving ourselves access")
        mebsession = None
        user, _, password = get_root_account_info(cluster)
        logger.info("Using account %s", user)
        try:
            mebsession = connect(user, password, logger, timeout=None)
            create_local_accounts(mebsession, logger)
        finally:
            if mebsession:
                mebsession.close()

    # Connect using localroot and check if the metadata schema already exists

    # note: we may have to wait for mysqld to startup, since the sidecar and
    # mysql containers are started at the same time.
    session = connect("localroot", "", logger, timeout=None)

    # Restore from MEB is done in init and the Backup may contain old metadata
    # mabe we could only check cluster create time in all cases, but that would
    # need a lot more testing in restart situations
    mdver = metadata_schema_version(session, logger)
    initdb_spec = cluster.parsed_spec.initDB
    if mdver and not (initdb_spec and initdb_spec.meb and cluster.get_create_time() is None):
        logger.info(f"InnoDB Cluster metadata (version={mdver}) found, skipping configuration...")
        pod.update_member_readiness_gate("configured", True)
        return 0

    # Check if the datadir already existed

    logger.info(
        f"Configuring mysql pod {namespace}/{name}, configured={gate} datadir={datadir}")

    try:
        initialize(session, datadir, pod, cluster, logger)

        pod.update_member_readiness_gate("configured", True)

        logger.info("Configuration finished")
    except Exception as e:
        import traceback
        traceback.print_exc()
        logger.critical(f"Unhandled exception while bootstrapping MySQL: {e}")
        # TODO post event to the Pod and the Cluster object if this is the seed
        return -1

    session.close()

    return 1

def ensure_correct_tls_sysvars(pod: MySQLPod, session: 'ClassicSession', enabled: bool, caller: str, logger: Logger) -> None:
    has_crl = os.path.exists("/etc/mysql-ssl/crl.pem")

    logger.info(f"Ensuring custom TLS certificates are {'enabled' if enabled else 'disabled'} {'(with crl)' if has_crl else ''} caller={caller}")

    def ensure_sysvar(var, value):
        logger.info(f"\tChecking if {var} is [{value}]")
        res = session.run_sql("SHOW VARIABLES LIKE ?", [var])
        row = res.fetch_one()
        if row:
            curval = row[1]
            if curval != value:
                logger.info(f"\t{var} is [{curval}] persisting to [{value}]")
                session.run_sql(f"SET PERSIST {var} = ?", [value])
        else:
            raise kopf.PermanentError(f"Variable {var} not found!")

    # first ensure configured paths are correct
    if enabled:
        ensure_sysvar("ssl_ca", "/etc/mysql-ssl/ca.pem")
        ensure_sysvar("ssl_crl", "/etc/mysql-ssl/crl.pem" if has_crl else "")
        ensure_sysvar("ssl_cert", "/etc/mysql-ssl/tls.crt")
        ensure_sysvar("ssl_key", "/etc/mysql-ssl/tls.key")
        if pod.instance_type == "group-member":
            ensure_sysvar("group_replication_recovery_ssl_verify_server_cert", "ON")
            ensure_sysvar("group_replication_ssl_mode", "VERIFY_IDENTITY")
            ensure_sysvar("group_replication_recovery_ssl_ca", "/etc/mysql-ssl/ca.pem")
            ensure_sysvar("group_replication_recovery_ssl_cert", "/etc/mysql-ssl/tls.crt")
            ensure_sysvar("group_replication_recovery_ssl_key", "/etc/mysql-ssl/tls.key")
    else:
        ensure_sysvar("ssl_ca", "ca.pem")
        ensure_sysvar("ssl_crl", "")
        ensure_sysvar("ssl_cert", "server-cert.pem")
        ensure_sysvar("ssl_key", "server-key.pem")
        if pod.instance_type == "group-member":
            ensure_sysvar("group_replication_recovery_ssl_verify_server_cert", "OFF")
            ensure_sysvar("group_replication_ssl_mode", "REQUIRED")
            ensure_sysvar("group_replication_recovery_ssl_ca", "")
            ensure_sysvar("group_replication_recovery_ssl_cert", "")
            ensure_sysvar("group_replication_recovery_ssl_key", "")


def reconfigure_tls(pod: MySQLPod, enabled: bool, caller: str, logger: Logger) -> None:

    session = connect("localroot", "", logger, timeout=None)

    ensure_correct_tls_sysvars(pod, session, enabled, caller, logger)

    try:
        logger.info("Reloading TLS")
        session.run_sql("ALTER INSTANCE RELOAD TLS")
    except Exception as e:
        logger.error(f"MySQL error reloading TLS certificates: {e}")
    finally:
        session.close()


def check_secret_mounted(secrets: dict, paths: list, logger: Logger) -> bool:
    logger.info("check_secret_mounted")

    for path in paths:
        secret_name = path.split("/")[-1]
        secret_value = secrets[secret_name]
        if not secret_value is None:
            if os.path.exists(path):
                dataf = open(path).read()
                if dataf != secret_value:
                    logger.info(f"check_secret_mounted: No match for {secret_name}")
                    return False
                logger.info(f"check_secret_mounted: {secret_name} matches")
            else:
                logger.info(f"check_secret_mounted: Path to secret {secret_name} doesn't exist")
                return False
        else:
            logger.info(f"check_secret_mounted: Not checking {path}, expected None value")

    logger.info(f"check_secret_mounted: Success")
    return True


def on_ca_secret_create_or_change(value: dict, useSelfSigned: bool, router_deployment: Optional[api_client.V1Deployment], logger: Logger) -> None:
    global g_pod_name
    global g_pod_namespace

    logger.info("on_ca_secret_create_or_change")

    ca_pem = utils.b64decode(value['data']['ca.pem']) if 'ca.pem' in value['data'] else None
    crl_pem = utils.b64decode(value['data']['crl.pem']) if 'crl.pem' in value['data'] else None
    secrets = {'ca.pem': ca_pem, 'crl.pem': crl_pem}

    max_time = 7 * 60
    delay = 5
    for _ in range(max_time//delay):
        if check_secret_mounted(secrets,
                                ["/etc/mysql-ssl/ca.pem",
                                 "/etc/mysql-ssl/crl.pem"],
                                logger):
            logger.info(f"TLS CA file change detected, reloading TLS configurations")
            pod = MySQLPod.read(g_pod_name, g_pod_namespace)
            reconfigure_tls(pod, False if useSelfSigned else True, "on_ca_secret_create_or_change", logger)

            if router_deployment:
                # give time to all other sidecars to reload the TLS and then restart the router deployment from -0
                time.sleep(delay)
                router_objects.restart_deployment_for_tls(router_deployment, router_tls_crt = None, router_tls_key = None, ca_pem = ca_pem, crl_pem = crl_pem, logger=logger)
            break
        else:
            logger.debug("Waiting for mounted TLS files to refresh...")
            time.sleep(delay)
            # TemporaryError was supposed to get this handler called again, but isn't
            # raise kopf.TemporaryError("TLS CA secret changed, but file didn't refresh yet")
    else:
        raise kopf.PermanentError("Timeout waiting for TLS files to get refreshed")


def on_tls_secret_create_or_change(value: dict, useSelfSigned: bool, router_deployment: Optional[api_client.V1Deployment], logger: Logger) -> None:
    global g_pod_name
    global g_pod_namespace

    logger.info("on_tls_secret_create_or_change")

    pod = MySQLPod.read(g_pod_name, g_pod_namespace)

    tls_crt = utils.b64decode(value['data']['tls.crt']) if 'tls.crt' in value['data'] else None
    tls_key = utils.b64decode(value['data']['tls.key']) if 'tls.key' in value['data'] else None
    secrets = {'tls.crt': tls_crt, 'tls.key': tls_key}

    max_time = 7 * 60
    delay = 5
    for _ in range(max_time//delay):
        if check_secret_mounted(secrets,
                                ["/etc/mysql-ssl/tls.crt",
                                 "/etc/mysql-ssl/tls.key"],
                                logger):
            logger.info(f"TLS certificate file change detected, reloading TLS configurations")
            reconfigure_tls(pod, False if useSelfSigned else True, "on_tls_secret_create_or_change", logger)
            break
        else:
            logger.info("Waiting for mounted TLS files to refresh...")
            time.sleep(delay)
            #raise kopf.TemporaryError("TLS certificate secret changed, but file didn't refresh yet")
    else:
        raise kopf.PermanentError("Timeout waiting for TLS files to get refreshed")


def on_router_tls_secret_create_or_change(value: dict, useSelfSigned: bool, router_deployment: Optional[api_client.V1Deployment], logger: Logger) -> None:
    logger.info("on_router_tls_secret_create_or_change")

    if router_deployment:
        router_tls_crt = utils.b64decode(value['data']['tls.crt']) if 'tls.crt' in value['data'] else None
        router_tls_key = utils.b64decode(value['data']['tls.key']) if 'tls.key' in value['data'] else None
        router_objects.restart_deployment_for_tls(router_deployment, router_tls_crt = router_tls_crt, router_tls_key = router_tls_key, ca_pem = None, crl_pem = None, logger=logger)


def secret_belongs_to_the_cluster_checker(namespace:str, name, **_) -> bool:
    # This should be always true but some precaution won't hurt
    # It is true when the sidecar runs as standalone operator. If for some reason it doesn't
    # that it will listen to all namespaces and then this won't hold true all the time
    if namespace == g_pod_namespace:
        ic = InnoDBCluster.read(namespace, g_cluster_name)
        return name in (ic.parsed_spec.tlsCASecretName, ic.parsed_spec.tlsSecretName, ic.parsed_spec.router.tlsSecretName)
    return False


@kopf.on.create("", "v1", "secrets", when=secret_belongs_to_the_cluster_checker) # type: ignore
@kopf.on.update("", "v1", "secrets", when=secret_belongs_to_the_cluster_checker) # type: ignore
def on_secret_create_or_update(name: str, namespace: str, spec, new, logger: Logger, **kwargs):
    # g_cluster_name
    # g_pod_index
    global g_tls_change_underway
    global g_ca_change_underway
    global g_ca_tls_change_underway_lock
    global g_ready_gate
    global g_ready_gate_lock

    logger.info(f"on_secret_create_or_update: name={name} pod_index={g_pod_index}")

    try:
        g_ready_gate_lock.acquire()
        if not g_ready_gate:
            logger.info("Cached value of gate[ready] is false, re-checking")
            ready = MySQLPod.read(g_pod_name, g_pod_namespace).get_member_readiness_gate("ready")
            if not ready:
                raise kopf.TemporaryError(f"Pod not ready - not yet part of the IC. Will retry", delay=15)
            g_ready_gate = True
            logger.info("Readiness gate 'ready' is true. Handling event.")
    finally:
        g_ready_gate_lock.release()

    ic = InnoDBCluster.read(namespace, g_cluster_name)
    tls_changed = False
    ca_changed = False
    handler = None
    router_deployment = None
    # In case the same secret is used for CA and TLS, and router TLS, then the order
    # here is very important. on_ca_secret_create_or_change() does what
    # on_tls_secret_create_or_change() does and restarts the deployment on top
    # So, either this order of checks or three separate if-statements.
    if ic.parsed_spec.tlsCASecretName == name:
        logger.info(f"on_secret_create_or_update: tlsCASecretName")
        g_ca_tls_change_underway_lock.acquire()
        try:
            if g_tls_change_underway:
                raise kopf.TemporaryError(f"TLS change underway. Wait to finish. {name}", delay=12)
            g_ca_change_underway = True
            ca_changed = True
        finally:
            g_ca_tls_change_underway_lock.release()

        handler = on_ca_secret_create_or_change
        router_deployment = ic.get_router_deployment() if g_pod_index == 0 else None
    elif ic.parsed_spec.tlsSecretName == name:
        logger.info(f"on_secret_create_or_update: tlsSecretName")
        g_ca_tls_change_underway_lock.acquire()
        try:
            if g_ca_change_underway:
                raise kopf.TemporaryError(f"CA change underway. Wait to finish. {name}", delay=14)
            g_tls_change_underway = True
            tls_changed = True
        finally:
            g_ca_tls_change_underway_lock.release()

        handler = on_tls_secret_create_or_change
    elif ic.parsed_spec.router.tlsSecretName == name:
        logger.info(f"on_secret_create_or_update: router.tlsSecretName")
        try:
            g_ca_tls_change_underway_lock.acquire()
            if g_ca_change_underway:
                raise kopf.TemporaryError(f"CA change underway. Wait to finish. {name}", delay=16)
            else:
                handler = on_router_tls_secret_create_or_change
                router_deployment = ic.get_router_deployment() if g_pod_index == 0 else None
        finally:
            g_ca_tls_change_underway_lock.release()

    if handler:
        try:
            handler(new, ic.parsed_spec.tlsUseSelfSigned, router_deployment , logger)
        finally:
            if ca_changed:
                g_ca_tls_change_underway_lock.acquire()
                g_ca_change_underway = False
                g_ca_tls_change_underway_lock.release()
            if tls_changed:
                g_ca_tls_change_underway_lock.acquire()
                g_tls_change_underway = False
                g_ca_tls_change_underway_lock.release()


@kopf.on.startup()
def configure(settings: kopf.OperatorSettings, logger: Logger, *args, **_):
    logger.info("sidecar: configure()")
    settings.peering.standalone = True
    settings.posting.enabled = False


def main(argv):
    global g_cluster_name
    global g_pod_index
    global g_pod_name
    global g_pod_namespace

    # const - when there is an argument without value
    # default - when there is no argument at all
    # nargs = "?" - zero or one arguments
    # nargs = "+" - one or more arguments, returns a list()
    # nargs = 8 - 8 arguments will be consumed
    # nargs = 1 - 1 argument will be consumed, returns a list with one element
    parser = argparse.ArgumentParser(description = "MySQL InnoDB Cluster Instance Sidecar Container")
    parser.add_argument('--debug',  type = int, nargs="?", const = 1, default = 0, help = "Debug")
    parser.add_argument('--logging-level', type = int, nargs="?", default = logging.INFO, help = "Logging Level")
    parser.add_argument('--pod-name', type = str, default = "", help = "Pod Name")
    parser.add_argument('--pod-namespace', type = str, default = "", help = "Pod Namespace")
    parser.add_argument('--datadir', type = str, nargs=1, help = "Path do data directory")
    args = parser.parse_args(argv)

    datadir = args.datadir

    kopf.configure(verbose=True if args.debug != 0 else False)

    mysqlsh.globals.shell.options.useWizards = False
    logging.basicConfig(level=args.logging_level,
                        format='%(asctime)s - [%(levelname)s] [%(name)s] %(message)s',
                        datefmt="%Y-%m-%dT%H:%M:%S")
    logger = logging.getLogger("sidecar")
    utils.log_banner(__file__, logger)

    g_pod_namespace = args.pod_namespace
    g_pod_name = args.pod_name

    name = args.pod_name
    namespace = args.pod_namespace
    pod = MySQLPod.read(name, namespace)
    logger.info(f"My pod is {name} in {namespace}")

    logger.info("Bootstrapping")
    r = bootstrap(pod, datadir, logger)
    if r < 0:
        logger.info(f"Bootstrap error {r}")
        return abs(r)

    cluster = pod.get_cluster()
    cluster.log_tls_info(logger)

    g_cluster_name = cluster.name
    g_pod_index = pod.index

    if r == 0:
        # refresh TLS settings if we're restarting in case something changed
        reconfigure_tls(pod, False if cluster.parsed_spec.tlsUseSelfSigned else True, "main", logger)

    logger.info("Starting Operator request handler...")
    try:
        loop = asyncio.get_event_loop()

        loop.run_until_complete(kopf.operator(namespace=namespace))
    except Exception as e:
        import traceback
        traceback.print_exc()
        logger.critical(f"Unhandled exception while bootstrapping MySQL: {e}")
        # TODO post event to the Pod and the Cluster object if this is the seed
        return 1