Misc. updates

- DB query and schema optimizations
- Event logging
- `DSTACK_FF_EVENTS` feature flag
- More input validation
- More unit tests
- Various docs

Author: jvstme

docs/docs/reference/environment-variables.md

@@ -132,6 +132,12 @@ For more details on the options below, refer to the [server deployment](../guide
 - `DSTACK_SERVER_INSTANCE_HEALTH_TTL_SECONDS`{ #DSTACK_SERVER_INSTANCE_HEALTH_TTL_SECONDS } – Maximum age of instance health checks.
 - `DSTACK_SERVER_INSTANCE_HEALTH_MIN_COLLECT_INTERVAL_SECONDS`{ #DSTACK_SERVER_INSTANCE_HEALTH_MIN_COLLECT_INTERVAL_SECONDS } – Minimum time interval between consecutive health checks of the same instance.
 
+<!--
+TODO: uncomment after dropping DSTACK_FF_EVENTS
+
+- `DSTACK_SERVER_EVENTS_TTL_SECONDS` { #DSTACK_SERVER_EVENTS_TTL_SECONDS } - Maximum age of event records. Set to `0` to disable event storage. Defaults to 30 days.
+-->
+
 ??? info "Internal environment variables"
      The following environment variables are intended for development purposes:
 

src/dstack/_internal/core/models/events.py

@@ -1,9 +1,9 @@
-# TODO: docs
-
 import uuid
 from datetime import datetime
 from enum import Enum
-from typing import Optional
+from typing import Annotated, Optional
+
+from pydantic import Field
 
 from dstack._internal.core.models.common import CoreModel
 
@@ -18,16 +18,50 @@ class EventTargetType(str, Enum):
 
 
 class EventTarget(CoreModel):
-    type: str  # Holds EventTargetType; str for adding new types without breaking compatibility
-    project_id: Optional[uuid.UUID]
-    id: uuid.UUID
-    name: str
+    type: Annotated[
+        str,  # not using EventTargetType to allow adding new types without breaking compatibility
+        Field(
+            description=(
+                f"Type of the target entity."
+                f" One of: {', '.join([f'`{t}`' for t in EventTargetType])}"
+            )
+        ),
+    ]
+    project_id: Annotated[
+        Optional[uuid.UUID],
+        Field(
+            description=(
+                "ID of the project the target entity belongs to,"
+                " or `null` for target types not bound to a project (e.g., users)"
+            )
+        ),
+    ]
+    id: Annotated[uuid.UUID, Field(description="ID of the target entity")]
+    name: Annotated[str, Field(description="Name of the target entity")]
 
 
 class Event(CoreModel):
     id: uuid.UUID
     message: str
     recorded_at: datetime
-    actor_user_id: Optional[uuid.UUID]
-    actor_user: Optional[str]
-    targets: list[EventTarget]
+    actor_user_id: Annotated[
+        Optional[uuid.UUID],
+        Field(
+            description=(
+                "ID of the user who performed the action that triggered the event,"
+                " or `null` if the action was performed by the system"
+            )
+        ),
+    ]
+    actor_user: Annotated[
+        Optional[str],
+        Field(
+            description=(
+                "Name of the user who performed the action that triggered the event,"
+                " or `null` if the action was performed by the system"
+            )
+        ),
+    ]
+    targets: Annotated[
+        list[EventTarget], Field(description="List of entities affected by the event")
+    ]

src/dstack/_internal/server/app.py

@@ -66,7 +66,7 @@
     get_client_version,
     get_server_client_error_details,
 )
-from dstack._internal.settings import DSTACK_VERSION
+from dstack._internal.settings import DSTACK_VERSION, FeatureFlags
 from dstack._internal.utils.logging import get_logger
 from dstack._internal.utils.ssh import check_required_ssh_version
 
@@ -229,7 +229,7 @@ def register_routes(app: FastAPI, ui: bool = True):
     app.include_router(model_proxy.router, prefix="/proxy/models", tags=["model-proxy"])
     app.include_router(prometheus.router)
     app.include_router(files.router)
-    app.include_router(events.root_router)
+    app.include_router(events.root_router, include_in_schema=FeatureFlags.EVENTS)
 
     @app.exception_handler(ForbiddenError)
     async def forbidden_error_handler(request: Request, exc: ForbiddenError):

src/dstack/_internal/server/background/__init__.py

@@ -33,6 +33,7 @@
     process_terminating_jobs,
 )
 from dstack._internal.server.background.tasks.process_volumes import process_submitted_volumes
+from dstack._internal.settings import FeatureFlags
 
 _scheduler = AsyncIOScheduler()
 
@@ -70,7 +71,8 @@ def start_background_tasks() -> AsyncIOScheduler:
     _scheduler.add_job(process_probes, IntervalTrigger(seconds=3, jitter=1))
     _scheduler.add_job(collect_metrics, IntervalTrigger(seconds=10), max_instances=1)
     _scheduler.add_job(delete_metrics, IntervalTrigger(minutes=5), max_instances=1)
-    _scheduler.add_job(delete_events, IntervalTrigger(minutes=7), max_instances=1)
+    if FeatureFlags.EVENTS:
+        _scheduler.add_job(delete_events, IntervalTrigger(minutes=7), max_instances=1)
     if settings.ENABLE_PROMETHEUS_METRICS:
         _scheduler.add_job(
             collect_prometheus_metrics, IntervalTrigger(seconds=10), max_instances=1

…4a29cd38_add_events_and_event_targets.py …4df9897e_add_events_and_event_targets.pysrc/dstack/_internal/server/migrations/versions/f27d4a29cd38_add_events_and_event_targets.py renamed to src/dstack/_internal/server/migrations/versions/22d74df9897e_add_events_and_event_targets.py src/dstack/_internal/server/migrations/versions/f27d4a29cd38_add_events_and_event_targets.py renamed to src/dstack/_internal/server/migrations/versions/22d74df9897e_add_events_and_event_targets.py

@@ -1,8 +1,8 @@
 """Add events and event_targets
 
-Revision ID: f27d4a29cd38
-Revises: 7d1ec2b920ac
-Create Date: 2025-11-26 01:01:46.305815
+Revision ID: 22d74df9897e
+Revises: 5fd659afca82
+Create Date: 2025-12-04 20:56:08.003504
 
 """
 
@@ -13,8 +13,8 @@
 import dstack._internal.server.models
 
 # revision identifiers, used by Alembic.
-revision = "f27d4a29cd38"
-down_revision = "7d1ec2b920ac"
+revision = "22d74df9897e"
+down_revision = "5fd659afca82"
 branch_labels = None
 depends_on = None
 
@@ -68,9 +68,15 @@ def upgrade() -> None:
         sa.PrimaryKeyConstraint("id", name=op.f("pk_event_targets")),
     )
     with op.batch_alter_table("event_targets", schema=None) as batch_op:
+        batch_op.create_index(
+            batch_op.f("ix_event_targets_entity_id"), ["entity_id"], unique=False
+        )
         batch_op.create_index(
             batch_op.f("ix_event_targets_entity_project_id"), ["entity_project_id"], unique=False
         )
+        batch_op.create_index(
+            batch_op.f("ix_event_targets_entity_type"), ["entity_type"], unique=False
+        )
         batch_op.create_index(batch_op.f("ix_event_targets_event_id"), ["event_id"], unique=False)
 
     # ### end Alembic commands ###
@@ -80,7 +86,9 @@ def downgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
     with op.batch_alter_table("event_targets", schema=None) as batch_op:
         batch_op.drop_index(batch_op.f("ix_event_targets_event_id"))
+        batch_op.drop_index(batch_op.f("ix_event_targets_entity_type"))
         batch_op.drop_index(batch_op.f("ix_event_targets_entity_project_id"))
+        batch_op.drop_index(batch_op.f("ix_event_targets_entity_id"))
 
     op.drop_table("event_targets")
     with op.batch_alter_table("events", schema=None) as batch_op:

src/dstack/_internal/server/migrations/versions/5fd659afca82_add_ix_instances_fleet_id.py

@@ -0,0 +1,31 @@
+"""Add ix_instances_fleet_id
+
+Revision ID: 5fd659afca82
+Revises: d4d9dc26cf58
+Create Date: 2025-12-04 20:52:07.015334
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "5fd659afca82"
+down_revision = "d4d9dc26cf58"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("instances", schema=None) as batch_op:
+        batch_op.create_index(batch_op.f("ix_instances_fleet_id"), ["fleet_id"], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("instances", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_instances_fleet_id"))
+
+    # ### end Alembic commands ###

src/dstack/_internal/server/migrations/versions/d4d9dc26cf58_add_ix_jobs_run_id.py

@@ -0,0 +1,31 @@
+"""Add ix_jobs_run_id
+
+Revision ID: d4d9dc26cf58
+Revises: 006512f572b4
+Create Date: 2025-12-04 20:48:10.543248
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "d4d9dc26cf58"
+down_revision = "006512f572b4"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("jobs", schema=None) as batch_op:
+        batch_op.create_index(batch_op.f("ix_jobs_run_id"), ["run_id"], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("jobs", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_jobs_run_id"))
+
+    # ### end Alembic commands ###

src/dstack/_internal/server/models.py

@@ -408,7 +408,9 @@ class JobModel(BaseModel):
     project_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"))
     project: Mapped["ProjectModel"] = relationship()
 
-    run_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("runs.id", ondelete="CASCADE"))
+    run_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("runs.id", ondelete="CASCADE"), index=True
+    )
     run: Mapped["RunModel"] = relationship()
 
     # Jobs need to reference fleets because we may choose an optimal fleet for a master job
@@ -602,7 +604,7 @@ class InstanceModel(BaseModel):
     )
     pool: Mapped[Optional["PoolModel"]] = relationship(back_populates="instances")
 
-    fleet_id: Mapped[Optional[uuid.UUID]] = mapped_column(ForeignKey("fleets.id"))
+    fleet_id: Mapped[Optional[uuid.UUID]] = mapped_column(ForeignKey("fleets.id"), index=True)
     fleet: Mapped[Optional["FleetModel"]] = relationship(back_populates="instances")
 
     compute_group_id: Mapped[Optional[uuid.UUID]] = mapped_column(ForeignKey("compute_groups.id"))
@@ -858,16 +860,14 @@ class SecretModel(BaseModel):
 class EventModel(BaseModel):
     __tablename__ = "events"
 
-    id: Mapped[uuid.UUID] = mapped_column(
-        UUIDType(binary=False), primary_key=True, default=uuid.uuid4
-    )
+    id: Mapped[uuid.UUID] = mapped_column(UUIDType(binary=False), primary_key=True)
     message: Mapped[str] = mapped_column(Text)
     recorded_at: Mapped[datetime] = mapped_column(NaiveDateTime, index=True)
 
     actor_user_id: Mapped[Optional[uuid.UUID]] = mapped_column(
         ForeignKey("users.id", ondelete="CASCADE"), nullable=True, index=True
     )
-    user: Mapped[Optional["UserModel"]] = relationship()
+    actor_user: Mapped[Optional["UserModel"]] = relationship()
 
     targets: Mapped[List["EventTargetModel"]] = relationship(back_populates="event")
 
@@ -889,6 +889,8 @@ class EventTargetModel(BaseModel):
     )
     entity_project: Mapped[Optional["ProjectModel"]] = relationship()
 
-    entity_type: Mapped[EventTargetType] = mapped_column(EnumAsString(EventTargetType, 100))
-    entity_id: Mapped[uuid.UUID] = mapped_column(UUIDType(binary=False))
+    entity_type: Mapped[EventTargetType] = mapped_column(
+        EnumAsString(EventTargetType, 100), index=True
+    )
+    entity_id: Mapped[uuid.UUID] = mapped_column(UUIDType(binary=False), index=True)
     entity_name: Mapped[str] = mapped_column(String(200))

src/dstack/_internal/server/routers/events.py

@@ -2,6 +2,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import dstack._internal.server.services.events as events_services
+from dstack._internal.core.errors import ServerClientError
 from dstack._internal.core.models.events import Event
 from dstack._internal.server.db import get_session
 from dstack._internal.server.models import UserModel
@@ -11,6 +12,7 @@
     CustomORJSONResponse,
     get_base_api_additional_responses,
 )
+from dstack._internal.settings import FeatureFlags
 
 root_router = APIRouter(
     prefix="/api/events",
@@ -28,9 +30,14 @@ async def list_events(
     """
     Returns events visible to the current user.
 
+    Regular users can see events related to themselves and to projects they are members of.
+    Global admins can see all events.
+
     The results are paginated. To get the next page, pass `recorded_at` and `id` of
     the last event from the previous page as `prev_recorded_at` and `prev_id`.
     """
+    if not FeatureFlags.EVENTS:
+        raise ServerClientError("Events are disabled on this server")
     return CustomORJSONResponse(
         await events_services.list_events(
             session=session,