add created_at column in script_log

Author: AlexCXC

scheduler/app/faas_scheduler/models.py

@@ -32,6 +32,7 @@ class ScriptLog(Base):
     output = Column(Text, nullable=True)
     operate_from = Column(String(255))
     state = Column(String(10))
+    created_at = Column(DateTime, index=True)
 
     PENDING = "pending"
     RUNNING = "running"
@@ -44,17 +45,17 @@ def __init__(
         org_id,
         script_name,
         context_data,
-        started_at,
         state,
+        created_at,
         operate_from=None,
     ):
         self.dtable_uuid = dtable_uuid
         self.owner = owner
         self.org_id = org_id
         self.script_name = script_name
         self.context_data = context_data
-        self.started_at = started_at
         self.state = state
+        self.created_at = created_at
         self.operate_from = operate_from
 
     def get_info(self):
@@ -77,14 +78,17 @@ def to_dict(self):
             "context_data": (
                 json.loads(self.context_data) if self.context_data else None
             ),
-            "started_at": datetime_to_isoformat_timestr(self.started_at),
+            "started_at": self.started_at
+            and datetime_to_isoformat_timestr(self.started_at),
             "finished_at": self.finished_at
             and datetime_to_isoformat_timestr(self.finished_at),
             "success": self.success,
             "return_code": self.return_code,
             "output": self.output,
             "operate_from": self.operate_from,
             "state": self.state,
+            "created_at": self.created_at
+            and datetime_to_isoformat_timestr(self.created_at),
         }
 
 

scheduler/app/faas_scheduler/utils.py

@@ -2,7 +2,7 @@
 import json
 import logging
 import requests
-from datetime import datetime
+from datetime import datetime, timedelta
 from uuid import UUID
 
 from tzlocal import get_localzone
@@ -380,8 +380,8 @@ def add_script(
         org_id,
         script_name,
         context_data,
-        datetime.now(),
         ScriptLog.PENDING,
+        datetime.now(),
         operate_from,
     )
     db_session.add(script)
@@ -390,8 +390,11 @@ def add_script(
     return script
 
 
-def update_script(db_session, script, success, return_code, output):
-    script.finished_at = datetime.now()
+def update_script(
+    db_session, script, success, return_code, output, started_at, finished_at
+):
+    script.started_at = started_at
+    script.finished_at = finished_at
     script.success = success
     script.return_code = return_code
     script.output = output
@@ -423,10 +426,15 @@ def run_script(
     return True
 
 
-def hook_update_script(db_session, script_id, success, return_code, output, spend_time):
+def hook_update_script(
+    db_session, script_id, success, return_code, output, started_at, spend_time
+):
     script = db_session.query(ScriptLog).filter_by(id=script_id).first()
     if script:
-        update_script(db_session, script, success, return_code, output)
+        finished_at = started_at + timedelta(seconds=spend_time)
+        update_script(
+            db_session, script, success, return_code, output, started_at, finished_at
+        )
         update_statistics(
             db_session, script.dtable_uuid, script.owner, script.org_id, spend_time
         )

scheduler/app/flask_server.py

@@ -14,7 +14,6 @@
     check_auth_token,
     get_script,
     get_run_script_statistics_by_month,
-    hook_update_script,
     can_run_task,
     get_run_scripts_count_monthly,
     ping_starter,
@@ -89,7 +88,7 @@ def scripts_api():
             owner, org_id, db_session, scripts_running_limit=scripts_running_limit
         ):
             return make_response(("The number of runs exceeds the limit"), 400)
-        script_log = scheduler.add_script_log(
+        script_log = scheduler.add(
             uuid_str_to_32_chars(dtable_uuid),
             org_id,
             owner,
@@ -137,23 +136,21 @@ def script_api(script_id):
         ):
             return make_response(("Bad request", 400))
 
-        if SUB_PROCESS_TIMEOUT and isinstance(SUB_PROCESS_TIMEOUT, int):
-            now = datetime.now()
-            duration_seconds = (now - script.started_at).seconds
-            if duration_seconds > SUB_PROCESS_TIMEOUT:
-                script.success = False
-                script.return_code = -1
-                script.finished_at = now
-                script.output = TIMEOUT_OUTPUT
-                db_session.commit()
+        # if SUB_PROCESS_TIMEOUT and isinstance(SUB_PROCESS_TIMEOUT, int):
+        #     now = datetime.now()
+        #     duration_seconds = (now - script.created_at).seconds
+        #     if duration_seconds > SUB_PROCESS_TIMEOUT:
+        #         script.success = False
+        #         script.return_code = -1
+        #         script.finished_at = now
+        #         script.output = TIMEOUT_OUTPUT
+        #         db_session.commit()
 
         return make_response(({"script": script.to_dict()}, 200))
 
     except Exception as e:
         logger.exception(e)
         return make_response(("Internal server error", 500))
-    finally:
-        db_session.close()
 
 
 # get python script statistics logs...
@@ -278,16 +275,14 @@ def record_script_result():
     success = data.get("success", False)
     return_code = data.get("return_code")
     output = data.get("output")
+    started_at = datetime.fromisoformat(data.get("started_at"))
     spend_time = data.get("spend_time")
     script_id = data.get("script_id")
     # update script_log and run-time statistics
     try:
         if script_id:
-            # 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
+                script_id, success, return_code, output, started_at, spend_time
             )
 
     except Exception as e:

scheduler/app/scheduler.py

@@ -2,7 +2,7 @@
 import logging
 import os
 import time
-from datetime import datetime
+from datetime import datetime, timedelta
 from threading import Lock, Thread
 
 from database import DBSession
@@ -24,8 +24,10 @@
 class ScriptQueue:
 
     def __init__(self):
-        self.q = []  # a list of ScriptLog
-        self.script_logs_dict = {}  # a dict of {id: ScriptLog}
+        self.script_queue = (
+            []
+        )  # a list of ScriptLog instances, but can not be used to update database records!!!
+        self.script_dict = {}  # a dict of {id: ScriptLog}
         self.lock = Lock()
         self.running_count = {}
         # a dict of
@@ -75,10 +77,10 @@ def can_run_script(self, script_log: ScriptLog):
 
         return True
 
-    def add_script_log(self, script_log: ScriptLog):
+    def add(self, script_log: ScriptLog):
         with self.lock:
-            self.q.append(script_log)
-            self.script_logs_dict[script_log.id] = script_log
+            self.script_queue.append(script_log)
+            self.script_dict[script_log.id] = script_log
             self.inspect_queue_and_running(
                 pre_msg=f"add script {script_log.get_info()} to queue"
             )
@@ -92,11 +94,11 @@ def get(self):
             return_task = None
 
             index = 0
-            while index < len(self.q):
-                script_log = self.q[index]
+            while index < len(self.script_queue):
+                script_log = self.script_queue[index]
                 if self.can_run_script(script_log):
                     return_task = script_log
-                    self.q.pop(index)
+                    self.script_queue.pop(index)
                     self.increase_running(script_log)
                     self.inspect_queue_and_running(
                         pre_msg=f"get script {script_log.get_info()} from queue"
@@ -154,7 +156,7 @@ def decrease_running(self, script_log):
 
     def script_done_callback(self, script_log: ScriptLog):
         with self.lock:
-            self.script_logs_dict.pop(script_log.id, None)
+            self.script_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"
@@ -172,25 +174,32 @@ def inspect_queue_and_running(self, pre_msg=None):
         lines.append(f"{'<' * 10} running {'<' * 10}")
 
         lines.append(f"{'>' * 10} queue {'>' * 10}")
-        for script_log in self.q:
+        for script_log in self.script_queue:
             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)
+        return self.script_dict.get(script_id)
 
-    def get_timeout_scripts(self):
+    def pop_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)
+            for index in range(len(self.script_queue) - 1, -1, -1):
+                script_log = self.script_queue[index]
+                if (
+                    script_log.state == ScriptLog.RUNNING
+                    and (now_time - script_log.started_at).seconds
+                    >= SUB_PROCESS_TIMEOUT
+                ):
+                    script_logs.append(self.script_queue.pop(index))
+                    self.decrease_running(script_log)
+                    self.inspect_queue_and_running(
+                        pre_msg=f"set script {script_log.get_info()} timeout from queue"
+                    )
         return script_logs
 
 
@@ -199,9 +208,7 @@ class Scheduelr:
     def __init__(self):
         self.script_queue = ScriptQueue()
 
-    def add_script_log(
-        self, dtable_uuid, org_id, owner, script_name, context_data, operate_from
-    ):
+    def add(self, dtable_uuid, org_id, owner, script_name, context_data, operate_from):
         script_log = add_script(
             DBSession(),
             dtable_uuid,
@@ -211,7 +218,7 @@ def add_script_log(
             context_data,
             operate_from,
         )
-        self.script_queue.add_script_log(script_log)
+        self.script_queue.add(script_log)
         return script_log
 
     def schedule(self):
@@ -225,7 +232,11 @@ def schedule(self):
                 db_session.query(ScriptLog).filter(
                     ScriptLog.id == script_log.id
                 ).update(
-                    {ScriptLog.state: ScriptLog.RUNNING}, synchronize_session=False
+                    {
+                        ScriptLog.state: ScriptLog.RUNNING,
+                        ScriptLog.started_at: script_log.started_at,
+                    },
+                    synchronize_session=False,
                 )
                 db_session.commit()
                 script_file_info = get_script_file(
@@ -244,49 +255,28 @@ def schedule(self):
             finally:
                 DBSession.remove()
 
-    def script_done_callback(self, script_id, success, return_code, output, spend_time):
+    def script_done_callback(
+        self, script_id, success, return_code, output, started_at, spend_time
+    ):
         hook_update_script(
-            DBSession(), script_id, success, return_code, output, spend_time
+            DBSession(), script_id, success, return_code, output, started_at, 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):
+    def load_pending_scripts(self):
         """load pending script logs, should be called only when server start"""
-        script_logs = (
+        script_logs = list(
             DBSession.query(ScriptLog)
             .filter_by(state=ScriptLog.PENDING)
+            .filter(ScriptLog.created_at > (datetime.now() - timedelta(hours=1)))
             .order_by(ScriptLog.id)
         )
+        logger.info(f"load {len(script_logs)} pending scripts created within 1 hour")
         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,
-                        },
-                        synchronize_session=False,
-                    )
-            except Exception as e:
-                logger.exception(e)
-            finally:
-                DBSession.remove()
-            time.sleep(60)
+            self.script_queue.add(script_log)
 
     def statistic_cleaner(self):
         while True:
@@ -301,10 +291,9 @@ def statistic_cleaner(self):
             time.sleep(24 * 60 * 60)
 
     def start(self):
-        self.load_pending_script_logs()
+        self.load_pending_scripts()
         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()