Skip to content

Instantly share code, notes, and snippets.

@ca-scribner
Last active January 6, 2025 16:45
Show Gist options
  • Save ca-scribner/9f7f903b3eefff64b2ba3824cb296f84 to your computer and use it in GitHub Desktop.
Save ca-scribner/9f7f903b3eefff64b2ba3824cb296f84 to your computer and use it in GitHub Desktop.
Automated handling of charm status requests from helpers

Often, Juju charms have helpers that encounter situations that should result in a Juju status (eg: a helper parsing charm config might see an invalid config that should result in BlockedStatus). Many charm code authors feel it is an antipattern to set charm status from a helper because it makes charms harder to read (from a top level, its hard to tell where in your charm statuses might be set), but in cases where the helper knows status should result of this situation (BlockedStatus("Blocked because X is unset")) or at least has context to add about the error it is hard to pass this information back to the charm. In particular, when trying to make resuable helpers that are shared across many charms, this can lead to a lot of complex logic and error handling that is implemented in every charm.

Below is a pattern modified/stolen from:

A simple version of this looks like (note this is untested code - think of it more as pseudocode rather than a full implementation):

# This could be a single exception like chisme's ErrorWithStatus, but Sunbean's way of using separate exceptions 
# feels more pythonic imo
class BaseStatusExceptionError(Exception):
    """Base status Exception."""
    def __init__(self, msg):
        self.msg = msg
        super().__init__(self.msg)


class BlockedExceptionError(BaseStatusExceptionError):
    """Charm is blocked."""
    pass


class MaintenanceExceptionError(BaseStatusExceptionError):
    """Charm is performing maintenance."""
    pass


class WaitingExceptionError(BaseStatusExceptionError):
    """Charm is waiting."""
    pass


class DemoStatusRaisedInHelpersCharm(ops.CharmBase):
    """Charm for demoing status raising in helpers."""

    def __init__(self, *args):
        super().__init__(*args)

        self.framework.observe(self.on.config_changed, self.on_config_changed)

    def on_config_changed(self, _event: ops.ConfigChangedEvent):
        try:
            helper_that_succeeds()
            helper_that_blocks()
        except BlockedExceptionError as e:
            logging.info(f"Charm helper raised BlockedExceptionError with message: '{e.msg}")
            self.unit.status = ops.BlockedStatus(e.msg)
        # ...and add more except blocks for other types


def helper_that_blocks():
    raise BlockedExceptionError("Blocked because X is unset")

Where the on_config_changed() event handler executes the helpers and catches the special exceptions so it can set the charm's status. This lets the helper express an opinion and provide a useful message, and the charm can accept or ignore this.

A downside to the above approach is that now every charm needs the except BlockedException as e; except WaitingException as e; .... But we can simplify this as Sunbeam does with a context manager:

@contextmanager
def status_handling_guard(section: str):
    """Handle Errors that request setting a charm status."""
    try:
        yield
    except BlockedExceptionError as e:
        logging.info(f"Early exit from {section} with message: '{e.msg}")
        self.unit.status = ops.BlockedStatus(e.msg)
    except MaintenanceExceptionError as e:
        logging.info(f"Early exit from {section} with message: '{e.msg}")
        self.unit.status = ops.MaintenanceStatus(e.msg)
    except WaitingExceptionError as e:
        logging.info(f"Early exit from {section} with message: '{e.msg}")
        self.unit.status = ops.WaitingStatus(e.msg)
    except Exception as e:
        # Something else went wrong
        # could do something here to add value
        raise


class DemoStatusRaisedInHelpersCharm(ops.CharmBase):
    """Charm for demoing status raising in helpers."""

    def __init__(self, *args):
        super().__init__(*args)

        self.framework.observe(self.on.config_changed, self.on_config_changed)

    def on_config_changed(self, _event: ops.ConfigChangedEvent):
        with status_handling_guard("Config parsing"):
            helper_that_succeeds()
            helper_that_blocks()

        # And even have multiple sections, if that's helpful...
        with status_handling_guard("Reconcile"):
            helper_that_waits()

This lets us make a single reusable status_handling_guard. Its easy to reuse, can have bulletproof testing in one place, and removes the chance that each charm is inconsistent about logging/forgets to catch a status, etc

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment