Resolve plugins via PEX rules in a bootstrap scheduler (#11568)

### Problem

#11536 moves from using POSIX-level replacement of the `stdio` file descriptors `{0, 1, 2}`, to replacing `sys.std*` with a thread-local implementation. Unfortunately, Python `subprocess`/`Popen` APIs hardcode `{0, 1, 2}` rather than actually inspecting `sys.std*.fileno()`, and so usages of those APIs that use `std*=None` (i.e. "inherit" mode), will inherit intentionally dead/closed file handles under `pantsd`.

PEX uses inherit mode when spawning `pip` (in order to pass through `stderr` to the parent process), and this runs afoul of the above behavior. Pants has used PEX as a library for a very long time, but 95% of usecases have now migrated to using PEX as a binary, with wrappers exposed via the `@rule` API. The `PluginResolver` that Pants uses to resolve its own code is the last usage of PEX as a library. 

### Solution

In a series of commits, introduce a "bootstrap" `Scheduler` that is able to resolve plugins after an `OptionsBootstrapper` has been created, but before creating the `BuildConfiguration` (which contains plugin information).

### Result

Although Pants has some stray references to PEX APIs, it no longer uses PEX as a library to resolve dependencies. #11536 and #7654 are unblocked.

In future, if the options required to construct a minimal scheduler can be further pruned, the bootstrap `Scheduler` might also be able to be used to create the `OptionsBootstrapper`, which would allow for addressing #10360.

src/python/pants/backend/python/util_rules/pex.py

@@ -321,6 +321,7 @@ class PexRequest(EngineAwareParameter):
     entry_point: str | None
     additional_args: Tuple[str, ...]
     pex_path: Tuple[Pex, ...]
+    apply_requirement_constraints: bool
     description: str | None = dataclasses.field(compare=False)
 
     def __init__(
@@ -336,6 +337,7 @@ def __init__(
         entry_point: str | None = None,
         additional_args: Iterable[str] = (),
         pex_path: Iterable[Pex] = (),
+        apply_requirement_constraints: bool = True,
         description: str | None = None,
     ) -> None:
         """A request to create a PEX from its inputs.
@@ -358,6 +360,8 @@ def __init__(
             left off, the Pex will open up as a REPL.
         :param additional_args: Any additional Pex flags.
         :param pex_path: Pex files to add to the PEX_PATH.
+        :param apply_requirement_constraints: Whether to apply any configured
+            requirement_constraints while building this PEX.
         :param description: A human-readable description to render in the dynamic UI when building
             the Pex.
         """
@@ -371,6 +375,7 @@ def __init__(
         self.entry_point = entry_point
         self.additional_args = tuple(additional_args)
         self.pex_path = tuple(pex_path)
+        self.apply_requirement_constraints = apply_requirement_constraints
         self.description = description
         self.__post_init__()
 
@@ -550,16 +555,14 @@ async def build_pex(
     if request.entry_point is not None:
         argv.extend(["--entry-point", request.entry_point])
 
-    if python_setup.requirement_constraints is not None:
-        argv.extend(["--constraints", python_setup.requirement_constraints])
-
     source_dir_name = "source_files"
     argv.append(f"--sources-directory={source_dir_name}")
 
     argv.extend(request.requirements)
 
     constraint_file_digest = EMPTY_DIGEST
-    if python_setup.requirement_constraints is not None:
+    if request.apply_requirement_constraints and python_setup.requirement_constraints is not None:
+        argv.extend(["--constraints", python_setup.requirement_constraints])
         constraint_file_digest = await Get(
             Digest,
             PathGlobs(

src/python/pants/bin/daemon_pants_runner.py

@@ -19,7 +19,6 @@
     set_logging_handlers,
     setup_logging,
 )
-from pants.init.options_initializer import BuildConfigInitializer
 from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.pantsd.pants_daemon_core import PantsDaemonCore
 from pants.util.contextutil import argv_as, hermetic_environment_as, stdio_as
@@ -144,7 +143,6 @@ def single_daemonized_run(
         # of a local run: once we allow for concurrent runs, this information should be
         # propagated down from the caller.
         #   see https://github.com/pantsbuild/pants/issues/7654
-        BuildConfigInitializer.reset()
         options_bootstrapper = OptionsBootstrapper.create(
             env=os.environ, args=sys.argv, allow_pantsrc=True
         )
@@ -153,11 +151,12 @@ def single_daemonized_run(
         # Run using the pre-warmed Session.
         with self._stderr_logging(global_bootstrap_options):
             try:
-                scheduler = self._core.prepare_scheduler(options_bootstrapper)
+                scheduler, options_initializer = self._core.prepare(options_bootstrapper)
                 runner = LocalPantsRunner.create(
                     os.environ,
                     options_bootstrapper,
                     scheduler=scheduler,
+                    options_initializer=options_initializer,
                     cancellation_latch=cancellation_latch,
                 )
                 return runner.run(start_time)

src/python/pants/bin/local_pants_runner.py

@@ -64,6 +64,7 @@ class LocalPantsRunner:
     @classmethod
     def _init_graph_session(
         cls,
+        options_initializer: OptionsInitializer,
         options_bootstrapper: OptionsBootstrapper,
         build_config: BuildConfiguration,
         run_id: str,
@@ -73,8 +74,10 @@ def _init_graph_session(
     ) -> GraphSession:
         native = Native()
         native.set_panic_handler()
-        graph_scheduler_helper = scheduler or EngineInitializer.setup_graph(options, build_config)
-        with OptionsInitializer.handle_unknown_flags(options_bootstrapper, raise_=True):
+        graph_scheduler_helper = scheduler or EngineInitializer.setup_graph(
+            options_bootstrapper, build_config
+        )
+        with options_initializer.handle_unknown_flags(options_bootstrapper, raise_=True):
             global_options = options.for_global_scope()
         return graph_scheduler_helper.new_session(
             run_id,
@@ -94,6 +97,7 @@ def create(
         cls,
         env: Mapping[str, str],
         options_bootstrapper: OptionsBootstrapper,
+        options_initializer: Optional[OptionsInitializer] = None,
         scheduler: Optional[GraphScheduler] = None,
         cancellation_latch: Optional[PySessionCancellationLatch] = None,
     ) -> LocalPantsRunner:
@@ -106,7 +110,8 @@ def create(
         :param options_bootstrapper: The OptionsBootstrapper instance to reuse.
         :param scheduler: If being called from the daemon, a warmed scheduler to use.
         """
-        build_config, options = OptionsInitializer.create_with_build_config(
+        options_initializer = options_initializer or OptionsInitializer(options_bootstrapper)
+        build_config, options = options_initializer.build_config_and_options(
             options_bootstrapper, raise_=True
         )
 
@@ -116,6 +121,7 @@ def create(
         # If we're running with the daemon, we'll be handed a warmed Scheduler, which we use
         # to initialize a session here.
         graph_session = cls._init_graph_session(
+            options_initializer,
             options_bootstrapper,
             build_config,
             run_tracker.run_id,
@@ -126,7 +132,7 @@ def create(
 
         # Option values are usually computed lazily on demand, but command line options are
         # eagerly computed for validation.
-        with OptionsInitializer.handle_unknown_flags(options_bootstrapper, raise_=True):
+        with options_initializer.handle_unknown_flags(options_bootstrapper, raise_=True):
             for scope in options.scope_to_flags.keys():
                 options.for_scope(scope)
 

src/python/pants/bin/remote_pants_runner.py

@@ -12,8 +12,8 @@
 from pants.base.exiter import ExitCode
 from pants.engine.internals.native import Native
 from pants.engine.internals.native_engine import PyExecutor
-from pants.init.options_initializer import OptionsInitializer
 from pants.nailgun.nailgun_protocol import NailgunProtocol
+from pants.option.global_options import GlobalOptions
 from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.pantsd.pants_daemon_client import PantsDaemonClient
 
@@ -103,7 +103,7 @@ def _connect_and_execute(self, pantsd_handle: PantsDaemonClient.Handle) -> ExitC
         native = Native()
 
         global_options = self._bootstrap_options.for_global_scope()
-        executor = PyExecutor(*OptionsInitializer.compute_executor_arguments(global_options))
+        executor = PyExecutor(*GlobalOptions.compute_executor_arguments(global_options))
 
         # Merge the nailgun TTY capability environment variables with the passed environment dict.
         ng_env = NailgunProtocol.ttynames_to_env(sys.stdin, sys.stdout.buffer, sys.stderr.buffer)

src/python/pants/build_graph/build_configuration.py

@@ -43,6 +43,7 @@ class BuildConfiguration:
     rules: FrozenOrderedSet[Rule]
     union_rules: FrozenOrderedSet[UnionRule]
     target_types: FrozenOrderedSet[Type[Target]]
+    allow_unknown_options: bool
 
     @property
     def all_optionables(self) -> FrozenOrderedSet[Type[Optionable]]:
@@ -93,6 +94,7 @@ class Builder:
         _rules: OrderedSet = field(default_factory=OrderedSet)
         _union_rules: OrderedSet = field(default_factory=OrderedSet)
         _target_types: OrderedSet[Type[Target]] = field(default_factory=OrderedSet)
+        _allow_unknown_options: bool = False
 
         def registered_aliases(self) -> BuildFileAliases:
             """Return the registered aliases exposed in BUILD files.
@@ -217,6 +219,14 @@ def register_target_types(self, target_types: typing.Iterable[Type[Target]] | An
                 )
             self._target_types.update(target_types)
 
+        def allow_unknown_options(self, allow: bool = True) -> None:
+            """Allows overriding whether Options parsing will fail for unrecognized Options.
+
+            Used to defer options failures while bootstrapping BuildConfiguration until after the
+            complete set of plugins is known.
+            """
+            self._allow_unknown_options = True
+
         def create(self) -> BuildConfiguration:
             registered_aliases = BuildFileAliases(
                 objects=self._exposed_object_by_alias.copy(),
@@ -228,4 +238,5 @@ def create(self) -> BuildConfiguration:
                 rules=FrozenOrderedSet(self._rules),
                 union_rules=FrozenOrderedSet(self._union_rules),
                 target_types=FrozenOrderedSet(self._target_types),
+                allow_unknown_options=self._allow_unknown_options,
             )

src/python/pants/engine/internals/engine_test.py

@@ -284,7 +284,7 @@ def add(self, **kwargs) -> None:
 
 
 def new_run_tracker() -> RunTracker:
-    # NB: A RunTracker usually observes "all options" (`get_full_options`), but it only actually
+    # NB: A RunTracker usually observes "all options" (`full_options_for_scopes`), but it only actually
     # directly consumes bootstrap options.
     return RunTracker(create_options_bootstrapper([]).bootstrap_options)
 

src/python/pants/engine/internals/options_parsing.py

@@ -6,7 +6,6 @@
 from pants.build_graph.build_configuration import BuildConfiguration
 from pants.engine.internals.session import SessionValues
 from pants.engine.rules import collect_rules, rule
-from pants.init.options_initializer import OptionsInitializer
 from pants.option.global_options import GlobalOptions
 from pants.option.options import Options
 from pants.option.options_bootstrapper import OptionsBootstrapper
@@ -33,7 +32,7 @@ def __post_init__(self) -> None:
 
     @memoized_property
     def options(self) -> Options:
-        return OptionsInitializer.create(self.options_bootstrapper, self.build_config)
+        return self.options_bootstrapper.full_options(self.build_config)
 
 
 @rule

src/python/pants/engine/internals/scheduler_integration_test.py

@@ -1,6 +1,7 @@
 # Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
+import os
 from pathlib import Path
 
 from pants.testutil.pants_integration_test import ensure_daemon, run_pants
@@ -11,7 +12,7 @@ def test_visualize_to():
     # Tests usage of the `--native-engine-visualize-to=` option, which triggers background
     # visualization of the graph. There are unit tests confirming the content of the rendered
     # results.
-    with temporary_dir() as destdir:
+    with temporary_dir(root_dir=os.getcwd()) as destdir:
         run_pants(
             [
                 f"--native-engine-visualize-to={destdir}",

src/python/pants/engine/process.py

@@ -108,6 +108,8 @@ def __init__(
             result = await Get(ProcessResult, Process(["/bin/echo", "hello world"], description="demo"))
             assert result.stdout == b"hello world"
         """
+        if isinstance(argv, str):
+            raise ValueError("argv must be a sequence of strings, but was a single string.")
         self.argv = tuple(argv)
         self.description = description
         self.level = level

src/python/pants/help/help_formatter_test.py

@@ -65,7 +65,7 @@ def _format_for_global_scope(show_advanced, show_deprecated, args, kwargs):
         )
         parser.register(*args, **kwargs)
         # Force a parse to generate the derivation history.
-        parser.parse_args(Parser.ParseArgsRequest((), OptionValueContainerBuilder(), []))
+        parser.parse_args(Parser.ParseArgsRequest((), OptionValueContainerBuilder(), [], False))
         oshi = HelpInfoExtracter("").get_option_scope_help_info("", parser, False)
         return HelpFormatter(
             show_advanced=show_advanced, show_deprecated=show_deprecated, color=False

src/python/pants/init/bootstrap_scheduler.py

@@ -0,0 +1,13 @@
+# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from dataclasses import dataclass
+
+from pants.engine.internals.scheduler import Scheduler
+
+
+@dataclass(frozen=True)
+class BootstrapScheduler:
+    """A Scheduler that has been configured with only the rules for bootstrapping."""
+
+    scheduler: Scheduler

src/python/pants/init/engine_initializer.py

@@ -30,9 +30,8 @@
 from pants.engine.target import RegisteredTargetTypes
 from pants.engine.unions import UnionMembership
 from pants.init import specs_calculator
-from pants.init.options_initializer import OptionsInitializer
-from pants.option.global_options import DEFAULT_EXECUTION_OPTIONS, ExecutionOptions
-from pants.option.options import Options
+from pants.option.global_options import DEFAULT_EXECUTION_OPTIONS, ExecutionOptions, GlobalOptions
+from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.option.subsystem import Subsystem
 from pants.util.ordered_set import FrozenOrderedSet
 from pants.vcs.changed import rules as changed_rules
@@ -164,25 +163,26 @@ def _make_goal_map_from_rules(rules):
 
     @staticmethod
     def setup_graph(
-        options: Options,
+        options_bootstrapper: OptionsBootstrapper,
         build_configuration: BuildConfiguration,
         executor: Optional[PyExecutor] = None,
+        execution_options: Optional[ExecutionOptions] = None,
     ) -> GraphScheduler:
         native = Native()
         build_root = get_buildroot()
-        bootstrap_options = options.bootstrap_option_values()
+        bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()
+        options = options_bootstrapper.full_options(build_configuration)
         assert bootstrap_options is not None
         executor = executor or PyExecutor(
-            *OptionsInitializer.compute_executor_arguments(bootstrap_options)
+            *GlobalOptions.compute_executor_arguments(bootstrap_options)
         )
+        execution_options = execution_options or ExecutionOptions.from_options(options)
         return EngineInitializer.setup_graph_extended(
             build_configuration,
-            ExecutionOptions.from_options(options),
+            execution_options,
             native=native,
             executor=executor,
-            pants_ignore_patterns=OptionsInitializer.compute_pants_ignore(
-                build_root, bootstrap_options
-            ),
+            pants_ignore_patterns=GlobalOptions.compute_pants_ignore(build_root, bootstrap_options),
             use_gitignore=bootstrap_options.pants_ignore_use_gitignore,
             local_store_dir=bootstrap_options.local_store_dir,
             local_execution_root_dir=bootstrap_options.local_execution_root_dir,