From 3ec0a5c4ae852a2c1fdaad97f32e20189bccc2d3 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 11 Aug 2018 05:15:29 +0300 Subject: [PATCH] remote: add additional checks and msgs for checkout Signed-off-by: Ruslan Kuprieiev --- dvc/output/hdfs.py | 2 +- dvc/output/local.py | 5 ----- dvc/remote/gs.py | 47 ++++++++++++++++++++++++++++++++++++++++ dvc/remote/hdfs.py | 49 ++++++++++++++++++++++++++++++++++++++++++ dvc/remote/local.py | 15 +++++++++++++ dvc/remote/s3.py | 46 +++++++++++++++++++++++++++++++++++++++ dvc/remote/ssh.py | 52 +++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 210 insertions(+), 6 deletions(-) diff --git a/dvc/output/hdfs.py b/dvc/output/hdfs.py index 6091bcde4..19cca3585 100644 --- a/dvc/output/hdfs.py +++ b/dvc/output/hdfs.py @@ -30,7 +30,7 @@ class OutputHDFS(DependencyHDFS): return True if self.use_cache \ - and self.project.cache.hdfs.save_info(self.path_info, self.info): + and self.project.cache.hdfs.changed(self.path_info, self.info): return True return False diff --git a/dvc/output/local.py b/dvc/output/local.py index 0149fd5fb..f23958530 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -68,11 +68,6 @@ class OutputLOCAL(DependencyLOCAL): if not self.use_cache: return - if not self.changed(): - msg = u'Data file \'{}\' didn\'t change.' - self.project.logger.info(msg.format(self.rel_path)) - return - self.project.cache.local.checkout(self.path_info, self.info) def _verify_metric(self): diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index c165056bb..67426aa7e 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -59,7 +59,35 @@ class RemoteGS(RemoteBase): return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])} + @staticmethod + def to_string(path_info): + return "{}://{}/{}".format(path_info['scheme'], + path_info['bucket'], + path_info['key']) + + def changed_cache(self, etag): + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + cache = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} + + if {self.PARAM_ETAG: etag} != self.save_info(cache): + if self.exists([cache])[0]: + msg = 'Corrupted cache file {}' + Logger.warn(msg.format(self.to_string(cache))) + self.remove(cache) + return True + return False + def changed(self, path_info, checksum_info): + if not self.exists([path_info])[0]: + return True + + etag = checksum_info.get(self.PARAM_ETAG, None) + if etag is None: + return True + + if self.changed_cache(etag): + return True + return checksum_info != self.save_info(path_info) def _copy(self, from_info, to_info, gs=None): @@ -95,6 +123,25 @@ class RemoteGS(RemoteBase): if not etag: return + if not self.changed(path_info, checksum_info): + msg = "Data '{}' didn't change." + Logger.info(msg.format(self.to_string(path_info))) + return + + if self.changed_cache(etag): + msg = "Cache '{}' not found. File '{}' won't be created." + Logger.warn(msg.format(etag, self.to_string(path_info))) + return + + if self.exists([path_info])[0]: + msg = "Data '{}' exists. Removing before checkout." + Logger.warn(msg.format(self.to_string(path_info))) + self.remove(path_info) + return + + msg = "Checking out '{}' with cache '{}'." + Logger.info(msg.format(self.to_string(path_info), etag)) + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) from_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 7b1e01e61..306806b8a 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -72,7 +72,37 @@ class RemoteHDFS(RemoteBase): return {self.PARAM_CHECKSUM: self.checksum(path_info)} + @staticmethod + def to_string(path_info): + return "{}://{}".format(path_info['scheme'], + path_info['url']) + + def changed_cache(self, checksum): + cache = {} + cache['scheme'] = 'hdfs' + cache['user'] = self.user + cache['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) + + if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): + if self.exists([cache])[0]: + msg = 'Corrupted cache file {}' + Logger.warn(msg.format(self.to_string(cache))) + self.remove(cache) + return True + + return False + def changed(self, path_info, checksum_info): + if not self.exists([path_info])[0]: + return True + + checksum = checksum_info.get(self.PARAM_CHECKSUM, None) + if checksum is None: + return True + + if self.changed_cache(self, checksum): + return True + return checksum_info != self.save_info(path_info) def save(self, path_info): @@ -99,6 +129,25 @@ class RemoteHDFS(RemoteBase): if not checksum: return + if not self.changed(path_info, checksum_info): + msg = "Data '{}' didn't change." + Logger.info(msg.format(self.to_string(path_info))) + return + + if self.changed_cache(checksum): + msg = "Cache '{}' not found. File '{}' won't be created." + Logger.warn(msg.format(checksum, self.to_string(path_info))) + return + + if self.exists([path_info])[0]: + msg = "Data '{}' exists. Removing before checkout." + Logger.warn(msg.format(self.to_string(path_info))) + self.remove(path_info) + return + + msg = "Checking out '{}' with cache '{}'." + Logger.info(msg.format(self.to_string(path_info), checksum)) + src = path_info.copy() src['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 45bb5a9a9..5cde2a3ea 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -235,6 +235,11 @@ class RemoteLOCAL(RemoteBase): Logger.warn(msg.format(os.path.relpath(path))) return + if not self.changed(path_info, checksum_info): + msg = "Data '{}' didn't change." + Logger.info(msg.format(os.path.relpath(path))) + return + if self.changed_cache(md5): msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.' Logger.warn(msg.format(md5, os.path.relpath(path))) @@ -334,6 +339,16 @@ class RemoteLOCAL(RemoteBase): return {self.PARAM_MD5: self.state.update(path_info['path'])} def changed(self, path_info, checksum_info): + if not self.exists([path_info])[0]: + return True + + md5 = checksum_info.get(self.PARAM_MD5, None) + if md5 is None: + return True + + if self.changed_cache(md5): + return True + return checksum_info != self.save_info(path_info) def remove(self, path_info): diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index b7c2628f4..5e07abc88 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -94,6 +94,16 @@ class RemoteS3(RemoteBase): path_info['key'])} def changed(self, path_info, checksum_info): + if not self.exists([path_info])[0]: + return True + + etag = checksum_info.get(self.PARAM_ETAG, None) + if etag is None: + return True + + if self.changed_cache(etag): + return True + return checksum_info != self.save_info(path_info) def _copy(self, from_info, to_info, s3=None): @@ -115,6 +125,23 @@ class RemoteS3(RemoteBase): return {self.PARAM_ETAG: etag} + @staticmethod + def to_string(path_info): + return "s3://{}/{}".format(path_info['bucket'], path_info['key']) + + def changed_cache(self, etag): + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key} + + if {self.PARAM_ETAG: etag} != self.save_info(cache): + if self.exists([cache])[0]: + msg = 'Corrupted cache file {}' + Logger.warn(msg.format(self.to_string(cache))) + self.remove(cache) + return True + + return False + def checkout(self, path_info, checksum_info): if path_info['scheme'] != 's3': raise NotImplementedError @@ -123,6 +150,25 @@ class RemoteS3(RemoteBase): if not etag: return + if not self.changed(path_info, checksum_info): + msg = "Data '{}' didn't change." + Logger.info(msg.format(self.to_string(path_info))) + return + + if self.changed_cache(etag): + msg = "Cache '{}' not found. File '{}' won't be created." + Logger.warn(msg.format(etag, self.to_string(path_info))) + return + + if self.exists([path_info])[0]: + msg = "Data '{}' exists. Removing before checkout." + Logger.warn(msg.format(self.to_string(path_info))) + self.remove(path_info) + return + + msg = "Checking out '{}' with cache '{}'." + Logger.info(msg.format(self.to_string(path_info), etag)) + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index ebc47586f..10236918d 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -148,7 +148,40 @@ class RemoteSSH(RemoteBase): return {self.PARAM_MD5: self.md5(path_info)} + @staticmethod + def to_string(path_info): + return "{}://{}@{}:{}".format(path_info['scheme'], + path_info['user'], + path_info['host'], + path_info['path']) + + def changed_cache(self, md5): + cache = {} + cache['scheme'] = 'ssh' + cache['host'] = self.host + cache['user'] = self.user + cache['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) + + if {self.PARAM_MD5: md5} != self.save_info(cache): + if self.exists([cache])[0]: + msg = 'Corrupted cache file {}' + Logger.warn(msg.format(self.to_string(cache))) + self.remove(cache) + return True + + return False + def changed(self, path_info, checksum_info): + if not self.exists([path_info])[0]: + return True + + md5 = checksum_info.get(self.PARAM_MD5, None) + if md5 is None: + return True + + if self.changed_code(md5): + return True + return checksum_info != self.save_info(path_info) def save(self, path_info): @@ -171,6 +204,25 @@ class RemoteSSH(RemoteBase): if not md5: return + if not self.changed(path_info, checksum_info): + msg = "Data '{}' didn't change." + Logger.info(msg.format(self.to_string(path_info))) + return + + if self.changed_cache(md5): + msg = "Cache '{}' not found. File '{}' won't be created." + Logger.warn(msg.format(md5, self.to_string(path_info))) + return + + if self.exists([path_info])[0]: + msg = "Data '{}' exists. Removing before checkout." + Logger.warn(msg.format(self.to_string(path_info))) + self.remove(path_info) + return + + msg = "Checking out '{}' with cache '{}'." + Logger.info(msg.format(self.to_string(path_info), md5)) + src = path_info.copy() src['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:])