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

dvc: cloud: use local state for cmp_checksum

Browse Source 
Fixes #633

Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2018-04-10 01:56:06 +03:00
parent 0c5cc47a86
commit 6dc9fbd08e
9 changed files with 43 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cloud/aws.py
+1 -26
View File
@@ -17,7 +17,6 @@ from dvc.logger import Logger
from dvc.progress import progress
from dvc.cloud.credentials_aws import AWSCredentials
from dvc.cloud.base import DataCloudError, DataCloudBase
from dvc.utils import file_md5
def sizeof_fmt(num, suffix='B'):
@@ -98,27 +97,6 @@ class DataCloudAWS(DataCloudBase):
if self.bucket is None:
raise DataCloudError('Storage path {} is not setup correctly'.format(self.storage_bucket))
@staticmethod
def _cmp_checksum(key, fname):
"""
Verify local and remote checksums. Used 'dvc-md5' metadata if supported
or falls back to etag.
"""
# Calling get_key() once more to avoid empty metadata.
# See http://blog.bidiuk.com/2014/02/get-amazon-s3-metadata-in-python-using-boto/
key = key.bucket.get_key(key.name)
md5_cloud = key.metadata.get('dvc-md5', None)
md5_local = file_md5(fname)[0]
if md5_cloud == None:
md5_cloud = key.etag[1:-1]
if md5_cloud == md5_local:
return True
return False
@staticmethod
def _upload_tracker(fname):
"""
@@ -221,10 +199,7 @@ class DataCloudAWS(DataCloudBase):
"""
Create multipart upload and save info to tracker file.
"""
# AWS doesn't provide easilly accessible md5 for multipart
# objects, so we have to store our own md5 sum to use later.
metadata = {'dvc-md5' : str(file_md5(fname)[0])}
multipart = key.bucket.initiate_multipart_upload(key.name, metadata=metadata)
multipart = key.bucket.initiate_multipart_upload(key.name)
self._write_upload_tracker(fname, multipart.id)
return multipart
dvc/cloud/base.py
+15
View File
@@ -31,6 +31,14 @@ class DataCloudError(DvcException):
super(DataCloudError, self).__init__('Data sync error: {}'.format(msg))
class CloudSettings(object):
def __init__(self, cache=None, state=None, global_storage_path=None, cloud_config=None):
self.cache = cache
self.state = state
self.cloud_config = cloud_config
self.global_storage_path = global_storage_path
class DataCloudBase(object):
""" Base class for DataCloud """
@@ -146,6 +154,13 @@ class DataCloudBase(object):
return ret
def _cmp_checksum(self, blob, fname):
md5 = self._cloud_settings.cache.path_to_md5(fname)
if self._cloud_settings.state.changed(fname, md5=md5):
return False
return True
def push(self, path):
key = self._get_key(path)
if key:
dvc/cloud/data_cloud.py
+14 -21
View File
@@ -1,7 +1,6 @@
import re
from multiprocessing.pool import ThreadPool
from dvc.cloud.instance_manager import CloudSettings
from dvc.logger import Logger
from dvc.exceptions import DvcException
from dvc.config import Config, ConfigError
@@ -11,7 +10,7 @@ from dvc.cloud.aws import DataCloudAWS
from dvc.cloud.gcp import DataCloudGCP
from dvc.cloud.ssh import DataCloudSSH
from dvc.cloud.local import DataCloudLOCAL
from dvc.cloud.base import DataCloudBase
from dvc.cloud.base import DataCloudBase, CloudSettings
class DataCloud(object):
@@ -24,9 +23,10 @@ class DataCloud(object):
'local' : DataCloudLOCAL,
}
def __init__(self, cache, config):
def __init__(self, cache=None, config=None, state=None):
self._cache = cache
self._config = config
self._state = state
remote = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_REMOTE, '')
if remote == '':
@@ -58,7 +58,7 @@ class DataCloud(object):
if not cloud_type:
raise ConfigError("Unsupported url '{}'".format(url))
return self._init_cloud(self._cache, cloud_config, cloud_type)
return self._init_cloud(cloud_config, cloud_type)
def __init__compat(self):
cloud_name = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, '').strip().lower()
@@ -74,29 +74,22 @@ class DataCloud(object):
if not cloud_config:
raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_name)
return self._init_cloud(self._cache, cloud_config, cloud_type)
return self._init_cloud(cloud_config, cloud_type)
def _init_cloud(self, cache, cloud_config, cloud_type):
cloud_settings = self.get_cloud_settings(cache,
self._config,
cloud_config)
def _init_cloud(self, cloud_config, cloud_type):
global_storage_path = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_STORAGEPATH, None)
if global_storage_path:
Logger.warn('Using obsoleted config format. Consider updating.')
cloud_settings = CloudSettings(cache=self._cache,
state=self._state,
global_storage_path=global_storage_path,
cloud_config=cloud_config)
cloud = cloud_type(cloud_settings)
cloud.sanity_check()
return cloud
@staticmethod
def get_cloud_settings(cache, config, cloud_config):
"""
Obtain cloud settings from config.
"""
global_storage_path = config[Config.SECTION_CORE].get(Config.SECTION_CORE_STORAGEPATH, None)
if global_storage_path:
Logger.warn('Using obsoleted config format. Consider updating.')
cloud_settings = CloudSettings(cache, global_storage_path, cloud_config)
return cloud_settings
def _collect(self, cloud, targets, jobs, local):
collected = set()
pool = ThreadPool(processes=jobs)
dvc/cloud/gcp.py
-14
View File
@@ -6,7 +6,6 @@ from google.cloud import storage as gc
from dvc.logger import Logger
from dvc.config import Config, ConfigError
from dvc.cloud.base import DataCloudError, DataCloudBase
from dvc.utils import file_md5
from dvc.progress import progress
@@ -28,19 +27,6 @@ class DataCloudGCP(DataCloudBase):
raise DataCloudError('sync up: google cloud bucket {} '
'doesn\'t exist'.format(self.storage_bucket))
@staticmethod
def _cmp_checksum(blob, fname):
"""
Verify local and remote checksums.
"""
md5 = file_md5(fname)[1]
b64_encoded_md5 = base64.b64encode(md5).decode() if md5 else None
if blob.md5_hash == b64_encoded_md5:
return True
return False
def _pull_key(self, key, path, no_progress_bar=False):
self._makedirs(path)
dvc/cloud/instance_manager.py
-5
View File
@@ -1,5 +0,0 @@
class CloudSettings(object):
def __init__(self, cache, global_storage_path, cloud_config):
self.cache = cache
self.cloud_config = cloud_config
self.global_storage_path = global_storage_path
dvc/cloud/local.py
-3
View File
@@ -25,9 +25,6 @@ class DataCloudLOCAL(DataCloudBase):
def cache_file_key(self, path):
return os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.cache_dir)
def _cmp_checksum(self, key, path):
return filecmp.cmp(path, key.path)
def _get_key(self, path):
key_name = self.cache_file_key(path)
key = LocalKey(self.storage_path, key_name)
dvc/cloud/ssh.py
-10
View File
@@ -64,16 +64,6 @@ class DataCloudSSH(DataCloudBase):
raise
return False
def _cmp_checksum(self, key, path):
#FIXME actually compare md5
path_exists = os.path.exists(path)
key_exists = self._isfile_remote(key.path)
if path_exists and key_exists:
return True
return False
def _get_key(self, path):
key_name = self.cache_file_key(path)
key = SSHKey(self.path, key_name)
dvc/project.py
+1 -1
View File
@@ -41,7 +41,7 @@ class Project(object):
self.state = State(self.root_dir, self.dvc_dir)
self.config = Config(self.dvc_dir)
self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None))
self.cloud = DataCloud(self.cache, self.config._config)
self.cloud = DataCloud(cache=self.cache, state=self.state, config=self.config._config)
@staticmethod
def init(root_dir=os.curdir):
tests/test_data_cloud.py
+12 -7
View File
@@ -8,8 +8,7 @@ import tempfile
from dvc.main import main
from dvc.config import Config, ConfigError
from dvc.cloud.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudGCP
from dvc.cloud.base import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED
from dvc.cloud.instance_manager import CloudSettings
from dvc.cloud.base import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED, CloudSettings
from tests.basic_env import TestDvc
@@ -95,7 +94,7 @@ def get_gcp_url():
class TestDataCloud(TestDvc):
def _test_cloud(self, config, cl):
cloud = DataCloud(self.dvc.cache, config)
cloud = DataCloud(cache=self.dvc.cache, state=self.dvc.state, config=config)
self.assertIsInstance(cloud._cloud, cl)
def test(self):
@@ -109,7 +108,7 @@ class TestDataCloud(TestDvc):
with self.assertRaises(ConfigError) as cx:
config = TEST_CONFIG
config[TEST_SECTION][Config.SECTION_REMOTE_URL] = 'notsupportedscheme://a/b'
DataCloud(self.dvc.cache, config)
DataCloud(cache=self.dvc.cache, state=self.dvc.state, config=config)
class TestDataCloudBase(TestDvc):
@@ -191,7 +190,9 @@ class TestDataCloudAWS(TestDataCloudBase):
config = TEST_CONFIG
config[TEST_SECTION][Config.SECTION_REMOTE_URL] = repo
config[TEST_SECTION][Config.SECTION_AWS_REGION] = TEST_AWS_REPO_REGION
cloud_settings = CloudSettings(self.dvc.cache, None, config[TEST_SECTION])
cloud_settings = CloudSettings(cache=self.dvc.cache,
state=self.dvc.state,
cloud_config=config[TEST_SECTION])
self.cloud = DataCloudAWS(cloud_settings)
def test(self):
@@ -209,7 +210,9 @@ class TestDataCloudGCP(TestDataCloudBase):
# Setup cloud
config = TEST_CONFIG
config[TEST_SECTION][Config.SECTION_REMOTE_URL] = repo
cloud_settings = CloudSettings(self.dvc.cache, None, config[TEST_SECTION])
cloud_settings = CloudSettings(cache=self.dvc.cache,
state=self.dvc.state,
cloud_config=config[TEST_SECTION])
self.cloud = DataCloudGCP(cloud_settings)
def test(self):
@@ -223,7 +226,9 @@ class TestDataCloudLOCAL(TestDataCloudBase):
config = TEST_CONFIG
config[TEST_SECTION][Config.SECTION_REMOTE_URL] = self.dname
cloud_settings = CloudSettings(self.dvc.cache, None, config[TEST_SECTION])
cloud_settings = CloudSettings(cache=self.dvc.cache,
state=self.dvc.state,
cloud_config=config[TEST_SECTION])
self.cloud = DataCloudLOCAL(cloud_settings)
def test(self):