Add initial code. Tests still need to be written.

Close #3821

pulpcore/tasking/_util.py

@@ -8,6 +8,7 @@
 import threading
 import time
 from gettext import gettext as _
+from opentelemetry.metrics import get_meter, Observation
 
 from django.conf import settings
 from django.db import connection, transaction
@@ -26,6 +27,7 @@
 from pulpcore.constants import TASK_FINAL_STATES, TASK_STATES, VAR_TMP_PULP
 from pulpcore.exceptions import AdvisoryLockError
 from pulpcore.tasking.tasks import dispatch, execute_task
+from pulpcore.app.models import Worker
 
 _logger = logging.getLogger(__name__)
 
@@ -58,8 +60,25 @@ def __exit__(self, exc_type, exc_value, traceback):
             raise RuntimeError("Lock not held.")
 
 
+def configure_telemetry():
+    def report_online_workers():
+        while True:
+            yield [
+                Observation(Worker.objects.online().count())
+            ]
+
+    meter = get_meter("pulpcore.tasking.worker")
+    worker_counter = meter.create_observable_gauge(
+        name="workers_online",
+        description="The number of online workers.",
+        unit="workers",
+        callbacks=[report_online_workers()]
+    )
+
+
 def startup_hook():
     configure_analytics()
+    configure_telemetry()
     configure_cleanup()
 
 

pulpcore/tasking/worker.py

@@ -125,6 +125,7 @@ def handle_worker_heartbeat(self):
 
     def shutdown(self):
         self.worker.delete()
+        # self.worker_counter.add(-1, attributes={"name": self.name})
         _logger.info(_("Worker %s was shut down."), self.name)
 
     def worker_cleanup(self):