Add sentry instrumentation for pipeline tasks

Author: r4victor

src/dstack/_internal/server/background/pipeline_tasks/volumes.py

@@ -36,6 +36,7 @@
     emit_volume_status_change_event,
     volume_model_to_volume,
 )
+from dstack._internal.server.utils import sentry_utils
 from dstack._internal.utils.common import get_current_datetime, run_async
 from dstack._internal.utils.logging import get_logger
 
@@ -119,6 +120,7 @@ def __init__(
             queue_check_delay=queue_check_delay,
         )
 
+    @sentry_utils.instrument_named_task("pipeline_tasks.VolumeFetcher.fetch")
     async def fetch(self, limit: int) -> list[VolumePipelineItem]:
         volume_lock, _ = get_locker(get_db().dialect_name).get_lockset(VolumeModel.__tablename__)
         async with volume_lock:
@@ -193,6 +195,7 @@ def __init__(
             heartbeater=heartbeater,
         )
 
+    @sentry_utils.instrument_named_task("pipeline_tasks.VolumeWorker.process")
     async def process(self, item: VolumePipelineItem):
         if item.to_be_deleted:
             await _process_to_be_deleted_item(item)

src/dstack/_internal/server/background/scheduled_tasks/compute_groups.py

@@ -39,7 +39,7 @@ async def process_compute_groups(batch_size: int = 1):
     await asyncio.gather(*tasks)
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def _process_next_compute_group():
     lock, lockset = get_locker(get_db().dialect_name).get_lockset(ComputeGroupModel.__tablename__)
     async with get_session_ctx() as session:

src/dstack/_internal/server/background/scheduled_tasks/events.py

@@ -9,7 +9,7 @@
 from dstack._internal.utils.common import get_current_datetime
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def delete_events():
     cutoff = get_current_datetime() - timedelta(seconds=settings.SERVER_EVENTS_TTL_SECONDS)
     stmt = delete(EventModel).where(EventModel.recorded_at < cutoff)

src/dstack/_internal/server/background/scheduled_tasks/fleets.py

@@ -39,7 +39,7 @@
 MIN_PROCESSING_INTERVAL = timedelta(seconds=30)
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def process_fleets():
     fleet_lock, fleet_lockset = get_locker(get_db().dialect_name).get_lockset(
         FleetModel.__tablename__

src/dstack/_internal/server/background/scheduled_tasks/gateways.py

@@ -35,7 +35,7 @@ async def process_gateways_connections():
     await _process_active_connections()
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def process_gateways():
     lock, lockset = get_locker(get_db().dialect_name).get_lockset(GatewayModel.__tablename__)
     async with get_session_ctx() as session:

src/dstack/_internal/server/background/scheduled_tasks/idle_volumes.py

@@ -27,7 +27,7 @@
 logger = get_logger(__name__)
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def process_idle_volumes():
     lock, lockset = get_locker(get_db().dialect_name).get_lockset(VolumeModel.__tablename__)
     async with get_session_ctx() as session:

src/dstack/_internal/server/background/scheduled_tasks/instances.py

@@ -152,7 +152,7 @@ async def process_instances(batch_size: int = 1):
     await asyncio.gather(*tasks)
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def delete_instance_health_checks():
     now = get_current_datetime()
     cutoff = now - timedelta(seconds=server_settings.SERVER_INSTANCE_HEALTH_TTL_SECONDS)
@@ -163,7 +163,7 @@ async def delete_instance_health_checks():
         await session.commit()
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def _process_next_instance():
     lock, lockset = get_locker(get_db().dialect_name).get_lockset(InstanceModel.__tablename__)
     async with get_session_ctx() as session:

src/dstack/_internal/server/background/scheduled_tasks/metrics.py

@@ -27,7 +27,7 @@
 MIN_COLLECT_INTERVAL_SECONDS = 9
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def collect_metrics():
     async with get_session_ctx() as session:
         res = await session.execute(
@@ -47,7 +47,7 @@ async def collect_metrics():
         await _collect_jobs_metrics(batch)
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def delete_metrics():
     now_timestamp_micro = int(get_current_datetime().timestamp() * 1_000_000)
     running_timestamp_micro_cutoff = (

src/dstack/_internal/server/background/scheduled_tasks/placement_groups.py

@@ -19,7 +19,7 @@
 logger = get_logger(__name__)
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def process_placement_groups():
     lock, lockset = get_locker(get_db().dialect_name).get_lockset(
         PlacementGroupModel.__tablename__

src/dstack/_internal/server/background/scheduled_tasks/prometheus_metrics.py

@@ -35,7 +35,7 @@
 METRICS_TTL_SECONDS = 600
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def collect_prometheus_metrics():
     now = get_current_datetime()
     cutoff = now - timedelta(seconds=MIN_COLLECT_INTERVAL_SECONDS)
@@ -63,7 +63,7 @@ async def collect_prometheus_metrics():
         await _collect_jobs_metrics(batch, now)
 
 
-@sentry_utils.instrument_background_task
+@sentry_utils.instrument_scheduled_task
 async def delete_prometheus_metrics():
     now = get_current_datetime()
     cutoff = now - timedelta(seconds=METRICS_TTL_SECONDS)