fix(taskworker) Improve metrics for scheduler (#93051)

Add a taskname dimension to scheduled tasks so that we can filter and
breakdown scheduler operations.

Author: markstory

src/sentry/taskworker/scheduler/runner.py

@@ -105,6 +105,14 @@ def __repr__(self) -> str:
     def fullname(self) -> str:
         return self._task.fullname
 
+    @property
+    def namespace(self) -> str:
+        return self._task.namespace.name
+
+    @property
+    def taskname(self) -> str:
+        return self._task.name
+
     def set_last_run(self, last_run: datetime | None) -> None:
         self._last_run = last_run
 
@@ -118,7 +126,6 @@ def runtime_after(self, start: datetime) -> datetime:
         return self._schedule.runtime_after(start)
 
     def delay_task(self) -> None:
-        logger.info("taskworker.scheduler.delay_task", extra={"task": self._task.fullname})
         monitor_config = self.monitor_config()
         headers: dict[str, Any] | None = None
         if monitor_config:
@@ -223,13 +230,21 @@ def _try_spawn(self, entry: ScheduleEntry) -> None:
             entry.delay_task()
             entry.set_last_run(now)
 
-            logger.info("taskworker.scheduler.delay_task", extra={"task": entry.fullname})
-            metrics.incr("taskworker.scheduler.delay_task")
+            logger.info("taskworker.scheduler.delay_task", extra={"fullname": entry.fullname})
+            metrics.incr(
+                "taskworker.scheduler.delay_task",
+                tags={
+                    "taskname": entry.taskname,
+                    "namespace": entry.namespace,
+                },
+            )
         else:
             # sync with last_run state in storage
             entry.set_last_run(self._run_storage.read(entry.fullname))
 
-            logger.info("taskworker.scheduler.sync_with_storage", extra={"task": entry.fullname})
+            logger.info(
+                "taskworker.scheduler.sync_with_storage", extra={"fullname": entry.fullname}
+            )
             metrics.incr("taskworker.scheduler.sync_with_storage")
 
     def _update_heap(self) -> None:

src/sentry/taskworker/task.py

@@ -64,6 +64,10 @@ def __init__(
     def fullname(self) -> str:
         return f"{self._namespace.name}:{self.name}"
 
+    @property
+    def namespace(self) -> TaskNamespace:
+        return self._namespace
+
     @property
     def retry(self) -> Retry | None:
         return self._retry

tests/sentry/taskworker/test_task.py

@@ -30,6 +30,7 @@ def test_define_task_defaults(task_namespace: TaskNamespace) -> None:
     task = Task(name="test.do_things", func=do_things, namespace=task_namespace)
     assert task.retry is None
     assert task.name == "test.do_things"
+    assert task.namespace == task_namespace
 
 
 def test_define_task_retry(task_namespace: TaskNamespace) -> None: