Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-10 07:18:27 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1405 from efiop/master

Browse Source 
remote: properly check status for remote outputs
This commit is contained in:
Ruslan Kuprieiev
2018-12-04 04:47:27 +02:00
committed by GitHub
parent b45e22e2da 8ad317e906
commit 2245ae0772
13 changed files with 127 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/dependency/hdfs.py
+3
View File
@@ -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):
dvc/dependency/http.py
+3
View File
@@ -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)
dvc/dependency/s3.py
+3
View File
@@ -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):
dvc/dependency/ssh.py
+3
View File
@@ -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):
dvc/remote/azure.py
+17 -7
View File
@@ -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):
dvc/remote/base.py
+6
View File
@@ -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 = {}
dvc/remote/gs.py
+12 -4
View File
@@ -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
dvc/remote/hdfs.py
+16 -18
View File
@@ -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<user>.*)@)?.*$'
@@ -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
dvc/remote/local.py
+1
View File
@@ -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
dvc/remote/s3.py
+16 -10
View File
@@ -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):
dvc/remote/ssh.py
+28 -14
View File
@@ -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)
tests/__main__.py
+1 -1
View File
@@ -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)
tests/test_repro.py
+18 -1
View File
@@ -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):