Merge pull request spulec#59 from spulec/fix-checking-ppid

Change orphan child checking of parent id.

Author: spulec

development.txt

@@ -1,6 +1,6 @@
 coverage==4.4.1
 mock==1.0.1
-moto==1.3.4
+moto==1.3.8
 nose==1.3.0
 pre-commit==0.7.6
 sure==1.2.2

pyqs/worker.py

@@ -38,6 +38,7 @@ def get_conn(region=None, access_key_id=None, secret_access_key=None):
 
 class BaseWorker(Process):
     def __init__(self, *args, **kwargs):
+        self.parent_id = kwargs.pop('parent_id')
         super(BaseWorker, self).__init__(*args, **kwargs)
         self.should_exit = Event()
 
@@ -48,7 +49,7 @@ def shutdown(self):
         self.should_exit.set()
 
     def parent_is_alive(self):
-        if os.getppid() == 1:
+        if os.getppid() != self.parent_id:
             logger.info(
                 "Parent process has gone away, exiting process {}!".format(
                     os.getpid()))
@@ -255,6 +256,7 @@ def __init__(self, queue_prefixes, worker_concurrency, interval, batchsize,
         self.setup_internal_queue(worker_concurrency)
         self.reader_children = []
         self.worker_children = []
+        self._pid = os.getpid()
         self._initialize_reader_children()
         self._initialize_worker_children(worker_concurrency)
         self._running = True
@@ -271,6 +273,7 @@ def _initialize_reader_children(self):
                 ReadWorker(
                     queue_url, self.internal_queue, self.batchsize,
                     connection_args=self.connection_args,
+                    parent_id=self._pid,
                 )
             )
 
@@ -280,6 +283,7 @@ def _initialize_worker_children(self, number):
                 ProcessWorker(
                     self.internal_queue, self.interval,
                     connection_args=self.connection_args,
+                    parent_id=self._pid,
                 )
             )
 
@@ -367,6 +371,7 @@ def _replace_reader_children(self):
                 worker = ReadWorker(
                     queue_url, self.internal_queue, self.batchsize,
                     connection_args=self.connection_args,
+                    parent_id=self._pid,
                 )
                 worker.start()
                 self.reader_children.append(worker)
@@ -381,6 +386,7 @@ def _replace_worker_children(self):
                 worker = ProcessWorker(
                     self.internal_queue, self.interval,
                     connection_args=self.connection_args,
+                    parent_id=self._pid,
                 )
                 worker.start()
                 self.worker_children.append(worker)

tests/test_manager_worker.py

@@ -100,7 +100,7 @@ def test_main_method(ManagerWorker):
 
     ManagerWorker.assert_called_once_with(
         ['email1', 'email2'], 2, 1, 10, prefetch_multiplier=2,
-        region='us-east-1', secret_access_key=None, access_key_id=None,
+        region=None, secret_access_key=None, access_key_id=None,
     )
     ManagerWorker.return_value.start.assert_called_once_with()
 

tests/test_worker.py

@@ -43,7 +43,7 @@ def test_worker_fills_internal_queue():
     conn.send_message(QueueUrl=queue_url, MessageBody=message)
 
     internal_queue = Queue()
-    worker = ReadWorker(queue_url, internal_queue, BATCHSIZE)
+    worker = ReadWorker(queue_url, internal_queue, BATCHSIZE, parent_id=1)
     worker.read_message()
 
     packed_message = internal_queue.get(timeout=1)
@@ -78,7 +78,7 @@ def test_worker_fills_internal_queue_only_until_maximum_queue_size():
         conn.send_message(QueueUrl=queue_url, MessageBody=message)
 
     internal_queue = Queue(maxsize=2)
-    worker = ReadWorker(queue_url, internal_queue, BATCHSIZE)
+    worker = ReadWorker(queue_url, internal_queue, BATCHSIZE, parent_id=1)
     worker.read_message()
 
     # The internal queue should only have two messages on it
@@ -110,7 +110,7 @@ def test_worker_fills_internal_queue_from_celery_task():
     conn.send_message(QueueUrl=queue_url, MessageBody=message)
 
     internal_queue = Queue()
-    worker = ReadWorker(queue_url, internal_queue, BATCHSIZE)
+    worker = ReadWorker(queue_url, internal_queue, BATCHSIZE, parent_id=1)
     worker.read_message()
 
     packed_message = internal_queue.get(timeout=1)
@@ -159,7 +159,7 @@ def test_worker_processes_tasks_from_internal_queue():
     )
 
     # Process message
-    worker = ProcessWorker(internal_queue, INTERVAL)
+    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
     worker.process_message()
 
     task_results.should.equal(['Test message'])
@@ -203,7 +203,7 @@ def test_worker_fills_internal_queue_and_respects_visibility_timeouts():
 
     # Run Reader
     internal_queue = Queue(maxsize=1)
-    worker = ReadWorker(queue_url, internal_queue, BATCHSIZE)
+    worker = ReadWorker(queue_url, internal_queue, BATCHSIZE, parent_id=1)
     worker.read_message()
 
     # Check log messages
@@ -251,7 +251,7 @@ def test_worker_processes_tasks_and_logs_correctly():
     )
 
     # Process message
-    worker = ProcessWorker(internal_queue, INTERVAL)
+    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
     worker.process_message()
 
     # Check output
@@ -301,7 +301,7 @@ def test_worker_processes_tasks_and_logs_warning_correctly():
     )
 
     # Process message
-    worker = ProcessWorker(internal_queue, INTERVAL)
+    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
     worker.process_message()
 
     # Check output
