mysqloperator/controller/innodbcluster/cluster_objects.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/
#

import random
import string
from logging import Logger, getLogger
import kopf
from typing import List, Dict, Optional, cast
from ..kubeutils import client as api_client
from .. import utils, config, consts
from .cluster_api import InnoDBCluster, AbstractServerSetSpec, InnoDBClusterSpec, ReadReplicaSpec, InnoDBClusterSpecProperties
from .. import fqdn
import yaml
from ..kubeutils import api_core, api_apps, api_customobj, k8s_cluster_domain, ApiException
from . import router_objects
import base64
import os
from ..backup import backup_objects

# TODO replace app field with component (mysqld,router) and tier (mysql)

# This service includes all instances, even those that are not ready

def prepare_cluster_service(spec: AbstractServerSetSpec, logger: Logger) -> dict:
    extra_label = ""
    if type(spec) is InnoDBClusterSpec:
        instance_type = "group-member"
        instances = spec.instances
    elif type(spec) is ReadReplicaSpec:
        instance_type = "read-replica"
        extra_label = f"mysql.oracle.com/read-replica: {spec.name}"
    else:
        raise NotImplementedError(f"Unknown subtype {type(spec)} for creating StatefulSet")

    tmpl = f"""
apiVersion: v1
kind: Service
metadata:
  name: {spec.headless_service_name}
  namespace: {spec.namespace}
  labels:
    tier: mysql
    mysql.oracle.com/cluster: {spec.cluster_name}
    mysql.oracle.com/instance-type: {instance_type}
    {extra_label}
spec:
  clusterIP: None
  publishNotReadyAddresses: true
  ports:
  - name: mysql
    port: {spec.mysql_port}
    targetPort: {spec.mysql_port}
  - name: mysqlx
    port: {spec.mysql_xport}
    targetPort: {spec.mysql_xport}
  - name: gr-xcom
    port: {spec.mysql_grport}
    targetPort: {spec.mysql_grport}
  selector:
    component: mysqld
    tier: mysql
    mysql.oracle.com/cluster: {spec.cluster_name}
    mysql.oracle.com/instance-type: {instance_type}
    {extra_label}
  type: ClusterIP
"""

    svc = yaml.safe_load(tmpl)

    if spec.instanceService.annotations:
        if not 'annotations' in svc['metadata']:
            svc['metadata']['annotations'] = {}
        svc['metadata']['annotations'] = spec.instanceService.annotations

    if spec.instanceService.labels:
        if not 'labels' in svc['metadata']:
            svc['metadata']['labels'] = {}
        svc['metadata']['labels'] = spec.instanceService.labels | svc['metadata']['labels']

    for subsystem in spec.get_add_to_svc_cbs:
        print(f"\t\tChecking subsystem {subsystem}")
        for add_to_svc_cb in spec.get_add_to_svc_cbs[subsystem]:
            print(f"\t\tAdding {subsystem} SVC bits")
            add_to_svc_cb(svc, logger)

    return svc


def prepare_secrets(spec: InnoDBClusterSpec) -> dict:
    def encode(s):
        return base64.b64encode(bytes(s, "ascii")).decode("ascii")

    # TODO: should we share the suffix with router&backup and stor in IC?
    # miught make it simpler to diagnose and remove
    characters = string.ascii_letters + string.digits
    suffix = ''.join(random.choice(characters) for _ in range(10))

    admin_user = encode(config.CLUSTER_ADMIN_USER_NAME + '-' + suffix)
    admin_pwd = encode(utils.generate_password())

    tmpl = f"""
apiVersion: v1
kind: Secret
metadata:
  name: {spec.name}-privsecrets
data:
  clusterAdminUsername: {admin_user}
  clusterAdminPassword: {admin_pwd}
"""
    return yaml.safe_load(tmpl)


def prepare_cluster_pod_disruption_budget(spec: InnoDBClusterSpec) -> dict:
    tmpl = f"""
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: {spec.name}-pdb
spec:
  maxUnavailable: 1
  selector:
    matchLabels:
      component: mysqld
      tier: mysql
      mysql.oracle.com/cluster: {spec.name}
"""
    pdb = yaml.safe_load(tmpl.replace("\n\n", "\n"))

    return pdb

def get_restore_container(cluster: InnoDBCluster, spec: AbstractServerSetSpec, cluster_domain: str):
    if not isinstance(spec, InnoDBClusterSpec):
        # restore happens on a primary, read replica doesn't need a restore container
        return ("", "")

    if not spec.initDB or not spec.initDB.meb:
        # No MEB Restore - no extra container
        return ("", "")

    container = f"""
      - name: restore
        command:
        - mysqlsh
        - --pym
        - meb.restore_main
        - --pod-name
        - "$(POD_NAME)"
        - --pod-namespace
        - "$(POD_NAMESPACE)"
        - --cluster-name
        - "{spec.name}"
        - --datadir
        - /var/lib/mysql

        env:
        - name: POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: POD_NAMESPACE
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace
        - name: MYSQL_OPERATOR_K8S_CLUSTER_DOMAIN
          value: {cluster_domain}
        - name: MYSQLSH_USER_CONFIG_HOME
          value: /tmp
        - name: MYSQLSH_CREDENTIAL_STORE_SAVE_PASSWORDS
          value: never
        image: {spec.mysql_image}
        imagePullPolicy: {spec.mysql_image_pull_policy}
        securityContext:
          allowPrivilegeEscalation: false
          capabilities:
            drop:
            - ALL
          privileged: false
          readOnlyRootFilesystem: true
          runAsGroup: 27
          runAsUser: 27
        volumeMounts:
        - mountPath: /usr/lib/mysqlsh/python-packages/meb
          name: mebcode
        - mountPath: /var/lib/mysql
          name: datadir
        - mountPath: /tmp
          # sharing with initconf to transfer restore information
          name: initconf-tmp
        - mountPath: /mysqlsh
          name: mebmyslhshhome
        - name: rundir
          mountPath: /var/run/mysqld
"""

    volumes = f"""
      - name: mebmyslhshhome
        emptyDir: {{}}
      - name: mebcode
        configMap:
            name: {cluster.name}-mebcode
"""

    return (container, volumes)


def get_meb_container(cluster: InnoDBCluster, spec: InnoDBClusterSpec,
                      cluster_domain: str):
    if not isinstance(spec, InnoDBClusterSpec):
        # Currently backup can't be done on RRs
        return ("", "")

    if all(not getattr(profile, 'meb', None) for profile in spec.backupProfiles):
        # No profile requests MEB
        return ("", "")

    if spec.tlsUseSelfSigned:
        ssl_cert = "/var/lib/mysql/server-cert.pem"
        ssl_key = "/var/lib/mysql/server-key.pem"
        mount = ""
    else:
        ssl_cert = "//etc/mysql-ssl/tls.crt"
        ssl_key = "/etc/mysql-ssl/tls.key"
        mount = """
        - mountPath: /etc/mysql-ssl"
          name: ssldata
       """

    container = f"""
      - name: meb
        command:
        - mysqlsh
        - --pym
        - meb.meb_main
        - --pod-name
        - "$(POD_NAME)"
        - --pod-namespace
        - "$(POD_NAMESPACE)"
        - --datadir
        - /var/lib/mysql
        - --ssl-cert
        - {ssl_cert}
        - --ssl-key
        - {ssl_key}

        env:
        - name: POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: POD_NAMESPACE
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace
        - name: MYSQL_OPERATOR_K8S_CLUSTER_DOMAIN
          value: {cluster_domain}
        - name: MYSQLSH_USER_CONFIG_HOME
          value: /tmp
        - name: MYSQLSH_CREDENTIAL_STORE_SAVE_PASSWORDS
          value: never
        image: {spec.mysql_image}
        imagePullPolicy: {spec.mysql_image_pull_policy}
        securityContext:
          allowPrivilegeEscalation: false
          capabilities:
            drop:
            - ALL
          privileged: false
          readOnlyRootFilesystem: true
          runAsGroup: 27
          runAsUser: 27
        volumeMounts:
        - mountPath: /usr/lib/mysqlsh/python-packages/meb
          name: mebcode
        - mountPath: /var/lib/mysql
          name: datadir
        - mountPath: /tmp
          # sharing with initconf to transfer restore information
          name: initconf-tmp
        - mountPath: /mysqlsh
          name: mebmyslhshhome
        - name: rundir
          mountPath: /var/run/mysqld
        - name: mebtls
          mountPath: /tls
{mount}
"""

    volumes = f"""
      - name: mebmyslhshhome
        emptyDir: {{}}
      - name: mebcode
        configMap:
            name: {cluster.name}-mebcode
      - name: mebtls
        secret:
            secretName: {cluster.name}-meb-tls
"""

    return (container, volumes)


