[feat] Introduce managed_kubernetes_cluster and clean up collect (#1939)

plugins/aws/resoto_plugin_aws/resource/ecs.py

@@ -1784,7 +1784,7 @@ class AwsEcsContainerInstance(EcsTaggable, AwsResource):
         "arn": S("containerInstanceArn"),
         "ec2_instance_id": S("ec2InstanceId"),
         "capacity_provider_name": S("capacityProviderName"),
-        "version": S("version"),
+        "version": S("version") >> F(str),
         "version_info": S("versionInfo") >> Bend(AwsEcsVersionInfo.mapping),
         "remaining_resources": S("remainingResources", default=[]) >> ForallBend(AwsEcsResource.mapping),
         "registered_resources": S("registeredResources", default=[]) >> ForallBend(AwsEcsResource.mapping),
@@ -1800,7 +1800,7 @@ class AwsEcsContainerInstance(EcsTaggable, AwsResource):
     }
     ec2_instance_id: Optional[str] = field(default=None)
     capacity_provider_name: Optional[str] = field(default=None)
-    version: Optional[int] = field(default=None)
+    version: Optional[str] = field(default=None)
     version_info: Optional[AwsEcsVersionInfo] = field(default=None)
     remaining_resources: List[AwsEcsResource] = field(factory=list)
     registered_resources: List[AwsEcsResource] = field(factory=list)

plugins/aws/resoto_plugin_aws/resource/eks.py

@@ -6,7 +6,7 @@
 from resoto_plugin_aws.resource.autoscaling import AwsAutoScalingGroup
 from resoto_plugin_aws.resource.base import AwsResource, GraphBuilder, AwsApiSpec
 from resoto_plugin_aws.resource.iam import AwsIamRole
-from resotolib.baseresources import ModelReference
+from resotolib.baseresources import ModelReference, BaseManagedKubernetesClusterProvider
 from resotolib.graph import Graph
 from resotolib.json_bender import Bender, S, Bend, ForallBend
 from resotolib.types import Json
@@ -399,7 +399,7 @@ class AwsEksConnectorConfig:
 
 
 @define(eq=False, slots=False)
-class AwsEksCluster(EKSTaggable, AwsResource):
+class AwsEksCluster(EKSTaggable, BaseManagedKubernetesClusterProvider, AwsResource):
     kind: ClassVar[str] = "aws_eks_cluster"
     kind_display: ClassVar[str] = "AWS EKS Cluster"
     aws_metadata: ClassVar[Dict[str, Any]] = {"provider_link_tpl": "https://{region_id}.console.aws.amazon.com/eks/home?region={region}#/clusters/{name}", "arn_tpl": "arn:{partition}:eks:{region}:{account}:cluster/{name}"}  # fmt: skip
@@ -421,8 +421,8 @@ class AwsEksCluster(EKSTaggable, AwsResource):
         "name": S("name"),
         "arn": S("arn"),
         "ctime": S("createdAt"),
-        "cluster_version": S("version"),
-        "cluster_endpoint": S("endpoint"),
+        "version": S("version"),
+        "endpoint": S("endpoint"),
         "cluster_role_arn": S("roleArn"),
         "cluster_resources_vpc_config": S("resourcesVpcConfig") >> Bend(AwsEksVpcConfigResponse.mapping),
         "cluster_kubernetes_network_config": S("kubernetesNetworkConfig")
@@ -436,8 +436,6 @@ class AwsEksCluster(EKSTaggable, AwsResource):
         "cluster_encryption_config": S("encryptionConfig", default=[]) >> ForallBend(AwsEksEncryptionConfig.mapping),
         "cluster_connector_config": S("connectorConfig") >> Bend(AwsEksConnectorConfig.mapping),
     }
-    cluster_version: Optional[str] = field(default=None)
-    cluster_endpoint: Optional[str] = field(default=None)
     cluster_role_arn: Optional[str] = field(default=None)
     cluster_resources_vpc_config: Optional[AwsEksVpcConfigResponse] = field(default=None)
     cluster_kubernetes_network_config: Optional[AwsEksKubernetesNetworkConfigResponse] = field(default=None)

plugins/digitalocean/resoto_plugin_digitalocean/client.py

@@ -1,14 +1,14 @@
 import logging
-from attrs import define
+from datetime import datetime
 from functools import lru_cache
 from typing import List, Any, Optional, Union, TypeVar, Callable, Mapping
