Merge pull request #1223 from MetPX/issue1214_another_scheduled_refactor

petersilva · web-flow · commit 46ff74720b5d · 2024-09-16T17:53:34.000-04:00
Issue1214 another scheduled refactor
diff --git a/sarracenia/flowcb/scheduled/__init__.py b/sarracenia/flowcb/scheduled/__init__.py
@@ -46,49 +46,18 @@ class hoho(Scheduled):
 
     """
 
-    def update_appointments(self,when):
-        """
-          # make a flat list from values where comma separated on a single or multiple lines.
-
-          set self.appointments to a list of when something needs to be run during the current day.
-        """
-        self.appointments=[]
-        if self.o.scheduled_minute or self.o.scheduled_hour:
-            for h in self.hours:
-               for m in self.minutes:
-                   if ( h > when.hour ) or ((h == when.hour) and ( m >= when.minute )):
-                       appointment = datetime.time(h, m, tzinfo=datetime.timezone.utc )
-                       next_time = datetime.datetime.combine(when,appointment)
-                       self.appointments.append(next_time)
-                   else:
-                       pass # that time is passed for today.
-        if self.o.scheduled_time:
-            for time in self.sched_times:
-                hour,minute=time.split(':')
-                hour = int(hour)
-                minute = int(minute)
-                if ( hour > when.hour ) or ((hour == when.hour) and ( minute >= when.minute )):
-                    appointment = datetime.time(hour, minute, tzinfo=datetime.timezone.utc )
-                    next_time = datetime.datetime.combine(when,appointment)
-                    self.appointments.append(next_time)
-                else:
-                    pass # that time is passed for today.
+    def __init__(self,options,class_logger=logger):
+        super().__init__(options,class_logger)
+        # if logLevel is set for a subclass, apply it here too
+        if hasattr(self.o,'logLevel') and logger:
+            logger.setLevel(getattr(logging, self.o.logLevel.upper()))
         
-        self.appointments.sort()
-
-
-        logger.info( f"for {when}: {json.dumps(list(map( lambda x: str(x), self.appointments))) } ")
-
-
-    def __init__(self,options,logger=logger):
-        super().__init__(options,logger)
         self.o.add_option( 'scheduled_interval', 'duration', 0 )
         self.o.add_option( 'scheduled_hour', 'list', [] )
         self.o.add_option( 'scheduled_minute', 'list', [] )
         self.o.add_option( 'scheduled_time', 'list', [] )
         
         self.housekeeping_needed=False
-        self.interrupted=None
 
         self.sched_times = sum([ x.split(',') for x in self.o.scheduled_time],[])
         #self.sched_times.sort()
@@ -104,7 +73,6 @@ def __init__(self,options,logger=logger):
         self.minutes = list(map( lambda x: int(x), sched_min))
         #self.minutes.sort()
 
-
         self.default_wait=datetime.timedelta(seconds=300)
 
         logger.debug( f'minutes: {self.minutes}')
@@ -114,104 +82,69 @@ def __init__(self,options,logger=logger):
         self.first_interval=True
 
         if self.o.scheduled_interval <= 0 and not self.appointments:
-            logger.info( f"no scheduled_interval or appointments (combination of scheduled_hour and scheduled_minute) set defaulting to every {self.default_wait.seconds} seconds" )
+            logger.warning(f"no scheduled_interval or appointments (combination of scheduled_hour and scheduled_minute) set defaulting to every {self.default_wait.seconds} seconds")
 
-    def gather(self,messageCountMax):
-
-        # for next expected post
-        self.wait_until_next()
-
-        if self.stop_requested or self.housekeeping_needed:
-            return (False, [])
+        # Determine the next gather time
+        # For scheduled_interval, gather immediately after starting
+        if self.o.scheduled_interval and self.o.scheduled_interval > 0:
+            self.next_gather_time = now
+        else:
+            self.next_gather_time = None
+            self.calc_next_gather_time()
 
-        logger.info('time to run')
+        self.last_gather_time = 0
 
-        # always post the same file at different time
-        gathered_messages = []
-
-        for relPath in self.o.path:
-            st = FmdStat()
-            m = sarracenia.Message.fromFileInfo(relPath, self.o, st)
-            gathered_messages.append(m)
-
-        return (True, gathered_messages)
-
-    def on_housekeeping(self):
-
-        self.housekeeping_needed = False
-
-
-    def wait_seconds(self,sleepfor):
-        """
-           sleep for the given number of seconds, like time.sleep() but broken into
-           shorter naps to be able to honour stop_requested, or when housekeeping is needed.
-           
+    def update_appointments(self,when):
         """
+          # make a flat list from values where comma separated on a single or multiple lines.
 
-        housekeeping=datetime.timedelta(seconds=self.o.housekeeping)
-        nap=datetime.timedelta(seconds=10)
-
-        if self.interrupted:
-            sleepfor = self.interrupted
-            now = datetime.datetime.fromtimestamp(time.time(),datetime.timezone.utc)
-
-            # update sleep remaining based on how long other processing took.
-            interruption_duration= now-self.interrupted_when
-            sleepfor -= interruption_duration
-
-        if sleepfor < nap:
-            nap=sleepfor
-    
-        sleptfor=datetime.timedelta(seconds=0)
-
-        while sleepfor > datetime.timedelta(seconds=0):
-            time.sleep(nap.total_seconds())
-            if self.stop_requested:
-                return
-
-            # how long is left to sleep.
-            sleepfor -= nap
-            self.interrupted=sleepfor
-            self.interrupted_when = datetime.datetime.fromtimestamp(time.time(),datetime.timezone.utc)
-
-            sleptfor += nap
-            if sleptfor > housekeeping:
-                self.housekeeping_needed=True
-                return
-
-        # got to the end of the interval...
-        self.interrupted=None
-       
-    def wait_until( self, appointment ):
-
-        now = datetime.datetime.fromtimestamp(time.time(),datetime.timezone.utc)
-
-        sleepfor=appointment-now
-
-        logger.info( f"appointment at: {appointment}, need to wait: {sleepfor})" )
-        self.wait_seconds( sleepfor )
-
-
-    def wait_until_next( self ):
-
-        one_min = datetime.timedelta(minutes=1)
-
-        if self.o.scheduled_interval > 0:
-            if self.first_interval:
-                self.first_interval=False
-                return
+          set self.appointments to a list of when something needs to be run during the current day.
+        """
+        self.appointments=[]
+        if self.o.scheduled_minute or self.o.scheduled_hour:
+            for h in self.hours:
+               for m in self.minutes:
+                   if ( h > when.hour ) or ((h == when.hour) and ( m >= when.minute )):
+                       appointment = datetime.time(h, m, tzinfo=datetime.timezone.utc )
+                       next_time = datetime.datetime.combine(when,appointment)
+                       self.appointments.append(next_time)
+                   else:
+                       pass # that time is passed for today.
+        if self.o.scheduled_time:
+            for time in self.sched_times:
+                hour,minute=time.split(':')
+                hour = int(hour)
+                minute = int(minute)
+                if ( hour > when.hour ) or ((hour == when.hour) and ( minute >= when.minute )):
+                    appointment = datetime.time(hour, minute, tzinfo=datetime.timezone.utc )
+                    next_time = datetime.datetime.combine(when,appointment)
+                    self.appointments.append(next_time)
+                else:
+                    pass # that time is passed for today.
+        
+        self.appointments.sort()
+        logger.info( f"for {when}: {self.appointments_to_string()} ")
 
