Fix pylint and Sphinx CI failures

- Add __init__ to CeleryInstrumentor and CeleryWorkerInstrumentor
  to satisfy attribute-defined-outside-init (W0201)
- Remove generic type parameter from CeleryGetter to fix Sphinx
  reference resolution for Request
- Split test_metrics.py into two files to stay under 1000 lines
- Mark _run_task as @staticmethod (R6301)
- Initialize einfo before try block (E0601)
- Add CHANGELOG entries

Author: diogosilva30

CHANGELOG.md

@@ -15,6 +15,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Bump `pylint` to `4.0.5`
   ([#4244](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4244))
+- `opentelemetry-instrumentation-celery`: Add task and worker lifecycle metrics matching Celery Flower
+  ([#4439](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4439))
+
+### Fixed
+
+- `opentelemetry-instrumentation-celery`: Fix memory leak in `task_id_to_start_time` dict never being cleaned up
+  ([#4439](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4439))
 
 ### Breaking changes
 

instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py

@@ -134,7 +134,7 @@ def init_celery_tracing(*args, **kwargs):
 _TASK_NAME_KEY = "celery.task_name"
 
 
-class CeleryGetter(Getter["Request"]):
+class CeleryGetter(Getter):
     def get(self, carrier: "Request", key: str) -> list[str] | None:
         value = getattr(carrier, key, None)
         if value is None:
@@ -280,6 +280,14 @@ class CeleryInstrumentor(BaseInstrumentor):
     Must be initialized in the worker subprocess via the
     ``worker_process_init`` signal."""
 
+    def __init__(self) -> None:
+        super().__init__()
+        self.metrics: Optional[_CeleryTaskMetrics] = None
+        self.task_id_to_start_time: dict = {}
+        self.task_id_to_received_time: dict = {}
+        self.prefetched_task_id_to_labels: dict = {}
+        self.executing_task_id_to_worker: dict = {}
+
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
@@ -810,6 +818,11 @@ def init_worker_metrics(sender, instance, conf, **kwargs):
             CeleryWorkerInstrumentor().instrument()
     """
 
+    def __init__(self) -> None:
+        super().__init__()
+        self.metrics: Optional[_CeleryWorkerMetrics] = None
+        self.online_workers: set = set()
+
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments