Redis crashes in the log shipping thread causing RAM issues (aka. postmortem on the 2018-08-09 incident)
On 2018-08-09, something happened on the control node which caused Redis to throw an error. This error was not caught in ArchiveBot, and this caused the thread sending logs to Redis to crash (below is an example from job 2tldvnvno8tzumwauaj6zoa08):

    Exception in thread Thread-2:
    Traceback (most recent call last):
      File "/usr/lib/python3.5/threading.py", line 914, in _bootstrap_inner
        self.run()
      File "/usr/lib/python3.5/threading.py", line 862, in run
        self._target(*self._args, **self._kwargs)
      File "/home/archivebot/ArchiveBot/pipeline/archivebot/control.py", line 266, in ship_logs
        pipe.execute()
      File "/home/archivebot/.local/lib/python3.5/site-packages/redis/client.py", line 2578, in execute
        return execute(conn, stack, raise_on_error)
      File "/home/archivebot/.local/lib/python3.5/site-packages/redis/client.py", line 2519, in _execute_pipeline
        self.raise_first_error(commands, response)
      File "/home/archivebot/.local/lib/python3.5/site-packages/redis/client.py", line 2526, in raise_first_error
        raise r
      File "/home/archivebot/.local/lib/python3.5/site-packages/redis/client.py", line 2514, in _execute_pipeline
        self.parse_response(connection, args[0], **options))
      File "/home/archivebot/.local/lib/python3.5/site-packages/redis/client.py", line 2536, in parse_response
        self, connection, command_name, **options)
      File "/home/archivebot/.local/lib/python3.5/site-packages/redis/client.py", line 577, in parse_response
        response = connection.read_response()
      File "/home/archivebot/.local/lib/python3.5/site-packages/redis/connection.py", line 574, in read_response
        raise response
    redis.exceptions.ResponseError: Command # 1 (EVALSHA f6fda5256637da887e5745b23660c71a16a25ed0 1 2tldvnvno8tzumwauaj6zoa08 {"url": "https://lancasteronline.com/obituaries/jean-ross/article_2c56fd48-cad1-547f-bdda-68cf13c215a4.html", "wget_code": "OK", "is_warning": false, "is_error": false, "ts": 1533839662.075921, "response_code": 200, "type": "download"} updates 2tldvnvno8tzumwauaj6zoa08_log) of pipeline caused error: Error running script (call to f_f6fda5256637da887e5745b23660c71a16a25ed0): @user_script:7: @user_script: 7: -MISCONF Redis is configured to save RDB snapshots, but is currently not able to persist on disk. Commands that may modify the data set are disabled. Please check Redis logs for details about the error.

A week later, on 2018-08-17, three jobs (all on different pipelines) were killed by the kernel's OOM killer because the machine ran out of RAM. The reason was the Redis code in `pipeline/archivebot/control.py`. ArchiveBot uses a Queue internally for storing the log lines that haven't been shipped to Redis yet. Because the thread consuming this queue crashed due to the issue on the control node, the queue started growing infinitely, eventually consuming all available memory. (It is not clear whether there were also jobs killed before 2018-08-17.)

There are two things here that I think need fixing:

1. Handle the exception so it doesn't crash the log shipper thread.
2. Limit the log queue size. When it's full, the oldest entry should just be dropped. (NB, the code already drops entries when it can't send them to Redis, so the lost log entries shouldn't be too much of a concern.)
diff --git a/pipeline/archivebot/control.py b/pipeline/archivebot/control.py
index 988a113..9ddd84b 100644
--- a/pipeline/archivebot/control.py
+++ b/pipeline/archivebot/control.py
@@ -3,11 +3,11 @@ import time
 import os
 import logging
 import threading
-from queue import Queue, Empty
+from queue import Queue, Empty, Full
 from contextlib import contextmanager
 
 import redis
-from redis.exceptions import ConnectionError as RedisConnectionError
+from redis.exceptions import ConnectionError as RedisConnectionError, RedisError
 
 logger = logging.getLogger('archivebot.control')
 
@@ -58,7 +58,7 @@ class Control(object):
         self.items_queued_outstanding = 0
         self.bytes_downloaded_outstanding = 0
         self.redis_url = redis_url
-        self.log_queue = Queue()
+        self.log_queue = Queue(maxsize = 10000)
 
         # if ITEM_IDENT is set, we are running inside a wpull process
         self.ident = os.getenv('ITEM_IDENT')
@@ -219,68 +219,81 @@ class Control(object):
         logger.info('Started log shipper thread with ident={}, thread={}'
                     .format(self.ident, threading.get_ident()))
 
