sentry-kube: argocd-sync command for autosync and self-heal

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,

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))

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()