+from urllib.parse import urljoin, urlencode
 
 import boto3
 import requests
+from attrs import define
 from botocore.exceptions import EndpointConnectionError, HTTPClientError
 from retrying import retry as retry_decorator
-from urllib.parse import urljoin, urlencode
-from datetime import datetime
 
 from resoto_plugin_digitalocean.utils import RetryableHttpError
 from resoto_plugin_digitalocean.utils import retry_on_error

plugins/digitalocean/resoto_plugin_digitalocean/collector.py

@@ -1,4 +1,5 @@
 import logging
+
 import math
 from pprint import pformat
 from typing import Tuple, Type, List, Dict, Callable, Any, Optional, cast, DefaultDict
@@ -833,7 +834,7 @@ def collect_k8s_clusters(self) -> None:
             attr_map={
                 "id": "id",
                 "urn": lambda c: kubernetes_id(c["id"]),
-                "k8s_version": "version",
+                "version": "version",
                 "k8s_cluster_subnet": "cluster_subnet",
                 "k8s_service_subnet": "service_subnet",
                 "ipv4_address": "ipv4",

plugins/digitalocean/resoto_plugin_digitalocean/resources.py

@@ -1,4 +1,6 @@
 import logging
+from dataclasses import field
+
 from attrs import define
 from typing import ClassVar, Dict, List, Optional, Tuple, Any
 
@@ -25,6 +27,7 @@
     BaseDNSRecord,
     ModelReference,
     PhantomBaseResource,
+    BaseManagedKubernetesClusterProvider,
 )
 from resotolib.graph import Graph
 import time
@@ -285,7 +288,7 @@ class DigitalOceanDropletNeighborhood(DigitalOceanResource, PhantomBaseResource)
 
 
 @define(eq=False, slots=False)
-class DigitalOceanKubernetesCluster(DigitalOceanResource, BaseResource):
+class DigitalOceanKubernetesCluster(DigitalOceanResource, BaseManagedKubernetesClusterProvider):
     """DigitalOcean Kubernetes Cluster"""
 
     kind: ClassVar[str] = "digitalocean_kubernetes_cluster"
@@ -301,13 +304,11 @@ class DigitalOceanKubernetesCluster(DigitalOceanResource, BaseResource):
         }
     }
 
-    k8s_version: Optional[str] = None
     k8s_cluster_subnet: Optional[str] = None
     k8s_service_subnet: Optional[str] = None
     ipv4_address: Optional[str] = None
-    endpoint: Optional[str] = None
     auto_upgrade_enabled: Optional[bool] = None
-    cluster_status: Optional[str] = None
+    cluster_status: Optional[str] = field(default=None, metadata=dict(ignore_history=True))
     surge_upgrade_enabled: Optional[bool] = None
     registry_enabled: Optional[bool] = None
     ha_enabled: Optional[bool] = None

plugins/digitalocean/test/test_collector.py

@@ -330,7 +330,7 @@ def test_collect_k8s_clusters() -> None:
     cluster: DigitalOceanKubernetesCluster = graph.search_first("urn", "do:kubernetes:e1c48631-b382-4001-2168-c47c54795a26")  # type: ignore # noqa: E501
     assert cluster.urn == "do:kubernetes:e1c48631-b382-4001-2168-c47c54795a26"
     assert cluster.name == "k8s-1-22-7-do-0-fra1-test"
-    assert cluster.k8s_version == "1.22.7-do.0"
+    assert cluster.version == "1.22.7-do.0"
     assert cluster.region().urn == "do:region:fra1"  # type: ignore
     assert cluster.k8s_cluster_subnet == "10.244.0.0/16"
     assert cluster.k8s_service_subnet == "10.245.0.0/16"

plugins/gcp/resoto_plugin_gcp/resources/container.py

@@ -5,7 +5,7 @@
 
 from resoto_plugin_gcp.gcp_client import GcpApiSpec
 from resoto_plugin_gcp.resources.base import GcpResource, GcpDeprecationStatus, GraphBuilder
-from resotolib.baseresources import ModelReference
+from resotolib.baseresources import ModelReference, BaseManagedKubernetesClusterProvider
 from resotolib.json_bender import Bender, S, Bend, ForallBend, MapDict
 from resotolib.types import Json
 
@@ -1083,7 +1083,7 @@ class GcpContainerResourceUsageExportConfig:
 
 
 @define(eq=False, slots=False)
