add clean old data schedule task

Author: AlexCXC

scheduler/app/faas_scheduler/models.py

@@ -57,6 +57,15 @@ def __init__(
         self.operate_from = operate_from
         self.state = state
 
+    def get_info(self):
+        return {
+            'id': self.id,
+            'org_id': self.org_id,
+            'owner': self.owner,
+            'dtable_uuid': self.dtable_uuid,
+            'script_name': self.script_name
+        }
+
     def to_dict(self):
         from faas_scheduler.utils import datetime_to_isoformat_timestr
 

scheduler/app/flask_server.py

@@ -131,7 +131,7 @@ def script_api(script_id):
         script = get_script(db_session, script_id)
         if not script:
             return make_response(("Not found", 404))
-        if dtable_uuid != script.dtable_uuid or script_name != script.script_name:
+        if uuid_str_to_32_chars(dtable_uuid) != script.dtable_uuid or script_name != script.script_name:
             return make_response(("Bad request", 400))
 
         if SUB_PROCESS_TIMEOUT and isinstance(SUB_PROCESS_TIMEOUT, int):
@@ -277,21 +277,19 @@ def record_script_result():
     output = data.get("output")
     spend_time = data.get("spend_time")
     script_id = data.get("script_id")
-
-    db_session = DBSession()
-
     # update script_log and run-time statistics
     try:
         if script_id:
-            hook_update_script(
-                db_session, script_id, success, return_code, output, spend_time
+            # hook_update_script(
+            #     db_session, script_id, success, return_code, output, spend_time
+            # )
+            scheduler.script_done_callback(
+                script_id, success, return_code, output, spend_time
             )
 
     except Exception as e:
         logger.exception(e)
         return make_response(("Internal server error", 500))
-    finally:
-        db_session.close()
 
     return "success"
 
@@ -376,5 +374,5 @@ def base_run_python_statistics():
 
 if __name__ == "__main__":
     scheduler.start()
-    http_server = WSGIServer(("127.0.0.1", 5055), app)
+    http_server = WSGIServer(("0.0.0.0", 5055), app)
     http_server.serve_forever()

scheduler/app/scheduler.py

@@ -2,18 +2,23 @@
 import logging
 import os
 import time
+from datetime import datetime
 from threading import Lock, Thread
 
 from database import DBSession
 from faas_scheduler.models import ScriptLog
 from faas_scheduler.utils import (
     add_script,
+    delete_log_after_days,
+    delete_statistics_after_days,
     run_script,
     get_script_file,
     hook_update_script
 )
 
 logger = logging.getLogger(__name__)
+SUB_PROCESS_TIMEOUT = int(os.environ.get("PYTHON_PROCESS_TIMEOUT", 60 * 15))
+TIMEOUT_OUTPUT = "Script running for too long time!"
 
 
 class ScriptQueue:
@@ -68,6 +73,7 @@ def add_script_log(self, script_log: ScriptLog):
         with self.lock:
             self.q.append(script_log)
             self.script_logs_dict[script_log.id] = script_log
+            self.inspect_queue_and_running(pre_msg=f'add script {script_log.get_info()} to queue')
 
     def get(self):
         """get the first valid task from self.q
@@ -83,25 +89,83 @@ def get(self):
                 if self.can_run_script(script_log):
                     return_task = script_log
                     self.q.pop(index)
+                    self.increase_running(script_log)
+                    self.inspect_queue_and_running(pre_msg=f'get script {script_log.get_info()} from queue')
                     break
                 index += 1
 
             return return_task
 
+    def increase_running(self, script_log):
+        if script_log.org_id != -1:
+            running_team_key = f'{script_log.org_id}'
+        else:
+            running_team_key = f'{script_log.owner}'
+        running_base_key = f'{running_team_key}_{script_log.dtable_uuid}'
+        running_script_key = f'{running_base_key}_{script_log.script_name}'
+        self.running_count[running_team_key] = self.running_count[running_team_key] + 1 if self.running_count.get(running_team_key) else 1
+        self.running_count[running_base_key] = self.running_count[running_base_key] + 1 if self.running_count.get(running_base_key) else 1
+        self.running_count[running_script_key] = self.running_count[running_script_key] + 1 if self.running_count.get(running_script_key) else 1
+
+    def decrease_running(self, script_log):
+        if script_log.org_id != -1:
+            running_team_key = f'{script_log.org_id}'
+        else:
+            running_team_key = f'{script_log.owner}'
+        running_base_key = f'{running_team_key}_{script_log.dtable_uuid}'
+        running_script_key = f'{running_base_key}_{script_log.script_name}'
+
+        if running_team_key in self.running_count:
+            self.running_count[running_team_key] -= 1
+        if not self.running_count.get(running_team_key):
+            self.running_count.pop(running_team_key, None)
+
+        if running_base_key in self.running_count:
+            self.running_count[running_base_key] -= 1
+        if not self.running_count.get(running_base_key):
+            self.running_count.pop(running_base_key, None)
+
+        if running_script_key in self.running_count:
+            self.running_count[running_script_key] -= 1
+        if not self.running_count.get(running_script_key):
+            self.running_count.pop(running_script_key, None)
+
     def script_done_callback(self, script_log: ScriptLog):
         with self.lock:
-            if script_log.org_id != -1:
-                running_team_key = f'{script_log.org_id}'
-            else:
-                running_team_key = f'{script_log.owner}'
-            running_base_key = f'{running_team_key}_{script_log.dtable_uuid}'
-            running_script_key = f'{running_base_key}_{script_log.script_name}'
-            if running_team_key in self.running_count:
-                self.running_count[running_team_key] -= 1
-            if running_base_key in self.running_count:
-                self.running_count[running_base_key] -= 1
-            if running_script_key in self.running_count:
-                self.running_count[running_script_key] -= 1
+            self.script_logs_dict.pop(script_log.id, None)
+            self.decrease_running(script_log)
+            self.inspect_queue_and_running(pre_msg=f'script {script_log.get_info()} run done')
+
+    def inspect_queue_and_running(self, pre_msg=None):
+        if logger.root.level != logging.DEBUG:
+            return
+        lines = ['\n']
+        if pre_msg:
+            lines.append(pre_msg)
+        lines.append(f"{'>' * 10} running {'>' * 10}")
+        for key, value in self.running_count.items():
+            lines.append(f'{key}: {value}')
+        lines.append(f"{'<' * 10} running {'<' * 10}")
+
+        lines.append(f"{'>' * 10} queue {'>' * 10}")
+        for script_log in self.q:
+            lines.append(f"org_id: {script_log.org_id} owner: {script_log.owner} dtable_uuid: {script_log.dtable_uuid} script_name: {script_log.script_name}")
+        lines.append(f"{'<' * 10} queue {'<' * 10}")
+        logger.debug('\n'.join(lines))
+
+    def get_script_log_by_id(self, script_id):
+        return self.script_logs_dict.get(script_id)
+
+    def get_timeout_scripts(self):
+        script_logs = []
+        now_time = datetime.now()
+        with self.lock:
+            for index in range(len(self.q) - 1, -1, -1):
+                script_log = self.q[index]
+                if (now_time - script_log.started_at).seconds >= SUB_PROCESS_TIMEOUT:
+                    script_logs.append(self.q.pop(index))
+                    self.script_logs_dict.pop(script_log.id, None)
+        return script_logs
 
 
 class Scheduelr:
@@ -157,7 +221,6 @@ def script_done_callback(
             output,
             spend_time
         ):
-        script_log = self.script_queue.script_logs_dict.pop(script_id)
         hook_update_script(
             DBSession(),
             script_id,
@@ -166,8 +229,10 @@ def script_done_callback(
             output,
             spend_time
         )
+        script_log = self.script_queue.get_script_log_by_id(script_id)
         if not script_log:  # not counted in memory, only update db record
             return
+        self.script_queue.script_done_callback(script_log)
 
     def load_pending_script_logs(self):
         """load pending script logs, should be called only when server start
@@ -176,9 +241,43 @@ def load_pending_script_logs(self):
         for script_log in script_logs:
             self.script_queue.add_script_log(script_log)
 
+    def timeout_setter(self):
+        while True:
+            db_session = DBSession()
+            now_time = datetime.now()
+            try:
+                script_logs = self.script_queue.get_timeout_scripts()
+                if script_logs:
+                    db_session.query(ScriptLog).filter(ScriptLog.id.in_([script_log.id for script_log in script_logs])).update(
+                        {
+                            ScriptLog.state: ScriptLog.FINISHED,
+                            ScriptLog.finished_at: now_time,
+                            ScriptLog.success: False,
+                            ScriptLog.output: TIMEOUT_OUTPUT,
+                            ScriptLog.return_code: -1
+                        }
+                    )
+            except Exception as e:
+                logger.exception(e)
+            finally:
+                DBSession.remove()
+            time.sleep(60)
+
+    def statistic_cleaner(self):
+        while True:
+            db_session = DBSession()
+            try:
+                delete_log_after_days(db_session)
+                delete_statistics_after_days(db_session)
+            except Exception as e:
+                logger.exception(e)
+            time.sleep(24 * 60 * 60)
+
     def start(self):
         self.load_pending_script_logs()
         Thread(target=self.schedule, daemon=True).start()
+        Thread(target=self.statistic_cleaner, daemon=True).start()
+        Thread(target=self.timeout_setter, daemon=True).start()
 
 
 scheduler = Scheduelr()