email-worker: Create EmailSendingWorker.

This commit just copies all the code from MissedMessageSendingWorker
class to a new EmailSendingWorker class. All the logic to send an email
through a queue was already there. This commit only makes the logic
generic. It does so by creating a special purpose queue called
'email_senders' to send any type of email. To make
MissedMessageSendingWorker still work we derive it from
EmailSendingWorker. All the tests that were testing
MissedMessageSendingWorker now run against EmailSendingWorker.

puppet/zulip/manifests/base.pp

@@ -47,6 +47,7 @@ class zulip::base {
              'feedback_messages',
              'invites',
              'missedmessage_email_senders',
+             'email_senders',
              'missedmessage_emails',
              'missedmessage_mobile_notifications',
              'outgoing_webhooks',

puppet/zulip_ops/files/nagios3/conf.d/services.cfg

@@ -555,6 +555,15 @@ define service {
         contact_groups                  admins
 }
 
+define service {
+        use                             generic-service
+        service_description             Check email_senders queue processor
+        check_command                   check_remote_arg_string!manage.py process_queue --queue_name=email_senders!1:1!1:1
+        max_check_attempts              3
+        hostgroup_name                  frontends
+        contact_groups                  admins
+}
+
 define service {
         use                             generic-service
         service_description             Check outgoing webhooks queue processor

scripts/nagios/check-rabbitmq-consumers

@@ -51,6 +51,7 @@ queues = {
     'message_sender',
     'missedmessage_emails',
     'missedmessage_email_senders',
+    'email_senders',
     'missedmessage_mobile_notifications',
     'outgoing_webhooks',
     'notify_tornado',

tools/test-queue-worker-reload

@@ -28,6 +28,7 @@ successful_worker_launches = [
     'launching queue worker thread test',
     'launching queue worker thread message_sender',
     'launching queue worker thread missedmessage_emails',
+    'launching queue worker thread email_senders',
     'launching queue worker thread missedmessage_email_senders',
     'launching queue worker thread email_mirror',
     'launching queue worker thread user_activity_interval',

zerver/tests/test_queue_worker.py

@@ -17,6 +17,7 @@ from zerver.worker import queue_processors
 from zerver.worker.queue_processors import (
     get_active_worker_queues,
     QueueProcessingWorker,
+    EmailSendingWorker,
     LoopQueueProcessingWorker,
     MissedMessageWorker,
 )
@@ -171,7 +172,7 @@ class WorkerTest(ZulipTestCase):
         fake_client = self.FakeClient()
 
         data = {'test': 'test', 'id': 'test_missed'}
-        fake_client.queue.append(('missedmessage_email_senders', data))
+        fake_client.queue.append(('email_senders', data))
 
         def fake_publish(queue_name: str,
                          event: Dict[str, Any],
@@ -179,7 +180,7 @@ class WorkerTest(ZulipTestCase):
             fake_client.queue.append((queue_name, event))
 
         with simulated_queue_client(lambda: fake_client):
-            worker = queue_processors.MissedMessageSendingWorker()
+            worker = queue_processors.EmailSendingWorker()
             worker.setup()
             with patch('zerver.worker.queue_processors.send_email_from_dict',
                        side_effect=smtplib.SMTPServerDisconnected), \
@@ -332,6 +333,7 @@ class WorkerTest(ZulipTestCase):
 
     def test_get_active_worker_queues(self) -> None:
         worker_queue_count = (len(QueueProcessingWorker.__subclasses__()) +
+                              len(EmailSendingWorker.__subclasses__()) +
                               len(LoopQueueProcessingWorker.__subclasses__()) - 1)
         self.assertEqual(worker_queue_count, len(get_active_worker_queues()))
         self.assertEqual(1, len(get_active_worker_queues(queue_type='test')))

zerver/worker/queue_processors.py

@@ -298,8 +298,8 @@ class MissedMessageWorker(LoopQueueProcessingWorker):
         for user_profile_id, events in by_recipient.items():
             handle_missedmessage_emails(user_profile_id, events)
 
-@assign_queue('missedmessage_email_senders')
-class MissedMessageSendingWorker(QueueProcessingWorker):
+@assign_queue('email_senders')
+class EmailSendingWorker(QueueProcessingWorker):
     @retry_send_email_failures
     def consume(self, data):
         # type: (Dict[str, Any]) -> None
@@ -309,6 +309,19 @@ class MissedMessageSendingWorker(QueueProcessingWorker):
             # TODO: Do something smarter here ..
             pass
 
+@assign_queue('missedmessage_email_senders')
+class MissedMessageSendingWorker(EmailSendingWorker):
+    """
+    Note: Class decorators are not inherited.
+
+    The `missedmessage_email_senders` queue was used up through 1.7.1, so we
+    keep consuming from it in case we've just upgraded from an old version.
+    After the 1.8 release, we can delete it and tell admins to upgrade to 1.8
+    first.
+    """
+    # TODO: zulip-1.8: Delete code related to missedmessage_email_senders queue.
+    pass
+
 @assign_queue('missedmessage_mobile_notifications')
 class PushNotificationsWorker(QueueProcessingWorker):
     def consume(self, data):