Control to interrupt a chaostoolkit experiment as soon as possible

monitor-and-interrupt.py

"""
Control that starts a thread in a background to perform some sort of bespoke
monitoring that interrupts the execution of the experiment as soon as possible
during the method.

To use in your experiment add the following to your experiment:

```json
    "controls": [
        {
            "name": "mykaboom",
            "provider": {
                "type": "python",
                "module": "monitor-and-interrupt"
            }
        }
    ],
```

Make sure this module lives in the `chaos` PYTHONPATH.
"""
from concurrent.futures import Future, ThreadPoolExecutor
import ctypes
import time
import threading

from chaoslib.exceptions import InterruptExecution
from chaoslib.types import Configuration, Experiment, Secrets

stop_event = threading.Event()
executor = ThreadPoolExecutor(max_workers=1)


def before_method_control(context: Experiment,
                          configuration: Configuration = None,
                          secrets: Secrets = None, **kwargs):
    """
    Start a thread that runs our validator and blows the experiment
    when the condition is not met anymore.
    """
    f = executor.submit(monitor, stop_event)
    f.add_done_callback(interrupt_experiment)


def after_method_control(context: Experiment, state,
                         configuration: Configuration = None,
                         secrets: Secrets = None, **kwargs):
    """
    Terminate the monitor thread gracefully.
    """
    stop_event.set()
    executor.shutdown(wait=True)



################################################################################
# Private functions
################################################################################
def monitor(event: threading.Event) -> None:
    """
    Runs a monitor until the event is set or the monitor runs into
    a dire condition.
    """
    while not event.is_set():
        # #########################
        # do stuff here and decide if we need to continue or not
        # you can raise any exception you need but its stack won't be
        # preserved to the main thread.
        # #########################
        print("Hello there")
        time.sleep(1)


def interrupt_experiment(f: Future) -> None:
    """
    Interrupts the main thread from the child monitor thread to throw an
    `InterruptExecution` which will interrupt the experiment's execution next
    chance it can.

    This is not pretty.
    """
    x = f.exception()
    if x is not None:
        main_thread = threading.main_thread()
        target_id = ctypes.c_long(main_thread.ident)

        gil = ctypes.pythonapi.PyGILState_Ensure()
        ret = ctypes.pythonapi.PyThreadState_SetAsyncExc(
            target_id, ctypes.py_object(InterruptExecution))
        ctypes.pythonapi.PyGILState_Release(gil)

        # not much we can do with this errors at this stage, this will
        # mainly be useful to debug after the facts...
        if ret == 0:
            raise ValueError("Invalid thread ID {}".format(target_id))

        elif ret > 1:
            ctypes.pythonapi.PyThreadState_SetAsyncExc(
                ctypes.c_long(target_id), None)
            raise SystemError("PyThreadState_SetAsyncExc failed")

sample-experiment.json

{
    "version": "1.0.0",
    "title": "Say hello and kaboom",
    "description": "n/a",
    "controls": [
        {
            "name": "mykaboom",
            "provider": {
                "type": "python",
                "module": "monitor-and-interrupt"
            }
        }
    ],
    "method": [
        {
            "type": "action",
            "name": "pretend-we-do-stuff",
            "provider": {
                "type": "process",
                "path": "sleep",
                "arguments": "10"
            }
        }
    ]
}

Unfortunately, in this particular example, we cannot interrupt the process immediatly because the exception doesn't have a mean to send a signal to the subprocess by itself.

So we need to change the chaostoolkit to be able to do this on this sort of exception.