Merge pull request #319 from jinyoungmoonDEV/master

add: add schedule field to DataSource

Author: jinyoungmoonDEV

src/spaceone/cost_analysis/info/data_source_info.py

@@ -37,16 +37,28 @@ def PluginInfo(vo):
         return None
 
 
+def ScheduleInfo(vo):
+    if vo:
+        info = {
+            "state": vo.state,
+            "hour": vo.hour,
+        }
+
+        return data_source_pb2.Schedule(**info)
+    else:
+        return None
+
+
 def DataSourceInfo(data_source_vo: DataSource, minimal=False):
     info = {
         "data_source_id": data_source_vo.data_source_id,
         "workspace_id": data_source_vo.workspace_id,
         "name": data_source_vo.name,
-        "state": data_source_vo.state,
         "data_source_type": data_source_vo.data_source_type,
         "secret_type": data_source_vo.secret_type,
         "provider": data_source_vo.provider,
         "resource_group": data_source_vo.resource_group,
+        "schedule": ScheduleInfo(data_source_vo.schedule),
     }
 
     if not minimal:

src/spaceone/cost_analysis/interface/grpc/data_source.py

@@ -63,26 +63,6 @@ def verify_plugin(self, request, context):
             data_source_service.verify_plugin(params)
             return self.locator.get_info("EmptyInfo")
 
-    def enable(self, request, context):
-        params, metadata = self.parse_request(request, context)
-
-        with self.locator.get_service(
-            "DataSourceService", metadata
-        ) as data_source_service:
-            return self.locator.get_info(
-                "DataSourceInfo", data_source_service.enable(params)
-            )
-
-    def disable(self, request, context):
-        params, metadata = self.parse_request(request, context)
-
-        with self.locator.get_service(
-            "DataSourceService", metadata
-        ) as data_source_service:
-            return self.locator.get_info(
-                "DataSourceInfo", data_source_service.disable(params)
-            )
-
     def deregister(self, request, context):
         params, metadata = self.parse_request(request, context)
 

src/spaceone/cost_analysis/interface/task/v1/data_source_sync_scheduler.py

@@ -20,7 +20,6 @@ def _init_config(self):
         self._token = config.get_global("TOKEN")
         if self._token is None:
             raise ERROR_CONFIGURATION(key="TOKEN")
-        self._data_source_sync_hour = config.get_global("DATA_SOURCE_SYNC_HOUR", 16)
         self._cost_report_sync_hour = config.get_global("COST_REPORT_RUN_HOUR", 0)
 
     def create_task(self) -> list:
@@ -30,34 +29,25 @@ def create_task(self) -> list:
         return tasks
 
     def _create_data_source_sync_task(self):
-        if datetime.utcnow().hour == self._data_source_sync_hour:
-            stp = {
-                "name": "data_source_sync_schedule",
-                "version": "v1",
-                "executionEngine": "BaseWorker",
-                "stages": [
-                    {
-                        "locator": "SERVICE",
-                        "name": "JobService",
-                        "metadata": {"token": self._token},
-                        "method": "create_jobs_by_data_source",
-                        "params": {"params": {}},
-                    }
-                ],
-            }
+        stp = {
+            "name": "data_source_sync_schedule",
+            "version": "v1",
+            "executionEngine": "BaseWorker",
+            "stages": [
+                {
+                    "locator": "SERVICE",
+                    "name": "JobService",
+                    "metadata": {"token": self._token},
+                    "method": "create_jobs_by_data_source",
+                    "params": {"params": {"sync_hour": datetime.utcnow().hour}},
+                }
+            ],
+        }
 
-            print(
-                f"{utils.datetime_to_iso8601(datetime.utcnow())} [INFO] [create_task] create_jobs_by_data_source => START"
-            )
-            return [stp]
-        else:
-            print(
-                f"{utils.datetime_to_iso8601(datetime.utcnow())} [INFO] [create_task] create_jobs_by_data_source => SKIP"
-            )
-            print(
-                f"{utils.datetime_to_iso8601(datetime.utcnow())} [INFO] [create_task] data_source_sync_time: {self._data_source_sync_hour} hour (UTC)"
-            )
-            return []
+        print(
+            f"{utils.datetime_to_iso8601(datetime.utcnow())} [INFO] [create_task] create_jobs_by_data_source => START"
+        )
+        return [stp]
 
     # todo: split scheduler
     def _create_cost_report_run_task(self):

src/spaceone/cost_analysis/model/data_source/request.py

@@ -25,6 +25,11 @@ class Plugin(BaseModel):
     upgrade_mode: Union[str, None] = None
 
 
+class Schedule(BaseModel):
+    state: str
+    hour: Union[int, None] = None
+
+
 class DataSourceRegisterRequest(BaseModel):
     name: str
     data_source_type: DataSourceType
@@ -33,6 +38,7 @@ class DataSourceRegisterRequest(BaseModel):
     secret_filter: Union[dict, None] = None
     template: Union[dict, None] = None
     plugin_info: Union[Plugin, None] = None
+    schedule: Schedule
     tags: Union[dict, None] = None
     resource_group: Union[ResourceGroup, None] = None
     workspace_id: Union[str, None] = None
@@ -55,7 +61,6 @@ class DataSourceSearchQueryRequest(BaseModel):
     query: Union[dict, None] = None
     data_source_id: Union[str, None] = None
     name: Union[str, None] = None
-    state: Union[State, None] = None
     data_source_type: Union[DataSourceType, None] = None
     provider: Union[str, None] = None
     connected_workspace_id: Union[str, None] = None

src/spaceone/cost_analysis/model/data_source/response.py

@@ -5,7 +5,6 @@
 from spaceone.core import utils
 
 from spaceone.cost_analysis.model.data_source.request import (
-    State,
     DataSourceType,
     SecretType,
     ResourceGroup,
@@ -20,13 +19,13 @@
 class DataSourceResponse(BaseModel):
     data_source_id: Union[str, None] = None
     name: Union[str, None] = None
-    state: Union[State, None] = None
     data_source_type: Union[DataSourceType, None] = None
     permissions: Union[dict, None] = None
     provider: Union[str, None] = None
     secret_type: Union[SecretType, None] = None
     secret_filter: Union[dict, None] = None
     plugin_info: Union[dict, None] = None
+    schedule: Union[dict, None] = None
     template: Union[dict, None] = None
     tags: Union[dict, None] = None
     cost_tag_keys: Union[list, None] = None

src/spaceone/cost_analysis/model/data_source_model.py

@@ -30,12 +30,19 @@ def to_dict(self):
         return dict(self.to_mongo())
 
 
-class DataSource(MongoModel):
-    data_source_id = StringField(max_length=40, generate_id="ds", unique=True)
-    name = StringField(max_length=255, unique_with="domain_id")
+class Schedule(EmbeddedDocument):
     state = StringField(
         max_length=20, default="ENABLED", choices=("ENABLED", "DISABLED")
     )
+    hour = IntField(min_value=0, max_length=23, null=True)
+
+    def to_dict(self):
+        return dict(self.to_mongo())
+
+
+class DataSource(MongoModel):
+    data_source_id = StringField(max_length=40, generate_id="ds", unique=True)
+    name = StringField(max_length=255, unique_with="domain_id")
     data_source_type = StringField(max_length=20, choices=("LOCAL", "EXTERNAL"))
     secret_type = StringField(
         max_length=32,
@@ -46,6 +53,7 @@ class DataSource(MongoModel):
     permissions = DictField(default=None, null=True)
     provider = StringField(max_length=40, default=None, null=True)
     plugin_info = EmbeddedDocumentField(PluginInfo, default=None, null=True)
+    schedule = EmbeddedDocumentField(Schedule)
     template = DictField(default={})
     tags = DictField(default={})
     cost_tag_keys = ListField(StringField())
@@ -65,11 +73,11 @@ class DataSource(MongoModel):
     meta = {
         "updatable_fields": [
             "name",
-            "state",
             "permissions",
             "plugin_info",
             "secret_filter",
             "template",
+            "schedule",
             "tags",
             "updated_at",
             "last_synchronized_at",
@@ -83,14 +91,13 @@ class DataSource(MongoModel):
             "data_source_id",
             "workspace_id",
             "name",
-            "state",
             "data_source_type",
             "secret_type",
             "provider",
+            "schedule",
         ],
         "ordering": ["name"],
         "indexes": [
-            "state",
             "data_source_type",
             "provider",
             "resource_group",

src/spaceone/cost_analysis/service/data_source_service.py

@@ -3,6 +3,7 @@
 from typing import Tuple, Union
 
 from mongoengine import QuerySet
+from spaceone.core import config
 from spaceone.core.service import *
 from spaceone.cost_analysis.error import *
 from spaceone.cost_analysis.model.data_source.request import *
@@ -61,7 +62,8 @@ def register(
                 'secret_filter': 'dict',
                 'template': 'dict',
                 'plugin_info': 'dict',
-                'tags': 'dict',
+                'schedule;: 'dict',         # required
+                'tags': 'dict',             # required
                 'resource_group': 'str      # required
                 'workspace_id': 'str'
                 'domain_id': 'str'          # injected from auth
@@ -85,6 +87,16 @@ def register(
         else:
             params["workspace_id"] = "*"
 
+        if not params.get("schedule"):
+            schedule = {
+                "state" : "ENABLED",
+                "hour": config.get_global("DATA_SOURCE_SYNC_HOUR", 16),
+            }
+
+            params["schedule"] = schedule
+        else:
+             self._check_schedule(params["schedule"])
+
         if data_source_type == "EXTERNAL":
             params["template"] = None
 
@@ -194,6 +206,7 @@ def update(self, params):
                 'name': 'str',
                 'secret_filter': 'dict',
                 'template': 'dict',
+                'schedule': 'dict',
                 'tags': 'dict'
                 'domain_id': 'str'          # injected from auth
             }
@@ -207,6 +220,9 @@ def update(self, params):
             data_source_id, domain_id
         )
 
+        if schedule := params.get("schedule"):
+            self._check_schedule(schedule)
+
         if "secret_filter" in params:
             if data_source_vo.secret_type == "USE_SERVICE_ACCOUNT_SECRET":
                 self.validate_secret_filter(
@@ -431,62 +447,6 @@ def update_plugin(self, params):
 
         return data_source_vo
 
-    @transaction(
-        permission="cost-analysis:DataSource.write",
-        role_types=["DOMAIN_ADMIN", "WORKSPACE_OWNER"],
-    )
-    @check_required(["data_source_id", "domain_id"])
-    def enable(self, params):
-        """Enable data source
-
-        Args:
-            params (dict): {
-                'data_source_id': 'str',  # required
-                'domain_id': 'str'        # injected from auth
-            }
-
-        Returns:
-            data_source_vo (object)
-        """
-
-        data_source_id = params["data_source_id"]
-        domain_id = params["domain_id"]
-        data_source_vo: DataSource = self.data_source_mgr.get_data_source(
-            data_source_id, domain_id
-        )
-
-        return self.data_source_mgr.update_data_source_by_vo(
-            {"state": "ENABLED"}, data_source_vo
-        )
-
-    @transaction(
-        permission="cost-analysis:DataSource.write",
-        role_types=["DOMAIN_ADMIN", "WORKSPACE_OWNER"],
-    )
-    @check_required(["data_source_id", "domain_id"])
-    def disable(self, params):
-        """Disable data source
-
-        Args:
-            params (dict): {
-                'data_source_id': 'str',    # required
-                'domain_id': 'str'          # injected from auth
-            }
-
-        Returns:
-            data_source_vo (object)
-        """
-
-        data_source_id = params["data_source_id"]
-        domain_id = params["domain_id"]
-        data_source_vo: DataSource = self.data_source_mgr.get_data_source(
-            data_source_id, domain_id
-        )
-
-        return self.data_source_mgr.update_data_source_by_vo(
-            {"state": "DISABLED"}, data_source_vo
-        )
-
     @transaction(
         permission="cost-analysis:DataSource.write",
         role_types=["DOMAIN_ADMIN", "WORKSPACE_OWNER"],
@@ -570,7 +530,7 @@ def sync(self, params):
             data_source_id, domain_id, workspace_id
         )
 
-        if data_source_vo.state == "DISABLED":
+        if data_source_vo.schedule.state == "DISABLED":
             raise ERROR_DATA_SOURCE_STATE(data_source_id=data_source_id)
 
         if data_source_vo.data_source_type == "LOCAL":
@@ -625,7 +585,6 @@ def get(self, params: DataSourceGetRequest) -> Union[DataSourceResponse, dict]:
         [
             "data_source_id",
             "name",
-            "state",
             "data_source_type",
             "provider",
             "workspace_id",
@@ -645,7 +604,6 @@ def list(
                 'query': 'dict (spaceone.api.core.v1.Query)'
                 'data_source_id': 'str',
                 'name': 'str',
-                'state': 'str',
                 'data_source_type': 'str',
                 'provider': 'str',
                 'connected_workspace_id': str,
@@ -969,6 +927,14 @@ def _filter_cost_data_keys_with_permissions_by_data_source_vo(
             data_source_vo.cost_data_keys = cost_data_keys
         return data_source_vo
 
+    @staticmethod
+    def _check_schedule(schedule: dict) -> None:
+        if schedule.get("state", "ENABLED") == "ENABLED":
+            if not schedule.get("hour"):
+                raise ERROR_INVALID_PARAMETER(
+                    key="schedule.hour", reason="Need to set an hour when the state is ENABLED."
+                )
+
     @staticmethod
     def _check_only_fields_for_permissions(query: dict) -> None:
         only_fields = query.get("only", [])