Add the ability to publish docker images non-interactively.

docs/notes/2.25.x.md

@@ -68,6 +68,14 @@ Previously we did ad-hoc coercion of some field values, so that, e.g., you could
 
 Fixed an error which was caused when the same tool appeaed in both the `--docker-tools` and `--docker-optional-tools` options.
 
+Added the ability to push docker images non-interactively. This, in turn, allows us to push docker images in parallel within one `pants publish` invocation.
+To enable this option, add the following to your `pants.toml`
+
+``` toml
+[docker]
+publish_noninteractively = true
+```
+
 #### Helm
 
 Strict adherence to the [schema of Helm OCI registry configuration](https://www.pantsbuild.org/2.25/reference/subsystems/helm#registries) is now required.

src/python/pants/backend/docker/goals/publish.py

@@ -21,7 +21,7 @@
     PublishRequest,
 )
 from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest
-from pants.engine.process import InteractiveProcess
+from pants.engine.process import InteractiveProcess, Process
 from pants.engine.rules import Get, collect_rules, rule
 
 logger = logging.getLogger(__name__)
@@ -80,7 +80,7 @@ async def push_docker_images(
     )
     jobs: list[PublishPackages] = []
     refs: list[str] = []
-    processes: list[InteractiveProcess] = []
+    processes: list[Process | InteractiveProcess] = []
 
     for tag in tags:
         for registry in options.registries().registries.values():
