refactor: refactor plugin server

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.8-slim
+FROM cloudforet/python-core:2.0
 
 ENV PYTHONUNBUFFERED 1
 ENV SPACEONE_PORT 50051
@@ -13,15 +13,11 @@ COPY pkg/*.txt ${PKG_DIR}/
 RUN pip install --upgrade pip && \
     pip install --upgrade -r ${PKG_DIR}/pip_requirements.txt
 
-ARG CACHEBUST=1
-RUN pip install --upgrade --pre spaceone-core spaceone-api
-
 COPY src ${SRC_DIR}
 WORKDIR ${SRC_DIR}
-RUN python3 setup.py install && \
-    rm -rf /tmp/*
+RUN python3 setup.py install && rm -rf /tmp/*
 
 EXPOSE ${SPACEONE_PORT}
 
 ENTRYPOINT ["spaceone"]
-CMD ["grpc", "spaceone.cost_analysis"]
+CMD ["run", "plugin-server", "plugin"]

pkg/pip_requirements.txt

@@ -1,8 +1,4 @@
+
 spaceone-core
 spaceone-api
-schematics
-requests
-boto3
-pyarrow
-pandas
-
+spaceone-cost-analysis

src/spaceone/cost_analysis/connector/aws_s3_connector.py → src/plugin/connector/aws_s3_connector.py

@@ -6,7 +6,7 @@
 
 from spaceone.core import utils
 from spaceone.core.connector import BaseConnector
-from spaceone.cost_analysis.error import *
+from spaceone.core.error import *
 
 __all__ = ['AWSS3Connector']
 

src/spaceone/cost_analysis/connector/spaceone_connector.py → src/plugin/connector/spaceone_connector.py

@@ -1,12 +1,11 @@
 import logging
 import re
 import requests
-from requests import HTTPError
 from google.protobuf.json_format import MessageToDict
 
 from spaceone.core.connector.space_connector import SpaceConnector
 from spaceone.core.connector import BaseConnector
-from spaceone.cost_analysis.error import *
+from spaceone.core.error import *
 
 __all__ = ['SpaceONEConnector']
 
@@ -96,7 +95,7 @@ def request(self, method, params, **kwargs):
         response = requests.post(url, json=params, headers=headers)
 
         if response.status_code >= 400:
-            raise HTTPError(f'HTTP {response.status_code} Error: {response.json()["detail"]}')
+            raise requests.HTTPError(f'HTTP {response.status_code} Error: {response.json()["detail"]}')
 
         response = response.json()
         return response

src/plugin/main.py

@@ -0,0 +1,126 @@
+from typing import Generator
+from spaceone.cost_analysis.plugin.data_source.lib.server import DataSourcePluginServer
+from .manager.data_source_manager import DataSourceManager
+from .manager.job_manager import JobManager
+from .manager.cost_manager import CostManager
+
+app = DataSourcePluginServer()
+
+
+@app.route('DataSource.init')
+def data_source_init(params: dict) -> dict:
+    """ init plugin by options
+
+    Args:
+        params (DataSourceInitRequest): {
+            'options': 'dict',    # Required
+            'domain_id': 'str'    # Required
+        }
+
+    Returns:
+        PluginResponse: {
+            'metadata': 'dict'
+        }
+    """
+    options = params['options']
+
+    data_source_mgr = DataSourceManager()
+    return data_source_mgr.init_response(options)
+
+
+@app.route('DataSource.verify')
+def data_source_verify(params: dict) -> None:
+    """ Verifying data source plugin
+
+    Args:
+        params (CollectorVerifyRequest): {
+            'options': 'dict',      # Required
+            'secret_data': 'dict',  # Required
+            'schema': 'str',
+            'domain_id': 'str'
+        }
+
+    Returns:
+        None
+    """
+
+    options = params['options']
+    secret_data = params['secret_data']
+    schema = params.get('schema')
+
+    data_source_mgr = DataSourceManager()
+    data_source_mgr.verify_plugin(options, secret_data, schema)
+
+
+@app.route('Job.get_tasks')
+def job_get_tasks(params: dict) -> dict:
+    """ Get job tasks
+
+    Args:
+        params (JobGetTaskRequest): {
+            'options': 'dict',      # Required
+            'secret_data': 'dict',  # Required
+            'schema': 'str',
+            'start': 'str',
+            'last_synchronized_at': 'datetime',
+            'domain_id': 'str'
+        }
+
+    Returns:
+        TasksResponse: {
+            'tasks': 'list',
+            'changed': 'list'
+        }
+
+    """
+
+    domain_id = params['domain_id']
+    options = params['options']
+    secret_data = params['secret_data']
+    schema = params.get('schema')
+    start = params.get('start')
+    last_synchronized_at = params.get('last_synchronized_at')
+
+    job_mgr = JobManager()
+    return job_mgr.get_tasks(domain_id, options, secret_data, schema, start, last_synchronized_at)
+
+
+@app.route('Cost.get_data')
+def cost_get_data(params: dict) -> Generator[dict, None, None]:
+    """ Get external cost data
+
+    Args:
+        params (CostGetDataRequest): {
+            'options': 'dict',      # Required
+            'secret_data': 'dict',  # Required
+            'schema': 'str',
+            'task_options': 'dict',
+            'domain_id': 'str'
+        }
+
+    Returns:
+        Generator[ResourceResponse, None, None]
+        {
+            'cost': 'float',
+            'usage_quantity': 'float',
+            'usage_unit': 'str',
+            'provider': 'str',
+            'region_code': 'str',
+            'product': 'str',
+            'usage_type': 'str',
+            'resource': 'str',
+            'tags': 'dict'
+            'additional_info': 'dict'
+            'data': 'dict'
+            'billed_date': 'str'
+        }
+    """
+
+    options = params['options']
+    secret_data = params['secret_data']
+
+    task_options = params.get('task_options', {})
+    schema = params.get('schema')
+
+    cost_mgr = CostManager()
+    return cost_mgr.get_data(options, secret_data, task_options, schema)

src/plugin/manager/__init__.py

@@ -0,0 +1 @@
+

src/spaceone/cost_analysis/manager/cost_manager.py → src/plugin/manager/cost_manager.py

@@ -1,13 +1,13 @@
 import logging
+from typing import Generator
 from datetime import datetime
 from dateutil import rrule
 
 from spaceone.core import utils
 from spaceone.core.manager import BaseManager
-from spaceone.cost_analysis.error import *
-from spaceone.cost_analysis.connector.aws_s3_connector import AWSS3Connector
-from spaceone.cost_analysis.connector.spaceone_connector import SpaceONEConnector
-from spaceone.cost_analysis.model.cost_model import Cost
+from spaceone.core.error import *
+from ..connector.aws_s3_connector import AWSS3Connector
+from ..connector.spaceone_connector import SpaceONEConnector
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -51,10 +51,11 @@ class CostManager(BaseManager):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.aws_s3_connector: AWSS3Connector = self.locator.get_connector('AWSS3Connector')
-        self.space_connector: SpaceONEConnector = self.locator.get_connector('SpaceONEConnector')
+        self.aws_s3_connector = AWSS3Connector()
+        self.space_connector = SpaceONEConnector()
 
-    def get_data(self, options, secret_data, schema, task_options):
+    def get_data(self, options: dict, secret_data: dict, task_options: dict, schema: str = None) \
+            -> Generator[dict, None, None]:
         self.aws_s3_connector.create_session(options, secret_data, schema)
         self._check_task_options(task_options)
 
@@ -79,7 +80,9 @@ def get_data(self, options, secret_data, schema, task_options):
                 for results in response_stream:
                     yield self._make_cost_data(results, account_id)
 
-        yield []
+        yield {
+            'results': []
+        }
 
     def _update_sync_state(self, options, secret_data, schema, service_account_id):
         self.space_connector.init_client(options, secret_data, schema)
@@ -156,13 +159,9 @@ class CostSummaryItem(BaseModel):
 
             costs_data.append(data)
 
-            # Excluded because schema validation is too slow
-            # cost_data = Cost(data)
-            # cost_data.validate()
-            #
-            # costs_data.append(cost_data.to_primitive())
-
-        return costs_data
+        return {
+            'results': costs_data
+        }
 
     @staticmethod
     def _get_tags_from_cost_data(cost_data: dict) -> dict:

src/plugin/manager/data_source_manager.py

@@ -0,0 +1,43 @@
+import logging
+
+from spaceone.core.manager import BaseManager
+from ..connector.aws_s3_connector import AWSS3Connector
+from ..connector.spaceone_connector import SpaceONEConnector
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class DataSourceManager(BaseManager):
+
+    @staticmethod
+    def init_response(options: dict) -> dict:
+        return {
+            'metadata': {
+                'currency': 'USD',
+                'supported_secret_types': ['MANUAL'],
+                'data_source_rules': [
+                    {
+                        'name': 'match_service_account',
+                        'conditions_policy': 'ALWAYS',
+                        'actions': {
+                            'match_service_account': {
+                                'source': 'additional_info.Account ID',
+                                'target': 'data.account_id'
+                            }
+                        },
+                        'options': {
+                            'stop_processing': True
+                        }
+                    }
+                ]
+            }
+        }
+
+    @staticmethod
+    def verify_plugin(options: dict, secret_data: dict, schema: str = None) -> None:
+        space_connector = SpaceONEConnector()
+        space_connector.init_client(options, secret_data, schema)
+        space_connector.verify_plugin()
+
+        aws_s3_connector = AWSS3Connector()
+        aws_s3_connector.create_session(options, secret_data, schema)

src/spaceone/cost_analysis/manager/job_manager.py → src/plugin/manager/job_manager.py

@@ -3,8 +3,7 @@
 
 from spaceone.core.error import *
 from spaceone.core.manager import BaseManager
-from spaceone.cost_analysis.connector.spaceone_connector import SpaceONEConnector
-from spaceone.cost_analysis.model.job_model import Tasks
+from ..connector.spaceone_connector import SpaceONEConnector
 
 _LOGGER = logging.getLogger(__name__)
 _DEFAULT_DATABASE = 'MZC'
@@ -14,9 +13,10 @@ class JobManager(BaseManager):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.space_connector: SpaceONEConnector = self.locator.get_connector('SpaceONEConnector')
+        self.space_connector = SpaceONEConnector()
 
-    def get_tasks(self, options, secret_data, schema, start, last_synchronized_at, domain_id):
+    def get_tasks(self, domain_id: str, options: dict, secret_data: dict, schema: str = None, start: str = None,
+                  last_synchronized_at: datetime = None) -> dict:
         tasks = []
         changed = []
 
@@ -75,11 +75,6 @@ def get_tasks(self, options, secret_data, schema, start, last_synchronized_at, d
             _LOGGER.debug(f'[get_tasks] tasks: {tasks}')
             _LOGGER.debug(f'[get_tasks] changed: {changed}')
 
-            tasks = Tasks({'tasks': tasks, 'changed': changed})
-
-            tasks.validate()
-            return tasks.to_primitive()
-
         else:
             _LOGGER.debug(f'[get_tasks] no project: tags.domain_id = {domain_id}')
 

src/setup.py

@@ -33,11 +33,7 @@
     install_requires=[
         'spaceone-core',
         'spaceone-api',
-        'schematics',
-        'requests',
-        'boto3',
-        'pyarrow',
-        'pandas'
+        'spaceone-cost-analysis'
     ],
     zip_safe=False,
 )