-        with conn(self):
-            with self.redis.pipeline(transaction=False) as pipe:
-                while not (self.ending and self.log_queue.empty()):
-
-                    try:
-                        # Ship a log entry
-                        try:
-                            entry = self.log_queue.get(timeout=5)
-                            with conn(self):
-                                self.log_script(keys=entry['keys'], args=entry['args'], client=pipe)
-
-                            shipping_count += 1
-
-                            self.log_queue.task_done()
-
-                        except Empty:
-                            pass #don't task_done() without tasks
-                        except RedisConnectionError:
-                            # If we couldn't ship it due to redis being down, discard
-                            self.log_queue.task_done()
-
-                        # If we have accreted enough or the queue is empty, commit logs and counts
-                        # The magic constant is necessary to resolve a race condition that might
-                        # prevent shipping
-                        if self.log_queue.empty() or shipping_count >= 64:
-                            with conn(self):
-                                # This locking structure is necessary to avoid a deadlock that
-                                # happens when redis is trying to send while another thread is
-                                # trying to acquire the lock
-                                with self.countslock:
-                                    t_bytes_downloaded = self.bytes_downloaded_outstanding
-                                    self.bytes_downloaded_outstanding = 0
-                                    t_items_downloaded = self.items_downloaded_outstanding
-                                    self.items_downloaded_outstanding = 0
-                                    t_items_queued = self.items_queued_outstanding
-                                    self.items_queued_outstanding = 0
-
-                                if t_bytes_downloaded > 0:
-                                    pipe.hincrby(self.ident, 'bytes_downloaded', t_bytes_downloaded)
-                                if t_items_downloaded > 0:
-                                    pipe.hincrby(self.ident, 'items_downloaded', t_items_downloaded)
-                                if t_items_queued > 0:
-                                    pipe.hincrby(self.ident, 'items_queued', t_items_queued)
-
-                                pipe.execute()
-
-                            shipping_count = 0
-
-                    except RedisConnectionError:
-                        logger.info('Log shipper got connection error while '
-                                    'incrementing counts or committing logs with '
-                                    'ident={}, thread={}'.format(self.ident, threading.get_ident()))
+        while not (self.ending and self.log_queue.empty()):
+            try:
+                with conn(self):
+                    with self.redis.pipeline(transaction=False) as pipe:
+                        while not (self.ending and self.log_queue.empty()):
+                            try:
+                                # Ship a log entry
+                                try:
+                                    entry = self.log_queue.get(timeout=5)
+                                    with conn(self):
+                                        self.log_script(keys=entry['keys'], args=entry['args'], client=pipe)
+
+                                    shipping_count += 1
+
+                                    self.log_queue.task_done()
+
+                                except Empty:
+                                    pass #don't task_done() without tasks
+                                except RedisConnectionError:
+                                    # If we couldn't ship it due to redis being down, discard
+                                    self.log_queue.task_done()
+
+                                # If we have accreted enough or the queue is empty, commit logs and counts
+                                # The magic constant is necessary to resolve a race condition that might
+                                # prevent shipping
+                                if self.log_queue.empty() or shipping_count >= 64:
+                                    with conn(self):
+                                        # This locking structure is necessary to avoid a deadlock that
+                                        # happens when redis is trying to send while another thread is
+                                        # trying to acquire the lock
+                                        with self.countslock:
+                                            t_bytes_downloaded = self.bytes_downloaded_outstanding
+                                            self.bytes_downloaded_outstanding = 0
+                                            t_items_downloaded = self.items_downloaded_outstanding
+                                            self.items_downloaded_outstanding = 0
+                                            t_items_queued = self.items_queued_outstanding
+                                            self.items_queued_outstanding = 0
+
+                                        if t_bytes_downloaded > 0:
+                                            pipe.hincrby(self.ident, 'bytes_downloaded', t_bytes_downloaded)
+                                        if t_items_downloaded > 0:
+                                            pipe.hincrby(self.ident, 'items_downloaded', t_items_downloaded)
+                                        if t_items_queued > 0:
+                                            pipe.hincrby(self.ident, 'items_queued', t_items_queued)
+
+                                        pipe.execute()
+
+                                    shipping_count = 0
+
+                            except RedisConnectionError:
+                                logger.info('Log shipper got connection error while '
+                                            'incrementing counts or committing logs with '
+                                            'ident={}, thread={}'.format(self.ident, threading.get_ident()))
+            except RedisError as e:
+                logger.info('Log shipper (ident={}, thread={}) got a Redis error: {!r}'.format(self.ident, threading.get_ident(), e))
 
         logger.info('Log shipper exiting with ident={}, thread={}'
                     .format(self.ident, threading.get_ident()))
         return True
 
     def log(self, packet, ident, log_key):
-        self.log_queue.put({'type': 'log',
-                            'keys': [ident],
-                            'args': [json.dumps(packet), self.log_channel, log_key]
-                           })
+        try:
+            self.log_queue.put_nowait({'type': 'log',
+                'keys': [ident],
+                'args': [json.dumps(packet), self.log_channel, log_key]
+              })
+        except Full:
+            # If the shipping is currently broken, pop an entry off the queue and retry.
+            try:
+                self.log_queue.get_nowait()
+            except Empty:
+                pass
+            else:
+                self.log_queue.task_done()
+                self.log(packet, ident, log_key)
 
     def get_url_file(self, ident):
         try:
['pipeline/archivebot/control.py']