Merge pull request #997 from efiop/master

Misc fixes

.travis.yml

@@ -1,4 +1,5 @@
 language: python
+cache: pip
 sudo: required
 dist: trusty
 group: deprecated-2017Q4

dvc/data_cloud.py

@@ -83,15 +83,18 @@ class DataCloud(object):
         cloud = cloud_type(self.project, cloud_config)
         return cloud
 
-    def _get_cloud(self, remote):
+    def _get_cloud(self, remote, cmd):
         if remote:
             return self._init_remote(remote)
 
         if self._cloud:
             return self._cloud
 
-        msg = "No remote repository specified. Setup default repository " \
-              "with 'dvc config core.remote <name>' or use '-r <name>'."
+        msg = "No remote repository specified. Setup default " \
+              "repository with\n"\
+              "    dvc config core.remote <name>\n" \
+              "or use:\n" \
+              "    dvc {} -r <name>".format(cmd)
         raise ConfigError(msg)
 
     def push(self, targets, jobs=1, remote=None, show_checksums=False):
@@ -100,7 +103,8 @@ class DataCloud(object):
         """
         return self.project.cache.local.push(targets,
                                              jobs=jobs,
-                                             remote=self._get_cloud(remote),
+                                             remote=self._get_cloud(remote,
+                                                                    'push'),
                                              show_checksums=show_checksums)
 
     def pull(self, targets, jobs=1, remote=None, show_checksums=False):
@@ -109,14 +113,16 @@ class DataCloud(object):
         """
         return self.project.cache.local.pull(targets,
                                              jobs=jobs,
-                                             remote=self._get_cloud(remote),
+                                             remote=self._get_cloud(remote,
+                                                                    'pull'),
                                              show_checksums=show_checksums)
 
     def status(self, targets, jobs=1, remote=None, show_checksums=False):
         """
         Check status of data items in a cloud-agnostic way.
         """
+        cloud = self._get_cloud(remote, 'status')
         return self.project.cache.local.status(targets,
                                                jobs=jobs,
-                                               remote=self._get_cloud(remote),
+                                               remote=cloud,
                                                show_checksums=show_checksums)

dvc/output/gs.py

@@ -30,7 +30,7 @@ class OutputGS(DependencyGS):
             return True
 
         if self.use_cache \
-           and self.info != self.project.cache.gs.save_info(self.path_info):
+           and self.project.cache.gs.changed(self.path_info, self.info):
             return True
 
         return False

dvc/output/hdfs.py

@@ -30,7 +30,7 @@ class OutputHDFS(DependencyHDFS):
             return True
 
         if self.use_cache \
-           and self.info != self.project.cache.hdfs.save_info(self.path_info):
+           and self.project.cache.hdfs.changed(self.path_info, self.info):
             return True
 
         return False

dvc/output/local.py

@@ -62,17 +62,12 @@ class OutputLOCAL(DependencyLOCAL):
         if not self.use_cache:
             return super(OutputLOCAL, self).changed()
 
-        return self.info != self.project.cache.local.save_info(self.path_info)
+        return self.project.cache.local.changed(self.path_info, self.info)
 
     def checkout(self):
         if not self.use_cache:
             return
 
-        if not self.changed():
-            msg = u'Data file \'{}\' didn\'t change, skipping checkout.'
-            self.project.logger.debug(msg.format(self.rel_path))
-            return
-
         self.project.cache.local.checkout(self.path_info, self.info)
 
     def _verify_metric(self):

dvc/output/s3.py

@@ -30,7 +30,7 @@ class OutputS3(DependencyS3):
             return True
 
         if self.use_cache \
-           and self.info != self.project.cache.s3.save_info(self.path_info):
+           and self.project.cache.s3.changed(self.path_info, self.info):
             return True
 
         return False

dvc/output/ssh.py

@@ -30,7 +30,7 @@ class OutputSSH(DependencySSH):
             return True
 
         if self.use_cache \
-           and self.info != self.project.cache.ssh.save_info(self.path_info):
+           and self.project.cache.ssh.changed(self.path_info, self.info):
             return True
 
         return False

dvc/project.py

@@ -383,7 +383,8 @@ class Project(object):
             self._do_gc('azure', self.cache.azure.gc, clist)
 
         if cloud:
