From 8ae01de0eb19bbc5f3d604c53ad03e2e7e0345b4 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 1 Aug 2019 23:35:21 +0300 Subject: [PATCH 01/21] metrics: allow multiple targets for show command Fix #1497 --- dvc/command/metrics.py | 17 ++++++----- dvc/exceptions.py | 7 +++-- dvc/repo/metrics/show.py | 40 +++++++++++++------------ tests/func/test_metrics.py | 60 +++++++++++++++++++++++++------------- 4 files changed, 75 insertions(+), 49 deletions(-) diff --git a/dvc/command/metrics.py b/dvc/command/metrics.py index 39c53f5b8..748b3bff7 100644 --- a/dvc/command/metrics.py +++ b/dvc/command/metrics.py @@ -35,13 +35,11 @@ def show_metrics(metrics, all_branches=False, all_tags=False): class CmdMetricsShow(CmdBase): def run(self): - typ = self.args.type - xpath = self.args.xpath try: metrics = self.repo.metrics.show( - self.args.path, - typ=typ, - xpath=xpath, + self.args.targets, + typ=self.args.type, + xpath=self.args.xpath, all_branches=self.args.all_branches, all_tags=self.args.all_tags, recursive=self.args.recursive, @@ -121,7 +119,10 @@ def add_parser(subparsers, parent_parser): formatter_class=argparse.RawDescriptionHelpFormatter, ) metrics_show_parser.add_argument( - "path", nargs="?", help="Path to a metric file or a directory." + "targets", + nargs="*", + help="Metric files or directories (see -R) to show " + "(leave empty to display all)", ) metrics_show_parser.add_argument( "-t", @@ -155,8 +156,8 @@ def add_parser(subparsers, parent_parser): action="store_true", default=False, help=( - "If path is a directory, recursively search and process metric " - "files in path." + "If any target is a directory, recursively search and process " + "metric files." ), ) metrics_show_parser.set_defaults(func=CmdMetricsShow) diff --git a/dvc/exceptions.py b/dvc/exceptions.py index da96fd0f9..508b1fb05 100644 --- a/dvc/exceptions.py +++ b/dvc/exceptions.py @@ -196,10 +196,11 @@ class ReproductionError(DvcException): class BadMetricError(DvcException): - def __init__(self, path): + def __init__(self, paths): super(BadMetricError, self).__init__( - "'{}' does not exist, not a metric or is malformed".format( - relpath(path) + "the following outputs do not exist, are not a metric file " + "or are malformed: {paths}".format( + paths=", ".join(relpath(path) for path in paths) ) ) diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index 42ceab3d0..e7f61024d 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -152,15 +152,16 @@ def _read_metric(fd, typ=None, xpath=None, fname=None, branch=None): return None -def _collect_metrics(repo, path, recursive, typ, xpath, branch): +def _collect_metrics(repo, targets, recursive, typ, xpath, branch): """Gather all the metric outputs. Args: - path (str): Path to a metric file or a directory. + targets (list): List of paths to metric files or directories. + If there's no target, collect all metrics. recursive (bool): If path is a directory, do a recursive search for metrics on the given path. - typ (str): The type of metric to search for, could be one of the - following (raw|json|tsv|htsv|csv|hcsv). + typ (str): The type that will be used to interpret the metric file, + one of the followings - (raw|json|tsv|htsv|csv|hcsv). xpath (str): Path to search for. branch (str): Branch to look up for metrics. @@ -171,20 +172,21 @@ def _collect_metrics(repo, path, recursive, typ, xpath, branch): - xpath: """ outs = [out for stage in repo.stages() for out in stage.outs] + found = [] - if path: - try: - outs = repo.find_outs_by_path(path, outs=outs, recursive=recursive) - except OutputNotFoundError: - logger.debug( - "DVC-file not for found for '{}' in branch '{}'".format( - path, branch + if targets: + for target in targets: + try: + found += repo.find_outs_by_path( + target, outs=outs, recursive=recursive ) - ) - return [] + except OutputNotFoundError: + pass + else: + found = outs res = [] - for o in outs: + for o in found: if not o.metric: continue @@ -259,7 +261,7 @@ def _read_metrics(repo, metrics, branch): def show( repo, - path=None, + targets=None, typ=None, xpath=None, all_branches=False, @@ -269,14 +271,16 @@ def show( res = {} for branch in repo.brancher(all_branches=all_branches, all_tags=all_tags): - entries = _collect_metrics(repo, path, recursive, typ, xpath, branch) + entries = _collect_metrics( + repo, targets, recursive, typ, xpath, branch + ) metrics = _read_metrics(repo, entries, branch) if metrics: res[branch] = metrics if not res: - if path: - raise BadMetricError(path) + if targets: + raise BadMetricError(targets) raise NoMetricsError() return res diff --git a/tests/func/test_metrics.py b/tests/func/test_metrics.py index e7892f506..a452dabe9 100644 --- a/tests/func/test_metrics.py +++ b/tests/func/test_metrics.py @@ -85,14 +85,14 @@ class TestMetricsBase(TestDvcGit): class TestMetrics(TestMetricsBase): def test_show(self): - ret = self.dvc.metrics.show("metric", all_branches=True) + ret = self.dvc.metrics.show(["metric"], all_branches=True) self.assertEqual(len(ret), 3) self.assertEqual(ret["foo"]["metric"], "foo") self.assertEqual(ret["bar"]["metric"], "bar") self.assertEqual(ret["baz"]["metric"], "baz") ret = self.dvc.metrics.show( - "metric_json", typ="json", xpath="branch", all_branches=True + ["metric_json"], typ="json", xpath="branch", all_branches=True ) self.assertEqual(len(ret), 3) self.assertSequenceEqual(ret["foo"]["metric_json"], ["foo"]) @@ -100,7 +100,7 @@ class TestMetrics(TestMetricsBase): self.assertSequenceEqual(ret["baz"]["metric_json"], ["baz"]) ret = self.dvc.metrics.show( - "metric_tsv", typ="tsv", xpath="0,0", all_branches=True + ["metric_tsv"], typ="tsv", xpath="0,0", all_branches=True ) self.assertEqual(len(ret), 3) self.assertSequenceEqual(ret["foo"]["metric_tsv"], ["foo"]) @@ -108,7 +108,7 @@ class TestMetrics(TestMetricsBase): self.assertSequenceEqual(ret["baz"]["metric_tsv"], ["baz"]) ret = self.dvc.metrics.show( - "metric_htsv", typ="htsv", xpath="0,branch", all_branches=True + ["metric_htsv"], typ="htsv", xpath="0,branch", all_branches=True ) self.assertEqual(len(ret), 3) self.assertSequenceEqual(ret["foo"]["metric_htsv"], ["foo"]) @@ -116,7 +116,7 @@ class TestMetrics(TestMetricsBase): self.assertSequenceEqual(ret["baz"]["metric_htsv"], ["baz"]) ret = self.dvc.metrics.show( - "metric_csv", typ="csv", xpath="0,0", all_branches=True + ["metric_csv"], typ="csv", xpath="0,0", all_branches=True ) self.assertEqual(len(ret), 3) self.assertSequenceEqual(ret["foo"]["metric_csv"], ["foo"]) @@ -124,7 +124,7 @@ class TestMetrics(TestMetricsBase): self.assertSequenceEqual(ret["baz"]["metric_csv"], ["baz"]) ret = self.dvc.metrics.show( - "metric_hcsv", typ="hcsv", xpath="0,branch", all_branches=True + ["metric_hcsv"], typ="hcsv", xpath="0,branch", all_branches=True ) self.assertEqual(len(ret), 3) self.assertSequenceEqual(ret["foo"]["metric_hcsv"], ["foo"]) @@ -132,7 +132,7 @@ class TestMetrics(TestMetricsBase): self.assertSequenceEqual(ret["baz"]["metric_hcsv"], ["baz"]) ret = self.dvc.metrics.show( - "metric_json_ext", + ["metric_json_ext"], typ="json", xpath="$.metrics[?(@.deviation_mse<0.30) & (@.value_mse>0.4)]", all_branches=True, @@ -153,7 +153,7 @@ class TestMetrics(TestMetricsBase): def test_unknown_type_ignored(self): ret = self.dvc.metrics.show( - "metric_hcsv", typ="unknown", xpath="0,branch", all_branches=True + ["metric_hcsv"], typ="unknown", xpath="0,branch", all_branches=True ) self.assertEqual(len(ret), 3) for b in ["foo", "bar", "baz"]: @@ -161,7 +161,7 @@ class TestMetrics(TestMetricsBase): def test_type_case_normalized(self): ret = self.dvc.metrics.show( - "metric_hcsv", typ=" hCSV ", xpath="0,branch", all_branches=True + ["metric_hcsv"], typ=" hCSV ", xpath="0,branch", all_branches=True ) self.assertEqual(len(ret), 3) for b in ["foo", "bar", "baz"]: @@ -169,7 +169,7 @@ class TestMetrics(TestMetricsBase): def test_xpath_is_empty(self): ret = self.dvc.metrics.show( - "metric_json", typ="json", xpath="", all_branches=True + ["metric_json"], typ="json", xpath="", all_branches=True ) self.assertEqual(len(ret), 3) for b in ["foo", "bar", "baz"]: @@ -177,7 +177,7 @@ class TestMetrics(TestMetricsBase): def test_xpath_is_none(self): ret = self.dvc.metrics.show( - "metric_json", typ="json", xpath=None, all_branches=True + ["metric_json"], typ="json", xpath=None, all_branches=True ) self.assertEqual(len(ret), 3) for b in ["foo", "bar", "baz"]: @@ -185,7 +185,7 @@ class TestMetrics(TestMetricsBase): def test_xpath_all_columns(self): ret = self.dvc.metrics.show( - "metric_hcsv", typ="hcsv ", xpath="0,", all_branches=True + ["metric_hcsv"], typ="hcsv ", xpath="0,", all_branches=True ) self.assertEqual(len(ret), 3) for b in ["foo", "bar", "baz"]: @@ -193,7 +193,7 @@ class TestMetrics(TestMetricsBase): def test_xpath_all_rows(self): ret = self.dvc.metrics.show( - "metric_csv", typ="csv", xpath=",0", all_branches=True + ["metric_csv"], typ="csv", xpath=",0", all_branches=True ) self.assertEqual(len(ret), 3) for b in ["foo", "bar", "baz"]: @@ -201,7 +201,7 @@ class TestMetrics(TestMetricsBase): def test_xpath_all(self): ret = self.dvc.metrics.show( - "metric_csv", typ="csv", xpath=",", all_branches=True + ["metric_csv"], typ="csv", xpath=",", all_branches=True ) self.assertEqual(len(ret), 3) for b in ["foo", "bar", "baz"]: @@ -209,7 +209,7 @@ class TestMetrics(TestMetricsBase): def test_xpath_all_with_header(self): ret = self.dvc.metrics.show( - "metric_hcsv", typ="hcsv", xpath=",", all_branches=True + ["metric_hcsv"], typ="hcsv", xpath=",", all_branches=True ) self.assertEqual(len(ret), 3) for b in ["foo", "bar", "baz"]: @@ -413,10 +413,12 @@ class TestMetricsRecursive(TestDvcGit): def test(self): with self.assertRaises(BadMetricError): - self.dvc.metrics.show("nested", all_branches=True, recursive=False) + self.dvc.metrics.show( + ["nested"], all_branches=True, recursive=False + ) ret = self.dvc.metrics.show( - "nested", all_branches=True, recursive=True + ["nested"], all_branches=True, recursive=True ) self.assertEqual(len(ret), 1) self.assertEqual( @@ -697,7 +699,7 @@ class TestCachedMetrics(TestDvcGit): self.dvc = DvcRepo(".") res = self.dvc.metrics.show( - "metrics.json", all_branches=True, typ="json", xpath="metrics" + ["metrics.json"], all_branches=True, typ="json", xpath="metrics" ) self.assertEqual( @@ -711,7 +713,7 @@ class TestCachedMetrics(TestDvcGit): ) res = self.dvc.metrics.show( - "", all_branches=True, typ="json", xpath="metrics" + all_branches=True, typ="json", xpath="metrics" ) self.assertEqual( @@ -777,7 +779,9 @@ class TestMetricsType(TestDvcGit): self._do_show(file_name, xpath) def _do_show(self, file_name, xpath): - ret = self.dvc.metrics.show(file_name, xpath=xpath, all_branches=True) + ret = self.dvc.metrics.show( + [file_name], xpath=xpath, all_branches=True + ) self.assertEqual(len(ret), 3) for branch in self.branches: if isinstance(ret[branch][file_name], list): @@ -844,3 +848,19 @@ def test_show_xpath_should_override_stage_xpath(dvc_repo): result = dvc_repo.metrics.show(xpath="m1") assert result == {"": {metric_file: [0.1]}} + + +def test_show_multiple_outputs(dvc_repo, caplog): + with open("1.json", "w") as fobj: + json.dump({"AUC": 1}, fobj) + + with open("2.json", "w") as fobj: + json.dump({"AUC": 2}, fobj) + + dvc_repo.run(cmd="", overwrite=True, metrics=["1.json"]) + dvc_repo.run(cmd="", overwrite=True, metrics=["2.json"]) + + with caplog.at_level(logging.INFO, logger="dvc"): + assert 0 == main(["metrics", "show", "1.json", "2.json"]) + assert '1.json: {"AUC": 1}' in caplog.text + assert '2.json: {"AUC": 2}' in caplog.text From 44f1cc62009eee2289d82c2b8e1fa8877272253e Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Mon, 5 Aug 2019 15:27:17 -0500 Subject: [PATCH 02/21] bring back debug message --- dvc/repo/metrics/show.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index e7f61024d..9c6f4d45a 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -181,6 +181,11 @@ def _collect_metrics(repo, targets, recursive, typ, xpath, branch): target, outs=outs, recursive=recursive ) except OutputNotFoundError: + logger.debug( + "DVC-file not for found for '{}' in branch '{}'".format( + target, branch + ) + ) pass else: found = outs From 21a2d3c6ffbf4872676df59283238ec3e5cb5c91 Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Mon, 5 Aug 2019 15:30:28 -0500 Subject: [PATCH 03/21] remove no-op Co-Authored-By: Ruslan Kuprieiev --- dvc/repo/metrics/show.py | 1 - 1 file changed, 1 deletion(-) diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index 9c6f4d45a..b7c2645bb 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -186,7 +186,6 @@ def _collect_metrics(repo, targets, recursive, typ, xpath, branch): target, branch ) ) - pass else: found = outs From d4ef110794538f69ed5595db2efd75ccfe7ddee5 Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Tue, 6 Aug 2019 21:10:48 -0500 Subject: [PATCH 04/21] exit with 1 if targets were not found --- dvc/command/metrics.py | 19 +++++++++++++++++++ dvc/exceptions.py | 10 ---------- dvc/repo/metrics/show.py | 6 ++---- tests/func/test_metrics.py | 10 +++++----- 4 files changed, 26 insertions(+), 19 deletions(-) diff --git a/dvc/command/metrics.py b/dvc/command/metrics.py index 748b3bff7..73bd9ae9f 100644 --- a/dvc/command/metrics.py +++ b/dvc/command/metrics.py @@ -2,6 +2,7 @@ from __future__ import unicode_literals import argparse import logging +import itertools from dvc.exceptions import DvcException from dvc.command.base import CmdBase, fix_subparsers, append_doc_link @@ -46,6 +47,24 @@ class CmdMetricsShow(CmdBase): ) show_metrics(metrics, self.args.all_branches, self.args.all_tags) + + found = set( + itertools.chain.from_iterable( + metric.keys() for metric in metrics.values() + ) + ) + + missing = set(self.args.targets) - found + + if missing: + logger.error( + "the following metrics do not exists, " + "are not metric files or are malformed: {paths}".format( + paths=", ".join("'{}'".format(x) for x in missing) + ) + ) + return 1 + except DvcException: logger.exception("failed to show metrics") return 1 diff --git a/dvc/exceptions.py b/dvc/exceptions.py index 508b1fb05..e8fd2df29 100644 --- a/dvc/exceptions.py +++ b/dvc/exceptions.py @@ -195,16 +195,6 @@ class ReproductionError(DvcException): super(ReproductionError, self).__init__(msg, cause=ex) -class BadMetricError(DvcException): - def __init__(self, paths): - super(BadMetricError, self).__init__( - "the following outputs do not exist, are not a metric file " - "or are malformed: {paths}".format( - paths=", ".join(relpath(path) for path in paths) - ) - ) - - class NoMetricsError(DvcException): def __init__(self): super(NoMetricsError, self).__init__( diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index b7c2645bb..0facfa8ae 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -8,7 +8,7 @@ import logging from jsonpath_ng.ext import parse -from dvc.exceptions import OutputNotFoundError, BadMetricError, NoMetricsError +from dvc.exceptions import OutputNotFoundError, NoMetricsError from dvc.utils.compat import builtin_str, open, StringIO, csv_reader NO_METRICS_FILE_AT_REFERENCE_WARNING = ( @@ -282,9 +282,7 @@ def show( if metrics: res[branch] = metrics - if not res: - if targets: - raise BadMetricError(targets) + if not res and not targets: raise NoMetricsError() return res diff --git a/tests/func/test_metrics.py b/tests/func/test_metrics.py index a452dabe9..46440ffdf 100644 --- a/tests/func/test_metrics.py +++ b/tests/func/test_metrics.py @@ -7,7 +7,7 @@ import logging from dvc.repo import Repo as DvcRepo from dvc.main import main -from dvc.exceptions import DvcException, BadMetricError, NoMetricsError +from dvc.exceptions import DvcException, NoMetricsError from dvc.repo.metrics.show import NO_METRICS_FILE_AT_REFERENCE_WARNING from dvc.stage import Stage from dvc.utils import relpath @@ -412,10 +412,10 @@ class TestMetricsRecursive(TestDvcGit): self.dvc.scm.checkout("master") def test(self): - with self.assertRaises(BadMetricError): - self.dvc.metrics.show( - ["nested"], all_branches=True, recursive=False - ) + ret = self.dvc.metrics.show( + ["nested"], all_branches=True, recursive=False + ) + self.assertEqual(len(ret), 0) ret = self.dvc.metrics.show( ["nested"], all_branches=True, recursive=True From 0e46edd9bb46d0464045876fd4df01cb311d4cad Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Fri, 9 Aug 2019 21:54:28 -0500 Subject: [PATCH 05/21] fail silently but with a more sound error message * Add a `_missing` key to the return value of `repo.metrics.show` to include targets not recognized as metrics or not found. --- dvc/command/metrics.py | 28 ++++++------------- dvc/exceptions.py | 10 +++++++ dvc/repo/metrics/show.py | 56 ++++++++++++++++++++++---------------- tests/func/test_metrics.py | 2 +- 4 files changed, 52 insertions(+), 44 deletions(-) diff --git a/dvc/command/metrics.py b/dvc/command/metrics.py index 73bd9ae9f..cfd1857ad 100644 --- a/dvc/command/metrics.py +++ b/dvc/command/metrics.py @@ -2,9 +2,8 @@ from __future__ import unicode_literals import argparse import logging -import itertools -from dvc.exceptions import DvcException +from dvc.exceptions import DvcException, BadMetricError from dvc.command.base import CmdBase, fix_subparsers, append_doc_link @@ -17,6 +16,10 @@ def show_metrics(metrics, all_branches=False, all_tags=False): metrics (list): Where each element is either a `list` if an xpath was specified, otherwise a `str` """ + # When `metrics` contains a `_missing` key, it means that some files + # specified as `targets` in `repo.metrics.show` didn't contain any metrics. + missing = metrics.pop("_missing", None) + for branch, val in metrics.items(): if all_branches or all_tags: logger.info("{branch}:".format(branch=branch)) @@ -33,6 +36,9 @@ def show_metrics(metrics, all_branches=False, all_tags=False): else: logger.info("\t{}: {}".format(fname, metric)) + if missing: + raise BadMetricError(missing) + class CmdMetricsShow(CmdBase): def run(self): @@ -47,24 +53,6 @@ class CmdMetricsShow(CmdBase): ) show_metrics(metrics, self.args.all_branches, self.args.all_tags) - - found = set( - itertools.chain.from_iterable( - metric.keys() for metric in metrics.values() - ) - ) - - missing = set(self.args.targets) - found - - if missing: - logger.error( - "the following metrics do not exists, " - "are not metric files or are malformed: {paths}".format( - paths=", ".join("'{}'".format(x) for x in missing) - ) - ) - return 1 - except DvcException: logger.exception("failed to show metrics") return 1 diff --git a/dvc/exceptions.py b/dvc/exceptions.py index e8fd2df29..bddf83ec9 100644 --- a/dvc/exceptions.py +++ b/dvc/exceptions.py @@ -195,6 +195,16 @@ class ReproductionError(DvcException): super(ReproductionError, self).__init__(msg, cause=ex) +class BadMetricError(DvcException): + def __init__(self, paths): + super(BadMetricError, self).__init__( + "the following metrics do not exists, " + "are not metric files or are malformed: {paths}".format( + paths=", ".join("'{}'".format(path) for path in paths) + ) + ) + + class NoMetricsError(DvcException): def __init__(self): super(NoMetricsError, self).__init__( diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index 0facfa8ae..11935e899 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -152,12 +152,11 @@ def _read_metric(fd, typ=None, xpath=None, fname=None, branch=None): return None -def _collect_metrics(repo, targets, recursive, typ, xpath, branch): +def _collect_metrics(repo, path, recursive, typ, xpath, branch): """Gather all the metric outputs. Args: - targets (list): List of paths to metric files or directories. - If there's no target, collect all metrics. + path (str): Path to a metric file or a directory. recursive (bool): If path is a directory, do a recursive search for metrics on the given path. typ (str): The type that will be used to interpret the metric file, @@ -172,25 +171,20 @@ def _collect_metrics(repo, targets, recursive, typ, xpath, branch): - xpath: """ outs = [out for stage in repo.stages() for out in stage.outs] - found = [] - if targets: - for target in targets: - try: - found += repo.find_outs_by_path( - target, outs=outs, recursive=recursive + if path: + try: + outs = repo.find_outs_by_path(path, outs=outs, recursive=recursive) + except OutputNotFoundError: + logger.debug( + "DVC-file not for found for '{}' in branch '{}'".format( + path, branch ) - except OutputNotFoundError: - logger.debug( - "DVC-file not for found for '{}' in branch '{}'".format( - target, branch - ) - ) - else: - found = outs + ) + return [] res = [] - for o in found: + for o in outs: if not o.metric: continue @@ -273,16 +267,32 @@ def show( recursive=False, ): res = {} + found = set() + + if not targets: + # Iterate once to call `_collect_metrics` on all the stages + targets = [None] for branch in repo.brancher(all_branches=all_branches, all_tags=all_tags): - entries = _collect_metrics( - repo, targets, recursive, typ, xpath, branch - ) - metrics = _read_metrics(repo, entries, branch) + metrics = {} + + for p in targets: + entries = _collect_metrics(repo, p, recursive, typ, xpath, branch) + metric = _read_metrics(repo, entries, branch) + + if metric: + found.add(p) + metrics.update(metric) + if metrics: res[branch] = metrics - if not res and not targets: + if not res and not any(targets): raise NoMetricsError() + missing = set(targets) - found + + if missing: + res["_missing"] = missing + return res diff --git a/tests/func/test_metrics.py b/tests/func/test_metrics.py index 46440ffdf..9d0ba2c70 100644 --- a/tests/func/test_metrics.py +++ b/tests/func/test_metrics.py @@ -415,7 +415,7 @@ class TestMetricsRecursive(TestDvcGit): ret = self.dvc.metrics.show( ["nested"], all_branches=True, recursive=False ) - self.assertEqual(len(ret), 0) + self.assertEqual(len(ret), 1) ret = self.dvc.metrics.show( ["nested"], all_branches=True, recursive=True From fa04c275803d521b0bb9edaad5868526a3bcd386 Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Fri, 9 Aug 2019 22:06:32 -0500 Subject: [PATCH 06/21] add tests for recursive operations and file not found --- tests/func/test_metrics.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/func/test_metrics.py b/tests/func/test_metrics.py index 9d0ba2c70..ba19bf771 100644 --- a/tests/func/test_metrics.py +++ b/tests/func/test_metrics.py @@ -857,10 +857,32 @@ def test_show_multiple_outputs(dvc_repo, caplog): with open("2.json", "w") as fobj: json.dump({"AUC": 2}, fobj) + os.mkdir("metrics") + with open("metrics/3.json", "w") as fobj: + json.dump({"AUC": 3}, fobj) + dvc_repo.run(cmd="", overwrite=True, metrics=["1.json"]) dvc_repo.run(cmd="", overwrite=True, metrics=["2.json"]) + dvc_repo.run(cmd="", overwrite=True, metrics=["metrics/3.json"]) with caplog.at_level(logging.INFO, logger="dvc"): assert 0 == main(["metrics", "show", "1.json", "2.json"]) assert '1.json: {"AUC": 1}' in caplog.text assert '2.json: {"AUC": 2}' in caplog.text + + caplog.clear() + + with caplog.at_level(logging.INFO, logger="dvc"): + assert 0 == main(["metrics", "show", "-R", "1.json", "metrics"]) + assert '1.json: {"AUC": 1}' in caplog.text + assert 'metrics/3.json: {"AUC": 3}' in caplog.text + + caplog.clear() + + with caplog.at_level(logging.INFO, logger="dvc"): + assert 1 == main(["metrics", "show", "1.json", "not-found"]) + assert '1.json: {"AUC": 1}' in caplog.text + assert ( + "the following metrics do not exists, " + "are not metric files or are malformed: 'not-found'" + ) in caplog.text From 163d872c92e98357d3336804f328071beb7de3ee Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Fri, 9 Aug 2019 22:46:25 -0500 Subject: [PATCH 07/21] fix tests for windows --- tests/func/test_metrics.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/func/test_metrics.py b/tests/func/test_metrics.py index ba19bf771..9f0b2a00f 100644 --- a/tests/func/test_metrics.py +++ b/tests/func/test_metrics.py @@ -875,7 +875,7 @@ def test_show_multiple_outputs(dvc_repo, caplog): with caplog.at_level(logging.INFO, logger="dvc"): assert 0 == main(["metrics", "show", "-R", "1.json", "metrics"]) assert '1.json: {"AUC": 1}' in caplog.text - assert 'metrics/3.json: {"AUC": 3}' in caplog.text + assert '3.json: {"AUC": 3}' in caplog.text caplog.clear() From a17caefd508103cb498a3b20ba2010deb0de396e Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Sat, 10 Aug 2019 11:37:35 -0500 Subject: [PATCH 08/21] integrate @efiop 's suggestions --- dvc/command/metrics.py | 4 ++-- dvc/repo/metrics/show.py | 10 ++++++---- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/dvc/command/metrics.py b/dvc/command/metrics.py index cfd1857ad..16e1287a2 100644 --- a/dvc/command/metrics.py +++ b/dvc/command/metrics.py @@ -16,9 +16,9 @@ def show_metrics(metrics, all_branches=False, all_tags=False): metrics (list): Where each element is either a `list` if an xpath was specified, otherwise a `str` """ - # When `metrics` contains a `_missing` key, it means that some files + # When `metrics` contains a `None` key, it means that some files # specified as `targets` in `repo.metrics.show` didn't contain any metrics. - missing = metrics.pop("_missing", None) + missing = metrics.pop(None, None) for branch, val in metrics.items(): if all_branches or all_tags: diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index 11935e899..cd1ab7459 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -276,12 +276,14 @@ def show( for branch in repo.brancher(all_branches=all_branches, all_tags=all_tags): metrics = {} - for p in targets: - entries = _collect_metrics(repo, p, recursive, typ, xpath, branch) + for target in targets: + entries = _collect_metrics( + repo, target, recursive, typ, xpath, branch + ) metric = _read_metrics(repo, entries, branch) if metric: - found.add(p) + found.add(target) metrics.update(metric) if metrics: @@ -293,6 +295,6 @@ def show( missing = set(targets) - found if missing: - res["_missing"] = missing + res[None] = missing return res From 71ab380d036208d0b96868e1bde06394c93142b2 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 11 Aug 2019 03:00:08 +0300 Subject: [PATCH 09/21] utils: remove: catch EPERM when chmoding a file Fix https://github.com/iterative/dvc/pull/2298#issuecomment-520184196 --- dvc/utils/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dvc/utils/__init__.py b/dvc/utils/__init__.py index af7356575..c5d1fbbc9 100644 --- a/dvc/utils/__init__.py +++ b/dvc/utils/__init__.py @@ -193,8 +193,8 @@ def _chmod(func, p, excinfo): try: os.chmod(p, perm) except OSError as exc: - # broken symlink - if exc.errno != errno.ENOENT: + # broken symlink or file is not owned by us + if exc.errno not in [errno.ENOENT, errno.EPERM]: raise func(p) From 2f30bddfab78398f93f27e0da5af0b460df7c7af Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 10 Aug 2019 05:54:26 +0300 Subject: [PATCH 10/21] travis: use azurite v3 Trying to fix azure error on our ci: https://travis-ci.com/iterative/dvc/jobs/224085273 Workaround for #2387 Signed-off-by: Ruslan Kuprieiev --- scripts/ci/before_install.sh | 2 +- scripts/ci/install.sh | 7 ++++++- scripts/ci/install_azurite.sh | 18 +++++++++++------- 3 files changed, 18 insertions(+), 9 deletions(-) diff --git a/scripts/ci/before_install.sh b/scripts/ci/before_install.sh index 19645bcad..4f48cc6f6 100644 --- a/scripts/ci/before_install.sh +++ b/scripts/ci/before_install.sh @@ -31,7 +31,7 @@ if [[ "$TRAVIS_BUILD_STAGE_NAME" == "Test" ]]; then ssh 0.0.0.0 ls &> /dev/null fi - if [ "$TRAVIS_OS_NAME" == "linux" ] && [ "$TRAVIS_PULL_REQUEST" == "false" ]; then + if [ "$TRAVIS_OS_NAME" == "linux" ]; then bash "$scriptdir/install_azurite.sh" bash "$scriptdir/install_oss.sh" bash "$scriptdir/install_hadoop.sh" diff --git a/scripts/ci/install.sh b/scripts/ci/install.sh index 658bf1fd1..86d10fef0 100644 --- a/scripts/ci/install.sh +++ b/scripts/ci/install.sh @@ -10,7 +10,12 @@ scriptdir="$(dirname $0)" # work or not. $scriptdir/retry.sh pip install --upgrade pip setuptools wheel $scriptdir/retry.sh pip install .[all,tests] -$scriptdir/retry.sh pip install psutil +# Installing specific packages to workaround some bugs. Please see [1] and [2] +# +# [1] https://github.com/iterative/dvc/issues/2284 +# [2] https://github.com/iterative/dvc/issues/2387 +$scriptdir/retry.sh pip uninstall -y azure-storage-blob +$scriptdir/retry.sh pip install psutil azure-storage-blob==1.5.0 git config --global user.email "dvctester@example.com" git config --global user.name "DVC Tester" diff --git a/scripts/ci/install_azurite.sh b/scripts/ci/install_azurite.sh index 094864db4..ad6d23e71 100755 --- a/scripts/ci/install_azurite.sh +++ b/scripts/ci/install_azurite.sh @@ -1,16 +1,20 @@ #!/usr/bin/env bash -set -euo pipefail +set -e +set -x # install docker -export DEBIAN_FRONTEND=noninteractive -curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add - -sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable" -sudo apt-get update -sudo apt-get install -y docker-ce + export DEBIAN_FRONTEND=noninteractive + curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add - + sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable" + sudo apt-get update + sudo apt-get install -y docker-ce # run azurite -sudo docker run --restart always -e executable=blob -p 10000:10000 --tmpfs /opt/azurite/folder -d arafato/azurite:2.6.5 +sudo docker run -d --restart always -e executable=blob -p 10000:10000 \ + --tmpfs /opt/azurite/folder \ + mcr.microsoft.com/azure-storage/azurite:3.1.2-preview \ + azurite -l /data --blobHost 0.0.0.0 # save secrets echo "export AZURE_STORAGE_CONTAINER_NAME='travis-tests'" >> env.sh From 71abef74f542ae5832c070089a3be132859f7c27 Mon Sep 17 00:00:00 2001 From: pared Date: Wed, 7 Aug 2019 12:34:37 +0200 Subject: [PATCH 11/21] remote: azure: implement batch_exists --- dvc/remote/azure.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 0b8330ae3..b94a67884 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -1,6 +1,7 @@ from __future__ import absolute_import from __future__ import unicode_literals +import itertools import os import re import logging @@ -154,3 +155,13 @@ class RemoteAZURE(RemoteBASE): path_info.bucket, path_info.path, sas_token=sas_token ) return download_url + + def batch_exists(self, path_infos, callback): + paths = [] + + for path_info in path_infos: + paths.append(self._list_paths(path_info.bucket, path_info.path)) + callback.update(str(path_info)) + + paths = set(itertools.chain.from_iterable(paths)) + return [path_info.path in paths for path_info in path_infos] From 237f3e248b93c690710ff9f0b684c8cce6a3cb82 Mon Sep 17 00:00:00 2001 From: pared Date: Wed, 7 Aug 2019 13:29:47 +0200 Subject: [PATCH 12/21] remote: base: modify cache_exists to use exists instead of batch_exists --- dvc/remote/azure.py | 6 +++++- dvc/remote/base.py | 10 +++------- dvc/remote/gs.py | 12 ------------ dvc/remote/http.py | 9 --------- dvc/remote/oss.py | 4 ++++ dvc/remote/s3.py | 11 ----------- dvc/remote/ssh/__init__.py | 32 -------------------------------- 7 files changed, 12 insertions(+), 72 deletions(-) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index b94a67884..ded3ad935 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -1,12 +1,12 @@ from __future__ import absolute_import from __future__ import unicode_literals -import itertools import os import re import logging from datetime import datetime, timedelta + from dvc.scheme import Schemes try: @@ -138,6 +138,10 @@ class RemoteAZURE(RemoteBASE): from_info.bucket, from_info.path, to_file, progress_callback=cb ) + def exists(self, path_info): + paths = self._list_paths(path_info.bucket, path_info.path) + return any(path_info.path == path for path in paths) + def open(self, path_info, mode="r", encoding=None): get_url = lambda: self._generate_download_url(path_info) # noqa: E731 return open_url(get_url, mode=mode, encoding=encoding) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 9a387774a..30df8696d 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -20,6 +20,7 @@ from dvc.exceptions import ( DvcIgnoreInCollectedDirError, ) from dvc.progress import progress, ProgressCallback +from dvc.utils import LARGE_DIR_SIZE, tmp_fname, move, relpath from dvc.utils import ( LARGE_DIR_SIZE, tmp_fname, @@ -644,15 +645,10 @@ class RemoteBASE(object): """ progress_callback = ProgressCallback(len(checksums)) - def exists_with_progress(chunks): - return self.batch_exists(chunks, callback=progress_callback) - - if self.no_traverse and hasattr(self, "batch_exists"): + if self.no_traverse: with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor: path_infos = [self.checksum_to_path_info(x) for x in checksums] - chunks = to_chunks(path_infos, num_chunks=self.JOBS) - results = executor.map(exists_with_progress, chunks) - in_remote = itertools.chain.from_iterable(results) + in_remote = list(executor.map(self.exists, path_infos)) ret = list(itertools.compress(checksums, in_remote)) progress_callback.finish("") return ret diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index b171b9c4e..4d07226cd 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -1,7 +1,6 @@ from __future__ import unicode_literals import logging -import itertools from funcy import cached_property try: @@ -89,17 +88,6 @@ class RemoteGS(RemoteBASE): paths = set(self._list_paths(path_info.bucket, path_info.path)) return any(path_info.path == path for path in paths) - def batch_exists(self, path_infos, callback): - paths = [] - - for path_info in path_infos: - paths.append(self._list_paths(path_info.bucket, path_info.path)) - callback.update(str(path_info)) - - paths = set(itertools.chain.from_iterable(paths)) - - return [path_info.path in paths for path_info in path_infos] - def _upload(self, from_file, to_info, **_kwargs): bucket = self.gs.bucket(to_info.bucket) blob = bucket.blob(to_info.path) diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 715465293..f377f0ddc 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -65,15 +65,6 @@ class RemoteHTTP(RemoteBASE): def exists(self, path_info): return bool(self._request("HEAD", path_info.url)) - def batch_exists(self, path_infos, callback): - results = [] - - for path_info in path_infos: - results.append(self.exists(path_info)) - callback.update(str(path_info)) - - return results - def _content_length(self, url): return self._request("HEAD", url).headers.get("Content-Length") diff --git a/dvc/remote/oss.py b/dvc/remote/oss.py index 79912edc3..7e1d2313d 100644 --- a/dvc/remote/oss.py +++ b/dvc/remote/oss.py @@ -119,3 +119,7 @@ class RemoteOSS(RemoteBASE): self.oss_service.get_object_to_file( from_info.path, to_file, progress_callback=cb ) + + def exists(self, path_info): + paths = self._list_paths(path_info.bucket, path_info.path) + return any(path_info.path == path for path in paths) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index d9adab21b..757db8980 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -3,7 +3,6 @@ from __future__ import unicode_literals import os import threading import logging -import itertools from funcy import cached_property try: @@ -216,16 +215,6 @@ class RemoteS3(RemoteBASE): paths = self._list_paths(path_info.bucket, path_info.path) return any(path_info.path == path for path in paths) - def batch_exists(self, path_infos, callback): - paths = [] - - for path_info in path_infos: - paths.append(self._list_paths(path_info.bucket, path_info.path)) - callback.update(str(path_info)) - - paths = set(itertools.chain.from_iterable(paths)) - return [path_info.path in paths for path_info in path_infos] - def _upload(self, from_file, to_info, name=None, no_progress_bar=False): total = os.path.getsize(from_file) cb = None if no_progress_bar else Callback(name, total) diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py index 22a7da5aa..8bb5c69f2 100644 --- a/dvc/remote/ssh/__init__.py +++ b/dvc/remote/ssh/__init__.py @@ -3,9 +3,6 @@ from __future__ import unicode_literals import os import getpass import logging -import itertools -import errno -from concurrent.futures import ThreadPoolExecutor import threading try: @@ -15,7 +12,6 @@ except ImportError: import dvc.prompt as prompt from dvc.config import Config -from dvc.utils import to_chunks from dvc.utils.compat import urlparse, StringIO from dvc.remote.base import RemoteBASE from dvc.scheme import Schemes @@ -148,34 +144,6 @@ class RemoteSSH(RemoteBASE): with self.ssh(path_info) as ssh: return ssh.exists(path_info.path) - def batch_exists(self, path_infos, callback): - def _exists(chunk_and_channel): - chunk, channel = chunk_and_channel - ret = [] - for path in chunk: - try: - channel.stat(path) - ret.append(True) - except IOError as exc: - if exc.errno != errno.ENOENT: - raise - ret.append(False) - callback.update(path) - return ret - - with self.ssh(path_infos[0]) as ssh: - channels = ssh.open_max_sftp_channels() - max_workers = len(channels) - - with ThreadPoolExecutor(max_workers=max_workers) as executor: - paths = [path_info.path for path_info in path_infos] - chunks = to_chunks(paths, num_chunks=max_workers) - chunks_and_channels = zip(chunks, channels) - outcome = executor.map(_exists, chunks_and_channels) - results = list(itertools.chain.from_iterable(outcome)) - - return results - def get_file_checksum(self, path_info): if path_info.scheme != self.scheme: raise NotImplementedError From 98de99f5a94f24515b3faa36c9d6b71523aef92a Mon Sep 17 00:00:00 2001 From: pared Date: Wed, 7 Aug 2019 13:31:27 +0200 Subject: [PATCH 13/21] remote: azure: make blob_service cached property --- dvc/remote/azure.py | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index ded3ad935..f51e0931d 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -6,6 +6,7 @@ import re import logging from datetime import datetime, timedelta +from funcy import cached_property from dvc.scheme import Schemes @@ -72,29 +73,27 @@ class RemoteAZURE(RemoteBASE): if not self.connection_string: raise ValueError("azure storage connection string missing") - self.__blob_service = None self.path_info = ( self.path_cls(url) if path else self.path_cls.from_parts(scheme=self.scheme, netloc=bucket) ) - @property + @cached_property def blob_service(self): - if self.__blob_service is None: - logger.debug("URL {}".format(self.path_info)) - logger.debug("Connection string {}".format(self.connection_string)) - self.__blob_service = BlockBlobService( - connection_string=self.connection_string + logger.debug("URL {}".format(self.path_info)) + logger.debug("Connection string {}".format(self.connection_string)) + blob_service = BlockBlobService( + connection_string=self.connection_string + ) + logger.debug("Container name {}".format(self.path_info.bucket)) + try: # verify that container exists + blob_service.list_blobs( + self.path_info.bucket, delimiter="/", num_results=1 ) - logger.debug("Container name {}".format(self.path_info.bucket)) - try: # verify that container exists - self.__blob_service.list_blobs( - self.path_info.bucket, delimiter="/", num_results=1 - ) - except AzureMissingResourceHttpError: - self.__blob_service.create_container(self.path_info.bucket) - return self.__blob_service + except AzureMissingResourceHttpError: + blob_service.create_container(self.path_info.bucket) + return blob_service def remove(self, path_info): if path_info.scheme != self.scheme: From 132b571e7232f2857064ddb1da4b749908b5a305 Mon Sep 17 00:00:00 2001 From: pared Date: Wed, 7 Aug 2019 15:22:56 +0200 Subject: [PATCH 14/21] remote: oss: fix _list_cache_args --- dvc/remote/oss.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dvc/remote/oss.py b/dvc/remote/oss.py index 7e1d2313d..cbf1fe3a7 100644 --- a/dvc/remote/oss.py +++ b/dvc/remote/oss.py @@ -121,5 +121,5 @@ class RemoteOSS(RemoteBASE): ) def exists(self, path_info): - paths = self._list_paths(path_info.bucket, path_info.path) + paths = self._list_paths(path_info.path) return any(path_info.path == path for path in paths) From 394d6ffeeb98e9b3a40b2c31c295e6db8cbf5725 Mon Sep 17 00:00:00 2001 From: pared Date: Wed, 7 Aug 2019 15:49:56 +0200 Subject: [PATCH 15/21] remote: ssh: connection: remove open_max_sftp_channels --- dvc/remote/ssh/connection.py | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py index 8362857b8..9d64ee5b4 100644 --- a/dvc/remote/ssh/connection.py +++ b/dvc/remote/ssh/connection.py @@ -289,15 +289,3 @@ class SSHConnection: def cp(self, src, dest): self.makedirs(posixpath.dirname(dest)) self.execute("cp {} {}".format(src, dest)) - - def open_max_sftp_channels(self): - # If there are more than 1 it means we've already opened max amount - if len(self._sftp_channels) <= 1: - while True: - try: - self._sftp_channels.append(self._ssh.open_sftp()) - except paramiko.ssh_exception.ChannelException: - if not self._sftp_channels: - raise - break - return self._sftp_channels From f224790aababaa488d97e9cf666c4ff64a2a403b Mon Sep 17 00:00:00 2001 From: pawel Date: Fri, 9 Aug 2019 11:56:27 +0200 Subject: [PATCH 16/21] remote: base: remove unnecessary list casting --- dvc/remote/base.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 30df8696d..275481f01 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -648,7 +648,7 @@ class RemoteBASE(object): if self.no_traverse: with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor: path_infos = [self.checksum_to_path_info(x) for x in checksums] - in_remote = list(executor.map(self.exists, path_infos)) + in_remote = executor.map(self.exists, path_infos) ret = list(itertools.compress(checksums, in_remote)) progress_callback.finish("") return ret From 8233f959c795ab6b0d2582dc1e8c45376f6364c8 Mon Sep 17 00:00:00 2001 From: pawel Date: Mon, 12 Aug 2019 11:22:51 +0200 Subject: [PATCH 17/21] remote: ssh: reintroduce batch_exists --- dvc/remote/azure.py | 10 ------- dvc/remote/base.py | 10 +------ dvc/remote/ssh/__init__.py | 57 ++++++++++++++++++++++++++++++++++++ dvc/remote/ssh/connection.py | 12 ++++++++ 4 files changed, 70 insertions(+), 19 deletions(-) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index f51e0931d..6e37545ed 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -158,13 +158,3 @@ class RemoteAZURE(RemoteBASE): path_info.bucket, path_info.path, sas_token=sas_token ) return download_url - - def batch_exists(self, path_infos, callback): - paths = [] - - for path_info in path_infos: - paths.append(self._list_paths(path_info.bucket, path_info.path)) - callback.update(str(path_info)) - - paths = set(itertools.chain.from_iterable(paths)) - return [path_info.path in paths for path_info in path_infos] diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 275481f01..2105372ae 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -20,15 +20,7 @@ from dvc.exceptions import ( DvcIgnoreInCollectedDirError, ) from dvc.progress import progress, ProgressCallback -from dvc.utils import LARGE_DIR_SIZE, tmp_fname, move, relpath -from dvc.utils import ( - LARGE_DIR_SIZE, - tmp_fname, - to_chunks, - move, - relpath, - makedirs, -) +from dvc.utils import LARGE_DIR_SIZE, tmp_fname, move, relpath, makedirs from dvc.state import StateNoop from dvc.path_info import PathInfo, URLInfo diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py index 8bb5c69f2..4202ba61a 100644 --- a/dvc/remote/ssh/__init__.py +++ b/dvc/remote/ssh/__init__.py @@ -1,9 +1,15 @@ from __future__ import unicode_literals +import errno +import itertools import os import getpass import logging import threading +from concurrent.futures import ThreadPoolExecutor + +from dvc.progress import ProgressCallback +from dvc.utils import to_chunks try: import paramiko @@ -208,3 +214,54 @@ class RemoteSSH(RemoteBASE): def makedirs(self, path_info): with self.ssh(path_info) as ssh: ssh.makedirs(path_info.path) + + def batch_exists(self, path_infos, callback): + def _exists(chunk_and_channel): + chunk, channel = chunk_and_channel + ret = [] + for path in chunk: + try: + channel.stat(path) + ret.append(True) + except IOError as exc: + if exc.errno != errno.ENOENT: + raise + ret.append(False) + callback.update(path) + return ret + + with self.ssh(path_infos[0]) as ssh: + channels = ssh.open_max_sftp_channels() + max_workers = len(channels) + + with ThreadPoolExecutor(max_workers=max_workers) as executor: + paths = [path_info.path for path_info in path_infos] + chunks = to_chunks(paths, num_chunks=max_workers) + chunks_and_channels = zip(chunks, channels) + outcome = executor.map(_exists, chunks_and_channels) + results = list(itertools.chain.from_iterable(outcome)) + + return results + + def cache_exists(self, checksums, jobs=None): + """This is older implementation used in remote/base.py + We are reusing it in RemoteSSH, because SSH's batch_exists proved to be + faster than current approach (relying on exists(path_info)) applied in + remote/base. + """ + progress_callback = ProgressCallback(len(checksums)) + + def exists_with_progress(chunks): + return self.batch_exists(chunks, callback=progress_callback) + + if self.no_traverse: + with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor: + path_infos = [self.checksum_to_path_info(x) for x in checksums] + chunks = to_chunks(path_infos, num_chunks=self.JOBS) + results = executor.map(exists_with_progress, chunks) + in_remote = itertools.chain.from_iterable(results) + ret = list(itertools.compress(checksums, in_remote)) + progress_callback.finish("") + return ret + + return list(set(checksums) & set(self.all())) diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py index 9d64ee5b4..8362857b8 100644 --- a/dvc/remote/ssh/connection.py +++ b/dvc/remote/ssh/connection.py @@ -289,3 +289,15 @@ class SSHConnection: def cp(self, src, dest): self.makedirs(posixpath.dirname(dest)) self.execute("cp {} {}".format(src, dest)) + + def open_max_sftp_channels(self): + # If there are more than 1 it means we've already opened max amount + if len(self._sftp_channels) <= 1: + while True: + try: + self._sftp_channels.append(self._ssh.open_sftp()) + except paramiko.ssh_exception.ChannelException: + if not self._sftp_channels: + raise + break + return self._sftp_channels From c308ad8a58dcce8e609b355d6bfcc82ed7a728c7 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 13 Aug 2019 02:08:29 +0300 Subject: [PATCH 18/21] remote: base: don't forget about the progress bar Signed-off-by: Ruslan Kuprieiev --- dvc/remote/base.py | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 2105372ae..fa6ab3b35 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -635,17 +635,21 @@ class RemoteBASE(object): Returns: A list with checksums that were found in the remote """ + if not self.no_traverse: + return list(set(checksums) & set(self.all())) + progress_callback = ProgressCallback(len(checksums)) + def exists_with_progress(path_info): + ret = self.exists(path_info) + progress_callback.update(str(path_info)) + return ret - if self.no_traverse: - with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor: - path_infos = [self.checksum_to_path_info(x) for x in checksums] - in_remote = executor.map(self.exists, path_infos) - ret = list(itertools.compress(checksums, in_remote)) - progress_callback.finish("") - return ret - - return list(set(checksums) & set(self.all())) + with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor: + path_infos = [self.checksum_to_path_info(x) for x in checksums] + in_remote = executor.map(exists_with_progress, path_infos) + ret = list(itertools.compress(checksums, in_remote)) + progress_callback.finish("") + return ret def already_cached(self, path_info): current = self.get_checksum(path_info) From 06d347178f352f4b318d35b439b4325182c3b3d7 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 13 Aug 2019 02:25:15 +0300 Subject: [PATCH 19/21] remote: base: remove whitespace Signed-off-by: Ruslan Kuprieiev --- dvc/remote/base.py | 1 + 1 file changed, 1 insertion(+) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index fa6ab3b35..20f8b7f16 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -639,6 +639,7 @@ class RemoteBASE(object): return list(set(checksums) & set(self.all())) progress_callback = ProgressCallback(len(checksums)) + def exists_with_progress(path_info): ret = self.exists(path_info) progress_callback.update(str(path_info)) From 641eb57b27ccc544266edd07997bd41f12bfa225 Mon Sep 17 00:00:00 2001 From: Alexander Schepanovski Date: Mon, 12 Aug 2019 20:21:51 +0700 Subject: [PATCH 20/21] api: stream data in api.open() for S3, GS and OSS Employ the same sign url and use resumable download approach tested for Azure. --- dvc/remote/azure.py | 5 ----- dvc/remote/base.py | 6 ++++++ dvc/remote/gs.py | 8 ++++++++ dvc/remote/oss.py | 5 +++++ dvc/remote/s3.py | 6 ++++++ 5 files changed, 25 insertions(+), 5 deletions(-) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 6e37545ed..aadc86a48 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -21,7 +21,6 @@ from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBASE from dvc.path_info import CloudURLInfo -from dvc.utils.http import open_url logger = logging.getLogger(__name__) @@ -141,10 +140,6 @@ class RemoteAZURE(RemoteBASE): paths = self._list_paths(path_info.bucket, path_info.path) return any(path_info.path == path for path in paths) - def open(self, path_info, mode="r", encoding=None): - get_url = lambda: self._generate_download_url(path_info) # noqa: E731 - return open_url(get_url, mode=mode, encoding=encoding) - def _generate_download_url(self, path_info, expires=3600): expires_at = datetime.utcnow() + timedelta(seconds=expires) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 20f8b7f16..050c16659 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -8,6 +8,7 @@ import json import logging import tempfile import itertools +from functools import partial from operator import itemgetter from multiprocessing import cpu_count from concurrent.futures import ThreadPoolExecutor @@ -23,6 +24,7 @@ from dvc.progress import progress, ProgressCallback from dvc.utils import LARGE_DIR_SIZE, tmp_fname, move, relpath, makedirs from dvc.state import StateNoop from dvc.path_info import PathInfo, URLInfo +from dvc.utils.http import open_url logger = logging.getLogger(__name__) @@ -508,6 +510,10 @@ class RemoteBASE(object): return 0 def open(self, path_info, mode="r", encoding=None): + if hasattr(self, "_generate_download_url"): + get_url = partial(self._generate_download_url, path_info) + return open_url(get_url, mode=mode, encoding=encoding) + raise RemoteActionNotImplemented("open", self.scheme) def remove(self, path_info): diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 4d07226cd..e64a41e42 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -1,6 +1,7 @@ from __future__ import unicode_literals import logging +from datetime import timedelta from funcy import cached_property try: @@ -97,3 +98,10 @@ class RemoteGS(RemoteBASE): bucket = self.gs.bucket(from_info.bucket) blob = bucket.get_blob(from_info.path) blob.download_to_filename(to_file) + + def _generate_download_url(self, path_info, expires=3600): + expiration = timedelta(seconds=int(expires)) + + bucket = self.gs.bucket(path_info.bucket) + blob = bucket.get_blob(path_info.path) + return blob.generate_signed_url(expiration=expiration) diff --git a/dvc/remote/oss.py b/dvc/remote/oss.py index cbf1fe3a7..f90fa6100 100644 --- a/dvc/remote/oss.py +++ b/dvc/remote/oss.py @@ -120,6 +120,11 @@ class RemoteOSS(RemoteBASE): from_info.path, to_file, progress_callback=cb ) + def _generate_download_url(self, path_info, expires=3600): + assert path_info.bucket == self.path_info.bucket + + return self.oss_service.sign_url("GET", path_info.path, expires) + def exists(self, path_info): paths = self._list_paths(path_info.path) return any(path_info.path == path for path in paths) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 757db8980..3d79f3eed 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -238,3 +238,9 @@ class RemoteS3(RemoteBASE): self.s3.download_file( from_info.bucket, from_info.path, to_file, Callback=cb ) + + def _generate_download_url(self, path_info, expires=3600): + params = {"Bucket": path_info.bucket, "Key": path_info.path} + return self.s3.generate_presigned_url( + ClientMethod="get_object", Params=params, ExpiresIn=int(expires) + ) From 5e4d26691a3851f75a9bf657754ebad479a49eda Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 14 Aug 2019 00:34:12 +0300 Subject: [PATCH 21/21] version: fix typo --- dvc/command/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dvc/command/version.py b/dvc/command/version.py index 770d9ad3c..ad041b4dd 100644 --- a/dvc/command/version.py +++ b/dvc/command/version.py @@ -109,7 +109,7 @@ class CmdVersion(CmdBaseNoRepo): def add_parser(subparsers, parent_parser): - VERSION_HELP = "Show DVC version and system/environment informaion." + VERSION_HELP = "Show DVC version and system/environment information." version_parser = subparsers.add_parser( "version",