remote: RFC: unify checksum <-> path convertions

Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2018-12-20 22:09:26 +02:00
parent 933d524ebe
commit 2a46247cb2
12 changed files with 84 additions and 107 deletions
+1 -1
View File
@@ -21,7 +21,7 @@ class DependencyHDFS(DependencyBase):
user = remote.user if remote else self.group('user')
self.path_info = {'scheme': 'hdfs',
'user': user,
'url': path}
'path': path}
def changed(self):
if not self.exists:
+1 -1
View File
@@ -21,7 +21,7 @@ class DependencyHTTP(DependencyBase):
self.path_info = {
'scheme': urlparse(path).scheme,
'url': path,
'path': path,
}
def save(self):
+1 -1
View File
@@ -30,7 +30,7 @@ class DependencyLOCAL(DependencyBase):
path = os.path.join(remote.prefix, urlparse(path).path.lstrip('/'))
if not os.path.isabs(path):
path = self.remote.ospath(path)
path = self.remote.to_ospath(path)
path = os.path.join(stage.cwd, path)
self.path = os.path.abspath(os.path.normpath(path))
+4 -13
View File
@@ -1,7 +1,6 @@
from __future__ import absolute_import
import os
import re
import posixpath
try:
from azure.storage.blob import BlockBlobService
@@ -85,7 +84,7 @@ class RemoteAzure(RemoteBase):
return [{
'scheme': self.scheme,
'bucket': self.bucket,
'path': posixpath.join(self.prefix, md5[0:2], md5[2:])
'path': self.checksum_to_path(md5[0:2]),
} for md5 in md5s]
def _list_paths(self, bucket, prefix):
@@ -111,9 +110,7 @@ class RemoteAzure(RemoteBase):
return []
ret = len(md5s) * [False]
paths = [posixpath.join(self.prefix,
md5[0:2],
md5[2:]) for md5 in md5s]
paths = [self.checksum_to_path(md5) for md5 in md5s]
for path in self._list_keys(self.bucket, self.prefix):
for i, k in enumerate(paths):
if k == path:
@@ -193,17 +190,13 @@ class RemoteAzure(RemoteBase):
if not no_progress_bar:
progress.finish_target(name)
def _path_to_etag(self, path):
relpath = posixpath.relpath(path, self.prefix)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all(self):
# NOTE: The list might be way too big(e.g. 100M entries, md5 for each
# is 32 bytes, so ~3200Mb list) and we don't really need all of it at
# the same time, so it makes sense to use a generator to gradually
# iterate over it, without keeping all of it in memory.
return (
self._path_to_etag(path)
self.path_to_checksum(path)
for path in self._list_keys(self.bucket, self.prefix)
)
@@ -218,9 +211,7 @@ class RemoteAzure(RemoteBase):
if etag in used:
continue
path_info = {'scheme': self.scheme,
'path': posixpath.join(self.prefix,
etag[0:2],
etag[2:]),
'path': self.checksum_to_path(etag),
'bucket': self.bucket}
self.remove(path_info)
removed = True
+12
View File
@@ -1,6 +1,7 @@
import os
import re
import errno
import posixpath
from multiprocessing import cpu_count
from dvc.config import Config
@@ -150,3 +151,14 @@ class RemoteBase(object):
assert len(names) == len(to_infos)
return names
@property
def ospath(self):
return posixpath
def checksum_to_path(self, checksum):
return self.ospath.join(self.prefix, checksum[0:2], checksum[2:])
def path_to_checksum(self, path):
relpath = self.ospath.relpath(path, self.prefix)
return self.ospath.dirname(relpath) + self.ospath.basename(relpath)
+7 -15
View File
@@ -1,5 +1,4 @@
import os
import posixpath
try:
from google.cloud import storage
@@ -74,7 +73,7 @@ class RemoteGS(RemoteBase):
path_info['path'])
def changed_cache(self, md5):
path = posixpath.join(self.prefix, md5[0:2], md5[2:])
path = self.checksum_to_path(md5)
cache = {'scheme': 'gs', 'bucket': self.bucket, 'path': path}
if {self.PARAM_MD5: md5} != self.save_info(cache):
@@ -117,7 +116,7 @@ class RemoteGS(RemoteBase):
raise NotImplementedError
md5 = self.get_md5(path_info['bucket'], path_info['path'])
path = posixpath.join(self.prefix, md5[0:2], md5[2:])
path = self.checksum_to_path(md5)
to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path}
self._copy(path_info, to_info)
@@ -151,7 +150,7 @@ class RemoteGS(RemoteBase):
msg = "Checking out '{}' with cache '{}'."
logger.info(msg.format(self.to_string(path_info), md5))
path = posixpath.join(self.prefix, md5[0:2], md5[2:])
path = self.checksum_to_path(md5)
from_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path}
self._copy(from_info, path_info)
@@ -172,8 +171,7 @@ class RemoteGS(RemoteBase):
def md5s_to_path_infos(self, md5s):
return [{'scheme': 'gs',
'bucket': self.bucket,
'path': posixpath.join(self.prefix,
md5[0:2], md5[2:])} for md5 in md5s]
'path': self.checksum_to_path(md5)} for md5 in md5s]
def _list_paths(self, bucket, prefix):
for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix):
@@ -193,8 +191,7 @@ class RemoteGS(RemoteBase):
return []
ret = len(md5s) * [False]
paths = [posixpath.join(self.prefix,
md5[0:2], md5[2:]) for md5 in md5s]
paths = [self.checksum_to_path(md5) for md5 in md5s]
for path in self._list_paths(self.bucket, self.prefix):
for i, k in enumerate(paths):
if k == path:
@@ -290,17 +287,13 @@ class RemoteGS(RemoteBase):
if not no_progress_bar:
progress.finish_target(name)
def _path_to_md5(self, path):
relpath = posixpath.relpath(path, self.prefix)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all_md5s(self):
# NOTE: The list might be way too big(e.g. 100M entries, md5 for each
# is 32 bytes, so ~3200Mb list) and we don't really need all of it at
# the same time, so it makes sense to use a generator to gradually
# iterate over it, without keeping all of it in memory.
return (
self._path_to_md5(path)
self.path_to_checksum(path)
for path in self._list_paths(self.bucket, self.prefix)
)
@@ -314,8 +307,7 @@ class RemoteGS(RemoteBase):
continue
path_info = {'scheme': 'gs',
'bucket': self.bucket,
'path': posixpath.join(self.prefix,
md5[0:2], md5[2:])}
'path': self.checksum_to_path(md5)}
self.remove(path_info)
removed = True
+26 -30
View File
@@ -22,7 +22,7 @@ class RemoteHDFS(RemoteBase):
def __init__(self, project, config):
self.project = project
self.url = config.get(Config.SECTION_REMOTE_URL, '/')
self.prefix = config.get(Config.SECTION_REMOTE_URL, '/')
self.user = self.group('user')
if not self.user:
self.user = config.get(Config.SECTION_REMOTE_USER,
@@ -59,38 +59,39 @@ class RemoteHDFS(RemoteBase):
def checksum(self, path_info):
regex = r'.*\t.*\t(?P<checksum>.*)'
stdout = self.hadoop_fs('checksum {}'.format(path_info['url']),
stdout = self.hadoop_fs('checksum {}'.format(path_info['path']),
user=path_info['user'])
return self._group(regex, stdout, 'checksum')
def cp(self, from_info, to_info):
self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])),
user=to_info['user'])
self.hadoop_fs('cp -f {} {}'.format(from_info['url'], to_info['url']),
dname = posixpath.dirname(to_info['path'])
self.hadoop_fs('mkdir -p {}'.format(dname), user=to_info['user'])
self.hadoop_fs('cp -f {} {}'.format(from_info['path'],
to_info['path']),
user=to_info['user'])
def rm(self, path_info):
self.hadoop_fs('rm {}'.format(path_info['url']),
self.hadoop_fs('rm {}'.format(path_info['path']),
user=path_info['user'])
def save_info(self, path_info):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
assert path_info.get('url')
assert path_info.get('path')
return {self.PARAM_CHECKSUM: self.checksum(path_info)}
@staticmethod
def to_string(path_info):
return "{}://{}".format(path_info['scheme'],
path_info['url'])
path_info['path'])
def changed_cache(self, checksum):
cache = {}
cache['scheme'] = 'hdfs'
cache['user'] = self.user
cache['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:])
cache['path'] = self.checksum_to_path(checksum)
if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache):
if self.exists(cache):
@@ -118,11 +119,11 @@ class RemoteHDFS(RemoteBase):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
assert path_info.get('url')
assert path_info.get('path')
checksum = self.checksum(path_info)
dest = path_info.copy()
dest['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:])
dest['path'] = self.checksum_to_path(checksum)
self.cp(path_info, dest)
@@ -132,7 +133,7 @@ class RemoteHDFS(RemoteBase):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
assert path_info.get('url')
assert path_info.get('path')
checksum = checksum_info.get(self.PARAM_CHECKSUM, None)
if not checksum:
@@ -158,7 +159,7 @@ class RemoteHDFS(RemoteBase):
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:])
src['path'] = self.checksum_to_path(checksum)
self.cp(src, path_info)
@@ -166,24 +167,23 @@ class RemoteHDFS(RemoteBase):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
assert path_info.get('url')
assert path_info.get('path')
logger.debug('Removing {}'.format(path_info['url']))
logger.debug('Removing {}'.format(path_info['path']))
self.rm(path_info)
def md5s_to_path_infos(self, md5s):
return [{'scheme': 'hdfs',
'user': self.user,
'url': posixpath.join(self.url,
md5[0:2], md5[2:])} for md5 in md5s]
'path': self.checksum_to_path(md5)} for md5 in md5s]
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info['scheme'] == 'hdfs'
try:
self.hadoop_fs('test -e {}'.format(path_info['url']))
self.hadoop_fs('test -e {}'.format(path_info['path']))
return True
except RemoteHDFSCmdError:
return False
@@ -213,11 +213,11 @@ class RemoteHDFS(RemoteBase):
if from_info['scheme'] != 'local':
raise NotImplementedError
cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['url']))
cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['path']))
self.hadoop_fs(cmd, user=to_info['user'])
cmd = 'copyFromLocal {} {}'.format(from_info['path'],
to_info['url'])
to_info['path'])
self.hadoop_fs(cmd, user=to_info['user'])
def download(self,
@@ -242,27 +242,24 @@ class RemoteHDFS(RemoteBase):
if not os.path.exists(dname):
os.makedirs(dname)
cmd = 'copyToLocal {} {}'.format(from_info['url'], to_info['path'])
cmd = 'copyToLocal {} {}'.format(from_info['path'],
to_info['path'])
self.hadoop_fs(cmd, user=from_info['user'])
def _path_to_checksum(self, path):
relpath = posixpath.relpath(path, self.url)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all_checksums(self):
try:
self.hadoop_fs('test -e {}'.format(self.url))
self.hadoop_fs('test -e {}'.format(self.prefix))
except RemoteHDFSCmdError:
return []
stdout = self.hadoop_fs('ls -R {}'.format(self.url))
stdout = self.hadoop_fs('ls -R {}'.format(self.prefix))
lines = stdout.split('\n')
flist = []
for line in lines:
if not line.startswith('-'):
continue
flist.append(line.split()[-1])
return [self._path_to_checksum(path) for path in flist]
return [self.path_to_checksum(path) for path in flist]
def gc(self, cinfos):
used = [info[self.PARAM_CHECKSUM] for info in cinfos['hdfs']]
@@ -274,8 +271,7 @@ class RemoteHDFS(RemoteBase):
continue
path_info = {'scheme': 'hdfs',
'user': self.user,
'url': posixpath.join(self.url,
checksum[0:2], checksum[2:])}
'path': self.checksum_to_path(checksum)}
self.remove(path_info)
removed = True
+8 -9
View File
@@ -1,7 +1,6 @@
import os
import threading
import requests
import posixpath
from dvc.logger import logger
from dvc.progress import progress
@@ -53,7 +52,7 @@ class RemoteHTTP(RemoteBase):
if to_info['scheme'] != 'local':
raise NotImplementedError
msg = "Downloading '{}' to '{}'".format(from_info['url'],
msg = "Downloading '{}' to '{}'".format(from_info['path'],
to_info['path'])
logger.debug(msg)
@@ -63,7 +62,7 @@ class RemoteHTTP(RemoteBase):
self._makedirs(to_info['path'])
total = self._content_length(from_info['url'])
total = self._content_length(from_info['path'])
if no_progress_bar or not total:
cb = None
@@ -71,9 +70,9 @@ class RemoteHTTP(RemoteBase):
cb = ProgressBarCallback(name, total)
try:
self._download_to(from_info['url'], tmp_file, callback=cb)
self._download_to(from_info['path'], tmp_file, callback=cb)
except Exception as exc:
msg = "Failed to download '{}'".format(from_info['url'])
msg = "Failed to download '{}'".format(from_info['path'])
logger.warn(msg, exc)
continue
@@ -85,13 +84,13 @@ class RemoteHTTP(RemoteBase):
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info['scheme'] in ['http', 'https']
return bool(self._request('HEAD', path_info.get('url')))
return bool(self._request('HEAD', path_info.get('path')))
def cache_exists(self, md5s):
assert isinstance(md5s, list)
ret = []
for md5 in md5s:
url = posixpath.join(self.prefix, md5[0:2], md5[2:])
url = self.checksum_to_path(md5)
exists = bool(self._request('HEAD', url))
ret.append(exists)
return ret
@@ -100,13 +99,13 @@ class RemoteHTTP(RemoteBase):
if path_info['scheme'] not in ['http', 'https']:
raise NotImplementedError
return {self.PARAM_ETAG: self._etag(path_info['url'])}
return {self.PARAM_ETAG: self._etag(path_info['path'])}
def md5s_to_path_infos(self, md5s):
return [
{
'scheme': 'http',
'url': posixpath.join(self.prefix, md5[0:2], md5[2:]),
'path': self.checksum_to_path(md5),
}
for md5 in md5s
]
+9 -7
View File
@@ -77,7 +77,7 @@ class RemoteLOCAL(RemoteBase):
for cache in os.listdir(subdir):
path = os.path.join(subdir, cache)
clist.append(self.path_to_md5(path))
clist.append(self.path_to_checksum(path))
return clist
@@ -87,10 +87,6 @@ class RemoteLOCAL(RemoteBase):
return os.path.join(self.cache_dir, md5[0:2], md5[2:])
def path_to_md5(self, path):
relpath = os.path.relpath(path, self.cache_dir)
return os.path.dirname(relpath) + os.path.basename(relpath)
def changed_cache_file(self, md5):
cache = self.get(md5)
if self.state.changed(cache, md5=md5):
@@ -165,8 +161,14 @@ class RemoteLOCAL(RemoteBase):
raise DvcException('No possible cache types left to try out.')
@property
def ospath(self):
if os.name == 'nt':
return ntpath
return posixpath
@classmethod
def ospath(cls, path):
def to_ospath(cls, path):
if os.name == 'nt':
return cls.ntpath(path)
return cls.unixpath(path)
@@ -242,7 +244,7 @@ class RemoteLOCAL(RemoteBase):
return []
for info in d:
info['relpath'] = self.ospath(info['relpath'])
info['relpath'] = self.to_ospath(info['relpath'])
return d
+7 -16
View File
@@ -1,6 +1,5 @@
import os
import threading
import posixpath
try:
import boto3
@@ -127,7 +126,7 @@ class RemoteS3(RemoteBase):
raise NotImplementedError
etag = self.get_etag(path_info['bucket'], path_info['path'])
path = posixpath.join(self.prefix, etag[0:2], etag[2:])
path = self.checksum_to_path(etag)
to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path}
self._copy(path_info, to_info)
@@ -139,7 +138,7 @@ class RemoteS3(RemoteBase):
return "s3://{}/{}".format(path_info['bucket'], path_info['path'])
def changed_cache(self, etag):
path = posixpath.join(self.prefix, etag[0:2], etag[2:])
path = self.checksum_to_path(etag)
cache = {'scheme': 's3', 'bucket': self.bucket, 'path': path}
if {self.PARAM_ETAG: etag} != self.save_info(cache):
@@ -178,7 +177,7 @@ class RemoteS3(RemoteBase):
msg = "Checking out '{}' with cache '{}'."
logger.info(msg.format(self.to_string(path_info), etag))
path = posixpath.join(self.prefix, etag[0:2], etag[2:])
path = self.checksum_to_path(etag)
from_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path}
self._copy(from_info, path_info)
@@ -196,8 +195,7 @@ class RemoteS3(RemoteBase):
def md5s_to_path_infos(self, md5s):
return [{'scheme': self.scheme,
'bucket': self.bucket,
'path': posixpath.join(self.prefix,
md5[0:2], md5[2:])} for md5 in md5s]
'path': self.checksum_to_path(md5)} for md5 in md5s]
def _list_paths(self, bucket, prefix):
s3 = self.s3
@@ -237,9 +235,7 @@ class RemoteS3(RemoteBase):
return []
ret = len(md5s) * [False]
paths = [posixpath.join(self.prefix,
md5[0:2],
md5[2:]) for md5 in md5s]
paths = [self.checksum_to_path(md5) for md5 in md5s]
for path in self._list_paths(self.bucket, self.prefix):
for i, k in enumerate(paths):
if k == path:
@@ -338,17 +334,13 @@ class RemoteS3(RemoteBase):
if not no_progress_bar:
progress.finish_target(name)
def _path_to_etag(self, path):
relpath = posixpath.relpath(path, self.prefix)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all(self):
# NOTE: The list might be way too big(e.g. 100M entries, md5 for each
# is 32 bytes, so ~3200Mb list) and we don't really need all of it at
# the same time, so it makes sense to use a generator to gradually
# iterate over it, without keeping all of it in memory.
return (
self._path_to_etag(path)
self.path_to_checksum(path)
for path in self._list_paths(self.bucket, self.prefix)
)
@@ -361,8 +353,7 @@ class RemoteS3(RemoteBase):
if etag in used_etags:
continue
path_info = {'scheme': 's3',
'path': posixpath.join(self.prefix,
etag[0:2], etag[2:]),
'path': self.checksum_to_path(etag),
'bucket': self.bucket}
self.remove(path_info)
removed = True
+6 -12
View File
@@ -78,8 +78,7 @@ class RemoteSSH(RemoteBase):
'host': self.host,
'user': self.user,
'port': self.port,
'path': posixpath.join(self.prefix,
md5[0:2], md5[2:])} for md5 in md5s]
'path': self.checksum_to_path(md5)} for md5 in md5s]
def ssh(self, host=None, user=None, port=None):
msg = "Establishing ssh connection with '{}' " \
@@ -244,7 +243,7 @@ class RemoteSSH(RemoteBase):
cache['host'] = self.host
cache['port'] = self.port
cache['user'] = self.user
cache['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:])
cache['path'] = self.checksum_to_path(md5)
if {self.PARAM_MD5: md5} != self.save_info(cache):
if self.exists(cache):
@@ -274,7 +273,7 @@ class RemoteSSH(RemoteBase):
md5 = self.md5(path_info)
dest = path_info.copy()
dest['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:])
dest['path'] = self.checksum_to_path(md5)
self.cp(path_info, dest)
@@ -308,7 +307,7 @@ class RemoteSSH(RemoteBase):
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:])
src['path'] = self.checksum_to_path(md5)
self.cp(src, path_info)
@@ -418,10 +417,6 @@ class RemoteSSH(RemoteBase):
sftp.close()
ssh.close()
def _path_to_md5(self, path):
relpath = posixpath.relpath(path, self.prefix)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all_md5s(self):
ssh = self.ssh(host=self.host,
user=self.user,
@@ -431,7 +426,7 @@ class RemoteSSH(RemoteBase):
flist = stdout.split()
ssh.close()
return [self._path_to_md5(path) for path in flist]
return [self.path_to_checksum(path) for path in flist]
def gc(self, cinfos):
used = [info[self.PARAM_MD5] for info in cinfos['ssh']]
@@ -445,8 +440,7 @@ class RemoteSSH(RemoteBase):
'user': self.user,
'host': self.host,
'port': self.port,
'path': posixpath.join(self.prefix,
md5[0:2], md5[2:])}
'path': self.checksum_to_path(md5)}
self.remove(path_info)
removed = True
+2 -2
View File
@@ -24,8 +24,8 @@ class TestCache(TestDvc):
def test_all(self):
md5_list = Cache(self.dvc).local.all()
self.assertEquals(len(md5_list), 2)
self.assertTrue(self.cache1_md5 in md5_list)
self.assertTrue(self.cache2_md5 in md5_list)
self.assertIn(self.cache1_md5, md5_list)
self.assertIn(self.cache2_md5, md5_list)
def test_get(self):
cache = Cache(self.dvc).local.get(self.cache1_md5)