Add queue flushing signal handler

[atai92]

Closes #

💸 TL;DR

Add a queue flushing signal handler, so when the event publisher receives a SIGINT/SIGTERM signal it'll continue flushing the queue, so events aren't dropped on shutdown.

📜 Details

Design Doc

Jira

🧪 Testing Steps / Validation

Tested using the included Dockerfile using the following steps:

docker run --sysctl fs.mqueue.msg_max=10000 --sysctl fs.mqueue.msgsize_max=102400 -it -v ~/baseplate.py:/src --ulimit msgqueue=-1 c725da7d50d0
python -m baseplate.sidecars.event_publisher --queue-name test --debug test.ini

# In a separate tab, exec into the container
docker exec -it 07e5b9a2bb37 /bin/bash
python
# Within Python, ran the code below
from baseplate.lib.message_queue import MessageQueue
import time

queue_name = "test"

event_queue = MessageQueue(
        "/events-" + queue_name,
        max_messages=10000,
        max_message_size=102400,
    )


message = b'[{"source":"test","action":"test","noun":"test","client_timestamp":1683823393779,"uuid":"20de9dec-6632-44fa-a3a3-ca29d177776d","action_info":{"reason":"test"}}]'

current_time = time.time()
while time.time() - current_time < 5:
    event_queue.put(message)

This screenshot below shows the reproduction of a bug where we would exit from a BatchFull exception:

This screenshot below shows that the BatchFull exception is no longer present using the same steps above and sending a SIGINT to event_publisher will still cause it to wait for the POSIX queue to be empty before exiting:

✅ Checks

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

[KTAtkinson]

The length here will always be greater than 0 because we add a header. We might want to add a items_count function to the batcher to count the number of items currently in the queue. It could use the length of batcher._items.

[redloaf]

Is this still needed?