[COST-5141] vaccum expired trino partitions

koku/masu/api/expired_data.py

@@ -36,3 +36,23 @@
     if simulate:
         response_key = response_key + " (simulated)"
     return Response({response_key: str(async_delete_results)})
+
+
+@never_cache
+@api_view(http_method_names=["GET", "DELETE"])
+@permission_classes((AllowAny,))
+@renderer_classes(tuple(api_settings.DEFAULT_RENDERER_CLASSES))
+def expired_trino_partitions(request):
+    """Return expired data."""
+    simulate = True
+    if request.method == "DELETE" and Config.DEBUG:
+        simulate = False
+    LOG.info("Simulate Flag: %s", simulate)
+
+    orchestrator = Orchestrator()
+    async_delete_results = orchestrator.remove_expired_trino_partitions(simulate=simulate)
+
+    response_key = "Async jobs for expired data removal"
+    if simulate:
+        response_key = response_key + " (simulated)"
+    return Response({response_key: str(async_delete_results)})

koku/masu/api/urls.py

@@ -21,6 +21,7 @@
 from masu.api.views import download_report
 from masu.api.views import EnabledTagView
 from masu.api.views import expired_data
+from masu.api.views import expired_trino_partitions
 from masu.api.views import explain_query
 from masu.api.views import fix_parquet
 from masu.api.views import get_status
@@ -58,6 +59,7 @@
     path("update_exchange_rates/", update_exchange_rates, name="update_exchange_rates"),
     path("update_azure_storage_capacity/", update_azure_storage_capacity, name="update_azure_storage_capacity"),
     path("enabled_tags/", EnabledTagView.as_view(), name="enabled_tags"),
+    path("expired_trino_partitions/", expired_trino_partitions, name="expired_trino_partitions"),
     path("expired_data/", expired_data, name="expired_data"),
     path("hcs_report_data/", hcs_report_data, name="hcs_report_data"),
     path("hcs_report_finalization/", hcs_report_finalization, name="hcs_report_finalization"),

koku/masu/api/views.py

@@ -18,6 +18,7 @@
 from masu.api.download import download_report
 from masu.api.enabled_tags import EnabledTagView
 from masu.api.expired_data import expired_data
+from masu.api.expired_data import expired_trino_partitions
 from masu.api.hcs_report_data import hcs_report_data
 from masu.api.hcs_report_finalization import hcs_report_finalization
 from masu.api.ingest_ocp_payload import ingest_ocp_payload

koku/masu/database/aws_report_db_accessor.py

@@ -26,6 +26,7 @@
 from masu.processor import is_feature_cost_3592_tag_mapping_enabled
 from reporting.models import OCP_ON_ALL_PERSPECTIVES
 from reporting.models import OCP_ON_AWS_PERSPECTIVES
+from reporting.models import OCP_ON_AWS_TEMP_MANAGED_TABLES
 from reporting.models import OCPAllCostLineItemDailySummaryP
 from reporting.models import OCPAllCostLineItemProjectDailySummaryP
 from reporting.models import OCPAWSCostLineItemProjectDailySummaryP
@@ -243,12 +244,7 @@ def populate_ocp_on_aws_cost_daily_summary_trino(
         days = self.date_helper.list_days(start_date, end_date)
         days_tup = tuple(str(day.day) for day in days)
         self.delete_ocp_on_aws_hive_partition_by_day(days_tup, aws_provider_uuid, openshift_provider_uuid, year, month)
-        tables = [
-            "reporting_ocpawscostlineitem_project_daily_summary_temp",
-            "aws_openshift_daily_resource_matched_temp",
-            "aws_openshift_daily_tag_matched_temp",
-        ]
-        for table in tables:
+        for table in OCP_ON_AWS_TEMP_MANAGED_TABLES:
             self.delete_hive_partition_by_month(table, openshift_provider_uuid, year, month)
 
         pod_column = "pod_effective_usage_cpu_core_hours"

koku/masu/database/azure_report_db_accessor.py

@@ -26,6 +26,7 @@
 from masu.processor import is_feature_cost_3592_tag_mapping_enabled
 from reporting.models import OCP_ON_ALL_PERSPECTIVES
 from reporting.models import OCP_ON_AZURE_PERSPECTIVES
+from reporting.models import OCP_ON_AZURE_TEMP_MANAGED_TABLES
 from reporting.models import OCPAllCostLineItemDailySummaryP
 from reporting.models import OCPAllCostLineItemProjectDailySummaryP
 from reporting.models import OCPAzureCostLineItemProjectDailySummaryP
@@ -296,12 +297,7 @@ def populate_ocp_on_azure_cost_daily_summary_trino(
         """Populate the daily cost aggregated summary for OCP on Azure."""
         year = start_date.strftime("%Y")
         month = start_date.strftime("%m")
-        tables = [
-            "reporting_ocpazurecostlineitem_project_daily_summary_temp",
-            "azure_openshift_daily_resource_matched_temp",
-            "azure_openshift_daily_tag_matched_temp",
-        ]
-        for table in tables:
+        for table in OCP_ON_AZURE_TEMP_MANAGED_TABLES:
             self.delete_hive_partition_by_month(table, openshift_provider_uuid, year, month)
         days = self.date_helper.list_days(start_date, end_date)
         days_tup = tuple(str(day.day) for day in days)

koku/masu/database/gcp_report_db_accessor.py

@@ -29,6 +29,7 @@
 from masu.processor import is_feature_cost_3592_tag_mapping_enabled
 from masu.util.gcp.common import check_resource_level
 from masu.util.ocp.common import get_cluster_alias_from_cluster_id
+from reporting.models import OCP_ON_GCP_TEMP_MANAGED_TABLES
 from reporting.provider.all.models import TagMapping
 from reporting.provider.gcp.models import GCPCostEntryBill
 from reporting.provider.gcp.models import GCPCostEntryLineItemDailySummary
@@ -322,12 +323,7 @@ def populate_ocp_on_gcp_cost_daily_summary_trino(
 
         year = start_date.strftime("%Y")
         month = start_date.strftime("%m")
-        tables = [
-            "reporting_ocpgcpcostlineitem_project_daily_summary_temp",
-            "gcp_openshift_daily_resource_matched_temp",
-            "gcp_openshift_daily_tag_matched_temp",
-        ]
-        for table in tables:
+        for table in OCP_ON_GCP_TEMP_MANAGED_TABLES:
             self.delete_hive_partition_by_month(table, openshift_provider_uuid, year, month)
 
         days = self.date_helper.list_days(start_date, end_date)

koku/masu/database/ocp_report_db_accessor.py

@@ -283,6 +283,26 @@
         LOG.info(log_json(msg="successfully deleted Hive partitions", context=ctx))
         return True
 
+    def find_expired_trino_partitions(self, table, source_column, date_str):
+        """Queries Trino for partitions less than the parition date."""
+        if not self.table_exists_trino(table):
+            LOG.info("Could not find table.")
+            return False
+        sql = f"""
+SELECT partitions.year, partitions.month, partitions.source
+FROM (
+    SELECT year as year,
+        month as month,
+        day as day,
+        cast(date_parse(concat(year, '-', month, '-', day), '%Y-%m-%d') as date) as partition_date,
+        {source_column} as source
+    FROM  "{table}$partitions"
+) as partitions
+WHERE partitions.partition_date < DATE '{date_str}'
+GROUP BY partitions.year, partitions.month, partitions.source
+"""
+        return self._execute_trino_raw_sql_query(sql, log_ref="finding expired partitions")
+
     def populate_line_item_daily_summary_table_trino(
         self, start_date, end_date, report_period_id, cluster_id, cluster_alias, source
     ):

koku/masu/database/report_db_accessor_base.py

@@ -221,7 +221,7 @@ def schema_exists_trino(self):
         check_sql = f"SHOW SCHEMAS LIKE '{self.schema}'"
         return bool(self._execute_trino_raw_sql_query(check_sql, log_ref="schema_exists_trino"))
 
-    def delete_hive_partition_by_month(self, table, source, year, month):
+    def delete_hive_partition_by_month(self, table, source, year, month, source_column="ocp_source"):
         """Deletes partitions individually by month."""
         retries = settings.HIVE_PARTITION_DELETE_RETRIES
         if self.schema_exists_trino() and self.table_exists_trino(table):
@@ -238,7 +238,7 @@ def delete_hive_partition_by_month(self, table, source, year, month):
                 try:
                     sql = f"""
                     DELETE FROM hive.{self.schema}.{table}
-                    WHERE ocp_source = '{source}'
+                    WHERE {source_column} = '{source}'
                     AND year = '{year}'
                     AND (month = replace(ltrim(replace('{month}', '0', ' ')),' ', '0') OR month = '{month}')
                     """

koku/masu/processor/_tasks/remove_expired.py

@@ -7,6 +7,7 @@
 
 from api.common import log_json
 from masu.processor.expired_data_remover import ExpiredDataRemover
+from masu.processor.expired_data_remover import ExpiredDataRemoverError
 
 LOG = logging.getLogger(__name__)
 
@@ -34,3 +35,38 @@
     if removed_data:
         status_msg = "Expired Data" if simulate else "Removed Data"
         LOG.info(log_json(msg=status_msg, removed_data=removed_data, context=context))
+    # We could extend the logic below here, or keep it as a separate celery task
+
+
+def _remove_expired_trino_partitions(schema_name, provider_type, simulate, provider_uuid=None):
+    """
+    Task to remove expired data.
+
+    Args:
+        schema_name (String) db schema name
+        provider    (String) provider type
+        simulate    (Boolean) Simulate report data removal
+
+    Returns:
+        None
+
+    """
+    log_statement = (
+        f"Remove expired data:\n"
+        f" schema_name: {schema_name}\n"
+        f" provider_type: {provider_type}\n"
+        f" simulate: {simulate}\n"
+    )
+    LOG.info(log_statement)
+
+    try:
+        remover = ExpiredDataRemover(schema_name, provider_type)
+    except ExpiredDataRemoverError:
+        return
+
+    removed_trino_partitions = remover.remove_expired_trino_partitions(simulate=simulate, provider_uuid=provider_uuid)
+    if removed_trino_partitions:
+        # TODO: Fix this log message
+        status_msg = "Expired Partitions" if simulate else "Removed Partitions"
+        result_msg = f"{status_msg}:\n {str(removed_trino_partitions)}"
+        LOG.info(result_msg)

koku/masu/processor/expired_data_remover.py

@@ -157,5 +157,18 @@
                         msg="Removed CostUsageReportManifest", provider=self._provider, expiration_date=expiration_date
                     )
                 )
-
         return removed_data
+
+    def remove_expired_trino_partitions(self, simulate=False, provider_uuid=None):
+        """
+        Removes expired trino partitions based on the retention policy.
+        """
+        if self._provider != Provider.PROVIDER_OCP:
+            LOG.info(f"{Provider.PROVIDER_OCP} is the only supported type for removing trino partitions.")
+            return
+        removed_partitions = []
+        expiration_date = self._calculate_expiration_date()
+        removed_partitions = self._cleaner.purge_expired_trino_partitions(
+            expired_date=expiration_date, simulate=simulate, provider_uuid=provider_uuid
+        )
+        return removed_partitions

koku/masu/processor/ocp/ocp_report_db_cleaner.py

@@ -12,6 +12,7 @@
 from koku.database import cascade_delete
 from koku.database import execute_delete_sql
 from masu.database.ocp_report_db_accessor import OCPReportDBAccessor
+from reporting.models import EXPIRE_MANAGED_TABLES
 from reporting.models import PartitionedTable
 from reporting.provider.ocp.models import UI_SUMMARY_TABLES
 
@@ -147,3 +148,28 @@
                 LOG.info(log_json(msg="deleted table partitions", count=del_count, schema=self._schema))
 
         return removed_items
+
+    def purge_expired_trino_partitions(self, expired_date, provider_uuid=None, simulate=False):
+        """Removes expired trino partitions."""
+        LOG.debug(f"purge_expired_trino_partitions: {expired_date}, {provider_uuid}, {simulate}")
+        if expired_date is None and provider_uuid is None:
+            err = "This method must be called with expired_date or provider_uuid"
+            raise OCPReportDBCleanerError(err)
+
+        with OCPReportDBAccessor(self._schema) as accessor:
+            LOG.info(EXPIRE_MANAGED_TABLES.items())
+            for table, source_column in EXPIRE_MANAGED_TABLES.items():
+                LOG.debug(f"{table}, {source_column}")
+                results = accessor.find_expired_trino_partitions(table, source_column, str(expired_date.date()))
+                if results:
+                    LOG.info(f"Discovered {len(results)} expired partitions")
+                else:
+                    LOG.info("No expired partitions")
+                    return
+                if simulate:
+                    for partition in results:
+                        LOG.info(f"partition_info: {partition}")
+                else:
+                    for result in results:
+                        year, month, source_value = result
+                        accessor.delete_hive_partition_by_month(table, source_value, year, month, source_column)

koku/masu/processor/orchestrator.py

@@ -33,6 +33,7 @@
 from masu.processor.tasks import record_all_manifest_files
 from masu.processor.tasks import record_report_status
 from masu.processor.tasks import remove_expired_data
+from masu.processor.tasks import remove_expired_trino_partitions
 from masu.processor.tasks import summarize_reports
 from masu.processor.worker_cache import WorkerCache
 from masu.util.aws.common import update_account_aliases
@@ -542,3 +543,31 @@
             )
             async_results.append({"customer": account.get("customer_name"), "async_id": str(async_result)})
         return async_results
+
+    def remove_expired_trino_partitions(self, simulate=False):
+        """
+        Removes expired trino partitions for each account.
+        """
+        async_results = []
+        schemas = set(
+            Provider.objects.filter(type=Provider.PROVIDER_OCP)
+            .values_list("customer__schema_name", flat=True)
+            .distinct()
+        )
+        # <QuerySet ['org1234567', 'org1234567', 'org1234567', 'org1234567']>
+        # distinct is not removing duplicates from this list, so using a set to reduce instead
+
+        for schema in schemas:
+            LOG.info("Calling remove_expired_trino_partitions with account: %s", schema)
+            async_result = remove_expired_trino_partitions.delay(
+                schema_name=schema,
+                provider_type=Provider.PROVIDER_OCP,
+                simulate=simulate,
+            )
+
+            LOG.info(
+                "Expired partition removal queued - schema_name: %s, Task ID: %s",
+                schema,
+                str(async_result),
+            )
+        return async_results

koku/masu/processor/tasks.py

@@ -50,6 +50,7 @@
 from masu.processor._tasks.download import _get_report_files
 from masu.processor._tasks.process import _process_report_file
 from masu.processor._tasks.remove_expired import _remove_expired_data
+from masu.processor._tasks.remove_expired import _remove_expired_trino_partitions
 from masu.processor.cost_model_cost_updater import CostModelCostUpdater
 from masu.processor.ocp.ocp_cloud_parquet_summary_updater import DELETE_TABLE
 from masu.processor.ocp.ocp_cloud_parquet_summary_updater import TRUNCATE_TABLE
