Import executor clean up (#1903)

Author: vish-cs

import-automation/executor/app/executor/cloud_batch.py

@@ -141,28 +141,26 @@ def get_gce_instance(required_cpu: float,
     return suitable_instances[0]['name']
 
 
-def create_job_request(import_name: str, import_config: dict, import_spec: dict,
-                       default_resources: dict, timeout: int) -> str:
-    resources = import_spec.get('resource_limits', default_resources)
-    machine_type = get_gce_instance(resources['cpu'], resources['memory'])
+def create_job_request(import_name: str, import_config: dict,
+                       import_spec: dict) -> str:
+    resources = {}
+    if 'resource_limits' in import_spec:
+        resources = import_spec.get('resource_limits', {})
+        machine_type = get_gce_instance(resources.get('cpu', 0),
+                                        resources.get('memory', 0))
+        resources[
+            "machine"] = machine_type if machine_type is not None else 'n2-standard-8'
+        resources["cpu"] = resources["cpu"] * 1000
+        resources["memory"] = resources["memory"] * 1024
 
-    resources[
-        "machine"] = machine_type if machine_type is not None else 'n2-standard-8'
-
-    resources["cpu"] = resources["cpu"] * 1000
-    resources["memory"] = resources["memory"] * 1024
-    schedule = import_spec.get('cron_schedule')
     import_config_string = json.dumps(import_config)
-    job_name = import_name.split(':')[1]
-    job_name = job_name.replace("_", "-").lower()
     argument_payload = {
-        "jobName": job_name,
         "importName": import_name,
         "importConfig": import_config_string,
-        "resources": resources,
-        "timeout": timeout,
-        "schedule": schedule
     }
+    if resources:
+        argument_payload["resources"] = resources
+
     argument_string = json.dumps(argument_payload)
     final_payload = {
         "argument": argument_string,

import-automation/executor/app/executor/import_executor.py

@@ -47,6 +47,7 @@
 from timer import Timer
 from import_differ.import_differ import ImportDiffer
 from tools.import_validation.runner import ValidationRunner
+from tools.import_validation.result import ValidationStatus, ValidationResult
 from tools.import_validation.validation_config import merge_and_save_config
 from app import configs
 from app import utils
@@ -568,6 +569,7 @@ def _invoke_import_validation(self, repo_dir: str, relative_import_dir: str,
         data_size = 0
         import_name = import_spec['import_name']
         validation_status = True
+        validation_results = []
 
         import_dir = f'{relative_import_dir}/{import_spec["import_name"]}'
         latest_version = self._get_latest_version(import_dir)
@@ -650,7 +652,8 @@ def _invoke_import_validation(self, repo_dir: str, relative_import_dir: str,
                     stats_summary=summary_stats,
                     lint_report=report_json,
                     validation_output=validation_output_file)
-                overall_status, _ = validation.run_validations()
+                overall_status, current_results = validation.run_validations()
+                validation_results.extend(current_results)
                 if validation_status:
                     validation_status = overall_status
             except ValueError as e:
@@ -683,13 +686,26 @@ def _invoke_import_validation(self, repo_dir: str, relative_import_dir: str,
         import_summary.import_stats[
             'validation_execution_time'] = start_timer.time()
         import_summary.import_stats['validation_data_size'] = data_size
+        validation_message = self._get_validation_message(validation_results)
         log_import_status(
             import_name, import_stage,
             ImportStatus.SUCCESS if validation_status else ImportStatus.FAILURE,
             import_summary.import_stats.get('validation_execution_time', 0),
-            import_summary.import_stats.get('validation_data_size', 0))
+            import_summary.import_stats.get('validation_data_size',
+                                            0), validation_message)
         return validation_status
 
+    def _get_validation_message(
+            self, validation_results: List[ValidationResult]) -> str:
+        """Generates a summary message of validation results."""
+        failed_validations = []
+
+        for res in validation_results:
+            if res.status.name == 'FAILED':
+                failed_validations.append(res.name)
+
+        return f"FAILED: {', '.join(failed_validations) if failed_validations else 'None'}"
+
     def _create_mount_point(self, gcs_volume_mount_dir: str,
                             cleanup_gcs_volume_mount: bool,
                             absolute_import_dir: str, import_name: str) -> None:
@@ -824,10 +840,6 @@ def _import_one_helper(
         output_dir = f'{relative_import_dir}/{import_name}'
         version = self.config.import_version_override if self.config.import_version_override else _clean_time(
             utils.pacific_time())
-        # Used for imports using CDA feed tranfers with a date placeholder in the GCS path,
-        # thus, we can determine the path using the current date (instead of a variable timestamp).
-        if version == 'DATE_VERSION_PLACEHOLDER':
-            version = datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%d")
         import_summary.latest_version = 'gs://' + os.path.join(
             self.config.storage_prod_bucket_name, output_dir, version)
         import_summary.graph_path = self.config.graph_data_path

import-automation/executor/app/executor/scheduler_job_manager.py

@@ -158,8 +158,7 @@ def create_or_update_import_schedule(absolute_import_name: str,
         # This set up a cloud scheduler job which periodically invokes a GCP workflow job.
         # The workflow job runs a CLOUD BATCH job with the specified configuration in the request.
         json_encoded_body = cloud_batch.create_job_request(
-            absolute_import_name, override_config, import_spec, resources,
-            timeout)
+            absolute_import_name, override_config, import_spec)
         cloud_batch_job_url = f'https://workflowexecutions.googleapis.com/v1/projects/{config.gcp_project_id}/locations/{config.scheduler_location}/workflows/{config.cloud_workflow_id}/executions'
         req = cloud_scheduler.cloud_batch_job_request(
             absolute_import_name, schedule, cloud_batch_job_url,

import-automation/executor/cloud_batch_import_test.sh

@@ -16,7 +16,7 @@ mapfile -t IMPORT_LIST < <(gcloud scheduler jobs list --location us-central1 --p
 SCRIPT_DIR=$(realpath $(dirname $0))
 for import in "${IMPORT_LIST[@]}"; do
   echo "Scheduling import: $import"
-  $SCRIPT_DIR/schedule_update_import.sh -s $PROJECT $import
+  $SCRIPT_DIR/../schedule_update_import.sh -s $PROJECT $import
 done
 echo "Batch update finished."
 

import-automation/executor/cloud_run_import_test.sh

@@ -14,8 +14,6 @@
             "source_files": [],
             "cron_schedule": "15 3 * * *",
             "config_override": {
-                "import_version_override": "DATE_VERSION_PLACEHOLDER",
-                "graph_data_path": "/**/*mcf*",
                 "invoke_import_validation": false,
                 "invoke_import_tool": false,
                 "invoke_differ_tool": false
@@ -35,8 +33,6 @@
             "source_files": [],
             "cron_schedule": "15 3 * * 1",
             "config_override": {
-                "import_version_override": "DATE_VERSION_PLACEHOLDER",
-                "graph_data_path": "/**/*mcf*",
                 "invoke_import_validation": false,
                 "invoke_import_tool": false,
                 "invoke_differ_tool": false