feat: add Asset interface

src/spaceone/inventory_v2/conf/collector_conf.py

@@ -0,0 +1,32 @@
+######################################################################
+#    ************ Very Important ************
+#
+# This is resource map for collector
+# If you add new service and manager for specific RESOURCE_TYPE,
+# add here for collector
+######################################################################
+
+RESOURCE_MAP = {
+    "inventory.CloudService": ("CloudServiceService", "CloudServiceManager"),
+    "inventory.CloudServiceType": (
+        "CloudServiceTypeService",
+        "CloudServiceTypeManager",
+    ),
+    "inventory.Region": ("RegionService", "RegionManager"),
+    "inventory.ErrorResource": ("CollectorService", "CollectingManager"),
+}
+
+
+OP_MAP = {"=": "eq", ">=": "gte", "<=": "lte", ">": "gt", "<": "lt", "!=": "not"}
+
+DB_QUEUE_NAME = "db_q"
+
+NOT_COUNT = 0
+CREATED = 1
+UPDATED = 2
+ERROR = 3
+
+JOB_TASK_STAT_EXPIRE_TIME = 3600  # 1 hour
+WATCHDOG_WAITING_TIME = 30  # wait 30 seconds, before watchdog works
+
+MAX_MESSAGE_LENGTH = 2000

src/spaceone/inventory_v2/conf/global_conf.py

@@ -47,13 +47,20 @@
             "identity": "grpc://identity:50051",
             "monitoring": "grpc://monitoring:50051",
             "file_manager": "grpc://file-manager:50051",
-            "secret": "grpc://secret:50051"
+            "secret": "grpc://secret:50051",
         },
     }
 }
 
+# Queue Settings
+QUEUES = {
+    "inventory_q": {
+        "backend": "spaceone.core.queue.redis_queue.RedisQueue",
+        "host": "redis",
+        "port": 6379,
+        "channel": "inventory_job",
+    },
+}
 # Scheduler Settings
-QUEUES = {}
 SCHEDULERS = {}
 WORKERS = {}
-TOKEN_INFO = {}

src/spaceone/inventory_v2/error/asset.py

@@ -0,0 +1,5 @@
+from spaceone.core.error import *
+
+
+class ERROR_RESOURCE_ALREADY_DELETED(ERROR_INVALID_ARGUMENT):
+    _message = "{resource_type} has already been deleted. ({resource_id})"

src/spaceone/inventory_v2/error/collector.py

