Event-Publisher Flush Queue on Shutdown

[atai92]

Closes #

💸 TL;DR

Event-Publisher sidecar is not flushing the entire POSIX queue when receiving a SIGINT or SIGTERM. This PR creates a signal handler which will flush the queue whenever a SIGINT or SIGTERM is received. If this is not addressed, some applications that use the Event Publisher sidecar may experience dropped messages if their Event Publisher sidecar is terminated and the queue still has messages (ex: during a deployment).

📜 Details

Design Doc

Jira

🧪 Testing Steps / Validation

Reproduced the issue by running the event-publisher, filling up the event queue, and adding events to the queue after each polling cycle by modifying the event-publisher

    try:
        [event_queue.put("{}",timeout=QUEUE_TIMEOUT) for i in range(0,10)]
    except:
        pass
    
    while True:
        message: Optional[bytes]

        try:
            message = event_queue.get(timeout=QUEUE_TIMEOUT)
            event_queue.put("{}",timeout=QUEUE_TIMEOUT)
        except TimedOutError:
            message = None

        if batcher.is_ready:
            serialize_and_publish_batch(publisher, batcher)
        batcher.add(message)

This keeps the queue saturated with something.
Then, force-closed the event-publisher using ctrl-c.

Afterwards, check the POSIX queue to see if there are messages left over:

>>> from baseplate.lib.message_queue import MessageQueue

        "/events-" + "test",
        max_messages=10,
        max_message_size=8192,
    )
message = event_queue.get(timeout=QUEUE_TIMEOUT)>>> QUEUE_TIMEOUT = 0.2
>>> event_queue = MessageQueue(
...         "/events-" + "test",
...         max_messages=10,
...         max_message_size=8192,
...     )
>>> message = event_queue.get(timeout=QUEUE_TIMEOUT)

If we don't see a timeout, then we know that there was a message that was retrieved.

Note that the bad request above should not impact the test because the client error will only result in the messages being discarded due to a network error which is not the objective of this test.

To test the solution, the same test was done with the flush_queue_signal_handler() set for SIGINT and SIGTERM. As can be seen below, a TimedOutError was received when trying to get a message from the queue signaling that the queue was empty and flushed correctly.

These tests were both repeated a few times.

✅ Checks

• CI tests (if present) are passing
• Adheres to code style for repo
• Contributor License Agreement (CLA) completed if not a Reddit employee

[sydjryan]

this looks good, approving to quiet harold - I know we're talking in slack about when to release it