# TODO - check if we need to add a finalizer to the sts and svc (and if so, what's the condition to remove them)
# TODO - check if we need to make readinessProbe take into account innodb recovery times

# TODO - create ServiceAccount ({cluster.name}-sidecar-sa) for the mysql pods and bind it to the mysql-sidecar role

# ## About lifecycle probes:
#
# ### startupProbe
#
# used to let k8s know that the container is still starting up.
#
# * Server startup can take anywhere from a few seconds to several minutes.
# * If the server is initializing for the first time, it will take a few seconds.
# * If the server is restarting after a clean shut down and there's not much data,
#   it will take even less to startup.
# * But if it's restarting after a crash and there's a lot of data, the InnoDB
#   recovery can take a very long time to finish.
# Since we want success to be reported asap, we set the interval to a small value.
# We also set the successThreshold to > 1, so that we can report success once
# every now and then to reset the failure counter.
# NOTE: Currently, the startup probe will never fail the startup. We assume that
# mysqld will abort if the startup fails. Once a method to check whether the
# server is actually frozen during startup, the probe should be updated to stop
# resetting the failure counter and let it actually fail.
#
# ### readinessProbe
#
# used to let k8s know that the container can be marked as ready, which means
# it can accept external connections. We need mysqld to be always accessible,
# so the probe should always succeed as soon as startup succeeds.
# Any failures that happen after it's up don't matter for the probe, because
# we want GR and the operator to control the fate of the container, not the
# probe.
#
# ### livenessProbe
#
# this checks that the server is still healthy. If it fails above the threshold
# (e.g. because of a deadlock), the container is restarted.
#
def prepare_cluster_stateful_set(cluster: InnoDBCluster, spec: AbstractServerSetSpec, logger: Logger) -> dict:
    init_mysql_argv = ["mysqld", "--user=mysql"]
