Add cluster label management feat (#111)

* add label management

* Fix CI issues

* fix lint

* fix Ci;s

---------

Co-authored-by: Rishi Mondal <[email protected]>

Author: Per0x1de-1337

src/shared/functions/__init__.py

@@ -2,10 +2,11 @@
 
 from src.shared.base_functions import function_registry
 from src.shared.functions.binding_policy_management import BindingPolicyManagement
+from src.shared.functions.check_cluster_upgrades import CheckClusterUpgradesFunction
+from src.shared.functions.cluster_label_management import ClusterLabelManagement
 from src.shared.functions.deploy_to import DeployToFunction
 from src.shared.functions.describe_resource import DescribeResourceFunction
 from src.shared.functions.edit_resource import EditResourceFunction
-from src.shared.functions.get_cluster_labels import GetClusterLabelsFunction
 from src.shared.functions.gvrc_discovery import GVRCDiscoveryFunction
 from src.shared.functions.helm.list import HelmListFunction
 from src.shared.functions.helm.repo import HelmRepoFunction
@@ -15,7 +16,6 @@
 from src.shared.functions.multicluster_create import MultiClusterCreateFunction
 from src.shared.functions.multicluster_logs import MultiClusterLogsFunction
 from src.shared.functions.namespace_utils import NamespaceUtilsFunction
-from src.shared.functions.check_cluster_upgrades import CheckClusterUpgradesFunction
 
 
 def initialize_functions():
@@ -44,7 +44,7 @@ def initialize_functions():
     function_registry.register(HelmDeployFunction())
 
     # Register cluster labels helper function
-    function_registry.register(GetClusterLabelsFunction())
+    function_registry.register(ClusterLabelManagement())
 
     # Register GVRC and namespace utilities
     function_registry.register(GVRCDiscoveryFunction())

src/shared/functions/cluster_label_management.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Dict, List
+
+from src.shared.base_functions import BaseFunction
+
+
+class ClusterLabelManagement(BaseFunction):
+    """
+    Add / update labels on an Open-Cluster-Management ManagedCluster.
+
+    Typical call:
+        • cluster_name  – managedcluster resource name
+        • labels        – dict of key → value
+        • kube_context  – context of the OCM Hub / ITS (default: its1)
+    """
+
+    def __init__(self) -> None:
+        super().__init__(
+            name="cluster_label_management",
+            description="Add or update labels on a ManagedCluster object."
+        )
+
+    # ────────────────────────── public entry ──────────────────────────
+    async def execute(
+        self,
+        cluster_name: str,
+        labels: Dict[str, str] | None = None,
+        remove_labels: List[str] | None = None,
+        kube_context: str = "its1",        # default is ITS / OCM hub
+        kubeconfig: str = "",
+        **_: Any,
+    ) -> Dict[str, Any]:
+        if not cluster_name:
+            return {"status": "error", "error": "cluster_name is required"}
+        if not labels and not remove_labels:
+            return {"status": "error", "error": "labels or remove_labels must be provided"}
+
+        label_args: List[str] = []
+        if labels:
+            label_args += [f"{k}={v}" for k, v in labels.items()]
+        if remove_labels:
+            label_args += [f"{key}-" for key in remove_labels]
+        cmd = [
+            "kubectl", "--context", kube_context,
+            "label", "managedcluster", cluster_name,
+            *label_args, "--overwrite",
+        ]
+        if kubeconfig:
+            cmd += ["--kubeconfig", kubeconfig]
+
+        proc = await asyncio.create_subprocess_exec(
+            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+        )
+        stdout, stderr = await proc.communicate()
+
+        if proc.returncode != 0:
+            # Surface full stderr so the LLM can show the exact cause
+            return {
+                "status": "error",
+                "stderr": stderr.decode().strip(),
+                "cmd": " ".join(cmd),
+            }
+
+        return {
+            "status": "success",
+            "stdout": stdout.decode().strip(),
+            "cmd": " ".join(cmd),
+        }
+
+    # ────────────────────────── JSON schema ──────────────────────────
+    def get_schema(self) -> Dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "cluster_name": {
+                    "type": "string",
+                    "description": "Name of the ManagedCluster resource",
+                },
+                "labels": {
+                    "type": "object",
+                    "description": "Dictionary of label key/value pairs to add/update",
+                },
+                "remove_labels": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "List of label keys to delete",
+                },
+                "kube_context": {
+                    "type": "string",
+                    "description": "kubectl context of the OCM hub",
+                    "default": "its1",
+                },
+                "kubeconfig": {
+                    "type": "string",
+                    "description": "Path to alternate kubeconfig (optional)",
+                },
+            },
+            "required": ["cluster_name"],
+        }