[plugin gcp][feat] Add GKE clusters resource (#644)

* [plugin gcp] Add GKE clusters

plugins/gcp/resoto_plugin_gcp/collector.py

@@ -9,6 +9,7 @@
 from resotolib.utils import except_log_and_pass
 from prometheus_client import Summary
 from .resources import (
+    GCPGKECluster,
     GCPProject,
     GCPQuota,
     GCPRegion,
@@ -222,6 +223,10 @@
     "resoto_plugin_gcp_collect_instance_templates_seconds",
     "Time it took the collect_instance_templates() method",
 )
+metrics_collect_gke_clusters = Summary(
+    "resoto_plugin_gcp_collect_gke_clusters_seconds",
+    "Time it took the collect_gke_clusters() method",
+)
 
 
 class GCPProjectCollector:
@@ -292,6 +297,7 @@ def __init__(self, project: GCPProject) -> None:
             "buckets": self.collect_buckets,
             "databases": self.collect_databases,
             "instance_templates": self.collect_instance_templates,
+            "gke_clusters": self.collect_gke_clusters,
         }
         # Region collectors collect resources in a single region.
         # They are being passed the GCPRegion resource object as `region` arg.
@@ -547,6 +553,7 @@ def collect_something(
         successors: List = None,
         predecessors: List = None,
         client_kwargs: Dict = None,
+        client_nested_callables: List[str] = None,
         resource_kwargs: Dict = None,
         paginate_subitems_name: str = None,
         post_process: Callable = None,
@@ -587,6 +594,8 @@ def collect_something(
             paginate_subitems_name = client_method_name
         if client_kwargs is None:
             client_kwargs = {}
+        if client_nested_callables is None:
+            client_nested_callables = []
         if resource_kwargs is None:
             resource_kwargs = {}
         if successors is None:
@@ -595,7 +604,9 @@ def collect_something(
             predecessors = []
         parent_map = {True: predecessors, False: successors}
 
-        if "project" in default_resource_args:
+        # For APIs that take a parent (`projects/*/locations/*`) parameter,
+        # setting the project is not expected.
+        if "parent" not in resource_kwargs and "project" in default_resource_args:
             resource_kwargs["project"] = self.project.id
 
         client = gcp_client(
@@ -604,6 +615,13 @@ def collect_something(
             credentials=self.credentials,
             **client_kwargs,
         )
+
+        # Some more recent client implementations have nested callables before
+        # the actual client_method_name method,
+        # e.g. discovery.build('container', 'v1').projects().locations().clusters()
+        for client_nested_callable in client_nested_callables:
+            client = getattr(client, client_nested_callable)()
+
         gcp_resource = getattr(client, client_method_name)
         if not callable(gcp_resource):
             raise RuntimeError(f"No method {client_method_name} on client {client}")
@@ -1513,3 +1531,19 @@ def collect_instance_templates(self):
             },
             predecessors=["__machine_type"],
         )
+
+    @metrics_collect_gke_clusters.time()
+    def collect_gke_clusters(self):
+        self.collect_something(
+            resource_class=GCPGKECluster,
+            resource_kwargs={"parent": f"projects/{self.project.id}/locations/-"},
+            client_nested_callables=["projects", "locations"],
+            paginate_items_name="clusters",
+            attr_map={
+                "ctime": lambda r: iso2datetime(r.get("createTime")),
+                "initial_cluster_version": "initialClusterVersion",
+                "current_master_version": "currentMasterVersion",
+                "cluster_status": "status",
+                "current_node_count": "currentNodeCount",
+            },
+        )

plugins/gcp/resoto_plugin_gcp/resources.py

@@ -637,3 +637,46 @@ def __post_init__(self) -> None:
                         break
             if cost > -1:
                 self.usage_unit_nanos = cost
+
+
+@dataclass(eq=False)
+class GCPGKECluster(GCPResource, BaseResource):
+    kind: ClassVar[str] = "gcp_gke_cluster"
+    api_identifier: ClassVar[str] = "cluster"
+    client: ClassVar[str] = "container"
+    api_version: ClassVar[str] = "v1"
+
+    initial_cluster_version: Optional[str] = None
+    current_master_version: Optional[str] = None
+    current_node_count: Optional[int] = None
+    cluster_status: Optional[str] = ""
+
+    cluster_status_map: ClassVar[Dict[str, InstanceStatus]] = {
+        "PROVISIONING": InstanceStatus.BUSY,
+        "STAGING": InstanceStatus.BUSY,
+        "RUNNING": InstanceStatus.RUNNING,
+        "STOPPING": InstanceStatus.BUSY,
+        "SUSPENDING": InstanceStatus.BUSY,
+        "SUSPENDED": InstanceStatus.STOPPED,
+        "REPAIRING": InstanceStatus.BUSY,
+        "TERMINATED": InstanceStatus.TERMINATED,
+        "busy": InstanceStatus.BUSY,
+        "running": InstanceStatus.RUNNING,
+        "stopped": InstanceStatus.STOPPED,
+        "terminated": InstanceStatus.TERMINATED,
+    }
+
+    def _cluster_status_setter(self, value: str) -> None:
+        self._cluster_status = self.cluster_status_map.get(
+            value, InstanceStatus.UNKNOWN
+        )
+        if self._cluster_status == InstanceStatus.TERMINATED:
+            self._cleaned = True
+
+    def _cluster_status_getter(self) -> str:
+        return self._cluster_status.value
+
+
+GCPGKECluster.cluster_status = property(
+    GCPGKECluster._cluster_status_getter, GCPGKECluster._cluster_status_setter
+)