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
This commit is contained in:
Peter Rowlands (변기호)
2021-01-04 19:25:13 +09:00
committed by GitHub
parent 8010005364
commit cffc019e1f
6 changed files with 68 additions and 50 deletions
+2 -3
View File
@@ -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."
),
)
+5 -14
View File
@@ -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 "
+50 -27
View File
@@ -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,
+4
View File
@@ -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,
+5 -5
View File
@@ -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()
+2 -1
View File
@@ -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)