fix: fix connect account policy logic (#199)

Author: ImMin5

src/spaceone/cost_analysis/manager/data_source_account_manager.py

@@ -26,12 +26,6 @@ def _rollback(vo: DataSourceAccount):
             )
             vo.delete()
 
-        params.update(
-            {
-                "v_workspace_id": utils.generate_id("workspace"),
-            }
-        )
-
         data_source_account_vo = self.data_source_account_model.create(params)
         self.transaction.add_rollback(_rollback, data_source_account_vo)
 
@@ -104,10 +98,13 @@ def connect_cost_data(self, cost_data: dict) -> Tuple[dict, bool]:
             )
             for account_connect_policy in account_connect_polices:
                 if account_connect_policy.get("name") == "connect_cost_to_account":
-                    source = account_connect_policy.get("source")
-                    target_key = account_connect_policy.get("target", "account_id")
+                    polices = account_connect_policy.get("polices")
+                    source = polices["connect_account_to_workspace"].get("source")
+                    target_key = polices["connect_account_to_workspace"].get(
+                        "target", "account_id"
+                    )
                     target_value = utils.get_dict_value(cost_data, source)
-                    operator = account_connect_policy.get("operator")
+                    operator = polices["connect_account_to_workspace"].get("operator")
 
                     if target_value:
                         ds_account_info = self._get_data_source_account(
@@ -138,17 +135,23 @@ def connect_account_by_data_source_vo(
 
         for account_connect_policy in account_connect_polices:
             if account_connect_policy.get("name") == "connect_account_to_workspace":
-                source = account_connect_policy.get("source")
-                target_key = account_connect_policy.get("target", "references")
+                polices = account_connect_policy.get("polices")
+                source = polices["connect_account_to_workspace"].get("source")
+                target_key = polices["connect_account_to_workspace"].get(
+                    "target", "references"
+                )
+
                 target_value = utils.get_dict_value(
-                    data_source_account_vo.to_dict(), source
+                    data_source_account_vo.to_dict(),
+                    source,
                 )
-                operator = account_connect_policy.get("operator")
+                operator = polices["connect_account_to_workspace"].get("operator")
 
                 if target_value:
                     workspace_info = self._get_workspace(
                         target_key, target_value, domain_id, operator
                     )
+
                     if workspace_info:
                         self.update_data_source_account_by_vo(
                             {"workspace_id": workspace_info.get("workspace_id")},
@@ -158,8 +161,10 @@ def connect_account_by_data_source_vo(
     def _get_workspace(
         self, target_key: str, target_value: str, domain_id: str, operator: str = "eq"
     ) -> Union[dict, None]:
-        if f"workspace:{domain_id}:{target_key}" in self._workspace_info:
-            return self._workspace_info[f"workspace:{domain_id}:{target_key}"]
+        if f"workspace:{domain_id}:{target_key}:{target_value}" in self._workspace_info:
+            return self._workspace_info[
+                f"workspace:{domain_id}:{target_key}:{target_value}"
+            ]
 
         query = {
             "filter": [
@@ -171,15 +176,17 @@ def _get_workspace(
         query["filter"].append({"k": target_key, "v": target_value, "o": operator})
 
         identity_mgr = self.locator.get_manager("IdentityManager")
-        response = identity_mgr.list_workspaces(query, domain_id)
+        response = identity_mgr.list_workspaces({"query": query}, domain_id)
         results = response.get("results", [])
         total_count = response.get("total_count", 0)
 
         workspace_info = None
         if total_count > 0:
             workspace_info = results[0]
 
-        self._workspace_info[f"workspace:{domain_id}:{target_key}"] = workspace_info
+        self._workspace_info[
+            f"workspace:{domain_id}:{target_key}:{target_value}"
+        ] = workspace_info
 
         return workspace_info
 

src/spaceone/cost_analysis/model/data_source_account/database.py

@@ -11,9 +11,9 @@ class DataSourceAccount(MongoModel):
     name = StringField(max_length=255)
     is_sync = BooleanField(default=False)
     v_service_account_id = StringField(max_length=40, generate_id="v_sa", unique=True)
-    v_project_id = StringField(max_length=40, generate_id="v_project", required=True)
+    v_project_id = StringField(max_length=40, generate_id="v_project", unique=True)
     v_workspace_id = StringField(
-        max_length=40, generate_id="v_workspace_id", required=True
+        max_length=40, generate_id="v_workspace_id", unique=True
     )
     service_account_id = StringField(max_length=255, default=None, null=True)
     project_id = StringField(max_length=255, default=None, null=True)

src/spaceone/cost_analysis/service/job_service.py

@@ -260,7 +260,7 @@ def get_cost_data(self, params):
                 is_canceled = False
 
                 for costs_data in self.ds_plugin_mgr.get_cost_data(
-                        options, secret_data, schema, task_options, domain_id
+                    options, secret_data, schema, task_options, domain_id
                 ):
                     results = costs_data.get("results", [])
                     for cost_data in results:
@@ -434,11 +434,11 @@ def create_cost_job(self, data_source_vo: DataSource, job_options):
         return job_vo
 
     def _list_secret_ids_from_secret_type(
-            self,
-            data_source_vo: DataSource,
-            secret_type: str,
-            workspace_id: str,
-            domain_id: str,
+        self,
+        data_source_vo: DataSource,
+        secret_type: str,
+        workspace_id: str,
+        domain_id: str,
     ):
         secret_ids = []
 
@@ -459,7 +459,7 @@ def _list_secret_ids_from_secret_type(
         return secret_ids
 
     def _list_secret_ids_from_secret_filter(
-            self, secret_filter, provider: str, workspace_id: str, domain_id: str
+        self, secret_filter, provider: str, workspace_id: str, domain_id: str
     ):
         secret_manager: SecretManager = self.locator.get_manager(SecretManager)
 
@@ -474,7 +474,7 @@ def _list_secret_ids_from_secret_filter(
 
     @staticmethod
     def _set_secret_filter(
-            secret_filter, provider: str, workspace_id: str, domain_id: str
+        secret_filter, provider: str, workspace_id: str, domain_id: str
     ):
         _filter = [{"k": "domain_id", "v": domain_id, "o": "eq"}]
 
@@ -489,8 +489,8 @@ def _set_secret_filter(
                     {"k": "secret_id", "v": secret_filter["secrets"], "o": "in"}
                 )
             if (
-                    "service_accounts" in secret_filter
-                    and secret_filter["service_accounts"]
+                "service_accounts" in secret_filter
+                and secret_filter["service_accounts"]
             ):
                 _filter.append(
                     {
@@ -586,10 +586,10 @@ def _create_cost_data(self, cost_data, job_task_vo, cost_options):
         self.cost_mgr.create_cost(cost_data, execute_rollback=False)
 
     def _is_job_failed(
-            self,
-            job_id: str,
-            domain_id: str,
-            workspace_id: str,
+        self,
+        job_id: str,
+        domain_id: str,
+        workspace_id: str,
     ):
         job_vo: Job = self.job_mgr.get_job(job_id, domain_id, workspace_id)
 
@@ -599,12 +599,12 @@ def _is_job_failed(
             return False
 
     def _close_job(
-            self,
-            job_id: str,
-            data_source_id: str,
-            domain_id: str,
-            data_keys: list,
-            workspace_id: str = None,
+        self,
+        job_id: str,
+        data_source_id: str,
+        domain_id: str,
+        data_keys: list,
+        workspace_id: str = None,
     ) -> None:
         job_vo: Job = self.job_mgr.get_job(job_id, domain_id, workspace_id)
         no_preload_cache = job_vo.options.get("no_preload_cache", False)
@@ -754,7 +754,7 @@ def _delete_old_cost_data(self, data_source_id, domain_id):
         monthly_cost_vos.delete()
 
     def _delete_changed_cost_data(
-            self, job_vo: Job, start, end, change_filter, domain_id
+        self, job_vo: Job, start, end, change_filter, domain_id
     ):
         query = {
             "filter": [
@@ -800,7 +800,7 @@ def _aggregate_cost_data(self, job_vo: Job, data_keys: list):
 
         for job_task_id in job_task_ids:
             for billed_month in self._distinct_billed_month(
-                    data_source_id, domain_id, job_id, job_task_id
+                data_source_id, domain_id, job_id, job_task_id
             ):
                 self._aggregate_monthly_cost_data(
                     data_source_id,
@@ -832,14 +832,14 @@ def _distinct_billed_month(self, data_source_id, domain_id, job_id, job_task_id)
         return values
 
     def _aggregate_monthly_cost_data(
-            self,
-            data_source_id: str,
-            domain_id: str,
-            job_id: str,
-            job_task_id: str,
-            billed_month: str,
-            data_keys: list,
-            workspace_id: str = None,
+        self,
+        data_source_id: str,
+        domain_id: str,
+        job_id: str,
+        job_task_id: str,
+        billed_month: str,
+        data_keys: list,
+        workspace_id: str = None,
     ):
         query = {
             "group_by": [
@@ -906,7 +906,7 @@ def _get_all_data_sources(self):
         )
 
     def _check_duplicate_job(
-            self, data_source_id: str, domain_id: str, this_job_vo: Job
+        self, data_source_id: str, domain_id: str, this_job_vo: Job
     ):
         query = {
             "filter": [
@@ -932,14 +932,6 @@ def _check_duplicate_job(
 
         return False
 
-    @staticmethod
-    def _get_start_last_synchronized_at(params):
-        start = params.get("start")
-        last_synchronized_at = utils.datetime_to_iso8601(
-            params.get("last_synchronized_at")
-        )
-        return start, last_synchronized_at
-
     def _get_job_task_ids(self, job_id, domain_id):
         job_task_ids = []
         job_task_vos = self.job_task_mgr.filter_job_tasks(
@@ -952,11 +944,11 @@ def _get_job_task_ids(self, job_id, domain_id):
         return job_task_ids
 
     def _get_data_source_account_map(
-            self,
-            data_source_id: str,
-            domain_id: str,
-            workspace_id: str,
-            resource_group: str,
+        self,
+        data_source_id: str,
+        domain_id: str,
+        workspace_id: str,
+        resource_group: str,
     ) -> Dict[str, DataSourceAccount]:
         data_source_account_map = {}
         conditions = {
@@ -978,11 +970,11 @@ def _get_data_source_account_map(
         return data_source_account_map
 
     def _get_linked_accounts_from_data_source_vo(
-            self,
-            data_source_vo: DataSource,
-            options: dict,
-            secret_data: dict,
-            schema: dict = None,
+        self,
+        data_source_vo: DataSource,
+        options: dict,
+        secret_data: dict,
+        schema: dict = None,
     ) -> list:
         linked_accounts = []
 
@@ -1044,6 +1036,14 @@ def _update_data_source_is_synced(self, job_vo: Job) -> None:
             f"[_update_data_source_account_sync_status] synced_account_ids: {synced_accounts} / {data_source_id} {domain_id}"
         )
 
+    @staticmethod
+    def _get_start_last_synchronized_at(params):
+        start = params.get("start")
+        last_synchronized_at = utils.datetime_to_iso8601(
+            params.get("last_synchronized_at")
+        )
+        return start, last_synchronized_at
+
     @staticmethod
     def _check_use_account_routing(data_source_vo: DataSource) -> bool:
         plugin_info = data_source_vo.plugin_info.to_dict() or {}