-            self.wait_seconds(datetime.timedelta(seconds=self.o.scheduled_interval))
-            return
+    def appointments_to_string(self):
+        return json.dumps(list(map( lambda x: str(x), self.appointments)))
 
-        if ( len(self.o.scheduled_hour) > 0 ) or ( len(self.o.scheduled_minute) > 0 ) or self.o.scheduled_time:
-            now = datetime.datetime.fromtimestamp(time.time(),datetime.timezone.utc)
+    def calc_next_gather_time(self, last_gather=0):
+        if self.next_gather_time in self.appointments:
+            self.appointments.remove(self.next_gather_time)
+        if last_gather == 0:
+            last_gather = datetime.datetime.now(datetime.timezone.utc)
+        
+        # Scheduled interval overrides other options
+        if self.o.scheduled_interval and self.o.scheduled_interval > 0:
+            self.next_gather_time = last_gather + datetime.timedelta(seconds=self.o.scheduled_interval)
+            logger.debug(f"next gather should be in {self.o.scheduled_interval}s, scheduled for {self.next_gather_time}")
+        
+        # No scheduled interval --> try to use configured schedule
+        elif len(self.o.scheduled_hour) > 0 or len(self.o.scheduled_minute) > 0 or len(self.o.scheduled_time) > 0:
             next_appointment=None
             missed_appointments=[]
             for t in self.appointments: 
-                # Need a little bit before or after apointment, to allow some wiggle room for sleep overhead.
-                # See issue #1214 for more details
-                if t < now < t + one_min: 
+                if last_gather < t: 
                     next_appointment=t
                     break
                 else:
@@ -230,21 +163,47 @@ def wait_until_next( self ):
                 self.update_appointments(midnight)
                 next_appointment=self.appointments[0]
 
