diff --git a/dvc/repo/experiments/diff.py b/dvc/repo/experiments/diff.py index 7ad7e7b02..f8a9fbe91 100644 --- a/dvc/repo/experiments/diff.py +++ b/dvc/repo/experiments/diff.py @@ -7,7 +7,7 @@ logger = logging.getLogger(__name__) def diff(repo, *args, a_rev=None, b_rev=None, param_deps=False, **kwargs): - from dvc.repo.experiments.show import _collect_experiment_commit + from dvc.repo.experiments.show import collect_experiment_commit from dvc.scm import resolve_rev if repo.scm.no_commits: @@ -15,15 +15,15 @@ def diff(repo, *args, a_rev=None, b_rev=None, param_deps=False, **kwargs): if a_rev: rev = resolve_rev(repo.scm, a_rev) - old = _collect_experiment_commit(repo, rev, param_deps=param_deps) + old = collect_experiment_commit(repo, rev, param_deps=param_deps) else: - old = _collect_experiment_commit(repo, "HEAD", param_deps=param_deps) + old = collect_experiment_commit(repo, "HEAD", param_deps=param_deps) if b_rev: rev = resolve_rev(repo.scm, b_rev) - new = _collect_experiment_commit(repo, rev, param_deps=param_deps) + new = collect_experiment_commit(repo, rev, param_deps=param_deps) else: - new = _collect_experiment_commit( + new = collect_experiment_commit( repo, "workspace", param_deps=param_deps ) diff --git a/dvc/repo/experiments/queue/utils.py b/dvc/repo/experiments/queue/utils.py index d77d33648..4488327fe 100644 --- a/dvc/repo/experiments/queue/utils.py +++ b/dvc/repo/experiments/queue/utils.py @@ -1,6 +1,8 @@ import logging from typing import TYPE_CHECKING, Dict +from scmrepo.exceptions import SCMError + from ..executor.base import ExecutorInfo, TaskStatus logger = logging.getLogger(__name__) @@ -52,7 +54,7 @@ def fetch_running_exp_from_temp_dir( result[rev]["last"] = last_rev if last_rev: result[last_rev] = info.asdict() - except InvalidRemoteSCMRepo: + except (InvalidRemoteSCMRepo, SCMError): # ignore stale info files del result[rev] return result diff --git a/dvc/repo/experiments/show.py b/dvc/repo/experiments/show.py index ee96a83c3..0ab3f4928 100644 --- a/dvc/repo/experiments/show.py +++ b/dvc/repo/experiments/show.py @@ -3,16 +3,29 @@ from collections import OrderedDict, defaultdict from datetime import datetime from enum import Enum from itertools import chain -from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + List, + Optional, + Tuple, + Union, +) + +from scmrepo.exceptions import SCMError as InnerScmError from dvc.repo.metrics.show import _gather_metrics from dvc.repo.params.show import _gather_params -from dvc.scm import iter_revs +from dvc.scm import SCMError, iter_revs, resolve_rev from dvc.utils import error_handler, onerror_collect, relpath from .refs import ExpRefInfo if TYPE_CHECKING: + from scmrepo.git.objects import GitCommit + from dvc.repo import Repo logger = logging.getLogger(__name__) @@ -25,34 +38,49 @@ class ExpStatus(Enum): Failed = 3 +def _is_scm_error(collected_exp: Dict[str, Any]) -> bool: + if "error" in collected_exp and ( + isinstance(collected_exp["error"], SCMError) + or isinstance(collected_exp["error"], InnerScmError) + ): + return True + return False + + +def _show_onerror_collect(result: Dict, exception: Exception, *args, **kwargs): + onerror_collect(result, exception, *args, **kwargs) + result["data"] = {} + + @error_handler -def _collect_experiment_commit( +def collect_experiment_commit( repo: "Repo", exp_rev: str, status: ExpStatus = ExpStatus.Success, param_deps=False, - running=None, + running: Optional[Dict[str, Any]] = None, onerror: Optional[Callable] = None, -): +) -> Dict[str, Any]: from dvc.dependency import ParamsDependency, RepoDependency - res: Dict[str, Optional[Any]] = defaultdict(dict) + result: Dict[str, Any] = defaultdict(dict) + running = running or {} for rev in repo.brancher(revs=[exp_rev]): if rev == "workspace": if exp_rev != "workspace": continue - res["timestamp"] = None + result["timestamp"] = None else: commit = repo.scm.resolve_commit(rev) - res["timestamp"] = datetime.fromtimestamp(commit.commit_time) + result["timestamp"] = datetime.fromtimestamp(commit.commit_time) params = _gather_params( repo, rev=rev, targets=None, deps=param_deps, onerror=onerror ) if params: - res["params"] = params + result["params"] = params - res["deps"] = { + result["deps"] = { relpath(dep.fs_path, repo.root_dir): { "hash": dep.hash_info.value, "size": dep.meta.size, @@ -61,7 +89,7 @@ def _collect_experiment_commit( for dep in repo.index.deps if not isinstance(dep, (ParamsDependency, RepoDependency)) } - res["outs"] = { + result["outs"] = { relpath(out.fs_path, repo.root_dir): { "hash": out.hash_info.value, "size": out.meta.size, @@ -73,14 +101,14 @@ def _collect_experiment_commit( if not (out.is_metric or out.is_plot) } - res["status"] = status.name + result["status"] = status.name if status == ExpStatus.Running: - res["executor"] = running.get(exp_rev, {}).get("location", None) + result["executor"] = running.get(exp_rev, {}).get("location", None) else: - res["executor"] = None + result["executor"] = None if status == ExpStatus.Failed: - res["error"] = { + result["error"] = { "msg": "Experiment run failed.", "type": "", } @@ -89,54 +117,102 @@ def _collect_experiment_commit( vals = _gather_metrics( repo, targets=None, rev=rev, recursive=False, onerror=onerror ) - res["metrics"] = vals + result["metrics"] = vals - return res + return result -def _collect_experiment_branch( - res, - repo, - branch, - baseline, - onerror: Optional[Callable] = None, - running=None, +def _collect_complete_experiment( + repo: "Repo", + baseline: str, + exp_rev: str, + running: Dict[str, Any], + revs: List[str], **kwargs, -): - from dvc.scm import resolve_rev +) -> Dict[str, Dict[str, Any]]: + results: "OrderedDict[str, Dict[str, Any]]" = OrderedDict() - exp_rev = resolve_rev(repo.scm, branch) - prev = None - revs = list(repo.scm.branch_revs(exp_rev, baseline)) + checkpoint: bool = True if len(revs) > 1 else False + prev = "" for rev in revs: status = ExpStatus.Running if rev in running else ExpStatus.Success - collected_exp = _collect_experiment_commit( + collected_exp = collect_experiment_commit( repo, rev, - onerror=onerror, status=status, running=running, **kwargs, ) - if len(revs) > 1: + if _is_scm_error(collected_exp): + return {} + if checkpoint: exp = {"checkpoint_tip": exp_rev} if prev: - res[prev]["data"][ # type: ignore[unreachable] + results[prev]["data"][ # type: ignore[unreachable] "checkpoint_parent" ] = rev - if rev in res: - res[rev]["data"].update(exp) - res.move_to_end(rev) + if rev in results: + results[rev]["data"].update(exp) + results.move_to_end(rev) else: exp.update(collected_exp["data"]) else: exp = collected_exp["data"] - if rev not in res: - res[rev] = {"data": exp} + if rev not in results: + results[rev] = {"data": exp} prev = rev - if len(revs) > 1: - res[prev]["data"]["checkpoint_parent"] = baseline - return res + if checkpoint and prev: + results[prev]["data"]["checkpoint_parent"] = baseline + return results + + +def _collect_branch( + repo: "Repo", + baseline: str, + running: Dict[str, Any], + **kwargs, +) -> Dict[str, Dict[str, Any]]: + results: Dict[str, Dict[str, Any]] = OrderedDict() + status = ExpStatus.Running if baseline in running else ExpStatus.Success + results["baseline"] = collect_experiment_commit( + repo, + baseline, + status=status, + running=running, + **kwargs, + ) + if baseline == "workspace" or _is_scm_error(results["baseline"]): + return results + + ref_info = ExpRefInfo(baseline_sha=baseline) + commits: List[Tuple[str, "GitCommit", str, List[str]]] = [] + + for ref in repo.scm.iter_refs(base=str(ref_info)): + try: + commit = repo.scm.resolve_commit(ref) + exp_rev = resolve_rev(repo.scm, ref) + revs = list(repo.scm.branch_revs(exp_rev, baseline)) + except (SCMError, InnerScmError): + continue + commits.append((ref, commit, exp_rev, revs)) + + for exp_ref, _, exp_rev, revs in sorted( + commits, key=lambda x: x[1].commit_time, reverse=True + ): + ref_info = ExpRefInfo.from_ref(exp_ref) + assert ref_info.baseline_sha == baseline + collected_exp = _collect_complete_experiment( + repo, + baseline=baseline, + exp_rev=exp_rev, + running=running, + revs=revs, + **kwargs, + ) + if _is_scm_error(collected_exp): + continue + results.update(collected_exp) + return results def get_names(repo: "Repo", result: Dict[str, Dict[str, Any]]): @@ -178,7 +254,6 @@ def get_names(repo: "Repo", result: Dict[str, Dict[str, Any]]): rev_result["data"]["name"] = name -# flake8: noqa: C901 def _collect_active_experiment( repo: "Repo", found_revs: Dict[str, List[str]], @@ -194,13 +269,16 @@ def _collect_active_experiment( if entry.baseline_rev in found_revs and ( stash_rev not in running or not running[stash_rev].get("last") ): - result[entry.baseline_rev][stash_rev] = _collect_experiment_commit( + collected_exp = collect_experiment_commit( repo, stash_rev, status=ExpStatus.Running, running=running, **kwargs, ) + if _is_scm_error(collected_exp): + continue + result[entry.baseline_rev][stash_rev] = collected_exp return result @@ -214,13 +292,16 @@ def _collect_queued_experiment( for entry in repo.experiments.celery_queue.iter_queued(): stash_rev = entry.stash_rev if entry.baseline_rev in found_revs: - result[entry.baseline_rev][stash_rev] = _collect_experiment_commit( + collected_exp = collect_experiment_commit( repo, stash_rev, status=ExpStatus.Queued, running=running, **kwargs, ) + if _is_scm_error(collected_exp): + continue + result[entry.baseline_rev][stash_rev] = collected_exp return result @@ -235,14 +316,16 @@ def _collect_failed_experiment( entry = queue_done_result.entry stash_rev = entry.stash_rev if entry.baseline_rev in found_revs: - experiment = _collect_experiment_commit( + collected_exp = collect_experiment_commit( repo, stash_rev, status=ExpStatus.Failed, running=running, **kwargs, ) - result[entry.baseline_rev][stash_rev] = experiment + if _is_scm_error(collected_exp): + continue + result[entry.baseline_rev][stash_rev] = collected_exp return result @@ -273,7 +356,7 @@ def show( return {} if onerror is None: - onerror = onerror_collect + onerror = _show_onerror_collect res: Dict[str, Dict] = defaultdict(OrderedDict) @@ -324,47 +407,20 @@ def show( else {} ) - for rev in found_revs: - status = ExpStatus.Running if rev in running else ExpStatus.Success - res[rev]["baseline"] = _collect_experiment_commit( + for baseline in found_revs: + res[baseline] = _collect_branch( repo, - rev, - status=status, - param_deps=param_deps, + baseline, running=running, + param_deps=param_deps, onerror=onerror, ) - if rev == "workspace": - continue - - ref_info = ExpRefInfo(baseline_sha=rev) - commits = [ - (ref, repo.scm.resolve_commit(ref)) - for ref in repo.scm.iter_refs(base=str(ref_info)) - ] - for exp_ref, _ in sorted( - commits, key=lambda x: x[1].commit_time, reverse=True - ): - ref_info = ExpRefInfo.from_ref(exp_ref) - assert ref_info.baseline_sha == rev - _collect_experiment_branch( - res[rev], - repo, - exp_ref, - rev, - param_deps=param_deps, - running=running, - onerror=onerror, - ) - - if not hide_failed: - update_new(res, failed_experiments) + update_new(res, failed_experiments) update_new(res, active_experiment) - if not hide_queued: - update_new(res, queued_experiment) + update_new(res, queued_experiment) if not sha_only: get_names(repo, res) diff --git a/tests/func/experiments/test_show.py b/tests/func/experiments/test_show.py index 59efcdd97..0171e7b6a 100644 --- a/tests/func/experiments/test_show.py +++ b/tests/func/experiments/test_show.py @@ -1,10 +1,12 @@ import logging import os +import random from datetime import datetime from unittest.mock import ANY import pytest from funcy import first, get_in +from scmrepo.exceptions import SCMError from dvc.cli import main from dvc.repo.experiments.executor.base import ( @@ -972,3 +974,122 @@ def test_show_sorted_deps(tmp_dir, dvc, scm, capsys): assert main(["exp", "show", "--csv"]) == 0 cap = capsys.readouterr() assert "a,b,c,z" in cap.out + + +@pytest.mark.vscode +def test_show_queued_error(tmp_dir, scm, dvc, exp_stage, mocker): + baseline_rev = scm.get_rev() + + dvc.experiments.run( + exp_stage.addressing, params=["foo=2"], queue=True, name="test_name" + ) + exp_rev_2 = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}") + commit_2 = scm.resolve_commit(exp_rev_2) + + dvc.experiments.run(exp_stage.addressing, params=["foo=3"], queue=True) + exp_rev_3 = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}") + + def resolve_commit(rev): + if rev == exp_rev_3: + raise SCMError + else: + return commit_2 + + mocker.patch.object( + scm, + "resolve_commit", + side_effect=mocker.MagicMock(side_effect=resolve_commit), + ) + + results = dvc.experiments.show()[baseline_rev] + assert len(results) == 2 + exp_2 = results[exp_rev_2]["data"] + assert exp_2["status"] == "Queued" + assert exp_2["params"]["params.yaml"] == {"data": {"foo": 2}} + + +@pytest.mark.vscode +def test_show_completed_error(tmp_dir, scm, dvc, exp_stage, mocker): + baseline_rev = scm.get_rev() + + result_2 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"]) + exp_rev_2 = first(result_2) + commit_2 = scm.resolve_commit(exp_rev_2) + result_3 = dvc.experiments.run(exp_stage.addressing, params=["foo=3"]) + exp_rev_3 = first(result_3) + + def resolve_commit(rev): + if rev == exp_rev_3: + raise SCMError + else: + return commit_2 + + mocker.patch.object( + scm, + "resolve_commit", + side_effect=mocker.MagicMock(side_effect=resolve_commit), + ) + experiments = dvc.experiments.show()[baseline_rev] + assert len(experiments) == 2 + assert exp_rev_2 in experiments + + +@pytest.mark.vscode +def test_show_checkpoint_error(tmp_dir, scm, dvc, checkpoint_stage, mocker): + baseline_rev = scm.get_rev() + results = dvc.experiments.run( + checkpoint_stage.addressing, params=["foo=2"] + ) + exp_rev = first(results) + exp_ref = str(first(exp_refs_by_rev(scm, exp_rev))) + + results = dvc.experiments.show()[baseline_rev] + assert len(results) == checkpoint_stage.iterations + 1 + + checkpoints = {} + for rev in results: + if rev != "baseline": + checkpoints[rev] = scm.resolve_commit(rev) + checkpoints[exp_ref] = scm.resolve_commit(exp_ref) + checkpoints[baseline_rev] = scm.resolve_commit(baseline_rev) + + failed_rev = random.choice(list(checkpoints.keys())) + + def resolve_commit(rev): + if rev == failed_rev: + raise SCMError + return checkpoints[rev] + + mocker.patch.object( + scm, + "resolve_commit", + side_effect=mocker.MagicMock(side_effect=resolve_commit), + ) + results = dvc.experiments.show()[baseline_rev] + assert len(results) == 1 + + +@pytest.mark.vscode +def test_show_baseline_error(tmp_dir, scm, dvc, exp_stage, mocker): + baseline_rev = scm.get_rev() + branch = scm.active_branch() + + result_2 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"]) + exp_rev_2 = first(result_2) + commit_2 = scm.resolve_commit(exp_rev_2) + + def resolve_commit(rev): + if rev == baseline_rev: + raise SCMError + else: + return commit_2 + + mocker.patch.object( + scm, + "resolve_commit", + side_effect=mocker.MagicMock(side_effect=resolve_commit), + ) + experiments = dvc.experiments.show()[baseline_rev] + assert len(experiments) == 1 + assert experiments["baseline"]["data"] == {"name": branch} + assert isinstance(experiments["baseline"]["error"], SCMError)