diff --git a/src/dstack/_internal/core/backends/aws/compute.py b/src/dstack/_internal/core/backends/aws/compute.py index 4fa3db952..39521cc7c 100644 --- a/src/dstack/_internal/core/backends/aws/compute.py +++ b/src/dstack/_internal/core/backends/aws/compute.py @@ -1291,6 +1291,19 @@ def _get_vpc_id_subnets_ids_by_vpc_name_or_error( "L-DB2E81BA": "G/OnDemand", } +# `GetServiceQuota` errors that say nothing about dstack: the quota stays unknown and +# the offer availability stays `UNKNOWN`. Any other error code is reported, since it +# may mean the request is malformed. +_EXPECTED_QUOTA_ERROR_CODES = { + "408", # request timed out + "TooManyRequestsException", # rate limits + "AccessDeniedException", # no servicequotas:GetServiceQuota permission + "AuthFailure", # invalid, expired, or deactivated credentials + "InvalidClientTokenId", + "RequestExpired", + "UnrecognizedClientException", +} + def _get_regions_to_quotas( session: boto3.Session, regions: List[str] @@ -1302,14 +1315,16 @@ def get_region_quotas(region_name: str, client: botocore.client.BaseClient) -> D resp = client.get_service_quota(ServiceCode="ec2", QuotaCode=quota_code) region_quotas[quota_class] = resp["Quota"]["Value"] except botocore.exceptions.ClientError as e: - if "TooManyRequestsException" in str(e): + error_code = e.response.get("Error", {}).get("Code", "") + if error_code in _EXPECTED_QUOTA_ERROR_CODES: logger.warning( - "Failed to get quota %s in %s due to rate limits", + "Failed to get quota %s in %s: %s", quota_code, region_name, + e, ) else: - logger.exception(e) + logger.exception("Failed to get quota %s in %s", quota_code, region_name) return region_quotas regions_to_quotas = {} diff --git a/src/dstack/_internal/server/background/pipeline_tasks/instances/check.py b/src/dstack/_internal/server/background/pipeline_tasks/instances/check.py index 9be54939c..793d0577d 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/instances/check.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/instances/check.py @@ -432,7 +432,7 @@ def _check_instance_inner( try: remove_dangling_tasks_from_instance(shim_client, instance) except Exception as exc: - logger.exception("%s: error removing dangling tasks: %s", fmt(instance), exc) + logger.warning("%s: error removing dangling tasks: %s", fmt(instance), exc) # There should be no shim API calls after this function call since it can request shim restart. _maybe_install_components(instance, shim_client) diff --git a/src/dstack/_internal/server/background/scheduled_tasks/gateways.py b/src/dstack/_internal/server/background/scheduled_tasks/gateways.py index 73d153730..2d7dbb93f 100644 --- a/src/dstack/_internal/server/background/scheduled_tasks/gateways.py +++ b/src/dstack/_internal/server/background/scheduled_tasks/gateways.py @@ -1,5 +1,6 @@ import asyncio +import httpx from sqlalchemy import select from dstack._internal.core.errors import SSHError @@ -60,4 +61,9 @@ async def _process_connection(conn: GatewayConnection): logger.warning("Connection to gateway %s failed: %s", conn.ip_address, e) return - await conn.try_collect_stats() + try: + await conn.try_collect_stats() + except httpx.HTTPError as e: + logger.warning("Failed to collect stats from gateway %s: %r", conn.ip_address, e) + except Exception: + logger.exception("Failed to collect stats from gateway %s", conn.ip_address) diff --git a/src/dstack/_internal/server/services/locking.py b/src/dstack/_internal/server/services/locking.py index 2a2b833c0..405681dc0 100644 --- a/src/dstack/_internal/server/services/locking.py +++ b/src/dstack/_internal/server/services/locking.py @@ -9,6 +9,10 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession +from dstack._internal.utils.logging import get_logger + +logger = get_logger(__name__) + KeyT = TypeVar("KeyT") @@ -134,12 +138,12 @@ async def advisory_lock_ctx( To prevent unreleased locks: - 1. When possible, prefer using `pg_advisory_xact_lock` instead of this context manager. + * When possible, prefer using `pg_advisory_xact_lock` instead of this context manager. `pg_advisory_xact_lock` is automatically released at the end of transaction. - 1. Prefer using `AsyncConnection` as `bind`. + * Prefer using `AsyncConnection` as `bind`. - 1. If using `AsyncSession` as `bind`, **do not** commit before exiting from the context manager. + * If using `AsyncSession` as `bind`, **do not** commit before exiting from the context manager. Committing will prompt `AsyncSession` to start a new transaction for releasing the lock, which may be assigned to a different database connection, which will fail to release. """ @@ -150,7 +154,7 @@ async def advisory_lock_ctx( yield finally: if dialect_name == "postgresql": - await bind.execute(select(func.pg_advisory_unlock(string_to_lock_id(resource)))) + await _release_advisory_lock(bind, resource) @asynccontextmanager @@ -165,7 +169,7 @@ async def try_advisory_lock_ctx( yield locked finally: if dialect_name == "postgresql" and locked: - await bind.execute(select(func.pg_advisory_unlock(string_to_lock_id(resource)))) + await _release_advisory_lock(bind, resource) _in_memory_locker = InMemoryResourceLocker() @@ -203,3 +207,18 @@ async def _wait_to_lock_many( if not left_to_lock: return await asyncio.sleep(delay) + + +async def _release_advisory_lock( + bind: Union[AsyncConnection, AsyncSession], resource: str +) -> None: + """ + Release an advisory lock, tolerating failures. + Releasing typically fails with `PendingRollbackError` because the connection has been + invalidated. In this case Postgres has already dropped the lock along with the + session and there is nothing left to release. + """ + try: + await bind.execute(select(func.pg_advisory_unlock(string_to_lock_id(resource)))) + except Exception as e: + logger.warning("Failed to release advisory lock on %s: %r", resource, e)