@@ -92,7 +92,11 @@ async def push_docker_images(
                 break
         else:
             refs.append(tag)
-            processes.append(InteractiveProcess.from_process(docker.push_image(tag, env)))
+            push_process = docker.push_image(tag, env)
+            if options.publish_noninteractively:
+                processes.append(push_process)
+            else:
+                processes.append(InteractiveProcess.from_process(push_process))
 
     for ref, process in zip(refs, processes):
         jobs.append(

src/python/pants/backend/docker/goals/publish_test.py

@@ -21,7 +21,7 @@
 from pants.core.goals.publish import PublishPackages, PublishProcesses
 from pants.engine.addresses import Address
 from pants.engine.fs import EMPTY_DIGEST
-from pants.engine.process import Process
+from pants.engine.process import InteractiveProcess, Process
 from pants.testutil.option_util import create_subsystem
 from pants.testutil.process_util import process_assertion
 from pants.testutil.rule_runner import QueryRule, RuleRunner
@@ -100,6 +100,7 @@ def assert_publish(
     assert publish.description == expect_description
     if expect_process:
         assert publish.process
+        assert isinstance(publish.process, InteractiveProcess)
         expect_process(publish.process.process)
     else:
         assert publish.process is None

src/python/pants/backend/docker/subsystems/docker_options.py

@@ -231,6 +231,15 @@ def env_vars(self) -> tuple[str, ...]:
             """
         ),
     )
+    publish_noninteractively = BoolOption(
+        default=False,
+        help=softwrap(
+            """
+            If true, publish images non-interactively. This allows for pushes to be parallelized, but requires
+            docker to be pre-authenticated to the registries to which it is pushing.
+            """
+        ),
+    )
     _tools = StrListOption(
         default=[],
         help=softwrap(

src/python/pants/backend/helm/goals/publish_test.py

@@ -20,6 +20,7 @@
 from pants.core.util_rules import external_tool
 from pants.engine.addresses import Address
 from pants.engine.fs import EMPTY_DIGEST
+from pants.engine.process import InteractiveProcess
 from pants.testutil.process_util import process_assertion
 from pants.testutil.rule_runner import QueryRule, RuleRunner
 
@@ -83,6 +84,7 @@ def assert_publish(
     assert publish.description == expect_description
     if expect_process:
         assert publish.process
+        assert isinstance(publish.process, InteractiveProcess)
         expect_process(publish.process.process)
     else:
         assert publish.process is None

src/python/pants/backend/python/goals/publish_test.py

@@ -21,7 +21,7 @@
 from pants.core.util_rules.config_files import rules as config_files_rules
 from pants.engine.addresses import Address
 from pants.engine.fs import EMPTY_DIGEST
-from pants.engine.process import Process
+from pants.engine.process import InteractiveProcess, Process
 from pants.testutil.process_util import process_assertion
 from pants.testutil.python_rule_runner import PythonRuleRunner
 from pants.testutil.rule_runner import QueryRule
@@ -109,6 +109,7 @@ def assert_package(
     assert package.description == expect_description
     if expect_process:
         assert package.process
+        assert isinstance(package.process, InteractiveProcess)
         expect_process(package.process.process)
     else:
         assert package.process is None
@@ -217,6 +218,8 @@ def test_twine_cert_arg(rule_runner, packages, options, cert_arg) -> None:
     process = result[0].process
     assert process
     if cert_arg:
+        assert isinstance(process, InteractiveProcess)
         assert cert_arg in process.process.argv
     else:
+        assert isinstance(process, InteractiveProcess)
         assert not any(arg.startswith("--cert") for arg in process.process.argv)

src/python/pants/core/goals/deploy.py

@@ -7,15 +7,20 @@
 from abc import ABCMeta
 from dataclasses import dataclass
 from itertools import chain
-from typing import Iterable
+from typing import Iterable, cast
 
 from pants.core.goals.package import PackageFieldSet
 from pants.core.goals.publish import PublishFieldSet, PublishProcesses, PublishProcessesRequest
 from pants.engine.console import Console
-from pants.engine.environment import EnvironmentName
+from pants.engine.environment import ChosenLocalEnvironmentName, EnvironmentName
 from pants.engine.goal import Goal, GoalSubsystem
 from pants.engine.intrinsics import run_interactive_process
-from pants.engine.process import InteractiveProcess
+from pants.engine.process import (
+    FallibleProcessResult,
+    InteractiveProcess,
+    InteractiveProcessResult,
+    Process,
+)
 from pants.engine.rules import Get, MultiGet, collect_rules, goal_rule, rule
 from pants.engine.target import (
     FieldSet,
@@ -144,27 +149,15 @@ async def _all_publish_processes(targets: Iterable[Target]) -> PublishProcesses:
     return PublishProcesses(chain.from_iterable(processes_per_target))
 
 
-async def _invoke_process(
+def _process_results_to_string(
     console: Console,
-    process: InteractiveProcess | None,
+    res: InteractiveProcessResult | FallibleProcessResult,
     *,
     names: Iterable[str],
     success_status: str,
     description: str | None = None,
 ) -> tuple[int, tuple[str, ...]]:
     results = []
-
-    if not process:
-        sigil = console.sigil_skipped()
-        status = "skipped"
-        if description:
-            status += f" {description}"
-        for name in names:
-            results.append(f"{sigil} {name} {status}.")
-        return 0, tuple(results)
-
-    logger.debug(f"Execute {process}")
-    res = await run_interactive_process(process)
     if res.exit_code == 0:
         sigil = console.sigil_succeeded()
         status = success_status
@@ -179,12 +172,41 @@ async def _invoke_process(
 
     for name in names:
         results.append(f"{sigil} {name} {status}")
-
     return res.exit_code, tuple(results)
 
 
+async def _invoke_process(
+    console: Console,
+    process: InteractiveProcess | None,
+    *,
+    names: Iterable[str],
+    success_status: str,
+    description: str | None = None,
+) -> tuple[int, tuple[str, ...]]:
+    results = []
+
+    if not process:
+        sigil = console.sigil_skipped()
+        status = "skipped"
+        if description:
+            status += f" {description}"
+        for name in names:
+            results.append(f"{sigil} {name} {status}.")
+        return 0, tuple(results)
+
+    logger.debug(f"Execute {process}")
+    res = await run_interactive_process(process)
+    return _process_results_to_string(
+        console, res, names=names, success_status=success_status, description=description
+    )
+
+
 @goal_rule
-async def run_deploy(console: Console, deploy_subsystem: DeploySubsystem) -> Deploy:
+async def run_deploy(
+    console: Console,
+    deploy_subsystem: DeploySubsystem,
+    local_environment: ChosenLocalEnvironmentName,
+) -> Deploy:
     target_roots_to_deploy_field_sets = await Get(
         TargetRootsToFieldSets,
         TargetRootsToFieldSetsRequest(
@@ -213,12 +235,40 @@ async def run_deploy(console: Console, deploy_subsystem: DeploySubsystem) -> Dep
 
     if publish_processes:
         logger.info(f"Publishing {pluralize(len(publish_processes), 'dependency')}...")
+        background_publish_processes = [
+            publish for publish in publish_processes if isinstance(publish.process, Process)
+        ]
+        foreground_publish_processes = [
+            publish
+            for publish in publish_processes
+            if isinstance(publish.process, InteractiveProcess) or publish.process is None
+        ]
+
+        # Publish all background deployments first
+        background_results = await MultiGet(
+            Get(
+                FallibleProcessResult,
+                {cast(Process, publish.process): Process, local_environment.val: EnvironmentName},
+            )
+            for publish in background_publish_processes
+        )
+        for pub, res in zip(background_publish_processes, background_results):
+            ec, statuses = _process_results_to_string(
+                console,
+                res,
+                names=pub.names,
+                description=pub.description,
+                success_status="published",
+            )
+            exit_code = ec if ec != 0 else exit_code
+            results.extend(statuses)
 
-        # Publish all deployment dependencies first.
-        for publish in publish_processes:
+        # Publish all foreground deployments next.
+        for publish in foreground_publish_processes:
+            process = cast(InteractiveProcess | None, publish.process)
             ec, statuses = await _invoke_process(
                 console,
-                publish.process,
+                process,
                 names=publish.names,
                 description=publish.description,
                 success_status="published",