Run the read-only content queries as a PULP-2346 A/B experiment

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 and skips the probe for file paths. On by default; disable with ``PULP_CONTENT_QUERY_AB_ENABLED=false`` and set the share of variant B with ``PULP_CONTENT_QUERY_AB_PROBABILITY``. The queries rely on the ``core_ca_relpath_pattern_idx`` index added by migration 0024.

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
 

deploy/clowdapp.yaml

@@ -621,6 +621,11 @@ objects:
             value: ${PULP_CONTENT_DIRECTORY_AB_PROBABILITY}
           - name: PULP_CONTENT_DIRECTORY_AB_REVISION
             value: ${PULP_CONTENT_DIRECTORY_AB_REVISION}
+          # PULP-2346 A/B experiment: read-only content queries (content app only).
+          - name: PULP_CONTENT_QUERY_AB_ENABLED
+            value: ${PULP_CONTENT_QUERY_AB_ENABLED}
+          - name: PULP_CONTENT_QUERY_AB_PROBABILITY
+            value: ${PULP_CONTENT_QUERY_AB_PROBABILITY}
           - name: PULP_FEATURE_SERVICE_API_CERT_PATH
             value: ${{PULP_FEATURE_SERVICE_API_CERT_PATH}}
           - name: PULP_FEATURE_SERVICE_API_URL
@@ -1819,6 +1824,12 @@ parameters:
   - name: PULP_CONTENT_DIRECTORY_AB_REVISION
     description: Deployed image git SHA tagged on PULP-2505 experiment log lines (required when enabled).
     value: ""
+  - name: PULP_CONTENT_QUERY_AB_ENABLED
+    description: Enable the PULP-2346 read-only content query A/B experiment (content app).
+    value: "true"
+  - name: PULP_CONTENT_QUERY_AB_PROBABILITY
+    description: Probability of running candidate variant B for PULP-2346.
+    value: "0.5"
   - name: PULP_AWS_DEFAULT_ACL
     description: AWS default ACL
     value: "@none None"

images/assets/patches/0080-Optimize-read-only-content-queries.patch

@@ -0,0 +1,104 @@
+--- 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,19 @@
+             # 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()
++            # File-ish paths (no trailing slash, dotted last segment) have no index.html to find.
++            last_segment = rel_path.rstrip("/").rsplit("/", 1)[-1]
++            probe_rows, _ = await self._retry_content_artifact_query(
++                lambda: sync_to_async(content_query_experiment)(
++                    "index_probe",
++                    repo_version,
++                    ContentArtifact.objects.all(),
++                    filters={"relative_path": index_path},
++                    limit=1,
++                    skip_candidate=not ends_in_slash and "." in last_segment,
++                )
+             )
++            contentartifact_exists = bool(probe_rows)
+             if contentartifact_exists:
+                 original_rel_path = index_path
+                 headers = self.response_headers(original_rel_path, distro)
+@@ -1047,11 +1065,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(

pulp_service/pulp_service/app/experiments.py

@@ -4,14 +4,16 @@
 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"
 
 
 def run_experiment(exp_id, control, candidate, *, p_candidate, context):
@@ -130,3 +132,71 @@ def directory_membership_dates(version, content_to_names, sources, *, serving_mo
         p_candidate=probability,
         context=context,
     )
+
+
+def _marked_content_query(variant, repository_id):
+    """Prefix each executed statement with its experiment, variant and repository (for RDS Performance Insights)."""
+    repository = UUID(str(repository_id))
+
+    def mark(execute, sql, params, many, context):
+        return execute(
+            f"/* pexp={CONTENT_QUERY_EXPERIMENT} v={variant} */ /* repo={repository} */ {sql}", params, many, context
+        )
+
+    return mark
+
+
+def content_query_experiment(query, version, queryset, *, filters, limit=None, skip_candidate=False):
+    """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, and with ``skip_candidate`` skips the
+    query. Both return the same rows. Disabled by default; then only variant A runs, unmarked.
+    """
+
+    def build(variant):
+        if variant == "A" or not settings_enabled():
+            return queryset.filter(content__in=version.content, **filters)
+        if skip_candidate:
+            return queryset.none()
+        membership = version._content_relationships().filter(content_id=OuterRef("content_id"))
+        return queryset.filter(Exists(membership), **filters)
+
+    used = {}
+
+    def run(variant):
+        def evaluate():
+            used["queryset"] = build(variant)
+            with connections[used["queryset"].db].execute_wrapper(
+                _marked_content_query(variant, version.repository_id)
+            ):
+                return list(used["queryset"] if limit is None else used["queryset"][:limit])
+
+        return evaluate
+
+    if not settings_enabled():
+        used["queryset"] = build("A")
+        return list(used["queryset"] if limit is None else used["queryset"][:limit]), used["queryset"]
+
+    rows = run_experiment(
+        CONTENT_QUERY_EXPERIMENT,
+        run("A"),
+        run("B"),
+        p_candidate=getattr(settings, "CONTENT_QUERY_AB_PROBABILITY", 0.5),
+        context={"query": query, "repository_id": str(version.repository_id)},
+    )
+    return rows, used["queryset"]
+
+
+def settings_enabled():
+    """Whether the PULP-2346 content query experiment is switched on."""
+    return bool(getattr(settings, "CONTENT_QUERY_AB_ENABLED", False))
+
+
+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]

