CUR proxy + CUR 2.0 - Major v4.0.0

cid/builtin/core/data/resources.yaml

@@ -268,6 +268,8 @@ views:
     File: cid/summary_view.sql
     dependsOn:
       cur:
+        - year
+        - month
         - bill_billing_entity
         - bill_billing_period_start_date
         - bill_invoice_id
@@ -332,7 +334,7 @@ views:
   compute_savings_plan_eligible_spend:
     File: cid/compute_savings_plan_eligible_spend.sql
     dependsOn:
-      cur: true
+      cur:
         - bill_billing_period_start_date
         - bill_payer_account_id
         - line_item_line_item_type
@@ -380,9 +382,9 @@ views:
         - savings_plan_savings_plan_a_r_n
 
   hourly_view:
-    spriFile: cudos/hourly_view_sp_ri.sql
-    spFile: cudos/hourly_view_sp.sql
-    riFile: cudos/hourly_view_ri.sql
+    #spriFile: cudos/hourly_view_sp_ri.sql
+    #spFile: cudos/hourly_view_sp.sql
+    #riFile: cudos/hourly_view_ri.sql
     File: cudos/hourly_view.sql
     dependsOn:
       cur:

cid/common.py

@@ -22,7 +22,7 @@
 from cid.plugin import Plugin
 from cid.utils import get_parameter, get_parameters, set_parameters, unset_parameter, get_yesno_parameter, cid_print, isatty, merge_objects, IsolatedParameters
 from cid.helpers.account_map import AccountMap
-from cid.helpers import Athena, S3, IAM, CUR, Glue, QuickSight, Dashboard, Dataset, Datasource, csv2view, Organizations
+from cid.helpers import Athena, S3, IAM, CUR, ProxyCUR, Glue, QuickSight, Dashboard, Dataset, Datasource, csv2view, Organizations
 from cid.helpers.quicksight.template import Template as CidQsTemplate
 from cid._version import __version__
 from cid.export import export_analysis
@@ -110,6 +110,14 @@ def organizations(self) -> Organizations:
     def s3(self) -> S3:
         return S3(self.base.session)
 
+    @cached_property
+    def cur1(self):
+        return ProxyCUR(self.cur, target_cur_version='1')
+
+    @cached_property
+    def cur2(self):
+        return ProxyCUR(self.cur, target_cur_version='2')
+
     @property
     def cur(self) -> CUR:
         if not self._clients.get('cur'):
@@ -131,6 +139,7 @@ def cur(self) -> CUR:
                     self._clients['cur'] = _cur
                     break
                 except CidCritical as exc:
+                    logger.exception(exc)
                     cid_print(f'CUR not found in {self.athena.DatabaseName}. If you have S3 bucket with CUR in this account you can create a CUR table with Crawler.')
                     self.create_cur_table()
         return self._clients['cur']
@@ -141,7 +150,6 @@ def accountMap(self) -> AccountMap:
             _account_map = AccountMap(self.base.session)
             _account_map.athena = self.athena
             _account_map.cur = self.cur
-
             self._clients.update({
                 'accountMap': _account_map
             })
