# Copyright (c) 2020, 2024, Oracle and/or its affiliates. # # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/ # from re import L from kubernetes.client.rest import ApiException from .innodbcluster.cluster_api import InnoDBCluster, MySQLPod import typing from enum import Enum from typing import Optional, TYPE_CHECKING, Tuple, List, Set, Dict, cast from . import shellutils, consts, errors import kopf import mysqlsh import enum import time if TYPE_CHECKING: from mysqlsh.mysql import ClassicSession from mysqlsh import Dba, Cluster mysql = mysqlsh.mysql # # InnoDB Cluster Instance Diagnostic Statuses # class ClusterInClusterSetType(Enum): PRIMARY = "PRIMARY" REPLICA = "REPLICA" PRIMARY_CANDIDATE = "PRIMARY_CANDIDATE" REPLICA_CANDIDATE = "REPLICA_CANDIDATE" UNKNOWN = "UNKNOWN" class InstanceDiagStatus(enum.Enum): # GR ONLINE ONLINE = "ONLINE" # GR RECOVERING RECOVERING = "RECOVERING" # GR ERROR ERROR = "ERROR" # GR OFFLINE or any indication that makes it's certain that GR is not ONLINE or RECOVERING OFFLINE = "OFFLINE" # Instance is not a member (and never was) # in addition to being OFFLINE NOT_MANAGED = "NOT_MANAGED" # Instance of an unmanaged replication group. Probably was already member but got removed UNMANAGED = "UNMANAGED" # Instance is not reachable, maybe networking issue UNREACHABLE = "UNREACHABLE" # Uncertain because we can't connect or query it UNKNOWN = "UNKNOWN" class InstanceStatus: pod: Optional[MySQLPod] = None status: InstanceDiagStatus = InstanceDiagStatus.UNKNOWN connect_error: Optional[int] = None view_id: Optional[str] = None is_primary: Optional[bool] = None in_quorum: Optional[bool] = None peers: Optional[dict] = None gtid_executed: Optional[str] = None cluster_in_cluster_set_type: ClusterInClusterSetType = ClusterInClusterSetType.UNKNOWN def __repr__(self) -> str: return f"InstanceStatus: pod={self.pod} status={self.status} connect_error={self.connect_error} view_id={self.view_id} is_primary={self.is_primary} in_quorum={self.in_quorum} peers={self.peers}" def diagnose_instance(pod: MySQLPod, logger, dba: 'Dba' = None) -> InstanceStatus: """ Check state of an instance in the given pod. State is checked in isolation. That is, using its own view of the group and its own local copy of the metadata (if there is one). Thus, it can be incorrect, if for example, the pod was deleted and didn't have its copy of the metadata updated or if there's a split-brain. """ status = InstanceStatus() status.pod = pod if not dba: try: dba = mysqlsh.connect_dba(pod.endpoint_co) except mysqlsh.Error as e: logger.info(f"Could not connect to {pod.endpoint}: error={e}") status.connect_error = e.code if mysql.ErrorCode.CR_MAX_ERROR >= e.code >= mysql.ErrorCode.CR_MIN_ERROR: # client side errors mean we can't connect to the server, but the # problem could be in the client or network and not the server # Check status of the pod pod.reload() logger.debug(f"{pod.endpoint}: pod.phase={pod.phase} deleting={pod.deleting}") if pod.phase != "Running" or not pod.check_containers_ready() or pod.deleting: # not ONLINE for sure if the Pod is not running status.status = InstanceDiagStatus.OFFLINE else: if shellutils.check_fatal_connect(e, pod.endpoint_url_safe, logger): raise return status cluster = None if dba: status.gtid_executed = dba.session.run_sql("select @@gtid_executed").fetch_one()[0] try: # TODO: we want to check from individual Pod's/Server's perspective # it will now check based from primary N times cluster = dba.get_cluster() except mysqlsh.Error as e: logger.info(f"get_cluster() error for {pod.endpoint}: error={e}") # TODO check for invalid metadata errors # Note: get_cluster() on a member that was previously removed # can fail as OFFLINE instead of NOT_MANAGED if its copy of the # metadata lacks the trx where it was removed if e.code == errors.SHERR_DBA_BADARG_INSTANCE_NOT_ONLINE: status.status = InstanceDiagStatus.OFFLINE elif e.code in (errors.SHERR_DBA_BADARG_INSTANCE_NOT_MANAGED, errors.SHERR_DBA_METADATA_NOT_FOUND): status.status = InstanceDiagStatus.NOT_MANAGED else: if shellutils.check_fatal( e, pod.endpoint_url_safe, "get_cluster()", logger): raise status.status = InstanceDiagStatus.UNKNOWN except RuntimeError as e: e_str = str(e) if e_str.find("unmanaged replication group"): status.status = InstanceDiagStatus.UNMANAGED else: logger.info(f"diagnose_instance: 2 Runtime Error [{e}]") status.status = InstanceDiagStatus.UNKNOWN if cluster: try: mstatus = cluster.status({"extended": 1}) status.cluster_in_cluster_set_type = ClusterInClusterSetType.PRIMARY if "clusterRole" in mstatus: logger.info("9.3.0+ cluster, ClusterSet enabled") if mstatus["clusterRole"] == "REPLICA": status.cluster_in_cluster_set_type = ClusterInClusterSetType.REPLICA else: status.cluster_in_cluster_set_type = ClusterInClusterSetType.UNKNOWN else: logger.info("pre 9.3.0 cluster, not ClusterSet enabled") cluster_status = mstatus["defaultReplicaSet"]["status"] status.view_id = mstatus["defaultReplicaSet"]["groupViewId"] if cluster_status.startswith("OK"): status.in_quorum = True else: logger.info( f"""No quorum visible from {pod.endpoint}: status={cluster_status} topology={";".join([f'{m},{i["status"]}' for m, i in mstatus["defaultReplicaSet"]["topology"].items()])}""") status.in_quorum = False members = {} mystate = None for member, info in mstatus["defaultReplicaSet"]["topology"].items(): if pod.instance_type == "group-member": state = info["status"] members[member] = state if member == pod.endpoint: mystate = state if state == "ONLINE": status.is_primary = info["memberRole"] == "PRIMARY" elif pod.instance_type == "read-replica": if "readReplicas" in info: for rr_member, rr_info in info["readReplicas"].items(): if rr_member == pod.endpoint: mystate = rr_info["status"] else: raise Exception(f"Unknown instance type for {pod.name}: {pod.instance_type}") if not mystate: # TODO raise Exception( f"Could not find {pod} in local cluster.status() output") status.peers = members if mystate == "ONLINE": status.status = InstanceDiagStatus.ONLINE elif mystate == "RECOVERING": status.status = InstanceDiagStatus.RECOVERING elif mystate == "ERROR": status.status = InstanceDiagStatus.ERROR elif mystate == "OFFLINE": status.status = InstanceDiagStatus.OFFLINE elif mystate == "UNREACHABLE": status.status = InstanceDiagStatus.UNREACHABLE else: logger.error(f"{pod.endpoint}: bad state {mystate}") assert False, f"{pod.endpoint}: bad state {mystate}" except mysqlsh.Error as e: if shellutils.check_fatal( e, pod.endpoint_url_safe, "status()", logger): raise logger.info(f"status() failed at {pod.endpoint}: error={e}") status.status = InstanceDiagStatus.UNKNOWN return status # # InnoDB Cluster Candidate Instance Statuses # class CandidateDiagStatus(enum.Enum): UNKNOWN = None # Instance is already a member of the cluster MEMBER = "MEMBER" # Instance is a member of the cluster but can rejoin it REJOINABLE = "REJOINABLE" # Instance is not yet a member of the cluster but can join it JOINABLE = "JOINABLE" # Instance is a member of the cluster but has a problem that prevents it # from rejoining BROKEN = "BROKEN" # Instance is not yet a member of the cluster and can't join it UNSUITABLE = "UNSUITABLE" # Instance can't be reached UNREACHABLE = "UNREACHABLE" class CandidateStatus: status: CandidateDiagStatus = CandidateDiagStatus.UNKNOWN # Reasons for broken/unsuitable bad_gtid_set: Optional[str] = None def check_errant_gtids(primary_session: 'ClassicSession', pod: MySQLPod, pod_dba: 'Dba', logger) -> Optional[str]: try: gtid_set = pod_dba.session.run_sql( "SELECT @@globals.GTID_EXECUTED").fetch_one()[0] except mysqlsh.Error as e: if e.code == mysql.ErrorCode.ER_UNKNOWN_SYSTEM_VARIABLE: return None else: raise if gtid_set: # find primary errants = primary_session.run_sql( "SELECT GTID_SUBTRACT(?, @@globals.GTID_EXECUTED)", [gtid_set]).fetch_one()[0] return errants return None def diagnose_cluster_candidate(primary_session: 'ClassicSession', cluster: 'Cluster', pod: MySQLPod, pod_dba: 'Dba', logger) -> CandidateStatus: """ Check status of an instance that's about to be added to the cluster or rejoin it, relative to the given cluster. Also checks whether the instance can join it. """ status = CandidateStatus() istatus = diagnose_instance(pod, logger, pod_dba) if istatus.status == InstanceDiagStatus.UNKNOWN: status.status = CandidateDiagStatus.UNREACHABLE elif istatus.status in (InstanceDiagStatus.ONLINE, InstanceDiagStatus.RECOVERING): status.status = CandidateDiagStatus.MEMBER logger.debug(f"{pod} is {istatus.status} -> {status.status}") elif istatus.status in (InstanceDiagStatus.NOT_MANAGED, InstanceDiagStatus.UNMANAGED): status.bad_gtid_set = check_errant_gtids( primary_session, pod, pod_dba, logger) if status.bad_gtid_set: logger.warning( f"{pod} has errant transactions relative to the cluster: errant_gtids={status.bad_gtid_set}") if not status.bad_gtid_set: status.status = CandidateDiagStatus.JOINABLE else: status.status = CandidateDiagStatus.UNSUITABLE logger.debug( f"{pod} is {istatus.status}, errant_gtids={status.bad_gtid_set} -> {status.status}") elif istatus.status in (InstanceDiagStatus.OFFLINE, InstanceDiagStatus.ERROR): if istatus.status == InstanceDiagStatus.ERROR: # check for fatal GR errors fatal_error = None else: fatal_error = None status.bad_gtid_set = check_errant_gtids( primary_session, pod, pod_dba, logger) if status.bad_gtid_set: logger.warning( f"{pod} has errant transactions relative to the cluster: errant_gtids={status.bad_gtid_set}") # TODO disable queryMembers if pod.endpoint in cluster.status()["defaultReplicaSet"]["topology"].keys(): # already a member of the cluster if not status.bad_gtid_set and not fatal_error: status.status = CandidateDiagStatus.REJOINABLE else: status.status = CandidateDiagStatus.BROKEN else: if not status.bad_gtid_set and not fatal_error: status.status = CandidateDiagStatus.JOINABLE else: status.status = CandidateDiagStatus.UNSUITABLE logger.debug( f"{pod} is {istatus.status} errant_gtids={status.bad_gtid_set} fatal_error={fatal_error} -> {status.status}") else: raise Exception( f"Unexpected pod state pod={pod} status={istatus.status}") return status # # InnoDB Cluster Diagnostic Statuses # class ClusterDiagStatus(enum.Enum): ONLINE = "ONLINE" # - All members are reachable or part of the quorum # - Reachable members form a quorum between themselves # - There are no unreachable members that are not in the quorum # - All members are ONLINE ONLINE_PARTIAL = "ONLINE_PARTIAL" # - All members are reachable or part of the quorum # - Some reachable members form a quorum between themselves # - There may be members outside of the quorum in any state, but they must not form a quorum # Note that there may be members that think are ONLINE, but minority in a view with UNREACHABLE members OFFLINE = "OFFLINE" # - All members are reachable # - All cluster members are OFFLINE/ERROR (or being deleted) # - GTID set of all members are consistent # We're sure that the cluster is completely down with no quorum hiding somewhere # The cluster can be safely rebooted NO_QUORUM = "NO_QUORUM" # - All members are reachable # - All cluster members are either OFFLINE/ERROR or ONLINE but with no quorum # A split-brain with no-quorum still falls in this category # The cluster can be safely restored SPLIT_BRAIN = "SPLIT_BRAIN" # - Some but not all members are unreachable # - There are multiple ONLINE/RECOVERING members forming a quorum, but with >1 # different views # If some members are not reachable, they could either be forming more errant # groups or be unavailable, but that doesn't make much dfifference. ONLINE_UNCERTAIN = "ONLINE_UNCERTAIN" # - Some members are unreachable # - Reachable members form a quorum between themselves # - There are unreachable members that are not in the quorum and have unknown state # Because there are members with unknown state, the possibility that there's a # split-brain exists. OFFLINE_UNCERTAIN = "OFFLINE_UNCERTAIN" # OFFLINE with unreachable members NO_QUORUM_UNCERTAIN = "NO_QUORUM_UNCERTAIN" # NO_QUORUM with unreachable members SPLIT_BRAIN_UNCERTAIN = "SPLIT_BRAIN_UNCERTAIN" # SPLIT_BRAIN with unreachable members UNKNOWN = "UNKNOWN" # - No reachable/connectable members # We have no idea about the state of the cluster, so nothing can be done about it # (even if we wanted) INITIALIZING = "INITIALIZING" # - Cluster is not marked as initialized in Kubernetes # The cluster hasn't been created/initialized yet, so we can safely create it FINALIZING = "FINALIZING" # - Cluster object is marked as being deleted INVALID = "INVALID" # - A (currently) undiagnosable and unrecoverable mess that doesn't fit any other state PENDING = "PENDING" def find_group_partitions(online_pod_info: Dict[str, InstanceStatus], pods: Set[MySQLPod], logger) -> Tuple[List[List[InstanceStatus]], List[Set[MySQLPod]]]: # List of group partitions that have quorum and can execute transactions. # If there's more than 1, then there's a split-brain. If there's none, then # we have no availability. active_partitions: List[List[InstanceStatus]] = [] # List of group partitions that have no quorum and can't execute transactions. blocked_partitions: List[Set[MySQLPod]] = [] all_pods = {} for pod in pods: all_pods[pod.endpoint] = pod no_primary_active_partitions = [] for ep, p in online_pod_info.items(): # logger.info(f"{ep}: {'QUORUM' if p.in_quorum else 'NOQUORUM'} {'PRIM' if p.is_primary else 'SEC'} ONLINE_PODS={online_pod_info.keys()}") # logger.info(f"PEERS OF {ep}={p.peers}") if p.in_quorum: online_peers = [peer for peer, state in p.peers.items() if state in ("ONLINE", "RECOVERING")] # A: UNMANAGED ? missing = set(online_peers) - set(online_pod_info.keys()) if missing: logger.info( f"Group view of {ep} has {p.peers.keys()} but these are not ONLINE: {missing}") raise kopf.TemporaryError( "Cluster status results inconsistent", delay=5) part = [online_pod_info[peer] for peer, state in p.peers.items() if state in ("ONLINE", "RECOVERING")] # A: NOT_MANAGED ? if p.is_primary: active_partitions.append(part) else: no_primary_active_partitions.append(part) if not active_partitions and no_primary_active_partitions: # it's possible for a group with quorum to not have a PRIMARY # for a short time if the PRIMARY is removed from the group raise kopf.TemporaryError( "Cluster has quorum but no PRIMARY", delay=10) def active_partition_with(pod): for part in active_partitions: if pod.endpoint in part: return part return None # print() for ep, p in online_pod_info.items(): # print(ep, p.status, p.in_quorum, p.peers) if not p.in_quorum: part = active_partition_with(p) assert not part, f"Inconsistent group view, {p} not expected to be in {part}" part = set([all_pods[peer] for peer, state in p.peers.items() if state not in ("(MISSING)", "UNREACHABLE")]) if part not in blocked_partitions: blocked_partitions.append(part) # print("ACTIVE PARTS", active_partitions) # print("BLOCKED PARTS", blocked_partitions) # print() # sort by partition size blocked_partitions.sort(key=lambda x: len(x), reverse=True) return active_partitions, blocked_partitions class ClusterStatus: status: ClusterDiagStatus = ClusterDiagStatus.UNKNOWN primary: Optional[MySQLPod] = None online_members: List[MySQLPod] = [] quorum_candidates: Optional[list] = None gtid_executed: Dict[int,str] = {} type: ClusterInClusterSetType = ClusterInClusterSetType.UNKNOWN def do_diagnose_cluster(cluster: InnoDBCluster, logger) -> ClusterStatus: if not cluster.deleting: cluster.reload() all_pods = set(cluster.get_pods()) last_known_quorum = cluster.get_last_known_quorum() # TODO last known quorum tracking log_msg = f"Diagnosing cluster {cluster.name} deleting={cluster.deleting} last_known_quorum={last_known_quorum}..." # Check if the cluster has already been initialized create_time = cluster.get_create_time() log_msg += f"create_time={create_time} deleting={cluster.deleting}" if not create_time and not cluster.deleting: cluster_status = ClusterStatus() cluster_status.status = ClusterDiagStatus.INITIALIZING log_msg += f"\nCluster {cluster.name} status={cluster_status.status}" return cluster_status all_member_pods = set() online_pods = set() offline_pods = set() unsure_pods = set() gtid_executed = {} online_pod_statuses = {} for pod in all_pods: # Diagnose the instance even if deleting - so we can remove it from the cluster and later re-add it # if pod.deleting: # logger.info(f"instance {pod} is deleting") # continue status = diagnose_instance(pod, logger) log_msg += f"\ndiag instance {pod} --> {status.status} quorum={status.in_quorum} gtid_executed={status.gtid_executed}" gtid_executed[pod.index] = status.gtid_executed if status.status == InstanceDiagStatus.UNKNOWN: unsure_pods.add(pod) all_member_pods.add(pod) elif status.status in (InstanceDiagStatus.OFFLINE, InstanceDiagStatus.ERROR, InstanceDiagStatus.UNMANAGED): offline_pods.add(pod) all_member_pods.add(pod) elif status.status in (InstanceDiagStatus.ONLINE, InstanceDiagStatus.RECOVERING): online_pod_statuses[pod.endpoint] = status online_pods.add(pod) all_member_pods.add(pod) elif status.status == InstanceDiagStatus.NOT_MANAGED: pass else: all_member_pods.add(pod) logger.error(f"Internal error processing pod {pod}") assert False log_msg += f"\n{cluster.name}: all={all_pods} members={all_member_pods} online={online_pods} offline={offline_pods} unsure={unsure_pods}" assert online_pods.union(offline_pods, unsure_pods) == all_member_pods cluster_status = ClusterStatus() cluster_status.gtid_executed = gtid_executed if online_pods: active_partitions, blocked_partitions = find_group_partitions( online_pod_statuses, all_member_pods, logger) log_msg += f"\nactive_partitions={active_partitions} blocked_partitions={blocked_partitions}" if not active_partitions: # no quorum if unsure_pods: cluster_status.status = ClusterDiagStatus.NO_QUORUM_UNCERTAIN else: cluster_status.status = ClusterDiagStatus.NO_QUORUM if blocked_partitions: cluster_status.quorum_candidates = list(blocked_partitions[0]) elif len(active_partitions) == 1: # ok if unsure_pods: cluster_status.status = ClusterDiagStatus.ONLINE_UNCERTAIN elif offline_pods: cluster_status.status = ClusterDiagStatus.ONLINE_PARTIAL else: cluster_status.status = ClusterDiagStatus.ONLINE cluster_status.online_members = [ p.pod for p in active_partitions[0] if p.pod] for p in active_partitions[0]: if p.is_primary: cluster_status.primary = p.pod break else: # split-brain if unsure_pods: cluster_status.status = ClusterDiagStatus.SPLIT_BRAIN_UNCERTAIN else: cluster_status.status = ClusterDiagStatus.SPLIT_BRAIN cluster_status.online_members = [] for part in active_partitions: cluster_status.online_members += [p.pod for p in part if p.pod] else: if cluster.deleting: cluster_status.status = ClusterDiagStatus.FINALIZING else: if offline_pods: if unsure_pods: cluster_status.status = ClusterDiagStatus.OFFLINE_UNCERTAIN else: cluster_status.status = ClusterDiagStatus.OFFLINE else: cluster_status.status = ClusterDiagStatus.UNKNOWN if cluster_status.status in (ClusterDiagStatus.UNKNOWN, ClusterDiagStatus.OFFLINE, ClusterDiagStatus.OFFLINE_UNCERTAIN, ClusterDiagStatus.SPLIT_BRAIN, ClusterDiagStatus.SPLIT_BRAIN_UNCERTAIN, ClusterDiagStatus.ONLINE_UNCERTAIN, ClusterDiagStatus.NO_QUORUM, ClusterDiagStatus.NO_QUORUM_UNCERTAIN): logger.info(log_msg) cluster_status.type = ClusterInClusterSetType.UNKNOWN else: cluster_status.type = status.cluster_in_cluster_set_type logger.debug(f"Cluster {cluster.name} status={cluster_status.status}") return cluster_status def diagnose_cluster(cluster: InnoDBCluster, logger) -> ClusterStatus: """ Diagnose the state of an InnoDB cluster, assuming it was already initialized. @param cluster - InnoDBCluster object to diagnose @param logger Returns (primary_pod, state) Notes: - we can only give a certain diagnostic if we can actually connect to and query from all pods - network related connectivity errors are not the same as server side errors, because server side errors (auth error, too many connections etc) mean we were able to at least connect to MySQL, so that means the server is up, but we don't know the GR status - we can consider a cluster to be ONLINE if there's a reachable majority, even if there are unreachable members. But the unreachable members could be forming their own split-brained cluster. Thus we separate between ONLINE (no possibility of split-brain) and ONLINE_PARTIAL - NO_QUORUM means there's no quorum anywhere at all, with certainty. If we can connect to only some members and none of them have quorum, then we can't report NO_QUORUM because the unreachable members could have a quorum of their own - If a member is part of the quorum but is not connectable, the state reported by the cluster takes precedence. - Exceptions that indicate there's something wrong with the deployment are bubbled up. For example: - auth error on a pod that's already initialized """ return cast(ClusterStatus, shellutils.RetryLoop(logger).call(do_diagnose_cluster, cluster, logger))