feat(beat): add beat shutdown signals

celery/beat.py

@@ -658,8 +658,10 @@ def sync(self):
 
     def stop(self, wait=False):
         info('beat: Shutting down...')
+        signals.beat_shutting_down.send(sender=self)
         self._is_shutdown.set()
         wait and self._is_stopped.wait()  # block until shutdown done.
+        signals.beat_shutdown.send(sender=self)
 
     def get_scheduler(self, lazy=False,
                       extension_namespace='celery.beat_schedulers'):

celery/signals.py

@@ -137,6 +137,8 @@
 # - Beat
 beat_init = Signal(name='beat_init')
 beat_embedded_init = Signal(name='beat_embedded_init')
+beat_shutting_down = Signal(name='beat_shutting_down')
+beat_shutdown = Signal(name='beat_shutdown')
 
 # - Eventlet
 eventlet_pool_started = Signal(name='eventlet_pool_started')

t/unit/app/test_beat.py

@@ -817,6 +817,19 @@ def test_start_manages_one_tick_before_shutdown(self):
         s.start()
         assert s._is_shutdown.is_set()
 
+    @patch("celery.signals.beat_shutdown")
+    @patch("celery.signals.beat_shutting_down")
+    @patch("celery.signals.beat_init")
+    def test_lifecycle_signals(self, beat_init: Mock, beat_shutting_down: Mock, beat_shutdown: Mock):
+        s, _ = self.get_service()
+        s.scheduler.shutdown_service = s
+        s.start()
+        beat_init.send.assert_called_with(sender=s)
+        s.stop(wait=False)
+        beat_shutting_down.send.assert_called_with(sender=s)
+        s.stop(wait=True)
+        beat_shutdown.send.assert_called_with(sender=s)
+
 
 class test_EmbeddedService: