Merge pull request #16 from gikang82/master

etc: support latest plugin framework

Dockerfile

@@ -1,25 +1,24 @@
-FROM cloudforet/python-core:1
+FROM cloudforet/python-core:2.0
 
 ENV PYTHONUNBUFFERED 1
-ENV CLOUDONE_PORT 50051
+ENV SPACEONE_PORT 50051
 ENV SERVER_TYPE grpc
 ENV PKG_DIR /tmp/pkg
 ENV SRC_DIR /tmp/src
 
 RUN apt update && apt upgrade -y
 
 COPY pkg/*.txt ${PKG_DIR}/
+
 RUN pip install --upgrade pip && \
-    pip install --upgrade --use-deprecated=legacy-resolver -r ${PKG_DIR}/pip_requirements.txt && \
-    pip install --upgrade spaceone-api
+    pip install --upgrade -r ${PKG_DIR}/pip_requirements.txt && \
+    pip install --upgrade --pre spaceone-core spaceone-api spaceone-cost-analysis==2.0.dev120
 
 COPY src ${SRC_DIR}
-
 WORKDIR ${SRC_DIR}
-RUN python3 setup.py install && \
-    rm -rf /tmp/*
+RUN python3 setup.py install && rm -rf /tmp/*
 
-EXPOSE ${CLOUDONE_PORT}
+EXPOSE ${SPACEONE_PORT}
 
 ENTRYPOINT ["spaceone"]
-CMD ["grpc", "spaceone.cost_analysis"]
+CMD ["run", "plugin-server", "plugin"]

src/plugin/__init__.py

@@ -0,0 +1,3 @@
+from cloudforet.cost_analysis.manager.data_source_manager import DataSourceManager
+from cloudforet.cost_analysis.manager.job_manager import JobManager
+from cloudforet.cost_analysis.manager.cost_manager import CostManager

src/plugin/conf/cost_conf.py

@@ -0,0 +1,3 @@
+DEFAULT_BILLING_DATASET = 'spaceone_billing_data'
+BIGQUERY_TABLE_PREFIX = 'gcp_billing_export_v1'
+SECRET_TYPE_DEFAULT = 'MANUAL'

src/plugin/connector/__init__.py

@@ -0,0 +1,2 @@
+from cloudforet.cost_analysis.connector.bigquery_connector import BigqueryConnector
+from cloudforet.cost_analysis.connector.cloud_billing_connector import CloudBillingConnector

src/plugin/connector/bigquery_connector.py

@@ -0,0 +1,52 @@
+import logging
+import google.oauth2.service_account
+import pandas_gbq
+from googleapiclient.discovery import build
+
+from spaceone.core.connector import BaseConnector
+from cloudforet.cost_analysis.error import *
+
+_LOGGER = logging.getLogger(__name__)
+
+REQUIRED_SECRET_KEYS = ["project_id", "private_key", "token_uri", "client_email"]
+
+
+class BigqueryConnector(BaseConnector):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.project_id = None
+        self.credentials = None
+        self.google_client = None
+
+    def create_session(self, options: dict, secret_data: dict, schema: str):
+        self._check_secret_data(secret_data)
+        self.project_id = secret_data['project_id']
+
+        self.credentials = google.oauth2.service_account.Credentials.from_service_account_info(secret_data)
+        self.google_client = build('bigquery', 'v2', credentials=self.credentials)
+
+    def list_tables(self, billing_export_project_id, dataset_id, **query):
+        table_list = []
+
+        query.update({'projectId': billing_export_project_id,
+                      'datasetId': dataset_id})
+
+        request = self.google_client.tables().list(**query)
+        while request is not None:
+            response = request.execute()
+            for table in response.get('tables', []):
+                table_list.append(table)
+            request = self.google_client.tables().list_next(previous_request=request, previous_response=response)
+
+        return table_list
+
+    def read_df_from_bigquery(self, query):
+        return pandas_gbq.read_gbq(query, project_id=self.project_id, credentials=self.credentials)
+
+    @staticmethod
+    def _check_secret_data(secret_data):
+        missing_keys = [key for key in REQUIRED_SECRET_KEYS if key not in secret_data]
+        if missing_keys:
+            for key in missing_keys:
+                raise ERROR_REQUIRED_PARAMETER(key=f"secret_data.{key}")

src/plugin/main.py

@@ -0,0 +1,157 @@
+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'      # Required
+        }
+
+    Returns:
+        None
+    """
+
+    options = params["options"]
+    secret_data = params["secret_data"]
+    domain_id = params.get("domain_id")
+    schema = params.get("schema")
+
+    data_source_mgr = DataSourceManager()
+    data_source_mgr.verify_plugin(options, secret_data, domain_id, 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'      # Required
+        }
+
+    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'      # Required
+        }
+
+    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)
+
+
+@app.route("Cost.get_linked_accounts")
+def cost_get_linked_accounts(params: dict) -> dict:
+    """ get linked accounts
+
+    Args:
+        params: (CostGetLinkedAccountsRequest): {
+            'options': 'dict'
+            'schema': 'str'
+            'secret_data': 'dict'
+            'domain_id': 'str'
+        }
+
+    Returns:
+        {
+            'account_id': 'str'
+            'name': 'str'
+        }
+    """
+    options = params["options"]
+    secret_data = params["secret_data"]
+
+    schema = params.get("schema")
+
+    cost_mgr = CostManager()
+    return cost_mgr.get_linked_accounts(options, secret_data, schema)

src/plugin/manager/__init__.py

@@ -0,0 +1,3 @@
+from cloudforet.cost_analysis.manager.data_source_manager import DataSourceManager
+from cloudforet.cost_analysis.manager.job_manager import JobManager
+from cloudforet.cost_analysis.manager.cost_manager import CostManager