-
Notifications
You must be signed in to change notification settings - Fork 18
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add observer that updates endpoints on relation (#58)
* Pass prebuilt charm to tests for local execution * Enable postgres process for full cluster restart test * Revert "Pass prebuilt charm to tests for local execution" This reverts commit cc7f199. * Add observer that updates endpoints on relation * Fixing logic * Bump up ops framework * Fix the logic to update the unit databag when it's clearead on another unit * Fix observer start * Fixes in the test * Fix part of the unit tests * Fix remaining unit tests * Minor adjustments in the code * Fix machine start * Add copyright notice * Add retries when dropping continuous_writes table * Terminating continuous_writes in test charm * Version bump for deps * Revert "Version bump for deps" This reverts commit 8e70a6f. * Version bump for libs * Switch test app charm to peer relation for storing cont writes PId * Code review tweaks * Fix TLS test * Comment logs * Fix get_member_ip * Add a call to relation endpoints update * Fix and revert changes * Change sleep time * Revert changes * Minor fixes * Revert lib changes * Revert requirements changes * Uncomment charms removal * Remove log calls * Add unit tests to the observer * Fix incorrect user deletion * Fix unit tests --------- Co-authored-by: Dragomir Penev <[email protected]>
- Loading branch information
1 parent
5b228ee
commit f4b372b
Showing
11 changed files
with
572 additions
and
220 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| # Copyright 2023 Canonical Ltd. | ||
| # See LICENSE file for licensing details. | ||
|
|
||
| """Cluster topology changes observer.""" | ||
|
|
||
| import logging | ||
| import os | ||
| import signal | ||
| import subprocess | ||
| import sys | ||
| from time import sleep | ||
|
|
||
| import requests | ||
| from ops.charm import CharmBase, CharmEvents | ||
| from ops.framework import EventBase, EventSource, Object | ||
| from ops.model import ActiveStatus | ||
|
|
||
| from constants import API_REQUEST_TIMEOUT, PATRONI_CLUSTER_STATUS_ENDPOINT | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| # File path for the spawned cluster topology observer process to write logs. | ||
| LOG_FILE_PATH = "/var/log/cluster_topology_observer.log" | ||
|
|
||
|
|
||
| class ClusterTopologyChangeEvent(EventBase): | ||
| """A custom event for cluster topology changes.""" | ||
|
|
||
|
|
||
| class ClusterTopologyChangeCharmEvents(CharmEvents): | ||
| """A CharmEvents extension for cluster topology changes. | ||
| Includes :class:`ClusterTopologyChangeEvent` in those that can be handled. | ||
| """ | ||
|
|
||
| cluster_topology_change = EventSource(ClusterTopologyChangeEvent) | ||
|
|
||
|
|
||
| class ClusterTopologyObserver(Object): | ||
| """Observes changing topology in the cluster. | ||
| Observed cluster topology changes cause :class"`ClusterTopologyChangeEvent` to be emitted. | ||
| """ | ||
|
|
||
| def __init__(self, charm: CharmBase): | ||
| """Constructor for ClusterTopologyObserver. | ||
| Args: | ||
| charm: the charm that is instantiating the library. | ||
| """ | ||
| super().__init__(charm, "cluster-topology-observer") | ||
|
|
||
| self._charm = charm | ||
|
|
||
| def start_observer(self): | ||
| """Start the cluster topology observer running in a new process.""" | ||
| if ( | ||
| not isinstance(self._charm.unit.status, ActiveStatus) | ||
| or self._charm._peers is None | ||
| or "observer-pid" in self._charm._peers.data[self._charm.unit] | ||
| ): | ||
| return | ||
|
|
||
| logging.info("Starting cluster topology observer process") | ||
|
|
||
| # We need to trick Juju into thinking that we are not running | ||
| # in a hook context, as Juju will disallow use of juju-run. | ||
| new_env = os.environ.copy() | ||
| if "JUJU_CONTEXT_ID" in new_env: | ||
| new_env.pop("JUJU_CONTEXT_ID") | ||
|
|
||
| pid = subprocess.Popen( | ||
| [ | ||
| "/usr/bin/python3", | ||
| "src/cluster_topology_observer.py", | ||
| self._charm._patroni._patroni_url, | ||
| f"{self._charm._patroni.verify}", | ||
| "/usr/bin/juju-run", | ||
| self._charm.unit.name, | ||
| self._charm.charm_dir, | ||
| ], | ||
| stdout=open(LOG_FILE_PATH, "a"), | ||
| stderr=subprocess.STDOUT, | ||
| env=new_env, | ||
| ).pid | ||
|
|
||
| self._charm._peers.data[self._charm.unit].update({"observer-pid": f"{pid}"}) | ||
| logging.info("Started cluster topology observer process with PID {}".format(pid)) | ||
|
|
||
| def stop_observer(self): | ||
| """Stop the running observer process if we have previously started it.""" | ||
| if ( | ||
| self._charm._peers is None | ||
| or "observer-pid" not in self._charm._peers.data[self._charm.unit] | ||
| ): | ||
| return | ||
|
|
||
| observer_pid = int(self._charm._peers.data[self._charm.unit].get("observer-pid")) | ||
|
|
||
| try: | ||
| os.kill(observer_pid, signal.SIGINT) | ||
| msg = "Stopped running cluster topology observer process with PID {}" | ||
| logging.info(msg.format(observer_pid)) | ||
| self._charm._peers.data[self._charm.unit].update({"observer-pid": ""}) | ||
| except OSError: | ||
| pass | ||
|
|
||
| @property | ||
| def unit_tag(self): | ||
| """Juju-style tag identifying the unit being run by this charm.""" | ||
| unit_num = self._charm.unit.name.split("/")[-1] | ||
| return "unit-{}-{}".format(self._charm.app.name, unit_num) | ||
|
|
||
|
|
||
| def dispatch(run_cmd, unit, charm_dir): | ||
| """Use the input juju-run command to dispatch a :class:`ClusterTopologyChangeEvent`.""" | ||
| dispatch_sub_cmd = "JUJU_DISPATCH_PATH=hooks/cluster_topology_change {}/dispatch" | ||
| subprocess.run([run_cmd, "-u", unit, dispatch_sub_cmd.format(charm_dir)]) | ||
|
|
||
|
|
||
| def main(): | ||
| """Main watch and dispatch loop. | ||
| Watch the Patroni API cluster info. When changes are detected, dispatch the change event. | ||
| """ | ||
| patroni_url, verify, run_cmd, unit, charm_dir = sys.argv[1:] | ||
|
|
||
| previous_cluster_topology = {} | ||
| while True: | ||
| cluster_status = requests.get( | ||
| f"{patroni_url}/{PATRONI_CLUSTER_STATUS_ENDPOINT}", | ||
| verify=verify, | ||
| timeout=API_REQUEST_TIMEOUT, | ||
| ) | ||
| current_cluster_topology = { | ||
| member["name"]: member["role"] for member in cluster_status.json()["members"] | ||
| } | ||
|
|
||
| # If it's the first time the cluster topology was retrieved, then store it and use | ||
| # it for subsequent checks. | ||
| if not previous_cluster_topology: | ||
| previous_cluster_topology = current_cluster_topology | ||
| # If the cluster topology changed, dispatch a charm event to handle this change. | ||
| elif current_cluster_topology != previous_cluster_topology: | ||
| previous_cluster_topology = current_cluster_topology | ||
| dispatch(run_cmd, unit, charm_dir) | ||
|
|
||
| # Wait some time before checking again for a cluster topology change. | ||
| sleep(30) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| main() |
Oops, something went wrong.