diff --git a/sentry_kube/cli/__init__.py b/sentry_kube/cli/__init__.py index 7bc8600..ecc0789 100644 --- a/sentry_kube/cli/__init__.py +++ b/sentry_kube/cli/__init__.py @@ -168,6 +168,13 @@ def main(ctx, *, cluster_name, root, customer, quiet, no_sentry, stage): context_name = cluster.services_data["context"] service_class = customer_config.k8s_config.service_class + if ctx.invoked_subcommand == "argocd-sync": + # ArgoCD runs on the region's default cluster and manages the others as spokes. + context_name = load_cluster_configuration( + customer_config.k8s_config, + customer_config.k8s_config.cluster_name or "default", + ).services_data["context"] + if ctx.invoked_subcommand == "ssh": ctx.obj = CliContext( context_name=context_name, diff --git a/sentry_kube/cli/argocd_sync.py b/sentry_kube/cli/argocd_sync.py new file mode 100644 index 0000000..8b13272 --- /dev/null +++ b/sentry_kube/cli/argocd_sync.py @@ -0,0 +1,99 @@ +from typing import Any, Optional + +import click +from kubernetes import client +from kubernetes.client.rest import ApiException + +from libsentrykube.cluster import Cluster +from libsentrykube.utils import die, kube_get_client + +__all__ = ("argocd_sync",) + +APPLICATION = { + "group": "argoproj.io", + "version": "v1alpha1", + "namespace": "argocd", + "plural": "applications", +} + + +def argocd_app_name(service: str, cluster: Cluster, hub_context: str) -> str: + context = cluster.services_data["context"] + if context == hub_context: + return service + tail = context.rsplit("_", 1)[-1] if context.startswith("gke_") else None + return ( + f"{cluster.services_data.get('cluster_name') or tail or cluster.name}-{service}" + ) + + +def autosync_enabled(automated: Optional[dict[str, Any]]) -> bool: + return automated is not None and automated.get("enabled", True) + + +def sync_patch( + automated: Optional[dict[str, Any]], + autosync: Optional[bool], + self_heal: Optional[bool], +) -> dict[str, Any]: + patch: dict[str, Any] = { + "enabled": autosync_enabled(automated) if autosync is None else autosync + } + if self_heal is not None: + patch["selfHeal"] = self_heal + return {"spec": {"syncPolicy": {"automated": patch}}} + + +def describe(app: dict[str, Any]) -> str: + automated = (app["spec"].get("syncPolicy") or {}).get("automated") + autosync = "on" if autosync_enabled(automated) else "off" + self_heal = "on" if (automated or {}).get("selfHeal") else "off" + return f"{app['metadata']['name']}: autosync={autosync} self-heal={self_heal}" + + +def on_off(ctx, param, value: Optional[str]) -> Optional[bool]: + return None if value is None else value == "on" + + +@click.command() +@click.argument("service") +@click.option( + "--autosync", + type=click.Choice(["on", "off"]), + callback=on_off, + help="Toggle automated sync.", +) +@click.option( + "--self-heal", + type=click.Choice(["on", "off"]), + callback=on_off, + help="Toggle self-heal.", +) +@click.pass_context +def argocd_sync( + ctx, service: str, autosync: Optional[bool], self_heal: Optional[bool] +) -> None: + """ + Show or toggle ArgoCD autosync and self-heal for a service. + + Targets the Application for SERVICE on the cluster selected with -c. + With no toggles, prints the current state. + """ + api = client.CustomObjectsApi(kube_get_client()) + name = argocd_app_name(service, ctx.obj.cluster, ctx.obj.context_name) + + try: + app = api.get_namespaced_custom_object(name=name, **APPLICATION) + if autosync is not None or self_heal is not None: + automated = (app["spec"].get("syncPolicy") or {}).get("automated") + app = api.patch_namespaced_custom_object( + name=name, + body=sync_patch(automated, autosync, self_heal), + **APPLICATION, + ) + except ApiException as e: + if e.status == 404: + die(f"ArgoCD Application '{name}' not found.") + raise + + click.echo(describe(app)) diff --git a/sentry_kube/tests/cli/test_argocd_sync.py b/sentry_kube/tests/cli/test_argocd_sync.py new file mode 100644 index 0000000..77530f7 --- /dev/null +++ b/sentry_kube/tests/cli/test_argocd_sync.py @@ -0,0 +1,108 @@ +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from click.testing import CliRunner +from kubernetes.client.rest import ApiException + +from sentry_kube.cli.argocd_sync import argocd_app_name, argocd_sync, sync_patch + +HUB = "gke_sentry-eu-west3_europe-west3_primary" + + +def cluster(name, **services_data): + return SimpleNamespace(name=name, services_data=services_data) + + +def app(name, automated=None): + sync_policy = {} if automated is None else {"automated": automated} + return {"metadata": {"name": name}, "spec": {"syncPolicy": sync_policy}} + + +@pytest.mark.parametrize( + "c, expected", + [ + (cluster("default", context=HUB), "relay"), + (cluster("seer-sandbox", context="gke_p_r_seer-sandbox"), "seer-sandbox-relay"), + ( + cluster("pop-au", context="gke_p_r_x", cluster_name="pop-au-1"), + "pop-au-1-relay", + ), + (cluster("local", context="colima"), "local-relay"), + ], +) +def test_argocd_app_name(c, expected): + assert argocd_app_name("relay", c, HUB) == expected + + +@pytest.mark.parametrize( + "automated, autosync, self_heal, expected", + [ + (None, True, None, {"enabled": True}), + ({"prune": True, "selfHeal": True}, False, None, {"enabled": False}), + (None, None, True, {"enabled": False, "selfHeal": True}), + ({}, None, False, {"enabled": True, "selfHeal": False}), + ({"enabled": False}, True, True, {"enabled": True, "selfHeal": True}), + ], +) +def test_sync_patch(automated, autosync, self_heal, expected): + assert sync_patch(automated, autosync, self_heal) == { + "spec": {"syncPolicy": {"automated": expected}} + } + + +@pytest.fixture +def api(): + with ( + patch("sentry_kube.cli.argocd_sync.kube_get_client"), + patch("sentry_kube.cli.argocd_sync.client.CustomObjectsApi") as mock, + ): + yield mock.return_value + + +def invoke(*args): + obj = SimpleNamespace(cluster=cluster("default", context=HUB), context_name=HUB) + return CliRunner().invoke(argocd_sync, ["relay", *args], obj=obj) + + +def test_status_does_not_patch(api): + api.get_namespaced_custom_object.return_value = app( + "relay", {"prune": True, "selfHeal": True} + ) + + result = invoke() + + assert result.exit_code == 0 + assert result.output == "relay: autosync=on self-heal=on\n" + api.patch_namespaced_custom_object.assert_not_called() + + +def test_toggle_patches_and_prints_result(api): + api.get_namespaced_custom_object.return_value = app("relay", {"prune": True}) + api.patch_namespaced_custom_object.return_value = app( + "relay", {"enabled": False, "prune": True} + ) + + result = invoke("--autosync", "off") + + assert result.exit_code == 0 + assert result.output == "relay: autosync=off self-heal=off\n" + assert api.patch_namespaced_custom_object.call_args.kwargs["body"] == { + "spec": {"syncPolicy": {"automated": {"enabled": False}}} + } + + +def test_missing_app(api): + api.get_namespaced_custom_object.side_effect = ApiException(status=404) + + result = invoke() + + assert result.exit_code != 0 + assert "ArgoCD Application 'relay' not found." in result.output + + +def test_rejects_invalid_toggle(api): + result = invoke("--self-heal", "maybe") + + assert result.exit_code == 2 + api.get_namespaced_custom_object.assert_not_called()