#    if config.enable_mysqld_general_log:
#        init_mysql_argv.append("--general-log=1")

    mysql_argv = init_mysql_argv

    # we only need this in initconf, we pass it to all operator images to be
    # on the safe side
    cluster_domain = k8s_cluster_domain(logger)

    fqdn_template = fqdn.idc_service_fqdn_template(spec)

    extra_label = ""
    if type(spec) is InnoDBClusterSpec:
        instance_type = "group-member"
    elif type(spec) is ReadReplicaSpec:
        instance_type = "read-replica"
        extra_label = f"mysql.oracle.com/read-replica: {spec.name}"
        # initial startup no replica, we scale up once the group is running
        # spec.instances therefore will be reduced by the caller!
    else:
        raise NotImplementedError(f"Unknown subtype {type(spec)} for creating StatefulSet")

    fixdatadir_container = ""
    if spec.dataDirPermissions.setRightsUsingInitContainer:
        fixdatadir_container = f"""
- name: fixdatadir
  image: {spec.operator_image}
  imagePullPolicy: {spec.sidecar_image_pull_policy}
  command: ["bash", "-c", "chown 27:27 /var/lib/mysql && chmod 0700 /var/lib/mysql"]
  securityContext:
    # make an exception for this one
    runAsNonRoot: false
    runAsUser: 0
    # These can't go to spec.template.spec.securityContext
    # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodTemplateSpec / https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSpec
    # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSecurityContext - for pods (top level)
    # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#Container
    # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#SecurityContext - for containers
    allowPrivilegeEscalation: false
    privileged: false
    readOnlyRootFilesystem: true
    capabilities:
      add:
      - CHOWN
      - FOWNER
      drop:
      - ALL
  volumeMounts:
  - name: datadir
    mountPath: /var/lib/mysql
  env:
  - name: MYSQL_OPERATOR_K8S_CLUSTER_DOMAIN
    value: {cluster_domain}
  - name: MYSQLSH_CREDENTIAL_STORE_SAVE_PASSWORDS
    value: never
"""
    logger.info(f"Fix data container {'EN' if fixdatadir_container else 'DIS'}ABLED")

    # if meb  restore ...
    (restore_container, restore_volumes) = get_restore_container(cluster, spec, cluster_domain)
    (meb_container, meb_volumes) = get_meb_container(cluster, spec, cluster_domain)

    # TODO re-add "--log-file=",
    tmpl = f"""
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: {spec.name}
  annotations:
      mysql.oracle.com/fqdn-template: '{fqdn_template}'
  labels:
    tier: mysql
    mysql.oracle.com/cluster: {spec.cluster_name}
    mysql.oracle.com/instance-type: {instance_type}
    {extra_label}
    app.kubernetes.io/name: mysql-innodbcluster
    app.kubernetes.io/instance: mysql-innodbcluster-{spec.name}
    app.kubernetes.io/component: database
    app.kubernetes.io/managed-by: mysql-operator
    app.kubernetes.io/created-by: mysql-operator
spec:
  serviceName: {spec.headless_service_name}
  replicas: {spec.instances}
  podManagementPolicy: Parallel
  selector:
    matchLabels:
      component: mysqld
      tier: mysql
      mysql.oracle.com/cluster: {spec.cluster_name}
      mysql.oracle.com/instance-type: {instance_type}
      {extra_label}
      app.kubernetes.io/name: mysql-innodbcluster-mysql-server
      app.kubernetes.io/instance: mysql-innodbcluster-{spec.name}-mysql-server
      app.kubernetes.io/component: database
      app.kubernetes.io/managed-by: mysql-operator
      app.kubernetes.io/created-by: mysql-operator
  template:
    metadata:
      annotations:
        mysql.oracle.com/fqdn-template: '{fqdn_template}'
      labels:
        component: mysqld
        tier: mysql
        mysql.oracle.com/cluster: {spec.cluster_name}
        mysql.oracle.com/instance-type: {instance_type}
        {extra_label}
        app.kubernetes.io/name: mysql-innodbcluster-mysql-server
        app.kubernetes.io/instance: mysql-innodbcluster-{spec.name}-mysql-server
        app.kubernetes.io/component: database
        app.kubernetes.io/managed-by: mysql-operator
        app.kubernetes.io/created-by: mysql-operator
    spec:
      readinessGates:
      - conditionType: "mysql.oracle.com/configured"
      - conditionType: "mysql.oracle.com/ready"
      serviceAccountName: {spec.serviceAccountName}
      securityContext:
        runAsUser: 27
        runAsGroup: 27
        fsGroup: 27
{utils.indent("fsGroupChangePolicy: " + spec.dataDirPermissions.fsGroupChangePolicy, 8) if spec.dataDirPermissions.fsGroupChangePolicy else ""}
        runAsNonRoot: true
      terminationGracePeriodSeconds: 120
      initContainers:
{utils.indent(fixdatadir_container, 6)}

      - name: initconf
        image: {spec.operator_image}
        imagePullPolicy: {spec.sidecar_image_pull_policy}
        # For datadir see the datadir volum mount
        command: ["mysqlsh", "--log-level=@INFO", "--pym", "mysqloperator", "init",
                  "--pod-name", "$(POD_NAME)",
                  "--pod-namespace", "$(POD_NAMESPACE)",
                  "--datadir", "/var/lib/mysql"
        ]
        securityContext:
          # These can't go to spec.template.spec.securityContext
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodTemplateSpec / https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSpec
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSecurityContext - for pods (top level)
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#Container
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#SecurityContext - for containers
          allowPrivilegeEscalation: false
          privileged: false
          readOnlyRootFilesystem: true
          # The value is is inherited from the PodSecurityContext but dumb sec checkers might not know that
          runAsNonRoot: true
          capabilities:
            drop:
            - ALL
        env:
        - name: POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: POD_NAMESPACE
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace
        - name: MYSQL_OPERATOR_K8S_CLUSTER_DOMAIN
          value: {cluster_domain}
        - name: MYSQLSH_USER_CONFIG_HOME
          value: /tmp
        - name: MYSQLSH_CREDENTIAL_STORE_SAVE_PASSWORDS
          value: never
        volumeMounts:
        - name: initconfdir
          mountPath: /mnt/initconf
          readOnly: true
        - name: datadir
          mountPath: /var/lib/mysql
        - name: mycnfdata
          mountPath: /mnt/mycnfdata
        - name: initconf-tmp
          mountPath: /tmp
        - name: rootcreds
          readOnly: true
          # rootHost is not obligatory and thus might not exist in the secret
          # Nevertheless K8s won't complain and instead of mounting an empty file
          # will create a directory (/rootcreds/rootHost will be an empty directory)
          # For more information see below the comment regarding rootcreds.
          subPath: rootHost
          mountPath: /rootcreds/rootHost

{restore_container}

      - name: initmysql
        image: {spec.mysql_image}
        imagePullPolicy: {spec.mysql_image_pull_policy}
        args: {init_mysql_argv}
        securityContext:
          # These can't go to spec.template.spec.securityContext
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodTemplateSpec / https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSpec
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSecurityContext - for pods (top level)
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#Container
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#SecurityContext - for containers
          allowPrivilegeEscalation: false
          privileged: false
          readOnlyRootFilesystem: true
          # The value is is inherited from the PodSecurityContext but dumb sec checkers might not know that
          runAsNonRoot: true
          capabilities:
            drop:
            - ALL
        env:
        - name: MYSQL_INITIALIZE_ONLY
          value: "1"
        - name: MYSQL_RANDOM_ROOT_PASSWORD
          value: "1"
        - name: MYSQLSH_USER_CONFIG_HOME
          value: /tmp
        volumeMounts:
        - name: datadir
          mountPath: /var/lib/mysql
        - name: rundir
          mountPath: /var/run/mysqld
        - name: mycnfdata
          mountPath: /etc/my.cnf.d
          subPath: my.cnf.d
        - name: mycnfdata
          mountPath: /docker-entrypoint-initdb.d
          subPath: docker-entrypoint-initdb.d
        - name: mycnfdata
          mountPath: /etc/my.cnf
          subPath: my.cnf
        - name: initmysql-tmp
          mountPath: /tmp
        - name: varlibmysqlfiles # The entrypoint of the container `touch`-es 2 files there
          mountPath: /var/lib/mysql-files
      containers:
      - name: sidecar
        image: {spec.operator_image}
        imagePullPolicy: {spec.sidecar_image_pull_policy}
        command: ["mysqlsh", "--pym", "mysqloperator", "sidecar",
                  "--pod-name", "$(POD_NAME)",
                  "--pod-namespace", "$(POD_NAMESPACE)",
                  "--datadir", "/var/lib/mysql"
        ]
        securityContext:
          # These can't go to spec.template.spec.securityContext
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodTemplateSpec / https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSpec
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSecurityContext - for pods (top level)
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#Container
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#SecurityContext - for containers
          allowPrivilegeEscalation: false
          privileged: false
          readOnlyRootFilesystem: true
          # The value is is inherited from the PodSecurityContext but dumb sec checkers might not know that
          runAsNonRoot: true
          capabilities:
            drop:
            - ALL
        env:
        - name: POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: POD_NAMESPACE
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace
        - name: MYSQL_UNIX_PORT
          value: /var/run/mysqld/mysql.sock
        - name: MYSQLSH_USER_CONFIG_HOME
          value: /mysqlsh
        - name: MYSQL_OPERATOR_K8S_CLUSTER_DOMAIN
          value: {cluster_domain}
        - name: MYSQLSH_CREDENTIAL_STORE_SAVE_PASSWORDS
          value: never
        volumeMounts:
        - name: rundir
          mountPath: /var/run/mysqld
        - name: mycnfdata
          mountPath: /etc/my.cnf.d
          subPath: my.cnf.d
        - name: mycnfdata
          mountPath: /etc/my.cnf
          subPath: my.cnf
        - name: shellhome
          mountPath: /mysqlsh
        - name: sidecar-tmp
          mountPath: /tmp
{utils.indent(spec.extra_sidecar_volume_mounts, 8)}
      - name: mysql
        image: {spec.mysql_image}
        imagePullPolicy: {spec.mysql_image_pull_policy}
        args: {mysql_argv}
        securityContext:
          # These can't go to spec.template.spec.securityContext
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodTemplateSpec / https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSpec
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#PodSecurityContext - for pods (top level)
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#Container
          # See: https://pkg.go.dev/k8s.io/api@v0.26.1/core/v1#SecurityContext - for containers
          allowPrivilegeEscalation: false
          privileged: false
          readOnlyRootFilesystem: true
          # The value is is inherited from the PodSecurityContext but dumb sec checkers might not know that
          runAsNonRoot: true
          capabilities:
            drop:
            - ALL
        lifecycle:
          preStop:
            exec:
              # 60 is the default value for dba.gtidWaitTimeout
              # see https://dev.mysql.com/doc/mysql-shell/8.0/en/mysql-innodb-cluster-working-with-cluster.html
              command: ["sh", "-c", "sleep 60 && mysqladmin -ulocalroot shutdown"]
        startupProbe:
          exec:
            command: ["/livenessprobe.sh", "8"]
          initialDelaySeconds: 5
          periodSeconds: 3
          failureThreshold: 10000
          successThreshold: 1
          timeout: 2
        readinessProbe:
          exec:
            command: ["/readinessprobe.sh"]
          periodSeconds: 5
          initialDelaySeconds: 10
          failureThreshold: 10000
        livenessProbe:
          exec:
            command: ["/livenessprobe.sh"]
          initialDelaySeconds: 15
          periodSeconds: 15
          failureThreshold: 10
          successThreshold: 1
          timeout: 5
        env:
        - name: MYSQL_UNIX_PORT
          value: /var/run/mysqld/mysql.sock
        - name: MYSQLSH_CREDENTIAL_STORE_SAVE_PASSWORDS
          value: never
{utils.indent(spec.extra_env, 8)}
        ports:
        - containerPort: {spec.mysql_port}
          name: mysql
        - containerPort: {spec.mysql_xport}
          name: mysqlx
        - containerPort: {spec.mysql_grport}
          name: gr-xcom
        volumeMounts:
        - name: datadir
          mountPath: /var/lib/mysql
        - name: rundir
          mountPath: /var/run/mysqld
        - name: mycnfdata
          mountPath: /etc/my.cnf.d
          subPath: my.cnf.d
        - name: mycnfdata
          mountPath: /etc/my.cnf
          subPath: my.cnf
        - name: initconfdir
          mountPath: /livenessprobe.sh
          subPath: livenessprobe.sh
        - name: initconfdir
          mountPath: /readinessprobe.sh
          subPath: readinessprobe.sh
        - name: varlibmysqlfiles # The entrypoint of the container `touch`-es 2 files there
          mountPath: /var/lib/mysql-files
        - name: mysql-tmp
          mountPath: /tmp
{utils.indent(spec.extra_volume_mounts, 8)}
{meb_container}
      volumes:
      - name: mycnfdata
        emptyDir: {{}}
      - name: rundir
        emptyDir: {{}}
      - name: varlibmysqlfiles
        emptyDir: {{}}
      - name: initconfdir
        configMap:
          name: {spec.name}-initconf
          defaultMode: 0755
      - name: shellhome
        emptyDir: {{}}
      - name: initconf-tmp
        emptyDir: {{}}
      - name: initmysql-tmp
        emptyDir: {{}}
      - name: mysql-tmp
        emptyDir: {{}}
      - name: sidecar-tmp
        emptyDir: {{}}
{meb_volumes}
{restore_volumes}
      # If we declare it and not use it anywhere as backing for a volumeMount K8s won't check
      # if the volume exists. K8s seems to be lazy in that regard. We don't need the information
      # from this secret directly, as the sidecar of pod 0 will fetch the information using the K8s API
      # However, we won't not to be lazy in checking if the secret exists and make it easier for the
      # administrator to find out if the secret is missing. If we mount it in a init or normal container,
      # the pod # will get stuck into "Ready:0/2 Init:0/3" with
      # Warning  FailedMount  XXs (....)  kubelet  "MountVolume.SetUp failed for volume "rootcreds" : secret ".........." not found" error to be seen in describe.
      - name: rootcreds
        secret:
          secretName: {spec.secretName}
          defaultMode: 0400
{utils.indent(spec.extra_volumes, 6)}
  volumeClaimTemplates:
  - metadata:
      name: datadir
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 2Gi
"""

    statefulset = yaml.safe_load(tmpl.replace("\n\n", "\n"))

    metadata = {}
    if spec.podAnnotations:
        print("\t\tAdding podAnnotations")
        metadata['annotations'] = spec.podAnnotations
    if spec.podLabels:
        print("\t\tAdding podLabels")
        metadata['labels'] = spec.podLabels

    if len(metadata):
        utils.merge_patch_object(statefulset["spec"]["template"], {"metadata" : metadata })

    if spec.keyring:
        print("\t\tAdding keyring STS bit")
        spec.keyring.add_to_sts_spec(statefulset)

    for subsystem in spec.add_to_sts_cbs:
        print(f"\t\tadd_to_sts_cb: Checking subsystem {subsystem}")
        for add_to_sts_cb in spec.add_to_sts_cbs[subsystem]:
            print(f"\t\tAdding {subsystem} STS bits")
            add_to_sts_cb(statefulset, None, logger)

    if os.getenv("MYSQL_OPERATOR_GLOBAL_PODSPEC_CM"):
        ps_cm_ns = os.getenv("MYSQL_OPERATOR_GLOBAL_PODSPEC_NS", "mysql-operator")
        ps_cm_name = os.getenv("MYSQL_OPERATOR_GLOBAL_PODSPEC_CM")
        ps_cm_item = os.getenv("MYSQL_OPERATOR_GLOBAL_PODSPEC_ITEM", "podspec.yaml")
        ps_cm = api_core.get_namespaced_config_map(ps_cm_name, ps_cm_name)
        ps_override = yaml.safe_load(ps_cm[ps_cm_item])
        print("\t\tAdding global podSpec")
        utils.merge_patch_object(statefulset["spec"]["template"]["spec"],
                                 ps_override, f"{ps_cm_ns}.{ps_cm_name}.{ps_cm_item}")

    if spec.podSpec:
        print("\t\tAdding podSpec")
        utils.merge_patch_object(statefulset["spec"]["template"]["spec"],
                                 spec.podSpec, "spec.podSpec")

    if spec.datadirVolumeClaimTemplate:
        print("\t\tAdding datadirVolumeClaimTemplate")
        utils.merge_patch_object(statefulset["spec"]["volumeClaimTemplates"][0]["spec"],
                                 spec.datadirVolumeClaimTemplate, "spec.volumeClaimTemplates[0].spec")
    return statefulset

