Support server monitoring with OpenTelemetry

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 (`OTEL_EXPORTER_OTLP_PROTOCOL` has no effect). 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 

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.

pyproject.toml

@@ -264,9 +264,20 @@ fluentbit = [
     "elasticsearch>=8.0.0",
     "dstack[server]",
 ]
+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 = [
     "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]",
 ]

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,
@@ -104,6 +104,11 @@ def create_app() -> FastAPI:
         ],
     )
     app.state.proxy_dependency_injector = ServerProxyDependencyInjector()
+    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(app, get_db().engine)
     return app
 
 

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(

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:

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)

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)

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:

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:

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:

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)

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(