# 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 kopf._cogs.structs.bodies import Body from .. import consts, errors, shellutils, utils, config, mysqlutils from .. import diagnose from ..backup import backup_objects from ..shellutils import DbaWrap from . import cluster_objects, router_objects from .cluster_api import MySQLPod, InnoDBCluster, client import typing from typing import Optional, TYPE_CHECKING, Dict, cast, Callable, Union from logging import Logger if TYPE_CHECKING: from mysqlsh.mysql import ClassicSession from mysqlsh import Dba, Cluster import os import copy import mysqlsh import kopf import datetime import time common_gr_options = { # Abort the server if member is kicked out of the group, which would trigger # an event from the container restart, which we can catch and act upon. # This also makes autoRejoinTries irrelevant. "exitStateAction": "ABORT_SERVER" } def select_pod_with_most_gtids(gtids: Dict[int, str]) -> int: pod_indexes = list(gtids.keys()) pod_indexes.sort(key = lambda a: mysqlutils.count_gtids(gtids[a])) return pod_indexes[-1] class ClusterMutex: def __init__(self, cluster: InnoDBCluster, pod: Optional[MySQLPod] = None, context: str = "n/a"): self.cluster = cluster self.pod = pod self.context = context def __enter__(self, *args): owner_lock_creation_time: datetime.datetime (owner, owner_context, owner_lock_creation_time) = utils.g_ephemeral_pod_state.testset( self.cluster, "cluster-mutex", self.pod.name if self.pod else self.cluster.name, context=self.context) if owner: raise kopf.TemporaryError( f"{self.cluster.name} busy. lock_owner={owner} owner_context={owner_context} lock_created_at={owner_lock_creation_time.isoformat()}", delay=10) def __exit__(self, *args): utils.g_ephemeral_pod_state.set(self.cluster, "cluster-mutex", None, context=self.context) class ClusterController: """ This is the controller for a innodbcluster object. It's the main controller for a cluster and drives the lifecycle of the cluster including creation, scaling and restoring from outages. """ def __init__(self, cluster: InnoDBCluster): self.cluster = cluster self.dba: Optional[Dba] = None self.dba_cluster: Optional[Cluster] = None @property def dba_cluster_name(self) -> str: """Return the name of the cluster as defined in the k8s resource as a InnoDB Cluster compatible name.""" return self.cluster.name.replace("-", "_").replace(".", "_") def publish_status(self, diag: diagnose.ClusterStatus, logger: Logger) -> None: cluster_status = self.cluster.get_cluster_status() if cluster_status and cluster_status["status"] != diag.status.name: self.cluster.info(action="ClusterStatus", reason="StatusChange", message=f"Cluster status changed to {diag.status.name}. {len(diag.online_members)} member(s) ONLINE") type = diag.type if (diag.status == diagnose.ClusterDiagStatus.PENDING or diag.status == diagnose.ClusterDiagStatus.INITIALIZING): if (not self.cluster.parsed_spec.initDB): type = diagnose.ClusterInClusterSetType.PRIMARY elif not self.cluster.parsed_spec.initDB.cluster_set: type = diagnose.ClusterInClusterSetType.PRIMARY else: # TODO: Should we declare it a replica or wait it to be diagnosed as replica, if initDB.cluster_set succeeds? type = diagnose.ClusterInClusterSetType.REPLICA_CANDIDATE logger.info("Publishing cluster status") cluster_status = { "status": diag.status.name, "onlineInstances": len(diag.online_members), "type": type.value, "lastProbeTime": utils.isotime() } self.cluster.set_cluster_status(cluster_status) def probe_status(self, logger: Logger) -> diagnose.ClusterStatus: diag = diagnose.diagnose_cluster(self.cluster, logger) if not self.cluster.deleting: self.publish_status(diag, logger) logger.info(f"cluster probe: status={diag.status} online={diag.online_members}") return diag def probe_status_if_needed(self, changed_pod: MySQLPod, logger: Logger) -> diagnose.ClusterDiagStatus: cluster_probe_time = self.cluster.get_cluster_status("lastProbeTime") member_transition_time = changed_pod.get_membership_info("lastTransitionTime") last_status = self.cluster.get_cluster_status("status") unreachable_states = (diagnose.ClusterDiagStatus.UNKNOWN, diagnose.ClusterDiagStatus.ONLINE_UNCERTAIN, diagnose.ClusterDiagStatus.OFFLINE_UNCERTAIN, diagnose.ClusterDiagStatus.NO_QUORUM_UNCERTAIN, diagnose.ClusterDiagStatus.SPLIT_BRAIN_UNCERTAIN) logger.info(f"cluster_probe_time={cluster_probe_time} member_transition_time={member_transition_time} last_status={last_status}") if cluster_probe_time and member_transition_time and cluster_probe_time <= member_transition_time or last_status in unreachable_states: return self.probe_status(logger).status logger.info("Returning last status") return last_status def probe_member_status(self, pod: MySQLPod, session: 'ClassicSession', joined: bool, logger: Logger) -> None: # TODO use diagnose? minfo = shellutils.query_membership_info(session) member_id, role, status, view_id, version, mcount, rmcount = minfo logger.debug( f"instance probe: role={role} status={status} view_id={view_id} version={version} members={mcount} reachable_members={rmcount}") pod.update_membership_status( member_id, role, status, view_id, version, joined=joined) # TODO if status == "ONLINE": pod.update_member_readiness_gate("ready", True) else: pod.update_member_readiness_gate("ready", False) return minfo def connect_to_primary(self, primary_pod: MySQLPod, logger: Logger) -> 'Cluster': if primary_pod: self.dba = shellutils.connect_dba( primary_pod.endpoint_co, logger, max_tries=2) self.dba_cluster = self.dba.get_cluster() else: # - check if we should consider pod marker for whether the instance joined self.connect_to_cluster(logger, need_primary=True) assert self.dba_cluster return self.dba_cluster def connect_to_cluster(self, logger: Logger, need_primary:bool = False) -> MySQLPod: # Get list of pods and try to connect to one of them def try_connect() -> MySQLPod: last_exc = None offline_pods = [] all_pods = self.cluster.get_pods() for pod in all_pods: if pod.name in offline_pods or pod.deleting: continue try: self.dba = mysqlsh.connect_dba(pod.endpoint_co) if need_primary: res = self.dba.session.run_sql( "SELECT member_role" " FROM performance_schema.replication_group_members" " WHERE member_host = @@report_host") r = res.fetch_one() if r[0] != "PRIMARY": logger.info(f"Primary requested, but {pod.name} is no primary") self.dba.session.close() continue except Exception as e: logger.debug(f"connect_dba: target={pod.name} error={e}") # Try another pod if we can't connect to it last_exc = e continue try: self.dba_cluster = self.dba.get_cluster() logger.info(f"Connected to {pod}") return pod except mysqlsh.Error as e: logger.info( f"get_cluster() from {pod.name} failed: {e}") if e.code == errors.SHERR_DBA_BADARG_INSTANCE_NOT_ONLINE: # This member is not ONLINE, so there's no chance of # getting a cluster handle from it offline_pods.append(pod.name) except Exception as e: logger.info( f"get_cluster() from {pod.name} failed: {e}") # If all pods are connectable but OFFLINE, then we have complete outage and need a reboot if len(offline_pods) == len(all_pods): raise kopf.TemporaryError( "Could not connect to any cluster member", delay=15) if last_exc: raise last_exc raise kopf.TemporaryError( "Could not connect to any cluster member", delay=15) return try_connect() def log_mysql_info(self, pod: MySQLPod, session: 'ClassicSession', logger: Logger) -> None: row = session.run_sql( "select @@server_id, @@server_uuid, @@report_host").fetch_one() server_id, server_uuid, report_host = row try: row = session.run_sql( "select @@globals.gtid_executed, @@globals.gtid_purged").fetch_one() gtid_executed, gtid_purged = row except: gtid_executed, gtid_purged = None, None logger.info( f"server_id={server_id} server_uuid={server_uuid} report_host={report_host} gtid_executed={gtid_executed} gtid_purged={gtid_purged}") def create_cluster(self, seed_pod: MySQLPod, logger: Logger) -> None: logger.info("Creating cluster at %s" % seed_pod.name) assume_gtid_set_complete = False initial_data_source = "blank" if self.cluster.parsed_spec.initDB: # TODO store version # TODO store last known quorum if self.cluster.parsed_spec.initDB.clone: initial_data_source = f"clone={self.cluster.parsed_spec.initDB.clone.uri}" elif self.cluster.parsed_spec.initDB.dump and seed_pod.index == 0: # A : Should we check for index? if self.cluster.parsed_spec.initDB.dump.storage.ociObjectStorage: initial_data_source = f"dump={self.cluster.parsed_spec.initDB.dump.storage.ociObjectStorage.bucketName}" elif self.cluster.parsed_spec.initDB.dump.storage.s3: initial_data_source = f"dump={self.cluster.parsed_spec.initDB.dump.storage.s3.bucketName}" elif self.cluster.parsed_spec.initDB.dump.storage.azure: initial_data_source = f"dump={self.cluster.parsed_spec.initDB.dump.storage.azure.containerName}", elif self.cluster.parsed_spec.initDB.dump.storage.persistentVolumeClaim: initial_data_source = f"dump={self.cluster.parsed_spec.initDB.dump.storage.persistentVolumeClaim}" else: assert 0, "Unknown Dump storage mechanism" elif self.cluster.parsed_spec.initDB.meb: initial_data_source = "meb" elif self.cluster.parsed_spec.initDB.cluster_set: initial_data_source = f"clusterSet={self.cluster.parsed_spec.initDB.cluster_set.uri}" else: print(f"{self.cluster.parsed_spec.initDB=} -> {self.cluster.parsed_spec.initDB.cluster_set=}") assert 0, "Unknown initDB source" else: # We're creating the cluster from scratch, so GTID set is sure to be complete assume_gtid_set_complete = True # The operator manages GR, so turn off start_on_boot to avoid conflicts create_options = { "gtidSetIsComplete": assume_gtid_set_complete, "manualStartOnBoot": True, "memberSslMode": "REQUIRED" if self.cluster.parsed_spec.tlsUseSelfSigned else "VERIFY_IDENTITY", } if not self.cluster.parsed_spec.tlsUseSelfSigned: logger.info("Using TLS GR authentication") rdns = seed_pod.get_cluster().get_tls_issuer_and_subject_rdns() create_options["memberAuthType"] = "CERT_SUBJECT" create_options["certIssuer"] = rdns["issuer"] create_options["certSubject"] = rdns["subject"] else: logger.info("Using PASSWORD GR authentication") create_options.update(common_gr_options) cluster_info = { "initialDataSource" : initial_data_source, "createOptions" : create_options, } self.cluster.update_cluster_info(cluster_info) def should_retry(err): if seed_pod.deleting: return False return True with DbaWrap(shellutils.connect_dba(seed_pod.endpoint_co, logger, is_retriable=should_retry)) as dba: try: self.dba_cluster = dba.get_cluster() # maybe from a previous incomplete create attempt logger.info("Cluster already exists") except: self.dba_cluster = None seed_pod.add_member_finalizer() if not self.dba_cluster: self.log_mysql_info(seed_pod, dba.session, logger) logger.info(f"CREATE CLUSTER: seed={seed_pod.name}, options={create_options}") try: self.dba_cluster = dba.create_cluster( self.dba_cluster_name, create_options) logger.info("create_cluster OK") except mysqlsh.Error as e: # If creating the cluster failed, remove the membership finalizer seed_pod.remove_member_finalizer() # can happen when retrying if e.code == errors.SHERR_DBA_BADARG_INSTANCE_ALREADY_IN_GR: logger.info( f"GR already running at {seed_pod.endpoint}, stopping before retrying...") try: dba.session.run_sql("STOP GROUP_REPLICATION") except mysqlsh.Error as e: logger.info(f"Could not stop GR plugin: {e}") # throw a temporary error for a full retry later raise kopf.TemporaryError( "GR already running while creating cluster but could not stop it", delay=3) raise # Only on first cluster! Not if this is supposed to join some other if not self.cluster.parsed_spec.initDB or not self.cluster.parsed_spec.initDB.cluster_set: try: _ = self.dba_cluster.get_cluster_set() # maybe from a previous incomplete create attempt logger.info("ClusterSet already exists") except: # TODO: what to do if this fails? self.dba_cluster_set = self.dba_cluster.create_cluster_set(self.dba_cluster_name) routing_options = self.cluster.parsed_spec.router.routingOptions for routing_option in routing_options: try: routing_value = routing_options[routing_option] logger.info(f"Setting Router Option [{routing_option}] to [{routing_value}]") self.dba_cluster_set.set_routing_option(routing_option, routing_value) except mysqlsh.Error as e: # We don't fail when setting an option fails logger.warning(f"Failed setting routing option {routing_option} to {routing_value}: {e}") self.probe_member_status(seed_pod, dba.session, True, logger) logger.debug("Cluster created %s" % self.dba_cluster.status()) # if there's just 1 pod, then the cluster is ready... otherwise, we # need to wait until all pods have joined if self.cluster.parsed_spec.instances == 1: self.post_create_actions(dba.session, self.dba_cluster, logger) self.probe_member_status(seed_pod, dba.session, True, logger) def post_create_actions(self, session: 'ClassicSession', dba_cluster: 'Cluster', logger: Logger) -> None: logger.info("cluster_controller::post_create_actions") # create router account user, password = self.cluster.get_router_account() update = True try: session.run_sql("show grants for ?@'%'", [user]) except mysqlsh.Error as e: if e.code == mysqlsh.mysql.ErrorCode.ER_NONEXISTING_GRANT: update = False else: raise logger.debug( f"{'Updating' if update else 'Creating'} router account {user}") dba_cluster.setup_router_account( user, {"password": password, "update": update}) # update read replicas for rr in self.cluster.parsed_spec.readReplicas: logger.debug(f"Setting {rr.name} replicas to {rr.instances}") cluster_objects.update_stateful_set_size(self.cluster, rr, logger) # update the router deployment n = self.cluster.parsed_spec.router.instances if n: logger.debug(f"Setting router replicas to {n}") router_objects.update_size(self.cluster, n, False, logger) def reboot_cluster(self, seed_pod_index: MySQLPod, logger: Logger) -> None: pods = self.cluster.get_pods() seed_pod = pods[seed_pod_index] logger.info(f"Rebooting cluster {self.cluster.name} from pod {seed_pod}...") self.dba = shellutils.connect_dba(seed_pod.endpoint_co, logger) self.log_mysql_info(seed_pod, self.dba.session, logger) seed_pod.add_member_finalizer() self.dba_cluster = self.dba.reboot_cluster_from_complete_outage() logger.info(f"reboot_cluster_from_complete_outage OK.") # rejoin other pods for pod in pods: if pod.index != seed_pod_index: with shellutils.connect_to_pod(pod, logger, timeout=5) as session: try: self.rejoin_instance(pod, session, logger) except: # TODO - verify this will be retried from elsewhere print("==================================================") print(f"INSTANCE REJOIN FAILED for {pod.name}") print("=================================================") import traceback traceback.print_exc() # TODO: May not be in a ClusterSet (old Cluster?) cs = self.dba_cluster.get_cluster_set() cs_status = cs.status() # TODO: is there really a valid case where # cs_status["clusters"][name] won't exist? if cs_status.get("clusters", {}).get(self.cluster.name, {}).get("globalStatus") == "INVALIDATED": try: cs.rejoin_cluster(self.cluster.name) except: print("========================") print("CLUSTERSET REJOIN FAILED") print("========================") import traceback traceback.print_exc() status = self.dba_cluster.status() logger.info(f"Cluster reboot successful. status={status}") self.probe_member_status(seed_pod, self.dba.session, True, logger) def force_quorum(self, seed_pod, logger: Logger) -> None: logger.info( f"Forcing quorum of cluster {self.cluster.name} using {seed_pod.name}...") self.connect_to_primary(seed_pod, logger) self.dba_cluster.force_quorum_using_partition_of(seed_pod.endpoint_co) status = self.dba_cluster.status() logger.info(f"Force quorum successful. status={status}") # TODO Rejoin OFFLINE members def destroy_cluster(self, last_pod, logger: Logger) -> None: logger.info(f"Stopping GR for last cluster member {last_pod.name}") try: with shellutils.connect_to_pod(last_pod, logger, timeout=5) as session: # Just stop GR session.run_sql("STOP group_replication") except Exception as e: logger.warning( f"Error stopping GR at last cluster member, ignoring... {e}") # Remove the pod membership finalizer even if we couldn't do final cleanup # (it's just stop GR, which should be harmless most of the time) last_pod.remove_member_finalizer() return logger.info("Stop GR OK") last_pod.remove_member_finalizer() def reconcile_pod(self, primary_pod: MySQLPod, pod: MySQLPod, logger: Logger) -> None: with DbaWrap(shellutils.connect_dba(pod.endpoint_co, logger)) as pod_dba_session: cluster = self.connect_to_primary(primary_pod, logger) status = diagnose.diagnose_cluster_candidate( self.dba.session, cluster, pod, pod_dba_session, logger) logger.info( f"Reconciling {pod}: state={status.status} deleting={pod.deleting} cluster_deleting={self.cluster.deleting}") if pod.deleting or self.cluster.deleting: return # TODO check case where a member pod was deleted and then rejoins with the same address but different uuid if status.status == diagnose.CandidateDiagStatus.JOINABLE: self.cluster.info(action="ReconcilePod", reason="Join", message=f"Joining {pod.name} to cluster") self.join_instance(pod, pod_dba_session, logger) elif status.status == diagnose.CandidateDiagStatus.REJOINABLE: self.cluster.info(action="ReconcilePod", reason="Rejoin", message=f"Rejoining {pod.name} to cluster") self.rejoin_instance(pod, pod_dba_session.session, logger) elif status.status == diagnose.CandidateDiagStatus.MEMBER: logger.info(f"{pod.endpoint} already a member") self.probe_member_status(pod, pod_dba_session.session, False, logger) elif status.status == diagnose.CandidateDiagStatus.UNREACHABLE: # TODO check if we should throw a tmp error or do nothing logger.error(f"{pod.endpoint} is unreachable") self.probe_member_status(pod, pod_dba_session.session, False, logger) else: # TODO check if we can repair broken instances # It would be possible to auto-repair an instance with errant # transactions by cloning over it, but that would mean these # errants are lost. logger.error(f"{pod.endpoint} is in state {status.status}") self.probe_member_status(pod, pod_dba_session.session, False, logger) def join_instance(self, pod: MySQLPod, pod_dba_session: 'Dba', logger: Logger) -> None: logger.info(f"Adding {pod.endpoint} to cluster") peer_pod = self.connect_to_cluster(logger) self.log_mysql_info(pod, pod_dba_session.session, logger) # TODO - always use clone when dataset is big # With Shell Bug #33900165 fixed we should use "auto" by default # and remove the retry logic below recovery_method = "incremental" if self.cluster.parsed_spec.initDB and self.cluster.parsed_spec.initDB.meb: # With a restore from a MEB backup server might not find the right # binlogs for incremental restore and provision an empty replica # clone does the right thing recovery_method = "clone" add_options = { "recoveryMethod": recovery_method, } # TODO : # add_replica_instance doesn't support cert base auth, thus certSubject works only for group-member-s - WL15056 # If a cluster was created with cert based auth between the group members no replica can join the cluster for option in self.dba_cluster.options()["defaultReplicaSet"]["globalOptions"]: if option["option"] == "memberAuthType" and option["value"] in ["CERT_SUBJECT", "CERT_SUBJECT_PASSWORD"]: rdns = pod.get_cluster().get_tls_issuer_and_subject_rdns() # add_instance() needs only certSubject and but not memberAuthType and certIssuer add_options["certSubject"] = rdns["subject"] if pod.instance_type == "group-member": add_options.update(common_gr_options) logger.info( f"ADD INSTANCE: target={pod.endpoint} instance_type={pod.instance_type} cluster_peer={peer_pod.endpoint} options={add_options}...") pod.add_member_finalizer() report_host = pod_dba_session.session.run_sql('SELECT @@report_host').fetch_one()[0] print(f"DBA SESSION GOES TO {report_host}!") try: if pod.instance_type == "read-replica": self.dba_cluster.add_replica_instance(pod.endpoint, add_options) else: self.dba_cluster.add_instance(pod.endpoint_co, add_options) logger.debug("add_instance OK") except (mysqlsh.Error, RuntimeError) as e: logger.warning(f"add_instance failed: error={e}") # Incremetnal may fail if transactions are missing from binlog # retry using clone add_options["recoveryMethod"] = "clone" logger.warning(f"trying add_instance with clone") try: if pod.instance_type == "read-replica": self.dba_cluster.add_replica_instance(pod.endpoint, add_options) else: self.dba_cluster.add_instance(pod.endpoint_co, add_options) except (mysqlsh.Error, RuntimeError) as e: logger.warning(f"add_instance failed second time: error={e}") raise if pod.instance_type == "read-replica": # This is not perfect, as we don't track this further, but async # replication gives us limited information only pod.update_member_readiness_gate("ready", True) else: with DbaWrap(shellutils.connect_dba(pod.endpoint_co, logger)) as dba_session: # TODO: pod_dba_session may be invalid on caller side if the # pod was provisioned via clone, which may lead to future # bugs, also always using a new connection here is "inefficient" # to a small degree. # In case clone is used and we need a reconnect we have # to communicate that to the caller, else we see bugs in # futre minfo = self.probe_member_status(pod, dba_session.session, True, logger) member_id, role, status, view_id, version, member_count, reachable_member_count = minfo logger.info(f"JOINED {pod.name}: {minfo}") # if the cluster size is complete, ensure routers are deployed if not router_objects.get_size(self.cluster) and member_count == self.cluster.parsed_spec.instances: self.post_create_actions(self.dba.session, self.dba_cluster, logger) def rejoin_instance(self, pod: MySQLPod, pod_session, logger: Logger) -> None: logger.info(f"Rejoining {pod.endpoint} to cluster") if not self.dba_cluster: self.connect_to_cluster(logger) self.log_mysql_info(pod, pod_session, logger) rejoin_options = {} logger.info( f"rejoin_instance: target={pod.endpoint} options={rejoin_options}...") try: self.dba_cluster.rejoin_instance(pod.endpoint, rejoin_options) logger.debug("rejoin_instance OK") except mysqlsh.Error as e: logger.warning(f"rejoin_instance failed: error={e}") raise self.probe_member_status(pod, pod_session, False, logger) def remove_instance(self, pod: MySQLPod, pod_body: Body, logger: Logger, force: bool = False) -> None: try: self.__remove_instance_aux(pod, logger, force) except Exception as e: logger.info(f"Exception {e} caught") pass finally: # Remove the membership finalizer to allow the pod to be removed pod.remove_member_finalizer(pod_body) logger.info(f"Removed finalizer for pod {pod_body['metadata']['name']}") def __remove_instance_aux(self, pod: MySQLPod, logger: Logger, force: bool = False) -> None: print(f"Removing {pod.endpoint} from cluster FORCE={force}") # TODO improve this check other_pods = self.cluster.get_pods() if len(other_pods) == 1 and pod.instance_type == 'group-member': print("There is only one pod left in the cluster. Won't remove it, as this will dissolve the cluster. It will be removed only if the cluster is being deleted.") if len(other_pods) > 1 or (len(other_pods) > 0 and pod.instance_type == 'read-replica'): try: print("connect_to_cluster") peer_pod = self.connect_to_cluster(logger) print(f"peer_pod={peer_pod}") except mysqlsh.Error as e: peer_pod = None if self.cluster.deleting: logger.warning( f"Could not connect to cluster, but ignoring because we're deleting: error={e}") else: logger.error(f"Could not connect to cluster: error={e}") raise if peer_pod: removed = False remove_options = {} if not force: logger.info( f"remove_instance: {pod.name} peer={peer_pod.name} options={remove_options}") try: self.dba_cluster.remove_instance(pod.endpoint, remove_options) removed = True logger.debug("remove_instance OK") except mysqlsh.Error as e: logger.warning(f"remove_instance failed: error={e}") if e.code == mysqlsh.mysql.ErrorCode.ER_OPTION_PREVENTS_STATEMENT: # super_read_only can still be true on a PRIMARY for a # short time raise kopf.TemporaryError( f"{peer_pod.name} is a PRIMARY but super_read_only is ON", delay=5) elif e.code == errors.SHERR_DBA_MEMBER_METADATA_MISSING: # already removed and we're probably just retrying removed = True print(f"removed={removed}") if not removed: remove_options["force"] = True logger.info( f"remove_instance: {pod.name} peer={peer_pod.name} options={remove_options}") try: self.dba_cluster.remove_instance(pod.endpoint, remove_options) logger.info("FORCED remove_instance OK") except mysqlsh.Error as e: logger.warning(f"remove_instance failed: error={e}") if e.code == errors.SHERR_DBA_MEMBER_METADATA_MISSING: pass else: deleting = not self.cluster or self.cluster.deleting if deleting: logger.info( f"force remove_instance failed. Ignoring because cluster is deleted: error={e} peer={peer_pod.name}") else: logger.error( f"force remove_instance failed. error={e} deleting_cluster={deleting} peer={peer_pod.name}") raise except RuntimeError as e: logger.info(f"force remove_instance failed. RuntimeError {e}") if str(e).find("The cluster object is disconnected") == -1: logger.info(f"Can't do anything to remove {pod.name} cleanly") raise else: logger.error( f"Cluster is not available, skipping clean removal of {pod.name}") def repair_cluster(self, pod: MySQLPod, diagnostic: diagnose.ClusterStatus, logger: Logger) -> None: # TODO check statuses where router has to be put down # Restore cluster to an ONLINE state if diagnostic.status == diagnose.ClusterDiagStatus.ONLINE: # Nothing to do return elif diagnostic.status == diagnose.ClusterDiagStatus.ONLINE_PARTIAL: # Nothing to do, rejoins handled on pod events return elif diagnostic.status == diagnose.ClusterDiagStatus.ONLINE_UNCERTAIN: # Nothing to do # TODO maybe delete unreachable pods if enabled? return elif diagnostic.status == diagnose.ClusterDiagStatus.OFFLINE: # Reboot cluster if all pods are reachable if len([g for g in diagnostic.gtid_executed.values() if g is not None]) == len(self.cluster.get_pods()): seed_pod = select_pod_with_most_gtids(diagnostic.gtid_executed) self.cluster.info(action="RestoreCluster", reason="Rebooting", message=f"Restoring OFFLINE cluster through pod {seed_pod}") shellutils.RetryLoop(logger).call(self.reboot_cluster, seed_pod, logger) else: logger.debug(f"Cannot reboot cluster because not all pods are reachable") raise kopf.TemporaryError( f"Cluster cannot be restored because there are unreachable pods", delay=5) elif diagnostic.status == diagnose.ClusterDiagStatus.OFFLINE_UNCERTAIN: # TODO delete unconnectable pods after timeout, if enabled raise kopf.TemporaryError( f"Unreachable members found while in state {diagnostic.status}, waiting...") elif diagnostic.status == diagnose.ClusterDiagStatus.NO_QUORUM: # Restore cluster self.cluster.info(action="RestoreCluster", reason="RestoreQuorum", message="Restoring quorum of cluster") shellutils.RetryLoop(logger).call( self.force_quorum, diagnostic.quorum_candidates[0], logger) elif diagnostic.status == diagnose.ClusterDiagStatus.NO_QUORUM_UNCERTAIN: # Restore cluster # TODO delete unconnectable pods after timeout, if enabled raise kopf.TemporaryError( f"Unreachable members found while in state {diagnostic.status}, waiting...") elif diagnostic.status == diagnose.ClusterDiagStatus.SPLIT_BRAIN: self.cluster.error(action="UnrecoverableState", reason="SplitBrain", message="Cluster is in a SPLIT-BRAIN state and cannot be restored automatically.") # TODO check if recoverable case # Fatal error, user intervention required raise kopf.PermanentError( f"Unable to recover from current cluster state. User action required. state={diagnostic.status}") elif diagnostic.status == diagnose.ClusterDiagStatus.SPLIT_BRAIN_UNCERTAIN: # TODO check if recoverable case and if NOT, then throw a permanent error self.cluster.error(action="UnrecoverableState", reason="SplitBrain", message="Cluster is in state SPLIT-BRAIN with unreachable instances and cannot be restored automatically.") raise kopf.PermanentError( f"Unable to recover from current cluster state. User action required. state={diagnostic.status}") # TODO delete unconnectable pods after timeout, if enabled raise kopf.TemporaryError( f"Unreachable members found while in state {diagnostic.status}, waiting...") elif diagnostic.status == diagnose.ClusterDiagStatus.UNKNOWN: # Nothing to do, but we can try again later and hope something comes back raise kopf.TemporaryError( f"No members of the cluster could be reached. state={diagnostic.status}") elif diagnostic.status == diagnose.ClusterDiagStatus.INVALID: self.cluster.error(action="UnrecoverableState", reason="Invalid", message="Cluster state is invalid and cannot be restored automatically.") raise kopf.PermanentError( f"Unable to recover from current cluster state. User action required. state={diagnostic.status}") elif diagnostic.status == diagnose.ClusterDiagStatus.FINALIZING: # Nothing to do return else: raise kopf.PermanentError( f"Invalid cluster state {diagnostic.status}") def on_router_tls_changed(self) -> None: """ Router pods need to be recreated in order for new certificates to get reloaded. """ pass def on_pod_created(self, pod: MySQLPod, logger: Logger) -> None: print("on_pod_created: probing cluster") diag = self.probe_status(logger) print(f"on_pod_created: pod={pod.name} primary={diag.primary} cluster_state={diag.status}") if diag.status == diagnose.ClusterDiagStatus.INITIALIZING: # If cluster is not yet created, then we create it at pod-0 if pod.index == 0: if self.cluster.get_create_time(): raise kopf.PermanentError( f"Internal inconsistency: cluster marked as initialized, but create requested again") print("Time to create the cluster") shellutils.RetryLoop(logger).call(self.create_cluster, pod, logger) # Mark the cluster object as already created self.cluster.set_create_time(datetime.datetime.now()) else: # Other pods must wait for the cluster to be ready raise kopf.TemporaryError("Cluster is not yet ready", delay=15) elif diag.status in (diagnose.ClusterDiagStatus.ONLINE, diagnose.ClusterDiagStatus.ONLINE_PARTIAL, diagnose.ClusterDiagStatus.ONLINE_UNCERTAIN): print("Reconciling pod") # Cluster exists and is healthy, join the pod to it shellutils.RetryLoop(logger).call( self.reconcile_pod, diag.primary, pod, logger) else: print("Attempting to repair the cluster") self.repair_cluster(pod, diag, logger) # Retry from scratch in another iteration raise kopf.TemporaryError(f"Cluster repair from state {diag.status} attempted", delay=5) def on_pod_restarted(self, pod: MySQLPod, logger: Logger) -> None: diag = self.probe_status(logger) logger.debug( f"on_pod_restarted: pod={pod.name} primary={diag.primary} cluster_state={diag.status}") if diag.status not in (diagnose.ClusterDiagStatus.ONLINE, diagnose.ClusterDiagStatus.ONLINE_PARTIAL): self.repair_cluster(pod, diag, logger) shellutils.RetryLoop(logger).call( self.reconcile_pod, diag.primary, pod, logger) def on_pod_deleted(self, pod: MySQLPod, pod_body: Body, logger: Logger) -> None: diag = self.probe_status(logger) print(f"on_pod_deleted: pod={pod.name} primary={diag.primary} cluster_state={diag.status} cluster.deleting={self.cluster.deleting}") if self.cluster.deleting: # cluster is being deleted, if this is pod-0 shut it down if pod.index == 0: self.destroy_cluster(pod, logger) pod.remove_member_finalizer(pod_body) return if pod.deleting and diag.status in (diagnose.ClusterDiagStatus.ONLINE, diagnose.ClusterDiagStatus.ONLINE_PARTIAL, diagnose.ClusterDiagStatus.ONLINE_UNCERTAIN, diagnose.ClusterDiagStatus.FINALIZING): print(f"REMOVING INSTANCE {pod.name}") shellutils.RetryLoop(logger).call( self.remove_instance, pod, pod_body, logger) else: print("ATTEMPTING CLUSTER REPAIR") self.repair_cluster(pod, diag, logger) # Retry from scratch in another iteration print("RETRYING ON POD DELETE") raise kopf.TemporaryError(f"Cluster repair from state {diag.status} attempted", delay=3) # TODO maybe not needed? need to make sure that shrinking cluster will be reported as ONLINE self.probe_status(logger) def on_group_view_change(self, members: list[tuple], view_id_changed) -> None: """ Query membership info about the cluster and update labels and annotations in each pod. This is for monitoring only and should not trigger any changes other than in informational k8s fields. """ for pod in self.cluster.get_pods(): info = pod.get_membership_info() if info: pod_member_id = info.get("memberId") else: pod_member_id = None for member_id, role, status, view_id, endpoint, version in members: if pod_member_id and member_id == pod_member_id: pass elif endpoint == pod.endpoint: pass else: continue pod.update_membership_status( member_id, role, status, view_id, version) if status == "ONLINE": pod.update_member_readiness_gate("ready", True) else: pod.update_member_readiness_gate("ready", False) break def on_server_image_change(self, version: str) -> None: return self.on_upgrade(version = version) def on_server_version_change(self, version: str) -> None: return self.on_upgrade(version = version) def on_upgrade(self, version: str) -> None: # TODO change status as needed - especially on version error, but make sure we recover [version_valid, version_error] = utils.version_in_range(version) if not version_valid: raise kopf.PermanentError(version_error) def on_router_upgrade(self, logger: Logger) -> None: def on_nonupdated() -> None: raise kopf.TemporaryError(f"Cluster {self.cluster.namespace}/{self.cluster.name} unreachable", delay=5) router_objects.update_router_account(self.cluster, on_nonupdated, logger) def on_change_metrics_user(self, logger: Logger) -> None: metrics = self.cluster.parsed_spec.metrics self.connect_to_primary(None, logger) if not metrics or not metrics.enable: # This will use default name. needs to adapt when supporting custom # names mysqlutils.remove_metrics_user(self.dba.session) return user = metrics.dbuser_name grants = metrics.dbuser_grants max_connections = metrics.dbuser_max_connections mysqlutils.setup_metrics_user(self.dba.session, user, grants, max_connections) def on_router_pod_delete(self, name: Union[str, list], logger: Logger) -> None: self.connect_to_cluster(logger) if type(name) is str: router_list = [name] elif type(name) is list: router_list = name for router_name in router_list: logger.info(f"Removing metadata for router {router_name} from {self.cluster.name}") ret = self.dba_cluster.remove_router_metadata(router_name + '::') logger.info(f"remove_router_metadata returned {ret}") def on_router_routing_option_chahnge(self, old: dict, new: dict, logger: Logger) -> None: self.connect_to_primary(None, logger) dba = self.dba_cluster try: dba = self.dba_cluster.get_cluster_set() logger.info("ClusterSet enabled cluster") except: logger.info("Old cluster without clusterset") pass # Unset removed entries for key in old: if key not in new: try: dba.set_routing_option(key, None) except mysqlsh.Error as e: # We don't fail when setting an option fails logger.warning(f"Failed unsetting routing option {key}: {e}") # Set new values, this resets existing values for key in new: try: dba.set_routing_option(key, new[key]) except mysqlsh.Error as e: # We don't fail when setting an option fails logger.warning(f"Failed setting routing option {key} to {new[key]}: {e}")