Merge pull request #73 from OpenDataAlex/process_tracker_python-12

OpenDataAlex · web-flow · commit 97f085ae19fb · 2019-07-10T10:35:08.000-04:00
process_tracker_python-12 Handling Process Dependency failures
diff --git a/.gitignore b/.gitignore
@@ -173,3 +173,5 @@ pip-selfcheck.json
 .idea/misc.xml
 .idea/modules.xml
 .idea/process_tracker_python.iml
+/tests/s3:/
+/tests/s3:\/
diff --git a/configs/mysql_config.ini b/configs/mysql_config.ini
@@ -1,5 +1,6 @@
 [DEFAULT]
 log_level = ERROR
+max_sequential_failures = 5
 data_store_type = mysql
 data_store_username = pt_admin
 data_store_password = Testing1!
diff --git a/configs/postgres_config.ini b/configs/postgres_config.ini
@@ -1,5 +1,6 @@
 [DEFAULT]
 log_level = ERROR
+max_sequential_failures = 5
 data_store_type = postgresql
 data_store_username = pt_admin
 data_store_password = Testing1!
diff --git a/dbscripts/mysql_process_tracker_defaults.sql b/dbscripts/mysql_process_tracker_defaults.sql
@@ -11,6 +11,7 @@ INSERT INTO process_tracker.extract_status_lkup (extract_status_id, extract_stat
 INSERT INTO process_tracker.process_status_lkup (process_status_id, process_status_name) VALUES (default, 'running');
 INSERT INTO process_tracker.process_status_lkup (process_status_id, process_status_name) VALUES (default, 'completed');
 INSERT INTO process_tracker.process_status_lkup (process_status_id, process_status_name) VALUES (default, 'failed');
+INSERT INTO process_tracker.process_status_lkup (process_status_id, process_status_name) VALUES (default, 'on hold');
 
 INSERT INTO process_tracker.error_type_lkup (error_type_id, error_type_name) VALUES (default, 'File Error');
 INSERT INTO process_tracker.error_type_lkup (error_type_id, error_type_name) VALUES (default, 'Data Error');
diff --git a/dbscripts/postgresql_process_tracker_defaults.sql b/dbscripts/postgresql_process_tracker_defaults.sql
@@ -11,6 +11,8 @@ INSERT INTO process_tracker.extract_status_lkup (extract_status_id, extract_stat
 INSERT INTO process_tracker.process_status_lkup (process_status_id, process_status_name) VALUES (default, 'running');
 INSERT INTO process_tracker.process_status_lkup (process_status_id, process_status_name) VALUES (default, 'completed');
 INSERT INTO process_tracker.process_status_lkup (process_status_id, process_status_name) VALUES (default, 'failed');
+INSERT INTO process_tracker.process_status_lkup (process_status_id, process_status_name) VALUES (default, 'on hold');
+
 
 INSERT INTO process_tracker.error_type_lkup (error_type_id, error_type_name) VALUES (default, 'File Error');
 INSERT INTO process_tracker.error_type_lkup (error_type_id, error_type_name) VALUES (default, 'Data Error');
diff --git a/process_tracker/process_tracker.py b/process_tracker/process_tracker.py
@@ -82,9 +82,8 @@ def __init__(
         :type dataset_types: list
         """
         self.config_location = config_location
-        log_level = SettingsManager(
-            config_location=self.config_location
-        ).determine_log_level()
+        self.config = SettingsManager(config_location=self.config_location)
+        log_level = self.config.determine_log_level()
 
         self.logger = logging.getLogger(__name__)
         self.logger.setLevel(log_level)
@@ -144,6 +143,7 @@ def __init__(
         self.process_status_running = self.process_status_types["running"]
         self.process_status_complete = self.process_status_types["completed"]
         self.process_status_failed = self.process_status_types["failed"]
+        self.process_status_hold = self.process_status_types["on hold"]
 
         self.process_tracking_run = self.register_new_process_run()
 
@@ -206,6 +206,61 @@ def change_run_status(self, new_status, end_date=None):
         else:
             raise Exception("The provided status type %s is invalid." % new_status)
 
+    def determine_hold_status(self, last_run_status, last_run_id):
+        """
+        Based on the setting 'max_concurrent_failures', count the number of failures for that number of process runs.
+        If the counts match, process will remain on hold.  If last run is 'on_hold' process will remain on hold.
+        :param last_run_status: The status of the previous run
+        :param last_run_id:  The process_run_id of the previous run
+        :return:
+        """
+        self.logger.debug("Determining if process should be put on or remain on hold.")
+
+        max_concurrent_failures = int(
+            self.config.config["DEFAULT"]["max_sequential_failures"]
+        )
+
+        self.logger.debug("Max Concurrent failures is %s" % max_concurrent_failures)
+        # last_runs = (
+        #     self.session.query(ProcessTracking.process_tracking_id)
+        #     .join(Process)
+        #     .filter(Process.process_name == self.process_name)
+        #     .order_by(ProcessTracking.process_run_id.desc())
+        #     .limit(max_concurrent_failures)
+        #     .subquery()
+        # )
+
+        failure_count = (
+            self.session.query(ProcessTracking)
+            .join(Process)
+            .filter(Process.process_name == self.process_name)
+            .filter(
+                ProcessTracking.process_run_id > (last_run_id - max_concurrent_failures)
+            )
+            .filter(ProcessTracking.process_status_id == self.process_status_failed)
+            .count()
+        )
+
+        # failure_count = (
+        #     self.session.query(ProcessTracking)
+        #     .filter(ProcessTracking.process_tracking_id.in_(last_runs))
+        #     .filter(ProcessTracking.process_status_id == self.process_status_failed)
+        #     .count()
+        # )
+
+        self.logger.debug("Number of failures in past runs is %s" % failure_count)
+
+        if last_run_status == self.process_status_hold:
+            self.logger.error("Last run still in hold status.  Need to remain in hold.")
+            return True
+        elif failure_count == max_concurrent_failures:
+            self.logger.error(
+                "Number of failures has reached max_concurrent_failures.  Putting process on hold until resolved."
+            )
+            return True
+        else:
+            return False
+
     def find_extracts_by_filename(self, filename, status="ready"):
         """
         For the given filename, or filename part, find all matching extracts that are ready for processing.
@@ -326,9 +381,6 @@ def get_latest_tracking_record(self, process):
             .first()
         )
 
-        if instance is None:
-            return False
-
         return instance
 
     def get_process_status_types(self):
@@ -480,16 +532,28 @@ def register_new_process_run(self):
                 "Processes that this process is dependent on are running or failed."
             )
 
-        if last_run:
+        if last_run is not None and last_run:
             # Must validate that the process is not currently running.
 
-            if last_run.process_status_id != self.process_status_running:
+            if (
+                last_run.process_status_id != self.process_status_running
+                and last_run.process_status_id != self.process_status_hold
+            ):
                 last_run.is_latest_run = False
                 new_run_flag = True
                 new_run_id = last_run.process_run_id + 1
             else:
                 new_run_flag = False
 
+            if self.determine_hold_status(
+                last_run_status=last_run.process_status_id,
+                last_run_id=last_run.process_run_id,
+            ):
+                self.logger.error(
+                    "Process is on hold due to number of concurrent failures or previous run is in on hold status."
+                )
+                new_run_flag = False
+
         if new_run_flag:
             new_run = ProcessTracking(
                 process_id=self.process.process_id,
@@ -508,7 +572,9 @@ def register_new_process_run(self):
             return new_run
 
         else:
-            raise Exception("The process %s is currently running." % self.process_name)
+            raise Exception(
+                "The process %s is currently running or on hold." % self.process_name
+            )
 
     def register_process_dataset_types(self, dataset_types):
         """
diff --git a/tests/test_process_tracker.py b/tests/test_process_tracker.py
@@ -114,6 +114,28 @@ def tearDown(self):
         self.session.query(ErrorType).delete()
         self.session.commit()
 
+    def process_run_setup(self, process_name, status, num_runs):
+        """
+        Helper function to setup mutliple process tracking runs for a given process and set to a given status. Used to
+        test the on_hold status logic.
+        :param process_name: Name of the process that runs will be created of.
+        :param status: Status the runs should be changed to
+        :param num_runs: Number of runs that should be created
+        :return:
+        """
+        i = 1
+        while i <= num_runs:
+            process_run = ProcessTracker(
+                process_name=process_name,
+                process_type="Extract",
+                actor_name="UnitTesting",
+                tool_name="Spark",
+            )
+
+            process_run.change_run_status(new_status=status)
+            i += 1
+            time.sleep(2)
+
     def test_bulk_change_extract_status(self):
         """
         Testing that bulk change occurs when extracts provided.
@@ -644,6 +666,90 @@ def test_initializing_process_tracking(self):
 
         self.assertEqual(expected_result, given_result)
 
+    @unittest.skip("Issue with hanging queries on database.")
+    def test_process_on_hold_max_failures(self):
+        """
+        Testing that when number of failed processes matches the maximum_sequential_failures (default 5), process run
+        goes on_hold.
+        :return:
+        """
+
+        self.process_run_setup(
+            process_name="On Hold Max Failures Test", status="failed", num_runs=5
+        )
+
+        with self.assertRaises(Exception) as context:
+            ProcessTracker(
+                process_name="On Hold Max Failures Test",
+                process_type="Extract",
+                actor_name="UnitTesting",
+                tool_name="Spark",
+            )
+
+        self.assertTrue(
+            "The process On Hold Max Failures Test is currently running or on_hold."
+            in str(context.exception)
+        )
+
+    @unittest.skip("Issue with hanging queries on database.")
+    def test_process_on_hold_under_max_failures(self):
+        """
+        Testing that when number of failed processes is less than the maximum_sequential_failures (default 5), process run
+        continues.
+        :return:
+        """
+        process_name = "On Hold Under Max Failures Test"
+        self.process_run_setup(process_name=process_name, status="failed", num_runs=3)
+
+        process_run = ProcessTracker(
+            process_name=process_name,
+            process_type="Extract",
+            actor_name="UnitTesting",
+            tool_name="Spark",
+        )
+
+        current_run_status = (
+            self.session.query(ProcessTracking)
+            .join(Process)
+            .filter(Process.process_name == process_name)
+            .filter(ProcessTracking.is_latest_run == True)
+        )
+        given_result = current_run_status[0].process_status_id
+
+        expected_result = process_run.process_status_running
+
+        self.assertEqual(expected_result, given_result)
+
+    def test_process_on_hold_previous_run_on_hold(self):
+        """
+        If the previous run does not get moved from on_hold status, then the next run will not kick off and the process
+        will remain on_hold.
+        :return:
+        """
+        process_name = "On Hold Previous Run Test"
+
+        process_run = ProcessTracker(
+            process_name=process_name,
+            process_type="Extract",
+            actor_name="UnitTesting",
+            tool_name="Spark",
+        )
+
+        process_run.change_run_status(new_status="on hold")
+
+        with self.assertRaises(Exception) as context:
+            ProcessTracker(
+                process_name=process_name,
+                process_type="Extract",
+                actor_name="UnitTesting",
+                tool_name="Spark",
+            )
+
+        self.assertTrue(
+            "The process On Hold Previous Run Test is currently running or on hold."
+            in str(context.exception)
+        )
+
     def test_register_extracts_by_location_local_file_count(self):
         """
         Testing that when the location is local, all the extracts are counted and registered in the location's file count.
@@ -865,8 +971,8 @@ def test_register_new_process_run_exception(self):
             self.process_tracker.register_new_process_run()
 
         return self.assertTrue(
-            "The process Testing Process Tracking Initialization "
-            "is currently running." in str(context.exception)
+            "The process Testing Process Tracking Initialization is currently running or on hold."
+            in str(context.exception)
         )
 
     def test_register_new_process_run_with_previous_run(self):
diff --git a/tests/utilities/test_utilities.py b/tests/utilities/test_utilities.py
@@ -26,7 +26,7 @@ def test_determine_low_high_date_invalid_date_type(self):
             utilities.determine_low_high_date(
                 date=lower_low_date, previous_date=low_date, date_type="blarg"
             )
-        print(context.exception)
+
         return self.assertTrue(
             "blarg is not a valid date_type." in str(context.exception)
         )

Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@ def test_determine_low_high_date_invalid_date_type(self):`
`26`	`26`	`utilities.determine_low_high_date(`
`27`	`27`	`date=lower_low_date, previous_date=low_date, date_type="blarg"`
`28`	`28`	`)`
`29`		`- print(context.exception)`
	`29`	`+`
`30`	`30`	`return self.assertTrue(`
`31`	`31`	`"blarg is not a valid date_type." in str(context.exception)`
`32`	`32`	`)`