Add more context for Job errors. (#2479)

This makes several error paths that manifest when creating locks and
PEXes present better error details.

Several changes support this:
1. Jobs can now have a context label which is used, when set, to prefix
   all job error output. This should help root cause the problem; i.e.:
   is it fundamentally Pex or is it Pip or is it a problem with a bad
   package?
2. When a Pip download errors, we now show all STDERR output instead of
   just the last line whenever the last line contains "See above for
   details", which Pip emits to indicate an error buried deeper in the
   logs.
3. A new log analyzer is added to all Pip downloads that knows how to
   spot sdist build errors and surface their details.
4. A fail-safe is added to the Pip download log scraper to just report
   the full Pip log content when no better analysis was found.

Fixes #2113

pex/build_system/pep_517.py

@@ -186,7 +186,13 @@ def _invoke_build_hook(
             stderr=subprocess.PIPE,
         )
         return SpawnedJob.file(
-            Job(command=args, process=process),
+            Job(
+                command=args,
+                process=process,
+                context="PEP-517:{hook_method} at {project_directory}".format(
+                    hook_method=hook_method, project_directory=project_directory
+                ),
+            ),
             output_file=fp.name,
             result_func=lambda file_content: json.loads(file_content.decode("utf-8")),
         )

pex/dist_metadata.py

@@ -651,22 +651,40 @@ def requires_dists(location):
         )
 
 
+# Frozen exception types don't work under 3.11+ where the `__traceback__` attribute can be set
+# after construction in some cases.
+@attr.s
 class RequirementParseError(Exception):
     """Indicates and invalid requirement string.
 
     See PEP-508: https://www.python.org/dev/peps/pep-0508
     """
 
+    error = attr.ib()  # type: Any
+    source = attr.ib(default=None)  # type: Optional[str]
+
+    def __str__(self):
+        # type: () -> str
+        if not self.source:
+            return str(self.error)
+        return "Failed to parse a requirement of {source}: {err}".format(
+            err=self.error, source=self.source
+        )
+
 
 @attr.s(frozen=True)
 class Constraint(object):
     @classmethod
-    def parse(cls, constraint):
-        # type: (Text) -> Constraint
+    def parse(
+        cls,
+        constraint,  # type: Text
+        source=None,  # type: Optional[str]
+    ):
+        # type: (...) -> Constraint
         try:
             return cls.from_packaging_requirement(PackagingRequirement(constraint))
         except InvalidRequirement as e:
-            raise RequirementParseError(str(e))
+            raise RequirementParseError(str(e), source=source)
 
     @classmethod
     def from_packaging_requirement(cls, requirement):
@@ -777,12 +795,16 @@ def as_requirement(self):
 @attr.s(frozen=True)
 class Requirement(Constraint):
     @classmethod
-    def parse(cls, requirement):
-        # type: (Text) -> Requirement
+    def parse(
+        cls,
+        requirement,  # type: Text
+        source=None,  # type: Optional[str]
+    ):
+        # type: (...) -> Requirement
         try:
             return cls.from_packaging_requirement(PackagingRequirement(requirement))
         except InvalidRequirement as e:
-            raise RequirementParseError(str(e))
+            raise RequirementParseError(str(e), source=source)
 
     @classmethod
     def from_packaging_requirement(cls, requirement):

pex/interpreter.py

@@ -46,7 +46,6 @@
         Mapping,
         MutableMapping,
         Optional,
-        Sequence,
         Text,
         Tuple,
         Union,
@@ -1566,45 +1565,6 @@ def __repr__(self):
         )
 
 
-def spawn_python_job(
-    args,  # type: Iterable[str]
-    env=None,  # type: Optional[Mapping[str, str]]
-    interpreter=None,  # type: Optional[PythonInterpreter]
-    expose=None,  # type: Optional[Sequence[str]]
-    pythonpath=None,  # type: Optional[Iterable[str]]
-    **subprocess_kwargs  # type: Any
-):
-    # type: (...) -> Job
-    """Spawns a python job.
-
-    :param args: The arguments to pass to the python interpreter.
-    :param env: The environment to spawn the python interpreter process in. Defaults to the ambient
-                environment.
-    :param interpreter: The interpreter to use to spawn the python job. Defaults to the current
-                        interpreter.
-    :param expose: The names of any vendored distributions to expose to the spawned python process.
-                   These will be appended to `pythonpath` if passed.
-    :param pythonpath: The PYTHONPATH to expose to the spawned python process. These will be
-                       pre-pended to the `expose` path if passed.
-    :param subprocess_kwargs: Any additional :class:`subprocess.Popen` kwargs to pass through.
-    :returns: A job handle to the spawned python process.
-    """
-    pythonpath = list(pythonpath or ())
-    subprocess_env = dict(env or os.environ)
-    if expose:
-        # In order to expose vendored distributions with their un-vendored import paths in-tact, we
-        # need to set `__PEX_UNVENDORED__`. See: vendor.__main__.ImportRewriter._modify_import.
-        subprocess_env["__PEX_UNVENDORED__"] = ",".join(expose)
-
-        pythonpath.extend(third_party.expose(expose, interpreter=interpreter))
-
-    interpreter = interpreter or PythonInterpreter.get()
-    cmd, process = interpreter.open_process(
-        args=args, pythonpath=pythonpath, env=subprocess_env, **subprocess_kwargs
-    )
-    return Job(command=cmd, process=process)
-
-
 # See the "Test results from various systems" table here:
 #  https://www.in-ulm.de/~mascheck/various/shebang/#length
 MAX_SHEBANG_LENGTH = 512 if sys.platform == "darwin" else 128

pex/jobs.py

@@ -63,19 +63,33 @@ def __init__(
             exitcode,  # type: int
             stderr,  # type: Optional[Text]
             message,  # type: str
+            context=None,  # type: Optional[str]
         ):
             # type: (...) -> None
-            super(Job.Error, self).__init__(message)
+            super(Job.Error, self).__init__(
+                "{ctx}: {msg}".format(ctx=context, msg=message) if context else message
+            )
             self.pid = pid
             self.command = command
             self.exitcode = exitcode
             self.stderr = stderr
+            self._context = context
+
+        def contextualized_stderr(self):
+            # type: () -> Iterator[Text]
+            if self.stderr:
+                for line in self.stderr.splitlines():
+                    if not self._context:
+                        yield line
+                    else:
+                        yield "{ctx}: {line}".format(ctx=self._context, line=line)
 
     def __init__(
         self,
         command,  # type: Iterable[str]
         process,  # type: subprocess.Popen
         finalizer=None,  # type: Optional[Callable[[int], None]]
+        context=None,  # type: Optional[str]
     ):
         # type: (...) -> None
         """
@@ -84,10 +98,13 @@ def __init__(
         :param finalizer: An optional cleanup function to call exactly once with the process return
                           code when the underlying process terminates in the course of calling this
                           job's public methods.
+        :param context: An optional context labeling the job that will be used to decorate error
+                        information.
         """
         self._command = tuple(command)
         self._process = process
         self._finalizer = finalizer
+        self._context = context
 
     def wait(self):
         # type: () -> None
@@ -153,6 +170,7 @@ def create_error(
             exitcode=self._process.returncode,
             stderr=err,
             message=msg,
+            context=self._context,
         )
 
     def _finalize_job(self):
@@ -407,7 +425,7 @@ def job_error_message(
             pid=job_error.pid,
             command=" ".join(job_error.command),
             exitcode=job_error.exitcode,
-            stderr=job_error.stderr,
+            stderr="\n".join(job_error.contextualized_stderr()),
         )
 
     @abstractmethod

pex/pip/dependencies/requires.py

@@ -22,7 +22,7 @@ def requires(self, *args, **kwargs):
             modified_requires = []
             orig = orig_requires(self, *args, **kwargs)
             for req in orig:
-                requirement = PexRequirement.parse(str(req))
+                requirement = PexRequirement.parse(str(req), source=repr(self))
                 excluded_by = dependency_configuration.excluded_by(requirement)
                 if excluded_by:
                     logger.debug(

pex/pip/log_analyzer.py

@@ -7,6 +7,7 @@
 import subprocess
 from abc import abstractmethod
 
+from pex.common import safe_delete
 from pex.jobs import Job
 from pex.typing import TYPE_CHECKING, Generic
 
@@ -81,7 +82,7 @@ def __init__(
         self._log = log
         self._log_analyzers = list(log_analyzers)
         self._preserve_log = preserve_log
-        super(LogScrapeJob, self).__init__(command, process, finalizer=finalizer)
+        super(LogScrapeJob, self).__init__(command, process, finalizer=finalizer, context="pip")
 
     def _check_returncode(self, stderr=None):
         # type: (Optional[bytes]) -> None
@@ -90,6 +91,7 @@ def _check_returncode(self, stderr=None):
             for analyzer in self._log_analyzers
             if analyzer.should_collect(self._process.returncode)
         ]
+        analyzed_stderr = b""  # type: bytes
         if activated_analyzers:
             collected = []
             # A process may fail so early that there is no log file to analyze.
@@ -100,7 +102,7 @@ def _check_returncode(self, stderr=None):
                     for line in fp:
                         if not activated_analyzers:
                             break
-                        for index, analyzer in enumerate(activated_analyzers):
+                        for index, analyzer in tuple(enumerate(activated_analyzers)):
                             result = analyzer.analyze(line)
                             if isinstance(result.data, ErrorMessage):
                                 collected.append(result.data)
@@ -110,7 +112,21 @@ def _check_returncode(self, stderr=None):
                                     break
                 for analyzer in activated_analyzers:
                     analyzer.analysis_completed()
-                if not self._preserve_log:
-                    os.unlink(self._log)
-                stderr = (stderr or b"") + "".join(collected).encode("utf-8")
-        super(LogScrapeJob, self)._check_returncode(stderr=stderr)
+            if collected:
+                analyzed_stderr = "".join(collected).encode("utf-8")
+
+        # Fall back to displaying the Pip logs in full if we have no stderr output or analysis. It's
+        # likely overwhelming, but better than silence and useful for debugging.
+        if (
+            not stderr
+            and not analyzed_stderr
+            and self._process.returncode != 0
+            and os.path.isfile(self._log)
+        ):
+            with open(self._log, "rb") as fp:
+                analyzed_stderr = fp.read()
+
+        if not self._preserve_log:
+            safe_delete(self._log)
+
+        super(LogScrapeJob, self)._check_returncode(stderr=(stderr or b"") + analyzed_stderr)