queue_processors: Pass the worker_num down into the class.

zerver/management/commands/process_queue.py

@@ -101,7 +101,7 @@ class Command(BaseCommand):
 
             logger.info("Worker %d connecting to queue %s", worker_num, queue_name)
             with log_and_exit_if_exception(logger, queue_name, threaded=False):
-                worker = get_worker(queue_name)
+                worker = get_worker(queue_name, worker_num=worker_num)
                 with configure_scope() as scope:
                     scope.set_tag("queue_worker", queue_name)
                     scope.set_tag("worker_num", worker_num)

zerver/worker/base.py

@@ -97,10 +97,16 @@ class QueueProcessingWorker(ABC):
     # startup and steady-state memory.
     PREFETCH = 100
 
-    def __init__(self, threaded: bool = False, disable_timeout: bool = False) -> None:
+    def __init__(
+        self,
+        threaded: bool = False,
+        disable_timeout: bool = False,
+        worker_num: Optional[int] = None,
+    ) -> None:
         self.q: Optional[SimpleQueueClient] = None
         self.threaded = threaded
         self.disable_timeout = disable_timeout
+        self.worker_num = worker_num
         if not hasattr(self, "queue_name"):
             raise WorkerDeclarationError("Queue worker declared without queue_name")
 

zerver/worker/email_senders.py

@@ -46,8 +46,13 @@ def retry_send_email_failures(
 
 @assign_queue("email_senders")
 class EmailSendingWorker(LoopQueueProcessingWorker):
-    def __init__(self, threaded: bool = False, disable_timeout: bool = False) -> None:
-        super().__init__(threaded, disable_timeout)
+    def __init__(
+        self,
+        threaded: bool = False,
+        disable_timeout: bool = False,
+        worker_num: Optional[int] = None,
+    ) -> None:
+        super().__init__(threaded, disable_timeout, worker_num)
         self.connection: Optional[BaseEmailBackend] = None
 
     @retry_send_email_failures

zerver/worker/queue_processors.py

@@ -1,14 +1,18 @@
 # Documented in https://zulip.readthedocs.io/en/latest/subsystems/queuing.html
 import importlib
 import pkgutil
-from typing import List
+from typing import List, Optional
 
 import zerver.worker
 from zerver.worker.base import QueueProcessingWorker, test_queues, worker_classes
 
 
 def get_worker(
-    queue_name: str, threaded: bool = False, disable_timeout: bool = False
+    queue_name: str,
+    *,
+    threaded: bool = False,
+    disable_timeout: bool = False,
+    worker_num: Optional[int] = None,
 ) -> QueueProcessingWorker:
     if queue_name in {"test", "noop", "noop_batch"}:
         import_module = "zerver.worker.test"
@@ -16,7 +20,9 @@ def get_worker(
         import_module = f"zerver.worker.{queue_name}"
 
     importlib.import_module(import_module)
-    return worker_classes[queue_name](threaded=threaded, disable_timeout=disable_timeout)
+    return worker_classes[queue_name](
+        threaded=threaded, disable_timeout=disable_timeout, worker_num=worker_num
+    )
 
 
 def get_active_worker_queues(only_test_queues: bool = False) -> List[str]:

zerver/worker/test.py

@@ -1,7 +1,7 @@
 # Documented in https://zulip.readthedocs.io/en/latest/subsystems/queuing.html
 import logging
 import time
-from typing import Any, Dict, List, Mapping, Sequence, Set
+from typing import Any, Dict, List, Mapping, Optional, Sequence, Set
 
 import orjson
 from django.conf import settings
@@ -35,10 +35,11 @@ class NoopWorker(QueueProcessingWorker):
         self,
         threaded: bool = False,
         disable_timeout: bool = False,
+        worker_num: Optional[int] = None,
         max_consume: int = 1000,
         slow_queries: Sequence[int] = [],
     ) -> None:
-        super().__init__(threaded, disable_timeout)
+        super().__init__(threaded, disable_timeout, worker_num)
         self.consumed = 0
         self.max_consume = max_consume
         self.slow_queries: Set[int] = set(slow_queries)