@@ -314,7 +314,7 @@ def test_worker_processes_tasks_and_logs_warning_correctly():
     logger.handlers[0].messages['error'][0].lower().should.contain(
         msg1.lower())
     msg2 = (
-        'raise ValueError("Need to be given basestring, was given '
+        '"Need to be given basestring, was given '
         '{}".format(message))\nValueError: Need to be given basestring, '
         'was given 23'
     )  # noqa
@@ -329,7 +329,7 @@ def test_worker_processes_empty_queue():
     """
     internal_queue = Queue()
 
-    worker = ProcessWorker(internal_queue, INTERVAL)
+    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
     worker.process_message()
 
 
@@ -338,9 +338,9 @@ def test_parent_process_death(os):
     """
     Test worker processes recognize parent process death
     """
-    os.getppid.return_value = 1
+    os.getppid.return_value = 123
 
-    worker = BaseWorker()
+    worker = BaseWorker(parent_id=1)
     worker.parent_is_alive().should.be.false
 
 
@@ -351,7 +351,7 @@ def test_parent_process_alive(os):
     """
     os.getppid.return_value = 1234
 
-    worker = BaseWorker()
+    worker = BaseWorker(parent_id=1234)
     worker.parent_is_alive().should.be.true
 
 
@@ -366,14 +366,14 @@ def test_read_worker_with_parent_process_alive_and_should_not_exit(os):
     queue_url = conn.create_queue(QueueName="tester")['QueueUrl']
 
     # Setup PPID
-    os.getppid.return_value = 1234
+    os.getppid.return_value = 1
 
     # Setup dummy read_message
     def read_message():
         raise Exception("Called")
 
     # When I have a parent process, and shutdown is not set
-    worker = ReadWorker(queue_url, "foo", BATCHSIZE)
+    worker = ReadWorker(queue_url, "foo", BATCHSIZE, parent_id=1)
     worker.read_message = read_message
 
     # Then read_message() is reached
@@ -397,7 +397,7 @@ def test_read_worker_with_parent_process_alive_and_should_exit(os):
     q = Queue(1)
 
     # When I have a parent process, and shutdown is set
-    worker = ReadWorker(queue_url, q, BATCHSIZE)
+    worker = ReadWorker(queue_url, q, BATCHSIZE, parent_id=1)
     worker.read_message = Mock()
     worker.shutdown()
 
@@ -416,13 +416,13 @@ def test_read_worker_with_parent_process_dead_and_should_not_exit(os):
     queue_url = conn.create_queue(QueueName="tester")['QueueUrl']
 
     # Setup PPID
-    os.getppid.return_value = 1
+    os.getppid.return_value = 123
 
     # Setup internal queue
     q = Queue(1)
 
     # When I have no parent process, and shutdown is not set
-    worker = ReadWorker(queue_url, q, BATCHSIZE)
+    worker = ReadWorker(queue_url, q, BATCHSIZE, parent_id=1)
     worker.read_message = Mock()
 
     # Then I return from run()
@@ -437,14 +437,14 @@ def test_process_worker_with_parent_process_alive_and_should_not_exit(os):
     is not set
     """
     # Setup PPID
-    os.getppid.return_value = 1234
+    os.getppid.return_value = 1
 
     # Setup dummy read_message
     def process_message():
         raise Exception("Called")
 
     # When I have a parent process, and shutdown is not set
-    worker = ProcessWorker("foo", INTERVAL)
+    worker = ProcessWorker("foo", INTERVAL, parent_id=1)
     worker.process_message = process_message
 
     # Then process_message() is reached
@@ -461,7 +461,7 @@ def test_process_worker_with_parent_process_dead_and_should_not_exit(os):
     os.getppid.return_value = 1
 
     # When I have no parent process, and shutdown is not set
-    worker = ProcessWorker("foo", INTERVAL)
+    worker = ProcessWorker("foo", INTERVAL, parent_id=1)
     worker.process_message = Mock()
 
     # Then I return from run()
@@ -478,7 +478,7 @@ def test_process_worker_with_parent_process_alive_and_should_exit(os):
     os.getppid.return_value = 1234
 
     # When I have a parent process, and shutdown is set
-    worker = ProcessWorker("foo", INTERVAL)
+    worker = ProcessWorker("foo", INTERVAL, parent_id=1)
     worker.process_message = Mock()
     worker.shutdown()
 
@@ -487,10 +487,14 @@ def test_process_worker_with_parent_process_alive_and_should_exit(os):
 
 
 @mock_sqs
-def test_worker_processes_shuts_down_after_processing_its_max_number_of_msgs():
+@patch("pyqs.worker.os")
+def test_worker_processes_shuts_down_after_processing_its_max_number_of_msgs(
+        os):
     """
     Test worker processes shutdown after processing maximum number of messages
     """
+    os.getppid.return_value = 1
+
     # Setup SQS Queue
     conn = boto3.client('sqs', region_name='us-east-1')
     queue_url = conn.create_queue(QueueName="tester")['QueueUrl']
@@ -535,7 +539,7 @@ def test_worker_processes_shuts_down_after_processing_its_max_number_of_msgs():
     )
 
     # When I Process messages
-    worker = ProcessWorker(internal_queue, INTERVAL)
+    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
     worker._messages_to_process_before_shutdown = 2
 
     # Then I return from run()
@@ -584,7 +588,7 @@ def test_worker_processes_discard_tasks_that_exceed_their_visibility_timeout():
     )
 
     # When I process the message
-    worker = ProcessWorker(internal_queue, INTERVAL)
+    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
     worker.process_message()
 
     # Then I get an error about exceeding the visibility timeout
@@ -648,7 +652,7 @@ def sleep_and_queue(internal_queue):
     thread.start()
 
     # When I Process messages
-    worker = ProcessWorker(internal_queue, INTERVAL)
+    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
     worker._messages_to_process_before_shutdown = 2
 
     # Then I return from run() after processing 2 messages