def update_stateful_set_size(cluster: InnoDBCluster, rr_spec: ReadReplicaSpec, logger: Logger) -> None:
    sts = cluster.get_read_replica_stateful_set(rr_spec.name)
    if sts:
        patch = {"spec": {"replicas": rr_spec.instances}}
        api_apps.patch_namespaced_stateful_set(
            sts.metadata.name, sts.metadata.namespace, body=patch)


def prepare_service_account_sidecar(spec: AbstractServerSetSpec) -> dict:
    account = f"""
apiVersion: v1
kind: ServiceAccount
metadata:
  name: {spec.serviceAccountName}
  namespace: {spec.namespace}
{spec.image_pull_secrets}
"""

    account = yaml.safe_load(account)

    return account


def prepare_service_account_patch_for_image_pull_secrets(spec: AbstractServerSetSpec) -> Optional[Dict]:
    if not spec.imagePullSecrets:
        return None
    return {
        "imagePullSecrets" : spec.imagePullSecrets
    }


def prepare_service_account_switchover(spec: AbstractServerSetSpec) -> dict:
    account = f"""
apiVersion: v1
kind: ServiceAccount
metadata:
  name: {spec.switchoverServiceAccountName}
  namespace: {spec.namespace}
{spec.image_pull_secrets}
"""

    account = yaml.safe_load(account)

    return account


def prepare_role_binding_sidecar(spec: AbstractServerSetSpec) -> dict:
    rolebinding = f"""
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: {spec.roleBindingName}
  namespace: {spec.namespace}
subjects:
  - kind: ServiceAccount
    name: {spec.serviceAccountName}
roleRef:
  kind: ClusterRole
  name: mysql-sidecar
  apiGroup: rbac.authorization.k8s.io
"""
    rolebinding = yaml.safe_load(rolebinding)

    return rolebinding


def prepare_role_binding_switchover(spec: AbstractServerSetSpec) -> dict:
    rolebinding = f"""
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: {spec.switchoverRoleBindingName}
  namespace: {spec.namespace}
subjects:
  - kind: ServiceAccount
    name: {spec.switchoverServiceAccountName}
roleRef:
  kind: ClusterRole
  name: mysql-switchover
  apiGroup: rbac.authorization.k8s.io
"""
    rolebinding = yaml.safe_load(rolebinding)

    return rolebinding


def prepare_additional_configmaps(spec: AbstractServerSetSpec, logger: Logger) -> List[Dict]:
    configmaps = []
    prefix = ''
    for subsystem in spec.get_configmaps_cbs:
        for cb in spec.get_configmaps_cbs[subsystem]:
            if cms := cb(prefix, logger):
              for (cm_name, cm) in cms:
                  if cm:
                      configmaps.append(cm)

    # TODO: This should use the CB mechanism like other CMs above
    if isinstance(spec, InnoDBClusterSpec):
        clusterspec = cast(InnoDBClusterSpec, spec)
        if  (clusterspec.initDB and clusterspec.initDB.meb) \
                or any(getattr(profile, 'meb', None) for profile in clusterspec.backupProfiles):

            if spec.edition != config.Edition.enterprise:
                raise kopf.TemporaryError("A BackupProfile requires MySQL Enterprise Backup, but the cluster doesn't use Enterprise Edition")

            configmaps.append(backup_objects.prepare_meb_code_configmap(clusterspec))

    return configmaps


def prepare_component_config_configmaps(spec: AbstractServerSetSpec, logger: Logger) -> List[Dict]:
    configmaps = [
        spec.keyring.get_component_config_configmap_manifest()
    ]

    return configmaps


def prepare_component_config_secrets(spec: AbstractServerSetSpec, logger: Logger) -> List[Dict]:
    secrets = []
    cm = spec.keyring.get_component_config_secret_manifest()
    if cm:
        secrets.append(cm)

    return secrets

def prepare_initconf(cluster: InnoDBCluster, spec: AbstractServerSetSpec, logger: Logger) -> dict:

    with open(os.path.dirname(os.path.abspath(__file__))+'/router-entrypoint-run.sh.tpl', 'r') as entryfile:
        router_entrypoint = "".join(entryfile.readlines())

    liveness_probe = """#!/bin/bash
# Copyright (c) 2020, 2021, Oracle and/or its affiliates.

# Insert 1 success every this amount of failures
# (assumes successThreshold is > 1)
max_failures_during_progress=$1

# Ping the server to see if it's up
mysqladmin -umysqlhealthchecker ping
# If it's up, we succeed
if [ $? -eq 0 ]; then
  exit 0
fi

if [ -z $max_failures_during_progress ]; then
  exit 1
fi

# If the init/startup/InnoDB recovery is still ongoing, we're
# not succeeded nor failed yet, so keep failing and getting time
# extensions until it succeeds.
# We currently rely on the server to exit/abort if the init/startup fails,
# but ideally there would be a way to check whether the server is
# still making progress and not just stuck waiting on a frozen networked
# volume, for example.

if [ -f /tmp/fail-counter ]; then
  fail_count=$(($(cat /tmp/fail-counter) + 1))
else
  fail_count=1
fi

if [ $fail_count -gt $max_failures_during_progress ]; then
  # Report success to reset the failure counter upstream and get
  # a time extension
  rm -f /tmp/fail-counter
  exit 0
else
  # Update the failure counter and fail out
  echo $fail_count > /tmp/fail-counter
  exit 1
fi
"""

    readiness_probe = """#!/bin/bash
# Copyright (c) 2020, 2022, Oracle and/or its affiliates.

# Once the container is ready, it's always ready.
if [ -f /tmp/mysql-ready ]; then
  exit 0
fi

# Ping server to see if it is ready
if mysqladmin -umysqlhealthchecker ping; then
  touch /tmp/mysql-ready
  exit 0
else
  exit 1
fi
"""

    has_crl = cluster.tls_has_crl()

    if not spec.tlsUseSelfSigned:
        ca_file_name = cluster.get_ca_and_tls().get("CA", "ca.pem")
    else:
        ca_file_name = ""

    tmpl = f"""
apiVersion: v1
kind: ConfigMap
metadata:
  name: {spec.name}-initconf
data:
  initdb-localroot.sql: |
    set sql_log_bin=0;
    # Create socket authenticated localroot@localhost account
    CREATE USER 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;
    # Drop the default account created by the docker image
    DROP USER IF EXISTS healthchecker@localhost;
    # Create account for liveness probe
    CREATE USER mysqlhealthchecker@localhost IDENTIFIED WITH auth_socket AS 'mysql';
    set sql_log_bin=1;


  readinessprobe.sh: |
{utils.indent(readiness_probe, 4)}


  livenessprobe.sh: |
{utils.indent(liveness_probe, 4)}

  router-entrypoint-run.sh.tpl: |
{utils.indent(router_entrypoint, 4)}


  my.cnf.in: |
    # Server identity related options (not shared across instances).
    # Do not edit.
    [mysqld]
    server_id=@@SERVER_ID@@
    report_host=@@HOSTNAME@@
    datadir=/var/lib/mysql
    loose_mysqlx_socket=/var/run/mysqld/mysqlx.sock
    socket=/var/run/mysqld/mysql.sock
    local-infile=1

    [mysql]
    socket=/var/run/mysqld/mysql.sock

    [mysqladmin]
    socket=/var/run/mysqld/mysql.sock

    !includedir /etc/my.cnf.d


  00-basic.cnf: |
    # Basic configuration.
    # Do not edit.
    [mysqld]
    plugin_load_add=auth_socket.so
    loose_auth_socket=FORCE_PLUS_PERMANENT
    skip_log_error
    log_error_verbosity=3

  01-group_replication.cnf: |
    # GR and replication related options
    # Do not edit.
    [mysqld]
    log_bin={spec.name}
    enforce_gtid_consistency=ON
    gtid_mode=ON
    skip_replica_start=1

  02-ssl.cnf: |
    # SSL configurations
    # Do not edit.
    [mysqld]
    {"# " if spec.tlsUseSelfSigned else ""}ssl-ca=/etc/mysql-ssl/{ca_file_name}
    {"# " if not has_crl else ""}ssl-crl=/etc/mysql-ssl/crl.pem
    {"# " if spec.tlsUseSelfSigned else ""}ssl-cert=/etc/mysql-ssl/tls.crt
    {"# " if spec.tlsUseSelfSigned else ""}ssl-key=/etc/mysql-ssl/tls.key

    loose_group_replication_recovery_use_ssl=1
    {"# " if spec.tlsUseSelfSigned else ""}loose_group_replication_recovery_ssl_verify_server_cert=1

    {"# " if spec.tlsUseSelfSigned else ""}loose_group_replication_recovery_ssl_ca=/etc/mysql-ssl/{ca_file_name}
    #{"# " if not has_crl else ""}loose_group_replication_recovery_ssl_crl=/etc/mysql-ssl/crl.pem
    {"# " if spec.tlsUseSelfSigned else ""}loose_group_replication_recovery_ssl_cert=/etc/mysql-ssl/tls.crt
    {"# " if spec.tlsUseSelfSigned else ""}loose_group_replication_recovery_ssl_key=/etc/mysql-ssl/tls.key

  99-extra.cnf: |
    # Additional user configurations taken from spec.mycnf in InnoDBCluster.
    # Do not edit directly.
{utils.indent(spec.mycnf, 4) if spec.mycnf else ""}
"""

    cm = yaml.safe_load(tmpl)

    prefix = 5
    for subsystem in spec.add_to_initconf_cbs:
        for add_to_initconf_cb in spec.add_to_initconf_cbs[subsystem]:
          add_to_initconf_cb(cm, f"{prefix:02d}-", logger)
          prefix = prefix + 1

    return cm


def prepare_service_failover_job(batchjob_name: str, force: str, options: dict, spec: InnoDBClusterSpec, logger: Logger) -> dict:
    cluster_domain = k8s_cluster_domain(logger)

    cluster_name = spec.cluster_name
    namespace = spec.namespace
    extras =", --force" if force else ""

    if options:
        extras += f", --timeout={options['timeout']}" if "timeout" in options else ""
        # TODO shell escape! Probably it's better to have the pod read this itself
        extras += f", --invalidate-replica-clusters={','.join(options['invalidateReplicaClusters'])}" if "invalidateReplicaClusters" in options else ""

    job = f"""
apiVersion: batch/v1
kind: Job
metadata:
  name: {batchjob_name}
  namespace: {namespace}
spec:
  template:
    spec:
      serviceAccountName: mysql-switchover-sa
      securityContext:
        runAsUser: 27
        runAsGroup: 27
        fsGroup: 27
      restartPolicy: Never
      containers:
      - name: failover
        image: {spec.operator_image}
        imagePullPolicy: {spec.mysql_image_pull_policy}
        command: ["mysqlsh",  "--pym", "mysqloperator", "csfo", --cluster-name, {cluster_name}, "--namespace", {namespace} {extras}]
        securityContext:
          allowPrivilegeEscalation: false
          privileged: false
          readOnlyRootFilesystem: true
          capabilities:
            drop:
            - ALL
        env:
        - name: MYSQLSH_USER_CONFIG_HOME
          value: /tmp
        - name: MYSQL_OPERATOR_K8S_CLUSTER_DOMAIN
          value: {cluster_domain}
        volumeMounts:
        - name: tmp
          mountPath: /tmp
      volumes:
      - name: tmp
        emptyDir: {{}}
"""

    return  yaml.safe_load(job)



def prepare_metrics_service_monitors(spec: AbstractServerSetSpec, logger: Logger) -> List[Dict]:
    monitors = []
    prefix = ''
    for subsystem in spec.get_svc_monitor_cbs:
        for cb in spec.get_svc_monitor_cbs[subsystem]:
            (monitor_name, monitor) = cb(logger)
            if monitor:
                monitors.append(monitor)

    return monitors


def update_stateful_set_spec(sts : api_client.V1StatefulSet, patch: dict) -> None:
    api_apps.patch_namespaced_stateful_set(
        sts.metadata.name, sts.metadata.namespace, body=patch)


def update_service(svc: api_client.V1Deployment, spec: InnoDBClusterSpec, logger: Logger) -> None:
    body = prepare_cluster_service(spec, logger)
    print(body)
    api_core.patch_namespaced_service(svc.metadata.name, svc.metadata.namespace, body=body)


def update_mysql_image(sts: api_client.V1StatefulSet, cluster: InnoDBCluster,
                       spec: AbstractServerSetSpec,
                       patcher,
                       logger: Logger) -> None:
    """Update MySQL Server image

    This will also update the sidecar container to the current operator version,
    so that a single rolling upgrade covers both and we don't require a restart
    for upgrading sidecar.
    """
    logger.info("update_mysql_image")
    # Operators <= 8.0.32-2.0.8 don't set this environment variable, we have to make sure it is there
    cluster_domain_env = [{
        "name": "MYSQL_OPERATOR_K8S_CLUSTER_DOMAIN",
        "value": k8s_cluster_domain(logger)
    }]

    patch = {"spec": {"template":
                      {"spec": {
                          "containers": [
                               {"name": "sidecar",
                                "image": spec.operator_image,
                                "env": cluster_domain_env
                               },
                               {"name": "mysql",
                                "image": spec.mysql_image,
                                "env": cluster_domain_env
                               },
                          ],
                          "initContainers": [
                              {"name": "fixdatadir",
                               "image": spec.operator_image,
                               "env": cluster_domain_env
                                },
                              {"name": "initconf",
                               "image": spec.operator_image,
                               "env": cluster_domain_env
                              },
                              {"name": "initmysql",
                               "image": spec.mysql_image,
                               "env": cluster_domain_env
                              },
                          ]}
                       }}}

    # TODO [compat8.3.0] remove this when compatibility pre 8.3.0 isn't needed anymore
    keyring_update = spec.keyring.upgrade_to_component(sts, spec, logger)

    if keyring_update:
        logger.info("Need to upgrade keyring from plugin to component")
        (cm, key_sts_patch) = keyring_update
        utils.merge_patch_object(patch["spec"]["template"], key_sts_patch)

        kopf.adopt(cm)
        patcher.create_configmap(spec.namespace, cm["metadata"]["name"], cm, on_apiexception_generic_handler)
        #api_core.create_namespaced_config_map(spec.namespace, cm)

        initconf_patch = [{"op": "remove", "path": "/data/03-keyring-oci.cnf"}]
        #try:
        #    api_core.patch_namespaced_config_map(f"{spec.cluster_name}-initconf",
        #                                            spec.namespace, initconf_patch)
        #except ApiException as exc:
        #    # This might happen during a retry or some other case where it was
        #    # removed already
        #    logger.info(f"Failed to remove keyring config from initconf, ignoring: {exc}")
        patcher.patch_configmap(spec.namespace, f"{spec.cluster_name}-initconf", initconf_patch, on_apiexception_404_handler)

    cm = prepare_initconf(cluster, spec, logger)
    patcher.patch_configmap(spec.namespace, cm['metadata']['name'], cm, on_apiexception_generic_handler)
    #api_core.patch_namespaced_config_map(
    #    cm['metadata']['name'], sts.metadata.namespace, body=cm)

    patcher.patch_sts(patch)
#    update_stateful_set_spec(sts, patch)


def update_operator_image(sts: api_client.V1StatefulSet, spec: InnoDBClusterSpec) -> None:
    patch = {"spec": {"template":
                      {"spec": {
                          "containers": [
                               {"name": "sidecar", "image": spec.operator_image}
                          ],
                          "initContainers": [
                              {"name": "fixdatadir", "image": spec.operator_image},
                              {"name": "initconf", "image": spec.operator_image}
                          ]}
                       }}}
    update_stateful_set_spec(sts, patch)


def update_pull_policy(sts: api_client.V1StatefulSet, spec: InnoDBClusterSpec, logger: Logger) -> dict:
    patch = {"spec": {"template":
                      {"spec": {
                          "initContainers": [
                              {"name": "initconf", "imagePullPolicy": spec.sidecar_image_pull_policy},
                              {"name": "initmysql", "imagePullPolicy": spec.mysql_image_pull_policy}
                          ],
                          "containers": [
                               {"name": "sidecar", "imagePullPolicy": spec.sidecar_image_pull_policy},
                               {"name": "mysql", "imagePullPolicy": spec.mysql_image_pull_policy}
                          ]}
                       }}}
    return patch

def update_template_property(sts: api_client.V1StatefulSet, property_name: str, property_value: str, logger: Logger) -> None:
    patch = {"spec": {"template": {"spec": { property_name: property_value }}}}
    update_stateful_set_spec(sts, patch)


def update_objects_for_subsystem(subsystem: InnoDBClusterSpecProperties,
                                 cluster: InnoDBCluster,
                                 patcher: 'InnoDBClusterObjectModifier',
                                 logger: Logger) -> None:
    logger.info(f"update_objects_for_subsystem: {subsystem}")

    sts = cluster.get_stateful_set()
    svc = cluster.get_service()

    spec = cluster.parsed_spec

    if subsystem in spec.get_configmaps_cbs:
        print(f"\t\tWalking over get_configmaps_cbs len={len(spec.get_configmaps_cbs[subsystem])}")
        #TODO: This won't delete old CMs but only replace old ones, if are still in use, with new content
        #      or create new ones. The solution is to use tuple returning like get_svc_monitor_cbs, where
        #      the cm name will be returned as first tuple element and second will be just None. This will
        #      signal that this CM should be removed, as not in use anymore.
        for get_configmap_cb in spec.get_configmaps_cbs[subsystem]:
            prefix = ''
            new_configmaps = get_configmap_cb(prefix, logger)
            if not new_configmaps:
                continue
            for (cm_name, new_cm) in new_configmaps:
                current_cm = cluster.get_configmap(cm_name)
                if current_cm:
                    if not new_cm:
                        print(f"\t\t\tDeleting CM {cluster.namespace}/{cm_name}")
                        #patcher.delete_configmap(cluster.namespace, cm_name, on_apiexception_404_handler)
                        cluster.delete_configmap(cm_name)
                        continue

                    data_differs = current_cm.data != new_cm["data"]
                    if data_differs:
                        print(f"\t\t\tReplacing CM {cluster.namespace}/{cm_name}")
                        current_cm.data = new_cm["data"]
                        #patcher.replace_configmap(cluster.namespace, cm_name, current_cm, on_apiexception_404_handler)
                        api_core.replace_namespaced_config_map(cm_name, cluster.namespace, body=current_cm)
                else:
                    print(f"\t\t\tNo such cm exists. Creating {cluster.namespace}/{new_cm}")
                    kopf.adopt(new_cm)
                    #patcher.create_configmap(cluster.namespace, new_cm['metadata']['name'], new_cm, on_apiexception_generic_handler)
                    api_core.create_namespaced_config_map(cluster.namespace, new_cm)

    if subsystem in spec.add_to_sts_cbs:
        print(f"\t\tCurrent container count: {len(sts.spec.template.spec.containers)}")
        print(f"\t\tWalking over add_to_sts_cbs len={len(spec.add_to_sts_cbs[subsystem])}")
        changed = False
        sts.spec = spec_to_dict(sts.spec)
        for add_to_sts_cb in spec.add_to_sts_cbs[subsystem]:
            changed = True
            print("\t\t\tPatching STS")
            add_to_sts_cb(sts, patcher, logger)
        if changed:
            new_container_names = [c["name"] for c in patcher.get_sts_path('/spec/template/spec/containers') if c["name"] not in ["mysql", "sidecar"]]
            print(f"\t\t\tNew containers: {new_container_names}")
            new_volumes_names = [c["name"] for c in patcher.get_sts_path('/spec/template/spec/volumes')]
            print(f"\t\t\tNew volumes: {new_volumes_names}")
            new_volume_mounts = [(c["name"], c["volumeMounts"]) for c in patcher.get_sts_path('/spec/template/spec/containers') if c["name"] not in ["mysql", "sidecar"]]
            print(f"\t\t\tNew volume mounts: {new_volume_mounts}")

            # There might be configmap changes, which when mounted will change the server, so we rollover
            # For fine grained approache the get_configmap should return whether there are such changes that require
            # a restart. With a restart, for example, the Cluster1LFSGeneralLogEnableDisableEnable test will hang
            restart_patch = {"spec":{"template":{"metadata":{"annotations":{"kubectl.kubernetes.io/restartedAt":utils.isotime()}}}}}
            patcher.patch_sts(restart_patch)
            #patcher.submit_patches(restart_sts=True)

        print(f"\t\t\tSTS {'patched' if changed else 'unchanged. No rollover upgrade!'}")

    if subsystem in spec.get_add_to_svc_cbs:
        print(f"\t\tWalking over get_add_to_svc_cbs len={len(spec.get_add_to_svc_cbs[subsystem])}")
        changed = False
        for add_to_svc_cb in spec.get_add_to_svc_cbs[subsystem]:
            changed = True
            print("\t\t\tPatching SVC")
            add_to_svc_cb(svc, logger)
        if changed:
            api_core.replace_namespaced_service(svc.metadata.name, svc.metadata.namespace, svc)

        print(f"\t\t\tSVC {'patched' if changed else 'unchanged'}")

    if subsystem in spec.get_svc_monitor_cbs:
      for subsystem in spec.get_svc_monitor_cbs:
          for cb in spec.get_svc_monitor_cbs[subsystem]:
              (monitor_name, monitor) = cb(logger)
              # monitor could be empty, this means - delete old monitor with monitor_name
              print(f"\t\t\tChecking for old ServiceMonitor {monitor_name}")
              if cluster.get_service_monitor(monitor_name):
                  print(f"\t\t\tRemoving old ServiceMonitor {monitor_name}")
                  try:
                      api_customobj.delete_namespaced_custom_object("monitoring.coreos.com", "v1", cluster.namespace,
                                                                    "servicemonitors", monitor_name)
                  except Exception as exc:
                      print(f"\t\t\tPrevious ServiceMonitor {monitor_name} was not removed. Reason: {exc}")
              if monitor:
                  kopf.adopt(monitor)
                  print(f"\t\t\tCreating ServiceMonitor {monitor} ...")
                  try:
                      api_customobj.create_namespaced_custom_object("monitoring.coreos.com", "v1", cluster.namespace,
                                                                    "servicemonitors", monitor)
                  except Exception as exc:
                      # This might be caused by Prometheus Operator missing
                      # we won't fail for that
                      print(f"\t\t\tServiceMonitor {monitor_name} NOT created!")
                      print(exc)
                      cluster.warn(action="CreateCluster", reason="CreateResourceFailed", message=f"{exc}")
              else:
                  print(f"\t\t\tNew ServiceMonitor {monitor_name} will not be created. Monitoring disabled.")


