From a61963fc04aedb430326fe814c07db13063a2d4b Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Tue, 14 Jul 2026 13:39:53 +0500 Subject: [PATCH 1/9] Support OTel tracing --- pyproject.toml | 11 ++- src/dstack/_internal/server/app.py | 4 +- .../pipeline_tasks/compute_groups.py | 6 +- .../background/pipeline_tasks/fleets.py | 6 +- .../pipeline_tasks/gateway_replicas.py | 6 +- .../background/pipeline_tasks/gateways.py | 6 +- .../pipeline_tasks/instances/__init__.py | 6 +- .../background/pipeline_tasks/jobs_running.py | 6 +- .../pipeline_tasks/jobs_submitted.py | 6 +- .../pipeline_tasks/jobs_terminating.py | 6 +- .../pipeline_tasks/placement_groups.py | 6 +- .../pipeline_tasks/runs/__init__.py | 6 +- .../service_router_worker_sync.py | 6 +- .../background/pipeline_tasks/volumes.py | 6 +- .../background/scheduled_tasks/events.py | 4 +- .../scheduled_tasks/idle_volumes.py | 4 +- .../scheduled_tasks/instance_healthchecks.py | 4 +- .../background/scheduled_tasks/metrics.py | 6 +- .../scheduled_tasks/prometheus_metrics.py | 6 +- src/dstack/_internal/server/settings.py | 11 +++ src/dstack/_internal/server/utils/common.py | 8 ++ .../_internal/server/utils/otel/__init__.py | 37 ++++++++ .../_internal/server/utils/otel/utils.py | 87 +++++++++++++++++++ .../_internal/server/utils/sentry_utils.py | 33 +------ src/dstack/_internal/server/utils/tracing.py | 28 ++++++ src/dstack/_internal/utils/common.py | 6 +- .../_internal/server/utils/test_common.py | 17 +++- .../_internal/server/utils/test_tracing.py | 61 +++++++++++++ 28 files changed, 316 insertions(+), 83 deletions(-) create mode 100644 src/dstack/_internal/server/utils/otel/__init__.py create mode 100644 src/dstack/_internal/server/utils/otel/utils.py create mode 100644 src/dstack/_internal/server/utils/tracing.py create mode 100644 src/tests/_internal/server/utils/test_tracing.py diff --git a/pyproject.toml b/pyproject.toml index b61f974ed3..2bf79c5aee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -264,9 +264,18 @@ fluentbit = [ "elasticsearch>=8.0.0", "dstack[server]", ] +otel = [ + "opentelemetry-sdk>=1.30.0", + "opentelemetry-exporter-otlp-proto-http>=1.30.0", + "opentelemetry-instrumentation-fastapi>=0.51b0", + "opentelemetry-instrumentation-sqlalchemy>=0.51b0", + "opentelemetry-instrumentation-httpx>=0.51b0", + "opentelemetry-instrumentation-requests>=0.51b0", + "dstack[server]", +] crusoe = [ "dstack[server]", ] all = [ - "dstack[gateway,server,aws,azure,gcp,verda,kubernetes,lambda,nebius,oci,crusoe,fluentbit]", + "dstack[gateway,server,aws,azure,gcp,verda,kubernetes,lambda,nebius,oci,crusoe,fluentbit,otel]", ] diff --git a/src/dstack/_internal/server/app.py b/src/dstack/_internal/server/app.py index 0f02806aa4..7ee097a0cb 100644 --- a/src/dstack/_internal/server/app.py +++ b/src/dstack/_internal/server/app.py @@ -67,7 +67,7 @@ SERVER_URL, UPDATE_DEFAULT_PROJECT, ) -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import otel, sentry_utils from dstack._internal.server.utils.logging import configure_logging from dstack._internal.server.utils.routers import ( CustomORJSONResponse, @@ -120,6 +120,8 @@ async def lifespan(app: FastAPI): profiles_sample_rate=settings.SENTRY_PROFILES_SAMPLE_RATE, before_send=sentry_utils.AsyncioCancelledErrorFilterEventProcessor(), ) + if settings.ENABLE_OTEL_TRACES: + otel.configure_tracing(app, get_db().engine) server_executor = ThreadPoolExecutor(max_workers=settings.SERVER_EXECUTOR_MAX_WORKERS) asyncio.get_running_loop().set_default_executor(server_executor) await migrate() diff --git a/src/dstack/_internal/server/background/pipeline_tasks/compute_groups.py b/src/dstack/_internal/server/background/pipeline_tasks/compute_groups.py index 78f23495db..5cccbd5204 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/compute_groups.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/compute_groups.py @@ -33,7 +33,7 @@ from dstack._internal.server.services.instances import emit_instance_status_change_event from dstack._internal.server.services.locking import get_locker from dstack._internal.server.services.pipelines import PipelineHinterProtocol -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime, run_async from dstack._internal.utils.logging import get_logger @@ -120,7 +120,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("ComputeGroupFetcher.fetch") + @tracing.instrument_pipeline_task("ComputeGroupFetcher.fetch") async def fetch(self, limit: int) -> list[PipelineItem]: compute_group_lock, _ = get_locker(get_db().dialect_name).get_lockset( ComputeGroupModel.__tablename__ @@ -188,7 +188,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("ComputeGroupWorker.process") + @tracing.instrument_pipeline_task("ComputeGroupWorker.process") async def process(self, item: PipelineItem): async with get_session_ctx() as session: res = await session.execute( diff --git a/src/dstack/_internal/server/background/pipeline_tasks/fleets.py b/src/dstack/_internal/server/background/pipeline_tasks/fleets.py index 8050c552c8..6b065557b2 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/fleets.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/fleets.py @@ -56,7 +56,7 @@ ) from dstack._internal.server.services.locking import get_locker from dstack._internal.server.services.pipelines import PipelineHinterProtocol -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime from dstack._internal.utils.logging import get_logger @@ -140,7 +140,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("FleetFetcher.fetch") + @tracing.instrument_pipeline_task("FleetFetcher.fetch") async def fetch(self, limit: int) -> list[PipelineItem]: fleet_lock, _ = get_locker(get_db().dialect_name).get_lockset(FleetModel.__tablename__) async with fleet_lock: @@ -209,7 +209,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("FleetWorker.process") + @tracing.instrument_pipeline_task("FleetWorker.process") async def process(self, item: PipelineItem): process_context = await _load_process_context(item) if process_context is None: diff --git a/src/dstack/_internal/server/background/pipeline_tasks/gateway_replicas.py b/src/dstack/_internal/server/background/pipeline_tasks/gateway_replicas.py index 29370a07b9..1c67f6133d 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/gateway_replicas.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/gateway_replicas.py @@ -38,7 +38,7 @@ from dstack._internal.server.services.locking import get_locker from dstack._internal.server.services.logging import fmt from dstack._internal.server.services.pipelines import PipelineHinterProtocol -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime, run_async from dstack._internal.utils.logging import get_logger @@ -127,7 +127,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("GatewayReplicaFetcher.fetch") + @tracing.instrument_pipeline_task("GatewayReplicaFetcher.fetch") async def fetch(self, limit: int) -> list[GatewayReplicaPipelineItem]: replica_lock, _ = get_locker(get_db().dialect_name).get_lockset( GatewayComputeModel.__tablename__ @@ -227,7 +227,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("GatewayReplicaWorker.process") + @tracing.instrument_pipeline_task("GatewayReplicaWorker.process") async def process(self, item: GatewayReplicaPipelineItem): if item.status == GatewayReplicaStatus.SUBMITTED: await _process_submitted_item(item) diff --git a/src/dstack/_internal/server/background/pipeline_tasks/gateways.py b/src/dstack/_internal/server/background/pipeline_tasks/gateways.py index 8e71568ccc..54b99708e2 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/gateways.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/gateways.py @@ -43,7 +43,7 @@ from dstack._internal.server.services.locking import get_locker from dstack._internal.server.services.logging import fmt from dstack._internal.server.services.pipelines import PipelineHinterProtocol -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime from dstack._internal.utils.logging import get_logger @@ -133,7 +133,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("GatewayFetcher.fetch") + @tracing.instrument_pipeline_task("GatewayFetcher.fetch") async def fetch(self, limit: int) -> list[GatewayPipelineItem]: gateway_lock, _ = get_locker(get_db().dialect_name).get_lockset(GatewayModel.__tablename__) async with gateway_lock: @@ -211,7 +211,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("GatewayWorker.process") + @tracing.instrument_pipeline_task("GatewayWorker.process") async def process(self, item: GatewayPipelineItem): if item.to_be_deleted: await _process_to_be_deleted_item(item) diff --git a/src/dstack/_internal/server/background/pipeline_tasks/instances/__init__.py b/src/dstack/_internal/server/background/pipeline_tasks/instances/__init__.py index 343b05f813..7b0cae9a1d 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/instances/__init__.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/instances/__init__.py @@ -56,7 +56,7 @@ from dstack._internal.server.services.placement import ( schedule_fleet_placement_groups_deletion, ) -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime from dstack._internal.utils.logging import get_logger @@ -152,7 +152,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("InstanceFetcher.fetch") + @tracing.instrument_pipeline_task("InstanceFetcher.fetch") async def fetch(self, limit: int) -> list[InstancePipelineItem]: instance_lock, _ = get_locker(get_db().dialect_name).get_lockset( InstanceModel.__tablename__ @@ -277,7 +277,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("InstanceWorker.process") + @tracing.instrument_pipeline_task("InstanceWorker.process") async def process(self, item: InstancePipelineItem): process_context: Optional[_ProcessContext] = None if item.status == InstanceStatus.PENDING: diff --git a/src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py b/src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py index 61599172b5..685d15cc88 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py @@ -110,7 +110,7 @@ ) from dstack._internal.server.services.secrets import get_project_secrets_mapping from dstack._internal.server.services.storage import get_default_storage -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime, get_or_error, run_async from dstack._internal.utils.interpolator import InterpolatorError from dstack._internal.utils.logging import get_logger @@ -209,7 +209,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("JobRunningFetcher.fetch") + @tracing.instrument_pipeline_task("JobRunningFetcher.fetch") async def fetch(self, limit: int) -> list[JobRunningPipelineItem]: job_lock, _ = get_locker(get_db().dialect_name).get_lockset(JobModel.__tablename__) async with job_lock: @@ -308,7 +308,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("JobRunningWorker.process") + @tracing.instrument_pipeline_task("JobRunningWorker.process") async def process(self, item: JobRunningPipelineItem): context = await _load_process_context(item=item) if context is None: diff --git a/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py b/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py index 0ae9bb4432..1c782ef943 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py @@ -139,7 +139,7 @@ ) from dstack._internal.server.services.secrets import get_project_secrets_mapping from dstack._internal.server.services.volumes import volume_model_to_volume -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime, get_or_error, run_async from dstack._internal.utils.interpolator import InterpolatorError from dstack._internal.utils.logging import get_logger @@ -229,7 +229,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("JobSubmittedFetcher.fetch") + @tracing.instrument_pipeline_task("JobSubmittedFetcher.fetch") async def fetch(self, limit: int) -> list[JobSubmittedPipelineItem]: now = get_current_datetime() if limit <= 0: @@ -314,7 +314,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("JobSubmittedWorker.process") + @tracing.instrument_pipeline_task("JobSubmittedWorker.process") async def process(self, item: JobSubmittedPipelineItem): context = await _load_process_context(item=item) if context is None: diff --git a/src/dstack/_internal/server/background/pipeline_tasks/jobs_terminating.py b/src/dstack/_internal/server/background/pipeline_tasks/jobs_terminating.py index be8d80948d..8b080661a0 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/jobs_terminating.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/jobs_terminating.py @@ -72,7 +72,7 @@ from dstack._internal.server.services.volumes import ( volume_model_to_volume, ) -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils import common from dstack._internal.utils.common import get_current_datetime, get_or_error from dstack._internal.utils.logging import get_logger @@ -162,7 +162,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("JobTerminatingFetcher.fetch") + @tracing.instrument_pipeline_task("JobTerminatingFetcher.fetch") async def fetch(self, limit: int) -> list[JobTerminatingPipelineItem]: job_lock, _ = get_locker(get_db().dialect_name).get_lockset(JobModel.__tablename__) async with job_lock: @@ -248,7 +248,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("JobTerminatingWorker.process") + @tracing.instrument_pipeline_task("JobTerminatingWorker.process") async def process(self, item: JobTerminatingPipelineItem): async with get_session_ctx() as session: job_model = await _refetch_locked_job(session=session, item=item) diff --git a/src/dstack/_internal/server/background/pipeline_tasks/placement_groups.py b/src/dstack/_internal/server/background/pipeline_tasks/placement_groups.py index e5160a8ea1..ff8439dadc 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/placement_groups.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/placement_groups.py @@ -33,7 +33,7 @@ from dstack._internal.server.services.locking import get_locker from dstack._internal.server.services.pipelines import PipelineHinterProtocol from dstack._internal.server.services.placement import placement_group_model_to_placement_group -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime, run_async from dstack._internal.utils.logging import get_logger @@ -117,7 +117,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("PlacementGroupFetcher.fetch") + @tracing.instrument_pipeline_task("PlacementGroupFetcher.fetch") async def fetch(self, limit: int) -> list[PipelineItem]: placement_group_lock, _ = get_locker(get_db().dialect_name).get_lockset( PlacementGroupModel.__tablename__ @@ -187,7 +187,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("PlacementGroupWorker.process") + @tracing.instrument_pipeline_task("PlacementGroupWorker.process") async def process(self, item: PipelineItem): async with get_session_ctx() as session: res = await session.execute( diff --git a/src/dstack/_internal/server/background/pipeline_tasks/runs/__init__.py b/src/dstack/_internal/server/background/pipeline_tasks/runs/__init__.py index 071af9fbd3..40aae39bc9 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/runs/__init__.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/runs/__init__.py @@ -35,7 +35,7 @@ from dstack._internal.server.services.prometheus.client_metrics import run_metrics from dstack._internal.server.services.runs import emit_run_status_change_event, get_run_spec from dstack._internal.server.services.secrets import get_project_secrets_mapping -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime from dstack._internal.utils.logging import get_logger @@ -129,7 +129,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("RunFetcher.fetch") + @tracing.instrument_pipeline_task("RunFetcher.fetch") async def fetch(self, limit: int) -> list[RunPipelineItem]: if limit <= 0: return [] @@ -246,7 +246,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("RunWorker.process") + @tracing.instrument_pipeline_task("RunWorker.process") async def process(self, item: RunPipelineItem): # Currently `dstack` supports runs with # * one multi-node replica (multi-node tasks) diff --git a/src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py b/src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py index 2b416fb823..5ded45afe6 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py @@ -36,7 +36,7 @@ run_model_has_sglang_router_replica_group, sync_router_workers_for_run_model, ) -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime from dstack._internal.utils.logging import get_logger @@ -107,7 +107,7 @@ def _workers(self) -> Sequence["ServiceRouterWorkerSyncWorker"]: class ServiceRouterWorkerSyncFetcher(Fetcher[ServiceRouterWorkerSyncPipelineItem]): - @sentry_utils.instrument_pipeline_task("ServiceRouterWorkerSyncFetcher.fetch") + @tracing.instrument_pipeline_task("ServiceRouterWorkerSyncFetcher.fetch") async def fetch(self, limit: int) -> list[ServiceRouterWorkerSyncPipelineItem]: sync_lock, _ = get_locker(get_db().dialect_name).get_lockset( ServiceRouterWorkerSyncModel.__tablename__ @@ -192,7 +192,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("ServiceRouterWorkerSyncWorker.process") + @tracing.instrument_pipeline_task("ServiceRouterWorkerSyncWorker.process") async def process(self, item: ServiceRouterWorkerSyncPipelineItem) -> None: async with get_session_ctx() as session: res = await session.execute( diff --git a/src/dstack/_internal/server/background/pipeline_tasks/volumes.py b/src/dstack/_internal/server/background/pipeline_tasks/volumes.py index cb8d6ae79d..66dabdeb07 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/volumes.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/volumes.py @@ -42,7 +42,7 @@ emit_volume_status_change_event, volume_model_to_volume, ) -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime, run_async from dstack._internal.utils.logging import get_logger @@ -132,7 +132,7 @@ def __init__( queue_check_delay=queue_check_delay, ) - @sentry_utils.instrument_pipeline_task("VolumeFetcher.fetch") + @tracing.instrument_pipeline_task("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: @@ -209,7 +209,7 @@ def __init__( pipeline_hinter=pipeline_hinter, ) - @sentry_utils.instrument_pipeline_task("VolumeWorker.process") + @tracing.instrument_pipeline_task("VolumeWorker.process") async def process(self, item: VolumePipelineItem): volume_model = await _refetch_locked_volume(item) if volume_model is None: diff --git a/src/dstack/_internal/server/background/scheduled_tasks/events.py b/src/dstack/_internal/server/background/scheduled_tasks/events.py index 1fbf602176..3e8f134156 100644 --- a/src/dstack/_internal/server/background/scheduled_tasks/events.py +++ b/src/dstack/_internal/server/background/scheduled_tasks/events.py @@ -5,11 +5,11 @@ from dstack._internal.server import settings from dstack._internal.server.db import get_session_ctx from dstack._internal.server.models import EventModel -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime -@sentry_utils.instrument_scheduled_task +@tracing.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) diff --git a/src/dstack/_internal/server/background/scheduled_tasks/idle_volumes.py b/src/dstack/_internal/server/background/scheduled_tasks/idle_volumes.py index 54fb68c282..e1625131e1 100644 --- a/src/dstack/_internal/server/background/scheduled_tasks/idle_volumes.py +++ b/src/dstack/_internal/server/background/scheduled_tasks/idle_volumes.py @@ -14,14 +14,14 @@ from dstack._internal.server.services.volumes import ( get_volume_configuration, ) -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime from dstack._internal.utils.logging import get_logger logger = get_logger(__name__) -@sentry_utils.instrument_scheduled_task +@tracing.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: diff --git a/src/dstack/_internal/server/background/scheduled_tasks/instance_healthchecks.py b/src/dstack/_internal/server/background/scheduled_tasks/instance_healthchecks.py index 41e83c71aa..0e3765a12c 100644 --- a/src/dstack/_internal/server/background/scheduled_tasks/instance_healthchecks.py +++ b/src/dstack/_internal/server/background/scheduled_tasks/instance_healthchecks.py @@ -5,11 +5,11 @@ from dstack._internal.server import settings from dstack._internal.server.db import get_session_ctx from dstack._internal.server.models import InstanceHealthCheckModel -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import get_current_datetime -@sentry_utils.instrument_scheduled_task +@tracing.instrument_scheduled_task async def delete_instance_healthchecks(): now = get_current_datetime() cutoff = now - timedelta(seconds=settings.SERVER_INSTANCE_HEALTH_TTL_SECONDS) diff --git a/src/dstack/_internal/server/background/scheduled_tasks/metrics.py b/src/dstack/_internal/server/background/scheduled_tasks/metrics.py index 1febe7fa52..10f87aa085 100644 --- a/src/dstack/_internal/server/background/scheduled_tasks/metrics.py +++ b/src/dstack/_internal/server/background/scheduled_tasks/metrics.py @@ -16,7 +16,7 @@ from dstack._internal.server.services.jobs import get_job_provisioning_data, get_job_runtime_data from dstack._internal.server.services.runner import client from dstack._internal.server.services.runner.ssh import runner_ssh_tunnel -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.utils.common import batched, get_current_datetime, get_or_error, run_async from dstack._internal.utils.logging import get_logger @@ -28,7 +28,7 @@ MIN_COLLECT_INTERVAL_SECONDS = 9 -@sentry_utils.instrument_scheduled_task +@tracing.instrument_scheduled_task async def collect_metrics(): async with get_session_ctx() as session: res = await session.execute( @@ -48,7 +48,7 @@ async def collect_metrics(): await _collect_jobs_metrics(batch) -@sentry_utils.instrument_scheduled_task +@tracing.instrument_scheduled_task async def delete_metrics(): now_timestamp_micro = int(get_current_datetime().timestamp() * 1_000_000) running_timestamp_micro_cutoff = ( diff --git a/src/dstack/_internal/server/background/scheduled_tasks/prometheus_metrics.py b/src/dstack/_internal/server/background/scheduled_tasks/prometheus_metrics.py index 96b8cb7742..1b5d21c797 100644 --- a/src/dstack/_internal/server/background/scheduled_tasks/prometheus_metrics.py +++ b/src/dstack/_internal/server/background/scheduled_tasks/prometheus_metrics.py @@ -20,7 +20,7 @@ from dstack._internal.server.services.jobs import get_job_provisioning_data, get_job_runtime_data from dstack._internal.server.services.runner import client from dstack._internal.server.services.runner.ssh import runner_ssh_tunnel -from dstack._internal.server.utils import sentry_utils +from dstack._internal.server.utils import tracing from dstack._internal.server.utils.common import gather_map_async from dstack._internal.utils.common import batched, get_current_datetime, get_or_error, run_async from dstack._internal.utils.logging import get_logger @@ -36,7 +36,7 @@ METRICS_TTL_SECONDS = 600 -@sentry_utils.instrument_scheduled_task +@tracing.instrument_scheduled_task async def collect_prometheus_metrics(): now = get_current_datetime() cutoff = now - timedelta(seconds=MIN_COLLECT_INTERVAL_SECONDS) @@ -64,7 +64,7 @@ async def collect_prometheus_metrics(): await _collect_jobs_metrics(batch, now) -@sentry_utils.instrument_scheduled_task +@tracing.instrument_scheduled_task async def delete_prometheus_metrics(): now = get_current_datetime() cutoff = now - timedelta(seconds=METRICS_TTL_SECONDS) diff --git a/src/dstack/_internal/server/settings.py b/src/dstack/_internal/server/settings.py index 27a97a6db5..211bf403c7 100644 --- a/src/dstack/_internal/server/settings.py +++ b/src/dstack/_internal/server/settings.py @@ -119,6 +119,17 @@ ) SENTRY_PROFILES_SAMPLE_RATE = float(os.getenv("DSTACK_SENTRY_PROFILES_SAMPLE_RATE", 0)) +ENABLE_OTEL_TRACES = os.getenv("DSTACK_ENABLE_OTEL_TRACES") is not None +"""Enables OpenTelemetry tracing. Requires the `otel` extra to be installed. +The exporter is configured via standard `OTEL_*` env vars, e.g. `OTEL_EXPORTER_OTLP_ENDPOINT`. +""" +OTEL_TRACES_SAMPLE_RATE = float(os.getenv("DSTACK_OTEL_TRACES_SAMPLE_RATE", 1.0)) +"""Head sampling rate for traces. The default assumes sampling is done in an OTel collector.""" +OTEL_TRACES_BACKGROUND_SAMPLE_RATE = float( + os.getenv("DSTACK_OTEL_TRACES_BACKGROUND_SAMPLE_RATE", 1.0) +) +"""Head sampling rate for background task traces.""" + DEFAULT_CREDS_DISABLED = os.getenv("DSTACK_DEFAULT_CREDS_DISABLED") is not None DEFAULT_CREDS_ENABLED = not DEFAULT_CREDS_DISABLED diff --git a/src/dstack/_internal/server/utils/common.py b/src/dstack/_internal/server/utils/common.py index 7d8b4290e6..624e957476 100644 --- a/src/dstack/_internal/server/utils/common.py +++ b/src/dstack/_internal/server/utils/common.py @@ -53,3 +53,11 @@ def join_byte_stream_checked(stream: Iterable[bytes], max_size: int) -> Optional return None result += chunk return result + + +SCHEDULED_TASKS_PREFIX = "scheduled_tasks" +PIPELINE_TASKS_PREFIX = "pipeline_tasks" + + +def is_background_task_name(name: str) -> bool: + return name.startswith(SCHEDULED_TASKS_PREFIX) or name.startswith(PIPELINE_TASKS_PREFIX) diff --git a/src/dstack/_internal/server/utils/otel/__init__.py b/src/dstack/_internal/server/utils/otel/__init__.py new file mode 100644 index 0000000000..4a8c004b0d --- /dev/null +++ b/src/dstack/_internal/server/utils/otel/__init__.py @@ -0,0 +1,37 @@ +from contextlib import contextmanager +from typing import TYPE_CHECKING, Generator + +if TYPE_CHECKING: + from fastapi import FastAPI + from sqlalchemy.ext.asyncio import AsyncEngine + +try: + from opentelemetry import context as otel_context + from opentelemetry import trace as otel_trace +except ImportError: + otel_context = None # type: ignore[assignment] + otel_trace = None # type: ignore[assignment] + +_TRACER_NAME = "dstack.server" + + +def configure_tracing(app: "FastAPI", engine: "AsyncEngine") -> None: + try: + from dstack._internal.server.utils.otel import utils + except ImportError as e: + raise RuntimeError( + "DSTACK_ENABLE_OTEL_TRACES is set but OpenTelemetry packages are not installed." + " Install them with `pip install 'dstack[otel]'`." + ) from e + utils.configure_tracing(app, engine) + + +@contextmanager +def task_span(name: str) -> Generator[None, None, None]: + """No-op if OpenTelemetry is not installed or tracing is not configured.""" + if otel_trace is None or otel_context is None: + yield + return + tracer = otel_trace.get_tracer(_TRACER_NAME) + with tracer.start_as_current_span(name, context=otel_context.Context()): + yield diff --git a/src/dstack/_internal/server/utils/otel/utils.py b/src/dstack/_internal/server/utils/otel/utils.py new file mode 100644 index 0000000000..544419f6ef --- /dev/null +++ b/src/dstack/_internal/server/utils/otel/utils.py @@ -0,0 +1,87 @@ +from typing import Optional, Sequence + +from fastapi import FastAPI +from opentelemetry import trace +from opentelemetry.context import Context +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor +from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor +from opentelemetry.instrumentation.requests import RequestsInstrumentor +from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.sdk.trace.sampling import ( + ParentBased, + Sampler, + SamplingResult, + TraceIdRatioBased, +) +from opentelemetry.trace import Link, SpanKind +from opentelemetry.trace.span import TraceState +from opentelemetry.util.types import Attributes +from sqlalchemy.ext.asyncio import AsyncEngine + +from dstack._internal import settings as core_settings +from dstack._internal.server import settings +from dstack._internal.server.utils.common import is_background_task_name +from dstack._internal.utils.logging import get_logger + +logger = get_logger(__name__) + + +def configure_tracing(app: FastAPI, engine: AsyncEngine) -> None: + resource = Resource.create( + { + "service.name": "dstack-server", + "service.version": core_settings.DSTACK_VERSION or "dev", + "deployment.environment.name": settings.SERVER_ENVIRONMENT, + } + ) + provider = TracerProvider( + resource=resource, + sampler=ParentBased( + root=_RootSpanNameSampler( + default_rate=settings.OTEL_TRACES_SAMPLE_RATE, + background_rate=settings.OTEL_TRACES_BACKGROUND_SAMPLE_RATE, + ) + ), + ) + provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) + trace.set_tracer_provider(provider) + FastAPIInstrumentor.instrument_app(app) + SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine) + HTTPXClientInstrumentor().instrument() + RequestsInstrumentor().instrument() + logger.info("OpenTelemetry tracing enabled") + + +class _RootSpanNameSampler(Sampler): + """Samples background task traces and other (HTTP) traces at different rates.""" + + def __init__(self, default_rate: float, background_rate: float): + self._default_sampler = TraceIdRatioBased(default_rate) + self._background_sampler = TraceIdRatioBased(background_rate) + + def should_sample( + self, + parent_context: Optional[Context], + trace_id: int, + name: str, + kind: Optional[SpanKind] = None, + attributes: Attributes = None, + links: Optional[Sequence[Link]] = None, + trace_state: Optional[TraceState] = None, + ) -> SamplingResult: + sampler = self._default_sampler + if is_background_task_name(name): + sampler = self._background_sampler + return sampler.should_sample( + parent_context, trace_id, name, kind, attributes, links, trace_state + ) + + def get_description(self) -> str: + return ( + f"RootSpanNameSampler(default={self._default_sampler.rate}," + f"background={self._background_sampler.rate})" + ) diff --git a/src/dstack/_internal/server/utils/sentry_utils.py b/src/dstack/_internal/server/utils/sentry_utils.py index 7eec46e4a2..4a2d3d22a3 100644 --- a/src/dstack/_internal/server/utils/sentry_utils.py +++ b/src/dstack/_internal/server/utils/sentry_utils.py @@ -1,35 +1,10 @@ import asyncio -import functools from typing import Optional -import sentry_sdk from sentry_sdk.types import Event, Hint, SamplingContext from dstack._internal.server import settings - -SCHEDULED_TASKS_PREFIX = "scheduled_tasks" -PIPELINE_TASKS_PREFIX = "pipeline_tasks" - - -def instrument_scheduled_task(f): - return instrument_named_task(f"{SCHEDULED_TASKS_PREFIX}.{f.__name__}")(f) - - -def instrument_pipeline_task(name: str): - return instrument_named_task(f"{PIPELINE_TASKS_PREFIX}.{name}") - - -def instrument_named_task(name: str): - def decorator(f): - @functools.wraps(f) - async def wrapper(*args, **kwargs): - with sentry_sdk.isolation_scope(): - with sentry_sdk.start_transaction(name=name): - return await f(*args, **kwargs) - - return wrapper - - return decorator +from dstack._internal.server.utils.common import is_background_task_name def sentry_traces_sampler(sampling_context: SamplingContext) -> float: @@ -39,7 +14,7 @@ def sentry_traces_sampler(sampling_context: SamplingContext) -> float: transaction_context = sampling_context["transaction_context"] name = transaction_context.get("name") if name is not None: - if _is_background_transaction(name): + if is_background_task_name(name): return settings.SENTRY_TRACES_BACKGROUND_SAMPLE_RATE return settings.SENTRY_TRACES_SAMPLE_RATE @@ -51,7 +26,3 @@ def __call__(self, event: Event, hint: Hint) -> Optional[Event]: if exc_info and isinstance(exc_info[1], asyncio.CancelledError): return None return event - - -def _is_background_transaction(name: str) -> bool: - return name.startswith(SCHEDULED_TASKS_PREFIX) or name.startswith(PIPELINE_TASKS_PREFIX) diff --git a/src/dstack/_internal/server/utils/tracing.py b/src/dstack/_internal/server/utils/tracing.py new file mode 100644 index 0000000000..4cf698a357 --- /dev/null +++ b/src/dstack/_internal/server/utils/tracing.py @@ -0,0 +1,28 @@ +import functools + +import sentry_sdk + +from dstack._internal.server.utils import otel +from dstack._internal.server.utils.common import PIPELINE_TASKS_PREFIX, SCHEDULED_TASKS_PREFIX + + +def instrument_scheduled_task(f): + return instrument_named_task(f"{SCHEDULED_TASKS_PREFIX}.{f.__name__}")(f) + + +def instrument_pipeline_task(name: str): + return instrument_named_task(f"{PIPELINE_TASKS_PREFIX}.{name}") + + +def instrument_named_task(name: str): + def decorator(f): + @functools.wraps(f) + async def wrapper(*args, **kwargs): + with sentry_sdk.isolation_scope(): + with sentry_sdk.start_transaction(name=name): + with otel.task_span(name): + return await f(*args, **kwargs) + + return wrapper + + return decorator diff --git a/src/dstack/_internal/utils/common.py b/src/dstack/_internal/utils/common.py index 2ee1f337ad..c38a04e820 100644 --- a/src/dstack/_internal/utils/common.py +++ b/src/dstack/_internal/utils/common.py @@ -1,4 +1,5 @@ import asyncio +import contextvars import enum import itertools import re @@ -47,7 +48,10 @@ class EntityID: async def run_async(func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R: - func_with_args = partial(func, *args, **kwargs) + # Copy the context so that context-dependent features such as tracing + # work in the executor thread, same as asyncio.to_thread() + ctx = contextvars.copy_context() + func_with_args = partial(ctx.run, func, *args, **kwargs) return await asyncio.get_running_loop().run_in_executor(None, func_with_args) diff --git a/src/tests/_internal/server/utils/test_common.py b/src/tests/_internal/server/utils/test_common.py index a31060e277..d16b5e7e06 100644 --- a/src/tests/_internal/server/utils/test_common.py +++ b/src/tests/_internal/server/utils/test_common.py @@ -1,6 +1,9 @@ import pytest -from dstack._internal.server.utils.common import join_byte_stream_checked +from dstack._internal.server.utils.common import ( + is_background_task_name, + join_byte_stream_checked, +) @pytest.mark.parametrize( @@ -31,3 +34,15 @@ def generator(stream): raise RuntimeError("Stream end reached, but next value was requested") assert join_byte_stream_checked(generator(stream), max_size) is None + + +@pytest.mark.parametrize( + ["name", "expected"], + [ + ["pipeline_tasks.VolumeWorker.process", True], + ["scheduled_tasks.process_metrics", True], + ["GET /api/project", False], + ], +) +def test_is_background_task_name(name, expected): + assert is_background_task_name(name) is expected diff --git a/src/tests/_internal/server/utils/test_tracing.py b/src/tests/_internal/server/utils/test_tracing.py new file mode 100644 index 0000000000..08caa1626a --- /dev/null +++ b/src/tests/_internal/server/utils/test_tracing.py @@ -0,0 +1,61 @@ +import pytest +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.sdk.trace.sampling import Decision + +from dstack._internal.server.utils import tracing +from dstack._internal.server.utils.otel.utils import _RootSpanNameSampler + +_span_exporter = InMemorySpanExporter() + + +@pytest.fixture +def span_exporter(): + # The global tracer provider can only be set once per process + if not isinstance(trace.get_tracer_provider(), TracerProvider): + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(_span_exporter)) + trace.set_tracer_provider(provider) + _span_exporter.clear() + return _span_exporter + + +class TestInstrumentNamedTask: + @pytest.mark.asyncio + async def test_creates_root_span(self, span_exporter: InMemorySpanExporter): + @tracing.instrument_pipeline_task("TestWorker.process") + async def task(): + return 42 + + assert await task() == 42 + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "pipeline_tasks.TestWorker.process" + assert spans[0].parent is None + + @pytest.mark.asyncio + async def test_each_run_starts_new_trace(self, span_exporter: InMemorySpanExporter): + @tracing.instrument_named_task("task") + async def task(): + pass + + tracer = trace.get_tracer(__name__) + with tracer.start_as_current_span("outer"): + await task() + task_span, outer_span = span_exporter.get_finished_spans() + assert task_span.name == "task" + assert task_span.parent is None + assert task_span.context.trace_id != outer_span.context.trace_id + + +class TestRootSpanNameSampler: + def test_samples_at_rate_by_name(self): + sampler = _RootSpanNameSampler(default_rate=1.0, background_rate=0.0) + http_result = sampler.should_sample(None, trace_id=123, name="GET /api/project") + background_result = sampler.should_sample( + None, trace_id=123, name="pipeline_tasks.VolumeWorker.process" + ) + assert http_result.decision == Decision.RECORD_AND_SAMPLE + assert background_result.decision == Decision.DROP From ee4294c357ab73d0eb0aa0a9fba005e47d1df1ca Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Tue, 14 Jul 2026 14:38:19 +0500 Subject: [PATCH 2/9] Fix otel.configure_tracing call site --- src/dstack/_internal/server/app.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/dstack/_internal/server/app.py b/src/dstack/_internal/server/app.py index 7ee097a0cb..ad2447cbac 100644 --- a/src/dstack/_internal/server/app.py +++ b/src/dstack/_internal/server/app.py @@ -104,6 +104,11 @@ def create_app() -> FastAPI: ], ) app.state.proxy_dependency_injector = ServerProxyDependencyInjector() + if settings.ENABLE_OTEL_TRACES: + # Must be configured before the app starts serving. In particular, + # the FastAPI instrumentation has no effect if the app's middleware + # stack is already built, which happens on the first ASGI event. + otel.configure_tracing(app, get_db().engine) return app @@ -120,8 +125,6 @@ async def lifespan(app: FastAPI): profiles_sample_rate=settings.SENTRY_PROFILES_SAMPLE_RATE, before_send=sentry_utils.AsyncioCancelledErrorFilterEventProcessor(), ) - if settings.ENABLE_OTEL_TRACES: - otel.configure_tracing(app, get_db().engine) server_executor = ThreadPoolExecutor(max_workers=settings.SERVER_EXECUTOR_MAX_WORKERS) asyncio.get_running_loop().set_default_executor(server_executor) await migrate() From 7c5e2b1f7f866d7552306b17352f975c668294aa Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Tue, 14 Jul 2026 16:29:53 +0500 Subject: [PATCH 3/9] Rename DB spans --- .../_internal/server/utils/otel/utils.py | 40 +++++++++++++++++++ .../_internal/server/utils/test_tracing.py | 20 +++++++++- 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/src/dstack/_internal/server/utils/otel/utils.py b/src/dstack/_internal/server/utils/otel/utils.py index 544419f6ef..47bd95d5ec 100644 --- a/src/dstack/_internal/server/utils/otel/utils.py +++ b/src/dstack/_internal/server/utils/otel/utils.py @@ -1,3 +1,4 @@ +import re from typing import Optional, Sequence from fastapi import FastAPI @@ -20,6 +21,8 @@ from opentelemetry.trace import Link, SpanKind from opentelemetry.trace.span import TraceState from opentelemetry.util.types import Attributes +from sqlalchemy import event +from sqlalchemy.engine import Engine from sqlalchemy.ext.asyncio import AsyncEngine from dstack._internal import settings as core_settings @@ -51,11 +54,48 @@ def configure_tracing(app: FastAPI, engine: AsyncEngine) -> None: trace.set_tracer_provider(provider) FastAPIInstrumentor.instrument_app(app) SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine) + _register_db_span_renaming(engine.sync_engine) HTTPXClientInstrumentor().instrument() RequestsInstrumentor().instrument() logger.info("OpenTelemetry tracing enabled") +def _register_db_span_renaming(engine: Engine) -> None: + """Renames DB spans from ` ` to ` `. + + The default names are not useful for trace lists and span metrics: + the db name is the same for all spans and is a file path on SQLite. + Must be called after the engine is instrumented so that the listener + runs after the instrumentation creates the span. + + NOTE: This hack is needed because SQLAlchemyInstrumentor does not provide any hooks + to update spans unlike most other instrumentors. + """ + + @event.listens_for(engine, "before_cursor_execute") + def _rename_db_span(conn, cursor, statement, parameters, context, executemany): + # The instrumentation stores the still-recording span on the execution context + span = getattr(context, "_otel_span", None) + if span is None or not span.is_recording(): + return + name = _get_db_span_name(statement) + if name is not None: + span.update_name(name) + + +_DB_TABLE_RE = re.compile(r'\b(?:FROM|INTO|UPDATE|JOIN)\s+["\'`]?(\w+)', re.IGNORECASE) + + +def _get_db_span_name(statement: str) -> Optional[str]: + operation = statement.split(None, 1)[0].upper() if statement.split() else None + if operation is None: + return None + match = _DB_TABLE_RE.search(statement) + if match is None: + return operation + return f"{operation} {match.group(1)}" + + class _RootSpanNameSampler(Sampler): """Samples background task traces and other (HTTP) traces at different rates.""" diff --git a/src/tests/_internal/server/utils/test_tracing.py b/src/tests/_internal/server/utils/test_tracing.py index 08caa1626a..97fda3d0f9 100644 --- a/src/tests/_internal/server/utils/test_tracing.py +++ b/src/tests/_internal/server/utils/test_tracing.py @@ -6,7 +6,7 @@ from opentelemetry.sdk.trace.sampling import Decision from dstack._internal.server.utils import tracing -from dstack._internal.server.utils.otel.utils import _RootSpanNameSampler +from dstack._internal.server.utils.otel.utils import _get_db_span_name, _RootSpanNameSampler _span_exporter = InMemorySpanExporter() @@ -50,6 +50,24 @@ async def task(): assert task_span.context.trace_id != outer_span.context.trace_id +class TestGetDBSpanName: + @pytest.mark.parametrize( + ("statement", "expected"), + [ + ("SELECT * FROM instances WHERE id = ?", "SELECT instances"), + ("select id\nfrom jobs join runs on ...", "SELECT jobs"), + ("INSERT INTO volumes (id) VALUES (?)", "INSERT volumes"), + ("UPDATE fleets SET status = ?", "UPDATE fleets"), + ('DELETE FROM "users"', "DELETE users"), + ("PRAGMA journal_mode=WAL;", "PRAGMA"), + ("BEGIN", "BEGIN"), + ("", None), + ], + ) + def test_returns_expected(self, statement, expected): + assert _get_db_span_name(statement) == expected + + class TestRootSpanNameSampler: def test_samples_at_rate_by_name(self): sampler = _RootSpanNameSampler(default_rate=1.0, background_rate=0.0) From a46d4468d53f365a28a3cc09be65ad889ce0c523 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Tue, 14 Jul 2026 17:12:03 +0500 Subject: [PATCH 4/9] Support OTel logs --- src/dstack/_internal/server/app.py | 6 ++- src/dstack/_internal/server/settings.py | 6 ++- .../_internal/server/utils/otel/__init__.py | 12 +++++- .../_internal/server/utils/otel/utils.py | 39 +++++++++++++++---- .../_internal/server/utils/test_tracing.py | 38 +++++++++++++++++- 5 files changed, 87 insertions(+), 14 deletions(-) diff --git a/src/dstack/_internal/server/app.py b/src/dstack/_internal/server/app.py index ad2447cbac..95c253d9fa 100644 --- a/src/dstack/_internal/server/app.py +++ b/src/dstack/_internal/server/app.py @@ -104,11 +104,13 @@ def create_app() -> FastAPI: ], ) app.state.proxy_dependency_injector = ServerProxyDependencyInjector() - if settings.ENABLE_OTEL_TRACES: + if settings.OTEL_TRACES_ENABLED: # Must be configured before the app starts serving. In particular, # the FastAPI instrumentation has no effect if the app's middleware - # stack is already built, which happens on the first ASGI event. + # stack is already built, which happens on the first ASGI event (lifespan). otel.configure_tracing(app, get_db().engine) + if settings.OTEL_LOGS_ENABLED: + otel.configure_log_export() return app diff --git a/src/dstack/_internal/server/settings.py b/src/dstack/_internal/server/settings.py index 211bf403c7..77c1c04dcd 100644 --- a/src/dstack/_internal/server/settings.py +++ b/src/dstack/_internal/server/settings.py @@ -119,7 +119,7 @@ ) SENTRY_PROFILES_SAMPLE_RATE = float(os.getenv("DSTACK_SENTRY_PROFILES_SAMPLE_RATE", 0)) -ENABLE_OTEL_TRACES = os.getenv("DSTACK_ENABLE_OTEL_TRACES") is not None +OTEL_TRACES_ENABLED = os.getenv("DSTACK_OTEL_TRACES_ENABLED") is not None """Enables OpenTelemetry tracing. Requires the `otel` extra to be installed. The exporter is configured via standard `OTEL_*` env vars, e.g. `OTEL_EXPORTER_OTLP_ENDPOINT`. """ @@ -129,6 +129,10 @@ os.getenv("DSTACK_OTEL_TRACES_BACKGROUND_SAMPLE_RATE", 1.0) ) """Head sampling rate for background task traces.""" +OTEL_LOGS_ENABLED = os.getenv("DSTACK_OTEL_LOGS_ENABLED") is not None +"""Enables log export via OTLP. Requires the `otel` extra to be installed. +The exporter is configured via standard `OTEL_*` env vars, e.g. `OTEL_EXPORTER_OTLP_ENDPOINT`. +""" DEFAULT_CREDS_DISABLED = os.getenv("DSTACK_DEFAULT_CREDS_DISABLED") is not None DEFAULT_CREDS_ENABLED = not DEFAULT_CREDS_DISABLED diff --git a/src/dstack/_internal/server/utils/otel/__init__.py b/src/dstack/_internal/server/utils/otel/__init__.py index 4a8c004b0d..16c0db5608 100644 --- a/src/dstack/_internal/server/utils/otel/__init__.py +++ b/src/dstack/_internal/server/utils/otel/__init__.py @@ -16,14 +16,22 @@ def configure_tracing(app: "FastAPI", engine: "AsyncEngine") -> None: + _import_utils("DSTACK_OTEL_TRACES_ENABLED").configure_tracing(app, engine) + + +def configure_log_export() -> None: + _import_utils("DSTACK_OTEL_LOGS_ENABLED").configure_log_export() + + +def _import_utils(enabled_by: str): try: from dstack._internal.server.utils.otel import utils except ImportError as e: raise RuntimeError( - "DSTACK_ENABLE_OTEL_TRACES is set but OpenTelemetry packages are not installed." + f"{enabled_by} is set but OpenTelemetry packages are not installed." " Install them with `pip install 'dstack[otel]'`." ) from e - utils.configure_tracing(app, engine) + return utils @contextmanager diff --git a/src/dstack/_internal/server/utils/otel/utils.py b/src/dstack/_internal/server/utils/otel/utils.py index 47bd95d5ec..5cdd76d52a 100644 --- a/src/dstack/_internal/server/utils/otel/utils.py +++ b/src/dstack/_internal/server/utils/otel/utils.py @@ -1,14 +1,19 @@ +import logging import re from typing import Optional, Sequence from fastapi import FastAPI from opentelemetry import trace +from opentelemetry._logs import set_logger_provider from opentelemetry.context import Context +from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor from opentelemetry.instrumentation.requests import RequestsInstrumentor from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor +from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor @@ -28,21 +33,15 @@ from dstack._internal import settings as core_settings from dstack._internal.server import settings from dstack._internal.server.utils.common import is_background_task_name +from dstack._internal.server.utils.logging import AsyncioCancelledErrorFilter from dstack._internal.utils.logging import get_logger logger = get_logger(__name__) def configure_tracing(app: FastAPI, engine: AsyncEngine) -> None: - resource = Resource.create( - { - "service.name": "dstack-server", - "service.version": core_settings.DSTACK_VERSION or "dev", - "deployment.environment.name": settings.SERVER_ENVIRONMENT, - } - ) provider = TracerProvider( - resource=resource, + resource=_get_resource(), sampler=ParentBased( root=_RootSpanNameSampler( default_rate=settings.OTEL_TRACES_SAMPLE_RATE, @@ -60,6 +59,30 @@ def configure_tracing(app: FastAPI, engine: AsyncEngine) -> None: logger.info("OpenTelemetry tracing enabled") +def configure_log_export() -> None: + logger_provider = LoggerProvider(resource=_get_resource()) + logger_provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter())) + set_logger_provider(logger_provider) + logging.getLogger().addHandler(_build_log_handler(logger_provider)) + logger.info("OpenTelemetry log export enabled") + + +def _get_resource() -> Resource: + return Resource.create( + { + "service.name": "dstack-server", + "service.version": core_settings.DSTACK_VERSION or "dev", + "deployment.environment.name": settings.SERVER_ENVIRONMENT, + } + ) + + +def _build_log_handler(logger_provider: LoggerProvider) -> logging.Handler: + handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider) + handler.addFilter(AsyncioCancelledErrorFilter()) + return handler + + def _register_db_span_renaming(engine: Engine) -> None: """Renames DB spans from ` ` to `
`. diff --git a/src/tests/_internal/server/utils/test_tracing.py b/src/tests/_internal/server/utils/test_tracing.py index 97fda3d0f9..82e77dfe7d 100644 --- a/src/tests/_internal/server/utils/test_tracing.py +++ b/src/tests/_internal/server/utils/test_tracing.py @@ -1,3 +1,6 @@ +import asyncio +import logging + import pytest from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider @@ -6,7 +9,11 @@ from opentelemetry.sdk.trace.sampling import Decision from dstack._internal.server.utils import tracing -from dstack._internal.server.utils.otel.utils import _get_db_span_name, _RootSpanNameSampler +from dstack._internal.server.utils.otel.utils import ( + _build_log_handler, + _get_db_span_name, + _RootSpanNameSampler, +) _span_exporter = InMemorySpanExporter() @@ -50,6 +57,35 @@ async def task(): assert task_span.context.trace_id != outer_span.context.trace_id +class TestBuildLogHandler: + def test_exports_records_with_trace_context(self, span_exporter: InMemorySpanExporter): + from opentelemetry.sdk._logs import LoggerProvider + from opentelemetry.sdk._logs.export import ( + InMemoryLogRecordExporter, + SimpleLogRecordProcessor, + ) + + log_exporter = InMemoryLogRecordExporter() + logger_provider = LoggerProvider() + logger_provider.add_log_record_processor(SimpleLogRecordProcessor(log_exporter)) + logger = logging.getLogger("test_otel_logs") + logger.addHandler(_build_log_handler(logger_provider)) + + tracer = trace.get_tracer(__name__) + with tracer.start_as_current_span("op") as span: + logger.warning("something happened") + try: + logger.error("cancelled", exc_info=asyncio.CancelledError()) + except Exception: + pass + + records = [d.log_record for d in log_exporter.get_finished_logs()] + assert len(records) == 1 + assert records[0].body == "something happened" + assert records[0].severity_text == "WARN" + assert records[0].trace_id == span.get_span_context().trace_id + + class TestGetDBSpanName: @pytest.mark.parametrize( ("statement", "expected"), From 7077bc66a704833a0c14e2e17696891914b9e2a3 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Wed, 15 Jul 2026 14:19:55 +0500 Subject: [PATCH 5/9] Support OTel metrics --- pyproject.toml | 2 + src/dstack/_internal/server/app.py | 6 +- src/dstack/_internal/server/settings.py | 7 ++ .../_internal/server/utils/otel/__init__.py | 15 +--- .../_internal/server/utils/otel/utils.py | 84 +++++++++++++++++-- .../_internal/server/utils/test_tracing.py | 19 +++++ 6 files changed, 109 insertions(+), 24 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 2bf79c5aee..3f9bf6e086 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -267,10 +267,12 @@ fluentbit = [ otel = [ "opentelemetry-sdk>=1.30.0", "opentelemetry-exporter-otlp-proto-http>=1.30.0", + "opentelemetry-exporter-prometheus>=0.51b0", "opentelemetry-instrumentation-fastapi>=0.51b0", "opentelemetry-instrumentation-sqlalchemy>=0.51b0", "opentelemetry-instrumentation-httpx>=0.51b0", "opentelemetry-instrumentation-requests>=0.51b0", + "opentelemetry-instrumentation-system-metrics>=0.51b0", "dstack[server]", ] crusoe = [ diff --git a/src/dstack/_internal/server/app.py b/src/dstack/_internal/server/app.py index 95c253d9fa..94fa7be584 100644 --- a/src/dstack/_internal/server/app.py +++ b/src/dstack/_internal/server/app.py @@ -104,13 +104,11 @@ def create_app() -> FastAPI: ], ) app.state.proxy_dependency_injector = ServerProxyDependencyInjector() - if settings.OTEL_TRACES_ENABLED: + if settings.OTEL_TRACES_ENABLED or settings.OTEL_METRICS_ENABLED or settings.OTEL_LOGS_ENABLED: # Must be configured before the app starts serving. In particular, # the FastAPI instrumentation has no effect if the app's middleware # stack is already built, which happens on the first ASGI event (lifespan). - otel.configure_tracing(app, get_db().engine) - if settings.OTEL_LOGS_ENABLED: - otel.configure_log_export() + otel.configure(app, get_db().engine) return app diff --git a/src/dstack/_internal/server/settings.py b/src/dstack/_internal/server/settings.py index 77c1c04dcd..2f82baafb5 100644 --- a/src/dstack/_internal/server/settings.py +++ b/src/dstack/_internal/server/settings.py @@ -133,6 +133,13 @@ """Enables log export via OTLP. Requires the `otel` extra to be installed. The exporter is configured via standard `OTEL_*` env vars, e.g. `OTEL_EXPORTER_OTLP_ENDPOINT`. """ +OTEL_METRICS_ENABLED = os.getenv("DSTACK_OTEL_METRICS_ENABLED") is not None +"""Enables OpenTelemetry metrics. Requires the `otel` extra to be installed.""" +OTEL_METRICS_EXPORTERS = os.getenv("DSTACK_OTEL_METRICS_EXPORTERS") +"""A comma-separated list of metrics exporters: `prometheus` (expose via the /metrics endpoint) +and/or `otlp` (push via OTLP, configured by standard `OTEL_*` env vars). +Defaults to `prometheus` if the /metrics endpoint is enabled, otherwise `otlp`. +""" DEFAULT_CREDS_DISABLED = os.getenv("DSTACK_DEFAULT_CREDS_DISABLED") is not None DEFAULT_CREDS_ENABLED = not DEFAULT_CREDS_DISABLED diff --git a/src/dstack/_internal/server/utils/otel/__init__.py b/src/dstack/_internal/server/utils/otel/__init__.py index 16c0db5608..767cd611e1 100644 --- a/src/dstack/_internal/server/utils/otel/__init__.py +++ b/src/dstack/_internal/server/utils/otel/__init__.py @@ -15,23 +15,16 @@ _TRACER_NAME = "dstack.server" -def configure_tracing(app: "FastAPI", engine: "AsyncEngine") -> None: - _import_utils("DSTACK_OTEL_TRACES_ENABLED").configure_tracing(app, engine) - - -def configure_log_export() -> None: - _import_utils("DSTACK_OTEL_LOGS_ENABLED").configure_log_export() - - -def _import_utils(enabled_by: str): +def configure(app: "FastAPI", engine: "AsyncEngine") -> None: + """Sets up the OTel signals enabled by the `DSTACK_OTEL_*_ENABLED` env vars.""" try: from dstack._internal.server.utils.otel import utils except ImportError as e: raise RuntimeError( - f"{enabled_by} is set but OpenTelemetry packages are not installed." + "DSTACK_OTEL_*_ENABLED is set but OpenTelemetry packages are not installed." " Install them with `pip install 'dstack[otel]'`." ) from e - return utils + utils.configure(app, engine) @contextmanager diff --git a/src/dstack/_internal/server/utils/otel/utils.py b/src/dstack/_internal/server/utils/otel/utils.py index 5cdd76d52a..3981227f6b 100644 --- a/src/dstack/_internal/server/utils/otel/utils.py +++ b/src/dstack/_internal/server/utils/otel/utils.py @@ -1,19 +1,24 @@ import logging import re -from typing import Optional, Sequence +from typing import List, Optional, Sequence from fastapi import FastAPI -from opentelemetry import trace +from opentelemetry import metrics, trace from opentelemetry._logs import set_logger_provider from opentelemetry.context import Context from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter +from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.exporter.prometheus import PrometheusMetricReader from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor from opentelemetry.instrumentation.requests import RequestsInstrumentor from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor +from opentelemetry.instrumentation.system_metrics import SystemMetricsInstrumentor from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler from opentelemetry.sdk._logs.export import BatchLogRecordProcessor +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import MetricReader, PeriodicExportingMetricReader from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor @@ -39,7 +44,20 @@ logger = get_logger(__name__) -def configure_tracing(app: FastAPI, engine: AsyncEngine) -> None: +def configure(app: FastAPI, engine: AsyncEngine) -> None: + if settings.OTEL_TRACES_ENABLED: + _configure_tracing() + if settings.OTEL_METRICS_ENABLED: + _configure_metrics() + if settings.OTEL_LOGS_ENABLED: + _configure_log_export() + if settings.OTEL_TRACES_ENABLED or settings.OTEL_METRICS_ENABLED: + # The instrumentors emit both traces and metrics — + # each signal is a no-op unless its provider is configured + _instrument(app, engine) + + +def _configure_tracing() -> None: provider = TracerProvider( resource=_get_resource(), sampler=ParentBased( @@ -51,15 +69,29 @@ def configure_tracing(app: FastAPI, engine: AsyncEngine) -> None: ) provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) trace.set_tracer_provider(provider) - FastAPIInstrumentor.instrument_app(app) - SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine) - _register_db_span_renaming(engine.sync_engine) - HTTPXClientInstrumentor().instrument() - RequestsInstrumentor().instrument() logger.info("OpenTelemetry tracing enabled") -def configure_log_export() -> None: +def _configure_metrics() -> None: + readers: List[MetricReader] = [] + for exporter in _get_metrics_exporters(): + if exporter == "prometheus": + # Registers into the default prometheus_client registry + # served by the /metrics endpoint + readers.append(PrometheusMetricReader()) + elif exporter == "otlp": + readers.append(PeriodicExportingMetricReader(OTLPMetricExporter())) + else: + raise ValueError( + f"Unknown exporter {exporter!r} in DSTACK_OTEL_METRICS_EXPORTERS." + " Supported exporters: prometheus, otlp" + ) + provider = MeterProvider(resource=_get_resource(), metric_readers=readers) + metrics.set_meter_provider(provider) + logger.info("OpenTelemetry metrics enabled") + + +def _configure_log_export() -> None: logger_provider = LoggerProvider(resource=_get_resource()) logger_provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter())) set_logger_provider(logger_provider) @@ -67,6 +99,40 @@ def configure_log_export() -> None: logger.info("OpenTelemetry log export enabled") +def _instrument(app: FastAPI, engine: AsyncEngine) -> None: + FastAPIInstrumentor.instrument_app(app) + SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine) + _register_db_span_renaming(engine.sync_engine) + HTTPXClientInstrumentor().instrument() + RequestsInstrumentor().instrument() + if settings.OTEL_METRICS_ENABLED: + SystemMetricsInstrumentor(config=_PROCESS_METRICS_CONFIG).instrument() + + +def _get_metrics_exporters() -> List[str]: + if settings.OTEL_METRICS_EXPORTERS is not None: + return [e.strip() for e in settings.OTEL_METRICS_EXPORTERS.split(",") if e.strip()] + if settings.ENABLE_PROMETHEUS_METRICS: + return ["prometheus"] + return ["otlp"] + + +# Process-level metrics only. Host-level (`system.*`) metrics are left +# to node exporters. `process.runtime.*` metrics are deprecated duplicates. +_PROCESS_METRICS_CONFIG = { + "process.cpu.time": ["user", "system"], + "process.cpu.utilization": ["user", "system"], + "process.memory.usage": None, + "process.memory.virtual": None, + "process.thread.count": None, + "process.open_file_descriptor.count": None, + "process.context_switches": ["involuntary", "voluntary"], + "cpython.gc.collections": None, + "cpython.gc.collected_objects": None, + "cpython.gc.uncollectable_objects": None, +} + + def _get_resource() -> Resource: return Resource.create( { diff --git a/src/tests/_internal/server/utils/test_tracing.py b/src/tests/_internal/server/utils/test_tracing.py index 82e77dfe7d..eb2e202869 100644 --- a/src/tests/_internal/server/utils/test_tracing.py +++ b/src/tests/_internal/server/utils/test_tracing.py @@ -12,6 +12,7 @@ from dstack._internal.server.utils.otel.utils import ( _build_log_handler, _get_db_span_name, + _get_metrics_exporters, _RootSpanNameSampler, ) @@ -86,6 +87,24 @@ def test_exports_records_with_trace_context(self, span_exporter: InMemorySpanExp assert records[0].trace_id == span.get_span_context().trace_id +class TestGetMetricsExporters: + @pytest.mark.parametrize( + ("exporters", "prometheus_enabled", "expected"), + [ + (None, True, ["prometheus"]), + (None, False, ["otlp"]), + ("otlp", True, ["otlp"]), + ("prometheus, otlp", False, ["prometheus", "otlp"]), + ], + ) + def test_returns_expected(self, monkeypatch, exporters, prometheus_enabled, expected): + from dstack._internal.server import settings + + monkeypatch.setattr(settings, "OTEL_METRICS_EXPORTERS", exporters) + monkeypatch.setattr(settings, "ENABLE_PROMETHEUS_METRICS", prometheus_enabled) + assert _get_metrics_exporters() == expected + + class TestGetDBSpanName: @pytest.mark.parametrize( ("statement", "expected"), From 427148a64a6f7aa5c4378faea1be3a8a235e5ebc Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Wed, 15 Jul 2026 15:07:45 +0500 Subject: [PATCH 6/9] Track background task runs metric --- .../_internal/server/utils/otel/__init__.py | 22 ++++++++++- src/dstack/_internal/server/utils/tracing.py | 12 +++++- .../_internal/server/utils/test_tracing.py | 39 +++++++++++++++++++ 3 files changed, 70 insertions(+), 3 deletions(-) diff --git a/src/dstack/_internal/server/utils/otel/__init__.py b/src/dstack/_internal/server/utils/otel/__init__.py index 767cd611e1..1b63bfd410 100644 --- a/src/dstack/_internal/server/utils/otel/__init__.py +++ b/src/dstack/_internal/server/utils/otel/__init__.py @@ -7,12 +7,23 @@ try: from opentelemetry import context as otel_context + from opentelemetry import metrics as otel_metrics from opentelemetry import trace as otel_trace except ImportError: otel_context = None # type: ignore[assignment] + otel_metrics = None # type: ignore[assignment] otel_trace = None # type: ignore[assignment] -_TRACER_NAME = "dstack.server" +_SCOPE_NAME = "dstack.server" + +if otel_metrics is not None: + _task_runs_counter = otel_metrics.get_meter(_SCOPE_NAME).create_counter( + "dstack.server.background.task.runs", + unit="{run}", + description="The number of background task runs", + ) +else: + _task_runs_counter = None def configure(app: "FastAPI", engine: "AsyncEngine") -> None: @@ -33,6 +44,13 @@ def task_span(name: str) -> Generator[None, None, None]: if otel_trace is None or otel_context is None: yield return - tracer = otel_trace.get_tracer(_TRACER_NAME) + tracer = otel_trace.get_tracer(_SCOPE_NAME) with tracer.start_as_current_span(name, context=otel_context.Context()): yield + + +def record_task_run(name: str, *, error: bool) -> None: + """No-op if OpenTelemetry is not installed or metrics are not configured.""" + if _task_runs_counter is None: + return + _task_runs_counter.add(1, {"task": name, "status": "error" if error else "success"}) diff --git a/src/dstack/_internal/server/utils/tracing.py b/src/dstack/_internal/server/utils/tracing.py index 4cf698a357..72d4a27c03 100644 --- a/src/dstack/_internal/server/utils/tracing.py +++ b/src/dstack/_internal/server/utils/tracing.py @@ -1,3 +1,4 @@ +import asyncio import functools import sentry_sdk @@ -21,7 +22,16 @@ async def wrapper(*args, **kwargs): with sentry_sdk.isolation_scope(): with sentry_sdk.start_transaction(name=name): with otel.task_span(name): - return await f(*args, **kwargs) + try: + result = await f(*args, **kwargs) + except asyncio.CancelledError: + # Interrupted, e.g. on server shutdown — not a task failure + raise + except Exception: + otel.record_task_run(name, error=True) + raise + otel.record_task_run(name, error=False) + return result return wrapper diff --git a/src/tests/_internal/server/utils/test_tracing.py b/src/tests/_internal/server/utils/test_tracing.py index eb2e202869..acda040944 100644 --- a/src/tests/_internal/server/utils/test_tracing.py +++ b/src/tests/_internal/server/utils/test_tracing.py @@ -58,6 +58,45 @@ async def task(): assert task_span.context.trace_id != outer_span.context.trace_id +class TestRecordTaskRun: + @pytest.mark.asyncio + async def test_counts_runs_by_status(self): + from opentelemetry import metrics + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import InMemoryMetricReader + + reader = InMemoryMetricReader() + # The global meter provider can only be set once per process + if not isinstance(metrics.get_meter_provider(), MeterProvider): + metrics.set_meter_provider(MeterProvider(metric_readers=[reader])) + else: + pytest.skip("global meter provider already set") + + @tracing.instrument_named_task("task") + async def ok_task(): + return 1 + + @tracing.instrument_named_task("task") + async def failing_task(): + raise ValueError() + + await ok_task() + await ok_task() + with pytest.raises(ValueError): + await failing_task() + + points = ( + reader.get_metrics_data() + .resource_metrics[0] + .scope_metrics[0] + .metrics[0] + .data.data_points + ) + counts = {p.attributes["status"]: p.value for p in points} + assert counts == {"success": 2, "error": 1} + assert all(p.attributes["task"] == "task" for p in points) + + class TestBuildLogHandler: def test_exports_records_with_trace_context(self, span_exporter: InMemorySpanExporter): from opentelemetry.sdk._logs import LoggerProvider From 92896aed02f979e505806726bf1d0be3f05bcd05 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Thu, 16 Jul 2026 11:50:30 +0500 Subject: [PATCH 7/9] Document Observability --- mkdocs/docs/guides/server-deployment.md | 64 +++++++++++++++++++++++++ mkdocs/docs/reference/env.md | 10 ++++ 2 files changed, 74 insertions(+) diff --git a/mkdocs/docs/guides/server-deployment.md b/mkdocs/docs/guides/server-deployment.md index f368b62d5a..8503b7f17c 100644 --- a/mkdocs/docs/guides/server-deployment.md +++ b/mkdocs/docs/guides/server-deployment.md @@ -386,6 +386,70 @@ To deploy the SSH proxy, follow its [deployment guide](https://github.com/dstack > For a local setup running [PostgreSQL](#postgresql) and the SSH proxy together, see the example > [`docker-compose.yml`](https://github.com/dstackai/dstack/blob/master/docker/server/docker-compose.yml). +## Observability + +Besides [workload metrics](../concepts/metrics.md), the `dstack` server can report its own errors, traces, logs, +and metrics for monitoring the server. Two integrations are supported: Sentry and OpenTelemetry. +They are independent and can be enabled together, e.g. Sentry for error tracking and OpenTelemetry for tracing. + +### Sentry + +To report server errors and traces via the Sentry SDK, set the `DSTACK_SENTRY_DSN` environment variable. + +The DSN can point to [Sentry :material-arrow-top-right-thin:{.external }](https://sentry.io/) +or any Sentry-compatible error tracker such as +[Bugsink :material-arrow-top-right-thin:{.external }](https://www.bugsink.com/) or +[GlitchTip :material-arrow-top-right-thin:{.external }](https://glitchtip.com/). + +The following optional environment variables control sampling: + +- `DSTACK_SENTRY_TRACES_SAMPLE_RATE` – The sample rate for API request traces. Defaults to `0.1`. +- `DSTACK_SENTRY_TRACES_BACKGROUND_SAMPLE_RATE` – The sample rate for background task traces. Defaults to `0.01`. +- `DSTACK_SENTRY_PROFILES_SAMPLE_RATE` – The profiling sample rate, relative to the traces sample rate. Defaults to `0`. + +### OpenTelemetry + +The server can export traces, logs, and metrics using OpenTelemetry. Each signal is enabled independently: + +- `DSTACK_OTEL_TRACES_ENABLED` – Enables tracing. API requests, DB queries, outgoing HTTP requests, and background tasks are instrumented automatically. +- `DSTACK_OTEL_LOGS_ENABLED` – Enables server log export. When tracing is also enabled, logs are correlated with traces via `trace_id`. +- `DSTACK_OTEL_METRICS_ENABLED` – Enables metrics: HTTP server and client request durations, process CPU/memory/GC metrics, etc. + +The export destination and protocol are configured via the standard +[`OTEL_*` environment variables :material-arrow-top-right-thin:{.external }](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/). +Traces and logs are exported via OTLP over HTTP. Example: + +```shell +$ DSTACK_OTEL_TRACES_ENABLED=1 \ + DSTACK_OTEL_LOGS_ENABLED=1 \ + DSTACK_OTEL_METRICS_ENABLED=1 \ + OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector.example.com:4318 \ + dstack server +``` + +??? info "Required dependencies" + To use the OpenTelemetry integration, install the `otel` extras: + + ```shell + $ pip install "dstack[all]" -U + # or + $ pip install "dstack[otel]" -U + ``` + +??? info "Trace sampling" + By default, all traces are exported (sample rate `1.0`), assuming sampling is done downstream, + e.g. in an OTel Collector. To sample in the server instead, set: + + - `DSTACK_OTEL_TRACES_SAMPLE_RATE` – The head sampling rate for API request traces. + - `DSTACK_OTEL_TRACES_BACKGROUND_SAMPLE_RATE` – The head sampling rate for background task traces. + +??? info "Metrics exporters" + By default, if the [Prometheus `/metrics` endpoint](../concepts/metrics.md) is enabled via + `DSTACK_ENABLE_PROMETHEUS_METRICS`, OpenTelemetry metrics are exposed there alongside the built-in + `dstack` metrics; otherwise they are pushed via OTLP. Set `DSTACK_OTEL_METRICS_EXPORTERS` + to a comma-separated list of `prometheus` and/or `otlp` to override, e.g. force OTLP push + even when `/metrics` is enabled. + ## Encryption By default, `dstack` stores data in plaintext. To enforce encryption, you diff --git a/mkdocs/docs/reference/env.md b/mkdocs/docs/reference/env.md index 86a7dd051d..e6623010ef 100644 --- a/mkdocs/docs/reference/env.md +++ b/mkdocs/docs/reference/env.md @@ -121,6 +121,16 @@ For more details on the options below, refer to the [server deployment](../guide - `DSTACK_SERVER_ELASTICSEARCH_INDEX`{ #DSTACK_SERVER_ELASTICSEARCH_INDEX } – The Elasticsearch/OpenSearch index pattern. Defaults to `dstack-logs`. - `DSTACK_SERVER_ELASTICSEARCH_API_KEY`{ #DSTACK_SERVER_ELASTICSEARCH_API_KEY } – The Elasticsearch/OpenSearch API key for authentication. - `DSTACK_ENABLE_PROMETHEUS_METRICS`{ #DSTACK_ENABLE_PROMETHEUS_METRICS } — Enables Prometheus metrics collection and export. +- `DSTACK_SENTRY_DSN`{ #DSTACK_SENTRY_DSN } – The Sentry DSN. If set, enables error reporting and tracing via the Sentry SDK. See [observability](../guides/server-deployment.md#observability). +- `DSTACK_SENTRY_TRACES_SAMPLE_RATE`{ #DSTACK_SENTRY_TRACES_SAMPLE_RATE } – The Sentry sample rate for API request traces. Defaults to `0.1`. +- `DSTACK_SENTRY_TRACES_BACKGROUND_SAMPLE_RATE`{ #DSTACK_SENTRY_TRACES_BACKGROUND_SAMPLE_RATE } – The Sentry sample rate for background task traces. Defaults to `0.01`. +- `DSTACK_SENTRY_PROFILES_SAMPLE_RATE`{ #DSTACK_SENTRY_PROFILES_SAMPLE_RATE } – The Sentry profiling sample rate, relative to the traces sample rate. Defaults to `0`. +- `DSTACK_OTEL_TRACES_ENABLED`{ #DSTACK_OTEL_TRACES_ENABLED } – Enables OpenTelemetry tracing if set to any value. Requires the `otel` extra. The exporter is configured via standard `OTEL_*` env vars such as `OTEL_EXPORTER_OTLP_ENDPOINT`. See [observability](../guides/server-deployment.md#observability). +- `DSTACK_OTEL_TRACES_SAMPLE_RATE`{ #DSTACK_OTEL_TRACES_SAMPLE_RATE } – The head sampling rate for API request traces. Defaults to `1.0`, which assumes sampling is done downstream, e.g. in an OTel collector. +- `DSTACK_OTEL_TRACES_BACKGROUND_SAMPLE_RATE`{ #DSTACK_OTEL_TRACES_BACKGROUND_SAMPLE_RATE } – The head sampling rate for background task traces. Defaults to `1.0`. +- `DSTACK_OTEL_LOGS_ENABLED`{ #DSTACK_OTEL_LOGS_ENABLED } – Enables server log export via OTLP if set to any value. Requires the `otel` extra. +- `DSTACK_OTEL_METRICS_ENABLED`{ #DSTACK_OTEL_METRICS_ENABLED } – Enables OpenTelemetry metrics if set to any value. Requires the `otel` extra. +- `DSTACK_OTEL_METRICS_EXPORTERS`{ #DSTACK_OTEL_METRICS_EXPORTERS } – A comma-separated list of OpenTelemetry metrics exporters: `prometheus` (expose on the `/metrics` endpoint) and/or `otlp` (push via OTLP). Defaults to `prometheus` if `DSTACK_ENABLE_PROMETHEUS_METRICS` is set, otherwise `otlp`. - `DSTACK_DEFAULT_SERVICE_CLIENT_MAX_BODY_SIZE`{ #DSTACK_DEFAULT_SERVICE_CLIENT_MAX_BODY_SIZE } – Request body size limit for services running with a gateway, in bytes. Defaults to 64 MiB. - `DSTACK_SERVICE_CLIENT_TIMEOUT`{ #DSTACK_SERVICE_CLIENT_TIMEOUT } – Timeout in seconds for HTTP requests sent from the in-server proxy and gateways to service replicas. Defaults to 60. - `DSTACK_FORBID_SERVICES_WITHOUT_GATEWAY`{ #DSTACK_FORBID_SERVICES_WITHOUT_GATEWAY } – Forbids registering new services without a gateway if set to any value. From 5995915f2e1222681f4ba3344d0b431780767b89 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Thu, 16 Jul 2026 11:52:30 +0500 Subject: [PATCH 8/9] Tests cleanup --- src/tests/_internal/server/utils/test_common.py | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/src/tests/_internal/server/utils/test_common.py b/src/tests/_internal/server/utils/test_common.py index d16b5e7e06..175fb042b8 100644 --- a/src/tests/_internal/server/utils/test_common.py +++ b/src/tests/_internal/server/utils/test_common.py @@ -1,7 +1,6 @@ import pytest from dstack._internal.server.utils.common import ( - is_background_task_name, join_byte_stream_checked, ) @@ -34,15 +33,3 @@ def generator(stream): raise RuntimeError("Stream end reached, but next value was requested") assert join_byte_stream_checked(generator(stream), max_size) is None - - -@pytest.mark.parametrize( - ["name", "expected"], - [ - ["pipeline_tasks.VolumeWorker.process", True], - ["scheduled_tasks.process_metrics", True], - ["GET /api/project", False], - ], -) -def test_is_background_task_name(name, expected): - assert is_background_task_name(name) is expected From bda09411904643abf8d64f9a1df55f3ca2d3ab4c Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Thu, 16 Jul 2026 15:20:22 +0500 Subject: [PATCH 9/9] Filter opentelemetry. logs --- mkdocs/docs/guides/server-deployment.md | 2 +- .../_internal/server/utils/otel/utils.py | 12 +++++++++++ .../_internal/server/utils/test_tracing.py | 20 +++++++++++++++++++ 3 files changed, 33 insertions(+), 1 deletion(-) diff --git a/mkdocs/docs/guides/server-deployment.md b/mkdocs/docs/guides/server-deployment.md index 8503b7f17c..d20e9d37e0 100644 --- a/mkdocs/docs/guides/server-deployment.md +++ b/mkdocs/docs/guides/server-deployment.md @@ -417,7 +417,7 @@ The server can export traces, logs, and metrics using OpenTelemetry. Each signal The export destination and protocol are configured via the standard [`OTEL_*` environment variables :material-arrow-top-right-thin:{.external }](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/). -Traces and logs are exported via OTLP over HTTP. Example: +Traces and logs are exported via OTLP over HTTP (`OTEL_EXPORTER_OTLP_PROTOCOL` has no effect). Example: ```shell $ DSTACK_OTEL_TRACES_ENABLED=1 \ diff --git a/src/dstack/_internal/server/utils/otel/utils.py b/src/dstack/_internal/server/utils/otel/utils.py index 3981227f6b..006591e7d8 100644 --- a/src/dstack/_internal/server/utils/otel/utils.py +++ b/src/dstack/_internal/server/utils/otel/utils.py @@ -146,9 +146,21 @@ def _get_resource() -> Resource: def _build_log_handler(logger_provider: LoggerProvider) -> logging.Handler: handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider) handler.addFilter(AsyncioCancelledErrorFilter()) + handler.addFilter(_OTelSDKLogFilter()) return handler +class _OTelSDKLogFilter(logging.Filter): + """Drops log records emitted by the OTel SDK itself. + + Exporting them would be a feedback loop: an OTLP export failure is logged, + the record is queued for the same failing exporter, and so on indefinitely. + """ + + def filter(self, record: logging.LogRecord) -> bool: + return not (record.name + ".").startswith("opentelemetry.") + + def _register_db_span_renaming(engine: Engine) -> None: """Renames DB spans from ` ` to `
`. diff --git a/src/tests/_internal/server/utils/test_tracing.py b/src/tests/_internal/server/utils/test_tracing.py index acda040944..02c09d1f17 100644 --- a/src/tests/_internal/server/utils/test_tracing.py +++ b/src/tests/_internal/server/utils/test_tracing.py @@ -125,6 +125,26 @@ def test_exports_records_with_trace_context(self, span_exporter: InMemorySpanExp assert records[0].severity_text == "WARN" assert records[0].trace_id == span.get_span_context().trace_id + def test_does_not_export_otel_sdk_records(self): + from opentelemetry.sdk._logs import LoggerProvider + from opentelemetry.sdk._logs.export import ( + InMemoryLogRecordExporter, + SimpleLogRecordProcessor, + ) + + log_exporter = InMemoryLogRecordExporter() + logger_provider = LoggerProvider() + logger_provider.add_log_record_processor(SimpleLogRecordProcessor(log_exporter)) + handler = _build_log_handler(logger_provider) + + # An OTLP export failure logged by the SDK must not be re-exported — + # that would be a feedback loop through the same failing exporter + sdk_logger = logging.getLogger("opentelemetry.exporter.otlp.proto.http._log_exporter") + sdk_logger.addHandler(handler) + sdk_logger.error("Failed to export logs batch") + + assert log_exporter.get_finished_logs() == () + class TestGetMetricsExporters: @pytest.mark.parametrize(