@@ -1356,6 +1364,7 @@ def create_or_update_dataset(self, dataset_definition: dict, dataset_id: str=Non
         data = self.get_data_from_definition('dataset', dataset_definition)
         template = Template(json.dumps(data))
         cur_required = dataset_definition.get('dependsOn', dict()).get('cur')
+        cur2_required = dataset_definition.get('dependsOn', dict()).get('cur2')
         athena_datasource = None
 
         # Manage datasource
@@ -1447,7 +1456,8 @@ def create_or_update_dataset(self, dataset_definition: dict, dataset_id: str=Non
 
         # Check for required views
         _views = dataset_definition.get('dependsOn', {}).get('views', [])
-        required_views = [(self.cur.table_name if cur_required and name =='${cur_table_name}' else name) for name in _views]
+        #FIXME: delete this : required_views = [(self.cur.table_name if cur_required and name =='${cur_table_name}' else name) for name in _views]
+        required_views = _views
 
         self.athena.discover_views(required_views)
         found_views = utils.intersection(required_views, self.athena._metadata.keys())
@@ -1456,9 +1466,9 @@ def create_or_update_dataset(self, dataset_definition: dict, dataset_id: str=Non
         if recursive:
             print(f"Detected views: {', '.join(found_views)}")
             for view_name in found_views:
-                if cur_required and view_name == self.cur.table_name:
-                    logger.debug(f'Dependency view {view_name} is a CUR. Skip.')
-                    continue
+                #if cur_required and view_name == self.cur.table_name:
+                #    logger.debug(f'Dependency view {view_name} is a CUR. Skip.')
+                #    continue
                 if view_name == 'account_map':
                     logger.debug(f'Dependency view is {view_name}. Skip.')
                     continue
@@ -1477,7 +1487,8 @@ def create_or_update_dataset(self, dataset_definition: dict, dataset_id: str=Non
         columns_tpl = {
             'athena_datasource_arn': athena_datasource.arn,
             'athena_database_name': self.athena.DatabaseName,
-            'cur_table_name': self.cur.table_name if cur_required else None
+            'cur_table_name': self.cur1.table_name if cur_required else None,
+            'cur2_table_name': self.cur2.table_name if cur2_required else None,
         }
 
         logger.debug(f'dataset_id={dataset_id}')
@@ -1507,7 +1518,7 @@ def create_or_update_dataset(self, dataset_definition: dict, dataset_id: str=Non
             elif found_dataset.name != compiled_dataset.get('Name'):
                 print(f"Dataset found with name {found_dataset.name}, but {compiled_dataset.get('Name')} expected. Updating.")
                 update_dataset = True
-            if update_dataset and get_parameters().get('on-drift', 'show').lower() != 'override' and isatty() and not cur_required:
+            if update_dataset and get_parameters().get('on-drift', 'show').lower() != 'override' and isatty() and not cur_required and not cur2_required:
                 while True:
                     diff = self.qs.dataset_diff(found_dataset.raw, compiled_dataset)
                     if diff and diff['diff']:
@@ -1588,17 +1599,17 @@ def create_or_update_view(self, view_name: str, recursive: bool=True, update: bo
         dependencies = view_definition.get('dependsOn', {})
 
         # Process CUR columns
-        if isinstance(dependencies.get('cur'), list):
-            for column in dependencies.get('cur'):
-                self.cur.ensure_column(column)
-        elif isinstance(dependencies.get('cur'), dict):
-            for column, column_type in dependencies.get('cur').items():
-                self.cur.ensure_column(column, column_type)
+        if dependencies.get('cur'):
+            self.cur1.ensure_columns(dependencies.get('cur'))
+        if dependencies.get('cur2'):
+            self.cur2.ensure_columns(dependencies.get('cur2'))
 
         if recursive:
             dependency_views = dependencies.get('views', [])
             if 'cur' in dependency_views:
                 dependency_views.remove('cur')
+            if 'cur2' in dependency_views:
+                dependency_views.remove('cur2')
             # Discover dependency views (may not be discovered earlier)
             self.athena.discover_views(dependency_views)
             logger.info(f"Dependency views: {', '.join(dependency_views)}" if dependency_views else 'No dependency views')
@@ -1727,17 +1738,18 @@ def get_view_query(self, view_name: str) -> str:
         # View path
         view_definition = self.get_definition("view", name=view_name)
         cur_required = view_definition.get('dependsOn', dict()).get('cur')
-        if cur_required and self.cur.has_savings_plans and self.cur.has_reservations and view_definition.get('spriFile'):
-            view_definition['File'] = view_definition.get('spriFile')
-        elif cur_required and self.cur.has_savings_plans and view_definition.get('spFile'):
-            view_definition['File'] = view_definition.get('spFile')
-        elif cur_required and self.cur.has_reservations and view_definition.get('riFile'):
-            view_definition['File'] = view_definition.get('riFile')
-        elif view_definition.get('File') or view_definition.get('Data') or view_definition.get('data'):
-            pass
-        else:
-            logger.critical(f'\nCannot find view {view_name}. View information is incorrect, please check resources.yaml')
-            raise Exception(f'\nCannot find view {view_name}')
+        cur2_required = view_definition.get('dependsOn', dict()).get('cur2')
+        #if cur_required and self.cur.has_savings_plans and self.cur.has_reservations and view_definition.get('spriFile'):
+        #    view_definition['File'] = view_definition.get('spriFile')
+        #elif cur_required and self.cur.has_savings_plans and view_definition.get('spFile'):
+        #    view_definition['File'] = view_definition.get('spFile')
+        #elif cur_required and self.cur.has_reservations and view_definition.get('riFile'):
+        #    view_definition['File'] = view_definition.get('riFile')
+        #if view_definition.get('File') or view_definition.get('Data') or view_definition.get('data'):
+        #    pass
+        #else:
+        #    logger.critical(f'\nCannot find view {view_name}. View information is incorrect, please check resources.yaml')
+        #    raise Exception(f'\nCannot find view {view_name}')
 
         # Load TPL file
         data = self.get_data_from_definition('view', view_definition)
@@ -1748,7 +1760,8 @@ def get_view_query(self, view_name: str) -> str:
 
         # Prepare template parameters
         columns_tpl = {
-            'cur_table_name': self.cur.table_name if cur_required else None,
+            'cur_table_name': self.cur1.table_name if cur_required else None,
+            'cur2_table_name': self.cur2.table_name if cur2_required else None,
             'athenaTableName': view_name,
             'athena_database_name': self.athena.DatabaseName,
         }

cid/helpers/__init__.py

@@ -2,17 +2,19 @@
 from cid.helpers.s3 import S3
 from cid.helpers.athena import Athena
 from cid.helpers.iam import IAM
-from cid.helpers.cur import CUR
+from cid.helpers.cur import CUR, ProxyCUR
 from cid.helpers.diff import diff
 from cid.helpers.quicksight import QuickSight, Dashboard, Dataset, Datasource, Template
 from cid.helpers.csv2view import csv2view
 from cid.helpers.organizations import Organizations
+from cid.helpers.cur_proxy import ProxyView
 
 __all__ = [
     "Athena",
     "S3",
     "IAM",
     "CUR",
+    "ProxyCUR",
     "Glue",
     "QuickSight",
     "Dashboard",
@@ -22,4 +24,5 @@
     "diff",
     "csv2view",
     "Organizations",
+    "ProxyView",
 ]

cid/helpers/account_map.py

@@ -107,7 +107,7 @@ def accounts(self) -> dict:
     @lru_cache(1000)
     def detect_metadata_table(self, name):
         """ detect meta table with the list of accounts """
-        cid_print('Autodiscovering metadata table')
+        cid_print('Autodiscover metadata table')
 
         # FIXME: This will only work for current Athena Database. We might want to check optimization_data base as well
         tables = self.athena.list_table_metadata()
@@ -142,19 +142,35 @@ def create(self, name) -> bool:
             if get_parameters().get('account-map-source'):
                 raise CidError('Skipping autodiscovery')
 
+            # try to find a table with data about accounts
             self._athena_table_name = self.detect_metadata_table(name)
             if not self._athena_table_name:
-                raise CidError('Metadata table not found')
-
-            # Query path
-            view_definition = self.athena._resources.get('views').get(name, {})
-            if view_definition.get('File'):
-                view_file = view_definition.get('File')
-                template = Template(resource_string(view_definition.get('providedBy'), f'data/queries/{view_file}').decode('utf-8'))
-            elif view_definition.get('data'):
-                template = Template(str(view_definition.get('data')))
+                raise CidError('Metadata table not found') # catch you later
+
+            if self.cur.version.startswith('2'):
+                template_str = '''
+                    CREATE OR REPLACE VIEW  ${athena_view_name} AS
+                    SELECT DISTINCT
+                        line_item_usage_account_id                                        account_id,
+                        MAX_BY(bill_payer_account_id,      line_item_usage_start_date)    parent_account_id,
+                        MAX_BY(line_item_usage_account_name, line_item_usage_start_date)  account_name,
+                        MAX_BY(bill_payer_account_name, line_item_usage_start_date)       parent_account_name
+                    FROM
+                        "${cur_table_name}"
+                    GROUP BY
+                        line_item_usage_account_id
+                '''
+                template = Template(template_str)
             else:
-                raise CidError(f'{name} definition does not contain File or data: {view_definition}')
+                # Query path
+                view_definition = self.athena._resources.get('views').get(name, {})
+                if view_definition.get('File'):
+                    view_file = view_definition.get('File')
+                    template = Template(resource_string(view_definition.get('providedBy'), f'data/queries/{view_file}').decode('utf-8'))
+                elif view_definition.get('data'):
+                    template = Template(str(view_definition.get('data')))
+                else:
+                    raise CidError(f'{name} definition does not contain File or data: {view_definition}')
 
             # Fill in TPLs
             columns_tpl = {

cid/helpers/athena.py

@@ -312,7 +312,7 @@ def execute_query(self, sql_query, sleep_duration=1, database: str=None, catalog
             raise CidCritical(f'InvalidRequestException: {exc}') from exc
         except Exception as exc:
             logger.debug(f'Full query: {sql_query}')
-            raise CidCritical(f'Athena query failed: {exc}') from exc
+            raise CidCritical(f'Query:\n{sql_query}\n\nAthena query failed: {exc}') from exc
 
         current_status = query_status['QueryExecution']['Status']['State']
 
@@ -330,7 +330,7 @@ def execute_query(self, sql_query, sleep_duration=1, database: str=None, catalog
         logger.info(f'Athena query failed: {failure_reason}')
         logger.debug(f'Full query: {sql_query}')
         if fail:
-            raise CidCritical(f'Athena query failed: {failure_reason}')
+            raise CidCritical(f'Query:\n{sql_query}\n\nAthena query status failed : {failure_reason}')
         return False
 
     def get_query_results(self, query_id):