def update_objects_for_logs(cluster: InnoDBCluster, patcher: 'InnoDBClusterObjectModifier', logger: Logger) -> None:
    subsystem = InnoDBClusterSpecProperties.LOGS.value
    update_objects_for_subsystem(subsystem, cluster, patcher, logger)

def update_objects_for_metrics(cluster: InnoDBCluster, patcher: 'InnoDBClusterObjectModifier', logger: Logger) -> None:
    subsystem = InnoDBClusterSpecProperties.METRICS.value
    update_objects_for_subsystem(subsystem, cluster, patcher, logger)


def remove_read_replica(cluster: InnoDBCluster, rr: ReadReplicaSpec):
    name = rr['name']
    try:
        api_core.delete_namespaced_config_map(f"{cluster.name}-{name}-initconf", cluster.namespace)
    except Exception as exc:
        print(f"ConfigMap for ReadReplica {name} was not removed. This is usually ok. Reason: {exc}")

    try:
        api_core.delete_namespaced_service(rr.headless_service_name, cluster.namespace)
    except Exception as exc:
        print(f"Service for ReadReplica {name} was not removed. This is usually ok. Reason: {exc}")

    try:
        api_apps.delete_namespaced_stateful_set(f"{cluster.name}-{name}", cluster.namespace)
    except Exception as exc:
        print(f"StatefulSet for ReadReplica  {name} was not removed. This is usually ok. Reason: {exc}")


def on_first_cluster_pod_created(cluster: InnoDBCluster, logger: Logger) -> None:
    # Add finalizer to the cluster object to prevent it from being deleted
    # until the last pod is properly deleted.
    cluster.add_cluster_finalizer()


def on_last_cluster_pod_removed(cluster: InnoDBCluster, logger: Logger) -> None:
    # Remove cluster finalizer because the last pod was deleted, this lets
    # the cluster object to be deleted too
    logger.info(
        f"Last pod for cluster {cluster.name} was deleted, removing cluster finalizer...")
    cluster.remove_cluster_finalizer()


from enum import Enum
from typing import Callable, cast
from .. import kubeutils

class PatchTarget(Enum):
    STS = "STS"
    DEPLOY = "DEPLOYMENT"
    CM = "CONFIGMAP"

class ApiCommandType(Enum):
    PATCH_STS = "PATCH_STS"
    PATCH_DEPLOY = "PATCH_DEPLOY"
    CREATE_CM = "CREATE_CM"
    DELETE_CM = "DELETE_CM"
    REPLACE_CM = "REPLACE_CM"
    PATCH_CM = "PATCH_CM"

OnApiExceptionHandler = Callable[[ApiException, Logger], None]

def on_apiexception_404_handler(exc: ApiException, logger: Logger):
    if exc.status == 404:
        logger.warning("Object not found! Exception: {exc}")
        return
    raise exc

def on_apiexception_generic_handler(exc: ApiException, logger: Logger):
    logger.warning("ApiException: {exc}")


class ApiCommand:
    def __init__(self,
                 type: ApiCommandType,
                 namespace: str,
                 name: str,
                 body: Optional[dict] = None,
                 on_api_exception: Optional[OnApiExceptionHandler] = None):
        self.type = type
        self.namespace = namespace
        self.name = name
        self.body = body
        self.on_api_exception = on_api_exception

    def run(self, logger: Logger) -> Optional[api_client.V1Status]:
        try:
            if self.type == ApiCommandType.CREATE_CM:
                status = cast(api_client.V1Status,
                              api_core.create_namespaced_config_map(self.namespace, self.body))
            elif self.type == ApiCommandType.DELETE_CM:
                delete_body = api_client.V1DeleteOptions(grace_period_seconds=0)
                status = cast(api_client.V1Status,
                              api_core.delete_namespaced_config_map(self.name, self.namespace, body=delete_body))
                return status
            elif self.type == ApiCommandType.REPLACE_CM:
                status = cast(api_client.V1Status,
                              api_core.replace_namespaced_config_map(self.name, self.namespace, body=self.body))
            elif self.type == ApiCommandType.PATCH_CM:
                status = cast(api_client.V1Status,
                              api_core.patch_namespaced_config_map(self.name, self.namespace, self.body))
        except kubeutils.ApiException as exc:
            if self.on_api_exception is not None:
                self.on_api_exception(exc, logger)
            else:
                raise

        return status


def snail_to_camel(s: str) -> str:
    if s.find("_") == -1:
        return s
    # Special case for '_exec'
    # For some reason for preStop with `exec` when dict-ified we get `pre_stop`` with `_exec`
    #  'lifecycle': {
    #    'post_start': None,
    #     'pre_stop': {'_exec': {'command': ['sh', '-c', 'sleep 60 && mysqladmin -ulocalroot shutdown']},
    # If we don't handle that it becomes `Exec`
    if len(s) and s[0] == "_":
        s = s[1:]

    words = s.split("_")
    ret = words[0] + "".join(word.title() for word in words[1:])
    return ret

def item_snail_to_camel(item):
    if isinstance(item, dict):
        # k8s API will return some fields as None, like
        # spec.containers[1].readinessProbe` : Required value: must specify a handler type
        # spec.containers[1].startupProbe: Required value: must specify a handler type
        # So we strip here the None values. Might hit somewhere where None is legit but for now it works!
        return {snail_to_camel(key):item_snail_to_camel(value) for key, value in item.items() if value is not None}
    if isinstance(item, list):
        return [item_snail_to_camel(value) for value in item]
    return item

def spec_to_dict(spec) -> dict:
    return item_snail_to_camel(spec.to_dict())

def strategic_merge(original, patch):
    if isinstance(original, dict) and isinstance(patch, dict):
        return merge_dicts(original, patch)
    elif isinstance(original, list) and isinstance(patch, list):
        return original + patch
    return patch

