[DPE-5588] Check against invalid arch (#539)

Author: sinclert-canonical

lib/charms/mysql/v0/architecture.py

@@ -0,0 +1,93 @@
+# Copyright 2024 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Library to provide hardware architecture checks for VMs and K8s charms.
+
+The WrongArchitectureWarningCharm class is designed to be used alongside
+the is-wrong-architecture helper function, as follows:
+
+```python
+from ops import main
+from charms.mysql.v0.architecture import WrongArchitectureWarningCharm, is_wrong_architecture
+
+if __name__ == "__main__":
+    if is_wrong_architecture():
+        main(WrongArchitectureWarningCharm)
+```
+"""
+
+import logging
+import os
+import pathlib
+import platform
+
+import yaml
+from ops.charm import CharmBase
+from ops.model import BlockedStatus
+
+# The unique Charmhub library identifier, never change it
+LIBID = "827e04542dba4c2a93bdc70ae40afdb1"
+LIBAPI = 0
+LIBPATCH = 1
+
+PYDEPS = ["ops>=2.0.0", "pyyaml>=5.0"]
+
+
+logger = logging.getLogger(__name__)
+
+
+class WrongArchitectureWarningCharm(CharmBase):
+    """A fake charm class that only signals a wrong architecture deploy."""
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        hw_arch = platform.machine()
+        self.unit.status = BlockedStatus(
+            f"Charm incompatible with {hw_arch} architecture. "
+            f"If this app is being refreshed, rollback"
+        )
+        raise RuntimeError(
+            f"Incompatible architecture: this charm revision does not support {hw_arch}. "
+            f"If this app is being refreshed, rollback with instructions from Charmhub docs. "
+            f"If this app is being deployed for the first time, remove it and deploy it again "
+            f"using a compatible revision."
+        )
+
+
+def is_wrong_architecture() -> bool:
+    """Checks if charm was deployed on wrong architecture."""
+    charm_path = os.environ.get("CHARM_DIR", "")
+    manifest_path = pathlib.Path(charm_path, "manifest.yaml")
+
+    if not manifest_path.exists():
+        logger.error("Cannot check architecture: manifest file not found in %s", manifest_path)
+        return False
+
+    manifest = yaml.safe_load(manifest_path.read_text())
+
+    manifest_archs = []
+    for base in manifest["bases"]:
+        base_archs = base.get("architectures", [])
+        manifest_archs.extend(base_archs)
+
+    hardware_arch = platform.machine()
+    if ("amd64" in manifest_archs and hardware_arch == "x86_64") or (
+        "arm64" in manifest_archs and hardware_arch == "aarch64"
+    ):
+        logger.debug("Charm architecture matches")
+        return False
+
+    logger.error("Charm architecture does not match")
+    return True

src/charm.py

@@ -4,6 +4,12 @@
 
 """Charm for MySQL."""
 
+from charms.mysql.v0.architecture import WrongArchitectureWarningCharm, is_wrong_architecture
+from ops.main import main
+
+if is_wrong_architecture() and __name__ == "__main__":
+    main(WrongArchitectureWarningCharm)
+
 import logging
 import random
 from socket import getfqdn
@@ -44,7 +50,6 @@
 from charms.tempo_coordinator_k8s.v0.tracing import TracingEndpointRequirer
 from ops import EventBase, RelationBrokenEvent, RelationCreatedEvent
 from ops.charm import RelationChangedEvent, UpdateStatusEvent
-from ops.main import main
 from ops.model import (
     ActiveStatus,
     BlockedStatus,

tests/integration/helpers.py

@@ -8,7 +8,8 @@
 import string
 import subprocess
 import tempfile
-from typing import Dict, List, Optional
+from pathlib import Path
+from typing import Dict, List, Optional, Union
 
 import mysql.connector
 import yaml
@@ -771,3 +772,17 @@ async def dispatch_custom_event_for_logrotate(ops_test: OpsTest, unit_name: str)
     )
 
     assert return_code == 0
+
+
+async def get_charm(charm_path: Union[str, Path], architecture: str, bases_index: int) -> Path:
+    """Fetches packed charm from CI runner without checking for architecture."""
+    charm_path = Path(charm_path)
+    charmcraft_yaml = yaml.safe_load((charm_path / "charmcraft.yaml").read_text())
+    assert charmcraft_yaml["type"] == "charm"
+
+    base = charmcraft_yaml["bases"][bases_index]
+    build_on = base.get("build-on", [base])[0]
+    version = build_on["channel"]
+    packed_charms = list(charm_path.glob(f"*{version}-{architecture}.charm"))
+
+    return packed_charms[0].resolve(strict=True)

tests/integration/test_architecture.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+from pathlib import Path
+
+import pytest
+import yaml
+from pytest_operator.plugin import OpsTest
+
+from . import markers
+from .helpers import get_charm
+
+METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
+APP_NAME = METADATA["name"]
+
+
+@pytest.mark.group(1)
+@markers.amd64_only
+async def test_arm_charm_on_amd_host(ops_test: OpsTest) -> None:
+    """Tries deploying an arm64 charm on amd64 host."""
+    charm = await get_charm(".", "arm64", 1)
+
+    await ops_test.model.deploy(
+        charm,
+        application_name=APP_NAME,
+        num_units=1,
+        config={"profile": "testing"},
+        resources={"mysql-image": METADATA["resources"]["mysql-image"]["upstream-source"]},
+        base="[email protected]",
+    )
+
+    await ops_test.model.wait_for_idle(
+        apps=[APP_NAME],
+        status="error",
+        raise_on_error=False,
+    )
+
+
+@pytest.mark.group(1)
+@markers.arm64_only
+async def test_amd_charm_on_arm_host(ops_test: OpsTest) -> None:
+    """Tries deploying an amd64 charm on arm64 host."""
+    charm = await get_charm(".", "amd64", 0)
+
+    await ops_test.model.deploy(
+        charm,
+        application_name=APP_NAME,
+        num_units=1,
+        config={"profile": "testing"},
+        resources={"mysql-image": METADATA["resources"]["mysql-image"]["upstream-source"]},
+        base="[email protected]",
+    )
+
+    await ops_test.model.wait_for_idle(
+        apps=[APP_NAME],
+        status="error",
+        raise_on_error=False,
+    )