diff --git a/dvc/cloud/aws.py b/dvc/cloud/aws.py index d8152b82d..af7ccb506 100644 --- a/dvc/cloud/aws.py +++ b/dvc/cloud/aws.py @@ -103,6 +103,10 @@ class DataCloudAWS(DataCloudBase): 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] @@ -128,23 +132,21 @@ class DataCloudAWS(DataCloudBase): """ return fname + '.download' - def _import(self, bucket_name, key_name, fname): - - bucket = self._get_bucket_aws(bucket_name) + def _pull_key(self, key, fname): + Logger.debug("Pulling key '{}' from bucket '{}' to file '{}'".format(key.name, + key.bucket.name, + fname)) + self._makedirs(fname) tmp_file = self.tmp_file(fname) name = os.path.basename(fname) - key = bucket.get_key(key_name) - if not key: - Logger.error('File "{}" does not exist in the cloud'.format(key_name)) - return None if self._cmp_checksum(key, fname): - Logger.debug('File "{}" matches with "{}".'.format(fname, key_name)) + Logger.debug('File "{}" matches with "{}".'.format(fname, key.name)) return fname - Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name, - key_name, + Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(key.bucket.name, + key.name, fname)) res_h = ResumableDownloadHandler(tracker_file_name=self._download_tracker(tmp_file), @@ -152,7 +154,7 @@ class DataCloudAWS(DataCloudBase): try: key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h) except Exception as exc: - Logger.error('Failed to download "{}": {}'.format(key_name, exc)) + Logger.error('Failed to download "{}": {}'.format(key.name, exc)) return None os.rename(tmp_file, fname) @@ -162,6 +164,24 @@ class DataCloudAWS(DataCloudBase): return fname + def _get_key(self, path): + key_name = self.cache_file_key(path) + bucket = self._get_bucket_aws(self.storage_bucket) + return bucket.get_key(key_name) + + def _get_keys(self, path): + key_name = self.cache_file_key(path) + bucket = self._get_bucket_aws(self.storage_bucket) + keys = bucket.list(prefix=key_name) + if not keys: + return None + return list(keys) + + def _new_key(self, path): + key_name = self.cache_file_key(path) + bucket = self._get_bucket_aws(self.storage_bucket) + return bucket.new_key(key_name) + def _write_upload_tracker(self, fname, mp_id): """ Write multipart id to upload tracker. @@ -265,23 +285,8 @@ class DataCloudAWS(DataCloudBase): multipart.complete_upload() self._unlink_upload_tracker(fname) - def push(self, path): + def _push_key(self, key, path): """ push, aws version """ - - aws_key = self.cache_file_key(path) - bucket = self._get_bucket_aws(self.storage_bucket) - key = bucket.get_key(aws_key) - if key: - Logger.debug('File already uploaded to the cloud. Checksum validation...') - - if self._cmp_checksum(key, path): - Logger.debug('File checksum matches. No uploading is needed.') - return path - - Logger.debug('Checksum miss-match. Re-uploading is required.') - - key = bucket.new_key(aws_key) - try: self._push_multipart(key, path) except Exception as exc: @@ -291,36 +296,3 @@ class DataCloudAWS(DataCloudBase): progress.finish_target(os.path.basename(path)) return path - - def _status(self, path): - aws_key = self.cache_file_key(path) - bucket = self._get_bucket_aws(self.storage_bucket) - key = bucket.get_key(aws_key) - - remote_exists = key is not None - local_exists = os.path.exists(path) - diff = None - if remote_exists and local_exists: - diff = self._cmp_checksum(key, path) - - return (local_exists, remote_exists, diff) - - def remove(self, path): - aws_file_name = self.cache_file_key(path) - - Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name)) - - if not self._aws_creds.access_key_id or not self._aws_creds.secret_access_key: - Logger.debug('[Cmd-Remove] Unable to check cache file in the cloud') - return - conn = S3Connection(self._aws_creds.access_key_id, self._aws_creds.secret_access_key) - bucket_name = self.storage_bucket - bucket = conn.lookup(bucket_name) - if bucket: - key = bucket.get_key(aws_file_name) - if not key: - Logger.warn('[Cmd-Remove] S3 remove warning: ' - 'file "{}" does not exist in S3'.format(aws_file_name)) - else: - key.delete() - Logger.info('[Cmd-Remove] File "{}" was removed from S3'.format(aws_file_name)) diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py index fa267578e..59538706f 100644 --- a/dvc/cloud/base.py +++ b/dvc/cloud/base.py @@ -1,8 +1,12 @@ import os +import yaml +import tempfile +from checksumdir import dirhash from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.config import ConfigError +from dvc.stage import Output STATUS_UNKNOWN = 0 @@ -71,7 +75,9 @@ class DataCloudBase(object): def cache_file_key(self, fname): """ Key of a file within the bucket """ - return '{}/{}'.format(self.storage_prefix, os.path.basename(fname)).strip('/') + relpath = os.path.relpath(fname, self._cloud_settings.cache_dir) + relpath.replace('\\', '/') + return '{}/{}'.format(self.storage_prefix, relpath).strip('/') @staticmethod def tmp_file(fname): @@ -84,35 +90,141 @@ class DataCloudBase(object): """ pass - def _import(self, bucket, fin, fout): - """ - Cloud-specific method for importing data file. - """ + def _push_key(self, key, path): pass def push(self, path): - """ Cloud-specific method for pushing data """ + """ Generic method for pushing data """ + if os.path.isfile(path): + return self._push(path) + + res = [] + for root, dirs, files in os.walk(path): + for fname in files: + p = os.path.join(root, fname) + res.append(self._push(p)) + + with open(p, 'r') as fd: + d = yaml.safe_load(fd) + md5 = d[Output.PARAM_MD5] + cache = os.path.join(self._cloud_settings.cache_dir, md5) + res.append(self._push(cache)) + return res + + def _push(self, path): + key = self._get_key(path) + if key: + Logger.debug("File '{}' already uploaded to the cloud. Validating checksum...".format(path)) + if self._cmp_checksum(key, path): + Logger.debug('File checksum matches. No uploading is needed.') + return [] + Logger.debug('Checksum mismatch. Reuploading is required.') + + key = self._new_key(path) + return [self._push_key(key, path)] + + def _makedirs(self, fname): + dname = os.path.dirname(fname) + if not os.path.exists(dname): + os.makedirs(dname) + + def _pull_key(self, key, path): + """ Cloud-specific method of pulling keys """ + pass + + def _get_key(self, path): + """ Cloud-specific method of getting keys """ + pass + + def _get_keys(self, path): + """ Cloud-specific method of getting keys """ pass def pull(self, path): """ Generic method for pulling data from the cloud """ - key_name = self.cache_file_key(path) - return self._import(self.storage_bucket, key_name, path) + key = self._get_key(path) + if key: + return [self._pull_key(key, path)] + + keys = self._get_keys(path) + if not keys: + Logger.error("File '{}' does not exist in the cloud".format(path)) + return [] + + res = [] + for k in keys: + fname = os.path.join(path, os.path.relpath(k.name, self.cache_file_key(path))) + res.append(self._pull_key(k, fname)) + with open(fname, 'r') as fd: + d = yaml.safe_load(fd) + md5 = d[Output.PARAM_MD5] + + cache = os.path.join(self._cloud_settings.cache_dir, md5) + cache_key = self._get_key(cache) + if not cache_key: + Logger.error("File '{}' does not exist in the cloud".format(path)) + continue + + res.append(self._pull_key(cache_key, cache)) + + return res def remove(self, path): """ - Cloud-specific method for removing data item from the cloud. + Generic method for removing data item from the cloud. """ - pass + key = self._get_key(path) + if key: + key.delete() + return [k.name] - def _status(self, path): - """ - Cloud-specific method for checking data item status. - """ - pass + keys = self._get_keys(path) + if not keys: + Logger.error("File '{}' does not exist in the cloud".format(path)) + return [] + + res = [] + for k in keys: + k.delete() + res.append(k.name) + return res + + def _status(self, key, path): + remote_exists = key != None + local_exists = os.path.exists(path) + + diff = None + if remote_exists and local_exists: + diff = self._cmp_checksum(key, path) + + return STATUS_MAP.get((local_exists, remote_exists, diff), STATUS_UNKNOWN) def status(self, path): """ Generic method for checking data item status. """ - return STATUS_MAP.get(self._status(path), STATUS_UNKNOWN) + key = self._get_key(path) + if key: + return self._status(key, path) + + keys = self._get_keys(path) + if not keys or len(keys) == 0: + return STATUS_NEW + + res = [] + for k in keys: + fname = os.path.join(path, os.path.relpath(k.name, self.cache_file_key(path))) + res.append(self._status(k, fname)) + + tmp = tempfile.NamedTemporaryFile() + self._pull_key(k, tmp.name) + with open(tmp.name, 'r') as fd: + d = yaml.safe_load(fd) + tmp.close() + md5 = d[Output.PARAM_MD5] + + cache = os.path.join(self._cloud_settings.cache_dir, md5) + cache_key = self._get_key(cache) + res.append(self._status(cache_key, cache)) + + return max(res) diff --git a/dvc/cloud/data_cloud.py b/dvc/cloud/data_cloud.py index 02727e94a..5298855ea 100644 --- a/dvc/cloud/data_cloud.py +++ b/dvc/cloud/data_cloud.py @@ -7,7 +7,6 @@ from dvc.utils import map_progress from dvc.cloud.aws import DataCloudAWS from dvc.cloud.gcp import DataCloudGCP from dvc.cloud.local import DataCloudLOCAL -from dvc.cloud.http import DataCloudHTTP class DataCloud(object): @@ -16,20 +15,16 @@ class DataCloud(object): CLOUD_MAP = { 'AWS' : DataCloudAWS, 'GCP' : DataCloudGCP, - 'HTTP' : DataCloudHTTP, 'LOCAL' : DataCloudLOCAL, } SCHEME_MAP = { 's3' : 'AWS', - 'http' : 'HTTP', - 'https' : 'HTTP', - 'ftp' : 'HTTP', 'gs' : 'GCP', '' : 'LOCAL', } - def __init__(self, config): + def __init__(self, cache_dir, config): self._config = config cloud_type = self._config['Global'].get('Cloud', '').strip().upper() @@ -39,7 +34,8 @@ class DataCloud(object): if cloud_type not in self._config.keys(): raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_type) - cloud_settings = self.get_cloud_settings(self._config, + cloud_settings = self.get_cloud_settings(cache_dir, + self._config, cloud_type) self.typ = cloud_type @@ -48,7 +44,7 @@ class DataCloud(object): self.sanity_check() @staticmethod - def get_cloud_settings(config, cloud_type): + def get_cloud_settings(cache_dir, config, cloud_type): """ Obtain cloud settings from config. """ @@ -57,7 +53,7 @@ class DataCloud(object): else: cloud_config = config[cloud_type] global_storage_path = config['Global'].get('StoragePath', None) - cloud_settings = CloudSettings(global_storage_path, cloud_config) + cloud_settings = CloudSettings(cache_dir, global_storage_path, cloud_config) return cloud_settings def sanity_check(self): @@ -89,12 +85,6 @@ class DataCloud(object): """ return map_progress(func, targets, jobs) - def sync(self, targets, jobs=1): - """ - Sync data items in a cloud-agnostic way. - """ - return self._map_targets(self._cloud.sync, targets, jobs) - def push(self, targets, jobs=1): """ Push data items in a cloud-agnostic way. diff --git a/dvc/cloud/gcp.py b/dvc/cloud/gcp.py index df9c386e8..5a36185cf 100644 --- a/dvc/cloud/gcp.py +++ b/dvc/cloud/gcp.py @@ -47,84 +47,63 @@ class DataCloudGCP(DataCloudBase): return False - def _import(self, bucket_name, key, fname): + def _pull_key(self, key, path): + self._makedirs(path) - bucket = self._get_bucket_gc(bucket_name) + name = os.path.basename(path) + tmp_file = self.tmp_file(path) - name = os.path.basename(fname) - tmp_file = self.tmp_file(fname) + if self._cmp_checksum(key, path): + Logger.debug('File "{}" matches with "{}".'.format(path, key.name)) + return path - blob = bucket.get_blob(key) - if not blob: - Logger.error('File "{}" does not exist in the cloud'.format(key)) - return None - - if self._cmp_checksum(blob, fname): - Logger.debug('File "{}" matches with "{}".'.format(fname, key)) - return fname - - Logger.debug('Downloading cache file from gc "{}/{}"'.format(bucket.name, key)) + Logger.debug('Downloading cache file from gc "{}/{}"'.format(key.bucket.name, key.name)) # percent_cb is not available for download_to_filename, so # lets at least update progress at keypoints(start, finish) progress.update_target(name, 0, None) try: - blob.download_to_filename(tmp_file) + key.download_to_filename(tmp_file) except Exception as exc: - Logger.error('Failed to download "{}": {}'.format(key, exc)) + Logger.error('Failed to download "{}": {}'.format(key.name, exc)) return None - os.rename(tmp_file, fname) + os.rename(tmp_file, path) progress.finish_target(name) Logger.debug('Downloading completed') - return fname - - def push(self, path): - """ push, gcp version """ + return path + def _get_key(self, path): + key_name = self.cache_file_key(path) bucket = self._get_bucket_gc(self.storage_bucket) - blob_name = self.cache_file_key(path) + return bucket.get_blob(key_name) + + def _get_keys(self, path): + key_name = self.cache_file_key(path) + bucket = self._get_bucket_gc(self.storage_bucket) + keys = bucket.list_blobs(prefix=key_name) + if not keys: + return None + return list(keys) + + def _new_key(self, path): + key_name = self.cache_file_key(path) + bucket = self._get_bucket_gc(self.storage_bucket) + return bucket.blob(key_name) + + def _push_key(self, key, path): + """ push, gcp version """ name = os.path.basename(path) - blob = bucket.get_blob(blob_name) - if blob is not None and blob.exists(): - if self._cmp_checksum(blob, path): - Logger.debug('checksum %s matches. Skipping upload' % path) - return path - Logger.debug('checksum %s mismatch. re-uploading' % path) - - # same as in _import progress.update_target(name, 0, None) - blob = bucket.blob(blob_name) - blob.upload_from_filename(path) + key.upload_from_filename(path) progress.finish_target(name) Logger.debug('uploading %s completed' % path) return path - - def _status(self, path): - """ status, gcp version """ - - bucket = self._get_bucket_gc(self.storage_bucket) - blob_name = self.cache_file_key(path) - blob = bucket.get_blob(blob_name) - - remote_exists = blob is not None and blob.exists() - local_exists = os.path.exists(path) - diff = None - if remote_exists and local_exists: - diff = self._cmp_checksum(blob, path) - - return (local_exists, remote_exists, diff) - - def remove(self, item): - bucket = self._get_bucket_gc(self.storage_bucket) - blob_name = self.cache_file_key(path) - blob = bucket.blob(blob_name) - blob.delete() diff --git a/dvc/cloud/http.py b/dvc/cloud/http.py deleted file mode 100644 index 506660944..000000000 --- a/dvc/cloud/http.py +++ /dev/null @@ -1,126 +0,0 @@ -import os -import requests - -from dvc.logger import Logger -from dvc.progress import progress -from dvc.utils import copyfile -from dvc.cloud.base import DataCloudError, DataCloudBase -from dvc.utils import file_md5 - - -class DataCloudHTTP(DataCloudBase): - """ - Driver for http cloud. - """ - def push(self, path): - raise Exception('Not implemented yet') - - def pull(self, path): - raise Exception('Not implemented yet') - - def remove(self, path): - raise Exception('Not implemented yet') - - def status(self, path): - raise Exception('Not implemented yet') - - @staticmethod - def _downloaded_size(fname): - """ - Check how much is already downloaded. - """ - if os.path.exists(fname): - downloaded = os.path.getsize(fname) - header = {'Range': 'bytes=%d-' % downloaded} - - Logger.debug('found existing {} file, resuming download'.format(fname)) - - return (downloaded, header) - - return (0, None) - - @staticmethod - def _get_header(req, name): - """ - Get header value from request. - """ - val = req.headers.get(name) - if val == None: - Logger.debug('\'{}\' not supported by the server'.format(name)) - - return val - - def _verify_downloaded_size(self, req, downloaded_size): - """ - Check that server supports resuming downloads. - """ - content_range = self._get_header(req, 'content-range') - if downloaded_size and content_range == None: - Logger.debug('Can\'t resume download') - return 0 - - return downloaded_size - - def _download(self, req, fname, downloaded): - """ - Download file with progress bar. - """ - mode = 'ab' if downloaded else 'wb' - name = os.path.basename(req.url) - total_length = self._get_header(req, 'content-length') - chunk_size = 1024 * 100 - - progress.update_target(name, downloaded, total_length) - - with open(fname, mode) as fobj: - for chunk in req.iter_content(chunk_size=chunk_size): - if not chunk: # filter out keep-alive new chunks - continue - - fobj.write(chunk) - downloaded += len(chunk) - progress.update_target(name, downloaded, total_length) - - - progress.finish_target(name) - - def _verify_md5(self, req, fname): - """ - Verify md5 of a downloaded file if server supports 'content-md5' header. - """ - md5 = file_md5(fname)[0] - content_md5 = self._get_header(req, 'content-md5') - - if content_md5 == None: - return True - - if md5 != content_md5: - Logger.error('Checksum mismatch') - return False - - Logger.debug('Checksum matches') - return True - - def import_data(self, url, path): - """ - Download single file from url. - """ - - tmp_file = self.tmp_file(path) - - downloaded, header = self._downloaded_size(tmp_file) - req = requests.get(url, stream=True, headers=header) - downloaded = self._verify_downloaded_size(req, downloaded) - - try: - self._download(req, tmp_file, downloaded) - except Exception as exc: - Logger.error('Failed to download "{}": {}'.format(url, exc)) - return None - - if not self._verify_md5(req, tmp_file): - return None - - os.rename(tmp_file, path) - - return path diff --git a/dvc/cloud/instance_manager.py b/dvc/cloud/instance_manager.py index 861693ae4..014b421e7 100644 --- a/dvc/cloud/instance_manager.py +++ b/dvc/cloud/instance_manager.py @@ -1,4 +1,5 @@ class CloudSettings(object): - def __init__(self, global_storage_path, cloud_config): + def __init__(self, cache_dir, global_storage_path, cloud_config): + self.cache_dir = cache_dir self.cloud_config = cloud_config self.global_storage_path = global_storage_path diff --git a/dvc/cloud/local.py b/dvc/cloud/local.py index ba6a1d73b..ab7471f15 100644 --- a/dvc/cloud/local.py +++ b/dvc/cloud/local.py @@ -6,48 +6,72 @@ from dvc.utils import copyfile, file_md5 from dvc.cloud.base import DataCloudError, DataCloudBase +class LocalKey(object): + def __init__(self, bucket, name): + self.name = name + self.bucket = bucket + + @property + def path(self): + return os.path.join(self.bucket, self.name) + + def delete(self): + os.unlink(self.path) + + class DataCloudLOCAL(DataCloudBase): """ Driver for local storage. """ - def push(self, path): - Logger.debug('sync to cloud ' + path + " " + self.storage_path) - copyfile(path, self.storage_path) + def cache_file_key(self, path): + return os.path.relpath(os.path.abspath(path), self._cloud_settings.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) + if os.path.exists(key.path) and os.path.isfile(key.path): + return key + return None + + def _get_keys(self, path): + key_name = self.cache_file_key(path) + key = LocalKey(self.storage_path, key_name) + + if not os.path.isdir(key.path): + return None + + res = [] + for root, dirs, files in os.walk(key.path): + for fname in files: + p = os.path.join(root, fname) + k_name = os.path.relpath(p, self.storage_path) + res.append(LocalKey(self.storage_path, k_name)) + return res + + def _new_key(self, path): + key_name = self.cache_file_key(path) + key = LocalKey(self.storage_path, key_name) + self._makedirs(key.path) + return key + + def _push_key(self, key, path): + self._makedirs(key.path) + copyfile(path, key.path) return path - def _import(self, bucket, i, path): - inp = os.path.join(self.storage_path, i) + def _pull_key(self, key, path): + self._makedirs(path) + tmp_file = self.tmp_file(path) try: - copyfile(inp, tmp_file) + copyfile(key.path, tmp_file) except Exception as exc: - Logger.error('Failed to copy "{}": {}'.format(i, exc)) + Logger.error('Failed to copy "{}": {}'.format(key.path, exc)) return None os.rename(tmp_file, path) return path - - def pull(self, path): - Logger.debug('sync from cloud ' + path) - return self._import(None, path, path) - - def remove(self, path): - Logger.debug('rm from cloud ' + path) - os.remove(path) - - def import_data(self, path, out): - Logger.debug('import from cloud ' + path + " " + out) - return self._import(None, path, out) - - def _status(self, path): - local = path - remote = '{}/{}'.format(self.storage_path, os.path.basename(local)) - - remote_exists = os.path.exists(remote) - local_exists = os.path.exists(local) - diff = None - if local_exists and remote_exists: - diff = filecmp.cmp(local, remote) - - return (local_exists, remote_exists, diff) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index d79b379ab..022abf4dc 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -1,4 +1,4 @@ -import dvc.cloud.data_cloud as cloud +import dvc.cloud.base as cloud from dvc.command.common.base import CmdBase class CmdDataPull(CmdBase): diff --git a/dvc/project.py b/dvc/project.py index fa7ea455a..148559c5e 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -1,4 +1,5 @@ import os +import stat import networkx as nx from dvc.logger import Logger @@ -77,7 +78,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) - self.cloud = DataCloud(self.config._config) + self.cloud = DataCloud(self.cache.cache_dir, self.config._config) @staticmethod def init(root_dir=os.curdir): @@ -208,12 +209,30 @@ class Project(object): clist.append(out.cache) return clist + def _remove_cache_file(self, cache): + os.chmod(cache, stat.S_IWRITE) + os.unlink(cache) + + def _remove_cache(self, cache): + if os.path.isfile(cache): + self._remove_cache_file(cache) + return + + for root, dirs, files in os.walk(cache, topdown=False): + for dname in dirs: + path = os.path.join(root, dname) + os.rmdir(path) + for fname in files: + path = os.path.join(root, fname) + self._remove_cache_file(path) + os.rmdir(cache) + def gc(self): clist = self._used_cache() for cache in self.cache.all(): if cache in clist: continue - os.unlink(cache) + self._remove_cache(cache) self.logger.info(u'\'{}\' was removed'.format(self.to_dvc_path(cache))) def push(self, jobs=1): @@ -221,10 +240,7 @@ class Project(object): def pull(self, jobs=1): self.cloud.pull(self._used_cache(), jobs) - for stage in self.stages(): - for out in stage.outs: - if out.use_cache: - out.link() + self.checkout() def status(self, jobs=1): return self.cloud.status(self._used_cache(), jobs) diff --git a/dvc/stage.py b/dvc/stage.py index 1de9c57a4..30098c977 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -3,6 +3,7 @@ import stat import yaml import itertools import subprocess +from checksumdir import dirhash from dvc.system import System from dvc.utils import file_md5 @@ -45,9 +46,9 @@ class CmdOutputDoesNotExistError(CmdOutputError): super(CmdOutputDoesNotExistError, self).__init__(path, 'does not exist') -class CmdOutputIsNotFileError(CmdOutputError): +class CmdOutputIsNotFileOrDirError(CmdOutputError): def __init__(self, path): - super(CmdOutputIsNotFileError, self).__init__(path, 'not a file') + super(CmdOutputIsNotFileOrDirError, self).__init__(path, 'not a file or directory') class CmdOutputAlreadyTrackedError(CmdOutputError): @@ -84,13 +85,19 @@ class Dependency(object): if state and state.mtime == self.mtime(): md5 = state.md5 else: - md5 = file_md5(self.path)[0] + md5 = self.compute_md5() return self.md5 != md5 def changed(self): return self._changed_md5() + def compute_md5(self): + if os.path.isdir(self.path): + return dirhash(self.path, hashfunc='md5') + else: + return file_md5(self.path)[0] + def mtime(self): return os.path.getmtime(self.path) @@ -100,8 +107,9 @@ class Dependency(object): def update(self): if not os.path.exists(self.path): raise CmdOutputDoesNotExistError(self.rel_path) - if not os.path.isfile(self.path): - raise CmdOutputIsNotFileError(self.path) + + if not os.path.isfile(self.path) and not os.path.isdir(self.path): + raise CmdOutputIsNotFileOrDirError(self.rel_path) state = self.project.state.get(self.path) if state and state.mtime == self.mtime() and state.inode == self.inode(): @@ -110,7 +118,7 @@ class Dependency(object): self.project.logger.debug(msg.format(self.path, md5)) self.md5 = md5 else: - self.md5 = file_md5(self.path)[0] + self.md5 = self.compute_md5() self.project.state.update(self.path, self.md5, self.mtime(), self.inode()) def dumpd(self, cwd): @@ -189,16 +197,32 @@ class Output(Dependency): return True - @staticmethod - def hardlink(src, link): + def hardlink(self, src, link): + self.project.logger.debug("creating hardlink {} -> {}".format(src, link)) System.hardlink(src, link) os.chmod(src, stat.S_IREAD) + def dir_cache(self): + res = {} + for root, dirs, files in os.walk(self.cache): + for fname in files: + path = os.path.join(root, fname) + relpath = os.path.relpath(path, self.cache) + with open(path, 'r') as fd: + d = yaml.safe_load(fd) + md5 = d[Output.PARAM_MD5] + res[relpath] = self.project.cache.get(md5) + return res + def checkout(self): if not self.use_cache: return + self.project.logger.debug("Checking out {} with cache {}".format(self.path, self.cache)) + if not self.changed(): + msg = "Data {} with cache {} didn't change" + self.project.logger.debug(msg.format(self.path, self.cache)) return if not os.path.exists(self.cache): @@ -207,45 +231,92 @@ class Output(Dependency): return if os.path.exists(self.path): + msg = "Data {} exists. Removing before checkout" + self.project.logger.debug(msg.format(self.path)) self.remove() - self.hardlink(self.cache, self.path) + if os.path.isfile(self.cache): + self.hardlink(self.cache, self.path) + return + + for relpath, cache in self.dir_cache().items(): + path = os.path.join(self.path, relpath) + os.makedirs(os.path.dirname(path)) + self.hardlink(cache, path) def save(self): if not self.use_cache: return + self.project.logger.debug("Saving {} to {}".format(self.path, self.cache)) + if not os.path.exists(self.path): raise CmdOutputDoesNotExistError(self.rel_path) - if not os.path.isfile(self.path): - raise CmdOutputIsNotFileError(self.rel_path) + if not os.path.isfile(self.path) and not os.path.isdir(self.path): + raise CmdOutputIsNotFileOrDirError(self.rel_path) if self.project.scm.is_tracked(self.path): - raise CmdOutputAlreadyTrackedError(self.path) + raise CmdOutputAlreadyTrackedError(self.rel_path) if not self.changed(): + msg = "Data {} with cache {} didn't change" + self.project.logger.debug(msg.format(self.path, self.cache)) return if os.path.exists(self.cache): # This means that we already have cache for this data. # We remove data and link it to existing cache to save # some space. - self.remove() - self.hardlink(self.cache, self.path) + msg = "Cache {} already exists, performing checkout for {}" + self.project.logger.debug(msg.format(self.cache, self.path)) + self.checkout() return - self.hardlink(self.path, self.cache) + if os.path.isfile(self.path): + self.hardlink(self.path, self.cache) + return + + for root, dirs, files in os.walk(self.path): + for fname in files: + path = os.path.join(root, fname) + relpath = os.path.relpath(path, self.path) + md5 = file_md5(path)[0] + cache = self.project.cache.get(md5) + cache_info = os.path.join(self.cache, relpath) + + self.hardlink(path, cache) + + os.makedirs(os.path.dirname(cache_info)) + with open(cache_info, 'w') as fd: + yaml.safe_dump({self.PARAM_MD5: md5}, fd, default_flow_style=False) + + def _remove(self, path, cache): + self.project.logger.debug("Removing '{}'".format(path)) + os.chmod(path, stat.S_IWUSR) + os.unlink(path) + if cache != None and os.path.exists(cache): + os.chmod(cache, stat.S_IREAD) def remove(self): if not os.path.exists(self.path): return - self.project.logger.debug("Removing '{}'".format(self.path)) - os.chmod(self.path, stat.S_IWUSR) - os.unlink(self.path) - if os.path.exists(self.cache): - os.chmod(self.cache, stat.S_IREAD) + if os.path.isfile(self.path): + self._remove(self.path, self.cache) + return + + caches = self.dir_cache() + for root, dirs, files in os.walk(self.path, topdown=False): + for d in dirs: + path = os.path.join(root, d) + os.rmdir(path) + for f in files: + path = os.path.join(root, f) + relpath = os.path.relpath(path, self.path) + cache = caches.get(relpath, None) + self._remove(path, cache) + os.rmdir(self.path) class StageCmdFailedError(DvcException): diff --git a/dvc/utils.py b/dvc/utils.py index 85729559b..55b3ee95b 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -83,7 +83,8 @@ def map_progress(func, targets, n_threads): try: ret = pool.map(func, targets) except Exception as exc: - Logger.error('Unexpected exception while processing targets: {}'.format(exc), exc_info=True) +# Logger.error('Unexpected exception while processing targets: {}'.format(exc), exc_info=True) + raise finally: progress.finish() diff --git a/requirements.txt b/requirements.txt index 822ec5f4e..978cc86ee 100644 --- a/requirements.txt +++ b/requirements.txt @@ -95,3 +95,4 @@ tinydb==3.7.0 gitpython==2.1.8 ntfsutils==0.1.3 pygraphviz==1.3.1 +checksumdir==1.1.4 diff --git a/setup.py b/setup.py index 4336fa8a2..bbbd51ff9 100644 --- a/setup.py +++ b/setup.py @@ -100,6 +100,7 @@ install_requires = [ 'gitpython==2.1.8', 'ntfsutils==0.1.3', 'pygraphviz==1.3.1', + 'checksumdir==1.1.4', ] setup( diff --git a/tests/basic_env.py b/tests/basic_env.py index 60c1bc798..6dcd98280 100644 --- a/tests/basic_env.py +++ b/tests/basic_env.py @@ -9,6 +9,12 @@ from dvc.project import Project class TestDir(TestCase): + DATA_DIR = 'data_dir' + DATA_SUB_DIR = os.path.join(DATA_DIR, 'data_sub_dir') + DATA = os.path.join(DATA_DIR, 'data') + DATA_SUB = os.path.join(DATA_SUB_DIR, 'data_sub') + DATA_CONTENTS = DATA + DATA_SUB_CONTENTS = DATA_SUB FOO = 'foo' FOO_CONTENTS = FOO BAR = 'bar' @@ -34,6 +40,10 @@ class TestDir(TestCase): self.create(self.FOO, self.FOO_CONTENTS) self.create(self.BAR, self.BAR_CONTENTS) self.create(self.CODE, self.CODE_CONTENTS) + os.mkdir(self.DATA_DIR) + os.mkdir(self.DATA_SUB_DIR) + self.create(self.DATA, self.DATA_CONTENTS) + self.create(self.DATA_SUB, self.DATA_SUB_CONTENTS) def tearDown(self): self._popd() diff --git a/tests/test_add.py b/tests/test_add.py index 90fee66ea..0f823bbfd 100644 --- a/tests/test_add.py +++ b/tests/test_add.py @@ -1,11 +1,12 @@ import os import shutil import filecmp +from checksumdir import dirhash from dvc.main import main from dvc.utils import file_md5 from dvc.stage import Stage, CmdOutputNoCacheError, CmdOutputOutsideOfRepoError -from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileError +from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileOrDirError from dvc.stage import CmdOutputAlreadyTrackedError from dvc.project import StageNotFoundError from dvc.command.add import CmdAdd @@ -43,8 +44,8 @@ class TestAddDirectory(TestDvc): def test(self): dname = 'directory' os.mkdir(dname) - with self.assertRaises(CmdOutputIsNotFileError) as cx: - self.dvc.add(dname) + self.create(os.path.join(dname, 'file'), 'file') + self.dvc.add(dname) class TestAddTrackedFile(TestDvc): diff --git a/tests/test_config.py b/tests/test_config.py index 0af5b3a58..69e08ec77 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -46,7 +46,7 @@ class TestConfigTest(TestCase): ) s = StringIO('\n'.join(c)) conf = self._conf(s) - cloud = DataCloud(conf) + cloud = DataCloud(None, conf) self.assertEqual(cloud.typ, 'AWS') self.assertEqual(cloud._cloud.storage_bucket, 'globalsb') self.assertEqual(cloud._cloud.storage_prefix, 'global_storage_path') @@ -65,7 +65,7 @@ class TestConfigTest(TestCase): ) s = StringIO('\n'.join(c)) conf = self._conf(s) - cloud = DataCloud(conf) + cloud = DataCloud(None, conf) self.assertEqual(cloud.typ, 'AWS') self.assertEqual(cloud._cloud.storage_bucket, 'awssb') self.assertEqual(cloud._cloud.storage_prefix, 'aws_storage_path') @@ -104,7 +104,7 @@ class TestConfigTest(TestCase): # patcher.start() conf = self._conf(s) - cloud = DataCloud(conf) + cloud = DataCloud(None, conf) aws_creds = cloud._cloud._get_credentials() patcher.stop() @@ -135,7 +135,7 @@ class TestConfigTest(TestCase): patcher.start() conf = self._conf(s) - cloud = DataCloud(conf) + cloud = DataCloud(None, conf) aws_creds = cloud._cloud._get_credentials() patcher.stop() diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index b4dfec7bb..a16f2f0c4 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -1,8 +1,9 @@ import os +import stat import uuid from dvc.config import ConfigError -from dvc.cloud.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudHTTP, DataCloudGCP +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 @@ -15,12 +16,12 @@ class TestDataCloud(TestDvc): config = {'Global': {'Cloud': k}, k: {'StoragePath': 'a/b', 'ProjectName': 'name'}} - cloud = DataCloud(config) + cloud = DataCloud(self.dvc.cache.cache_dir, config) self.assertIsInstance(cloud._cloud, v) with self.assertRaises(ConfigError) as cx: config = {'Global': {'Cloud': 'not_supported_type'}} - DataCloud(config) + DataCloud(self.dvc.cache.cache_dir, config) class TestDataCloudBase(TestDvc): @@ -30,45 +31,77 @@ class TestDataCloudBase(TestDvc): def _test_cloud(self): self._setup_cloud() + stage = self.dvc.add(self.FOO) + cache = stage.outs[0].cache + + stage_dir = self.dvc.add(self.DATA_DIR) + cache_dir = stage_dir.outs[0].cache + # Check status - status = self.cloud.status(self.FOO) + status = self.cloud.status(cache) self.assertEqual(status, STATUS_NEW) - # Push and check status - self.cloud.push(self.FOO) - self.assertTrue(os.path.exists(self.FOO)) - self.assertTrue(os.path.isfile(self.FOO)) + status_dir = self.cloud.status(cache_dir) + self.assertEqual(status_dir, STATUS_NEW) - status = self.cloud.status(self.FOO) + # Push and check status + self.cloud.push(cache) + self.assertTrue(os.path.exists(cache)) + self.assertTrue(os.path.isfile(cache)) + + self.cloud.push(cache_dir) + self.assertTrue(os.path.exists(cache_dir)) + self.assertTrue(os.path.isdir(cache_dir)) + + status = self.cloud.status(cache) self.assertEqual(status, STATUS_OK) + status_dir = self.cloud.status(cache_dir) + self.assertEqual(status_dir, STATUS_OK) + # Modify and check status - with open(self.FOO, 'a') as fd: + os.chmod(cache, stat.S_IWRITE) + with open(cache, 'a') as fd: fd.write('addon') - status = self.cloud.status(self.FOO) + os.chmod(cache, 0o777) + status = self.cloud.status(cache) self.assertEqual(status, STATUS_MODIFIED) # Remove and check status - os.unlink(self.FOO) - status = self.cloud.status(self.FOO) + os.unlink(cache) + status = self.cloud.status(cache) self.assertEqual(status, STATUS_DELETED) + self.dvc._remove_cache(cache_dir) + status_dir = self.cloud.status(cache_dir) + self.assertEqual(status_dir, STATUS_DELETED) + # Pull and check status - self.cloud.pull(self.FOO) - self.assertTrue(os.path.exists(self.FOO)) - self.assertTrue(os.path.isfile(self.FOO)) - with open(self.FOO, 'r') as fd: + self.cloud.pull(cache) + self.assertTrue(os.path.exists(cache)) + self.assertTrue(os.path.isfile(cache)) + with open(cache, 'r') as fd: self.assertEqual(fd.read(), self.FOO_CONTENTS) - status = self.cloud.status(self.FOO) + self.cloud.pull(cache_dir) + self.assertTrue(os.path.exists(cache_dir)) + self.assertTrue(os.path.isdir(cache_dir)) + + status = self.cloud.status(cache) self.assertEqual(status, STATUS_OK) + status_dir = self.cloud.status(cache_dir) + self.assertEqual(status_dir, STATUS_OK) + class TestDataCloudAWS(TestDataCloudBase): TEST_REPO_BUCKET = 'dvc-test' TEST_REPO_REGION = 'us-east-2' def _should_test(self): + if os.name == 'nt': + return False + if os.getenv("TRAVIS_PULL_REQUEST") == "false" and \ os.getenv("TRAVIS_SECURE_ENV_VARS") == "true": return True @@ -83,7 +116,7 @@ class TestDataCloudAWS(TestDataCloudBase): # Setup cloud config = {'StoragePath': repo, 'Region': self.TEST_REPO_REGION} - cloud_settings = CloudSettings(None, config) + cloud_settings = CloudSettings(self.dvc.cache.cache_dir, None, config) self.cloud = DataCloudAWS(cloud_settings) def test(self): @@ -96,6 +129,9 @@ class TestDataCloudGCP(TestDataCloudBase): TEST_REPO_GCP_PROJECT='dvc-project' def _should_test(self): + if os.name == 'nt': + return False + if os.getenv("TRAVIS") == "true" or os.getenv("APPVEYOR") == "True" or os.getenv("DVC_TEST_GCP") != "true": return False return True @@ -110,7 +146,7 @@ class TestDataCloudGCP(TestDataCloudBase): # Setup cloud config = {'StoragePath': self.TEST_REPO_GCP, 'Region': self.TEST_REPO_GCP_PROJECT} - cloud_settings = CloudSettings(None, config) + cloud_settings = CloudSettings(self.dvc.cache.cache_dir, None, config) self.cloud = DataCloudGCP(cloud_settings) def test(self): @@ -119,26 +155,19 @@ class TestDataCloudGCP(TestDataCloudBase): class TestDataCloudLOCAL(TestDataCloudBase): + def _should_test(self): + if os.name == 'nt': + return False + return True + def _setup_cloud(self): self.dname = 'cloud' os.mkdir(self.dname) config = {'StoragePath': self.dname} - cloud_settings = CloudSettings(None, config) + cloud_settings = CloudSettings(self.dvc.cache.cache_dir, None, config) self.cloud = DataCloudLOCAL(cloud_settings) def test(self): - self._test_cloud() - self.assertTrue(os.path.isdir(self.dname)) - path = os.path.join(self.dname, self.FOO) - self.assertTrue(os.path.isfile(path)) - - -class TestDataCloudHTTP(TestDvc): - def test(self): - cloud = DataCloudHTTP(CloudSettings(None, None)) - path = 'out' - url = 'https://github.com/dataversioncontrol/dvc/LICENSE' - ret = cloud.import_data(url, path) - self.assertEqual(ret, path) - self.assertTrue(os.path.exists(path)) - self.assertTrue(os.path.isfile(path)) + if self._should_test(): + self._test_cloud() + self.assertTrue(os.path.isdir(self.dname)) diff --git a/tests/test_gc.py b/tests/test_gc.py index a4d7e3ae9..99fcbc22c 100644 --- a/tests/test_gc.py +++ b/tests/test_gc.py @@ -5,7 +5,7 @@ import filecmp from dvc.main import main from dvc.utils import file_md5 from dvc.stage import Stage, CmdOutputNoCacheError, CmdOutputOutsideOfRepoError -from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileError +from dvc.stage import CmdOutputDoesNotExistError from dvc.project import StageNotFoundError from dvc.command.gc import CmdGC