-            self._do_gc('remote', self.cloud._get_cloud(remote).gc, clist)
+            self._do_gc('remote', self.cloud._get_cloud(remote,
+                                                        'gc -c').gc, clist)
 
     def push(self,
              target=None,

dvc/remote/azure.py

@@ -92,6 +92,9 @@ class RemoteAzure(RemoteBase):
 #
 #        return {self.PARAM_ETAG: etag}
 #
+#    def changed(self, path_info, checksum_info):
+#        return checksum_info != self.save_info(path_info)
+#
 #    def _copy_blob(self, to_bucket, to_key, from_bucket, from_key):
 #        source = self.blob_service.make_blob_url(from_bucket, from_key)
 #

dvc/remote/base.py

@@ -81,6 +81,9 @@ class RemoteBase(object):
     def save_info(self, path_info):
         raise NotImplementedError
 
+    def changed(self, path_info, checksum_info):
+        raise NotImplementedError
+
     def save(self, path_info):
         raise NotImplementedError
 

dvc/remote/gs.py

@@ -59,6 +59,37 @@ 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):
         gs = gs if gs else self.gs
 
@@ -92,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}
 

dvc/remote/hdfs.py

@@ -72,6 +72,39 @@ 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):
         if path_info['scheme'] != 'hdfs':
             raise NotImplementedError
@@ -96,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:])
 

dvc/remote/local.py

@@ -82,7 +82,7 @@ class RemoteLOCAL(RemoteBase):
         relpath = os.path.relpath(path, self.cache_dir)
         return os.path.dirname(relpath) + os.path.basename(relpath)
 
-    def changed(self, md5):
+    def changed_cache(self, md5):
         cache = self.get(md5)
         if self.state.changed(cache, md5=md5):
             if os.path.exists(cache):
@@ -104,6 +104,10 @@ class RemoteLOCAL(RemoteBase):
         while i > 0:
             try:
                 self.CACHE_TYPE_MAP[self.cache_types[0]](cache, path)
+                msg = "Created '{}': {} -> {}"
+                Logger.info(msg.format(self.cache_types[0],
+                                       os.path.relpath(cache),
+                                       os.path.relpath(path)))
                 return
             except Exception as exc:
                 msg = 'Cache type \'{}\' is not supported: {}'
@@ -172,7 +176,7 @@ class RemoteLOCAL(RemoteBase):
         dir_info = sorted(dir_info, key=itemgetter(self.PARAM_RELPATH))
 
         md5 = dict_md5(dir_info) + self.MD5_DIR_SUFFIX
-        if self.changed(md5):
+        if self.changed_cache(md5):
             self.dump_dir_cache(md5, dir_info)
 
         return (md5, dir_info)
@@ -231,14 +235,19 @@ class RemoteLOCAL(RemoteBase):
             Logger.warn(msg.format(os.path.relpath(path)))
             return
 
-        if self.changed(md5):
+        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)))
             remove(path)
             return
 
         if os.path.exists(path):
-            msg = u'Data \'{}\' exists. Removing before checkout'
+            msg = u'Data \'{}\' exists. Removing before checkout.'
             Logger.warn(msg.format(os.path.relpath(path)))
             remove(path)
 
@@ -277,7 +286,7 @@ class RemoteLOCAL(RemoteBase):
 
         cache = self.get(md5)
 
-        if self.changed(md5):
+        if self.changed_cache(md5):
             self._move(path, cache)
         else:
             remove(path)
@@ -297,7 +306,7 @@ class RemoteLOCAL(RemoteBase):
             p = os.path.join(path, relpath)
             c = self.get(m)
 
-            if self.changed(m):
+            if self.changed_cache(m):
                 self._move(p, c)
             else:
                 remove(p)
@@ -329,6 +338,19 @@ 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):
         if path_info['scheme'] != 'local':
             raise NotImplementedError
@@ -498,7 +520,7 @@ class RemoteLOCAL(RemoteBase):
 
         progress.update_target(title, 90, 100)
 
-        local_exists = [not self.changed(md5) for md5 in md5s]
+        local_exists = [not self.changed_cache(md5) for md5 in md5s]
 
         progress.finish_target(title)
 
@@ -525,7 +547,7 @@ class RemoteLOCAL(RemoteBase):
         names = []
         # NOTE: filter files that are not corrupted
         for md5, name in grouped:
-            if self.changed(md5):
+            if self.changed_cache(md5):
                 md5s.append(md5)
                 names.append(name)
 
@@ -594,7 +616,7 @@ class RemoteLOCAL(RemoteBase):
 
         # NOTE: verifying that our cache is not corrupted
         def func(info):
-            return not self.changed(info[self.PARAM_MD5])
+            return not self.changed_cache(info[self.PARAM_MD5])
         checksum_infos = list(filter(func, checksum_infos))
 
         progress.update_target(title, 20, 100)

dvc/remote/s3.py

@@ -93,6 +93,19 @@ class RemoteS3(RemoteBase):
         return {self.PARAM_ETAG: self.get_etag(path_info['bucket'],
                                                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):
         s3 = s3 if s3 else self.s3
 
@@ -112,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
@@ -120,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}
 

dvc/remote/ssh.py

@@ -148,6 +148,42 @@ 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):
         if path_info['scheme'] != 'ssh':
             raise NotImplementedError
@@ -168,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:])