From 8ad317e9067c6543d24b37b7ef753522f252a2c5 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 4 Dec 2018 03:16:41 +0200 Subject: [PATCH] remote: properly check status for remote outputs Fix #1402 Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/hdfs.py | 3 +++ dvc/dependency/http.py | 3 +++ dvc/dependency/s3.py | 3 +++ dvc/dependency/ssh.py | 3 +++ dvc/remote/azure.py | 24 +++++++++++++++++------- dvc/remote/base.py | 6 ++++++ dvc/remote/gs.py | 16 ++++++++++++---- dvc/remote/hdfs.py | 34 ++++++++++++++++------------------ dvc/remote/local.py | 1 + dvc/remote/s3.py | 26 ++++++++++++++++---------- dvc/remote/ssh.py | 42 ++++++++++++++++++++++++++++-------------- tests/__main__.py | 2 +- tests/test_repro.py | 19 ++++++++++++++++++- 13 files changed, 127 insertions(+), 55 deletions(-) diff --git a/dvc/dependency/hdfs.py b/dvc/dependency/hdfs.py index aab808ff8..4ea959309 100644 --- a/dvc/dependency/hdfs.py +++ b/dvc/dependency/hdfs.py @@ -24,6 +24,9 @@ class DependencyHDFS(DependencyBase): 'url': path} def changed(self): + if not self.exists: + return True + return self.info != self.remote.save_info(self.path_info) def save(self): diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py index 23f52b316..f05bcbed4 100644 --- a/dvc/dependency/http.py +++ b/dvc/dependency/http.py @@ -33,4 +33,7 @@ class DependencyHTTP(DependencyBase): return ret def changed(self): + if not self.exists: + return True + return self.info != self.remote.save_info(self.path_info) diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 4ca6cd14c..9eb2a48d0 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -26,6 +26,9 @@ class DependencyS3(DependencyBase): 'key': key} def changed(self): + if not self.exists: + return True + return self.info != self.remote.save_info(self.path_info) def save(self): diff --git a/dvc/dependency/ssh.py b/dvc/dependency/ssh.py index 7ff346354..f4463213f 100644 --- a/dvc/dependency/ssh.py +++ b/dvc/dependency/ssh.py @@ -33,6 +33,9 @@ class DependencySSH(DependencyBase): 'path': path} def changed(self): + if not self.exists: + return True + return self.info != self.remote.save_info(self.path_info) def save(self): diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 3b58a5a6d..2bc3112b6 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -42,6 +42,9 @@ class RemoteAzure(RemoteBase): match.group('container_name') or os.getenv('AZURE_STORAGE_CONTAINER_NAME')) + # FIXME: currently Azure doesn't support prefixes + self.prefix = '/' + self.connection_string = ( match.group('connection_string') or os.getenv('AZURE_STORAGE_CONNECTION_STRING')) @@ -138,19 +141,26 @@ class RemoteAzure(RemoteBase): 'key': '{}/{}'.format(md5[0:2], md5[2:]) } for md5 in md5s] - def _all_keys(self): + def _all_keys(self, bucket, prefix): return {blob.name - for blob in self.blob_service.list_blobs(self.bucket)} + for blob in self.blob_service.list_blobs(bucket, + prefix=prefix)} def exists(self, path_infos): - keys = self._all_keys() ret = [] + + if len(path_infos) == 0: + return ret + + bucket = path_infos[0]['bucket'] + key = path_infos[0]['key'] + for path_info in path_infos: - if path_info['scheme'] != self.scheme: - raise NotImplementedError + assert path_info['scheme'] == self.scheme + assert path_info['bucket'] == bucket key = path_info['key'] - ret.append(key in keys) + ret.append(key in self._all_keys(bucket, key)) return ret @@ -229,7 +239,7 @@ class RemoteAzure(RemoteBase): return posixpath.dirname(path) + posixpath.basename(path) def _all(self): - keys = self._all_keys() + keys = self._all_keys(self.bucket, self.prefix) return [self._path_to_etag(key) for key in keys] def gc(self, cinfos): diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 39844aefa..8868e4156 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -28,6 +28,12 @@ class DataCloudError(DvcException): super(DataCloudError, self).__init__('Data sync error: {}'.format(msg)) +class RemoteBaseCmdError(DvcException): + def __init__(self, cmd, ret, err): + m = "SSH command '{}' finished with non-zero return code '{}': {}" + super(RemoteBaseCmdError, self).__init__(m.format(cmd, ret, err)) + + class RemoteBase(object): REGEX = None REQUIRES = {} diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 922b9caa9..39848cbf3 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -175,14 +175,22 @@ class RemoteGS(RemoteBase): def exists(self, path_infos): ret = [] - gs = self.gs - bucket = gs.bucket(self.bucket) - keys = [blob.name for blob in bucket.list_blobs(prefix=self.prefix)] + if len(path_infos) == 0: + return ret + + gs = self.gs + bucket_name = path_infos[0]['bucket'] + bucket = gs.bucket(bucket_name) for path_info in path_infos: + assert self.scheme == path_info['scheme'] + assert bucket_name == path_info['bucket'] + exists = False - if path_info['key'] in keys: + key = path_info['key'] + keys = [blob.name for blob in bucket.list_blobs(prefix=key)] + if key in keys: exists = True ret.append(exists) return ret diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 5be1ee37a..1f5a337ed 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -5,13 +5,16 @@ import posixpath from subprocess import Popen, PIPE from dvc.config import Config -from dvc.remote.base import RemoteBase +from dvc.remote.base import RemoteBase, RemoteBaseCmdError from dvc.remote.local import RemoteLOCAL -from dvc.exceptions import DvcException from dvc.logger import Logger from dvc.utils import fix_env +class RemoteHDFSCmdError(RemoteBaseCmdError): + pass + + class RemoteHDFS(RemoteBase): scheme = 'hdfs' REGEX = r'^hdfs://((?P.*)@)?.*$' @@ -45,7 +48,7 @@ class RemoteHDFS(RemoteBase): stderr=PIPE) out, err = p.communicate() if p.returncode != 0: - raise DvcException('HDFS command failed: {}: {}'.format(cmd, err)) + raise RemoteHDFSCmdError(cmd, p.returncode, err) return out.decode('utf-8') @staticmethod @@ -106,7 +109,7 @@ class RemoteHDFS(RemoteBase): if checksum is None: return True - if self.changed_cache(self, checksum): + if self.changed_cache(checksum): return True return checksum_info != self.save_info(path_info) @@ -176,23 +179,18 @@ class RemoteHDFS(RemoteBase): md5[0:2], md5[2:])} for md5 in md5s] def exists(self, path_infos): - try: - stdout = self.hadoop_fs('ls -R {}'.format(self.url)) - except DvcException: - return len(path_infos) * [False] - - lines = stdout.split('\n') - lurl = [] - for line in lines: - if not line.startswith('-'): - continue - lurl.append(line.split()[-1]) - ret = [] + + if len(path_infos) == 0: + return ret + for path_info in path_infos: - exists = False - if path_info['url'] in lurl: + assert path_info['scheme'] == self.scheme + try: + self.hadoop_fs('test -e {}'.format(path_info['url'])) exists = True + except RemoteHDFSCmdError: + exists = False ret.append(exists) return ret diff --git a/dvc/remote/local.py b/dvc/remote/local.py index b6c4b35d3..7d5845530 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -505,6 +505,7 @@ class RemoteLOCAL(RemoteBase): def exists(self, path_infos): ret = [] for path_info in path_infos: + assert path_info['scheme'] == 'local' ret.append(os.path.exists(path_info['path'])) return ret diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 56e75adb4..38735bcc1 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -198,13 +198,17 @@ class RemoteS3(RemoteBase): 'key': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s] - def _all_keys(self): + def _all_keys(self, bucket, prefix): s3 = self.s3 keys = [] - kwargs = {'Bucket': self.bucket, - 'Prefix': self.prefix} + kwargs = {'Bucket': bucket, + 'Prefix': prefix} while True: + # NOTE: list_objects_v2() is 90% faster than head_object [1] + # + # [1] https://www.peterbe.com/plog/ + # fastest-way-to-find-out-if-a-file-exists-in-s3 resp = s3.list_objects_v2(**kwargs) contents = resp.get('Contents', None) if not contents: @@ -222,17 +226,19 @@ class RemoteS3(RemoteBase): return keys def exists(self, path_infos): - # NOTE: We mostly use exists() method when filtering a bulk of cache - # files to decide if we need to download/upload them and in s3 - # list_objects_v2() is much-much faster than trying to check keys - # one-by-one. ret = [] - keys = self._all_keys() + if len(path_infos) == 0: + return ret + + bucket = path_infos[0]['bucket'] for path_info in path_infos: + assert path_info['scheme'] == self.scheme + assert path_info['bucket'] == bucket exists = False - if path_info['key'] in keys: + key = path_info['key'] + if key in self._all_keys(bucket, key): exists = True ret.append(exists) @@ -332,7 +338,7 @@ class RemoteS3(RemoteBase): return posixpath.dirname(relpath) + posixpath.basename(relpath) def _all(self): - keys = self._all_keys() + keys = self._all_keys(self.bucket, self.prefix) return [self._path_to_etag(key) for key in keys] def gc(self, cinfos): diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 121a5fcf4..5da03e866 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -9,7 +9,7 @@ except ImportError: from dvc.logger import Logger from dvc.progress import progress -from dvc.remote.base import RemoteBase +from dvc.remote.base import RemoteBase, RemoteBaseCmdError from dvc.remote.local import RemoteLOCAL from dvc.config import Config from dvc.exceptions import DvcException @@ -37,6 +37,10 @@ def create_cb(name): return (lambda cur, tot: percent_cb(name, cur, tot)) +class RemoteSSHCmdError(RemoteBaseCmdError): + pass + + class RemoteSSH(RemoteBase): scheme = 'ssh' @@ -101,20 +105,30 @@ class RemoteSSH(RemoteBase): def exists(self, path_infos): ret = [] - ssh = self.ssh(host=self.host, - user=self.user, - port=self.port) - cmd = 'find {} -type f -follow -print'.format(self.prefix) - stdout = self._exec(ssh, cmd) - plist = stdout.split() - ssh.close() + + if len(path_infos) == 0: + return ret + + host = path_infos[0]['host'] + user = path_infos[0]['user'] + port = path_infos[0]['port'] + ssh = self.ssh(host=host, + user=user, + port=port) for path_info in path_infos: - exists = False - if path_info['path'] in plist: + assert host == path_info['host'] + assert user == path_info['user'] + assert port == path_info['port'] + try: + self._exec(ssh, 'test -e {}'.format(path_info['path'])) exists = True + except RemoteSSHCmdError: + exists = False ret.append(exists) + ssh.close() + return ret def _exec(self, ssh, cmd): @@ -154,10 +168,10 @@ class RemoteSSH(RemoteBase): stdout.close() stderr.close() - if stdout.channel.recv_exit_status() != 0: + ret = stdout.channel.recv_exit_status() + if ret != 0: err = b''.join(stderr_chunks).decode('utf-8') - msg = 'SSH command \'{}\' failed: {}'.format(cmd, err) - raise DvcException(msg) + raise RemoteSSHCmdError(cmd, ret, err) return b''.join(stdout_chunks).decode('utf-8') @@ -245,7 +259,7 @@ class RemoteSSH(RemoteBase): if md5 is None: return True - if self.changed_code(md5): + if self.changed_cache(md5): return True return checksum_info != self.save_info(path_info) diff --git a/tests/__main__.py b/tests/__main__.py index 441399c4d..1c15c89df 100644 --- a/tests/__main__.py +++ b/tests/__main__.py @@ -8,5 +8,5 @@ os.putenv('PATH', '{}:{}'.format(os.path.join(REPO_ROOT, 'bin'), os.getenv('PATH os.putenv('DVC_HOME', REPO_ROOT) os.putenv('DVC_TEST', 'true') -cmd = 'nosetests -v --processes=-1 --process-timeout=200 --cover-inclusive --cover-erase --cover-package=dvc --with-coverage' +cmd = 'nosetests -v --processes=-1 --process-timeout=500 --cover-inclusive --cover-erase --cover-package=dvc --with-coverage' check_call(cmd, shell=True) diff --git a/tests/test_repro.py b/tests/test_repro.py index 72cba2bac..0b8e8544e 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -616,6 +616,7 @@ class TestReproExternalBase(TestDvc): self.assertEqual(ret, 0) self.dvc = Project('.') + self.dvc.logger.be_verbose() foo_key = remote_key + self.sep + self.FOO bar_key = remote_key + self.sep + self.BAR @@ -634,31 +635,47 @@ class TestReproExternalBase(TestDvc): import_stage = self.dvc.imp(out_foo_path, 'import') self.assertTrue(os.path.exists('import')) self.assertTrue(filecmp.cmp('import', self.FOO, shallow=False)) + self.assertEqual(self.dvc.status(import_stage.path), {}) import_remote_stage = self.dvc.imp(out_foo_path, out_foo_path + '_imported') + self.assertEqual(self.dvc.status(import_remote_stage.path), {}) cmd_stage = self.dvc.run(outs=[out_bar_path], deps=[out_foo_path], cmd=self.cmd(foo_path, bar_path)) + self.assertEqual(self.dvc.status(cmd_stage.path), {}) + + self.assertEqual(self.dvc.status(), {}) self.write(self.bucket, foo_key, self.BAR_CONTENTS) sleep() - self.dvc.status() + self.assertNotEqual(self.dvc.status(), {}) stages = self.dvc.reproduce(import_stage.path) self.assertEqual(len(stages), 1) self.assertTrue(os.path.exists('import')) self.assertTrue(filecmp.cmp('import', self.BAR, shallow=False)) + self.assertEqual(self.dvc.status(import_stage.path), {}) + + stages = self.dvc.reproduce(import_remote_stage.path) + self.assertEqual(len(stages), 1) + self.assertEqual(self.dvc.status(import_remote_stage.path), {}) stages = self.dvc.reproduce(cmd_stage.path) self.assertEqual(len(stages), 1) + self.assertEqual(self.dvc.status(cmd_stage.path), {}) + self.assertEqual(self.dvc.status(), {}) self.dvc.gc() + self.assertEqual(self.dvc.status(), {}) self.dvc.remove(cmd_stage.path, outs_only=True) + self.assertNotEqual(self.dvc.status(cmd_stage.path), {}) + self.dvc.checkout(cmd_stage.path, force=True) + self.assertEqual(self.dvc.status(cmd_stage.path), {}) class TestReproExternalS3(TestReproExternalBase):