cms-SiteForge/app/worker_health.py

"""Docker worker liveness probe: no localhost Web dependency."""
import os
import time
import socket
import re
from pathlib import Path
from . import db

def heartbeat_path():
    instance=os.environ.get('SITEFORGE_WORKER_INSTANCE', '')
    if not instance and db.POSTGRES:
        instance=socket.gethostname()
    if instance:
        if not re.fullmatch(r'[A-Za-z0-9._-]{1,120}',instance):
            raise ValueError('Invalid worker instance identifier')
        return db.DATA/'worker-heartbeats'/instance
    return db.DATA/'worker-heartbeat'

def healthy(path=None, max_age=None, check_store=False):
    file=Path(path) if path is not None else heartbeat_path()
    limit=float(max_age if max_age is not None else os.getenv('SITEFORGE_WORKER_HEALTH_MAX_AGE','45'))
    if limit <= 0:return False
    try:
        age=time.time()-file.stat().st_mtime
        fresh=-5 <= age < limit
        if not fresh:return False
        if check_store:
            from .healthchecks import readiness
            return readiness()
        return True
    except (OSError,ValueError):
        return False

if __name__=='__main__':
    raise SystemExit(0 if healthy(check_store=True) else 1)