19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

34.51

/src/python/pants/goal/run_tracker.py

# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import logging
import platform
import socket
import time
import uuid
from hashlib import sha256
from pathlib import Path
from typing import Any

from pants.base.build_environment import get_buildroot
from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE, ExitCode
from pants.engine.internals import native_engine
from pants.option.errors import ConfigValidationError
from pants.option.options import Options
from pants.option.scope import GLOBAL_SCOPE, GLOBAL_SCOPE_CONFIG_SECTION
from pants.util.osutil import getuser
from pants.version import VERSION

logger = logging.getLogger(__name__)


class RunTracker:
    """Tracks and times the execution of a single Pants run."""

    # TODO: Find a way to know from a goal name whether it's a standard or a custom
    #  goal whose name could, in theory, reveal something proprietary. That's more work than
    #  we want to do at the moment, so we maintain this manual list for now.
    STANDARD_GOALS = frozenset(
        (
            "check",
            "count-loc",
            "dependents",
            "dependencies",
            "export-codegen",
            "filedeps",
            "fmt",
            "lint",
            "list",
            "package",
            "py-constraints",
            "repl",
            "roots",
            "run",
            "tailor",
            "test",
            "typecheck",
            "validate",
        )
    )

    def __init__(self, args: tuple[str, ...], options: Options):
        """
        :API: public
        """
        self._has_started: bool = False
        self._has_ended: bool = False

        # Select a globally unique ID for the run, that sorts by time.
        run_timestamp = time.time()
        run_uuid = uuid.uuid4().hex
        str_time = time.strftime("%Y_%m_%d_%H_%M_%S", time.localtime(run_timestamp))
        millis = int((run_timestamp * 1000) % 1000)
        self.run_id = f"pants_run_{str_time}_{millis}_{run_uuid}"

        self._args = args
        self._all_options = options
        info_dir = Path(self._all_options.for_global_scope().pants_workdir) / "run-tracker"
        self._run_info: dict[str, Any] = {}

        # pantsd stats.
        self._pantsd_metrics: dict[str, int] = dict()

        self.run_logs_file = info_dir / self.run_id / "logs"
        self.run_logs_file.parent.mkdir(exist_ok=True, parents=True)
        native_engine.set_per_run_log_path(str(self.run_logs_file))

        # Initialized in `start()`.
        self._run_start_time: float | None = None
        self._run_total_duration: float | None = None

    @property
    def goals(self) -> list[str]:
        return self._all_options.goals if self._all_options else []

    @property
    def active_standard_backends(self) -> list[str]:
        return [
            backend
            for backend in self._all_options.for_global_scope().backend_packages
            if backend.startswith("pants.backend.")
        ]

    def start(self, run_start_time: float, specs: list[str]) -> None:
        """Start tracking this pants run."""
        if self._has_started:
            raise AssertionError("RunTracker.start must not be called multiple times.")
        self._has_started = True

        # Initialize the run.
        self._run_start_time = run_start_time

        datetime = time.strftime("%A %b %d, %Y %H:%M:%S", time.localtime(run_start_time))
        cmd_line = " ".join(("pants",) + self._args[1:])

        self._run_info.update(
            {
                "id": self.run_id,
                "timestamp": run_start_time,
                "datetime": datetime,
                "user": getuser(),
                "machine": socket.gethostname(),
                "buildroot": get_buildroot(),
                "path": get_buildroot(),
                "version": VERSION,
                "cmd_line": cmd_line,
                "specs_from_command_line": specs,
            }
        )

    def get_anonymous_telemetry_data(self, unhashed_repo_id: str) -> dict[str, str | list[str]]:
        def maybe_hash_with_repo_id_prefix(s: str) -> str:
            qualified_str = f"{unhashed_repo_id}.{s}" if s else unhashed_repo_id
            # If the repo_id is the empty string we return a blank string.
            return sha256(qualified_str.encode()).hexdigest() if unhashed_repo_id else ""

        return {
            "run_id": str(self._run_info.get("id", uuid.uuid4())),
            "timestamp": str(self._run_info.get("timestamp")),
            # Note that this method is called after the StreamingWorkunitHandler.session() ends,
            # i.e., after end_run() has been called, so duration will be set.
            "duration": str(self._run_total_duration),
            "outcome": str(self._run_info.get("outcome")),
            "platform": platform.platform(),
            "python_implementation": platform.python_implementation(),
            "python_version": platform.python_version(),
            "pants_version": str(self._run_info.get("version")),
            # Note that if repo_id is the empty string then these three fields will be empty.
            "repo_id": maybe_hash_with_repo_id_prefix(""),
            "machine_id": maybe_hash_with_repo_id_prefix(str(uuid.getnode())),
            "user_id": maybe_hash_with_repo_id_prefix(getuser()),
            # Note that we conserve the order in which the goals were specified on the cmd line.
            "standard_goals": [goal for goal in self.goals if goal in self.STANDARD_GOALS],
            # Lets us know of any custom goals were used, without knowing their names.
            "num_goals": str(len(self.goals)),
            "active_standard_backends": sorted(self.active_standard_backends),
        }

    def set_pantsd_scheduler_metrics(self, metrics: dict[str, int]) -> None:
        self._pantsd_metrics = metrics

    @property
    def pantsd_scheduler_metrics(self) -> dict[str, int]:
        return dict(self._pantsd_metrics)  # defensive copy

    def run_information(self) -> dict[str, Any]:
        """Basic information about this run."""
        return self._run_info

    def has_ended(self) -> bool:
        return self._has_ended

    def end_run(self, exit_code: ExitCode) -> None:
        """This pants run is over, so stop tracking it.

        Note: If end_run() has been called once, subsequent calls are no-ops.
        """

        if self.has_ended():
            return
        self._has_ended = True

        if self._run_start_time is None:
            raise Exception("RunTracker.end_run() called without calling .start()")

        duration = time.time() - self._run_start_time
        self._run_total_duration = duration

        outcome_str = "SUCCESS" if exit_code == PANTS_SUCCEEDED_EXIT_CODE else "FAILURE"
        self._run_info["outcome"] = outcome_str

        native_engine.set_per_run_log_path(None)

    def get_cumulative_timings(self) -> list[dict[str, Any]]:
        return [{"label": "main", "timing": self._run_total_duration}]

    def get_options_to_record(self) -> dict:
        recorded_options = {}
        scopes = self._all_options.for_global_scope().stats_record_option_scopes
        if "*" in scopes:
            scopes = self._all_options.known_scope_to_info.keys() if self._all_options else []
        for scope in scopes:
            scope_and_maybe_option = scope.split("^")
            if scope == GLOBAL_SCOPE:
                scope = GLOBAL_SCOPE_CONFIG_SECTION
            recorded_options[scope] = self._get_option_to_record(*scope_and_maybe_option)
        return recorded_options

    def _get_option_to_record(self, scope, option=None):
        """Looks up an option scope (and optionally option therein) in the options parsed by Pants.

        Returns a dict of of all options in the scope, if option is None. Returns the specific
        option if option is not None. Raises ValueError if scope or option could not be found.
        """
        scope_to_look_up = scope if scope != GLOBAL_SCOPE_CONFIG_SECTION else ""
        try:
            value = self._all_options.for_scope(
                scope_to_look_up, check_deprecations=False
            ).as_dict()
            if option is None:
                return value
            else:
                return value[option]
        except (ConfigValidationError, AttributeError) as e:
            option_str = "" if option is None else f" option {option}"
            raise ValueError(
                f"Couldn't find option scope {scope}{option_str} for recording ({e!r})"
            )

    def retrieve_logs(self) -> list[str]:
        """Get a list of every log entry recorded during this run."""

        if not self.run_logs_file:
            return []

        output = []
        try:
            with open(self.run_logs_file) as f:
                output = f.readlines()
        except OSError as e:
            logger.warning("Error retrieving per-run logs from RunTracker.", exc_info=e)

        return output

    @property
    def counter_names(self) -> tuple[str, ...]:
        return tuple(native_engine.all_counter_names())

1	# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	import logging	1✔
7	import platform	1✔
8	import socket	1✔
9	import time	1✔
10	import uuid	1✔
11	from hashlib import sha256	1✔
12	from pathlib import Path	1✔
13	from typing import Any	1✔
14
15	from pants.base.build_environment import get_buildroot	1✔
16	from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE, ExitCode	1✔
17	from pants.engine.internals import native_engine	1✔
18	from pants.option.errors import ConfigValidationError	1✔
19	from pants.option.options import Options	1✔
20	from pants.option.scope import GLOBAL_SCOPE, GLOBAL_SCOPE_CONFIG_SECTION	1✔
21	from pants.util.osutil import getuser	1✔
22	from pants.version import VERSION	1✔
23
24	logger = logging.getLogger(__name__)	1✔
25
26
27	class RunTracker:	1✔
28	"""Tracks and times the execution of a single Pants run."""
29
30	# TODO: Find a way to know from a goal name whether it's a standard or a custom
31	# goal whose name could, in theory, reveal something proprietary. That's more work than
32	# we want to do at the moment, so we maintain this manual list for now.
33	STANDARD_GOALS = frozenset(	1✔
34	(
35	"check",
36	"count-loc",
37	"dependents",
38	"dependencies",
39	"export-codegen",
40	"filedeps",
41	"fmt",
42	"lint",
43	"list",
44	"package",
45	"py-constraints",
46	"repl",
47	"roots",
48	"run",
49	"tailor",
50	"test",
51	"typecheck",
52	"validate",
53	)
54	)
55
56	def __init__(self, args: tuple[str, ...], options: Options):	1✔
57	"""
58	:API: public
59	"""
UNCOV 60	self._has_started: bool = False	×
UNCOV 61	self._has_ended: bool = False	×
62
63	# Select a globally unique ID for the run, that sorts by time.
UNCOV 64	run_timestamp = time.time()	×
UNCOV 65	run_uuid = uuid.uuid4().hex	×
UNCOV 66	str_time = time.strftime("%Y_%m_%d_%H_%M_%S", time.localtime(run_timestamp))	×
UNCOV 67	millis = int((run_timestamp * 1000) % 1000)	×
UNCOV 68	self.run_id = f"pants_run_{str_time}_{millis}_{run_uuid}"	×
69
UNCOV 70	self._args = args	×
UNCOV 71	self._all_options = options	×
UNCOV 72	info_dir = Path(self._all_options.for_global_scope().pants_workdir) / "run-tracker"	×
UNCOV 73	self._run_info: dict[str, Any] = {}	×
74
75	# pantsd stats.
UNCOV 76	self._pantsd_metrics: dict[str, int] = dict()	×
77
UNCOV 78	self.run_logs_file = info_dir / self.run_id / "logs"	×
UNCOV 79	self.run_logs_file.parent.mkdir(exist_ok=True, parents=True)	×
UNCOV 80	native_engine.set_per_run_log_path(str(self.run_logs_file))	×
81
82	# Initialized in `start()`.
UNCOV 83	self._run_start_time: float \| None = None	×
UNCOV 84	self._run_total_duration: float \| None = None	×
85
86	@property	1✔
87	def goals(self) -> list[str]:	1✔
UNCOV 88	return self._all_options.goals if self._all_options else []	×
89
90	@property	1✔
91	def active_standard_backends(self) -> list[str]:	1✔
UNCOV 92	return [	×
93	backend
94	for backend in self._all_options.for_global_scope().backend_packages
95	if backend.startswith("pants.backend.")
96	]
97
98	def start(self, run_start_time: float, specs: list[str]) -> None:	1✔
99	"""Start tracking this pants run."""
UNCOV 100	if self._has_started:	×
101	raise AssertionError("RunTracker.start must not be called multiple times.")	×
UNCOV 102	self._has_started = True	×
103
104	# Initialize the run.
UNCOV 105	self._run_start_time = run_start_time	×
106
UNCOV 107	datetime = time.strftime("%A %b %d, %Y %H:%M:%S", time.localtime(run_start_time))	×
UNCOV 108	cmd_line = " ".join(("pants",) + self._args[1:])	×
109
UNCOV 110	self._run_info.update(	×
111	{
112	"id": self.run_id,
113	"timestamp": run_start_time,
114	"datetime": datetime,
115	"user": getuser(),
116	"machine": socket.gethostname(),
117	"buildroot": get_buildroot(),
118	"path": get_buildroot(),
119	"version": VERSION,
120	"cmd_line": cmd_line,
121	"specs_from_command_line": specs,
122	}
123	)
124
125	def get_anonymous_telemetry_data(self, unhashed_repo_id: str) -> dict[str, str \| list[str]]:	1✔
UNCOV 126	def maybe_hash_with_repo_id_prefix(s: str) -> str:	×
UNCOV 127	qualified_str = f"{unhashed_repo_id}.{s}" if s else unhashed_repo_id	×
128	# If the repo_id is the empty string we return a blank string.
UNCOV 129	return sha256(qualified_str.encode()).hexdigest() if unhashed_repo_id else ""	×
130
UNCOV 131	return {	×
132	"run_id": str(self._run_info.get("id", uuid.uuid4())),
133	"timestamp": str(self._run_info.get("timestamp")),
134	# Note that this method is called after the StreamingWorkunitHandler.session() ends,
135	# i.e., after end_run() has been called, so duration will be set.
136	"duration": str(self._run_total_duration),
137	"outcome": str(self._run_info.get("outcome")),
138	"platform": platform.platform(),
139	"python_implementation": platform.python_implementation(),
140	"python_version": platform.python_version(),
141	"pants_version": str(self._run_info.get("version")),
142	# Note that if repo_id is the empty string then these three fields will be empty.
143	"repo_id": maybe_hash_with_repo_id_prefix(""),
144	"machine_id": maybe_hash_with_repo_id_prefix(str(uuid.getnode())),
145	"user_id": maybe_hash_with_repo_id_prefix(getuser()),
146	# Note that we conserve the order in which the goals were specified on the cmd line.
147	"standard_goals": [goal for goal in self.goals if goal in self.STANDARD_GOALS],
148	# Lets us know of any custom goals were used, without knowing their names.
149	"num_goals": str(len(self.goals)),
150	"active_standard_backends": sorted(self.active_standard_backends),
151	}
152
153	def set_pantsd_scheduler_metrics(self, metrics: dict[str, int]) -> None:	1✔
154	self._pantsd_metrics = metrics	×
155
156	@property	1✔
157	def pantsd_scheduler_metrics(self) -> dict[str, int]:	1✔
158	return dict(self._pantsd_metrics) # defensive copy	×
159
160	def run_information(self) -> dict[str, Any]:	1✔
161	"""Basic information about this run."""
UNCOV 162	return self._run_info	×
163
164	def has_ended(self) -> bool:	1✔
UNCOV 165	return self._has_ended	×
166
167	def end_run(self, exit_code: ExitCode) -> None:	1✔
168	"""This pants run is over, so stop tracking it.
169
170	Note: If end_run() has been called once, subsequent calls are no-ops.
171	"""
172
UNCOV 173	if self.has_ended():	×
174	return	×
UNCOV 175	self._has_ended = True	×
176
UNCOV 177	if self._run_start_time is None:	×
178	raise Exception("RunTracker.end_run() called without calling .start()")	×
179
UNCOV 180	duration = time.time() - self._run_start_time	×
UNCOV 181	self._run_total_duration = duration	×
182
UNCOV 183	outcome_str = "SUCCESS" if exit_code == PANTS_SUCCEEDED_EXIT_CODE else "FAILURE"	×
UNCOV 184	self._run_info["outcome"] = outcome_str	×
185
UNCOV 186	native_engine.set_per_run_log_path(None)	×
187
188	def get_cumulative_timings(self) -> list[dict[str, Any]]:	1✔
UNCOV 189	return [{"label": "main", "timing": self._run_total_duration}]	×
190
191	def get_options_to_record(self) -> dict:	1✔
192	recorded_options = {}	×
193	scopes = self._all_options.for_global_scope().stats_record_option_scopes	×
194	if "*" in scopes:	×
195	scopes = self._all_options.known_scope_to_info.keys() if self._all_options else []	×
196	for scope in scopes:	×
197	scope_and_maybe_option = scope.split("^")	×
198	if scope == GLOBAL_SCOPE:	×
199	scope = GLOBAL_SCOPE_CONFIG_SECTION	×
200	recorded_options[scope] = self._get_option_to_record(*scope_and_maybe_option)	×
201	return recorded_options	×
202
203	def _get_option_to_record(self, scope, option=None):	1✔
204	"""Looks up an option scope (and optionally option therein) in the options parsed by Pants.
205
206	Returns a dict of of all options in the scope, if option is None. Returns the specific
207	option if option is not None. Raises ValueError if scope or option could not be found.
208	"""
209	scope_to_look_up = scope if scope != GLOBAL_SCOPE_CONFIG_SECTION else ""	×
210	try:	×
211	value = self._all_options.for_scope(	×
212	scope_to_look_up, check_deprecations=False
213	).as_dict()
214	if option is None:	×
215	return value	×
216	else:
217	return value[option]	×
218	except (ConfigValidationError, AttributeError) as e:	×
219	option_str = "" if option is None else f" option {option}"	×
220	raise ValueError(	×
221	f"Couldn't find option scope {scope}{option_str} for recording ({e!r})"
222	)
223
224	def retrieve_logs(self) -> list[str]:	1✔
225	"""Get a list of every log entry recorded during this run."""
226
227	if not self.run_logs_file:	×
228	return []	×
229
230	output = []	×
231	try:	×
232	with open(self.run_logs_file) as f:	×
233	output = f.readlines()	×
234	except OSError as e:	×
235	logger.warning("Error retrieving per-run logs from RunTracker.", exc_info=e)	×
236
237	return output	×
238
239	@property	1✔
240	def counter_names(self) -> tuple[str, ...]:	1✔
241	return tuple(native_engine.all_counter_names())	×

pantsbuild / pants / 19015773527

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous