diff --git a/deployments/CI_DEPLOYMENTS.md b/deployments/CI_DEPLOYMENTS.md index c5f749ff..497e7ff2 100644 --- a/deployments/CI_DEPLOYMENTS.md +++ b/deployments/CI_DEPLOYMENTS.md @@ -43,11 +43,31 @@ It then handles deployments for: - the database, assuming an existing Azure PostgreSQL flexible server - the API Container App, assuming an existing Container Apps environment + (see "Log routing" below) - the entity-linkage Container App in the same environment - the stitch-llm Container App in the same environment - the ETL Container App (`etl`) in the same environment, on non-`development` lanes only (see below) +### Log routing + +Container App logs (structured JSON on stdout) are forwarded to a Log Analytics +workspace by the **Container Apps environment** (`appLogsConfiguration`), not by +this pipeline or the app — so every app in an environment, including per-PR +preview apps, shares one workspace. Current topology: `development` and +`staging` lanes both feed the non-prod workspace **`stitch-staging`** +(`STITCH-DEV-RG`); `production` is isolated in its own workspace. + +To change where a lane's logs go, edit that lane's environment (named by its +`AZURE_CONTAINER_APP_ENVIRONMENT` variable) — Portal: **Settings → Logging**, or +CLI: `az containerapp env update --logs-destination log-analytics +--logs-workspace-id --logs-workspace-key `. Keep "Parse JSON +logs into columns" **off** on every environment sharing a workspace, so the +table schema stays uniform (the KQL in [`PERFORMANCE.md`](./PERFORMANCE.md) and +`tools/analyze_logs.py` assume the raw-`Log_s` form). This binding lives outside +the repo, so recreating an environment reverts it to `None` (a greyed-out Logs +blade) — reconfigure it when that happens. + ### ETL pipelines (temporary POC wiring) The single `etl` Container App is deployed from a pre-built image published by diff --git a/deployments/PERFORMANCE.md b/deployments/PERFORMANCE.md index ab1b0215..5bd2dc21 100644 --- a/deployments/PERFORMANCE.md +++ b/deployments/PERFORMANCE.md @@ -7,6 +7,9 @@ the cloud, and readable straight from the terminal locally — so you can find slow/frequent queries from real data instead of guessing. This doc covers the basic loop: **enable capture → drive traffic → analyze**. +For *where* deployed logs are routed (which Log Analytics workspace each lane +feeds, and how to change it), see "Log routing" in +[`CI_DEPLOYMENTS.md`](./CI_DEPLOYMENTS.md). > The instrumentation lives in the app code > ([`deployments/api/src/stitch/api/observability/`](api/src/stitch/api/observability/)), @@ -23,7 +26,7 @@ Two structured log streams, distinguished by the `logger` field: | Logger | Emitted | Key fields | |---|---|---| | `stitch.observability.request` | once per HTTP request (always) | `route`, `method`, `status_code`, `duration_ms`, `db_query_count`, `db_time_ms`, `request_id` | -| `stitch.api.observability.query` | once per query above the slow threshold | `statement` (parameterized SQL, **no bound values**), `duration_ms`, `rowcount`, `route`, `request_id` | +| `stitch.api.observability.query` | once per query above the slow threshold | `statement` (parameterized SQL, **no bound values**), `duration_ms`, `rowcount`, `route`, `request_id`, `query_name` (when the query runs in a labeled scope) | > The request summary is emitted by the shared `stitch.observability` > middleware, so it logs under `stitch.observability.request` (the API's @@ -34,6 +37,14 @@ Two structured log streams, distinguished by the `logger` field: `db_query_count` on a request is the N+1 detector; the `query` stream tells you *which* statement is expensive. +`query_name` is a stable label (e.g. `resources.list_ids`, `resources.count`, +`resources.filter_options`) attached to the queries a request handler runs, so you +can pick out a specific query without matching on SQL text — useful when two +statements share a near-identical prefix (both list queries open with +`WITH resource_universe AS …`). It is present only for queries executed inside a +labeled scope; unlabeled queries (e.g. ORM-internal statements outside a handler) +omit the field entirely. + --- ## Step 1 — Enable capture (the knobs) diff --git a/deployments/api/src/stitch/api/db/merge_candidate_actions.py b/deployments/api/src/stitch/api/db/merge_candidate_actions.py index 2162884e..ad9e214e 100644 --- a/deployments/api/src/stitch/api/db/merge_candidate_actions.py +++ b/deployments/api/src/stitch/api/db/merge_candidate_actions.py @@ -23,6 +23,7 @@ MergeCandidateStatus, MergeCandidateView, ) +from stitch.api.observability.context import named_query from stitch.ogsi.model import OGFieldSource from stitch.ogsi.model.og_field import OilGasFieldBase from stitch.ogsi.model.types import OGSISrcKey @@ -204,7 +205,8 @@ async def list_merge_candidates(session: AsyncSession) -> list[MergeCandidateVie .options(selectinload(MergeCandidateModel.items)) .order_by(MergeCandidateModel.created.desc()) ) - candidates = (await session.scalars(stmt)).all() + with named_query("merge_candidates.list"): + candidates = (await session.scalars(stmt)).all() return [_candidate_to_view(candidate) for candidate in candidates] @@ -213,7 +215,8 @@ async def get_merge_candidate( candidate_id: int, licensed_sources: Collection[OGSISrcKey] | None = None, ) -> MergeCandidateDetailView: - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.detail.load"): + candidate = await _load_candidate_model(session, candidate_id) resource_ids = [ item.resource_id for item in sorted(candidate.items, key=lambda i: i.position) @@ -228,14 +231,16 @@ async def get_merge_candidate( # never delete). So a null-shell view always means "emptied by a merge", # never "missing"; no existence check is needed here. Revisit if a resource # hard-delete path is ever added. - by_id = await coalesce_resources_with_sources( - session, resource_ids, licensed_sources - ) + with named_query("merge_candidates.detail.coalesce"): + by_id = await coalesce_resources_with_sources( + session, resource_ids, licensed_sources + ) # `status` compares the resources' coalesced values; `values` lists every # contributing source tagged with the resource it's attached to, ranked by # the default source order (winner-first). - default_priority = await _default_source_priority(session) + with named_query("merge_candidates.detail.default_priority"): + default_priority = await _default_source_priority(session) fallback_priority = max(default_priority.values(), default=0) + 1 sources_with_priority = [ (rid, source, default_priority.get(source.source, fallback_priority)) @@ -254,14 +259,16 @@ async def create_merge_candidate( request: MergeCandidateCreateRequest, ) -> MergeCandidateView: resource_ids = _normalize_resource_ids(request.resource_ids) - await _load_mergeable_resources(session, resource_ids) + with named_query("merge_candidates.create.load_resources"): + await _load_mergeable_resources(session, resource_ids) fingerprint = _fingerprint(resource_ids) - existing = await session.scalar( - select(MergeCandidateModel) - .options(selectinload(MergeCandidateModel.items)) - .where(MergeCandidateModel.fingerprint == fingerprint) - ) + with named_query("merge_candidates.create.check_existing"): + existing = await session.scalar( + select(MergeCandidateModel) + .options(selectinload(MergeCandidateModel.items)) + .where(MergeCandidateModel.fingerprint == fingerprint) + ) if existing is not None: if existing.status == MergeCandidateStatus.PENDING: raise InvalidActionError( @@ -275,22 +282,23 @@ async def create_merge_candidate( f"An approved merge candidate already exists for resources {resource_ids}." ) - candidate = MergeCandidateModel.create(created_by=user, fingerprint=fingerprint) - session.add(candidate) - await session.flush() + with named_query("merge_candidates.create.persist"): + candidate = MergeCandidateModel.create(created_by=user, fingerprint=fingerprint) + session.add(candidate) + await session.flush() - session.add_all( - [ - MergeCandidateItemModel( - merge_candidate_id=candidate.id, - resource_id=resource_id, - position=position, - ) - for position, resource_id in enumerate(resource_ids) - ] - ) - await session.flush() - await session.refresh(candidate, ["items"]) + session.add_all( + [ + MergeCandidateItemModel( + merge_candidate_id=candidate.id, + resource_id=resource_id, + position=position, + ) + for position, resource_id in enumerate(resource_ids) + ] + ) + await session.flush() + await session.refresh(candidate, ["items"]) return _candidate_to_view(candidate) @@ -300,7 +308,8 @@ async def approve_merge_candidate( candidate_id: int, request: MergeCandidateReviewRequest | None = None, ) -> MergeCandidateView: - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.approve.load"): + candidate = await _load_candidate_model(session, candidate_id) if candidate.status != MergeCandidateStatus.PENDING: raise InvalidActionError( f"Merge candidate {candidate_id} is not pending; current status={candidate.status}." @@ -309,7 +318,8 @@ async def approve_merge_candidate( resource_ids = [ item.resource_id for item in sorted(candidate.items, key=lambda i: i.position) ] - await _load_mergeable_resources(session, resource_ids) + with named_query("merge_candidates.approve.load_resources"): + await _load_mergeable_resources(session, resource_ids) merged_resource = await apply_resource_merge( session=session, user=user, @@ -322,9 +332,9 @@ async def approve_merge_candidate( candidate.reviewed_by_id = user.id candidate.last_updated_by_id = user.id candidate.merged_resource_id = merged_resource.id - await session.flush() - - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.approve.persist"): + await session.flush() + candidate = await _load_candidate_model(session, candidate_id) return _candidate_to_view(candidate) @@ -334,7 +344,8 @@ async def deny_merge_candidate( candidate_id: int, request: MergeCandidateReviewRequest | None = None, ) -> MergeCandidateView: - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.deny.load"): + candidate = await _load_candidate_model(session, candidate_id) if candidate.status != MergeCandidateStatus.PENDING: raise InvalidActionError( f"Merge candidate {candidate_id} is not pending; current status={candidate.status}." @@ -345,6 +356,7 @@ async def deny_merge_candidate( candidate.reviewed_at = datetime.now(timezone.utc) candidate.reviewed_by_id = user.id candidate.last_updated_by_id = user.id - await session.flush() - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.deny.persist"): + await session.flush() + candidate = await _load_candidate_model(session, candidate_id) return _candidate_to_view(candidate) diff --git a/deployments/api/src/stitch/api/db/og_field_resource_actions.py b/deployments/api/src/stitch/api/db/og_field_resource_actions.py index 0565aab2..8485d358 100644 --- a/deployments/api/src/stitch/api/db/og_field_resource_actions.py +++ b/deployments/api/src/stitch/api/db/og_field_resource_actions.py @@ -13,6 +13,7 @@ ResourceNotFoundError, ) from stitch.api.auth import CurrentUser +from stitch.api.observability.context import named_query from stitch.api.entities import ( FILTER_OPTION_FIELDS, OGFieldQueryParams, @@ -73,16 +74,19 @@ async def query( ids_stmt = base_resource_query(params, licensed_sources) count_stmt = select(func.count()).select_from(ids_stmt.subquery()) - total = (await session.scalar(count_stmt)) or 0 + with named_query("resources.count"): + total = (await session.scalar(count_stmt)) or 0 ids_stmt = ids_stmt.limit(params.limit).offset(params.offset) - ids = list((await session.scalars(ids_stmt)).all()) + with named_query("resources.list_ids"): + ids = list((await session.scalars(ids_stmt)).all()) if not ids: return [], total # Hydrate the page with the shared SQL coalescer (same one the detail path # uses), then the shared list-item projection, in phase-1 order. - coalesced = await coalesce_resources(session, ids, licensed_sources) + with named_query("resources.list_hydrate"): + coalesced = await coalesce_resources(session, ids, licensed_sources) items = [resource_to_list_item_view(coalesced[rid]) for rid in ids] return items, total @@ -93,8 +97,11 @@ async def filter_options( ) -> dict[str, list[str]]: """Distinct coalesced values for every filterable field, in one query.""" options: dict[str, list[str]] = {field: [] for field in FILTER_OPTION_FIELDS} - for colname, value in await session.execute(filter_option_rows(licensed_sources)): - options[colname].append(value) + with named_query("resources.filter_options"): + for colname, value in await session.execute( + filter_option_rows(licensed_sources) + ): + options[colname].append(value) return options @@ -103,20 +110,22 @@ async def get( id: int, licensed_sources: Collection[OGSISrcKey] | None = None, ) -> OGFieldResource: - stmt = ( - select(ResourceModel) - .options(selectinload(ResourceModel.memberships)) - .where(ResourceModel.id == id) - ) - model = await session.scalar(stmt) - if model is None: - raise HTTPException( - status_code=HTTP_404_NOT_FOUND, detail=f"No Resource with id `{id}` found." + with named_query("resources.detail"): + stmt = ( + select(ResourceModel) + .options(selectinload(ResourceModel.memberships)) + .where(ResourceModel.id == id) + ) + model = await session.scalar(stmt) + if model is None: + raise HTTPException( + status_code=HTTP_404_NOT_FOUND, + detail=f"No Resource with id `{id}` found.", + ) + await session.refresh(model, ["memberships"]) + return await resource_model_to_entity( + session, model, licensed_sources=licensed_sources ) - await session.refresh(model, ["memberships"]) - return await resource_model_to_entity( - session, model, licensed_sources=licensed_sources - ) async def resolve_root_id(session: AsyncSession, id: int) -> int: @@ -132,23 +141,29 @@ async def resolve_root_id(session: AsyncSession, id: int) -> int: (``apply_resource_merge`` always targets a brand-new row), so ``get_root`` terminates. """ - model = await session.scalar(select(ResourceModel).where(ResourceModel.id == id)) - if model is None: - raise HTTPException( - status_code=HTTP_404_NOT_FOUND, detail=f"No Resource with id `{id}` found." + with named_query("resources.resolve_root"): + model = await session.scalar( + select(ResourceModel).where(ResourceModel.id == id) ) - if model.repointed_id is None: - return id - try: - return (await model.get_root(session)).id - except ResourceNotFoundError as exc: - # Unreachable under current invariants (FK + acyclic merges + the - # self-reference validator), but a corrupt repoint chain must not surface - # as an unhandled 500 on a read path. - raise HTTPException( - status_code=HTTP_404_NOT_FOUND, - detail=f"Resource `{id}` could not be resolved to a current resource.", - ) from exc + if model is None: + raise HTTPException( + status_code=HTTP_404_NOT_FOUND, + detail=f"No Resource with id `{id}` found.", + ) + if model.repointed_id is None: + return id + try: + return (await model.get_root(session)).id + except ResourceNotFoundError as exc: + # Unreachable under current invariants (FK + acyclic merges + the + # self-reference validator), but a corrupt repoint chain must not + # surface as an unhandled 500 on a read path. + raise HTTPException( + status_code=HTTP_404_NOT_FOUND, + detail=( + f"Resource `{id}` could not be resolved to a current resource." + ), + ) from exc async def get_resolved( @@ -193,9 +208,12 @@ async def field_source_values( # field_source_candidates ranks in SQL by the same tiered key as the coalesce # winner, so the row order is winner-first and rank is just the enumerate # index. Empty text can't be persisted, so every returned row is a real value. - rows = ( - await session.execute(field_source_candidates(root_id, field, licensed_sources)) - ).all() + with named_query("resources.field_source_values"): + rows = ( + await session.execute( + field_source_candidates(root_id, field, licensed_sources) + ) + ).all() return [ OGFieldSourceValueView( source=row.source, @@ -237,7 +255,8 @@ async def set_field_source_priority( status_code=422, detail=f"field={field} is not a known resource field.", ) - resource = await session.get(ResourceModel, id) + with named_query("resources.set_field_source_priority.load"): + resource = await session.get(ResourceModel, id) if resource is None: raise HTTPException( status_code=HTTP_404_NOT_FOUND, detail=f"No Resource with id `{id}` found." @@ -262,9 +281,10 @@ async def set_field_source_priority( # Eligibility + current effective order come from the same ranked read the GET # endpoint uses: licensed sources with a non-empty value for the field, # winner-first. - rows = ( - await session.execute(field_source_candidates(id, field, licensed_sources)) - ).all() + with named_query("resources.set_field_source_priority.candidates"): + rows = ( + await session.execute(field_source_candidates(id, field, licensed_sources)) + ).all() current_order = [row.source_pk for row in rows] eligible = set(current_order) requested = set(ordered_source_pks) @@ -281,24 +301,25 @@ async def set_field_source_priority( return await field_source_values(session, id, field, licensed_sources) source_by_pk = {row.source_pk: row.source for row in rows} - await session.execute( - delete(OGFieldResourceSourcePriority).where( - OGFieldResourceSourcePriority.resource_id == id, - OGFieldResourceSourcePriority.colname == field, - ) - ) - for priority, source_pk in enumerate(ordered_source_pks): - session.add( - OGFieldResourceSourcePriority.create( - created_by=user, - resource_id=id, - source=source_by_pk[source_pk], - source_pk=source_pk, - colname=field, - priority=priority, + with named_query("resources.set_field_source_priority.persist"): + await session.execute( + delete(OGFieldResourceSourcePriority).where( + OGFieldResourceSourcePriority.resource_id == id, + OGFieldResourceSourcePriority.colname == field, ) ) - await session.flush() + for priority, source_pk in enumerate(ordered_source_pks): + session.add( + OGFieldResourceSourcePriority.create( + created_by=user, + resource_id=id, + source=source_by_pk[source_pk], + source_pk=source_pk, + colname=field, + priority=priority, + ) + ) + await session.flush() return await field_source_values(session, id, field, licensed_sources) @@ -317,17 +338,20 @@ async def create( raise ResourceIntegrityError( f"Cannot create resource that has been repointed.\n\tNew: {repr(resource)}" ) - model = ResourceModel.create(created_by=user) - session.add(model) - await session.flush() - if resource.source_data: - src_models = await get_or_create_sources(session, user, resource.source_data) - res = await attach_sources_to_resource( - session=session, resource_id=model.id, source_rows=src_models, user=user - ) - return res - await session.refresh(model, ["memberships"]) - return await resource_model_to_entity(session, model) + with named_query("resources.create"): + model = ResourceModel.create(created_by=user) + session.add(model) + await session.flush() + if resource.source_data: + src_models = await get_or_create_sources( + session, user, resource.source_data + ) + res = await attach_sources_to_resource( + session=session, resource_id=model.id, source_rows=src_models, user=user + ) + return res + await session.refresh(model, ["memberships"]) + return await resource_model_to_entity(session, model) async def apply_resource_merge( @@ -351,7 +375,8 @@ async def apply_resource_merge( stmt = select(ResourceModel).where(ResourceModel.id.in_(unique_ids)) - results = (await session.scalars(stmt)).all() + with named_query("resources.merge.load"): + results = (await session.scalars(stmt)).all() missing_ids = set(unique_ids).difference(set([r.id for r in results])) if len(missing_ids) > 0: msg = f"Resources not found for ids: [{','.join(map(str, missing_ids))}]" @@ -372,20 +397,21 @@ async def apply_resource_merge( # originals are intentionally NOT carried over -- merging resets ordering to # default. (No-op reset today since the target is fresh; a later PR handles # an explicit reset if merge semantics ever preserve an existing resource.) - new_resource = ResourceModel.create(created_by=user) - session.add(new_resource) - await session.flush() + with named_query("resources.merge.apply"): + new_resource = ResourceModel.create(created_by=user) + session.add(new_resource) + await session.flush() - # all results are still members of the session - # changes will be picked up on commit - for res in results: - res.repointed_id = new_resource.id + # all results are still members of the session + # changes will be picked up on commit + for res in results: + res.repointed_id = new_resource.id - _ = await _repoint_memberships(session, user, new_resource.id, unique_ids) + _ = await _repoint_memberships(session, user, new_resource.id, unique_ids) - # Return the canonical resource entity - await session.refresh(new_resource, ["memberships"]) - return await resource_model_to_entity(session, new_resource) + # Return the canonical resource entity + await session.refresh(new_resource, ["memberships"]) + return await resource_model_to_entity(session, new_resource) async def _repoint_memberships( diff --git a/deployments/api/src/stitch/api/db/og_field_source_actions.py b/deployments/api/src/stitch/api/db/og_field_source_actions.py index 7924f34c..2abd5eec 100644 --- a/deployments/api/src/stitch/api/db/og_field_source_actions.py +++ b/deployments/api/src/stitch/api/db/og_field_source_actions.py @@ -11,6 +11,7 @@ ) from stitch.api.db.utils import partition_by_id_none from stitch.api.entities import OGFieldQueryParams, User +from stitch.api.observability.context import named_query from stitch.ogsi.model import OGFieldSource, OGFieldResource from stitch.ogsi.model.types import OGSISrcKey @@ -36,7 +37,8 @@ async def create_source( model = OilGasFieldSourceModel.create_from_entity(source, created_by=user) session.add(model) - await session.flush() + with named_query("sources.create"): + await session.flush() return model.as_entity() @@ -95,10 +97,10 @@ async def create_and_attach_sources( ) # Fail fast: validate the target before creating any source, so an invalid # resource_id never leaves a source insert to roll back. - resource = await _get_attachable_resource(session, resource_id) - - models = await _create_source_models(session, user, sources) - await _attach_source_models(session, resource, models, user) + with named_query("sources.create_and_attach"): + resource = await _get_attachable_resource(session, resource_id) + models = await _create_source_models(session, user, sources) + await _attach_source_models(session, resource, models, user) return [model.as_entity() for model in models] @@ -128,10 +130,9 @@ async def get_or_create_sources( data: Sequence[OGFieldSource], ) -> Sequence[OGFieldSource]: - return [ - src.as_entity() - for src in await _get_or_create_source_models(session, user, data) - ] + with named_query("sources.get_or_create"): + models = await _get_or_create_source_models(session, user, data) + return [src.as_entity() for src in models] async def _get_or_create_source_models( @@ -202,15 +203,17 @@ async def attach_sources_to_resource( user: User, ) -> OGFieldResource: """Link an OG field source to a resource via membership.""" - resource = await _get_attachable_resource(session, resource_id) - if len(source_rows) < 1: - raise ResourceIntegrityError( - f"Must pass at least 1 source row to attach to resource (id: `{resource_id}`)." - ) + with named_query("sources.attach"): + resource = await _get_attachable_resource(session, resource_id) + if len(source_rows) < 1: + raise ResourceIntegrityError( + f"Must pass at least 1 source row to attach to resource " + f"(id: `{resource_id}`)." + ) - src_models = await _get_or_create_source_models(session, user, source_rows) - await _attach_source_models(session, resource, src_models, user) - return await resource_model_to_entity(session, resource) + src_models = await _get_or_create_source_models(session, user, source_rows) + await _attach_source_models(session, resource, src_models, user) + return await resource_model_to_entity(session, resource) async def get_source( @@ -218,7 +221,8 @@ async def get_source( id: int, licensed_sources: Collection[OGSISrcKey] | None = None, ) -> OGFieldSource: - model = await session.get(OilGasFieldSourceModel, id) + with named_query("sources.detail"): + model = await session.get(OilGasFieldSourceModel, id) if model is None: raise SourceNotFoundError(f"No OG Field Source found for id `{id}`") if licensed_sources is not None and model.source not in licensed_sources: @@ -239,7 +243,8 @@ async def get_sources( session: AsyncSession, ids: Sequence[int] ) -> Sequence[OGFieldSource]: stmt = select(OilGasFieldSourceModel).where(OilGasFieldSourceModel.id.in_(ids)) - models = (await session.scalars(stmt)).all() + with named_query("sources.get_by_ids"): + models = (await session.scalars(stmt)).all() return [model.as_entity() for model in models] @@ -251,17 +256,20 @@ async def query( """Filtered/sorted/paginated source records (id-ordered) plus total count.""" stmt = base_source_query(params, licensed_sources) count_stmt = select(func.count()).select_from(stmt.subquery()) - total = (await session.scalar(count_stmt)) or 0 + with named_query("sources.count"): + total = (await session.scalar(count_stmt)) or 0 stmt = stmt.limit(params.limit).offset(params.offset) - ids = list((await session.scalars(stmt)).all()) + with named_query("sources.list_ids"): + ids = list((await session.scalars(stmt)).all()) if not ids: return (), total - headers = ( - await session.scalars( - select(OilGasFieldSourceModel).where(OilGasFieldSourceModel.id.in_(ids)) - ) - ).all() + with named_query("sources.list_hydrate"): + headers = ( + await session.scalars( + select(OilGasFieldSourceModel).where(OilGasFieldSourceModel.id.in_(ids)) + ) + ).all() by_id = {h.id: h for h in headers} return tuple(by_id[i].as_entity() for i in ids if i in by_id), total diff --git a/deployments/api/src/stitch/api/observability/context.py b/deployments/api/src/stitch/api/observability/context.py index 4104e70e..b6ea4ad8 100644 --- a/deployments/api/src/stitch/api/observability/context.py +++ b/deployments/api/src/stitch/api/observability/context.py @@ -1,20 +1,25 @@ -"""API-specific per-request DB stats for the query-timing listener. +"""API-specific per-request DB stats and query labels for the timing listener. The generic request context (request id / route / scenario) is owned by the shared ``stitch.observability.middleware``; consumers import those directly from -there. This module owns only the API-specific ``db_stats`` dict: the request -middleware sets it before handing off to the rest of the app, the SQLAlchemy -listener mutates it in place, and the middleware reads back the aggregated query -count / time once the request completes. +there. This module owns the API-specific pieces the query-timing listener reads: +the ``db_stats`` dict (the request middleware sets it before handing off to the +rest of the app, the SQLAlchemy listener mutates it in place, and the middleware +reads back the aggregated query count / time once the request completes) and the +optional ``query_name`` label a call site attaches to the queries it runs. """ +from collections.abc import Iterator +from contextlib import contextmanager from contextvars import ContextVar from typing import TypedDict __all__ = [ "DbStats", "db_stats_var", + "named_query", "new_db_stats", + "query_name_var", ] @@ -25,6 +30,28 @@ class DbStats(TypedDict): db_stats_var: ContextVar[DbStats | None] = ContextVar("stitch_db_stats", default=None) +# Optional label a call site attaches to the queries it runs, so the timing +# listener can tag its events with a stable ``query_name`` instead of relying on +# SQL-text matching. Defaults to ``None`` (unlabeled), and every statement run +# inside a ``named_query`` scope -- including ORM-emitted secondary queries and +# shared helpers -- inherits the active name. +query_name_var: ContextVar[str | None] = ContextVar("stitch_query_name", default=None) + def new_db_stats() -> DbStats: return {"count": 0, "time_ms": 0.0} + + +@contextmanager +def named_query(name: str) -> Iterator[None]: + """Label every DB query run within this scope with ``name``. + + Set at a logical-operation boundary (typically a DB action function). The + label is read by the query-timing listener and added to its emitted events; + it resets on exit so nothing leaks to later queries. + """ + token = query_name_var.set(name) + try: + yield + finally: + query_name_var.reset(token) diff --git a/deployments/api/src/stitch/api/observability/query_timing.py b/deployments/api/src/stitch/api/observability/query_timing.py index ffe53c5c..46eab2f2 100644 --- a/deployments/api/src/stitch/api/observability/query_timing.py +++ b/deployments/api/src/stitch/api/observability/query_timing.py @@ -20,7 +20,7 @@ scenario_var, ) -from .context import db_stats_var +from .context import db_stats_var, query_name_var from .sinks import emit_query_event try: # py3.12+: monotonic, nanosecond resolution @@ -93,16 +93,18 @@ def _after(conn, cursor, statement, parameters, context, executemany): except Exception: # pragma: no cover - driver dependent rowcount = None - emit_query_event( - { - "duration_ms": round(elapsed_ms, 2), - "rowcount": rowcount - if rowcount is not None and rowcount >= 0 - else None, - "executemany": executemany, - "statement": _normalize_statement(statement, statement_max_chars), - "request_id": request_id_var.get(), - "route": route_var.get(), - "scenario": scenario_var.get(), - } - ) + event = { + "duration_ms": round(elapsed_ms, 2), + "rowcount": rowcount if rowcount is not None and rowcount >= 0 else None, + "executemany": executemany, + "statement": _normalize_statement(statement, statement_max_chars), + "request_id": request_id_var.get(), + "route": route_var.get(), + "scenario": scenario_var.get(), + } + # Optional label from the active ``named_query`` scope. Omitted entirely + # when unset so existing log consumers see no new field for unlabeled + # queries. + if (query_name := query_name_var.get()) is not None: + event["query_name"] = query_name + emit_query_event(event) diff --git a/deployments/api/tests/observability/test_query_name_actions.py b/deployments/api/tests/observability/test_query_name_actions.py new file mode 100644 index 00000000..619d2af3 --- /dev/null +++ b/deployments/api/tests/observability/test_query_name_actions.py @@ -0,0 +1,428 @@ +"""End-to-end coverage that the wrapped DB action call sites emit the expected +``query_name`` labels. + +The unit tests in ``test_query_timing.py`` exercise ``named_query`` around raw +statements; these drive the real endpoints and action functions through the +router / action / engine path so a typo in a label or a misplaced scope is +caught. Query timing is registered on the integration engine with +``log_all_queries=True`` and the sink is captured, so every statement a request +runs is inspected. + +Assertions compare the *exact* set of labels an operation emits (not just a +subset), so a secondary statement that carries an unexpected label -- or a scope +that leaks onto the wrong query -- fails the test. Unlabeled statements +(connection setup, transaction control, ORM-internal reads that legitimately run +outside any ``named_query`` scope) are excluded: the label is optional by design, +and asserting their absence would just pin SQLite/driver noise. +""" + +import pytest +from httpx import AsyncClient +from sqlalchemy.ext.asyncio import AsyncSession + +from stitch.api.db import merge_candidate_actions as mca +from stitch.api.db import og_field_resource_actions as resource_actions +from stitch.api.db import og_field_source_actions as source_actions +from stitch.api.db.model import MembershipModel, MembershipStatus, ResourceModel +from stitch.api.db.model.oil_gas_field_source_value import ATTRIBUTE_NAMES +from stitch.api.entities import ( + MergeCandidateCreateRequest, + MergeCandidateReviewRequest, + OGFieldQueryParams, + User, +) +from stitch.api.observability import query_timing +from tests.utils import make_source_model + + +@pytest.fixture +def captured_query_events(integration_engine, monkeypatch) -> list[dict]: + """Capture every query event emitted while hitting the integration engine.""" + events: list[dict] = [] + monkeypatch.setattr( + query_timing, "emit_query_event", lambda event: events.append(event) + ) + query_timing.register_query_timing( + integration_engine.sync_engine, slow_query_ms=0, log_all_queries=True + ) + return events + + +def _labels(events: list[dict]) -> set[str]: + """The set of query_name labels present among captured events.""" + return {event["query_name"] for event in events if "query_name" in event} + + +def _assert_labels(events: list[dict], expected: set[str]) -> None: + """Assert the labeled events are *exactly* ``expected`` (no missing/extra labels).""" + actual = _labels(events) + assert actual == expected, f"unexpected labels: {actual ^ expected}" + + +class TestEndpointQueryLabels: + """Drive the real HTTP read endpoints and assert the exact labels each emits.""" + + @pytest.mark.anyio + async def test_list_endpoint_labels( + self, + integration_client: AsyncClient, + og_create_res_fact, + captured_query_events: list[dict], + ): + # A row must exist for the hydrate query to run (skipped when empty). + create = await integration_client.post( + "/oil-gas-fields/", + json=og_create_res_fact(name="Labeled Resource").model_dump(mode="json"), + ) + assert create.status_code == 200, create.text + + captured_query_events.clear() + response = await integration_client.get("/oil-gas-fields/") + assert response.status_code == 200, response.text + + _assert_labels( + captured_query_events, + {"resources.count", "resources.list_ids", "resources.list_hydrate"}, + ) + + @pytest.mark.anyio + async def test_filter_options_endpoint_labels( + self, + integration_client: AsyncClient, + captured_query_events: list[dict], + ): + # filter-options returns every filterable field in one combined query. + response = await integration_client.get("/oil-gas-fields/filter-options") + assert response.status_code == 200, response.text + + _assert_labels(captured_query_events, {"resources.filter_options"}) + + @pytest.mark.anyio + async def test_detail_endpoint_labels( + self, + integration_client: AsyncClient, + og_create_res_fact, + captured_query_events: list[dict], + ): + create = await integration_client.post( + "/oil-gas-fields/", + json=og_create_res_fact(name="Detail Resource").model_dump(mode="json"), + ) + assert create.status_code == 200, create.text + created_id = create.json()["id"] + + captured_query_events.clear() + response = await integration_client.get(f"/oil-gas-fields/{created_id}") + assert response.status_code == 200, response.text + + # get_resolved -> resolve_root_id then get; both statements labeled. + _assert_labels( + captured_query_events, {"resources.resolve_root", "resources.detail"} + ) + + +async def _attach_source( + session: AsyncSession, user: User, resource_id: int, source: str, **values +) -> int: + """Attach one active source (carrying ``values``) to a resource; return its pk.""" + model = make_source_model(source=source, created_by_id=user.id, **values) + session.add(model) + await session.flush() + session.add( + MembershipModel.create( + created_by=user, + resource_id=resource_id, + source=model.source, + source_pk=model.id, + status=MembershipStatus.ACTIVE, + ) + ) + await session.flush() + return model.id + + +class TestActionCallSiteLabels: + """Call the DB action functions directly and assert the exact labels each emits. + + Each operation is captured on its own (clear -> call -> assert exact set) so + the fan-out of every action -- including nested labeled helpers -- is pinned, + covering the write paths, sources.*, merge_candidates.*, and every + third-level sub-label that the endpoint tests above do not reach. + """ + + @pytest.mark.anyio + async def test_resource_read_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + captured_query_events: list[dict], + ): + session = seeded_integration_session + created = await resource_actions.create( + session, test_user, og_create_res_fact(name="Labelled") + ) + await session.commit() + + captured_query_events.clear() + await resource_actions.query(session, OGFieldQueryParams()) + _assert_labels( + captured_query_events, + {"resources.count", "resources.list_ids", "resources.list_hydrate"}, + ) + + captured_query_events.clear() + await resource_actions.filter_options(session) + _assert_labels(captured_query_events, {"resources.filter_options"}) + + captured_query_events.clear() + await resource_actions.get_resolved(session, created.id) + _assert_labels( + captured_query_events, {"resources.resolve_root", "resources.detail"} + ) + + captured_query_events.clear() + await resource_actions.field_source_values( + session, created.id, next(iter(ATTRIBUTE_NAMES)) + ) + _assert_labels( + captured_query_events, + {"resources.resolve_root", "resources.field_source_values"}, + ) + + @pytest.mark.anyio + async def test_create_label( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + captured_query_events: list[dict], + ): + session = seeded_integration_session + + captured_query_events.clear() + await resource_actions.create( + session, test_user, og_create_res_fact(name="Created") + ) + # create() with source_data fans out to the source helpers. + _assert_labels( + captured_query_events, + {"resources.create", "sources.get_or_create", "sources.attach"}, + ) + + @pytest.mark.anyio + async def test_merge_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + captured_query_events: list[dict], + ): + session = seeded_integration_session + first = await resource_actions.create( + session, test_user, og_create_res_fact(name="Merge A") + ) + second = await resource_actions.create( + session, test_user, og_create_res_fact(name="Merge B") + ) + await session.commit() + + captured_query_events.clear() + await resource_actions.apply_resource_merge( + session, test_user, [first.id, second.id] + ) + _assert_labels( + captured_query_events, {"resources.merge.load", "resources.merge.apply"} + ) + + @pytest.mark.anyio + async def test_set_field_source_priority_labels( + self, + seeded_integration_session: AsyncSession, + integration_session_factory, + test_user: User, + captured_query_events: list[dict], + ): + session = seeded_integration_session + resource = ResourceModel.create(created_by=test_user) + session.add(resource) + await session.flush() + rid = resource.id + await _attach_source(session, test_user, rid, "gem", country="Gemland") + await _attach_source(session, test_user, rid, "rmi", country="Rmiland") + await session.commit() + + # Both sources carry `country`, so both are eligible; reverse the current + # winner-first order to force an actual re-prioritization (the persist path). + current = await resource_actions.field_source_values(session, rid, "country") + reordered = [view.source_id for view in reversed(current)] + assert len(reordered) == 2 + + # Run in a fresh session (one per request in prod) so the `.load` get() + # issues SQL rather than hitting this session's identity map. + captured_query_events.clear() + async with integration_session_factory() as fresh: + await resource_actions.set_field_source_priority( + fresh, test_user, rid, "country", reordered + ) + await fresh.commit() + + # The action re-prioritizes, then returns field_source_values(), which + # itself resolves the root and re-reads the candidates. + _assert_labels( + captured_query_events, + { + "resources.set_field_source_priority.load", + "resources.set_field_source_priority.candidates", + "resources.set_field_source_priority.persist", + "resources.resolve_root", + "resources.field_source_values", + }, + ) + + @pytest.mark.anyio + async def test_source_action_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + source_maker, + captured_query_events: list[dict], + ): + session = seeded_integration_session + parent = await resource_actions.create( + session, test_user, og_create_res_fact(name="Parent") + ) + await session.commit() + + captured_query_events.clear() + created = await source_actions.create_source( + session, test_user, source_maker(managed=False, source="gem") + ) + _assert_labels(captured_query_events, {"sources.create"}) + + captured_query_events.clear() + await source_actions.create_and_attach_sources( + session, test_user, [source_maker(managed=False, source="rmi")], parent.id + ) + _assert_labels(captured_query_events, {"sources.create_and_attach"}) + + captured_query_events.clear() + await source_actions.get_or_create_sources( + session, test_user, [source_maker(managed=False, source="wm")] + ) + _assert_labels(captured_query_events, {"sources.get_or_create"}) + + captured_query_events.clear() + await source_actions.attach_sources_to_resource( + session, parent.id, [source_maker(managed=False, source="bc")], test_user + ) + _assert_labels(captured_query_events, {"sources.attach"}) + + captured_query_events.clear() + await source_actions.get_source(session, created.id) + _assert_labels(captured_query_events, {"sources.detail"}) + + captured_query_events.clear() + await source_actions.get_sources(session, [created.id]) + _assert_labels(captured_query_events, {"sources.get_by_ids"}) + + captured_query_events.clear() + await source_actions.query(session, OGFieldQueryParams()) + _assert_labels( + captured_query_events, + {"sources.count", "sources.list_ids", "sources.list_hydrate"}, + ) + + @pytest.mark.anyio + async def test_merge_candidate_action_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + captured_query_events: list[dict], + ): + session = seeded_integration_session + resources = [ + await resource_actions.create( + session, test_user, og_create_res_fact(name=f"Cand {i}") + ) + for i in range(4) + ] + await session.commit() + + captured_query_events.clear() + approved = await mca.create_merge_candidate( + session, + test_user, + MergeCandidateCreateRequest( + resource_ids=[resources[0].id, resources[1].id] + ), + ) + _assert_labels( + captured_query_events, + { + "merge_candidates.create.load_resources", + "merge_candidates.create.check_existing", + "merge_candidates.create.persist", + }, + ) + + captured_query_events.clear() + await mca.list_merge_candidates(session) + _assert_labels(captured_query_events, {"merge_candidates.list"}) + + captured_query_events.clear() + await mca.get_merge_candidate(session, approved.id) + _assert_labels( + captured_query_events, + { + "merge_candidates.detail.load", + "merge_candidates.detail.coalesce", + "merge_candidates.detail.default_priority", + }, + ) + + captured_query_events.clear() + await mca.approve_merge_candidate( + session, + test_user, + approved.id, + MergeCandidateReviewRequest(review_notes="ok"), + ) + # approve delegates the actual merge to apply_resource_merge. + _assert_labels( + captured_query_events, + { + "merge_candidates.approve.load", + "merge_candidates.approve.load_resources", + "merge_candidates.approve.persist", + "resources.merge.load", + "resources.merge.apply", + }, + ) + + captured_query_events.clear() + denied = await mca.create_merge_candidate( + session, + test_user, + MergeCandidateCreateRequest( + resource_ids=[resources[2].id, resources[3].id] + ), + ) + await mca.deny_merge_candidate( + session, + test_user, + denied.id, + MergeCandidateReviewRequest(review_notes="no"), + ) + _assert_labels( + captured_query_events, + { + "merge_candidates.create.load_resources", + "merge_candidates.create.check_existing", + "merge_candidates.create.persist", + "merge_candidates.deny.load", + "merge_candidates.deny.persist", + }, + ) diff --git a/deployments/api/tests/observability/test_query_timing.py b/deployments/api/tests/observability/test_query_timing.py index 1405f97b..6771db79 100644 --- a/deployments/api/tests/observability/test_query_timing.py +++ b/deployments/api/tests/observability/test_query_timing.py @@ -8,7 +8,11 @@ from sqlalchemy.ext.asyncio import create_async_engine from stitch.api.observability import query_timing -from stitch.api.observability.context import db_stats_var, new_db_stats +from stitch.api.observability.context import ( + db_stats_var, + named_query, + new_db_stats, +) from stitch.api.observability.query_timing import _START_KEY from stitch.observability import JsonFormatter, configure_logging @@ -57,6 +61,41 @@ async def test_accumulates_into_request_stats(self, timed_engine): assert stats["count"] == 2 assert stats["time_ms"] >= 0 + @pytest.mark.anyio + async def test_named_query_labels_event(self, timed_engine): + engine, captured = timed_engine + + async with engine.connect() as conn: + with named_query("resources.count"): + await conn.execute(text("SELECT 1")) + + assert len(captured) == 1 + assert captured[0]["query_name"] == "resources.count" + + @pytest.mark.anyio + async def test_query_name_absent_when_unset(self, timed_engine): + engine, captured = timed_engine + + async with engine.connect() as conn: + await conn.execute(text("SELECT 1")) + + assert len(captured) == 1 + # Field is omitted entirely for unlabeled queries so existing log + # consumers see no new key. + assert "query_name" not in captured[0] + + @pytest.mark.anyio + async def test_named_query_scope_resets(self, timed_engine): + engine, captured = timed_engine + + async with engine.connect() as conn: + with named_query("resources.count"): + await conn.execute(text("SELECT 1")) + await conn.execute(text("SELECT 2")) + + assert captured[0]["query_name"] == "resources.count" + assert "query_name" not in captured[1] + @pytest.mark.anyio async def test_respects_slow_query_threshold(self, monkeypatch): captured: list[dict] = []