Add in-place upgrades (#88)

Derived from
canonical/mysql-router-k8s-operator#138

Author: carlcsaposs-canonical

actions.yaml

@@ -0,0 +1,7 @@
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+resume-upgrade:
+  description: Upgrade remaining units (after you manually verified that upgraded units are healthy).
+force-upgrade:
+  description: Force upgrade of this unit.

charm_version

@@ -0,0 +1 @@
+1

charmcraft.yaml

@@ -28,3 +28,6 @@ parts:
           exit 1
       fi
     charm-entrypoint: src/machine_charm.py
+    prime:
+      - charm_version
+      - workload_version

metadata.yaml

@@ -36,6 +36,11 @@ requires:
     interface: juju-info
     scope: container
 peers:
+  upgrade-version-a:
+    # Relation versioning scheme:
+    # DA056 - Upgrading in-place upgrade protocol
+    # https://docs.google.com/document/d/1H7qy5SAwLiCOKO9xMQJbbQP5_-jGV6Lhi-mJOk4gZ08/edit
+    interface: upgrade
 # TODO TLS VM: re-enable peer relation
 #  mysql-router-peers:
 #    interface: mysql_router_peers

poetry.lock

@@ -12,6 +12,7 @@ authors = []
 python = "^3.8.1"  # ^3.8.1 required by flake8
 ops = "^2.6.0"
 tenacity = "^8.2.3"
+poetry-core = "^1.7.0"
 jinja2 = "^3.1.2"
 
 [tool.poetry.group.charm-libs.dependencies]

pyproject.toml

@@ -14,8 +14,10 @@
 import container
 import lifecycle
 import logrotate
+import machine_upgrade
 import relations.database_provides
 import relations.database_requires
+import upgrade
 import workload
 
 logger = logging.getLogger(__name__)
@@ -35,11 +37,26 @@ def __init__(self, *args) -> None:
         self._authenticated_workload_type = workload.AuthenticatedWorkload
         self._database_requires = relations.database_requires.RelationEndpoint(self)
         self._database_provides = relations.database_provides.RelationEndpoint(self)
-        self.framework.observe(self.on.update_status, self.reconcile_database_relations)
-        # Set status on first start if no relations active
-        self.framework.observe(self.on.start, self.reconcile_database_relations)
+        self.framework.observe(self.on.update_status, self.reconcile)
+        self.framework.observe(
+            self.on[upgrade.PEER_RELATION_ENDPOINT_NAME].relation_changed, self.reconcile
+        )
+        self.framework.observe(
+            self.on[upgrade.RESUME_ACTION_NAME].action, self._on_resume_upgrade_action
+        )
+        # (For Kubernetes) Reset partition after scale down
+        self.framework.observe(
+            self.on[upgrade.PEER_RELATION_ENDPOINT_NAME].relation_departed, self.reconcile
+        )
+        # Handle upgrade & set status on first start if no relations active
+        self.framework.observe(self.on.start, self.reconcile)
         # Update app status
-        self.framework.observe(self.on.leader_elected, self.reconcile_database_relations)
+        self.framework.observe(self.on.leader_elected, self.reconcile)
+        # Set versions in upgrade peer relation app databag
+        self.framework.observe(
+            self.on[upgrade.PEER_RELATION_ENDPOINT_NAME].relation_created,
+            self._upgrade_relation_created,
+        )
 
     @property
     @abc.abstractmethod
@@ -60,6 +77,11 @@ def _tls_certificate_saved(self) -> bool:
     def _container(self) -> container.Container:
         """Workload container (snap or ROCK)"""
 
+    @property
+    @abc.abstractmethod
+    def _upgrade(self) -> typing.Optional[upgrade.Upgrade]:
+        pass
+
     @property
     @abc.abstractmethod
     def _logrotate(self) -> logrotate.LogRotate:
@@ -95,8 +117,8 @@ def _prioritize_statuses(statuses: typing.List[ops.StatusBase]) -> ops.StatusBas
         """
         status_priority = (
             ops.BlockedStatus,
-            ops.WaitingStatus,
             ops.MaintenanceStatus,
+            ops.WaitingStatus,
             # Catch any unknown status type
             ops.StatusBase,
         )
@@ -108,6 +130,11 @@ def _prioritize_statuses(statuses: typing.List[ops.StatusBase]) -> ops.StatusBas
 
     def _determine_app_status(self, *, event) -> ops.StatusBase:
         """Report app status."""
+        if self._upgrade and (upgrade_status := self._upgrade.app_status):
+            # Upgrade status should take priority over relation status—even if the status level is
+            # normally lower priority.
+            # (Relations should not be modified during upgrade.)
+            return upgrade_status
         statuses = []
         for endpoint in (self._database_requires, self._database_provides):
             if status := endpoint.get_status(event):
@@ -117,25 +144,28 @@ def _determine_app_status(self, *, event) -> ops.StatusBase:
     def _determine_unit_status(self, *, event) -> ops.StatusBase:
         """Report unit status."""
         statuses = []
-        workload_ = self.get_workload(event=event)
-        statuses.append(workload_.get_status(event))
+        workload_status = self.get_workload(event=event).status
+        if self._upgrade:
+            statuses.append(self._upgrade.get_unit_juju_status(workload_status=workload_status))
+        statuses.append(workload_status)
         return self._prioritize_statuses(statuses)
 
-    def set_status(self, *, event) -> None:
+    def set_status(self, *, event, app=True, unit=True) -> None:
         """Set charm status."""
-        if self._unit_lifecycle.authorized_leader:
+        if app and self._unit_lifecycle.authorized_leader:
             self.app.status = self._determine_app_status(event=event)
             logger.debug(f"Set app status to {self.app.status}")
-        self.unit.status = self._determine_unit_status(event=event)
-        logger.debug(f"Set unit status to {self.unit.status}")
+        if unit:
+            self.unit.status = self._determine_unit_status(event=event)
+            logger.debug(f"Set unit status to {self.unit.status}")
 
     def wait_until_mysql_router_ready(self) -> None:
         """Wait until a connection to MySQL Router is possible.
 
         Retry every 5 seconds for up to 30 seconds.
         """
         logger.debug("Waiting until MySQL Router is ready")
-        self.unit.status = ops.WaitingStatus("MySQL Router starting")
+        self.unit.status = ops.MaintenanceStatus("MySQL Router starting")
         try:
             for attempt in tenacity.Retrying(
                 reraise=True,
@@ -156,21 +186,63 @@ def wait_until_mysql_router_ready(self) -> None:
     #  Handlers
     # =======================
 
-    def reconcile_database_relations(self, event=None) -> None:
-        """Handle database requires/provides events."""
+    def _upgrade_relation_created(self, _) -> None:
+        if self._unit_lifecycle.authorized_leader:
+            # `self._upgrade.is_compatible` should return `True` during first charm
+            # installation/setup
+            self._upgrade.set_versions_in_app_databag()
+
+    def reconcile(self, event=None) -> None:  # noqa: C901
+        """Handle most events."""
+        if not self._upgrade:
+            logger.debug("Peer relation not available")
+            return
+        if not self._upgrade.versions_set:
+            logger.debug("Peer relation not ready")
+            return
         workload_ = self.get_workload(event=event)
+        if self._upgrade.unit_state == "restarting":  # Kubernetes only
+            if not self._upgrade.is_compatible:
+                logger.info(
+                    "Upgrade incompatible. If you accept potential *data loss* and *downtime*, you can continue with `resume-upgrade force=true`"
+                )
+                self.unit.status = ops.BlockedStatus(
+                    "Upgrade incompatible. Rollback to previous revision with `juju refresh`"
+                )
+                self.set_status(event=event, unit=False)
+                return
+        elif isinstance(self._upgrade, machine_upgrade.Upgrade):  # Machines only
+            if not self._upgrade.is_compatible:
+                self.set_status(event=event)
+                return
+            if self._upgrade.unit_state == "outdated":
+                if self._upgrade.authorized:
+                    self._upgrade.upgrade_unit(
+                        workload_=workload_, tls=self._tls_certificate_saved
+                    )
+                else:
+                    self.set_status(event=event)
+                    logger.debug("Waiting to upgrade")
+                    return
         logger.debug(
             "State of reconcile "
             f"{self._unit_lifecycle.authorized_leader=}, "
             f"{isinstance(workload_, workload.AuthenticatedWorkload)=}, "
             f"{workload_.container_ready=}, "
-            f"{self._database_requires.is_relation_breaking(event)=}"
+            f"{self._database_requires.is_relation_breaking(event)=}, "
+            f"{self._upgrade.in_progress=}"
         )
         if self._unit_lifecycle.authorized_leader:
             if self._database_requires.is_relation_breaking(event):
+                if self._upgrade.in_progress:
+                    logger.warning(
+                        "Modifying relations during an upgrade is not supported. The charm may be in a broken, unrecoverable state. Re-deploy the charm"
+                    )
                 self._database_provides.delete_all_databags()
             elif (
-                isinstance(workload_, workload.AuthenticatedWorkload) and workload_.container_ready
+                not self._upgrade.in_progress
+                and isinstance(workload_, workload.AuthenticatedWorkload)
+                and workload_.container_ready
             ):
                 self._database_provides.reconcile_users(
                     event=event,
@@ -182,4 +254,25 @@ def reconcile_database_relations(self, event=None) -> None:
             workload_.enable(tls=self._tls_certificate_saved, unit_name=self.unit.name)
         elif workload_.container_ready:
             workload_.disable()
+        # Empty waiting status means we're waiting for database requires relation before starting
+        # workload
+        if not workload_.status or workload_.status == ops.WaitingStatus():
+            self._upgrade.unit_state = "healthy"
+        if self._unit_lifecycle.authorized_leader:
+            self._upgrade.reconcile_partition()
+            if not self._upgrade.in_progress:
+                self._upgrade.set_versions_in_app_databag()
         self.set_status(event=event)
+
+    def _on_resume_upgrade_action(self, event: ops.ActionEvent) -> None:
+        if not self._unit_lifecycle.authorized_leader:
+            message = f"Must run action on leader unit. (e.g. `juju run {self.app.name}/leader {upgrade.RESUME_ACTION_NAME}`)"
+            logger.debug(f"Resume upgrade event failed: {message}")
+            event.fail(message)
+            return
+        if not self._upgrade or not self._upgrade.in_progress:
+            message = "No upgrade in progress"
+            logger.debug(f"Resume upgrade event failed: {message}")
+            event.fail(message)
+            return
+        self._upgrade.reconcile_partition(action_event=event)