From cffc019e1f2f5489aa5f2fc7efca38b0535a2b46 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20Rowlands=20=28=EB=B3=80=EA=B8=B0=ED=98=B8=29?= Date: Mon, 4 Jan 2021 19:25:13 +0900 Subject: [PATCH] experiments: fix logging bugs (#5195) * experiments: use relpaths when tracking changed files * experiments: suppress SCM hints in tempdir executors * experiments: move multiprocessing error logging into child * exp run: clarify help text for --temp --- dvc/command/experiments.py | 5 +- dvc/repo/experiments/__init__.py | 19 ++----- dvc/repo/experiments/executor/base.py | 77 +++++++++++++++++--------- dvc/repo/experiments/executor/local.py | 4 ++ dvc/repo/reproduce.py | 10 ++-- dvc/scm/git/__init__.py | 3 +- 6 files changed, 68 insertions(+), 50 deletions(-) diff --git a/dvc/command/experiments.py b/dvc/command/experiments.py index 4c8648bae..bb5290a2b 100644 --- a/dvc/command/experiments.py +++ b/dvc/command/experiments.py @@ -1093,7 +1093,7 @@ def _add_run_common(parser): "--run-all", action="store_true", default=False, - help="Execute all experiments in the run queue.", + help="Execute all experiments in the run queue. Implies --temp.", ) parser.add_argument( "-j", @@ -1108,7 +1108,6 @@ def _add_run_common(parser): dest="tmp_dir", help=( "Run this experiment in a separate temporary directory instead of " - "your workspace. Only applies when running a single experiment " - "without --queue." + "your workspace." ), ) diff --git a/dvc/repo/experiments/__init__.py b/dvc/repo/experiments/__init__.py index 33817963e..8248d75cc 100644 --- a/dvc/repo/experiments/__init__.py +++ b/dvc/repo/experiments/__init__.py @@ -4,7 +4,6 @@ import re import signal from collections import defaultdict, namedtuple from concurrent.futures import CancelledError, ProcessPoolExecutor, wait -from contextlib import contextmanager from functools import wraps from multiprocessing import Manager from typing import Dict, Iterable, Mapping, Optional @@ -112,10 +111,6 @@ class Experiments: def dvc_dir(self): return relpath(self.repo.dvc_dir, self.repo.scm.root_dir) - @contextmanager - def chdir(self): - yield - @cached_property def args_file(self): from .executor.base import BaseExecutor @@ -607,22 +602,18 @@ class Experiments: for future, (rev, executor) in futures.items(): rev, executor = futures[future] - exc = future.exception() try: + exc = future.exception() if exc is None: exec_result = future.result() result[rev].update( self._collect_executor(executor, exec_result) ) - else: - # Checkpoint errors have already been logged - if not isinstance(exc, CheckpointKilledError): - logger.exception( - "Failed to reproduce experiment '%s'", - rev[:7], - exc_info=exc, - ) + elif not isinstance(exc, CheckpointKilledError): + logger.error( + "Failed to reproduce experiment '%s'", rev[:7], + ) except CancelledError: logger.error( "Cancelled before attempting to reproduce experiment " diff --git a/dvc/repo/experiments/executor/base.py b/dvc/repo/experiments/executor/base.py index c3326d321..31f4f3a2c 100644 --- a/dvc/repo/experiments/executor/base.py +++ b/dvc/repo/experiments/executor/base.py @@ -2,6 +2,7 @@ import logging import os import pickle from abc import ABC, abstractmethod +from contextlib import contextmanager from functools import partial from typing import ( TYPE_CHECKING, @@ -14,6 +15,7 @@ from typing import ( from funcy import cached_property +from dvc.exceptions import DvcException from dvc.path_info import PathInfo from dvc.repo.experiments.base import ( EXEC_BASELINE, @@ -31,6 +33,7 @@ from dvc.repo.experiments.base import ( ) from dvc.scm import SCM from dvc.stage import PipelineStage +from dvc.stage.run import CheckpointKilledError from dvc.stage.serialize import to_lockfile from dvc.utils import dict_sha256 from dvc.utils.fs import remove @@ -62,6 +65,7 @@ class BaseExecutor(ABC): PACKED_ARGS_FILE = "repro.dat" WARN_UNTRACKED = False + QUIET = False def __init__( self, @@ -224,7 +228,6 @@ class BaseExecutor(ABC): and force is a bool specifying whether or not this experiment should force overwrite any existing duplicates. """ - from dvc.repo import Repo from dvc.repo.checkout import checkout as dvc_checkout from dvc.repo.reproduce import reproduce as dvc_reproduce @@ -244,27 +247,8 @@ class BaseExecutor(ABC): exp_ref: Optional["ExpRefInfo"] = None repro_force: bool = False - try: - dvc = Repo(dvc_dir) - if dvc_dir is not None: - old_cwd: Optional[str] = os.getcwd() - if rel_cwd: - os.chdir(os.path.join(dvc.root_dir, rel_cwd)) - else: - os.chdir(dvc.root_dir) - else: - old_cwd = None - logger.debug("Running repro in '%s'", os.getcwd()) - - args_path = os.path.join( - dvc.tmp_dir, BaseExecutor.PACKED_ARGS_FILE - ) - if os.path.exists(args_path): - args, kwargs = BaseExecutor.unpack_repro_args(args_path) - remove(args_path) - else: - args = [] - kwargs = {} + with cls._repro_dvc(dvc_dir, rel_cwd) as dvc: + args, kwargs = cls._repro_args(dvc) repro_force = kwargs.get("force", False) logger.debug("force = %s", str(repro_force)) @@ -317,17 +301,56 @@ class BaseExecutor(ABC): "\t%s", ", ".join(untracked), ) - finally: - if dvc: - dvc.scm.close() - if old_cwd: - os.chdir(old_cwd) # ideally we would return stages here like a normal repro() call, but # stages is not currently picklable and cannot be returned across # multiprocessing calls return ExecutorResult(exp_hash, exp_ref, repro_force) + @classmethod + @contextmanager + def _repro_dvc(cls, dvc_dir: Optional[str], rel_cwd: Optional[str]): + from dvc.repo import Repo + + dvc = Repo(dvc_dir) + if cls.QUIET: + dvc.scm.quiet = cls.QUIET + if dvc_dir is not None: + old_cwd: Optional[str] = os.getcwd() + if rel_cwd: + os.chdir(os.path.join(dvc.root_dir, rel_cwd)) + else: + os.chdir(dvc.root_dir) + else: + old_cwd = None + logger.debug("Running repro in '%s'", os.getcwd()) + + try: + yield dvc + except CheckpointKilledError: + raise + except DvcException: + logger.exception("") + raise + except Exception: + logger.exception("unexpected error") + raise + finally: + dvc.scm.close() + if old_cwd: + os.chdir(old_cwd) + + @classmethod + def _repro_args(cls, dvc): + args_path = os.path.join(dvc.tmp_dir, cls.PACKED_ARGS_FILE) + if os.path.exists(args_path): + args, kwargs = cls.unpack_repro_args(args_path) + remove(args_path) + else: + args = [] + kwargs = {} + return args, kwargs + @classmethod def checkpoint_callback( cls, diff --git a/dvc/repo/experiments/executor/local.py b/dvc/repo/experiments/executor/local.py index 8fdb0d571..52ac6acb7 100644 --- a/dvc/repo/experiments/executor/local.py +++ b/dvc/repo/experiments/executor/local.py @@ -25,7 +25,11 @@ class BaseLocalExecutor(BaseExecutor): class TempDirExecutor(BaseLocalExecutor): """Temp directory experiment executor.""" + # Temp dir executors should warn if untracked files exist (to help with + # debugging user code), and suppress other DVC hints (like `git add` + # suggestions) that are not applicable outside of workspace runs WARN_UNTRACKED = True + QUIET = True def __init__( self, diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index b36395588..e4872b9bd 100644 --- a/dvc/repo/reproduce.py +++ b/dvc/repo/reproduce.py @@ -1,11 +1,11 @@ import logging -import os import typing from functools import partial from dvc.exceptions import ReproductionError from dvc.repo.scm_context import scm_context from dvc.stage.run import CheckpointKilledError +from dvc.utils import relpath from . import locked from .graph import get_pipeline, get_pipelines @@ -60,19 +60,19 @@ def _dump_stage(stage): def _track_stage(stage): - stage.repo.scm.track_file(stage.dvcfile.path) + stage.repo.scm.track_file(stage.dvcfile.relpath) for dep in stage.deps: if not dep.use_scm_ignore and dep.is_in_repo: - stage.repo.scm.track_file(os.fspath(dep.path_info)) + stage.repo.scm.track_file(relpath(dep.path_info)) for out in stage.outs: if not out.use_scm_ignore and out.is_in_repo: - stage.repo.scm.track_file(os.fspath(out.path_info)) + stage.repo.scm.track_file(relpath(out.path_info)) if out.live: from dvc.repo.live import summary_path_info summary = summary_path_info(out) if summary: - stage.repo.scm.track_file(os.fspath(summary)) + stage.repo.scm.track_file(relpath(summary)) stage.repo.scm.track_changed_files() diff --git a/dvc/scm/git/__init__.py b/dvc/scm/git/__init__.py index 6f851d81e..6a2cd757f 100644 --- a/dvc/scm/git/__init__.py +++ b/dvc/scm/git/__init__.py @@ -45,6 +45,7 @@ class Git(Base): ): self.ignored_paths: List[str] = [] self.files_to_track: Set[str] = set() + self.quiet: bool = False selected = backends if backends else self.DEFAULT_BACKENDS.keys() self.backends = OrderedDict( @@ -243,7 +244,7 @@ class Git(Base): self.files_to_track = set() def remind_to_track(self): - if not self.files_to_track: + if self.quiet or not self.files_to_track: return files = " ".join(shlex.quote(path) for path in self.files_to_track)