in mysqloperator/controller/innodbcluster/cluster_api.py [0:0]
def _load(self, spec_root: dict, spec_specific: dict, where_specific: str) -> None:
# initialize now or all instances will share the same list initialized before the ctor
self.add_to_initconf_cbs: dict[str, List[AddToInitconfHandler]] = {}
self.remove_from_sts_cbs: dict[str, List[RemoveFromStsHandler]] = {}
self.add_to_sts_cbs: dict[str, List[AddToStsHandler]] = {}
self.get_configmaps_cbs: dict[str, List[GetConfigMapHandler]] = {}
self.get_add_to_svc_cbs: dict[str, List[AddToSvcHandler]] = {}
self.get_svc_monitor_cbs: dict[str, List[GetSvcMonitorHandler]] = {}
self.secretName = dget_str(spec_root, "secretName", "spec")
if "tlsCASecretName" in spec_root:
self.tlsCASecretName = dget_str(spec_root, "tlsCASecretName", "spec")
else:
self.tlsCASecretName = f"{self.name}-ca"
if "tlsSecretName" in spec_root:
self.tlsSecretName = dget_str(spec_root, "tlsSecretName", "spec")
else:
self.tlsSecretName = f"{self.name}-tls"
if "tlsUseSelfSigned" in spec_root:
self.tlsUseSelfSigned = dget_bool(spec_root, "tlsUseSelfSigned", "spec")
self.instances = dget_int(spec_specific, "instances", where_specific)
if "version" in spec_specific:
self.version = dget_str(spec_specific, "version", where_specific)
if "edition" in spec_root:
self.edition = dget_enum(
spec_root, "edition", "spec", default_value=config.OPERATOR_EDITION,
enum_type=Edition)
if "imageRepository" not in spec_root:
self.imageRepository = config.DEFAULT_IMAGE_REPOSITORY
if "imagePullPolicy" in spec_root:
self.imagePullPolicy = dget_enum(
spec_root, "imagePullPolicy", "spec",
default_value=config.default_image_pull_policy,
enum_type=ImagePullPolicy)
if "imagePullSecrets" in spec_root:
self.imagePullSecrets = dget_list(
spec_root, "imagePullSecrets", "spec", content_type=dict)
self.serviceAccountName = dget_str(spec_root, "serviceAccountName", "spec", default_value=f"{self.name}-sidecar-sa")
self.roleBindingName = f"{self.name}-sidecar-rb"
self.switchoverServiceAccountName = "mysql-switchover-sa"
self.switchoverRoleBindingName = "mysql-switchover-rb"
if "imageRepository" in spec_root:
self.imageRepository = dget_str(spec_root, "imageRepository", "spec")
if "podSpec" in spec_specific: # TODO - replace with something more specific
self.podSpec = dget_dict(spec_specific, "podSpec", where_specific)
if "podAnnotations" in spec_specific:
self.podAnnotations = dget_dict(spec_specific, "podAnnotations", where_specific)
if "podLabels" in spec_specific:
self.podLabels = dget_dict(spec_specific, "podLabels", where_specific)
if "datadirVolumeClaimTemplate" in spec_specific:
self.datadirVolumeClaimTemplate = dget_dict(spec_specific, "datadirVolumeClaimTemplate", where_specific)
self.keyring = KeyringSpec(self.namespace, self.name)
if "keyring" in spec_root:
self.keyring.parse(dget_dict(spec_root, "keyring", "spec"), "spec.keyring")
if "mycnf" in spec_root:
self.mycnf = dget_str(spec_root, "mycnf", "spec")
self.metrics = MetriscSpec(self.namespace, self.cluster_name)
if "metrics" in spec_root:
self.metrics.parse(dget_dict(spec_root, "metrics", "spec"), "spec.metrics")
if cb := self.metrics.get_configmaps_cb():
self.get_configmaps_cbs[InnoDBClusterSpecProperties.METRICS.value] = [cb]
if cb := self.metrics.get_add_to_sts_cb():
self.add_to_sts_cbs[InnoDBClusterSpecProperties.METRICS.value] = [cb]
if cb := self.metrics.get_add_to_svc_cb():
self.get_add_to_svc_cbs[InnoDBClusterSpecProperties.METRICS.value] = [cb]
if cb := self.metrics.get_svc_monitor_cb():
self.get_svc_monitor_cbs[InnoDBClusterSpecProperties.METRICS.value] = [cb]
self.logs = LogsSpec(self.namespace, self.cluster_name)
if (section:= InnoDBClusterSpecProperties.LOGS.value) in spec_root:
self.logs.parse(dget_dict(spec_root, section, "spec"), f"spec.{section}", getLogger())
if cb := self.logs.get_configmaps_cb():
self.get_configmaps_cbs[InnoDBClusterSpecProperties.LOGS.value] = [cb]
if cb := self.logs.get_add_to_initconf_cb():
self.add_to_initconf_cbs[InnoDBClusterSpecProperties.LOGS.value] = [cb]
if cb := self.logs.get_remove_from_sts_cb():
self.remove_from_sts_cbs[InnoDBClusterSpecProperties.LOGS.value] = [cb]
if cb := self.logs.get_add_to_sts_cb():
self.add_to_sts_cbs[InnoDBClusterSpecProperties.LOGS.value] = [cb]
# Initialization Options
section = InnoDBClusterSpecProperties.INITDB.value
if section in spec_root:
self.load_initdb(dget_dict(spec_root, section, "spec"))
# TODO keep a list of base_server_id in the operator to keep things globally unique?
if "baseServerId" in spec_specific:
self.baseServerId = dget_int(spec_specific, "baseServerId", "spec")
self.serviceFqdnTemplate = None
if "serviceFqdnTemplate" in spec_root:
self.serviceFqdnTemplate = dget_str(spec_root, "serviceFqdnTemplate",
"spec",
default_value="{service}.{namespace}.svc.{domain}")
self.dataDirPermissions = DataDirPermissionsSpec()
section = "datadirPermissions"
if section in spec_root:
self.dataDirPermissions.parse(dget_dict(spec_root, section, "spec"), f"spec.{section}")
self.instanceService = InstanceServiceSpec()
section = InnoDBClusterSpecProperties.INSTANCE_SERVICE.value
if section in spec_root:
self.instanceService.parse(dget_dict(spec_root, section, "spec"), f"spec.{section}")