Created
October 31, 2021 14:44
-
-
Save xi/a518039354e4253be5eae789fe2ae091 to your computer and use it in GitHub Desktop.
simple cron implementation with useful logging
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import datetime | |
import logging | |
import subprocess | |
import sys | |
import time | |
MINUTE = datetime.timedelta(seconds=60) | |
logging.basicConfig( | |
stream=sys.stderr, | |
level=logging.INFO, | |
format='%(asctime)s %(levelname)-8s %(message)s', | |
) | |
def parse_crontab(path): | |
entries = [] | |
with open(path) as fh: | |
for line in fh: | |
if line.startswith('#') or not line.strip(): | |
continue | |
else: | |
entries.append(line.rstrip().split('\t', 5)) | |
return entries | |
def is_match(entry, now): | |
return ( | |
(entry[0] == '*' or int(entry[0]) == now.minute) | |
and (entry[1] == '*' or int(entry[1]) == now.hour) | |
and (entry[2] == '*' or int(entry[2]) == now.day) | |
and (entry[3] == '*' or int(entry[3]) == now.month) | |
and (entry[4] == '*' or int(entry[4]) == now.weekday()) | |
) | |
def log_errors(processes): | |
processes_keep = [] | |
for cmd, p in processes: | |
if p.poll() is None: | |
processes_keep.append((cmd, p)) | |
elif p.returncode != 0: | |
logging.error('%s exited with %i' % (cmd, p.returncode)) | |
return processes_keep | |
if __name__ == '__main__': | |
crontab = parse_crontab(sys.argv[1]) | |
processes = [] | |
now = datetime.datetime.now() | |
while True: | |
# NOTE: as a side effect, this mechanism correctly handles | |
# daylight saving time: ``now`` always increases monotonously. | |
# If ``datetime.now()`` moves back, ``now``, waits for it to | |
# catch up. If ``datetime.now()`` jumps forward, ``now`` | |
# immediately moves through the complete hour. | |
while now <= datetime.datetime.now(): | |
for entry in crontab: | |
if is_match(entry, now): | |
cmd = entry[-1] | |
logging.info('Running %s' % cmd) | |
p = subprocess.Popen(cmd, shell=True) | |
processes.append((cmd, p)) | |
processes = log_errors(processes) | |
now += MINUTE | |
time.sleep(60) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment