Google Cloud Storage outputs always considered changed
DVC version: 0.21.3+b45e22.mod
Running from source
Also happens on official 0.21.3 release from homebrew

Going exactly by the tutorial: https://dvc.org/doc/user-guide/external-outputs
and working with a GS bucket and cache.

Immediately after the `dvc run` executes, running `dvc status`shows the output file on GS as changed. `dvc repro` will execute the command again, and it will still count as changed afterwards.

Seems like the problem is probably in `RemoteGS.exists(self, path_infos)`

The function is called with `path_infos=[ {'scheme': 'gs', 'bucket': 'mybucket, 'key': 'data.txt'}]`

However, since it is a cache remote, it lists the files in the cache bucket, whose names are etags and not the names of the files in the output. So exists returns False, and the output file is marked as changed due to not existing.
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<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]
+        ret = []
 
-        lines = stdout.split('\\n')
-        lurl = []
-        for line in lines:
-            if not line.startswith('-'):
-                continue
-            lurl.append(line.split()[-1])
+        if len(path_infos) == 0:
+            return ret
 
-        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):
['dvc/remote/local.py', 'dvc/remote/ssh.py', 'dvc/dependency/http.py', 'tests/test_repro.py', 'dvc/remote/hdfs.py', 'dvc/remote/gs.py', 'dvc/remote/s3.py', 'dvc/remote/azure.py', 'dvc/dependency/ssh.py', 'dvc/dependency/hdfs.py', 'dvc/remote/base.py', 'dvc/dependency/s3.py', 'tests/__main__.py']