Add UpstreamPulp.remote_policy for remotes created during replication

CHANGES/+remote-policy.feature

@@ -0,0 +1 @@
+Added `UpstreamPulp.remote_policy` so remotes created during replication can use `on_demand` or `streamed` instead of defaulting to `immediate`.

CHANGES/+replicate-ssl-tempfiles.bugfix

@@ -0,0 +1 @@
+Fixed replicate() deleting temporary TLS files before pulp-glue could use them, and stopped leaking PULP_CA_BUNDLE into later worker tasks.

docs/user/guides/replication.md

@@ -48,6 +48,7 @@ pulp upstream-pulp create \
 | `tls_validation` | Whether to verify the upstream server's TLS certificate. Defaults to `True`. |
 | `q_select` | A filter expression to select which upstream distributions to replicate. See [Filtering Distributions](#filtering-distributions-with-q_select). |
 | `policy` | Controls how replication manages local objects. One of `all`, `labeled`, or `nodelete`. See [Replication Policies](#replication-policies). Defaults to `all`. |
+| `remote_policy` | Download policy for remotes created during replication. One of `immediate`, `on_demand`, or `streamed`. Distinct from `policy`. When unset, remotes use Pulp's default (`immediate`). |
 
 ## Running Replication
 
@@ -151,7 +152,8 @@ pulp upstream-pulp replicate --upstream-pulp "my-upstream"
 ## Replication Policies
 
 The `policy` field controls how replication handles local objects, particularly when upstream
-distributions are removed or no longer match a `q_select` filter.
+distributions are removed or no longer match a `q_select` filter. It is not the same as a remote's
+download policy (`immediate`, `on_demand`, or `streamed`); set that with `remote_policy`.
 
 ### `all` (default)
 

pulpcore/app/migrations/0157_upstreampulp_remote_policy.py

@@ -0,0 +1,34 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0156_alter_contentartifact_relative_path_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="upstreampulp",
+            name="remote_policy",
+            field=models.TextField(
+                choices=[
+                    ("immediate", "When syncing, download all metadata and content now."),
+                    (
+                        "on_demand",
+                        "When syncing, download metadata, but do not download content now. "
+                        "Instead, download content as clients request it, and save it in Pulp "
+                        "to be served for future client requests.",
+                    ),
+                    (
+                        "streamed",
+                        "When syncing, download metadata, but do not download content now. "
+                        "Instead,download content as clients request it, but never save it in "
+                        "Pulp. This causes future requests for that same content to have to be "
+                        "downloaded again.",
+                    ),
+                ],
+                null=True,
+            ),
+        ),
+    ]

pulpcore/app/models/replica.py

@@ -11,6 +11,8 @@
 from pulpcore.app.util import get_domain_pk
 from pulpcore.plugin.models import AutoAddObjPermsMixin, BaseModel, EncryptedTextField
 
+from .repository import Remote
+
 
 class UpstreamPulp(BaseModel, AutoAddObjPermsMixin):
     ALL = "all"
@@ -59,6 +61,7 @@ class UpstreamPulp(BaseModel, AutoAddObjPermsMixin):
     sock_read_timeout = models.FloatField(
         null=True, validators=[MinValueValidator(0.0, "Timeout must be >= 0")]
     )
+    remote_policy = models.TextField(choices=Remote.POLICY_CHOICES, null=True)
 
     q_select = models.TextField(null=True)
     policy = models.TextField(choices=POLICY_CHOICES, default=ALL)

pulpcore/app/serializers/replica.py

@@ -3,7 +3,7 @@
 from rest_framework import serializers
 from rest_framework.validators import UniqueValidator
 
-from pulpcore.app.models import UpstreamPulp
+from pulpcore.app.models import Remote, UpstreamPulp
 from pulpcore.app.serializers import (
     HiddenFieldsMixin,
     IdentityField,
@@ -122,6 +122,16 @@ class UpstreamPulpSerializer(ModelSerializer, HiddenFieldsMixin):
         ),
         min_value=0.0,
     )
+    remote_policy = serializers.ChoiceField(
+        choices=Remote.POLICY_CHOICES,
+        help_text=_(
+            "Download policy for remotes created during replication. One of 'immediate', "
+            "'on_demand', or 'streamed'. Distinct from 'policy', which controls how replicate "
+            "manages local objects. Defaults to the Remote default ('immediate') when unset."
+        ),
+        required=False,
+        allow_null=True,
+    )
 
     pulp_last_updated = serializers.DateTimeField(
         help_text="Timestamp of the most recent update of the remote.", read_only=True
@@ -178,6 +188,7 @@ class Meta:
             "connect_timeout",
             "sock_connect_timeout",
             "sock_read_timeout",
+            "remote_policy",
             "pulp_last_updated",
             "hidden_fields",
             "q_select",

pulpcore/app/tasks/replica.py

@@ -1,6 +1,7 @@
 import os
 import platform
 import sys
+from contextlib import contextmanager
 from tempfile import NamedTemporaryFile
 
 from django.db import transaction
@@ -28,36 +29,31 @@ def user_agent():
     return f"pulpcore/{pulp_version} ({python}, {system}) (pulp-glue {pulp_glue_version})"
 
 
-def replicate_distributions(server_pk, q_select=None, **kwargs):
-    server = UpstreamPulp.objects.get(pk=server_pk)
-
-    # Write out temporary files related to SSL
+@contextmanager
+def _ssl_temp_files(server):
+    """Write UpstreamPulp TLS material to temp files that live for this context."""
     ssl_files = {}
-    for key in ["ca_cert", "client_cert", "client_key"]:
-        if value := getattr(server, key):
-            f = NamedTemporaryFile(dir=".")
-            f.write(bytes(value, "utf-8"))
-            f.flush()
-            ssl_files[key] = f.name
-
-    if "ca_cert" in ssl_files:
-        os.environ["PULP_CA_BUNDLE"] = ssl_files["ca_cert"]
-
-    ctx = ReplicaContext.from_config(
-        {
-            "base_url": server.base_url,
-            "api_root": server.api_root,
-            "domain": server.domain,
-            "username": server.username,
-            "password": server.password,
-            "cert": ssl_files.get("client_cert"),
-            "key": ssl_files.get("client_key"),
-            "user_agent": user_agent(),
-            "verify_ssl": server.tls_validation,
-            "dry_run": True,  # We only want to read from upstream anyway.
-        }
-    )
-
+    try:
+        for key in ["ca_cert", "client_cert", "client_key"]:
+            if value := getattr(server, key):
+                suffix = ".key" if key == "client_key" else ".pem"
+                with NamedTemporaryFile(
+                    dir=".", mode="w", encoding="utf-8", delete=False, suffix=suffix
+                ) as f:
+                    f.write(value)
+                    f.flush()
+                    ssl_files[key] = f.name
+        yield ssl_files
+    finally:
+        for path in ssl_files.values():
+            try:
+                os.unlink(path)
+            except FileNotFoundError:
+                pass
+
+
+def _build_remote_settings(server):
+    """Build fields copied onto remotes created during replication."""
     remote_settings = {
         "ca_cert": server.ca_cert,
         "tls_validation": server.tls_validation,
@@ -70,73 +66,102 @@ def replicate_distributions(server_pk, q_select=None, **kwargs):
         "sock_connect_timeout": server.sock_connect_timeout,
         "sock_read_timeout": server.sock_read_timeout,
     }
+    # Omit policy when unset so new remotes keep Remote.policy's default (immediate).
+    if (remote_policy := getattr(server, "remote_policy", None)) is not None:
+        remote_settings["policy"] = remote_policy
+    return remote_settings
 
-    try:
-        task_group = TaskGroup.current()
-        supported_replicators = []
-        # Load all the available replicators
-        for config in pulp_plugin_configs():
-            if config.replicator_classes:
-                for replicator_class in config.replicator_classes:
-                    req = PluginRequirement(
-                        config.label, specifier=replicator_class.required_version
-                    )
-                    if ctx.has_plugin(req):
-                        replicator = replicator_class(ctx, task_group, remote_settings, server)
-                        supported_replicators.append(replicator)
-
-        effective_q_select = q_select if q_select is not None else server.q_select
-        distro_repo_pairs = []
-        for replicator in supported_replicators:
-            distro_names = []
-            pending_distributions = []
-            distros = replicator.upstream_distributions(q=effective_q_select)
-            for distro in distros:
-                # Create remote
-                remote = replicator.create_or_update_remote(upstream_distribution=distro)
-                if not remote:
-                    # The upstream distribution is not serving any content,
-                    # let it fall through the cracks and be cleaned up below.
-                    continue
-                # Check if there is already a repository
-                repository = replicator.create_or_update_repository(remote=remote)
-                if not repository:
-                    # No update occurred because server.policy==LABELED and there was
-                    # an already existing local repository with the same name
-                    continue
-
-                # Dispatch a sync task if needed
-                if replicator.requires_syncing(distro):
-                    replicator.sync(repository, remote)
-
-                # Add name to the list of known distribution names
-                distro_names.append(distro["name"])
-                distro_repo_pairs.append((distro["name"], str(repository.pk)))
-                pending_distributions.append((repository, distro))
-
-            # Get or create distributions BEFORE remove_missing so that
-            # create_or_update_distribution can synchronously rename any existing
-            # distribution matched by base_path.  remove_missing then sees the
-            # updated name in the DB and won't schedule it for deletion.
-            for repository, distro in pending_distributions:
-                replicator.create_or_update_distribution(repository, distro)
-
-            # When a per-request q_select override is used, this is a selective sync
-            # of a subset of distributions.  Skipping remove_missing avoids deleting
-            # distributions that simply weren't included in the filter — but it also
-            # means that distributions removed from upstream won't be cleaned up until
-            # a full (non-overridden) replication runs.
-            if q_select is None:
-                replicator.remove_missing(distro_names)
-    except GluePulpException as e:
-        raise ExternalServiceError(service_name=server.base_url, details=str(e))
-
-    dispatch(
-        finalize_replication,
-        task_group=task_group,
-        exclusive_resources=[server, distros_lock_uri(server.pulp_domain_id)],
-        args=[server.pk, distro_repo_pairs],
-    )
+
+def replicate_distributions(server_pk, q_select=None, **kwargs):
+    server = UpstreamPulp.objects.get(pk=server_pk)
+    with _ssl_temp_files(server) as ssl_files:
+        verify_ssl = (
+            ssl_files["ca_cert"]
+            if server.tls_validation and "ca_cert" in ssl_files
+            else server.tls_validation
+        )
+        ctx = ReplicaContext.from_config(
+            {
+                "base_url": server.base_url,
+                "api_root": server.api_root,
+                "domain": server.domain,
+                "username": server.username,
+                "password": server.password,
+                "cert": ssl_files.get("client_cert"),
+                "key": ssl_files.get("client_key"),
+                "user_agent": user_agent(),
+                "verify_ssl": verify_ssl,
+                "dry_run": True,  # We only want to read from upstream anyway.
+            }
+        )
+
+        remote_settings = _build_remote_settings(server)
+        try:
+            task_group = TaskGroup.current()
+            supported_replicators = []
+            # Load all the available replicators
+            for config in pulp_plugin_configs():
+                if config.replicator_classes:
+                    for replicator_class in config.replicator_classes:
+                        req = PluginRequirement(
+                            config.label, specifier=replicator_class.required_version
+                        )
+                        if ctx.has_plugin(req):
+                            replicator = replicator_class(ctx, task_group, remote_settings, server)
+                            supported_replicators.append(replicator)
+
+            effective_q_select = q_select if q_select is not None else server.q_select
+            distro_repo_pairs = []
+            for replicator in supported_replicators:
+                distro_names = []
+                pending_distributions = []
+                distros = replicator.upstream_distributions(q=effective_q_select)
+                for distro in distros:
+                    # Create remote
+                    remote = replicator.create_or_update_remote(upstream_distribution=distro)
+                    if not remote:
+                        # The upstream distribution is not serving any content,
+                        # let it fall through the cracks and be cleaned up below.
+                        continue
+                    # Check if there is already a repository
+                    repository = replicator.create_or_update_repository(remote=remote)
+                    if not repository:
+                        # No update occurred because server.policy==LABELED and there was
+                        # an already existing local repository with the same name
+                        continue
+
+                    # Dispatch a sync task if needed
+                    if replicator.requires_syncing(distro):
+                        replicator.sync(repository, remote)
+
+                    # Add name to the list of known distribution names
+                    distro_names.append(distro["name"])
+                    distro_repo_pairs.append((distro["name"], str(repository.pk)))
+                    pending_distributions.append((repository, distro))
+
+                # Get or create distributions BEFORE remove_missing so that
+                # create_or_update_distribution can synchronously rename any existing
+                # distribution matched by base_path.  remove_missing then sees the
+                # updated name in the DB and won't schedule it for deletion.
+                for repository, distro in pending_distributions:
+                    replicator.create_or_update_distribution(repository, distro)
+
+                # When a per-request q_select override is used, this is a selective sync
+                # of a subset of distributions.  Skipping remove_missing avoids deleting
+                # distributions that simply weren't included in the filter — but it also
+                # means that distributions removed from upstream won't be cleaned up until
+                # a full (non-overridden) replication runs.
+                if q_select is None:
+                    replicator.remove_missing(distro_names)
+        except GluePulpException as e:
+            raise ExternalServiceError(service_name=server.base_url, details=str(e))
+
+        dispatch(
+            finalize_replication,
+            task_group=task_group,
+            exclusive_resources=[server, distros_lock_uri(server.pulp_domain_id)],
+            args=[server.pk, distro_repo_pairs],
+        )
 
 
 def finalize_replication(server_pk, distro_repo_pairs, **kwargs):