How to implement a custom SIGTERM handler?

[Ailothaen]

I have a Python Flask app:

from flask import Flask, Response
import signal

app = Flask("myapp")

def exit_gracefully(*args):
    print("Doing cleanup actions here")

signal.signal(signal.SIGINT, exit_gracefully)
signal.signal(signal.SIGTERM, exit_gracefully)

@app.route("/myendpoint")
def myendpoint():
    print("Doing something here")
    return Response("Everything is fine", status=200, mimetype="application/json")

This app is ran by gunicorn, using a classic wsgi.py file and a systemd service:

[Unit]
Description=myservice

[Service]
User=root
Group=root
WorkingDirectory=/opt/myservice
ExecStart=/opt/myservice/run.sh

[Install]
WantedBy=multi-user.target

The content of run.sh:

#!/bin/bash
source env/bin/activate
env/bin/gunicorn --bind unix:gunicorn.sock --workers 1 --timeout 10 wsgi:app

I need to intercept SIGTERM signals sent to the service, because I need to perform some cleanup and interruptions (especially setting threading.Event() flags)

However, the issue is that, when I type systemctl stop myservice, the service hangs for 30 seconds before shutting down (which is probably gunicorn's graceful_timeout setting), instead of shutting down immediately, as I intend to.

When looking in the source code of gunicorn, I notice that the Worker class uses a signal handler to shutdown itself gracefully:

# gunicorn/workers/base.py:173
signal.signal(signal.SIGTERM, self.handle_exit)

My guess is that, since it is possible to use only one handler for each signal, I am overriding this handler by using my own handler.
But I still need both handlers, mine for my cleanup operations to be done and the default one for the gunicorn worker to stop properly.

Something maybe worth to mention is that, when I run run.sh in my shell and send SIGINT to it (by pressing Ctrl+C), my app stopped immediately, despite gunicorn also implements an handler for it.

Any ideas on how I should proceed? I thought about calling the function handle_exit in my own handler, but since my code is "gunicorn-agnostic", I have no idea about how I should do that. Should I change the design of my app?

[aawilson]

I'm also interested in what the "porcelain" way to do this is. The best solution I've run into (the one gunicorn_thrift uses) is to have your worker class implement config that can be used for cleanup. E.g.:

from gunicorn.config import Setting
from gunicorn.workers.ggevent import GeventWorker  # Of course this would be whichever worker you actually want as your base

class WorkerTerm(Setting):
    name = "worker_term"
    section = "Server Hooks"
    validator = validate_callable(1)
    type = six.callable

    def worker_term(worker):
        pass

    default = staticmethod(worker_term)
    desc = """\
        Called just after a worker received SIGTERM, and about to gracefully
        shutdown.

        The callable needs to accept one instance variable for the initialized
        Worker.
        """
class CustomGeventWorker(GeventWorker):
    def handle_exit(self, sig, frame):
        ret = super(GeventThriftPyWorker, self).handle_exit(sig, frame)
        self.cfg.worker_term(self)
        return ret    

then wherever your gunicorn config is, you would implement worker_term there. It gets tricky at this point, though, because that config doesn't necessarily know what you've passed to gunicorn as the app, so if you e.g. reach in to your wsgi app submodule to grab the threading.Signal instance, you will be violating a separation of concerns (and making that config file not usable with a different app). Depending on how you instantiate your app instances you might be able to get around this with e.g. the post_worker_init setting, but it's all quite messy.

[rolfosian]

This is how I did it for the Flask debug server with reloader handling:

import threading
from flask import Flask
from signal import SIGINT, SIGTERM, signal
from time import sleep

go_flag = threading.Event()
go_flag.set()

def dummy(go_flag):
    print('starting')
    while go_flag.is_set():
        sleep(0.2)
    print('closing')

threads = [threading.Thread(target=dummy, args=(go_flag,)) for _ in range(10)]

def start_threads():
    for thread in threads:
        thread.start()
        
def join_threads(go_flag):
    go_flag.clear()
    for thread in threads:
        thread.join()

def signal_handler_exit(sig, frame, go_flag):
    print(f"Received signal {sig}. Exiting...")
    join_threads(go_flag)
    exit(0)
    
def set_signal_handlers(SIGINT, SIGTERM, go_flag):
    signal(SIGINT, lambda signum, frame: signal_handler_exit(signum, frame, go_flag))
    signal(SIGTERM, lambda signum, frame: signal_handler_exit(signum, frame, go_flag))

from werkzeug._reloader import StatReloaderLoop, reloader_loops
class DummyReloaderLoop(StatReloaderLoop):        
    def trigger_reload(self, filename: str) -> None:
        join_threads(go_flag)
        return super().trigger_reload(filename)

reloader_loops['auto'] = DummyReloaderLoop

app = Flask(__name__)

@app.route('/')
def hello_world():
    return 'Hello, World!'

if __name__ == '__main__':
    start_threads()
    set_signal_handlers(SIGINT, SIGTERM, go_flag)
    app.run(threaded=True, host='127.0.0.1', debug=True, use_reloader=True, port=5000)

However for gunicorn.conf.py I settled on this which isn't exactly clean as a whistle, as it runs for every worker? It gets the job done though. I couldn't get it to work properly with on_starting and on_reload unfortunately.

def post_worker_init(worker):
    from main import start_threads, join_threads, go_flag, set_signal_handlers
    from signal import SIGINT, SIGTERM
    
    def custom_changed(func):
        def wrapper(*args, **kwargs):
            join_threads(go_flag)
            return func(*args, **kwargs)
        return wrapper
    worker.reloader._callback = custom_changed(worker.reloader._callback)
    
    start_threads()
    set_signal_handlers(SIGINT, SIGTERM, go_flag)

edit: I found this to be much better and more responsive

def set_signal_handlers(SIGINT, SIGTERM, go_flag):
    def handler_wrapper(func):
        #functools.wraps
        @wraps(func)
        def wrapper(*args, **kwargs):
            # print('Joining threads...')
            join_threads(go_flag)
            return func(*args, **kwargs)
        return wrapper

    sigint_handler = getsignal(SIGINT)
    sigterm_handler = getsignal(SIGTERM)
    
    wrapped_sigint_handler = handler_wrapper(sigint_handler)
    wrapped_sigterm_handler = handler_wrapper(sigterm_handler)
    
    signal(SIGINT, wrapped_sigint_handler)
    signal(SIGTERM, wrapped_sigterm_handler)
    
class ReloaderLoop_(StatReloaderLoop):
    def trigger_reload(self, filename: str) -> None:
        join_threads(go_flag)
        return super().trigger_reload(filename)
    # because the signal handlers arent cleared on the initial restart by the flask debug server so this resets them
    def restart_with_reloader(self) -> int:
        signal(SIGINT, SIG_DFL)
        signal(SIGTERM, SIG_DFL)
        return super().restart_with_reloader()