-            self.wait_until(next_appointment)
-            if self.interrupted:
-                logger.info( f"sleep interrupted, returning for housekeeping." )
-            else:
-                self.appointments.remove(next_appointment)
-                logger.info( f"ok {len(self.appointments)} appointments left today" )
-            return
-
-        # default wait...
-
-        if self.first_interval:
-            self.first_interval=False
-            return
+            self.next_gather_time = next_appointment
+            logger.debug(f"next gather scheduled for {self.next_gather_time} from appointments {self.appointments_to_string()}")
+
+        # No scheduled interval and no scheduled hour/minutes/time
+        else:
+            self.next_gather_time = last_gather + self.default_wait
+            logger.debug(f"next gather should be in {self.default_wait.seconds}s, scheduled for {self.next_gather_time} (default_wait")
+
+    def ready_to_gather(self):
+        current_time = datetime.datetime.now(datetime.timezone.utc )
+        if current_time >= self.next_gather_time and not self.stop_requested:
+            late = (current_time - self.next_gather_time).total_seconds()
+            logger.info(f"--> yes, now >= {self.next_gather_time} ({late}s late)")
+            # NOTE: could also pass self.next_gather_time to calc_next_gather_time to get more precise intervals
+            # See https://github.com/MetPX/sarracenia/issues/1214#issuecomment-2344711046 for discussion.
+            self.calc_next_gather_time(current_time)
+            self.last_gather_time = current_time
+            return True
+        else:
+            logger.debug(f"--> no, next gather scheduled for {self.next_gather_time}")
+
+    def gather(self, messageCountMax):
+        if self.ready_to_gather():
+            # always post the same file at different time
+            gathered_messages = []
+            for relPath in self.o.path:
+                st = FmdStat()
+                m = sarracenia.Message.fromFileInfo(relPath, self.o, st)
+                gathered_messages.append(m)
+            return (True, gathered_messages)
+        else:
+            logger.debug(f"nothing to do")
+            return (False, [])
 
-        self.wait_seconds(self.default_wait)
+    def on_housekeeping(self):
+        logger.info(f"next gather scheduled for {self.next_gather_time}")
+        n_appointments = len(self.appointments)
+        if n_appointments > 0:
+            logger.info(f"{n_appointments} appointments remaining for today")
+            logger.debug(f"remaining appointments: {self.appointments_to_string()}")
+        self.housekeeping_needed = False
 
 if __name__ == '__main__':
     
diff --git a/sarracenia/flowcb/scheduled/http_with_metadata.py b/sarracenia/flowcb/scheduled/http_with_metadata.py
@@ -34,10 +34,7 @@ def __init__(self, options, logger=logger):
 
     def gather(self,messageCountMax):
 
-        # for next expected post
-        self.wait_until_next()
-
-        if self.stop_requested or self.housekeeping_needed:
+        if not self.ready_to_gather():
             return (False, [])
 
         logger.info('time to run')
diff --git a/sarracenia/flowcb/scheduled/poll.py b/sarracenia/flowcb/scheduled/poll.py
@@ -1,37 +1,26 @@
 import logging
-import requests
-import base64
-
-import datetime
-import os
 import sys
-import time
-
-from datetime import date
 
 import sarracenia
 from sarracenia.flowcb.scheduled import Scheduled
 
 logger = logging.getLogger(__name__)
 
-
-
-
 class Poll(Scheduled):
     """
       
     """
+    def __init__(self,options,logger=logger):
+        super().__init__(options,logger)
 
     def gather(self,messageCountMax): # placeholder
         """
-
            This gather aborts further gathers if the next interval has not yet arrived.
-
         """
-        logger.info( f"waiting for next poll")
-        self.wait_until_next()
-
-        return  not (self.stop_requested or self.housekeeping_needed), [] 
+        ready = self.ready_to_gather()
+        if ready:
+            logger.info("poll routine will run")
+        return (ready, [])
 
 
 if __name__ == '__main__':
diff --git a/sarracenia/flowcb/scheduled/wiski.py b/sarracenia/flowcb/scheduled/wiski.py
@@ -135,8 +135,10 @@ def gather(self,messageCountMax): # placeholder
         
         messages=[]
 
-        self.wait_until_next()
+        if not self.ready_to_gather():
+            return (False, [])
 
+        start_time = datetime.datetime.fromtimestamp(time.time(),datetime.timezone.utc)
         while (1):
             if self.stop_requested or self.housekeeping_needed:
                 return (False, messages)
@@ -156,6 +158,9 @@ def gather(self,messageCountMax): # placeholder
                 break
             else:
                 logger.info( f"request failed. Status code: {response.status_code}: {response.text}" )
+            
+            now = datetime.datetime.fromtimestamp(time.time(),datetime.timezone.utc)
+            self.housekeeping_needed = ((now-start_time).total_seconds() >= self.o.housekeeping)
         
         k = KIWIS(self.main_url, headers=headers )
 
diff --git a/tests/sarracenia/flowcb/scheduled/__scheduled___test.py b/tests/sarracenia/flowcb/scheduled/__scheduled___test.py