pulp_service/pulp_service/tests/unit/test_content_query_experiment.py

@@ -0,0 +1,137 @@
+"""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 experiment_on(settings):
+    settings.CONTENT_QUERY_AB_ENABLED = True
+    settings.CONTENT_QUERY_AB_PROBABILITY = 0.5
+
+
+@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, False, id="directory-listing"),
+    pytest.param({"relative_path__startswith": "org/a/"}, None, False, id="specific-folder"),
+    pytest.param({"relative_path__startswith": "nothing/"}, None, False, id="missing-folder"),
+    pytest.param({"relative_path": "org/b/index.html"}, 1, False, id="index-probe-hit"),
+    pytest.param({"relative_path": "org/a/index.html"}, 1, False, id="index-probe-miss"),
+    pytest.param({"relative_path": "org/a/a-1.0.jar"}, 2, False, id="file-lookup-hit"),
+    pytest.param({"relative_path": "org/b/b-2.0.jar"}, 2, False, id="file-lookup-not-in-version"),
+]
+
+
+@pytest.mark.parametrize(("filters", "limit", "skip"), LOOKUPS)
+def test_variants_return_same_rows(version, experiment_on, force_variant, filters, limit, skip):  # noqa: ARG001
+    """Variant A (stock) and variant B (EXISTS) must return identical content artifacts."""
+    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"]
+
+
+def test_skip_candidate_returns_no_rows_without_querying(version, experiment_on, force_variant):  # noqa: ARG001
+    force_variant("B")
+    with CaptureQueriesContext(connection) as ctx:
+        rows, _ = content_query_experiment(
+            "index_probe",
+            version,
+            ContentArtifact.objects.all(),
+            filters={"relative_path": "org/a/a-1.0.jarindex.html"},
+            limit=1,
+            skip_candidate=True,
+        )
+    assert rows == []
+    assert not ctx.captured_queries
+
+
+def test_disabled_runs_stock_query_only_and_logs_nothing(version, settings, caplog, force_variant):
+    settings.CONTENT_QUERY_AB_ENABLED = False
+    force_variant("B")
+    with CaptureQueriesContext(connection) as ctx:
+        rows, _ = content_query_experiment(
+            "test", version, ContentArtifact.objects.all(), filters={"relative_path__startswith": "org/a/"}
+        )
+    assert sorted(row.relative_path for row in rows) == ["org/a/a-1.0.jar", "org/a/a-1.0.pom"]
+    assert not [record for record in caplog.records if record.name == "pulp.experiment"]
+    assert not any("pexp=" in query["sql"] for query in ctx.captured_queries)
+    assert any("content_ids" in query["sql"] for query in ctx.captured_queries)
+
+
+def test_enabled_logs_variant_query_and_repository(version, experiment_on, caplog, force_variant):  # noqa: ARG001
+    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, experiment_on, force_variant):  # noqa: ARG001
+    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
+
+
+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]))