diff --git a/CHANGES/2346.misc b/CHANGES/2346.misc new file mode 100644 index 00000000..c485daa3 --- /dev/null +++ b/CHANGES/2346.misc @@ -0,0 +1 @@ +Add the PULP-2346 A/B experiment for the content app's read-only content queries (the ``index.html`` probe, the directory listing and the exact file lookup). Variant B checks repository membership per matching row instead of walking the whole repository version's content set. The experiment is always on: half of the calls run variant B, and each call logs its variant, query, repository and duration. The queries rely on the ``core_ca_relpath_pattern_idx`` index added by migration 0024. diff --git a/Dockerfile b/Dockerfile index a133965d..5864dc35 100644 --- a/Dockerfile +++ b/Dockerfile @@ -187,6 +187,9 @@ RUN patch -p1 -d /usr/local/lib/pulp/lib/python${PYTHON_VERSION}/site-packages < COPY images/assets/patches/0079-Log-Maven-path-index-fallback-reasons.patch /tmp/ RUN patch -p1 -d /usr/local/lib/pulp/lib/python${PYTHON_VERSION}/site-packages < /tmp/0079-Log-Maven-path-index-fallback-reasons.patch +COPY images/assets/patches/0080-Optimize-read-only-content-queries.patch /tmp/ +RUN patch -p1 -d /usr/local/lib/pulp/lib/python${PYTHON_VERSION}/site-packages < /tmp/0080-Optimize-read-only-content-queries.patch + RUN mkdir /licenses COPY LICENSE /licenses/LICENSE diff --git a/images/assets/patches/0080-Optimize-read-only-content-queries.patch b/images/assets/patches/0080-Optimize-read-only-content-queries.patch new file mode 100644 index 00000000..7294e974 --- /dev/null +++ b/images/assets/patches/0080-Optimize-read-only-content-queries.patch @@ -0,0 +1,101 @@ +--- a/pulpcore/content/handler.py ++++ b/pulpcore/content/handler.py +@@ -50,7 +50,11 @@ + ) + from jinja2 import Template # noqa: E402 + +-from pulp_service.app.experiments import directory_membership_dates # noqa: E402 ++from pulp_service.app.experiments import ( # noqa: E402 ++ content_query_experiment, ++ directory_membership_dates, ++ single_or_raise, ++) + + from pulpcore.app import mime_types # noqa: E402 + from pulpcore.app.models import ( # noqa: E402 +@@ -785,11 +789,14 @@ + artifacts_to_find[pa.content_artifact.pk] = name + + if repo_version or publication.pass_through: +- cas = ContentArtifact.objects.select_related("artifact").filter( +- content__in=content_repo_ver.content, relative_path__startswith=path ++ ca_rows, cas = content_query_experiment( ++ "directory_listing", ++ content_repo_ver, ++ ContentArtifact.objects.select_related("artifact"), ++ filters={"relative_path__startswith": path}, + ) + content_sources.append((cas, "content_id")) +- for ca in cas: ++ for ca in ca_rows: + name = file_or_directory_name(path, ca.relative_path) + directory_list.add(name) + update_content_date(content_dates, ca.content_id, ca.pulp_created) +@@ -977,15 +984,18 @@ + # pass-through + if publication.pass_through: + try: +- ca = await self._retry_content_artifact_query( +- lambda: ContentArtifact.objects.select_related( +- "artifact", "artifact__pulp_domain" +- ) +- .filter( +- content__in=publication.repository_version.content, ++ rows, _ = await self._retry_content_artifact_query( ++ lambda: sync_to_async(content_query_experiment)( ++ "pass_through_file_lookup", ++ publication.repository_version, ++ ContentArtifact.objects.select_related( ++ "artifact", "artifact__pulp_domain" ++ ), ++ filters={"relative_path": original_rel_path}, ++ limit=2, + ) +- .aget(relative_path=original_rel_path) + ) ++ ca = single_or_raise(ContentArtifact, rows) + + except MultipleObjectsReturned: + log.error( +@@ -1022,11 +1032,16 @@ + # Look for index.html or list the directory + index_path = "{}index.html".format(rel_path) + +- contentartifact_exists = await self._retry_content_artifact_query( +- lambda: ContentArtifact.objects.filter( +- content__in=repo_version.content, relative_path=index_path +- ).aexists() ++ probe_rows, _ = await self._retry_content_artifact_query( ++ lambda: sync_to_async(content_query_experiment)( ++ "index_probe", ++ repo_version, ++ ContentArtifact.objects.values_list("pk", flat=True), ++ filters={"relative_path": index_path}, ++ limit=1, ++ ) + ) ++ contentartifact_exists = bool(probe_rows) + if contentartifact_exists: + original_rel_path = index_path + headers = self.response_headers(original_rel_path, distro) +@@ -1047,11 +1062,16 @@ + ) + + try: +- ca = await self._retry_content_artifact_query( +- lambda: ContentArtifact.objects.select_related( +- "artifact", "artifact__pulp_domain" +- ).aget(content__in=repo_version.content, relative_path=original_rel_path) ++ rows, _ = await self._retry_content_artifact_query( ++ lambda: sync_to_async(content_query_experiment)( ++ "file_lookup", ++ repo_version, ++ ContentArtifact.objects.select_related("artifact", "artifact__pulp_domain"), ++ filters={"relative_path": original_rel_path}, ++ limit=2, ++ ) + ) ++ ca = single_or_raise(ContentArtifact, rows) + + except MultipleObjectsReturned: + log.error( diff --git a/pulp_service/pulp_service/app/experiments.py b/pulp_service/pulp_service/app/experiments.py index 7056e700..79881f3d 100644 --- a/pulp_service/pulp_service/app/experiments.py +++ b/pulp_service/pulp_service/app/experiments.py @@ -4,14 +4,17 @@ import logging import random from time import perf_counter +from uuid import UUID from django.conf import settings from django.db import connections -from django.db.models import Q +from django.db.models import Exists, OuterRef, Q from django_guid import get_guid logger = logging.getLogger("pulp.experiment") DIRECTORY_EXPERIMENT = "PULP-2505" +CONTENT_QUERY_EXPERIMENT = "PULP-2346" +CONTENT_QUERY_P_CANDIDATE = 0.5 def run_experiment(exp_id, control, candidate, *, p_candidate, context): @@ -130,3 +133,59 @@ def directory_membership_dates(version, content_to_names, sources, *, serving_mo p_candidate=probability, context=context, ) + + +def _content_query_marker(variant, repository_id): + """Return the PULP-2346 marker for a variant and repository.""" + repository = UUID(str(repository_id)) + return f"/* pexp={CONTENT_QUERY_EXPERIMENT} v={variant} */ /* repo={repository} */" + + +def mark_lazy_queryset(queryset, marker): + """Embed an experiment marker in the SQL, so it also survives later evaluation (e.g. ``.values()``).""" + # The marker is built from constants and a UUID-validated repository id, never from user input. + return queryset.extra(where=[f"1 = 1 {marker}"]) # noqa: S610 + + +def content_query_experiment(query, version, queryset, *, filters, limit=None): + """Run one content app lookup as PULP-2346 and return ``(rows, queryset used)``. + + Variant A is stock pulpcore (``content__in=version.content``). Variant B checks repository + membership per matching row with a correlated EXISTS. Both return the same rows. Always on: + half of the calls run variant B. + """ + + def build(variant): + if variant == "A": + selected = queryset.filter(content__in=version.content, **filters) + else: + membership = version._content_relationships().filter(content_id=OuterRef("content_id")) + selected = queryset.filter(Exists(membership), **filters) + return mark_lazy_queryset(selected, _content_query_marker(variant, version.repository_id)) + + used = {} + + def run(variant): + def evaluate(): + used["queryset"] = build(variant) + return list(used["queryset"] if limit is None else used["queryset"][:limit]) + + return evaluate + + rows = run_experiment( + CONTENT_QUERY_EXPERIMENT, + run("A"), + run("B"), + p_candidate=CONTENT_QUERY_P_CANDIDATE, + context={"query": query, "repository_id": str(version.repository_id)}, + ) + return rows, used["queryset"] + + +def single_or_raise(model, rows): + """Return the only row, raising what ``QuerySet.get()`` would for none or several.""" + if not rows: + raise model.DoesNotExist(f"{model.__name__} matching query does not exist.") + if len(rows) > 1: + raise model.MultipleObjectsReturned(f"get() returned more than one {model.__name__}.") + return rows[0] diff --git a/pulp_service/pulp_service/tests/unit/test_content_query_experiment.py b/pulp_service/pulp_service/tests/unit/test_content_query_experiment.py new file mode 100644 index 00000000..a7af124c --- /dev/null +++ b/pulp_service/pulp_service/tests/unit/test_content_query_experiment.py @@ -0,0 +1,157 @@ +"""Database-backed checks for the PULP-2346 content query experiment.""" + +import json +from uuid import uuid4 + +import pytest +from django.db import connection +from django.test.utils import CaptureQueriesContext + +from pulpcore.app.models import Content, ContentArtifact, Repository + +from pulp_service.app.experiments import CONTENT_QUERY_EXPERIMENT, content_query_experiment, single_or_raise + +MEMBER_PATHS = ["org/a/a-1.0.jar", "org/a/a-1.0.pom", "org/b/index.html", "top.txt"] +OTHER_PATHS = ["org/b/b-2.0.jar", "org/a/not-a-member.jar"] + + +@pytest.fixture +def version(db): # noqa: ARG001 + """A repository version holding MEMBER_PATHS, with OTHER_PATHS existing outside of it.""" + repository = Repository.objects.create(name=uuid4()) + repository.CONTENT_TYPES = [Content] + members = [] + for path in [*MEMBER_PATHS, *OTHER_PATHS]: + content = Content.objects.create(pulp_type="core.content") + ContentArtifact.objects.create(content=content, relative_path=path) + if path in MEMBER_PATHS: + members.append(content.pk) + with repository.new_version() as new_version: + new_version.add_content(Content.objects.filter(pk__in=members)) + return new_version + + +@pytest.fixture +def force_variant(monkeypatch): + def _force(variant): + # random() < p_candidate (0.5) selects B, otherwise A. + monkeypatch.setattr("pulp_service.app.experiments.random.random", lambda: 0.1 if variant == "B" else 0.9) + + return _force + + +LOOKUPS = [ + pytest.param( + {"relative_path__startswith": "org/"}, + None, + ["org/a/a-1.0.jar", "org/a/a-1.0.pom", "org/b/index.html"], + id="directory-listing", + ), + pytest.param( + {"relative_path__startswith": "org/a/"}, None, ["org/a/a-1.0.jar", "org/a/a-1.0.pom"], id="specific-folder" + ), + pytest.param({"relative_path__startswith": "nothing/"}, None, [], id="missing-folder"), + pytest.param({"relative_path": "org/b/index.html"}, 1, ["org/b/index.html"], id="index-probe-hit"), + pytest.param({"relative_path": "org/a/index.html"}, 1, [], id="index-probe-miss"), + pytest.param({"relative_path": "org/a/a-1.0.jar"}, 2, ["org/a/a-1.0.jar"], id="file-lookup-hit"), + # Exists as a ContentArtifact, but its content is not in the version. + pytest.param({"relative_path": "org/b/b-2.0.jar"}, 2, [], id="file-lookup-not-in-version"), +] + + +@pytest.mark.parametrize(("filters", "limit", "expected"), LOOKUPS) +def test_variants_return_expected_rows(version, force_variant, filters, limit, expected): + """Both variants return exactly the version's matching artifacts, not just each other's rows.""" + results = {} + for variant in ("A", "B"): + force_variant(variant) + rows, _ = content_query_experiment( + "test", version, ContentArtifact.objects.select_related("artifact"), filters=filters, limit=limit + ) + results[variant] = sorted(row.relative_path for row in rows) + assert results["A"] == results["B"] == expected + + +def test_variants_use_their_own_query_shape(version, force_variant): + """A is the stock unnest(content_ids) query; B is the correlated EXISTS and never unnests.""" + sql = {} + for variant in ("A", "B"): + force_variant(variant) + with CaptureQueriesContext(connection) as ctx: + content_query_experiment( + "test", version, ContentArtifact.objects.all(), filters={"relative_path": "org/a/a-1.0.jar"}, limit=2 + ) + sql[variant] = " ".join(query["sql"] for query in ctx.captured_queries) + assert "unnest" in sql["A"] and "EXISTS" not in sql["A"] + assert "EXISTS" in sql["B"] and "unnest" not in sql["B"] + + +def test_candidate_index_probe_executes_membership_query(version, force_variant): + force_variant("B") + with CaptureQueriesContext(connection) as ctx: + rows, _ = content_query_experiment( + "index_probe", + version, + ContentArtifact.objects.all(), + filters={"relative_path": "org/b/index.html"}, + limit=1, + ) + assert [row.relative_path for row in rows] == ["org/b/index.html"] + assert any("EXISTS" in query["sql"] for query in ctx.captured_queries) + + +def test_logs_variant_query_and_repository(version, caplog, force_variant): + caplog.set_level("INFO", logger="pulp.experiment") + for variant in ("B", "A"): + force_variant(variant) + content_query_experiment( + "directory_listing", version, ContentArtifact.objects.all(), filters={"relative_path__startswith": "org/"} + ) + records = [json.loads(r.getMessage()) for r in caplog.records if r.name == "pulp.experiment"] + assert [record["variant"] for record in records] == ["B", "A"] + for record in records: + assert record["event"] == "ab_experiment" + assert record["exp_id"] == CONTENT_QUERY_EXPERIMENT + assert record["query"] == "directory_listing" + assert record["repository_id"] == str(version.repository_id) + assert record["outcome"] == "success" + assert record["result_count"] == 3 + assert record["duration_ms"] >= 0 + + +def test_sql_carries_variant_and_repository_markers(version, force_variant): + for variant in ("A", "B"): + force_variant(variant) + with CaptureQueriesContext(connection) as ctx: + content_query_experiment( + "test", version, ContentArtifact.objects.all(), filters={"relative_path__startswith": "org/"} + ) + marker = f"/* pexp={CONTENT_QUERY_EXPERIMENT} v={variant} */ /* repo={version.repository_id} */" + assert any(marker in query["sql"] for query in ctx.captured_queries), marker + # Once per statement: the collector adds a digest's load once for every marker it finds. + assert all(query["sql"].count("pexp=") == 1 for query in ctx.captured_queries) + + +def test_lazy_queryset_keeps_variant_marker(version, force_variant): + """A later subquery evaluation must retain the PULP-2346 marker.""" + for variant in ("A", "B"): + force_variant(variant) + _, queryset = content_query_experiment( + "directory_listing", + version, + ContentArtifact.objects.all(), + filters={"relative_path__startswith": "org/"}, + ) + marker = f"/* pexp={CONTENT_QUERY_EXPERIMENT} v={variant} */ /* repo={version.repository_id} */" + with CaptureQueriesContext(connection) as ctx: + list(queryset.values("content_id")) + assert any(marker in query["sql"] for query in ctx.captured_queries), marker + + +def test_single_or_raise_matches_get_semantics(version): # noqa: ARG001 + rows = list(ContentArtifact.objects.filter(relative_path="top.txt")) + assert single_or_raise(ContentArtifact, rows) == rows[0] + with pytest.raises(ContentArtifact.DoesNotExist): + single_or_raise(ContentArtifact, []) + with pytest.raises(ContentArtifact.MultipleObjectsReturned): + single_or_raise(ContentArtifact, list(ContentArtifact.objects.all()[:2]))