@@ -310,6 +311,31 @@
     _remove_expired_data(schema_name, provider, simulate, provider_uuid)
 
 
+@celery_app.task(name="masu.processor.tasks.remove_expired_trino_partitions", queue=DEFAULT)
+def remove_expired_trino_partitions(schema_name, provider_type, simulate, provider_uuid=None, queue_name=None):
+    """
+    Remove expired report data.
+
+    Args:
+        schema_name (String) db schema name
+        provider    (String) provider type
+        simulate    (Boolean) Simulate report data removal
+
+    Returns:
+        None
+
+    """
+
+    context = {
+        "schema": schema_name,
+        "provider_type": provider_type,
+        "provider_uuid": provider_uuid,
+    }
+    LOG.info(log_json("remove_expired_data", msg="removing expired partitions", context=context))
+    _remove_expired_trino_partitions(schema_name, provider_type, simulate, provider_uuid)
+
+
+
 @celery_app.task(name="masu.processor.tasks.summarize_reports", queue=SummaryQueue.DEFAULT)  # noqa: C901
 def summarize_reports(  # noqa: C901
     reports_to_summarize, queue_name=None, manifest_list=None, ingress_report_uuid=None

koku/reporting/models.py

@@ -161,6 +161,7 @@
     OCPAzureDatabaseSummaryP,
 )
 
+# These are cleaned during source delete
 TRINO_MANAGED_TABLES = {
     "reporting_ocpusagelineitem_daily_summary": "source",
     "reporting_ocpawscostlineitem_project_daily_summary": "ocp_source",
@@ -176,3 +177,30 @@
     "gcp_openshift_daily_resource_matched_temp": "ocp_source",
     "gcp_openshift_daily_tag_matched_temp": "ocp_source",
 }
+
+# These are cleaned during expired_data flow
+EXPIRE_MANAGED_TABLES = {
+    "reporting_ocpusagelineitem_daily_summary": "source",
+    "reporting_ocpawscostlineitem_project_daily_summary": "ocp_source",
+    "reporting_ocpgcpcostlineitem_project_daily_summary": "ocp_source",
+    "reporting_ocpazurecostlineitem_project_daily_summary": "ocp_source",
+}
+
+# TEMP tables are cleaned during day to day processing
+OCP_ON_AWS_TEMP_MANAGED_TABLES = {
+    "reporting_ocpawscostlineitem_project_daily_summary_temp",
+    "aws_openshift_daily_resource_matched_temp",
+    "aws_openshift_daily_tag_matched_temp",
+}
+
+OCP_ON_AZURE_TEMP_MANAGED_TABLES = {
+    "reporting_ocpazurecostlineitem_project_daily_summary_temp",
+    "azure_openshift_daily_resource_matched_temp",
+    "azure_openshift_daily_tag_matched_temp",
+}
+
+OCP_ON_GCP_TEMP_MANAGED_TABLES = {
+    "reporting_ocpgcpcostlineitem_project_daily_summary_temp",
+    "gcp_openshift_daily_resource_matched_temp",
+    "gcp_openshift_daily_tag_matched_temp",
+}