def merge_dicts(original, patch):
    for key, value in patch.items():
        if key in original:
            original[key] = strategic_merge(original[key], value)
        else:
            original[key] = value
    return original


class InnoDBClusterObjectModifier:
    def __init__(self, cluster: InnoDBCluster, logger: Logger):
        self.server_sts_patch = {}
        self.sts_changed = False
        self.sts_template_changed = False
        self.deploy_changed = False
        self.router_deploy_patch = {}
        self.cluster = cluster
        self.logger = logger
        self.commands: list[ApiCommand] = []
        self.sts = self.cluster.get_stateful_set()
        self.sts.spec = spec_to_dict(self.sts.spec)
        self.sts_spec_changed = False

    def _apply_server_sts_patch_to_sts_spec_if_needed(self):
        if len(self.server_sts_patch):
            # update with accumulated patches before overwriting
            self.logger.info(f"Applying accumulated patches {self.server_sts_patch['spec']} to sts.spec")
            utils.merge_patch_object(self.sts.spec, self.server_sts_patch["spec"], none_deletes=True)
            self.sts_spec_changed = True
            self.server_sts_patch = {}

    def _get_or_patch_sts_path(self, path: str, patch: Optional[dict] = None):
        self.logger.info(f"get_sts_path: patch_path={path}\n")
        # patches could be cached in self.server_sts_patch, so apply them, if any, before returning parts of self.sts.spec
        self._apply_server_sts_patch_to_sts_spec_if_needed()
        base = self.sts.spec
        # first is leading backslash, then is 'spec', so we skip
        path_elements = path.split("/")[2:]
        if len(path) > 1:
            for path_element in path_elements[0:-1]:
                #self.logger.info(f"{path_element} in base = {path_element in base}\n")
                assert path_element in base
                base = base[path_element]
            if patch is not None:
                base[path_elements[-1]] = patch
                self.logger.info(f"get_sts_path: after patching self.sts.spec={self.sts.spec}")
        return base[path_elements[-1]]

    def get_sts_path(self, path: str):
        return self._get_or_patch_sts_path(path, None)

    def patch_sts(self, patch: dict) -> None:
        self.sts_changed = True
        if "template" in patch:
            self.sts_template_changed = True
        self.logger.info(f"Accumulating patch={patch}\n")
        # cache the patches without merging into self.sts.spec
        # in case there is no call to patch_sts_overwrite then we won't "replace"
        # the existing sts object but "patch" it
        # if an sts_overwrite happens, we have to apply the patches to the self.sts.spec before overwriting
        utils.merge_patch_object(self.server_sts_patch, patch, none_deletes=True)

    def patch_sts_overwrite(self, patch: dict, patch_path: str) -> None:
        self.sts_changed = True
        if "template" in patch:
            self.sts_template_changed = True

        self.sts_spec_changed = True
        self._get_or_patch_sts_path(patch_path, patch)
        return

        if len(self.server_sts_patch):
            # update with accumulated patches before overwriting
            self.logger.info(f"Applying accumulated patches before applying overwriting patch patch={self.server_sts_patch['spec']}")
            #self.logger.info(f"STS.spec before apply={self.sts.spec}")
            utils.merge_patch_object(self.sts.spec, self.server_sts_patch["spec"], none_deletes=True)
            #self.logger.info(f"STS.spec after  apply={self.sts.spec}")
            self.server_sts_patch = {}

        #self.logger.info(f"patch_sts_overwrite: patch_path={patch_path} patch={patch}\n")

        base = self.sts.spec
        # first is leading backslash, then is spec, so we skip
        patch_path_elements = patch_path.split("/")[2:]
        if len(patch_path) > 1:
            for patch_path_element in patch_path_elements[0:-1]:
                #self.logger.info(f"{patch_path_element} in base = {patch_path_element in base}")
                assert patch_path_element in base
                base = base[patch_path_element]
            #  base[]
            #self.logger.info(f"\nExchanging {base[patch_path_elements[-1]]} \nwith\n{patch}")
            base[patch_path_elements[-1]] = patch
            self.logger.info(f"\n\nself.sts.spec={self.sts.spec}\n\n")


    def patch_deploy(self, patch: dict) -> None:
        self.deploy_changed = True
        self.logger.info(f"patch={patch}")
        utils.merge_patch_object(self.router_deploy_patch, patch, none_deletes=True)

    def create_configmap(self, namespace: str, name: str, body: dict, on_api_exception: Optional[OnApiExceptionHandler]) -> None:
        self.commands.append(ApiCommand(ApiCommandType.CREATE_CM, namespace, name, body, on_api_exception))

    def delete_configmap(self, namespace: str, name: str, on_api_exception: Optional[OnApiExceptionHandler]) -> None:
        self.commands.append(ApiCommand(ApiCommandType.DELETE_CM, namespace, name, None, on_api_exception))

    def replace_configmap(self, namespace: str, name: str, body: dict, on_api_exception: Optional[OnApiExceptionHandler]) -> None:
        self.commands.append(ApiCommand(ApiCommandType.REPLACE_CM, namespace, name, body, on_api_exception))

    def patch_configmap(self, namespace: str, name: str, patch: dict, on_api_exception: Optional[OnApiExceptionHandler]) -> None:
        self.commands.append(ApiCommand(ApiCommandType.PATCH_CM, namespace, name, patch, on_api_exception))

    def submit_patches(self) -> None:
        self.logger.info(f"InnoDBClusterObjectModifier::submit_patches sts_changed={self.sts_changed} sts_spec_changed={self.sts_spec_changed} len(router_deploy_patch)={len(self.router_deploy_patch)} len(commands)={len(self.commands)}")
        if (self.sts_changed or len(self.router_deploy_patch) or len(self.commands)):
              if len(self.commands):
                  for command in self.commands:
                      command.run(self.logger)
              if self.sts_changed:
                  if self.sts_spec_changed:
                      # this should apply server_sts_patch over self.sts.spec and empty self.server_sts_patch
                      # in the next step we will `replace` the STS and not `patch` it
                      # Only if the self.sts.spec is not touched should be server_sts_patch be applied, as otherwise
                      # changes to self.sts.spec will be skipped/forgotten
                      self._apply_server_sts_patch_to_sts_spec_if_needed()

                  if len(self.server_sts_patch):
                      self.logger.info(f"Patching STS.spec with {self.server_sts_patch}")
                      if self.sts_template_changed:
                          restart_patch = {"spec":{"template":{"metadata":{"annotations":{"kubectl.kubernetes.io/restartedAt":utils.isotime()}}}}}
                          utils.merge_patch_object(self.server_sts_patch, restart_patch)
                      api_apps.patch_namespaced_stateful_set(self.sts.metadata.name, self.sts.metadata.namespace, body=self.server_sts_patch)
                      self.server_sts_patch = {}
                  else:
                      self.logger.info(f"Replacing STS.spec with {self.sts.spec}")
                      # only if template has been changed. It could be that only scale up/down (spec['replica'] changed) has happened
                      # in that case if we set an annotation in the template the whole STS will be rolled over and this is not
                      # what is wanted. If replica count is increased only new pods are needed, respectively when replica is decreased.
                      # if replica is up and we set the annotation actually what will happen is rollover update and then scale up - we disturb the cluster
                      # (and tests will fail)
                      if self.sts_template_changed:
                          if not "annotations" in self.sts.spec["template"]["metadata"] or self.sts.spec["template"]["metadata"]["annotations"] is None:
                              self.sts.spec["template"]["metadata"]["annotations"] = {}
                          self.sts.spec["template"]["metadata"]["annotations"]["kubectl.kubernetes.io/restartedAt"] = utils.isotime()
                      api_apps.replace_namespaced_stateful_set(self.sts.metadata.name, self.sts.metadata.namespace, body=self.sts)
              if len(self.router_deploy_patch) and (deploy:= self.cluster.get_router_deployment()):
                  self.logger.info(f"Patching Deployment with {self.router_deploy_patch}")
                  router_objects.update_deployment_spec(deploy, self.router_deploy_patch)