feat: add collector plugin connector for v1 and v2

Signed-off-by: ImMin5 <[email protected]>

src/spaceone/inventory_v2/conf/collector_conf.py

@@ -7,16 +7,15 @@
 ######################################################################
 
 RESOURCE_MAP = {
-    "inventory.CloudService": ("CloudServiceService", "CloudServiceManager"),
-    "inventory.CloudServiceType": (
-        "CloudServiceTypeService",
-        "CloudServiceTypeManager",
+    "inventory.Asset": ("AssetService", "AssetManager"),
+    "inventory.AssetType": (
+        "AssetTypeService",
+        "AssetTypeManager",
     ),
     "inventory.Region": ("RegionService", "RegionManager"),
     "inventory.ErrorResource": ("CollectorService", "CollectingManager"),
 }
 
-
 OP_MAP = {"=": "eq", ">=": "gte", "<=": "lte", ">": "gt", "<": "lt", "!=": "not"}
 
 DB_QUEUE_NAME = "db_q"

src/spaceone/inventory_v2/connector/__init__.py

@@ -0,0 +1,9 @@
+from spaceone.inventory_v2.connector.collector_plugin_connector import (
+    BaseCollectorPluginConnector,
+)
+from spaceone.inventory_v2.connector.collector_plugin_connector.collector_plugin_v1_connector import (
+    CollectorPluginV1Connector,
+)
+from spaceone.inventory_v2.connector.collector_plugin_connector.collector_plugin_v2_connector import (
+    CollectorPluginV2Connector,
+)

src/spaceone/inventory_v2/connector/collector_plugin_connector/__init__.py

@@ -0,0 +1,27 @@
+from spaceone.core.connector import BaseConnector
+from spaceone.core.connector.space_connector import SpaceConnector
+
+
+class BaseCollectorPluginConnector(BaseConnector):
+    collector_version = None
+
+    @classmethod
+    def init_plugin(cls, endpoint: str, options: dict) -> dict:
+        plugin_connector = SpaceConnector(endpoint=endpoint, token="NO_TOKEN")
+        return plugin_connector.dispatch("Collector.init", {"options": options})
+
+    def verify_plugin(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    def get_tasks(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    def collect(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    @classmethod
+    def get_connector_by_collector_version(cls, collector_version: str):
+        for subclass in cls.__subclasses__():
+            if subclass.collector_version == collector_version:
+                return subclass()
+        raise Exception(f"Not found collector plugin connector: {collector_version}")

src/spaceone/inventory_v2/connector/collector_plugin_connector/collector_plugin_v1_connector.py

@@ -0,0 +1,114 @@
+import logging
+
+from typing import Generator
+
+from spaceone.core.connector.space_connector import SpaceConnector
+
+from spaceone.inventory_v2.connector.collector_plugin_connector import (
+    BaseCollectorPluginConnector,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class CollectorPluginV1Connector(BaseCollectorPluginConnector):
+    collector_version = "v1"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    def verify_plugin(self, endpoint: str, secret_data: dict, options: dict) -> dict:
+        plugin_connector = SpaceConnector(endpoint=endpoint, token="NO_TOKEN")
+        params = {"options": options, "secret_data": secret_data}
+        return plugin_connector.dispatch("Collect.collect", params)
+
+    def get_tasks(self, endpoint: str, options: dict, secret_data: dict) -> dict:
+        try:
+            plugin_connector = SpaceConnector(endpoint=endpoint, token="NO_TOKEN")
+            params = {"options": options, "secret_data": secret_data}
+            return plugin_connector.dispatch("Job.get_tasks", params)
+        except Exception as e:
+            return {"tasks": []}
+
+    def collect(
+        self,
+        endpoint: str,
+        options: dict,
+        secret_data: dict,
+        task_options: dict = None,
+    ) -> Generator[dict, None, None]:
+        plugin_connector = SpaceConnector(endpoint=endpoint, token="NO_TOKEN")
+
+        params = {"options": options, "secret_data": secret_data, "filter": {}}
+
+        if task_options:
+            params["task_options"] = task_options
+
+        for resource_data in plugin_connector.dispatch("Collector.collect", params):
+            yield self._convert_resource_data(resource_data)
+
+    @staticmethod
+    def _convert_resource_data(resource_data: dict) -> dict:
+        _LOGGER.debug(
+            f"[_convert_resource_data] before convert resource_data: {resource_data}"
+        )
+
+        resource_type = resource_data.get("resource_type")
+
+        if resource_type in ["inventory.CloudService", "inventory.CloudServiceType"]:
+            if resource_type == "inventory.CloudService":
+                resource_data["resource_type"] = "inventory.Asset"
+                resource_data["resource"]["asset_type_id"] = resource_data[
+                    "resource"
+                ].get("asset_type")
+                # resource_data["resource"]["asset_group_id"] = resource_data[
+                #     "resource"
+                # ].get("cloud_service_group")
+            elif resource_type == "inventory.CloudServiceType":
+                resource_data["resource_type"] = "inventory.AssetType"
+
+            resource_type = resource_data.get("resource_type")
+
+            # convert match rule
+            for rule_values in resource_data.get("match_rules", {}).values():
+                for index, rule_value in enumerate(rule_values):
+                    if rule_value == "cloud_service_id":
+                        rule_values[index] = "asset_id"
+                    elif rule_value == "cloud_service_type":
+                        rule_values[index] = "asset_type_id"
+                    elif rule_value == "cloud_service_group":
+                        del rule_values[index]
+                    elif rule_value == "reference.resource_id":
+                        rule_values[index] = "resource_id"
+                    elif rule_value == "group":
+                        rule_values[index] = "asset_group_id"
+
+            if _resource := resource_data.get("resource"):
+                _resource_v1 = {}
+                if "instance_size" in _resource:
+                    _resource_v1["instance_size"] = _resource.pop("instance_size")
+                if "instance_type" in _resource:
+                    _resource_v1["instance_type"] = _resource.pop("instance_type")
+                del _resource["metadata"]
+
+                if resource_type == "inventory.Asset":
+                    asset_type_id = f"{_resource['provider']}.{_resource['cloud_service_group']}.{_resource['cloud_service_type']}"
+                    _resource["asset_type_id"] = asset_type_id
+                    resource_data["asset_type_id"] = asset_type_id
+                elif resource_type == "inventory.AssetType":
+
+                    asset_type_id = f"at-{_resource['provider']}-{_resource['group']}-{_resource['name']}"
+                    asset_groups = [
+                        f"ag-{_resource['provider']}-{_resource['group']}",
+                        f"ag-{_resource['provider']}",
+                    ]
+                    _resource["asset_type_id"] = asset_type_id
+                    resource_data["asset_type_id"] = asset_type_id
+                    resource_data["asset_groups"] = asset_groups
+                    resource_data["icon"] = _resource.get("tags", {}).get("icon", "")
+
+                resource_data["resource"]["v1"] = _resource_v1
+
+        _LOGGER.debug(f"[_convert_resource_data] resource_data: {resource_data}")
+
+        return resource_data

src/spaceone/inventory_v2/connector/collector_plugin_connector/collector_plugin_v2_connector.py

@@ -0,0 +1,17 @@
+from spaceone.inventory_v2.connector import BaseCollectorPluginConnector
+
+
+class CollectorPluginV2Connector(BaseCollectorPluginConnector):
+    collector_version = "v1"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    def verify_plugin(self, endpoint, secret_data, options):
+        pass
+
+    def get_tasks(self, params):
+        pass
+
+    def collect(self, params):
+        pass

src/spaceone/inventory_v2/error/__init__.py

@@ -1,2 +1,3 @@
 from spaceone.inventory_v2.error.region import *
 from spaceone.inventory_v2.error.collector import *
+from spaceone.inventory_v2.error.metric import *

src/spaceone/inventory_v2/error/metric.py

@@ -0,0 +1,17 @@
+from spaceone.core.error import *
+
+
+class ERROR_NOT_SUPPORT_RESOURCE_TYPE(ERROR_INVALID_ARGUMENT):
+    _message = "Not support resource type. (resource_type = {resource_type})"
+
+
+class ERROR_INVALID_DATE_RANGE(ERROR_INVALID_ARGUMENT):
+    _message = "{reason}"
+
+
+class ERROR_METRIC_QUERY_RUN_FAILED(ERROR_BASE):
+    _message = "Metric query run failed. (metric_id = {metric_id})"
+
+
+class ERROR_WRONG_QUERY_OPTIONS(ERROR_INVALID_ARGUMENT):
+    _message = "Wrong query options. (query_options = {query_options})"

src/spaceone/inventory_v2/interface/grpc/metric.py

@@ -1,6 +1,6 @@
 from spaceone.core.pygrpc import BaseAPI
 from spaceone.api.inventory_v2.v1 import metric_pb2, metric_pb2_grpc
-from spaceone.inventory.service.metric_service import MetricService
+from spaceone.inventory_v2.service.metric_service import MetricService
 
 
 class Metric(BaseAPI, metric_pb2_grpc.MetricServicer):

src/spaceone/inventory_v2/manager/__init__.py

@@ -1 +1,5 @@
+from spaceone.inventory_v2.manager.asset_manager import AssetManager
+from spaceone.inventory_v2.manager.asset_type_manager import AssetTypeManager
+from spaceone.inventory_v2.lib.resource_manager import ResourceManager
 from spaceone.inventory_v2.manager.region_manager import RegionManager
+from spaceone.inventory_v2.manager.collecting_manager import CollectingManager