Add --cleanup CLI flag with idempotent deploy

CHANGELOG.md

@@ -15,6 +15,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Dependabot updates for `github-actions` and `terraform` ecosystems. (#29)
 * CI workflow to build and push service images to GitHub Container Registry on master merge. (#97)
 * Pull-first with local build fallback for container images during deployment. (#97)
+* `--cleanup` CLI flag with idempotent deploy and cleanup actions. (#49)
+* Credential recovery for partial redeploy from existing healthy nodes. (#49)
+* PEM serialization support in `crypto.py` for credential recovery. (#49)
+* InnoDB Cluster auto-join for fresh MySQL nodes added to an existing cluster. (#49)
+* Disaster recovery test flow with node replacement and partial redeploy verification. (#49)
 * Swarm overlay network encryption via IPsec and explicit data path port. (#92)
 * MySQL slow query log for queries exceeding 10 seconds or not using indexes. (#76)
 * Strengthen MySQL superset password: secrets module, expanded charset, 24 chars. (#89)
@@ -26,6 +31,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Completed [ARCHITECTURE.md](./docs/ARCHITECTURE.md) (#93)
 * Migrated CI from self-hosted to GitHub-hosted runners with Docker-in-Docker test infrastructure. (#94)
 * Test workflow always builds service images locally for reproducibility. (#97)
+* Restructured test suite into 7-stage flow: sanity, deploy, functional, disaster,
+  recovery, redeploy, post-redeploy functional. (#49)
+* InnoDB Cluster initcontainer scans all MySQL nodes to find existing cluster and adds missing
+  members instead of only checking the primary node. (#49)
+* Made Terraform test infrastructure idempotent: removed `always_run` triggers, node recovery
+  reruns `terraform apply` which only recreates missing containers. (#49)
 * Enabled parallel replication applier threads on secondary MySQL nodes. (#75)
 * Clean up temporary `.pyc` files on remote nodes after execution. (#51)
 * Set `innodb_flush_method=O_DIRECT` to eliminate double caching in containers. (#74)

services/mysql-mgmt/docker_compose.yml

@@ -13,9 +13,17 @@ services:
       - |
         set -euo pipefail
         export MYSQL_SUPERSET_PASSWORD=$(</run/secrets/mysql_superset_password)
-        if ! mysqlsh \
-               --login-path="${PRIMARY_MYSQL_NODE}" \
-               --execute="dba.getCluster();"; then
+        CLUSTER_NODE=""
+        for attempt in 1 2 3; do
+          for node in "${PRIMARY_MYSQL_NODE}" "${SECONDARY_FIRST_MYSQL_NODE}" "${SECONDARY_SECOND_MYSQL_NODE}"; do
+            if mysqlsh --login-path="$${node}" --execute="dba.getCluster('superset');" 2>/dev/null; then
+              CLUSTER_NODE="$${node}"
+              break 2
+            fi
+          done
+          [ "$${attempt}" -lt 3 ] && sleep 5 || true
+        done
+        if [ -z "$${CLUSTER_NODE}" ]; then
           mysqlsh \
             --login-path="${PRIMARY_MYSQL_NODE}" \
             --execute="dba.configureInstance('${PRIMARY_MYSQL_NODE}')"
@@ -46,18 +54,32 @@ services:
             --login-path="${PRIMARY_MYSQL_NODE}" \
             --sql \
             --file="/opt/superset_user.sql"
+          CLUSTER_NODE="${PRIMARY_MYSQL_NODE}"
+        else
+          CLUSTER_STATUS=$(mysqlsh --login-path="$${CLUSTER_NODE}" \
+            --execute="print(dba.getCluster('superset').status())" 2>/dev/null)
+          for node in "${PRIMARY_MYSQL_NODE}" "${SECONDARY_FIRST_MYSQL_NODE}" "${SECONDARY_SECOND_MYSQL_NODE}"; do
+            if ! echo "$${CLUSTER_STATUS}" | grep -q "\"address\": \"$${node}:3306\""; then
+              mysqlsh --login-path="$${node}" \
+                --execute="dba.configureInstance('$${node}')"
+              mysqlsh --login-path="$${node}" --sql \
+                --execute="RESET MASTER;"
+              mysqlsh --login-path="$${CLUSTER_NODE}" \
+                --execute="dba.getCluster('superset').addInstance('$${node}',{recoveryMethod:'incremental'});"
+            fi
+          done
         fi
         export NODE_IP_ADDRESS=$(mysqlsh --python --execute "exec(open('/opt/interfaces.py').read())")
         /opt/envsubst-Linux-x86_64 < "/opt/superset_user.sql.tpl" > "/opt/superset_user.sql"
         mysqlsh \
-          --login-path="${PRIMARY_MYSQL_NODE}" \
+          --login-path="$${CLUSTER_NODE}" \
           --sql \
           --file="/opt/superset_user.sql"
         mysqlrouter \
           --user \
             "superset" \
           --bootstrap \
-            "superset:$(cat /run/secrets/mysql_superset_password)@${PRIMARY_MYSQL_NODE}:3306" \
+            "superset:$(cat /run/secrets/mysql_superset_password)@$${CLUSTER_NODE}:3306" \
           --directory \
             "/opt/default/mysql_router" \
           --conf-use-sockets \
@@ -72,7 +94,7 @@ services:
           --client-ssl-key \
             "/opt/default/mysql_router_key.pem"
         mysqlsh \
-          --login-path="${PRIMARY_MYSQL_NODE}" --sql \
+          --login-path="$${CLUSTER_NODE}" --sql \
           --execute="DROP USER 'superset'@'$(mysqlsh --python --execute "exec(open('/opt/interfaces.py').read())")';"
         ifmetric ${VIRTUAL_NETWORK_INTERFACE} 2
         /opt/envsubst-Linux-x86_64 < "/opt/keepalived.conf.tpl" > "/opt/default/keepalived.conf";

src/crypto.py

@@ -129,6 +129,23 @@ def generate_certificate(
                 backend=cryptography.hazmat.backends.default_backend()
         )
 
+    @staticmethod
+    def serialization(pem: str
+                      ) -> cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey | cryptography.x509.Certificate:
+        pem_bytes = pem.encode('utf-8')
+        if 'PRIVATE KEY' in pem:
+            private_key = cryptography.hazmat.primitives.serialization.load_pem_private_key(
+                pem_bytes,
+                password=None,
+                backend=cryptography.hazmat.backends.default_backend()
+            )
+            assert isinstance(private_key, cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey)
+            return private_key
+        return cryptography.x509.load_pem_x509_certificate(
+            pem_bytes,
+            backend=cryptography.hazmat.backends.default_backend()
+        )
+
     @staticmethod
     def deserialization(pki:
                         cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey |

src/initialize.py

@@ -62,6 +62,8 @@
 import re
 import socket
 
+import cryptography.hazmat.primitives.asymmetric.rsa
+
 import crypto
 import decorators
 import remote
@@ -123,13 +125,69 @@ def __init__(self) -> None:
         ]  # type: ignore[assignment]
         self.cert_manager = crypto.OpenSSL()
 
+    def _recover_existing_credentials(self) -> bool:
+        for node in self.mysql_nodes:
+            try:
+                _, stdout, _ = node.ssh_client.exec_command(
+                    "cat /opt/superset-cluster/mysql-server/mysql_root_password"
+                )
+                password = stdout.read().decode().strip()
+                if not password:
+                    continue
+                _, stdout, _ = node.ssh_client.exec_command(
+                    "cat /opt/superset-cluster/mysql-server/superset_cluster_ca_key.pem"
+                )
+                ca_key_pem = stdout.read().decode().strip()
+                _, stdout, _ = node.ssh_client.exec_command(
+                    "cat /opt/superset-cluster/mysql-server/superset_cluster_ca_certificate.pem"
+                )
+                ca_bundle = stdout.read().decode().strip()
+                if not (ca_key_pem and ca_bundle):
+                    continue
+                certs = ca_bundle.split('-----END CERTIFICATE-----')
+                ca_cert_pem = [c for c in certs if '-----BEGIN CERTIFICATE-----' in c][-1]
+                ca_cert_pem = ca_cert_pem.strip() + '\n-----END CERTIFICATE-----\n'
+                deserialized_key = self.cert_manager.serialization(ca_key_pem)
+                assert isinstance(deserialized_key, cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey)
+                self.ca_key = deserialized_key
+                self.ca_certificate = self.cert_manager.serialization(ca_cert_pem)
+                self.mysql_root_password = password
+                break
+            except (OSError, IndexError, ValueError):
+                continue
+        else:
+            return False
+        for node in self.mgmt_nodes:
+            try:
+                _, stdout, _ = node.ssh_client.exec_command(
+                    "cat /opt/superset-cluster/mysql-mgmt/mysql_superset_password"
+                )
+                superset_pw = stdout.read().decode().strip()
+                if not superset_pw:
+                    continue
+                _, stdout, _ = node.ssh_client.exec_command(
+                    "docker exec $(docker ps --filter name=superset"
+                    " --format '{{.ID}}' | head -1)"
+                    " cat /run/secrets/superset_secret_key 2>/dev/null"
+                )
+                secret_key = stdout.read().decode().strip()
+                if not secret_key:
+                    continue
+                self.mysql_superset_password = superset_pw
+                self.superset_secret_key = secret_key
+                return True
+            except OSError:
+                continue
+        return False
+
     @decorators.Overlay.run_selected_methods_once
     def credentials(self) -> None:
-        self.ca_key = self.cert_manager.generate_private_key()
-        self.ca_certificate = self.cert_manager.generate_certificate('Superset-Cluster', self.ca_key)
-        self.mysql_root_password = self.cert_manager.generate_mysql_root_password()
-        self.mysql_superset_password = self.cert_manager.generate_mysql_superset_password()
-        self.superset_secret_key = self.cert_manager.generate_superset_secret_key()
+        if not self._recover_existing_credentials():
+            self.ca_key = self.cert_manager.generate_private_key()
+            self.ca_certificate = self.cert_manager.generate_certificate('Superset-Cluster', self.ca_key)
+            self.mysql_root_password = self.cert_manager.generate_mysql_root_password()
+            self.mysql_superset_password = self.cert_manager.generate_mysql_superset_password()
+            self.superset_secret_key = self.cert_manager.generate_superset_secret_key()
         for node in list(itertools.chain(self.mysql_nodes, self.mgmt_nodes)):
             node.key = self.cert_manager.generate_private_key()
             node.csr = self.cert_manager.generate_csr(f'Superset-Cluster-{node.node}', node.key)
@@ -166,6 +224,13 @@ def get_mylogin_cnf(self, node: remote.RemoteConnection) -> bytes:
 
     def start_mysql_servers(self) -> None:
         for node in self.mysql_nodes:
+            _, stdout, _ = node.ssh_client.exec_command(
+                "docker inspect --format='{{.State.Health.Status}}'"
+                " mysql 2>/dev/null"
+            )
+            if stdout.read().decode().strip() == "healthy":
+                continue
+            node.ssh_client.exec_command("docker rm -f mysql 2>/dev/null")
             node.upload_directory(
                 local_directory_path="./services/mysql-server",
                 remote_directory_path="/opt/superset-cluster/mysql-server"
@@ -197,6 +262,16 @@ def start_mysql_servers(self) -> None:
             )
 
     def start_mysql_mgmt(self, node: remote.RemoteConnection, state: str, priority: int) -> None:
+        _, stdout, _ = node.ssh_client.exec_command(
+            "docker inspect --format='{{.State.Health.Status}}'"
+            " mysql-mgmt 2>/dev/null"
+        )
+        if stdout.read().decode().strip() == "healthy":
+            return
+        node.ssh_client.exec_command(
+            "docker rm -f mysql-mgmt mysql-mgmt-initcontainer 2>/dev/null;"
+            " docker volume rm mysql-mgmt_default_generated 2>/dev/null"
+        )
         node.upload_directory(
             local_directory_path="./services/mysql-mgmt",
             remote_directory_path="/opt/superset-cluster/mysql-mgmt"
@@ -253,6 +328,25 @@ def start_mysql_mgmt(self, node: remote.RemoteConnection, state: str, priority:
 
     def start_superset(self) -> None:
         for node in self.mgmt_nodes:
+            _, redis_out, _ = node.ssh_client.exec_command(
+                "docker inspect --format='{{.State.Health.Status}}'"
+                " redis 2>/dev/null"
+            )
+            _, svc_out, _ = node.ssh_client.exec_command(
+                "docker service ps superset"
+                " --format='{{.CurrentState}}'"
+                " --filter desired-state=running 2>/dev/null"
+            )
+            redis_healthy = redis_out.read().decode().strip() == "healthy"
+            superset_running = "Running" in svc_out.read().decode()
+            if redis_healthy and superset_running:
+                continue
+            node.ssh_client.exec_command(
+                "docker service rm superset 2>/dev/null;"
+                " docker rm -f redis 2>/dev/null;"
+                " docker swarm leave --force 2>/dev/null;"
+                " docker network rm superset-network docker_gwbridge 2>/dev/null"
+            )
             node.upload_directory(
                 local_directory_path="./services/superset",
                 remote_directory_path='/opt/superset-cluster/superset'
@@ -285,6 +379,41 @@ def start_superset(self) -> None:
                           mysql_superset_password=self.mysql_superset_password)
             )
 
+    def teardown_node(self, node: remote.RemoteConnection, is_mgmt: bool = False) -> None:
+        commands = [
+            "if docker info --format '{{.Swarm.LocalNodeState}}' 2>/dev/null"
+            " | grep -qvx inactive; then docker swarm leave --force; fi",
+            "docker ps -aq | xargs -r docker rm -f",
+            "docker volume ls -q | xargs -r docker volume rm",
+            "docker network ls --filter name=superset-network -q"
+            " | xargs -r docker network rm",
+            "docker network ls --filter name=docker_gwbridge -q"
+            " | xargs -r docker network rm",
+            "if [ -d /opt/superset-cluster ]; then"
+            " rm -rf /opt/superset-cluster; fi",
+        ]
+        if is_mgmt:
+            commands.extend([
+                "if ip addr show dev {iface} 2>/dev/null"
+                " | grep -q '{vip}/'; then"
+                " ip addr del {vip}/{mask} dev {iface}; fi".format(
+                    iface=self.virtual_network_interface,
+                    vip=self.virtual_ip_address,
+                    mask=self.virtual_network_mask),
+                "if ip route show | grep -q '{vip} '; then"
+                " ip route del {vip}; fi".format(
+                    vip=self.virtual_ip_address),
+            ])
+        _, stdout, stderr = node.ssh_client.exec_command(
+            "rc=0; " + "; ".join(f"{{ {cmd}; }} || rc=$?" for cmd in commands) + "; exit $rc"
+        )
+        exit_status = stdout.channel.recv_exit_status()
+        if exit_status != 0:
+            raise RuntimeError(
+                f"Teardown failed on {node.node} (exit {exit_status}): "
+                f"{stderr.read().decode().strip()}"
+            )
+
     def start_cluster(self) -> None:
         try:
             self.start_mysql_servers()
@@ -296,9 +425,33 @@ def start_cluster(self) -> None:
                 node.ssh_client.close()
                 node.sftp_client.close()
 
+    def cleanup(self) -> None:
+        try:
+            for node in self.mgmt_nodes:
+                try:
+                    self.teardown_node(node, is_mgmt=True)
+                except (OSError, RuntimeError):
+                    pass
+            for node in self.mysql_nodes:
+                try:
+                    self.teardown_node(node, is_mgmt=False)
+                except (OSError, RuntimeError):
+                    pass
+        finally:
+            for node in list(itertools.chain(self.mysql_nodes, self.mgmt_nodes)):
+                node.ssh_client.close()
+                node.sftp_client.close()
+
 
 if __name__ == "__main__":
-    if len(sys.argv) != 6:
+    if len(sys.argv) < 6:
         print("Invalid form of arguments provided")
         sys.exit(1)
-    Controller().start_cluster()
+
+    action = sys.argv[6] if len(sys.argv) > 6 else "deploy"
+    controller = Controller()
+
+    if action == "cleanup":
+        controller.cleanup()
+    else:
+        controller.start_cluster()

src/remote.py

@@ -124,15 +124,15 @@ def run_python_container_command(self, command: str) -> dict:
             encoding="utf-8"
         ) as memfile:
             source = memfile.read() + command
-        self.upload_file(content=source, remote_file_path=f'/opt/{nonce}.py')
+        self.upload_file(content=source, remote_file_path=f'/opt/superset-cluster/{nonce}.py')
         _, stdout, stderr = self.ssh_client.exec_command(
-            f"PYTHONPATH=/home/superset/.local/lib/python3.10/site-packages python3 /opt/{nonce}.py"
+            f"PYTHONPATH=/home/superset/.local/lib/python3.10/site-packages python3 /opt/superset-cluster/{nonce}.py"
         )
         result = {
             "output": stdout.read().decode(),
             "error": stderr.read().decode()
         }
-        self.ssh_client.exec_command(f"rm -f /opt/{nonce}.py")
+        self.ssh_client.exec_command(f"rm -f /opt/superset-cluster/{nonce}.py")
         return result
 
     def upload_directory(self, local_directory_path: str, remote_directory_path: str) -> None:

superset-cluster

@@ -20,6 +20,8 @@ display_help() {
     echo "  --virtual-network-mask <mask>           Network mask for the virtual network gateway."
     echo "                                          Example: --virtual-network-mask 24"
     echo
+    echo "  --cleanup                               Remove all cluster resources from the specified nodes."
+    echo
     echo "  -h, --help                              Show this help message and exit."
     echo
     echo "Example:"
@@ -33,6 +35,7 @@ display_help() {
 
 
 parse_arguments() {
+    action="deploy"
     while [[ "$#" -gt 0 ]]; do
         case $1 in
             --mgmt-nodes)
@@ -59,6 +62,9 @@ parse_arguments() {
                 display_help
                 exit 0
                 ;;
+            --cleanup)
+                action="cleanup"
+                ;;
             *)
                 echo "Unknown option: $1"
                 display_help
@@ -77,14 +83,16 @@ if python --version &>/dev/null; then
     '${virtual_network_interface}' \
     '${virtual_network_mask}' \
     '${mgmt_nodes[*]}' \
-    '${mysql_nodes[*]}'"
+    '${mysql_nodes[*]}' \
+    '${action}'"
 elif python3 --version &>/dev/null; then
   eval "python3 ./src/initialize.py \
     '${virtual_ip_address}' \
     '${virtual_network_interface}' \
     '${virtual_network_mask}' \
     '${mgmt_nodes[*]}' \
-    '${mysql_nodes[*]}'"
+    '${mysql_nodes[*]}' \
+    '${action}'"
 else
   echo "Neither python nor python3 is set as the main executable in this environment, check python location with 'which python'."
   exit 1