Skip capture of TRACE workunits by default

src/python/pants/backend/python/target_types_rules.py

@@ -62,7 +62,6 @@
 from pants.source.source_root import SourceRoot, SourceRootRequest
 from pants.util.docutil import doc_url
 from pants.util.frozendict import FrozenDict
-from pants.util.logging import LogLevel
 from pants.util.ordered_set import OrderedSet
 
 logger = logging.getLogger(__name__)
@@ -136,7 +135,7 @@ def create_pex_binary(entry_point_spec: str) -> PexBinary:
 # -----------------------------------------------------------------------------------------------
 
 
-@rule(desc="Determining the entry point for a `pex_binary` target", level=LogLevel.DEBUG)
+@rule(desc="Determining the entry point for a `pex_binary` target")
 async def resolve_pex_entry_point(request: ResolvePexEntryPointRequest) -> ResolvedPexEntryPoint:
     ep_val = request.entry_point_field.value
     if ep_val is None:
@@ -266,7 +265,7 @@ def _classify_entry_points(
             )
 
 
-@rule(desc="Determining the entry points for a `python_distribution` target", level=LogLevel.DEBUG)
+@rule(desc="Determining the entry points for a `python_distribution` target")
 async def resolve_python_distribution_entry_points(
     request: ResolvePythonDistributionEntryPointsRequest,
 ) -> ResolvedPythonDistributionEntryPoints:

src/python/pants/bin/local_pants_runner.py

@@ -34,6 +34,7 @@
 from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.util.collections import deep_getsizeof
 from pants.util.contextutil import maybe_profiled
+from pants.util.logging import LogLevel
 
 logger = logging.getLogger(__name__)
 
@@ -87,6 +88,14 @@ def _init_graph_session(
             ui_use_prodash=global_options.dynamic_ui_renderer
             == DynamicUIRenderer.experimental_prodash,
             use_colors=global_options.get("colors", True),
+            max_workunit_level=max(
+                global_options.streaming_workunits_level,
+                global_options.level,
+                *(
+                    LogLevel[level.upper()]
+                    for level in global_options.log_levels_by_target.values()
+                ),
+            ),
             session_values=SessionValues(
                 {
                     OptionsBootstrapper: options_bootstrapper,
@@ -281,6 +290,7 @@ def run(self, start_time: float) -> ExitCode:
                 allow_async_completion=(
                     global_options.pantsd and global_options.streaming_workunits_complete_async
                 ),
+                max_workunit_verbosity=global_options.streaming_workunits_level,
             )
             with streaming_reporter:
                 engine_result = PANTS_FAILED_EXIT_CODE

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

@@ -303,7 +303,12 @@ class TestStreamingWorkunit(SchedulerTestBase):
     def _fixture_for_rules(
         self, tmp_path: Path, rules, max_workunit_verbosity: LogLevel = LogLevel.INFO
     ) -> Tuple[SchedulerSession, WorkunitTracker, StreamingWorkunitHandler]:
-        scheduler = self.mk_scheduler(tmp_path, rules, include_trace_on_error=False)
+        scheduler = self.mk_scheduler(
+            tmp_path,
+            rules,
+            include_trace_on_error=False,
+            max_workunit_verbosity=max_workunit_verbosity,
+        )
         tracker = WorkunitTracker()
         handler = StreamingWorkunitHandler(
             scheduler,
@@ -648,7 +653,7 @@ def artifacts(self):
         return {"snapshot_1": self.snapshot_1, "snapshot_2": self.snapshot_2}
 
 
-@rule(desc="a_rule")
+@rule(desc="a_rule", level=LogLevel.DEBUG)
 def a_rule(input: ComplicatedInput) -> Output:
     return Output(snapshot_1=input.snapshot_1, snapshot_2=input.snapshot_2)
 
@@ -662,6 +667,10 @@ def rule_runner() -> RuleRunner:
             QueryRule(ProcessResult, (Process,)),
         ],
         isolated_local_store=True,
+        # NB: The Sessions's configured verbosity is applied before a `StreamingWorkunitHandler`
+        # can poll, and prevents things from being stored at all. So in order to observe TRACE
+        # workunits in a poll, we must also configure TRACE on the Session.
+        max_workunit_verbosity=LogLevel.TRACE,
     )
 
 

src/python/pants/engine/internals/native_engine.pyi

@@ -344,6 +344,7 @@ class PySession:
         scheduler: PyScheduler,
         dynamic_ui: bool,
         ui_use_prodash: bool,
+        max_workunit_level: int,
         build_id: str,
         session_values: SessionValues,
         cancellation_latch: PySessionCancellationLatch,

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

@@ -308,6 +308,7 @@ def new_session(
         build_id: str,
         dynamic_ui: bool = False,
         ui_use_prodash: bool = False,
+        max_workunit_level: LogLevel = LogLevel.DEBUG,
         session_values: SessionValues | None = None,
         cancellation_latch: PySessionCancellationLatch | None = None,
     ) -> SchedulerSession:
@@ -318,6 +319,7 @@ def new_session(
                 scheduler=self.py_scheduler,
                 dynamic_ui=dynamic_ui,
                 ui_use_prodash=ui_use_prodash,
+                max_workunit_level=max_workunit_level.level,
                 build_id=build_id,
                 session_values=session_values or SessionValues(),
                 cancellation_latch=cancellation_latch or PySessionCancellationLatch(),

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

@@ -10,12 +10,13 @@
 from pants.option.global_options import DEFAULT_EXECUTION_OPTIONS, DEFAULT_LOCAL_STORE_OPTIONS
 from pants.util.contextutil import temporary_file_path
 from pants.util.dirutil import safe_mkdtemp
+from pants.util.logging import LogLevel
 
 
 class SchedulerTestBase:
-    """A mixin for classes (tests, presumably) which need to create temporary schedulers.
+    """A mixin for classes which need to create temporary schedulers.
 
-    TODO: In the medium term, this should be part of pants_test.test_base.TestBase.
+    TODO: In the medium term, this should be removed in favor of RuleRunner.
     """
 
     _executor = PyExecutor(core_threads=2, max_threads=4)
@@ -25,6 +26,7 @@ def mk_scheduler(
         tmp_path: Path,
         rules,
         include_trace_on_error: bool = True,
+        max_workunit_verbosity: LogLevel = LogLevel.DEBUG,
     ) -> SchedulerSession:
         """Creates a SchedulerSession for a Scheduler with the given Rules installed."""
 
@@ -49,6 +51,7 @@ def mk_scheduler(
         )
         return scheduler.new_session(
             build_id="buildid_for_test",
+            max_workunit_level=max_workunit_verbosity,
         )
 
     def execute(self, scheduler, product, *subjects):

src/python/pants/engine/streaming_workunit_handler.py

@@ -186,7 +186,7 @@ def __init__(
         specs: Specs,
         report_interval_seconds: float,
         allow_async_completion: bool,
-        max_workunit_verbosity: LogLevel = LogLevel.TRACE,
+        max_workunit_verbosity: LogLevel,
     ) -> None:
         scheduler = scheduler.isolated_shallow_clone("streaming_workunit_handler_session")
         self.callbacks = callbacks

src/python/pants/init/engine_initializer.py

@@ -39,6 +39,7 @@
 )
 from pants.option.option_value_container import OptionValueContainer
 from pants.option.subsystem import Subsystem
+from pants.util.logging import LogLevel
 from pants.util.ordered_set import FrozenOrderedSet
 from pants.vcs.changed import rules as changed_rules
 
@@ -58,13 +59,15 @@ def new_session(
         dynamic_ui: bool = False,
         ui_use_prodash: bool = False,
         use_colors=True,
+        max_workunit_level: LogLevel = LogLevel.DEBUG,
         session_values: SessionValues | None = None,
         cancellation_latch: PySessionCancellationLatch | None = None,
     ) -> GraphSession:
         session = self.scheduler.new_session(
             build_id,
             dynamic_ui,
             ui_use_prodash,
+            max_workunit_level=max_workunit_level,
             session_values=session_values,
             cancellation_latch=cancellation_latch,
         )

src/python/pants/option/global_options.py

@@ -1440,6 +1440,18 @@ def register_options(cls, register):
         help="Interval in seconds between when streaming workunit event receivers will be polled.",
         advanced=True,
     )
+    streaming_workunits_level = EnumOption(
+        "--streaming-workunits-level",
+        default=LogLevel.DEBUG,
+        help=(
+            "The level of workunits that will be reported to streaming workunit event "
+            "receivers.\n\n"
+            "Workunits form a tree, and even when workunits are filtered out by this setting, the "
+            "workunit tree structure will be preserved (by adjusting the parent pointers of the "
+            "remaining workunits)."
+        ),
+        advanced=True,
+    )
     streaming_workunits_complete_async = BoolOption(
         "--streaming-workunits-complete-async",
         default=not is_in_container(),

src/python/pants/testutil/rule_runner.py

@@ -190,6 +190,7 @@ def __init__(
         bootstrap_args: Iterable[str] = (),
         use_deprecated_python_macros: bool = False,
         extra_session_values: dict[Any, Any] | None = None,
+        max_workunit_verbosity: LogLevel = LogLevel.DEBUG,
     ) -> None:
 
         bootstrap_args = [*bootstrap_args]
@@ -272,6 +273,7 @@ def __init__(
                     **(extra_session_values or {}),
                 }
             ),
+            max_workunit_level=max_workunit_verbosity,
         )
         self.scheduler = graph_session.scheduler_session
 

src/python/pants/util/logging.py

@@ -5,12 +5,21 @@
 
 import logging
 from enum import Enum
+from functools import total_ordering
 
 # A custom log level for pants trace logging.
 TRACE = 5
 
 
+@total_ordering
 class LogLevel(Enum):
+    """Exposes an enum of the Python `logging` module's levels, with the addition of TRACE.
+
+    NB: The `logging` module uses the opposite integer ordering of levels from Rust's `log` crate,
+    but the ordering implementation of this enum inverts its comparison to make its ordering align
+    with Rust's.
+    """
+
     TRACE = ("trace", TRACE)
     DEBUG = ("debug", logging.DEBUG)
     INFO = ("info", logging.INFO)
@@ -34,3 +43,8 @@ def log(self, logger: logging.Logger, *args, **kwargs) -> None:
 
     def set_level_for(self, logger: logging.Logger):
         logger.setLevel(self.level)
+
+    def __lt__(self, other):
+        if not isinstance(other, LogLevel):
+            return NotImplemented
+        return self._level > other._level

src/rust/engine/process_execution/src/lib.rs

@@ -648,7 +648,7 @@ pub struct Context {
 impl Default for Context {
   fn default() -> Self {
     Context {
-      workunit_store: WorkunitStore::new(false),
+      workunit_store: WorkunitStore::new(false, log::Level::Debug),
       build_id: String::default(),
       run_id: RunId(0),
     }

src/rust/engine/process_execution/src/local.rs

@@ -651,7 +651,7 @@ pub async fn prepare_workdir(
     context.workunit_store.clone(),
     "setup_sandbox".to_owned(),
     WorkunitMetadata {
-      level: Level::Trace,
+      level: Level::Debug,
       ..WorkunitMetadata::default()
     },
     |_workunit| async move {

src/rust/engine/process_execution/src/remote_tests.rs

@@ -999,7 +999,7 @@ async fn sends_headers() {
   )
   .unwrap();
   let context = Context {
-    workunit_store: WorkunitStore::new(false),
+    workunit_store: WorkunitStore::new(false, log::Level::Debug),
     build_id: String::from("marmosets"),
     run_id: RunId(0),
   };

src/rust/engine/process_executor/src/main.rs

@@ -210,7 +210,7 @@ struct Opt {
 #[tokio::main]
 async fn main() {
   env_logger::init();
-  let workunit_store = WorkunitStore::new(false);
+  let workunit_store = WorkunitStore::new(false, log::Level::Debug);
   workunit_store.init_thread_state(None);
 
   let args = Opt::from_args();

src/rust/engine/src/externs/interface.rs

@@ -354,13 +354,17 @@ impl PySession {
     scheduler: &PyScheduler,
     dynamic_ui: bool,
     ui_use_prodash: bool,
+    max_workunit_level: u64,
     build_id: String,
     session_values: PyObject,
     cancellation_latch: &PySessionCancellationLatch,
     py: Python,
   ) -> PyO3Result<Self> {
     let core = scheduler.0.core.clone();
     let cancellation_latch = cancellation_latch.0.clone();
+    let py_level: PythonLogLevel = max_workunit_level
+      .try_into()
+      .map_err(|e| PyException::new_err(format!("{e}")))?;
     // NB: Session creation interacts with the Graph, which must not be accessed while the GIL is
     // held.
     let session = py
@@ -369,6 +373,7 @@ impl PySession {
           core,
           dynamic_ui,
           ui_use_prodash,
+          py_level.into(),
           build_id,
           session_values,
           cancellation_latch,

src/rust/engine/src/session.rs

@@ -145,11 +145,20 @@ impl Session {
     core: Arc<Core>,
     dynamic_ui: bool,
     ui_use_prodash: bool,
+    mut max_workunit_level: log::Level,
     build_id: String,
     session_values: PyObject,
     cancelled: AsyncLatch,
   ) -> Result<Session, String> {
-    let workunit_store = WorkunitStore::new(!dynamic_ui);
+    // We record workunits with the maximum level of:
+    // 1. the given `max_workunit_verbosity`, which should be computed from:
+    //     * the log level, to ensure that workunit events are logged
+    //     * the levels required by any consumers who will call `with_latest_workunits`.
+    // 2. the level required by the ConsoleUI (if any): currently, DEBUG.
+    if dynamic_ui {
+      max_workunit_level = std::cmp::max(max_workunit_level, log::Level::Debug);
+    }
+    let workunit_store = WorkunitStore::new(!dynamic_ui, max_workunit_level);
     let display = tokio::sync::Mutex::new(SessionDisplay::new(
       &workunit_store,
       core.local_parallelism,