@@ -0,0 +1,105 @@
+from spaceone.core.error import *
+
+
+class ERROR_NO_COLLECTOR(ERROR_BASE):
+    _message = "{collector_id} does not exist in {domain_id}"
+
+
+class ERROR_COLLECTOR_STATE(ERROR_BASE):
+    _message = "collector state is {state}"
+
+
+class ERROR_INIT_PLUGIN_FAILURE(ERROR_BASE):
+    _message = "Fail to init plugin, params={params}"
+
+
+class ERROR_VERIFY_PLUGIN_FAILURE(ERROR_BASE):
+    _message = "Fail to verify plugin, params={params}"
+
+
+class ERROR_NO_PLUGIN_PARAMETER(ERROR_BASE):
+    _message = "parameter: {param} is required"
+
+
+class ERROR_TOKEN_AUTHENTICATION_FAILURE(ERROR_BASE):
+    _message = "A access token or refresh token is invalid."
+
+
+class ERROR_AUTHENTICATION_FAILURE_PLUGIN(ERROR_BASE):
+    _message = (
+        "External plugin authentication exception. (plugin_error_message={message})"
+    )
+
+
+class ERROR_JOB_STATE_CHANGE(ERROR_BASE):
+    _message = "Job {job_id} state change: {status} -> {action}"
+
+
+class ERROR_JOB_TASK_STATE_CHANGE(ERROR_BASE):
+    _message = "Job task {job_task_id} state change: {status} -> {action}"
+
+
+class ERROR_COLLECT_FILTER(ERROR_BASE):
+    _message = "collect failed, plugin_info: {plugin_info}, filter: {param}"
+
+
+class ERROR_COLLECTOR_SECRET(ERROR_BASE):
+    _message = "collect failed, plugin_info: {plugin_info}, secret_id: {param}"
+
+
+class ERROR_JOB_UPDATE(ERROR_BASE):
+    _message = "job update failed, param={param}"
+
+
+class ERROR_COLLECTOR_COLLECTING(ERROR_BASE):
+    _message = "collecting failed, plugin_info: {plugin_info}, filter: {filter}"
+
+
+class ERROR_COLLECT_CANCELED(ERROR_BASE):
+    _message = "collecting canceled, job_id: {job_id}"
+
+
+class ERROR_UNSUPPORTED_RESOURCE_TYPE(ERROR_BASE):
+    _message = "collector can not find resource_type: {resource_type}"
+
+
+class ERROR_UNSUPPORTED_FILTER_KEY(ERROR_BASE):
+    _message = "request unsupported filter_key {filter_key} : {filter_value}"
+
+
+class ERROR_COLLECT_INITIALIZE(ERROR_BASE):
+    _message = "failed on stage {stage}, params: {params}"
+
+
+class ERROR_INVALID_PLUGIN_VERSION(ERROR_INVALID_ARGUMENT):
+    _message = (
+        "Plugin version is invalid. (plugin_id = {plugin_id}, version = {version})"
+    )
+
+
+class ERROR_NOT_ALLOWED_PLUGIN_ID(ERROR_INVALID_ARGUMENT):
+    _message = "Changing plugin_id is not allowed. (old_plugin_id = {old_plugin_id}, new_plugin_id = {new_plugin_id})"
+
+
+class ERROR_WRONG_PLUGIN_SETTINGS(ERROR_BASE):
+    _message = "The plugin settings is incorrect. (key = {key})"
+
+
+class ERROR_INVALID_PLUGIN_OPTIONS(ERROR_INTERNAL_API):
+    _message = "The options received from the plugin is invalid. (reason = {reason})"
+
+
+class ERROR_RESOURCE_KEYS_NOT_DEFINED(ERROR_BASE):
+    _message = "{resource_type} manager does not define resource_keys field"
+
+
+class ERROR_TOO_MANY_MATCH(ERROR_BASE):
+    _message = "The same resource exists. (match_key = {match_key}, matched_resources = {resources}, more = {more})"
+
+
+class ERROR_UNSUPPORTED_SCHEDULE(ERROR_BASE):
+    _message = "supported schedules: {supported}, requested: {requested}"
+
+
+class ERROR_NOT_ALLOWED_SECRET_ID(ERROR_BASE):
+    _message = "Not allowed secret_id: {secret_id}"

src/spaceone/inventory_v2/info/region_info.py

@@ -3,12 +3,13 @@
 from spaceone.api.inventory.v2 import region_pb2
 from spaceone.core.pygrpc.message_type import *
 from spaceone.core import utils
-from spaceone.inventory_v2.model.region_model import Region
+from spaceone.inventory_v2.model.region.region_model import Region
 
 __all__ = ["RegionInfo", "RegionsInfo"]
 
 _LOGGER = logging.getLogger(__name__)
 
+
 def RegionInfo(region_vo: Region, minimal=False):
     info = {
         "region_id": region_vo.region_id,
@@ -35,4 +36,4 @@ def RegionsInfo(region_vos, total_count, **kwargs):
     return region_pb2.RegionsInfo(
         results=list(map(functools.partial(RegionInfo, **kwargs), region_vos)),
         total_count=total_count,
-    )
+    )

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

@@ -0,0 +1,56 @@
+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
+
+
+class Metric(BaseAPI, metric_pb2_grpc.MetricServicer):
+    pb2 = metric_pb2
+    pb2_grpc = metric_pb2_grpc
+
+    def create(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_svc = MetricService(metadata)
+        response: dict = metric_svc.create(params)
+        return self.dict_to_message(response)
+
+    def update(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_svc = MetricService(metadata)
+        response: dict = metric_svc.update(params)
+        return self.dict_to_message(response)
+
+    def delete(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_svc = MetricService(metadata)
+        metric_svc.delete(params)
+        return self.empty()
+
+    def run(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_svc = MetricService(metadata)
+        metric_svc.run(params)
+        return self.empty()
+
+    def test(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_svc = MetricService(metadata)
+        response: dict = metric_svc.test(params)
+        return self.dict_to_message(response)
+
+    def get(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_svc = MetricService(metadata)
+        response: dict = metric_svc.get(params)
+        return self.dict_to_message(response)
+
+    def list(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_svc = MetricService(metadata)
+        response: dict = metric_svc.list(params)
+        return self.dict_to_message(response)
+
+    def stat(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_svc = MetricService(metadata)
+        response: dict = metric_svc.stat(params)
+        return self.dict_to_message(response)

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

@@ -0,0 +1,26 @@
+from spaceone.core.pygrpc import BaseAPI
+from spaceone.api.inventory_v2.v1 import metric_data_pb2, metric_data_pb2_grpc
+from spaceone.inventory.service.metric_data_service import MetricDataService
+
+
+class MetricData(BaseAPI, metric_data_pb2_grpc.MetricDataServicer):
+    pb2 = metric_data_pb2
+    pb2_grpc = metric_data_pb2_grpc
+
+    def list(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_data_svc = MetricDataService(metadata)
+        response: dict = metric_data_svc.list(params)
+        return self.dict_to_message(response)
+
+    def analyze(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_data_svc = MetricDataService(metadata)
+        response: dict = metric_data_svc.analyze(params)
+        return self.dict_to_message(response)
+
+    def stat(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_data_svc = MetricDataService(metadata)
+        response: dict = metric_data_svc.stat(params)
+        return self.dict_to_message(response)

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

@@ -0,0 +1,44 @@
+from spaceone.core.pygrpc import BaseAPI
+from spaceone.api.inventory_v2.v1 import metric_example_pb2, metric_example_pb2_grpc
+from spaceone.inventory.service.metric_example_service import MetricExampleService
+
+
+class MetricExample(BaseAPI, metric_example_pb2_grpc.MetricExampleServicer):
+    pb2 = metric_example_pb2
+    pb2_grpc = metric_example_pb2_grpc
+
+    def create(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_example_svc = MetricExampleService(metadata)
+        response: dict = metric_example_svc.create(params)
+        return self.dict_to_message(response)
+
+    def update(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_example_svc = MetricExampleService(metadata)
+        response: dict = metric_example_svc.update(params)
+        return self.dict_to_message(response)
+
+    def delete(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_example_svc = MetricExampleService(metadata)
+        metric_example_svc.delete(params)
+        return self.empty()
+
+    def get(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_example_svc = MetricExampleService(metadata)
+        response: dict = metric_example_svc.get(params)
+        return self.dict_to_message(response)
+
+    def list(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_example_svc = MetricExampleService(metadata)
+        response: dict = metric_example_svc.list(params)
+        return self.dict_to_message(response)
+
+    def stat(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        metric_example_svc = MetricExampleService(metadata)
+        response: dict = metric_example_svc.stat(params)
+        return self.dict_to_message(response)

src/spaceone/inventory_v2/interface/task/v1/inventory_scheduler.py

@@ -0,0 +1,73 @@
+import logging
+from datetime import datetime
+from spaceone.core.error import ERROR_CONFIGURATION
+from spaceone.core.locator import Locator
+from spaceone.core.scheduler import HourlyScheduler
+from spaceone.core import config, utils
+from spaceone.inventory_v2.service.collector_service import CollectorService
+
+__all__ = ["InventoryHourlyScheduler"]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class InventoryHourlyScheduler(HourlyScheduler):
+    def __init__(self, queue, interval, minute=":00"):
+        super().__init__(queue, interval, minute)
+        self.locator = Locator()
+        self._init_config()
+
+    def _init_config(self):
+        self._token = config.get_global("TOKEN")
+        if self._token is None:
+            raise ERROR_CONFIGURATION(key="TOKEN")
+
+    def create_task(self):
+        current_hour = datetime.utcnow().hour
+        return [
+            self._create_job_request(collector_vo)
+            for collector_vo in self.list_schedule_collectors(current_hour)
+        ]
+
+    def list_schedule_collectors(self, current_hour: int):
+        try:
+            collector_svc: CollectorService = self.locator.get_service(
+                CollectorService, {"token": self._token}
+            )
+            collector_vos, total_count = collector_svc.scheduled_collectors(
+                {"hour": current_hour}
+            )
+            _LOGGER.debug(
+                f"[list_schedule_collectors] scheduled collectors count (UTC {current_hour}): {total_count}"
+            )
+            return collector_vos
+        except Exception as e:
+            _LOGGER.error(e, exc_info=True)
+            return []
+
+    def _create_job_request(self, collector_vo):
+        schedule_job = {
+            "locator": "SERVICE",
+            "name": "CollectorService",
+            "metadata": {
+                "token": self._token,
+            },
+            "method": "collect",
+            "params": {
+                "params": {
+                    "collector_id": collector_vo.collector_id,
+                    "domain_id": collector_vo.domain_id,
+                }
+            },
+        }
+
+        _LOGGER.debug(
+            f"[_create_job_request] tasks: inventory_collect_schedule: {collector_vo.collector_id}"
+        )
+
+        return {
+            "name": "inventory_collect_schedule",
+            "version": "v1",
+            "executionEngine": "BaseWorker",
+            "stages": [schedule_job],
+        }