-class GcpContainerCluster(GcpResource):
+class GcpContainerCluster(BaseManagedKubernetesClusterProvider, GcpResource):
     kind: ClassVar[str] = "gcp_container_cluster"
     kind_display: ClassVar[str] = "GCP Container Cluster"
     kind_description: ClassVar[str] = (
@@ -1124,6 +1124,7 @@ class GcpContainerCluster(GcpResource):
         "cost_management_config": S("costManagementConfig", "enabled"),
         "create_time": S("createTime"),
         "current_master_version": S("currentMasterVersion"),
+        "version": S("currentMasterVersion"),
         "current_node_count": S("currentNodeCount"),
         "current_node_version": S("currentNodeVersion"),
         "database_encryption": S("databaseEncryption", default={}) >> Bend(GcpContainerDatabaseEncryption.mapping),
@@ -1192,7 +1193,6 @@ class GcpContainerCluster(GcpResource):
     default_max_pods_constraint: Optional[str] = field(default=None)
     enable_kubernetes_alpha: Optional[bool] = field(default=None)
     enable_tpu: Optional[bool] = field(default=None)
-    endpoint: Optional[str] = field(default=None)
     etag: Optional[str] = field(default=None)
     expire_time: Optional[datetime] = field(default=None)
     identity_service_config: Optional[bool] = field(default=None)

plugins/k8s/resoto_plugin_k8s/__init__.py

@@ -5,14 +5,14 @@
 from tempfile import TemporaryDirectory
 from typing import Dict, Any, Type, Optional
 
-import resotolib.logger
-import resotolib.proc
 from kubernetes.client import ApiException
 from kubernetes.client import Configuration
 
+import resotolib.logger
+import resotolib.proc
 from resoto_plugin_k8s.base import K8sApiClient, K8sClient
-from resoto_plugin_k8s.collector import KubernetesCollector
 from resoto_plugin_k8s.base import K8sConfig
+from resoto_plugin_k8s.collector import KubernetesCollector
 from resoto_plugin_k8s.deferred_edges import create_deferred_edges
 from resotolib.args import ArgumentParser, Namespace
 from resotolib.baseplugin import BaseCollectorPlugin

resotolib/resotolib/baseplugin.py

@@ -1,9 +1,9 @@
 import time
 from abc import ABC, abstractmethod
 from enum import Enum, auto
+from queue import Queue
 from threading import Thread, current_thread
 from typing import Dict, Optional, Any
-from queue import Queue
 
 from prometheus_client import Counter
 
@@ -19,7 +19,6 @@
 from resotolib.logger import log
 from resotolib.types import Json
 
-
 metrics_unhandled_plugin_exceptions = Counter(
     "resoto_unhandled_plugin_exceptions_total",
     "Unhandled plugin exceptions",

resotolib/resotolib/baseresources.py

@@ -1264,6 +1264,16 @@ class BaseOrganizationalUnit(BaseResource):
     metadata: ClassVar[Dict[str, Any]] = {"icon": "resource", "group": "misc"}
 
 
+@define(eq=False, slots=False)
+class BaseManagedKubernetesClusterProvider(BaseResource):
+    kind: ClassVar[str] = "managed_kubernetes_cluster_provider"
+    kind_display: ClassVar[str] = "Managed Kubernetes Cluster Provider"
+    kind_description: ClassVar[str] = "A managed kubernetes cluster provider."
+    metadata: ClassVar[Dict[str, Any]] = {"icon": "cluster", "group": "compute"}
+    version: Optional[str] = field(default=None, metadata={"description": "The kubernetes version"})
+    endpoint: Optional[str] = field(default=None, metadata={"description": "The kubernetes API endpoint"})
+
+
 @define(eq=False, slots=False)
 class UnknownCloud(BaseCloud):
     kind: ClassVar[str] = "unknown_cloud"

resotoworker/resotoworker/__main__.py

@@ -236,6 +236,12 @@ def core_actions_processor(
     if kind == "action":
         try:
             if message_type == "collect":
+                if len(collectors) == 0:
+                    log.error("No no collector plugins loaded - skipping collect")
+                    return None
+                if config.resotoworker.pool_size == 0:
+                    log.error("Zero workers configured - skipping collect")
+                    return None
                 collect_event.set()
                 start_time = time.time()
                 collector.collect_and_send(collectors, task_data=data)