From 0cd30e727a1879a6a897499f5d932a9c9e574b92 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 6 Aug 2017 02:40:50 +0300 Subject: [PATCH 01/33] init: check user.{name,email} to make sure git is ready Making sure git is configured at least with name and email, so our commits can be commited without any problems. Fixes #64. Signed-off-by: Ruslan Kuprieiev --- dvc/git_wrapper.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/dvc/git_wrapper.py b/dvc/git_wrapper.py index f466db785..4cc5d5e32 100644 --- a/dvc/git_wrapper.py +++ b/dvc/git_wrapper.py @@ -46,6 +46,15 @@ class GitWrapperI(object): return GitWrapper.parse_porcelain_files(out) + @staticmethod + def git_config_get(name): + code, out, err = Executor.exec_cmd(['git', 'config', '--get', name]) + Logger.debug('[dvc-git] "git config --get {}": code({}), out({}), err({})'.format( + name, code, out, err)) + if code != 0: + return None + return out + @staticmethod def git_path_to_system_path(path): if os.name == 'nt': @@ -86,6 +95,17 @@ class GitWrapper(GitWrapperI): Logger.error("{} {}".format(status, file)) return False + # Sanity check to make sure we will be able to commit + name = self.git_config_get('user.name') + if name == None: + Logger.error('[dvc-git] Please setup user.name in git config') + return False + + email = self.git_config_get('user.email') + if email == None: + Logger.error('[dvc-git] Please setup user.email in git config') + return False + return True @property From c7df4bd0ba5e34018a447021d96a1a155374a4be Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 7 Aug 2017 01:27:19 +0300 Subject: [PATCH 02/33] tests: add test for utils.py Some sanity tests to make sure something like #132 doesn't happen anymore. Signed-off-by: Ruslan Kuprieiev --- tests/test_utils.py | 60 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 tests/test_utils.py diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 000000000..0ce5b9e09 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,60 @@ +import os +import shutil +import filecmp +from random import randint +from time import sleep + +from dvc import utils +from tests.basic_env import DirHierarchyEnvironment + +class TestExecutor(DirHierarchyEnvironment): + def setUp(self): + pass + + def tearDown(self): + pass + + def test_rmtree(self): + root = 'testdir' + + os.makedirs(root + '/subdir') + with open(root + '/file1', 'w+') as f: + f.write('file1contents') + with open(root + '/subdir/file2', 'w+') as f: + f.write('file2contents') + + utils.rmtree(root) + self.assertFalse(os.path.exists(root)) + + def test_copyfile(self): + src = 'file1' + dest = 'file2' + dest_dir = 'testdir' + + with open(src, 'w+') as f: + f.write('file1contents') + + os.mkdir(dest_dir) + + utils.copyfile(src, dest) + self.assertTrue(filecmp.cmp(src, dest)) + + utils.copyfile(src, dest_dir) + self.assertTrue(filecmp.cmp(src, '{}/{}'.format(dest_dir, src))) + + shutil.rmtree(dest_dir) + os.remove(src) + os.remove(dest) + + def test_map_progress(self): + def f(target): + sleep(randint(1,2)) + with open(target, 'w+') as o: + o.write(target) + sleep(randint(1,2)) + + targets = ['map{}'.format(i) for i in range(1, 10)] + n_threads = [1, 10, 20] + + for n in n_threads: + utils.map_progress(f, targets, n) From 93326c99ad2c9724a98b394312b3f8a9cf0e1d7a Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 7 Aug 2017 02:26:26 +0300 Subject: [PATCH 03/33] tests: utils: don't forget to init env --- tests/test_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_utils.py b/tests/test_utils.py index 0ce5b9e09..ba2862a34 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -9,7 +9,7 @@ from tests.basic_env import DirHierarchyEnvironment class TestExecutor(DirHierarchyEnvironment): def setUp(self): - pass + DirHierarchyEnvironment.init_environment(self) def tearDown(self): pass From afcbb148b94cae9610b3880b1688927dd56495fb Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 7 Aug 2017 02:36:02 +0300 Subject: [PATCH 04/33] RFC: fix codeclimate warnings codeclimate noted a few similiar code chunks and it looks like it is worth fixing them. Signed-off-by: Ruslan Kuprieiev --- dvc/command/gc.py | 11 +---------- dvc/command/remove.py | 11 +---------- dvc/command/traverse.py | 16 +++++++++++++++- 3 files changed, 17 insertions(+), 21 deletions(-) diff --git a/dvc/command/gc.py b/dvc/command/gc.py index 6ff3a5051..c303fadaf 100644 --- a/dvc/command/gc.py +++ b/dvc/command/gc.py @@ -2,7 +2,6 @@ import os from dvc.command.traverse import Traverse from dvc.logger import Logger -from dvc.path.data_item import DataItemError class CmdGC(Traverse): @@ -12,11 +11,7 @@ class CmdGC(Traverse): def process_file(self, target): Logger.debug(u'[Cmd-GC] GC file {}.'.format(target)) - try: - data_item = self.settings.path_factory.existing_data_item(target) - except DataItemError: - Logger.warn(u'[Cmd-GC] Data item {} is not a valid symbolic link'.format(target)) - data_item = self.settings.path_factory.data_item(target) + data_item = self._get_data_item(target) for cache_data_item in data_item.get_all_caches(): if cache_data_item.cache.relative != data_item.resolved_cache.relative: @@ -27,10 +22,6 @@ class CmdGC(Traverse): Logger.debug(u'[Cmd-GC] GC data item {}. Success.'.format(data_item.data.relative)) pass - def _remove_cloud_cache(self, data_item): - if not self.parsed_args.keep_in_cloud: - self.cloud.remove_from_cloud(data_item) - def is_recursive(self): return self.parsed_args.recursive diff --git a/dvc/command/remove.py b/dvc/command/remove.py index 21e2f4e94..24c365467 100644 --- a/dvc/command/remove.py +++ b/dvc/command/remove.py @@ -2,7 +2,6 @@ import os from dvc.command.traverse import Traverse from dvc.logger import Logger -from dvc.path.data_item import DataItemError class CmdRemove(Traverse): @@ -12,11 +11,7 @@ class CmdRemove(Traverse): def process_file(self, target): Logger.debug(u'[Cmd-Remove] Remove file {}.'.format(target)) - try: - data_item = self.settings.path_factory.existing_data_item(target) - except DataItemError: - Logger.warn(u'[Cmd-Remove] Data file {} is not valid symbolic link'.format(target)) - data_item = self.settings.path_factory.data_item(target) + data_item = self._get_data_item(target) self._remove_cache_file(data_item) self._remove_state_file(data_item) @@ -26,10 +21,6 @@ class CmdRemove(Traverse): Logger.debug(u'[Cmd-Remove] Remove data item {}. Success.'.format(data_item.data.relative)) pass - def _remove_cloud_cache(self, data_item): - if not self.parsed_args.keep_in_cloud: - self.cloud.remove_from_cloud(data_item) - def _remove_state_file(self, data_item): if os.path.isfile(data_item.state.relative): self._remove_dvc_path(data_item.state, 'state') diff --git a/dvc/command/traverse.py b/dvc/command/traverse.py index 8dc07069a..b000611e3 100644 --- a/dvc/command/traverse.py +++ b/dvc/command/traverse.py @@ -4,7 +4,7 @@ from dvc.command.base import CmdBase, DvcLock from dvc.exceptions import DvcException from dvc.logger import Logger from dvc.data_cloud import DataCloud - +from dvc.path.data_item import DataItemError class TraverseError(DvcException): def __init__(self, msg): @@ -69,6 +69,20 @@ class Traverse(CmdBase): self.traverse_dir_finalize(target) pass + # Common helpers + def _get_data_item(self, target): + try: + data_item = self.settings.path_factory.existing_data_item(target) + except DataItemError: + Logger.warn(u'[TraverseFileTree] Data file {} is not valid symbolic link'.format(target)) + data_item = self.settings.path_factory.data_item(target) + + return data_item + + def _remove_cloud_cache(self, data_item): + if not self.parsed_args.keep_in_cloud: + self.cloud.remove_from_cloud(data_item) + ## API: def is_recursive(self): return True From 10f19f5c2e958aaef06df43f9b1324ed1f897980 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 12 Aug 2017 08:00:50 +0300 Subject: [PATCH 05/33] cloud: parse and store cloud in settings Cloud is determined from config and stays the same from that moment, so it makes sense to parse it once and store in settings, so whoever needs it, could just use self.settings.cloud. This is going to be more useful in the few next commits, where push/pull/status are introduced. Signed-off-by: Ruslan Kuprieiev --- dvc/command/data_sync.py | 4 +--- dvc/command/traverse.py | 4 +--- dvc/settings.py | 6 ++++++ 3 files changed, 8 insertions(+), 6 deletions(-) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index fbe9ff211..070365bdc 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -4,7 +4,6 @@ from multiprocessing.pool import ThreadPool from dvc.command.base import CmdBase, DvcLock from dvc.exceptions import DvcException from dvc.system import System -from dvc.data_cloud import DataCloud from dvc.utils import map_progress @@ -19,7 +18,6 @@ class CmdDataSync(CmdBase): def run(self): with DvcLock(self.is_locker, self.git): - cloud = DataCloud(self.settings) targets = [] if len(self.parsed_args.targets) == 0: @@ -35,5 +33,5 @@ class CmdDataSync(CmdBase): else: raise DataSyncError('File "{}" does not exit'.format(target)) - map_progress(cloud.sync, targets, self.parsed_args.jobs) + map_progress(self.settings.cloud.sync, targets, self.parsed_args.jobs) pass diff --git a/dvc/command/traverse.py b/dvc/command/traverse.py index b000611e3..6c3a94e23 100644 --- a/dvc/command/traverse.py +++ b/dvc/command/traverse.py @@ -3,7 +3,6 @@ import os from dvc.command.base import CmdBase, DvcLock from dvc.exceptions import DvcException from dvc.logger import Logger -from dvc.data_cloud import DataCloud from dvc.path.data_item import DataItemError class TraverseError(DvcException): @@ -15,7 +14,6 @@ class Traverse(CmdBase): def __init__(self, settings, cmd_name, do_not_start_from_root=True): super(Traverse, self).__init__(settings) self._cmd_name = cmd_name - self.cloud = DataCloud(self.settings) self._do_not_start_from_root = do_not_start_from_root def run(self): @@ -81,7 +79,7 @@ class Traverse(CmdBase): def _remove_cloud_cache(self, data_item): if not self.parsed_args.keep_in_cloud: - self.cloud.remove_from_cloud(data_item) + self.settings.cloud.remove_from_cloud(data_item) ## API: def is_recursive(self): diff --git a/dvc/settings.py b/dvc/settings.py index a20a4784a..9d93cd00e 100644 --- a/dvc/settings.py +++ b/dvc/settings.py @@ -5,6 +5,7 @@ from dvc.config import ConfigI, Config from dvc.exceptions import DvcException from dvc.path.factory import PathFactory from dvc.cli import parse_args +from dvc.data_cloud import DataCloud class SettingsError(DvcException): def __init__(self, msg): @@ -33,6 +34,7 @@ class Settings(object): self._config = config self._path_factory = PathFactory(git, config) self._parsed_args = args + self._cloud = DataCloud(self) @property def args(self): @@ -63,3 +65,7 @@ class Settings(object): @property def path_factory(self): return self._path_factory + + @property + def cloud(self): + return self._cloud From a2343119809b71beb9a52cd3649c82f0e7b0ac62 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 12 Aug 2017 08:37:58 +0300 Subject: [PATCH 06/33] sync/cloud: move target logic to DataCloud This should come in handy for pull and push commands. Signed-off-by: Ruslan Kuprieiev --- dvc/command/data_sync.py | 30 +----------------------------- dvc/data_cloud.py | 35 +++++++++++++++++++++++++++++++++-- 2 files changed, 34 insertions(+), 31 deletions(-) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 070365bdc..6ee12e64b 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -1,16 +1,4 @@ -import os -from multiprocessing.pool import ThreadPool - from dvc.command.base import CmdBase, DvcLock -from dvc.exceptions import DvcException -from dvc.system import System -from dvc.utils import map_progress - - -class DataSyncError(DvcException): - def __init__(self, msg): - super(DataSyncError, self).__init__('Data sync error: {}'.format(msg)) - class CmdDataSync(CmdBase): def __init__(self, settings): @@ -18,20 +6,4 @@ class CmdDataSync(CmdBase): def run(self): with DvcLock(self.is_locker, self.git): - targets = [] - - if len(self.parsed_args.targets) == 0: - raise DataSyncError('Sync target is not specified') - - for target in self.parsed_args.targets: - if System.islink(target): - targets.append(target) - elif os.path.isdir(target): - for root, dirs, files in os.walk(target): - for f in files: - targets.append(os.path.join(root, f)) - else: - raise DataSyncError('File "{}" does not exit'.format(target)) - - map_progress(self.settings.cloud.sync, targets, self.parsed_args.jobs) - pass + self.settings.cloud.sync(self.parsed_args.targets, self.parsed_args.jobs) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index d1aaa3c54..118c78cde 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -15,6 +15,8 @@ from dvc.config import ConfigError from dvc.progress import progress from dvc.utils import copyfile from dvc.utils import cached_property +from dvc.system import System +from dvc.utils import map_progress class DataCloudError(DvcException): @@ -434,8 +436,37 @@ class DataCloud(object): self._cloud.sanity_check() - def sync(self, fname): - return self._cloud.sync(fname) + def _collect_dir(self, d): + targets = [] + + for root, dirs, files in os.walk(d): + for f in files: + targets.append(os.path.join(root, f)) + + return targets + + def _collect_target(self, target): + if System.islink(target): + return [target] + elif os.path.isdir(target): + return self._collect_dir(target) + + Logger.warn('Target "{}" does not exist'.format(target)) + + return [] + + def _collect_targets(self, targets): + collected = [] + + for t in targets: + collected += self._collect_target(t) + + return collected + + def sync(self, targets, jobs=1): + collected = self._collect_targets(targets) + + map_progress(self._cloud.sync, collected, jobs) def remove_from_cloud(self, item): return self._cloud.remove_from_cloud(item) From db43e6bc91767f0db04d0d8efab714ea2d3a2a30 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 12 Aug 2017 20:16:33 +0300 Subject: [PATCH 07/33] dvc: add pull/push commands We already have 'sync' command, but users report it to be confusing when both local and remote files exist, so lets introduce two direct commands to implicitly push and pull from cloud. Fixes #157. Signed-off-by: Ruslan Kuprieiev --- dvc/cli.py | 35 +++++++++++++++++++++++++++-------- dvc/command/base.py | 4 ++++ dvc/command/data_sync.py | 20 +++++++++++++++++++- dvc/data_cloud.py | 29 ++++++++++++++++++++--------- 4 files changed, 70 insertions(+), 18 deletions(-) diff --git a/dvc/cli.py b/dvc/cli.py index 51bc22372..742b11aa9 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -10,7 +10,7 @@ from dvc.command.init import CmdInit from dvc.command.remove import CmdRemove from dvc.command.run import CmdRun from dvc.command.repro import CmdRepro -from dvc.command.data_sync import CmdDataSync +from dvc.command.data_sync import CmdDataSync, CmdDataPush, CmdDataPull from dvc.command.lock import CmdLock from dvc.command.gc import CmdGC from dvc.command.import_file import CmdImportFile @@ -121,22 +121,41 @@ def parse_args(argv=None): help='Arguments of a command') run_parser.set_defaults(func=CmdRun) - # Sync - sync_parser = subparsers.add_parser( - 'sync', - parents=[parent_parser], - help='Synchronize data file with cloud (cloud settings already setup.') - sync_parser.add_argument( + # Parent parser used in sync/pull/push + parent_sync_parser = argparse.ArgumentParser( + add_help=False, + parents=[parent_parser]) + parent_sync_parser.add_argument( 'targets', nargs='+', help='File or directory to sync.') - sync_parser.add_argument('-j', + parent_sync_parser.add_argument('-j', '--jobs', type=int, default=cpu_count(), help='Number of jobs to run simultaneously.') + + # Sync + sync_parser = subparsers.add_parser( + 'sync', + parents=[parent_sync_parser], + help='Synchronize data file with cloud (cloud settings already setup.') sync_parser.set_defaults(func=CmdDataSync) + # Pull + pull_parser = subparsers.add_parser( + 'pull', + parents=[parent_sync_parser], + help='Pull data files from the cloud') + pull_parser.set_defaults(func=CmdDataPull) + + # Push + push_parser = subparsers.add_parser( + 'push', + parents=[parent_sync_parser], + help='Push data files to the cloud') + push_parser.set_defaults(func=CmdDataPush) + # Repro repro_parser = subparsers.add_parser( 'repro', diff --git a/dvc/command/base.py b/dvc/command/base.py index 208939b6f..440a302e9 100644 --- a/dvc/command/base.py +++ b/dvc/command/base.py @@ -55,6 +55,10 @@ class CmdBase(object): def args(self): return self._settings.args + @property + def cloud(self): + return self._settings.cloud + @property def parsed_args(self): return self._settings._parsed_args diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 6ee12e64b..ce6c0b317 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -6,4 +6,22 @@ class CmdDataSync(CmdBase): def run(self): with DvcLock(self.is_locker, self.git): - self.settings.cloud.sync(self.parsed_args.targets, self.parsed_args.jobs) + self.cloud.sync(self.parsed_args.targets, self.parsed_args.jobs) + + +class CmdDataPull(CmdBase): + def __init__(self, settings): + super(CmdDataPull, self).__init__(settings) + + def run(self): + with DvcLock(self.is_locker, self.git): + self.cloud.pull(self.parsed_args.targets, self.parsed_args.jobs) + + +class CmdDataPush(CmdBase): + def __init__(self, settings): + super(CmdDataPush, self).__init__(settings) + + def run(self): + with DvcLock(self.is_locker, self.git): + self.cloud.push(self.parsed_args.targets, self.parsed_args.jobs) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 118c78cde..76f8297ae 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -105,15 +105,14 @@ class DataCloudBase(object): def sync_from_cloud(self, item): pass - def sync(self, fname): - item = self._settings.path_factory.data_item(fname) + def sync(self, item): if os.path.isfile(item.resolved_cache.dvc): self.sync_to_cloud(item) else: - self.create_directory(fname, item) + self.create_directory(item) self.sync_from_cloud(item) - def create_directory(self, fname, item): + def create_directory(self, item): self._lock.acquire() try: dir = os.path.dirname(item.cache.relative) @@ -125,7 +124,7 @@ class DataCloudBase(object): raise DataCloudError(u'Cannot create directory {}: {}'.format(dir, ex)) elif not os.path.isdir(dir): msg = u'File {} cannot be synced because {} is not a directory' - raise DataCloudError(msg.format(fname, dir)) + raise DataCloudError(msg.format(item.cache.relative, dir)) finally: self._lock.release() @@ -441,13 +440,16 @@ class DataCloud(object): for root, dirs, files in os.walk(d): for f in files: - targets.append(os.path.join(root, f)) + path = os.path.join(root, f) + item = self._settings.path_factory.data_item(path) + targets.append(item) return targets def _collect_target(self, target): if System.islink(target): - return [target] + item = self._settings.path_factory.data_item(target) + return [item] elif os.path.isdir(target): return self._collect_dir(target) @@ -463,10 +465,19 @@ class DataCloud(object): return collected - def sync(self, targets, jobs=1): + def _map_targets(self, f, targets, jobs): collected = self._collect_targets(targets) - map_progress(self._cloud.sync, collected, jobs) + map_progress(f, collected, jobs) + + def sync(self, targets, jobs=1): + self._map_targets(self._cloud.sync, targets, jobs) + + def push(self, targets, jobs=1): + self._map_targets(self._cloud.sync_to_cloud, targets, jobs) + + def pull(self, targets, jobs=1): + self._map_targets(self._cloud.sync_from_cloud, targets, jobs) def remove_from_cloud(self, item): return self._cloud.remove_from_cloud(item) From f21defdc6874b74fcd31a5973e1500b241bf137d Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 12 Aug 2017 20:27:21 +0300 Subject: [PATCH 08/33] RFC: cloud: rename methods New names fit much better: sync_to_cloud -> push sync_from_cloud -> pull remove_from_cloud -> remove Signed-off-by: Ruslan Kuprieiev --- dvc/command/traverse.py | 2 +- dvc/data_cloud.py | 40 ++++++++++++++++++++-------------------- 2 files changed, 21 insertions(+), 21 deletions(-) diff --git a/dvc/command/traverse.py b/dvc/command/traverse.py index 6c3a94e23..63e030521 100644 --- a/dvc/command/traverse.py +++ b/dvc/command/traverse.py @@ -79,7 +79,7 @@ class Traverse(CmdBase): def _remove_cloud_cache(self, data_item): if not self.parsed_args.keep_in_cloud: - self.settings.cloud.remove_from_cloud(data_item) + self.settings.cloud.remove(data_item) ## API: def is_recursive(self): diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 76f8297ae..fd8d08788 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -99,18 +99,18 @@ class DataCloudBase(object): def sanity_check(self): pass - def sync_to_cloud(self, item): + def push(self, item): pass - def sync_from_cloud(self, item): + def pull(self, item): pass def sync(self, item): if os.path.isfile(item.resolved_cache.dvc): - self.sync_to_cloud(item) + self.push(item) else: self.create_directory(item) - self.sync_from_cloud(item) + self.pull(item) def create_directory(self, item): self._lock.acquire() @@ -128,20 +128,20 @@ class DataCloudBase(object): finally: self._lock.release() - def remove_from_cloud(self, item): + def remove(self, item): pass class DataCloudLOCAL(DataCloudBase): - def sync_to_cloud(self, item): + def push(self, item): Logger.debug('sync to cloud ' + item.resolved_cache.dvc + " " + self.storage_path) copyfile(item.resolved_cache.dvc, self.storage_path) - def sync_from_cloud(self, item): + def pull(self, item): Logger.debug('sync from cloud ' + self.storage_path + " " + item.resolved_cache.dvc) copyfile(self.storage_path, item.resolved_cache.dvc) - def remove_from_cloud(self, item): + def remove(self, item): Logger.debug('rm from cloud ' + item.resolved_cache.dvc) os.remove(item.resolved_cache.dvc) @@ -247,7 +247,7 @@ class DataCloudAWS(DataCloudBase): raise DataCloudError('Storage path is not setup correctly') return bucket - def sync_from_cloud(self, item): + def pull(self, item): """ sync from cloud, aws version """ bucket = self._get_bucket_aws() @@ -274,8 +274,8 @@ class DataCloudAWS(DataCloudBase): progress.finish_target(os.path.basename(item.resolved_cache.relative)) Logger.info('Downloading completed') - def sync_to_cloud(self, data_item): - """ sync_to_cloud, aws version """ + def push(self, data_item): + """ push, aws version """ aws_key = self.cache_file_key(data_item.resolved_cache.dvc) bucket = self._get_bucket_aws() @@ -302,7 +302,7 @@ class DataCloudAWS(DataCloudBase): progress.finish_target(os.path.basename(data_item.resolved_cache.relative)) - def remove_from_cloud(self, data_item): + def remove(self, data_item): aws_file_name = self.cache_file_key(data_item.cache.dvc) Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name)) @@ -342,7 +342,7 @@ class DataCloudGCP(DataCloudBase): raise DataCloudError('sync up: google cloud bucket {} doesn\'t exist'.format(self.storage_bucket)) return bucket - def sync_from_cloud(self, item): + def pull(self, item): """ sync from cloud, gcp version """ bucket = self._get_bucket_gc() @@ -358,8 +358,8 @@ class DataCloudGCP(DataCloudBase): blob.download_to_filename(item.resolved_cache.dvc) Logger.info('Downloading completed') - def sync_to_cloud(self, data_item): - """ sync_to_cloud, gcp version """ + def push(self, data_item): + """ push, gcp version """ bucket = self._get_bucket_gc() blob_name = self.cache_file_key(data_item.resolved_cache.dvc) @@ -377,7 +377,7 @@ class DataCloudGCP(DataCloudBase): blob.upload_from_filename(data_item.resolved_cache.relative) Logger.info('uploading %s completed' % data_item.resolved_cache.relative) - def remove_from_cloud(self, item): + def remove(self, item): raise Exception('NOT IMPLEMENTED YET') @@ -474,10 +474,10 @@ class DataCloud(object): self._map_targets(self._cloud.sync, targets, jobs) def push(self, targets, jobs=1): - self._map_targets(self._cloud.sync_to_cloud, targets, jobs) + self._map_targets(self._cloud.push, targets, jobs) def pull(self, targets, jobs=1): - self._map_targets(self._cloud.sync_from_cloud, targets, jobs) + self._map_targets(self._cloud.pull, targets, jobs) - def remove_from_cloud(self, item): - return self._cloud.remove_from_cloud(item) + def remove(self, item): + return self._cloud.remove(item) From 178834e5726c14e7c70c8b74b58075911cab2773 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 12 Aug 2017 22:26:49 +0300 Subject: [PATCH 09/33] cloud: aws: check md5 when pulling data Currently dvc downloads data from cloud even if we already have the same file downloaded. This patch fixes it by checking md5 before attempting to download. Fixes #159 Signed-off-by: Ruslan Kuprieiev --- dvc/data_cloud.py | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index fd8d08788..7c61a4c3e 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -247,17 +247,31 @@ class DataCloudAWS(DataCloudBase): raise DataCloudError('Storage path is not setup correctly') return bucket + def _cmp_checksum(self, key, fname): + md5_cloud = key.etag[1:-1] + md5_local = file_md5(fname)[0] + + if md5_cloud == md5_local: + return True + + return False + def pull(self, item): """ sync from cloud, aws version """ bucket = self._get_bucket_aws() - key_name = self.cache_file_key(item.resolved_cache.dvc) + fname = item.resolved_cache.dvc + key_name = self.cache_file_key(fname) key = bucket.get_key(key_name) if not key: Logger.error('File "{}" does not exist in the cloud'.format(key_name)) return + if self._cmp_checksum(key, fname): + Logger.debug('File "{}" matches with "{}".'.format(fname, key_name)) + return + Logger.info('Downloading cache file from S3 "{}/{}"'.format(bucket.name, key_name)) try: @@ -283,9 +297,7 @@ class DataCloudAWS(DataCloudBase): if key: Logger.debug('File already uploaded to the cloud. Checksum validation...') - md5_cloud = key.etag[1:-1] - md5_local = file_md5(data_item.resolved_cache.dvc)[0] - if md5_cloud == md5_local: + if self._cmp_checksum(key, data_item.resolved_cache.dvc): Logger.debug('File checksum matches. No uploading is needed.') return From f34bcdd9592efd438cbe53e146d117b447bd7e8e Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 12 Aug 2017 14:38:13 -0700 Subject: [PATCH 10/33] build: cleanup build_macos.sh Signed-off-by: Ruslan Kuprieiev --- build_macos.sh | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) diff --git a/build_macos.sh b/build_macos.sh index c51eca7ac..c16938719 100755 --- a/build_macos.sh +++ b/build_macos.sh @@ -1,5 +1,7 @@ #!/bin/sh +set -e + BUILD_DIR=build INSTALL_DIR=usr/local BIN_DIR=$BUILD_DIR/$INSTALL_DIR/bin @@ -40,16 +42,6 @@ install_dependencies() pip install -r requirements.txt print_info "Installing fpm..." - if command_exists dnf; then - sudo dnf install ruby-devel gcc make rpm-build - elif command_exists yum; then - sudo yum install ruby-devel gcc make rpm-build - elif command_exists apt-get; then - sudo apt-get install ruby ruby-dev rubygems build-essential - #else - # echo "Unable to install fpm dependencies" && exit 1 - fi - gem install --no-ri --no-rdoc fpm print_info "Installing pyinstaller..." @@ -65,8 +57,7 @@ build_dvc() cleanup install_dependencies build_dvc -#fpm_build rpm fpm_build osxpkg -#cleanup +cleanup print_info "Successfully built dvc rpm and deb packages" From 45253ba225d638cb384c088ee6e89458a52c669b Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 13 Aug 2017 04:07:37 +0300 Subject: [PATCH 11/33] travis: add workaround for mac build Travis doesn't natively support python builds on Mac yet, so we need to workaround it by installing everything ourselves. Fixes #133. Signed-off-by: Ruslan Kuprieiev --- .travis.yml | 26 +++++++++++++++++++++++++- build_macos.sh | 2 +- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/.travis.yml b/.travis.yml index 0efc91eb5..d20cf3f2e 100644 --- a/.travis.yml +++ b/.travis.yml @@ -7,6 +7,29 @@ python: - "3.5" # - "3.6" Not supported by pyinstaller yet cache: pip +# Travis doesn't support testing python on Mac yet, so we need +# to workaround it by installing it directly with brew. +# +# Note that we specify language as ruby-2.4.0 to help us satisfy +# fpm dependencies, as travis runs el capitan with an old ruby +# version, that causes fpm to crash. +matrix: + include: + - os: osx + language: ruby + rvm: 2.4.0 + env: PYTHON_VER=2.7.10 + - os: osx + language: ruby + rvm: 2.4.0 + env: PYTHON_VER=3.4.3 + - os: osx + language: ruby + rvm: 2.4.0 + env: PYTHON_VER=3.5.0 +before_install: + - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then pyenv install $PYTHON_VER; fi + - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then pyenv global $PYTHON_VER; fi install: - pip install --upgrade pip - pip install -r requirements.txt @@ -16,7 +39,7 @@ install: - mkdir -p ~/.aws && printf "[default]\naws_access_key_id = ABCDEFG\naws_secret_access_key = 123456789" > ~/.aws/credentials script: - ./unittests.sh - - ./build_linux.sh + - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then ./build_macos.sh; else ./build_linux.sh; fi - ./build_package.sh after_sucess: - CODECLIMATE_REPO_TOKEN=7350efa14a67c3911af96dadaabe811b43eead7b466309b47687499bf5d72287 codeclimate-test-reporter @@ -28,6 +51,7 @@ deploy: file: - dvc*.rpm - dvc*.deb + - dvc*.pkg skip_cleanup: true on: tags: true diff --git a/build_macos.sh b/build_macos.sh index c16938719..f0c09ead8 100755 --- a/build_macos.sh +++ b/build_macos.sh @@ -60,4 +60,4 @@ build_dvc fpm_build osxpkg cleanup -print_info "Successfully built dvc rpm and deb packages" +print_info "Successfully built dvc osx package" From ebeb32a8acbcf05b33efdcc165872a0f3b7b4246 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 14 Aug 2017 00:53:24 +0300 Subject: [PATCH 12/33] import: resume download Download to .part file first and rename at the very end, so if we lose connection we could resume download. Note: not all servers support 'Range' header, so this feature is not going to work everywhere. This feature is activated by -c/--continue cmdline option. Fixes #108 Signed-off-by: Ruslan Kuprieiev --- dvc/cli.py | 6 ++++++ dvc/command/import_file.py | 34 ++++++++++++++++++++++++++++++---- 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/dvc/cli.py b/dvc/cli.py index 742b11aa9..10701d03f 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -223,6 +223,12 @@ def parse_args(argv=None): type=int, default=cpu_count(), help='Number of jobs to run simultaneously.') + import_parser.add_argument('-c', + '--continue', + dest='cont', + action='store_true', + default=False, + help='Resume downloading file from url') import_parser.set_defaults(func=CmdImportFile) # Lock diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py index 201f93cc7..981fd6f4c 100644 --- a/dvc/command/import_file.py +++ b/dvc/command/import_file.py @@ -142,21 +142,45 @@ class CmdImportFile(CmdBase): """ Download single file from url. """ - r = requests.get(from_url, stream=True) + tmp_file = to_file + '.part' name = os.path.basename(from_url) chunk_size = 1024 * 100 downloaded = 0 last_reported = 0 report_bucket = 100*1024*10 - total_length = r.headers.get('content-length') - with open(to_file, 'wb') as f: + resume_header = None + mode = 'wb' + + # Resume download if we can + if os.path.exists(tmp_file) and self.parsed_args.cont: + mode = 'ab' + + downloaded = os.path.getsize(tmp_file) + resume_header = {'Range': 'bytes=%d-' % downloaded} + + Logger.debug('found existing {} file, resuming download'.format(tmp_file)) + + r = requests.get(from_url, stream=True, headers=resume_header) + + content_range = r.headers.get('content-range') + if resume_header and content_range == None: + mode = 'wb' + downloaded = 0 + + Logger.debug('\'range\' is not supported by the server. Can\'t resume download') + + total_length = r.headers.get('content-length') + if total_length == None: + Logger.debug('\'content-length\' is not supported by the server') + + with open(tmp_file, mode) as f: for chunk in r.iter_content(chunk_size=chunk_size): if not chunk: # filter out keep-alive new chunks continue - downloaded += chunk_size + downloaded += len(chunk) last_reported += chunk_size if last_reported >= report_bucket: @@ -171,6 +195,8 @@ class CmdImportFile(CmdBase): # tell progress bar that this target is finished downloading progress.finish_target(name) + os.rename(tmp_file, to_file) + def download_targets(self, targets): """ Download targets in a number of threads. From c21261b69ffb0f5fe9736ceac0e5fd752fea7d4d Mon Sep 17 00:00:00 2001 From: Ophir Yoktan Date: Mon, 14 Aug 2017 23:27:05 +0300 Subject: [PATCH 13/33] return empty md5 if file doesnt exist --- dvc/data_cloud.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 7c61a4c3e..2adf09052 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -46,11 +46,14 @@ def create_cb(name): def file_md5(fname): """ get the (md5 hexdigest, md5 digest) of a file """ - hash_md5 = hashlib.md5() - with open(fname, "rb") as f: - for chunk in iter(lambda: f.read(1024*1000), b""): - hash_md5.update(chunk) - return (hash_md5.hexdigest(), hash_md5.digest()) + if os.path.exists(fname): + hash_md5 = hashlib.md5() + with open(fname, "rb") as f: + for chunk in iter(lambda: f.read(1024*1000), b""): + hash_md5.update(chunk) + return (hash_md5.hexdigest(), hash_md5.digest()) + else: + return (None, None) class DataCloudBase(object): From 81b39b2ad746941126fc2a3d663877a30a42be6b Mon Sep 17 00:00:00 2001 From: Ophir Yoktan Date: Tue, 15 Aug 2017 09:00:24 +0300 Subject: [PATCH 14/33] replace eq\neq to is is not none --- dvc/settings.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/dvc/settings.py b/dvc/settings.py index 9d93cd00e..9be3733f7 100644 --- a/dvc/settings.py +++ b/dvc/settings.py @@ -15,17 +15,17 @@ class SettingsError(DvcException): class Settings(object): def __init__(self, argv=None, git=None, config=None): self._args = None - args = None + args is None - if argv != None and len(argv) != 0: + if argv is not None and len(argv) != 0: args = parse_args(argv) self._args = argv[2:] - if git == None: + if git is None: git = GitWrapper() - if config == None: - if args != None and args.cmd != 'init': + if config is None: + if args is not None and args.cmd != 'init': config = Config() else: config = ConfigI() From b73798270c8573c0e7b5b11bfa4d14e831a791b0 Mon Sep 17 00:00:00 2001 From: Ophir Yoktan Date: Tue, 15 Aug 2017 09:04:38 +0300 Subject: [PATCH 15/33] replace eq\neq to is is not none - bug fix --- dvc/settings.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dvc/settings.py b/dvc/settings.py index 9be3733f7..7ac62581b 100644 --- a/dvc/settings.py +++ b/dvc/settings.py @@ -15,7 +15,7 @@ class SettingsError(DvcException): class Settings(object): def __init__(self, argv=None, git=None, config=None): self._args = None - args is None + args = None if argv is not None and len(argv) != 0: args = parse_args(argv) From 7a3b500390f6b272c8baed4ce4c2dae496de4bb9 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 16 Aug 2017 19:33:04 +0300 Subject: [PATCH 16/33] dvc: bring back frozen versions for deps Fixes #173. Signed-off-by: Ruslan Kuprieiev --- requirements.txt | 174 ++++++++++++++++++++++----------------------- setup.py | 178 ++++++++++++++++++++++++----------------------- 2 files changed, 178 insertions(+), 174 deletions(-) diff --git a/requirements.txt b/requirements.txt index 253e3144e..5ee7b96f0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,86 +1,88 @@ -altgraph -appdirs -backports.shutil-get-terminal-size -boto -cachetools -configparser -decorator -dill -enum34 -fasteners -funcsigs -future -futures -gapic-google-cloud-datastore-v1 -gapic-google-cloud-error-reporting-v1beta1 -gapic-google-cloud-logging-v2 -gapic-google-cloud-pubsub-v1 -gapic-google-cloud-spanner-admin-database-v1 -gapic-google-cloud-spanner-admin-instance-v1 -gapic-google-cloud-spanner-v1 -gapic-google-cloud-speech-v1beta1 -gapic-google-cloud-vision-v1 -google-auth -google-auth-httplib2 -google-cloud -google-cloud-bigquery -google-cloud-bigtable -google-cloud-core -google-cloud-datastore -google-cloud-dns -google-cloud-error-reporting -google-cloud-language -google-cloud-logging -google-cloud-monitoring -google-cloud-pubsub -google-cloud-resource-manager -google-cloud-runtimeconfig -google-cloud-spanner -google-cloud-speech -google-cloud-storage -google-cloud-translate -google-cloud-vision -google-gax -googleapis-common-protos -grpc-google-iam-v1 -grpcio -httplib2 -ipython -ipython-genutils -macholib -mock -modulegraph -monotonic -nose -oauth2client -packaging -pbr -pexpect -pickleshare -ply -prompt-toolkit -proto-google-cloud-datastore-v1 -proto-google-cloud-error-reporting-v1beta1 -proto-google-cloud-logging-v2 -proto-google-cloud-pubsub-v1 -proto-google-cloud-spanner-admin-database-v1 -proto-google-cloud-spanner-admin-instance-v1 -proto-google-cloud-spanner-v1 -proto-google-cloud-speech-v1beta1 -proto-google-cloud-vision-v1 -protobuf -ptyprocess -pyasn1 -pyasn1-modules -Pygments -PyInstaller -pyparsing -requests -rsa -scandir -simplegeneric -six -traitlets -wcwidth -colorama -google-compute-engine +altgraph==0.13 +appdirs==1.4.3 +backports.shutil-get-terminal-size==1.0.0 +boto==2.46.1 +cachetools==2.0.0 +configparser==3.5.0 +decorator==4.0.11 +dill==0.2.6 +enum34==1.1.6 +fasteners==0.14.1 +funcsigs==1.0.2 +future==0.16.0 +futures==3.0.5 +gapic-google-cloud-datastore-v1==0.15.3 +gapic-google-cloud-error-reporting-v1beta1==0.15.3 +gapic-google-cloud-logging-v2==0.91.3 +gapic-google-cloud-pubsub-v1==0.15.3 +gapic-google-cloud-spanner-admin-database-v1==0.15.3 +gapic-google-cloud-spanner-admin-instance-v1==0.15.3 +gapic-google-cloud-spanner-v1==0.15.3 +gapic-google-cloud-speech-v1beta1==0.15.3 +gapic-google-cloud-vision-v1==0.90.3 +google-auth==0.10.0 +google-auth-httplib2==0.0.2 +google-cloud==0.24.0 +google-cloud-bigquery==0.24.0 +google-cloud-bigtable==0.24.0 +google-cloud-core==0.24.0 +google-cloud-datastore==1.0.0 +google-cloud-dns==0.24.0 +google-cloud-error-reporting==0.24.0 +google-cloud-language==0.24.0 +google-cloud-logging==0.24.0 +google-cloud-monitoring==0.24.0 +google-cloud-pubsub==0.24.0 +google-cloud-resource-manager==0.24.0 +google-cloud-runtimeconfig==0.24.0 +google-cloud-spanner==0.24.0 +google-cloud-speech==0.24.0 +google-cloud-storage==1.0.0 +google-cloud-translate==0.24.0 +google-cloud-vision==0.24.0 +google-gax==0.15.8 +googleapis-common-protos==1.5.2 +grpc-google-iam-v1==0.11.1 +grpcio==1.2.1 +httplib2==0.10.3 +ipython==5.3.0 +ipython-genutils==0.2.0 +macholib==1.8 +mock==2.0.0 +modulegraph==0.14 +monotonic==1.3 +nose==1.3.7 +oauth2client==3.0.0 +packaging==16.8 +pathlib==1.0.1 +pathlib2==2.2.1 +pbr==3.0.0 +pexpect==4.2.1 +pickleshare==0.7.4 +ply==3.8 +prompt-toolkit==1.0.14 +proto-google-cloud-datastore-v1==0.90.3 +proto-google-cloud-error-reporting-v1beta1==0.15.3 +proto-google-cloud-logging-v2==0.91.3 +proto-google-cloud-pubsub-v1==0.15.3 +proto-google-cloud-spanner-admin-database-v1==0.15.3 +proto-google-cloud-spanner-admin-instance-v1==0.15.3 +proto-google-cloud-spanner-v1==0.15.3 +proto-google-cloud-speech-v1beta1==0.15.3 +proto-google-cloud-vision-v1==0.90.3 +protobuf==3.2.0 +ptyprocess==0.5.1 +pyasn1==0.2.3 +pyasn1-modules==0.0.8 +Pygments==2.2.0 +PyInstaller==3.2.1 +pyparsing==2.2.0 +requests==2.13.0 +rsa==3.4.2 +scandir==1.5 +simplegeneric==0.8.1 +six==1.10.0 +traitlets==4.3.2 +wcwidth==0.1.7 +colorama==0.3.9 +google-compute-engine==2.4.1 diff --git a/setup.py b/setup.py index 0821df687..faef14732 100644 --- a/setup.py +++ b/setup.py @@ -4,97 +4,99 @@ from distutils.errors import DistutilsPlatformError from dvc import VERSION install_requires = [ - 'altgraph', - 'appdirs', - 'backports.shutil-get-terminal-size', - 'boto', - 'cachetools', - 'configparser', - 'decorator', - 'dill', - 'enum34', - 'fasteners', - 'funcsigs', - 'future', - 'futures', - 'gapic-google-cloud-datastore-v1', - 'gapic-google-cloud-error-reporting-v1beta1', - 'gapic-google-cloud-logging-v2', - 'gapic-google-cloud-pubsub-v1', - 'gapic-google-cloud-spanner-admin-database-v1', - 'gapic-google-cloud-spanner-admin-instance-v1', - 'gapic-google-cloud-spanner-v1', - 'gapic-google-cloud-speech-v1beta1', - 'gapic-google-cloud-vision-v1', - 'google-auth', - 'google-auth-httplib2', - 'google-cloud', - 'google-cloud-bigquery', - 'google-cloud-bigtable', - 'google-cloud-core', - 'google-cloud-datastore', - 'google-cloud-dns', - 'google-cloud-error-reporting', - 'google-cloud-language', - 'google-cloud-logging', - 'google-cloud-monitoring', - 'google-cloud-pubsub', - 'google-cloud-resource-manager', - 'google-cloud-runtimeconfig', - 'google-cloud-spanner', - 'google-cloud-speech', - 'google-cloud-storage', - 'google-cloud-translate', - 'google-cloud-vision', - 'google-gax', - 'googleapis-common-protos', - 'grpc-google-iam-v1', - 'grpcio', - 'httplib2', - 'ipython', - 'ipython-genutils', - 'macholib', - 'mock', - 'modulegraph', - 'monotonic', - 'nose', - 'oauth2client', - 'packaging', - 'pbr', - 'pexpect', - 'pickleshare', - 'ply', - 'prompt-toolkit', - 'proto-google-cloud-datastore-v1', - 'proto-google-cloud-error-reporting-v1beta1', - 'proto-google-cloud-logging-v2', - 'proto-google-cloud-pubsub-v1', - 'proto-google-cloud-spanner-admin-database-v1', - 'proto-google-cloud-spanner-admin-instance-v1', - 'proto-google-cloud-spanner-v1', - 'proto-google-cloud-speech-v1beta1', - 'proto-google-cloud-vision-v1', - 'protobuf', - 'ptyprocess', - 'pyasn1', - 'pyasn1-modules', - 'Pygments', - 'PyInstaller', - 'pyparsing', - 'requests', - 'rsa', - 'scandir', - 'simplegeneric', - 'six', - 'traitlets', - 'wcwidth', - 'colorama', - 'google-compute-engine' + 'altgraph==0.13', + 'appdirs==1.4.3', + 'backports.shutil-get-terminal-size==1.0.0', + 'boto==2.46.1', + 'cachetools==2.0.0', + 'configparser==3.5.0', + 'decorator==4.0.11', + 'dill==0.2.6', + 'enum34==1.1.6', + 'fasteners==0.14.1', + 'funcsigs==1.0.2', + 'future==0.16.0', + 'futures==3.0.5', + 'gapic-google-cloud-datastore-v1==0.15.3', + 'gapic-google-cloud-error-reporting-v1beta1==0.15.3', + 'gapic-google-cloud-logging-v2==0.91.3', + 'gapic-google-cloud-pubsub-v1==0.15.3', + 'gapic-google-cloud-spanner-admin-database-v1==0.15.3', + 'gapic-google-cloud-spanner-admin-instance-v1==0.15.3', + 'gapic-google-cloud-spanner-v1==0.15.3', + 'gapic-google-cloud-speech-v1beta1==0.15.3', + 'gapic-google-cloud-vision-v1==0.90.3', + 'google-auth==0.10.0', + 'google-auth-httplib2==0.0.2', + 'google-cloud==0.24.0', + 'google-cloud-bigquery==0.24.0', + 'google-cloud-bigtable==0.24.0', + 'google-cloud-core==0.24.0', + 'google-cloud-datastore==1.0.0', + 'google-cloud-dns==0.24.0', + 'google-cloud-error-reporting==0.24.0', + 'google-cloud-language==0.24.0', + 'google-cloud-logging==0.24.0', + 'google-cloud-monitoring==0.24.0', + 'google-cloud-pubsub==0.24.0', + 'google-cloud-resource-manager==0.24.0', + 'google-cloud-runtimeconfig==0.24.0', + 'google-cloud-spanner==0.24.0', + 'google-cloud-speech==0.24.0', + 'google-cloud-storage==1.0.0', + 'google-cloud-translate==0.24.0', + 'google-cloud-vision==0.24.0', + 'google-gax==0.15.8', + 'googleapis-common-protos==1.5.2', + 'grpc-google-iam-v1==0.11.1', + 'grpcio==1.2.1', + 'httplib2==0.10.3', + 'ipython==5.3.0', + 'ipython-genutils==0.2.0', + 'macholib==1.8', + 'mock==2.0.0', + 'modulegraph==0.14', + 'monotonic==1.3', + 'nose==1.3.7', + 'oauth2client==3.0.0', + 'packaging==16.8', + 'pathlib==1.0.1', + 'pathlib2==2.2.1', + 'pbr==3.0.0', + 'pexpect==4.2.1', + 'pickleshare==0.7.4', + 'ply==3.8', + 'prompt-toolkit==1.0.14', + 'proto-google-cloud-datastore-v1==0.90.3', + 'proto-google-cloud-error-reporting-v1beta1==0.15.3', + 'proto-google-cloud-logging-v2==0.91.3', + 'proto-google-cloud-pubsub-v1==0.15.3', + 'proto-google-cloud-spanner-admin-database-v1==0.15.3', + 'proto-google-cloud-spanner-admin-instance-v1==0.15.3', + 'proto-google-cloud-spanner-v1==0.15.3', + 'proto-google-cloud-speech-v1beta1==0.15.3', + 'proto-google-cloud-vision-v1==0.90.3', + 'protobuf==3.2.0', + 'ptyprocess==0.5.1', + 'pyasn1==0.2.3', + 'pyasn1-modules==0.0.8', + 'Pygments==2.2.0', + 'PyInstaller==3.2.1', + 'pyparsing==2.2.0', + 'requests==2.13.0', + 'rsa==3.4.2', + 'scandir==1.5', + 'simplegeneric==0.8.1', + 'six==1.10.0', + 'traitlets==4.3.2', + 'wcwidth==0.1.7', + 'colorama==0.3.9', + 'google-compute-engine==2.4.1', ] if platform.system() == 'Darwin': - install_requires.append('appnope') - install_requires.append('py2app') + install_requires.append('appnope==0.1.0') + install_requires.append('py2app==0.12') setup( name='dvc', From 7f78331003e6f1722037b5e15462dd292acb4925 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Fri, 18 Aug 2017 10:30:36 +0300 Subject: [PATCH 17/33] import: use content-md5 to verify download Fixes #167 Signed-off-by: Ruslan Kuprieiev --- dvc/command/import_file.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py index 981fd6f4c..c1e9c6b87 100644 --- a/dvc/command/import_file.py +++ b/dvc/command/import_file.py @@ -3,7 +3,7 @@ import re import requests from dvc.command.base import CmdBase, DvcLock -from dvc.data_cloud import sizeof_fmt +from dvc.data_cloud import sizeof_fmt, file_md5 from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.state_file import StateFile @@ -195,6 +195,18 @@ class CmdImportFile(CmdBase): # tell progress bar that this target is finished downloading progress.finish_target(name) + # Verify checksum if we can + content_md5 = r.headers.get('content-md5') + if content_md5 != None: + md5 = file_md5(tmp_file)[0] + if md5 != content_md5: + Logger.error('Checksum mismatch') + return + + Logger.debug('Checksum matches') + else: + Logger.debug('\'content-md5\' is not supported by the server. Can\'t verify download') + os.rename(tmp_file, to_file) def download_targets(self, targets): From 5f7f608529731888124c59e157af5e0ca4fb8795 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 19 Aug 2017 06:24:04 +0300 Subject: [PATCH 18/33] import: import from clouds From now on, 'dvc import' supports s3 and gcp links. This patch also does some RFC to properly parse url's scheme using urllib instead of relying on hand-made regex. Code for downloding files from http urls is moved to data_cloud.py in preparation for pull/push support for http/https/ftp schemes in the future. Also added dummy progress updates(start/finish) for gcp, as i couldn't find an API call that allows callbacks to update progress properly. Fixes #49 Signed-off-by: Ruslan Kuprieiev --- dvc/command/import_file.py | 128 ++------------------ dvc/data_cloud.py | 241 ++++++++++++++++++++++++++++++++----- 2 files changed, 224 insertions(+), 145 deletions(-) diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py index c1e9c6b87..6f4dc3984 100644 --- a/dvc/command/import_file.py +++ b/dvc/command/import_file.py @@ -1,6 +1,11 @@ import os import re -import requests + +try: + from urlparse import urlparse +except ImportError: + # Python 3 + from urllib.parse import urlparse from dvc.command.base import CmdBase, DvcLock from dvc.data_cloud import sizeof_fmt, file_md5 @@ -8,8 +13,6 @@ from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.state_file import StateFile from dvc.system import System -from dvc.progress import progress -from dvc.utils import copyfile, map_progress class ImportFileError(DvcException): def __init__(self, msg): @@ -106,123 +109,22 @@ class CmdImportFile(CmdBase): if not os.path.exists(cache_dir): os.makedirs(cache_dir) - data_targets.append((input, output, data_item)) + data_targets.append((input, data_item)) return data_targets def import_files(self, targets, lock=False): data_targets = self.collect_targets(targets) - self.download_targets(data_targets) + self.cloud.import_data(data_targets, self.parsed_args.jobs) self.create_state_files(data_targets, lock) - def download_target(self, target): - """ - Download single target from url or from local path. - """ - input = target[0] - output = target[2].cache.relative - - if self.is_url(input): - Logger.debug("Downloading {} -> {}.".format(input, output)) - self.download_file(input, output) - Logger.debug("Done downloading {} -> {}.".format(input, output)) - else: - Logger.debug("Copying {} -> {}".format(input, output)) - self.copy_file(input, output) - Logger.debug("Dony copying {} -> {}".format(input, output)) - - @staticmethod - def copy_file(input, output): - """ - Copy single file from local path. - """ - copyfile(input, output) - - def download_file(self, from_url, to_file): - """ - Download single file from url. - """ - - tmp_file = to_file + '.part' - name = os.path.basename(from_url) - chunk_size = 1024 * 100 - downloaded = 0 - last_reported = 0 - report_bucket = 100*1024*10 - - resume_header = None - mode = 'wb' - - # Resume download if we can - if os.path.exists(tmp_file) and self.parsed_args.cont: - mode = 'ab' - - downloaded = os.path.getsize(tmp_file) - resume_header = {'Range': 'bytes=%d-' % downloaded} - - Logger.debug('found existing {} file, resuming download'.format(tmp_file)) - - r = requests.get(from_url, stream=True, headers=resume_header) - - content_range = r.headers.get('content-range') - if resume_header and content_range == None: - mode = 'wb' - downloaded = 0 - - Logger.debug('\'range\' is not supported by the server. Can\'t resume download') - - total_length = r.headers.get('content-length') - if total_length == None: - Logger.debug('\'content-length\' is not supported by the server') - - with open(tmp_file, mode) as f: - for chunk in r.iter_content(chunk_size=chunk_size): - if not chunk: # filter out keep-alive new chunks - continue - - downloaded += len(chunk) - - last_reported += chunk_size - if last_reported >= report_bucket: - last_reported = 0 - Logger.debug('Downloaded {}'.format(sizeof_fmt(downloaded))) - - # update progress bar - progress.update_target(name, downloaded, total_length) - - f.write(chunk) - - # tell progress bar that this target is finished downloading - progress.finish_target(name) - - # Verify checksum if we can - content_md5 = r.headers.get('content-md5') - if content_md5 != None: - md5 = file_md5(tmp_file)[0] - if md5 != content_md5: - Logger.error('Checksum mismatch') - return - - Logger.debug('Checksum matches') - else: - Logger.debug('\'content-md5\' is not supported by the server. Can\'t verify download') - - os.rename(tmp_file, to_file) - - def download_targets(self, targets): - """ - Download targets in a number of threads. - """ - map_progress(self.download_target, targets, self.parsed_args.jobs) - def create_state_files(self, targets, lock): """ Create state files for all targets. """ for t in targets: - input = t[0] - output = t[1] - data_item = t[2] + input, data_item = t + output = data_item.data.relative Logger.debug('Creating symlink {} --> {}'.format(data_item.symlink_file, data_item.data.relative)) System.symlink(data_item.symlink_file, data_item.data.relative) @@ -237,14 +139,6 @@ class CmdImportFile(CmdBase): state_file.save() Logger.debug('State file "{}" was created'.format(data_item.state.relative)) - URL_REGEX = re.compile( - r'^(?:http|ftp)s?://' # http:// or https:// - r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|' # domain... - r'localhost|' # localhost... - r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip - r'(?::\d+)?' # optional port - r'(?:/?|[/?]\S+)$', re.IGNORECASE) - @staticmethod def is_url(url): - return CmdImportFile.URL_REGEX.match(url) is not None + return len(urlparse(url).scheme) != 0 diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 2adf09052..f635e83aa 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -4,10 +4,18 @@ import os import threading import configparser import tempfile +import requests +import configparser from boto.s3.connection import S3Connection from google.cloud import storage as gc +try: + from urlparse import urlparse +except ImportError: + # Python 3 + from urllib.parse import urlparse + import dvc from dvc.logger import Logger from dvc.exceptions import DvcException @@ -99,6 +107,9 @@ class DataCloudBase(object): def cache_file_key(self, file): return '{}/{}'.format(self.storage_prefix, file).strip('/') + def tmp_file(self, fname): + return fname + '.part' + def sanity_check(self): pass @@ -134,20 +145,119 @@ class DataCloudBase(object): def remove(self, item): pass + def import_data(self, url, item): + pass + class DataCloudLOCAL(DataCloudBase): def push(self, item): Logger.debug('sync to cloud ' + item.resolved_cache.dvc + " " + self.storage_path) copyfile(item.resolved_cache.dvc, self.storage_path) + def _import(self, i, out): + tmp_file = self.tmp_file(out) + try: + copyfile(i, tmp_file) + os.rename(tmp_file, out) + except Exception as exc: + Logger.error('Failed to copy "{}": {}'.format(i, exc)) + return + def pull(self, item): Logger.debug('sync from cloud ' + self.storage_path + " " + item.resolved_cache.dvc) - copyfile(self.storage_path, item.resolved_cache.dvc) + self._import(self.storage_path, item.resolved_cache.dvc) def remove(self, item): Logger.debug('rm from cloud ' + item.resolved_cache.dvc) os.remove(item.resolved_cache.dvc) + def import_data(self, path, item): + Logger.debug('import from cloud ' + path + " " + item.cache.relative) + self._import(path, item.cache.relative) + +class DataCloudHTTP(DataCloudBase): + def push(self, item): + raise Exception('Not implemented yet') + + def pull(self, item): + raise Exception('Not implemented yet') + + def remove(self, item): + raise Exception('Not implemented yet') + + def import_data(self, url, item): + """ + Download single file from url. + """ + + to_file = item.cache.relative + + tmp_file = self.tmp_file(to_file) + name = os.path.basename(url) + chunk_size = 1024 * 100 + downloaded = 0 + last_reported = 0 + report_bucket = 100*1024*10 + + resume_header = None + mode = 'wb' + + # Resume download if we can + if os.path.exists(tmp_file) and self.parsed_args.cont: + mode = 'ab' + + downloaded = os.path.getsize(tmp_file) + resume_header = {'Range': 'bytes=%d-' % downloaded} + + Logger.debug('found existing {} file, resuming download'.format(tmp_file)) + + r = requests.get(url, stream=True, headers=resume_header) + + content_range = r.headers.get('content-range') + if resume_header and content_range == None: + mode = 'wb' + downloaded = 0 + + Logger.debug('\'range\' is not supported by the server. Can\'t resume download') + + total_length = r.headers.get('content-length') + if total_length == None: + Logger.debug('\'content-length\' is not supported by the server') + + with open(tmp_file, mode) as f: + for chunk in r.iter_content(chunk_size=chunk_size): + if not chunk: # filter out keep-alive new chunks + continue + + downloaded += len(chunk) + + last_reported += chunk_size + if last_reported >= report_bucket: + last_reported = 0 + Logger.debug('Downloaded {}'.format(sizeof_fmt(downloaded))) + + # update progress bar + progress.update_target(name, downloaded, total_length) + + f.write(chunk) + + # tell progress bar that this target is finished downloading + progress.finish_target(name) + + # Verify checksum if we can + content_md5 = r.headers.get('content-md5') + if content_md5 != None: + md5 = file_md5(tmp_file)[0] + if md5 != content_md5: + Logger.error('Checksum mismatch') + return + + Logger.debug('Checksum matches') + else: + Logger.debug('\'content-md5\' is not supported by the server. Can\'t verify download') + + os.rename(tmp_file, to_file) + class DataCloudAWS(DataCloudBase): """ DataCloud class for Amazon Web Services """ @@ -238,13 +348,12 @@ class DataCloudAWS(DataCloudBase): Logger.info("can't find aws credetials, assuming envirment variables or iam role") self._aws_creds = creds - def _get_bucket_aws(self): + def _get_bucket_aws(self, bucket_name): """ get a bucket object, aws """ if all([self.aws_access_key_id, self.aws_secret_access_key, self.aws_region_host]): conn = S3Connection(self.aws_access_key_id, self.aws_secret_access_key, host=self.aws_region_host) else: conn = S3Connection() - bucket_name = self.storage_bucket bucket = conn.lookup(bucket_name) if bucket is None: raise DataCloudError('Storage path is not setup correctly') @@ -259,13 +368,12 @@ class DataCloudAWS(DataCloudBase): return False - def pull(self, item): - """ sync from cloud, aws version """ + def _import(self, bucket_name, key_name, fname): - bucket = self._get_bucket_aws() + bucket = self._get_bucket_aws(bucket_name) - fname = item.resolved_cache.dvc - key_name = self.cache_file_key(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)) @@ -275,27 +383,38 @@ class DataCloudAWS(DataCloudBase): Logger.debug('File "{}" matches with "{}".'.format(fname, key_name)) return - Logger.info('Downloading cache file from S3 "{}/{}"'.format(bucket.name, key_name)) + Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name, key_name, fname)) try: - temp_file = tempfile.NamedTemporaryFile(dir=item.resolved_cache.dirname, delete=False) - key.get_contents_to_filename(temp_file.name, - cb=create_cb(item.resolved_cache.relative)) - os.rename(temp_file.name, item.resolved_cache.relative) + key.get_contents_to_filename(tmp_file, cb=create_cb(name)) + os.rename(tmp_file, fname) except Exception as exc: Logger.error('Failed to download "{}": {}'.format(key_name, exc)) - if temp_file and os.path.exists(temp_file.name): - os.remove(temp_file.name) return - progress.finish_target(os.path.basename(item.resolved_cache.relative)) - Logger.info('Downloading completed') + progress.finish_target(name) + Logger.debug('Downloading completed') + + def pull(self, data_item): + """ pull, aws version """ + + fname = data_item.resolved_cache.dvc + key_name = self.cache_file_key(fname) + self._import(self.storage_bucket, key_name, fname) + + def import_data(self, url, item): + """ import, aws version """ + + o = urlparse(url) + assert o.scheme == 's3' + + self._import(o.netloc, o.path, item.cache.relative) def push(self, data_item): """ push, aws version """ aws_key = self.cache_file_key(data_item.resolved_cache.dvc) - bucket = self._get_bucket_aws() + 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...') @@ -349,19 +468,20 @@ class DataCloudGCP(DataCloudBase): if project is None or len(project) < 1: raise ConfigError('can\'t read google cloud project name. Please set ProjectName in section GC.') - def _get_bucket_gc(self): + def _get_bucket_gc(self, storage_bucket): """ get a bucket object, gc """ client = gc.Client(project=self.gc_project_name) - bucket = client.bucket(self.storage_bucket) + bucket = client.bucket(storage_bucket) if not bucket.exists(): raise DataCloudError('sync up: google cloud bucket {} doesn\'t exist'.format(self.storage_bucket)) return bucket - def pull(self, item): - """ sync from cloud, gcp version """ + def _import(self, bucket_name, key, fname): - bucket = self._get_bucket_gc() - key = self.cache_file_key(item.resolved_cache.dvc) + bucket = self._get_bucket_gc(bucket_name) + + name = os.path.basename(fname) + tmp_file = self.tmp_file(fname) blob = bucket.get_blob(key) if not blob: @@ -370,14 +490,42 @@ class DataCloudGCP(DataCloudBase): Logger.info('Downloading cache file from gc "{}/{}"'.format(bucket.name, key)) - blob.download_to_filename(item.resolved_cache.dvc) + # 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) + os.rename(tmp_file, fname) + except Exception as exc: + Logger.error('Failed to download "{}": {}'.format(key, exc)) + return + + progress.finish_target(name) + Logger.info('Downloading completed') + def pull(self, item): + """ pull, gcp version """ + + fname = item.resolved_cache.dvc + key_name = self.cache_file_key(fname) + self._import(self.storage_bucket, key_name, data_item) + + def import_data(self, url, item): + """ import, gcp version """ + + o = urlparse(url) + assert o.scheme == 'gs' + + self._import(o.netloc, o.path, item.cache.relative) + def push(self, data_item): """ push, gcp version """ bucket = self._get_bucket_gc() blob_name = self.cache_file_key(data_item.resolved_cache.dvc) + name = os.path.basename(data_item.resolved_cache.dvc) blob = bucket.get_blob(blob_name) if blob is not None and blob.exists(): @@ -388,8 +536,13 @@ class DataCloudGCP(DataCloudBase): return Logger.debug('checksum %s mismatch. re-uploading' % data_item.cache.relative) + # same as in _import + progress.update_target(name, 0, None) + blob = bucket.blob(blob_name) blob.upload_from_filename(data_item.resolved_cache.relative) + + progress.finish_target(name) Logger.info('uploading %s completed' % data_item.resolved_cache.relative) def remove(self, item): @@ -400,9 +553,19 @@ class DataCloud(object): """ Generic class to do initial config parsing and redirect to proper DataCloud methods """ CLOUD_MAP = { - 'AWS': DataCloudAWS, - 'GCP': DataCloudGCP, - 'LOCAL': DataCloudLOCAL, + 'AWS' : DataCloudAWS, + 'GCP' : DataCloudGCP, + 'HTTP' : DataCloudHTTP, + 'LOCAL' : DataCloudLOCAL, + } + + SCHEME_MAP = { + 's3' : 'AWS', + 'http' : 'HTTP', + 'https' : 'HTTP', + 'ftp' : 'HTTP', + 'gs' : 'GCP', + '' : 'LOCAL', } def __init__(self, settings): @@ -410,6 +573,8 @@ class DataCloud(object): #To handle ConfigI case if not hasattr(settings.config, '_config'): + self._settings = settings + self._config = configparser.ConfigParser() self._cloud = DataCloudBase(None, None, None) return @@ -485,6 +650,23 @@ class DataCloud(object): map_progress(f, collected, jobs) + def _import(self, target): + url, item = target + o = urlparse(url) + + typ = self.SCHEME_MAP.get(o.scheme, None) + if typ == None: + Logger.error('Not supported scheme \'{}\''.format(o.scheme)) + return + + cloud_config = None + if self._config.has_section(typ): + cloud_config = self._config[typ] + + cloud = self.CLOUD_MAP[typ](self._settings, self._config, cloud_config) + + cloud.import_data(url, item) + def sync(self, targets, jobs=1): self._map_targets(self._cloud.sync, targets, jobs) @@ -494,5 +676,8 @@ class DataCloud(object): def pull(self, targets, jobs=1): self._map_targets(self._cloud.pull, targets, jobs) + def import_data(self, targets, jobs=1): + map_progress(self._import, targets, jobs) + def remove(self, item): return self._cloud.remove(item) From ec76c1aa8a4a5c1c92ab9926a54cb0b90e2a6407 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 19 Aug 2017 14:15:12 +0300 Subject: [PATCH 19/33] utils: collect return value from map Zip it together with the targets list and return back to the caller so he can use it if he wants. Signed-off-by: Ruslan Kuprieiev --- dvc/utils.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/dvc/utils.py b/dvc/utils.py index bb4e8e11f..f2045f76b 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -68,8 +68,10 @@ def map_progress(func, targets, n_threads): p = ThreadPool(processes=n_threads) try: - p.map(func, targets) + ret = p.map(func, targets) except Exception as exc: Logger.error('Unexpected exception while processing targets: {}'.format(exc)) finally: progress.finish() + + return zip(targets, ret) From 149c921576d77c01fac7f8f5e2dc80ffea01f5e9 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 19 Aug 2017 14:52:47 +0300 Subject: [PATCH 20/33] dvc: add status command This command shows status for data files. I.e. $ dvc status data/ new file: .cache/Posts-train.tsv_2ccd8e7 new file: data/llss modified: .cache/Posts.xml_512a612 new file: .cache/empty_0000000 new file: data/oss new file: .cache/matrix-test.p_69b924d new file: .cache/dir21111/Posts.xml.tgz_cd71d01 so it will tell us if there are any new/modified/deleted data files that you might want to push/pull/etc. Fixes #158 Signed-off-by: Ruslan Kuprieiev --- dvc/cli.py | 9 +++- dvc/command/data_sync.py | 25 ++++++++++ dvc/data_cloud.py | 102 +++++++++++++++++++++++++++++++++++++-- 3 files changed, 130 insertions(+), 6 deletions(-) diff --git a/dvc/cli.py b/dvc/cli.py index 10701d03f..5711dd11d 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -10,7 +10,7 @@ from dvc.command.init import CmdInit from dvc.command.remove import CmdRemove from dvc.command.run import CmdRun from dvc.command.repro import CmdRepro -from dvc.command.data_sync import CmdDataSync, CmdDataPush, CmdDataPull +from dvc.command.data_sync import CmdDataSync, CmdDataPush, CmdDataPull, CmdDataStatus from dvc.command.lock import CmdLock from dvc.command.gc import CmdGC from dvc.command.import_file import CmdImportFile @@ -156,6 +156,13 @@ def parse_args(argv=None): help='Push data files to the cloud') push_parser.set_defaults(func=CmdDataPush) + # Status + status_parser = subparsers.add_parser( + 'status', + parents=[parent_sync_parser], + help='Show status for data files') + status_parser.set_defaults(func=CmdDataStatus) + # Repro repro_parser = subparsers.add_parser( 'repro', diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index ce6c0b317..9b377c7b2 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -1,4 +1,7 @@ from dvc.command.base import CmdBase, DvcLock +from dvc.logger import Logger + +import dvc.data_cloud as cloud class CmdDataSync(CmdBase): def __init__(self, settings): @@ -25,3 +28,25 @@ class CmdDataPush(CmdBase): def run(self): with DvcLock(self.is_locker, self.git): self.cloud.push(self.parsed_args.targets, self.parsed_args.jobs) + +class CmdDataStatus(CmdBase): + def __init__(self, settings): + super(CmdDataStatus, self).__init__(settings) + + def run(self): + with DvcLock(self.is_locker, self.git): + status = self.cloud.status(self.parsed_args.targets, self.parsed_args.jobs) + + for s in status: + target, ret = s + + if ret == cloud.STATUS_UNKNOWN or ret == cloud.STATUS_OK: + continue + + prefix_map = { + cloud.STATUS_DELETED : 'deleted: ', + cloud.STATUS_MODIFIED : 'modified:', + cloud.STATUS_NEW : 'new file:', + } + + Logger.info('\t{}\t{}'.format(prefix_map[ret], target.resolved_cache.dvc)) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index f635e83aa..554fd0349 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -6,6 +6,7 @@ import configparser import tempfile import requests import configparser +import filecmp from boto.s3.connection import S3Connection from google.cloud import storage as gc @@ -27,6 +28,12 @@ from dvc.system import System from dvc.utils import map_progress +STATUS_UNKNOWN = 0 +STATUS_OK = 1 +STATUS_MODIFIED = 2 +STATUS_NEW = 3 +STATUS_DELETED = 4 + class DataCloudError(DvcException): def __init__(self, msg): super(DataCloudError, self).__init__('Data sync error: {}'.format(msg)) @@ -148,6 +155,9 @@ class DataCloudBase(object): def import_data(self, url, item): pass + def status(self, item): + pass + class DataCloudLOCAL(DataCloudBase): def push(self, item): @@ -175,6 +185,28 @@ class DataCloudLOCAL(DataCloudBase): Logger.debug('import from cloud ' + path + " " + item.cache.relative) self._import(path, item.cache.relative) + def status(self, data_item): + local = data_item.resolved_cache.relative + remote = '{}/{}'.format(self.storage_path, os.path.basename(local)) + + remote_exists = os.path.exists(remove) + local_exists = os.path.exists(local) + + if remote_exists and not local_exists: + return STATUS_DELETED + + if not remote_exists and local_exists: + return STATUS_NEW + + if remote_exists and local_exists: + if filecmp.cmp(local, remote): + return STATUS_OK + else: + return STATUS_MODIFIED + + return STATUS_UNKNOWN + + class DataCloudHTTP(DataCloudBase): def push(self, item): raise Exception('Not implemented yet') @@ -185,6 +217,9 @@ class DataCloudHTTP(DataCloudBase): def remove(self, item): raise Exception('Not implemented yet') + def status(self, item): + raise Exception('Not implemented yet') + def import_data(self, url, item): """ Download single file from url. @@ -436,6 +471,30 @@ class DataCloudAWS(DataCloudBase): progress.finish_target(os.path.basename(data_item.resolved_cache.relative)) + def status(self, data_item): + """ status, aws version """ + + aws_key = self.cache_file_key(data_item.resolved_cache.dvc) + 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(data_item.resolved_cache.relative) + + if remote_exists and not local_exists: + return STATUS_DELETED + + if not remote_exists and local_exists: + return STATUS_NEW + + if remote_exists and local_exists: + if self._cmp_checksum(key, data_item.resolved_cache.dvc): + return STATUS_OK + else: + return STATUS_MODIFIED + + return STATUS_UNKNOWN + def remove(self, data_item): aws_file_name = self.cache_file_key(data_item.cache.dvc) @@ -476,6 +535,14 @@ class DataCloudGCP(DataCloudBase): raise DataCloudError('sync up: google cloud bucket {} doesn\'t exist'.format(self.storage_bucket)) return bucket + def _cmp_checksum(self, blob, fname): + b64_encoded_md5 = base64.b64encode(file_md5(fname)[1]) + + if blob.md5_hash == b64_encoded_md5: + return True + + return False + def _import(self, bucket_name, key, fname): bucket = self._get_bucket_gc(bucket_name) @@ -523,15 +590,13 @@ class DataCloudGCP(DataCloudBase): def push(self, data_item): """ push, gcp version """ - bucket = self._get_bucket_gc() + bucket = self._get_bucket_gc(self.storage_bucket) blob_name = self.cache_file_key(data_item.resolved_cache.dvc) name = os.path.basename(data_item.resolved_cache.dvc) blob = bucket.get_blob(blob_name) if blob is not None and blob.exists(): - b64_encoded_md5 = base64.b64encode(file_md5(data_item.resolved_cache.dvc)[1]) - - if blob.md5_hash == b64_encoded_md5: + if self._cmp_checksum(blob, data_item.resolved_cache.dvc): Logger.debug('checksum %s matches. Skipping upload' % data_item.cache.relative) return Logger.debug('checksum %s mismatch. re-uploading' % data_item.cache.relative) @@ -545,6 +610,30 @@ class DataCloudGCP(DataCloudBase): progress.finish_target(name) Logger.info('uploading %s completed' % data_item.resolved_cache.relative) + def status(self, data_item): + """ status, gcp version """ + + bucket = self._get_bucket_gc(self.storage_bucket) + blob_name = self.cache_file_key(data_item.resolved_cache.dvc) + blob = bucket.get_blob(blob_name) + + remote_exists = blob is not None and blob.exists() + local_exists = os.path.exists(data_item.resolved_cache.relative) + + if remote_exists and not local_exists: + return STATUS_DELETED + + if not remote_exists and local_exists: + return STATUS_NEW + + if remote_exists and local_exists: + if self._cmp_checksum(blob, data_item.resolved_cache.dvc): + return STATUS_OK + else: + return STATUS_MODIFIED + + return STATUS_UNKNOWN + def remove(self, item): raise Exception('NOT IMPLEMENTED YET') @@ -648,7 +737,7 @@ class DataCloud(object): def _map_targets(self, f, targets, jobs): collected = self._collect_targets(targets) - map_progress(f, collected, jobs) + return map_progress(f, collected, jobs) def _import(self, target): url, item = target @@ -681,3 +770,6 @@ class DataCloud(object): def remove(self, item): return self._cloud.remove(item) + + def status(self, targets, jobs=1): + return self._map_targets(self._cloud.status, targets, jobs) From c3a43c6d84746303c53b1cdbf2005b3bc4e09925 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 19 Aug 2017 15:19:09 +0300 Subject: [PATCH 21/33] cloud: return data_item in sync/pull/push/import This will help us distinguish if target was processed successfully. Signed-off-by: Ruslan Kuprieiev --- dvc/data_cloud.py | 69 ++++++++++++++++++++++++++++------------------- 1 file changed, 42 insertions(+), 27 deletions(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 554fd0349..62c072212 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -128,10 +128,10 @@ class DataCloudBase(object): def sync(self, item): if os.path.isfile(item.resolved_cache.dvc): - self.push(item) + return self.push(item) else: self.create_directory(item) - self.pull(item) + return self.pull(item) def create_directory(self, item): self._lock.acquire() @@ -163,19 +163,22 @@ class DataCloudLOCAL(DataCloudBase): def push(self, item): Logger.debug('sync to cloud ' + item.resolved_cache.dvc + " " + self.storage_path) copyfile(item.resolved_cache.dvc, self.storage_path) + return item - def _import(self, i, out): + def _import(self, i, out, item): tmp_file = self.tmp_file(out) try: copyfile(i, tmp_file) os.rename(tmp_file, out) except Exception as exc: Logger.error('Failed to copy "{}": {}'.format(i, exc)) - return + return None + + return item def pull(self, item): Logger.debug('sync from cloud ' + self.storage_path + " " + item.resolved_cache.dvc) - self._import(self.storage_path, item.resolved_cache.dvc) + return self._import(self.storage_path, item.resolved_cache.dvc, item) def remove(self, item): Logger.debug('rm from cloud ' + item.resolved_cache.dvc) @@ -183,7 +186,7 @@ class DataCloudLOCAL(DataCloudBase): def import_data(self, path, item): Logger.debug('import from cloud ' + path + " " + item.cache.relative) - self._import(path, item.cache.relative) + return self._import(path, item.cache.relative, item) def status(self, data_item): local = data_item.resolved_cache.relative @@ -285,7 +288,7 @@ class DataCloudHTTP(DataCloudBase): md5 = file_md5(tmp_file)[0] if md5 != content_md5: Logger.error('Checksum mismatch') - return + return None Logger.debug('Checksum matches') else: @@ -293,6 +296,8 @@ class DataCloudHTTP(DataCloudBase): os.rename(tmp_file, to_file) + return item + class DataCloudAWS(DataCloudBase): """ DataCloud class for Amazon Web Services """ @@ -403,7 +408,7 @@ class DataCloudAWS(DataCloudBase): return False - def _import(self, bucket_name, key_name, fname): + def _import(self, bucket_name, key_name, fname, data_item): bucket = self._get_bucket_aws(bucket_name) @@ -412,11 +417,11 @@ class DataCloudAWS(DataCloudBase): key = bucket.get_key(key_name) if not key: Logger.error('File "{}" does not exist in the cloud'.format(key_name)) - return + return None if self._cmp_checksum(key, fname): Logger.debug('File "{}" matches with "{}".'.format(fname, key_name)) - return + return data_item Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name, key_name, fname)) @@ -425,17 +430,20 @@ class DataCloudAWS(DataCloudBase): os.rename(tmp_file, fname) except Exception as exc: Logger.error('Failed to download "{}": {}'.format(key_name, exc)) - return + return None progress.finish_target(name) Logger.debug('Downloading completed') + return data_item + def pull(self, data_item): """ pull, aws version """ fname = data_item.resolved_cache.dvc key_name = self.cache_file_key(fname) - self._import(self.storage_bucket, key_name, fname) + + return self._import(self.storage_bucket, key_name, fname, data_item) def import_data(self, url, item): """ import, aws version """ @@ -443,7 +451,7 @@ class DataCloudAWS(DataCloudBase): o = urlparse(url) assert o.scheme == 's3' - self._import(o.netloc, o.path, item.cache.relative) + return self._import(o.netloc, o.path, item.cache.relative, item) def push(self, data_item): """ push, aws version """ @@ -456,7 +464,7 @@ class DataCloudAWS(DataCloudBase): if self._cmp_checksum(key, data_item.resolved_cache.dvc): Logger.debug('File checksum matches. No uploading is needed.') - return + return data_item Logger.debug('Checksum miss-match. Re-uploading is required.') @@ -467,10 +475,12 @@ class DataCloudAWS(DataCloudBase): cb=create_cb(data_item.resolved_cache.relative)) except Exception as exc: Logger.error('Failed to upload "{}": {}'.format(data_item.resolved_cache.relative, exc)) - return + return None progress.finish_target(os.path.basename(data_item.resolved_cache.relative)) + return data_item + def status(self, data_item): """ status, aws version """ @@ -543,7 +553,7 @@ class DataCloudGCP(DataCloudBase): return False - def _import(self, bucket_name, key, fname): + def _import(self, bucket_name, key, fname, data_item): bucket = self._get_bucket_gc(bucket_name) @@ -553,7 +563,7 @@ class DataCloudGCP(DataCloudBase): blob = bucket.get_blob(key) if not blob: Logger.error('File "{}" does not exist in the cloud'.format(key)) - return + return None Logger.info('Downloading cache file from gc "{}/{}"'.format(bucket.name, key)) @@ -566,18 +576,21 @@ class DataCloudGCP(DataCloudBase): os.rename(tmp_file, fname) except Exception as exc: Logger.error('Failed to download "{}": {}'.format(key, exc)) - return + return None progress.finish_target(name) Logger.info('Downloading completed') + return data_item + def pull(self, item): """ pull, gcp version """ fname = item.resolved_cache.dvc key_name = self.cache_file_key(fname) - self._import(self.storage_bucket, key_name, data_item) + + return self._import(self.storage_bucket, key_name, fname, data_item) def import_data(self, url, item): """ import, gcp version """ @@ -585,7 +598,7 @@ class DataCloudGCP(DataCloudBase): o = urlparse(url) assert o.scheme == 'gs' - self._import(o.netloc, o.path, item.cache.relative) + return self._import(o.netloc, o.path, item.cache.relative, item) def push(self, data_item): """ push, gcp version """ @@ -598,7 +611,7 @@ class DataCloudGCP(DataCloudBase): if blob is not None and blob.exists(): if self._cmp_checksum(blob, data_item.resolved_cache.dvc): Logger.debug('checksum %s matches. Skipping upload' % data_item.cache.relative) - return + return data_item Logger.debug('checksum %s mismatch. re-uploading' % data_item.cache.relative) # same as in _import @@ -610,6 +623,8 @@ class DataCloudGCP(DataCloudBase): progress.finish_target(name) Logger.info('uploading %s completed' % data_item.resolved_cache.relative) + return data_item + def status(self, data_item): """ status, gcp version """ @@ -746,7 +761,7 @@ class DataCloud(object): typ = self.SCHEME_MAP.get(o.scheme, None) if typ == None: Logger.error('Not supported scheme \'{}\''.format(o.scheme)) - return + return None cloud_config = None if self._config.has_section(typ): @@ -754,19 +769,19 @@ class DataCloud(object): cloud = self.CLOUD_MAP[typ](self._settings, self._config, cloud_config) - cloud.import_data(url, item) + return cloud.import_data(url, item) def sync(self, targets, jobs=1): - self._map_targets(self._cloud.sync, targets, jobs) + return self._map_targets(self._cloud.sync, targets, jobs) def push(self, targets, jobs=1): - self._map_targets(self._cloud.push, targets, jobs) + return self._map_targets(self._cloud.push, targets, jobs) def pull(self, targets, jobs=1): - self._map_targets(self._cloud.pull, targets, jobs) + return self._map_targets(self._cloud.pull, targets, jobs) def import_data(self, targets, jobs=1): - map_progress(self._import, targets, jobs) + return map_progress(self._import, targets, jobs) def remove(self, item): return self._cloud.remove(item) From 3acc2146fa3bec91d33261049e665a760bb564a2 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 19 Aug 2017 15:42:47 +0300 Subject: [PATCH 22/33] data_item: handle __eq__ if other == None Currently it is not possible to check if data_item == None, as __eq__ tries to access .data in 'None' object and thus throws an exception. Signed-off-by: Ruslan Kuprieiev --- dvc/path/data_item.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/dvc/path/data_item.py b/dvc/path/data_item.py index 04fb341a1..90e2fbff4 100644 --- a/dvc/path/data_item.py +++ b/dvc/path/data_item.py @@ -62,6 +62,9 @@ class DataItem(object): return self.data.dvc.__hash__() def __eq__(self, other): + if other == None: + return False + return self.data.dvc == other.data.dvc @property From 4fe831712e93e01344bb78d214caec9969c84151 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 19 Aug 2017 15:23:26 +0300 Subject: [PATCH 23/33] import: don't create state files for failed imports Signed-off-by: Ruslan Kuprieiev --- dvc/command/import_file.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py index 6f4dc3984..e6530bf8b 100644 --- a/dvc/command/import_file.py +++ b/dvc/command/import_file.py @@ -115,17 +115,22 @@ class CmdImportFile(CmdBase): def import_files(self, targets, lock=False): data_targets = self.collect_targets(targets) - self.cloud.import_data(data_targets, self.parsed_args.jobs) - self.create_state_files(data_targets, lock) + imported_targets = self.cloud.import_data(data_targets, self.parsed_args.jobs) + self.create_state_files(imported_targets, lock) def create_state_files(self, targets, lock): """ Create state files for all targets. """ for t in targets: - input, data_item = t + orig_target, processed_data_item = t + input, data_item = orig_target output = data_item.data.relative + if processed_data_item == None: + Logger.debug('Skipping creating state file for failed import {}'.format(data_item.state.relative)) + continue + Logger.debug('Creating symlink {} --> {}'.format(data_item.symlink_file, data_item.data.relative)) System.symlink(data_item.symlink_file, data_item.data.relative) From eb533f9a9ca697457c19f24ba7980bb28b3ea642 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 19 Aug 2017 20:17:36 +0300 Subject: [PATCH 24/33] RFC: fix issues reported by codeclimate Signed-off-by: Ruslan Kuprieiev --- dvc/data_cloud.py | 244 ++++++++++++++++++++-------------------------- 1 file changed, 105 insertions(+), 139 deletions(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 62c072212..c79c0e186 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -34,6 +34,14 @@ STATUS_MODIFIED = 2 STATUS_NEW = 3 STATUS_DELETED = 4 +STATUS_MAP = { + # (local_exists, remote_exists, cmp) + (True, True, True) : STATUS_OK, + (True, True, False) : STATUS_MODIFIED, + (True, False, None) : STATUS_NEW, + (False, True, None) : STATUS_DELETED, +} + class DataCloudError(DvcException): def __init__(self, msg): super(DataCloudError, self).__init__('Data sync error: {}'.format(msg)) @@ -120,11 +128,22 @@ class DataCloudBase(object): def sanity_check(self): pass + def _import(self, bucket, path, fname, item): + pass + def push(self, item): pass def pull(self, item): - pass + fname = item.resolved_cache.dvc + key_name = self.cache_file_key(fname) + + return self._import(self.storage_bucket, key_name, fname, data_item) + + def import_data(self, url, item): + o = urlparse(url) + + return self._import(o.netloc, o.path, item.cache.relative, item) def sync(self, item): if os.path.isfile(item.resolved_cache.dvc): @@ -152,12 +171,11 @@ class DataCloudBase(object): def remove(self, item): pass - def import_data(self, url, item): + def _status(self, item): pass def status(self, item): - pass - + return STATUS_MAP.get(self._status(item), STATUS_UNKNOWN) class DataCloudLOCAL(DataCloudBase): def push(self, item): @@ -188,27 +206,17 @@ class DataCloudLOCAL(DataCloudBase): Logger.debug('import from cloud ' + path + " " + item.cache.relative) return self._import(path, item.cache.relative, item) - def status(self, data_item): + def _status(self, data_item): local = data_item.resolved_cache.relative remote = '{}/{}'.format(self.storage_path, os.path.basename(local)) remote_exists = os.path.exists(remove) local_exists = os.path.exists(local) + c = None + if local_exists and remote_exists: + c = filecmp.cmp(local, remote) - if remote_exists and not local_exists: - return STATUS_DELETED - - if not remote_exists and local_exists: - return STATUS_NEW - - if remote_exists and local_exists: - if filecmp.cmp(local, remote): - return STATUS_OK - else: - return STATUS_MODIFIED - - return STATUS_UNKNOWN - + return (local_exists, remote_exists, c) class DataCloudHTTP(DataCloudBase): def push(self, item): @@ -223,76 +231,86 @@ class DataCloudHTTP(DataCloudBase): def status(self, item): raise Exception('Not implemented yet') + def _downloaded_size(self, fname): + if os.path.exists(fname) and self.parsed_args.cont: + downloaded = os.path.getsize(fname) + header = {'Range': 'bytes=%d-' % downloaded} + + Logger.debug('found existing {} file, resuming download'.format(tmp_file)) + + return (downloaded, header) + + return (0, None) + + def _get_header(self, r, name): + val = r.headers.get(name) + if val == None: + Logger.debug('\'{}\' not supported by the server'.format(name)) + + return val + + def _verify_downloaded_size(self, r, downloaded_size): + content_range = self._get_header(r, 'content-range') + if downloaded_size and content_range == None: + Logger.debug('Can\'t resume download') + return 0 + + return downloaded_size + + def _download(self, r, fname, downloaded): + mode = 'ab' if downloaded else 'wb' + name = os.path.basename(r.url) + total_length = self._get_header(r, 'content-length') + chunk_size = 1024 * 100 + + progress.update_target(name, downloaded, total_length) + + with open(fname, mode) as f: + for chunk in r.iter_content(chunk_size=chunk_size): + if not chunk: # filter out keep-alive new chunks + continue + + f.write(chunk) + downloaded += len(chunk) + progress.update_target(name, downloaded, total_length) + + + progress.finish_target(name) + + def _verify_md5(self, r, fname): + md5 = file_md5(fname)[0] + content_md5 = self._get_header(r, '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, item): """ Download single file from url. """ to_file = item.cache.relative - tmp_file = self.tmp_file(to_file) - name = os.path.basename(url) - chunk_size = 1024 * 100 - downloaded = 0 - last_reported = 0 - report_bucket = 100*1024*10 - resume_header = None - mode = 'wb' + downloaded, header = self._downloaded_size(tmp_file) + r = requests.get(url, stream=True, headers=header) + downloaded = self._verify_downloaded_size(r, downloaded) - # Resume download if we can - if os.path.exists(tmp_file) and self.parsed_args.cont: - mode = 'ab' + try: + self._download(r, tmp_file, downloaded) + except Exception as exc: + Logger.error('Failed to download "{}": {}'.format(url, exc)) + return None - downloaded = os.path.getsize(tmp_file) - resume_header = {'Range': 'bytes=%d-' % downloaded} - - Logger.debug('found existing {} file, resuming download'.format(tmp_file)) - - r = requests.get(url, stream=True, headers=resume_header) - - content_range = r.headers.get('content-range') - if resume_header and content_range == None: - mode = 'wb' - downloaded = 0 - - Logger.debug('\'range\' is not supported by the server. Can\'t resume download') - - total_length = r.headers.get('content-length') - if total_length == None: - Logger.debug('\'content-length\' is not supported by the server') - - with open(tmp_file, mode) as f: - for chunk in r.iter_content(chunk_size=chunk_size): - if not chunk: # filter out keep-alive new chunks - continue - - downloaded += len(chunk) - - last_reported += chunk_size - if last_reported >= report_bucket: - last_reported = 0 - Logger.debug('Downloaded {}'.format(sizeof_fmt(downloaded))) - - # update progress bar - progress.update_target(name, downloaded, total_length) - - f.write(chunk) - - # tell progress bar that this target is finished downloading - progress.finish_target(name) - - # Verify checksum if we can - content_md5 = r.headers.get('content-md5') - if content_md5 != None: - md5 = file_md5(tmp_file)[0] - if md5 != content_md5: - Logger.error('Checksum mismatch') - return None - - Logger.debug('Checksum matches') - else: - Logger.debug('\'content-md5\' is not supported by the server. Can\'t verify download') + if not self._verify_md5(r, tmp_file): + return None os.rename(tmp_file, to_file) @@ -437,22 +455,6 @@ class DataCloudAWS(DataCloudBase): return data_item - def pull(self, data_item): - """ pull, aws version """ - - fname = data_item.resolved_cache.dvc - key_name = self.cache_file_key(fname) - - return self._import(self.storage_bucket, key_name, fname, data_item) - - def import_data(self, url, item): - """ import, aws version """ - - o = urlparse(url) - assert o.scheme == 's3' - - return self._import(o.netloc, o.path, item.cache.relative, item) - def push(self, data_item): """ push, aws version """ @@ -481,29 +483,18 @@ class DataCloudAWS(DataCloudBase): return data_item - def status(self, data_item): - """ status, aws version """ - + def _status(self, data_item): aws_key = self.cache_file_key(data_item.resolved_cache.dvc) 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(data_item.resolved_cache.relative) - - if remote_exists and not local_exists: - return STATUS_DELETED - - if not remote_exists and local_exists: - return STATUS_NEW - + c = None if remote_exists and local_exists: - if self._cmp_checksum(key, data_item.resolved_cache.dvc): - return STATUS_OK - else: - return STATUS_MODIFIED + c = self._cmp_checksum(key, data_item.resolved_cache.dvc) - return STATUS_UNKNOWN + return (local_exists, remote_exists, c) def remove(self, data_item): aws_file_name = self.cache_file_key(data_item.cache.dvc) @@ -584,22 +575,6 @@ class DataCloudGCP(DataCloudBase): return data_item - def pull(self, item): - """ pull, gcp version """ - - fname = item.resolved_cache.dvc - key_name = self.cache_file_key(fname) - - return self._import(self.storage_bucket, key_name, fname, data_item) - - def import_data(self, url, item): - """ import, gcp version """ - - o = urlparse(url) - assert o.scheme == 'gs' - - return self._import(o.netloc, o.path, item.cache.relative, item) - def push(self, data_item): """ push, gcp version """ @@ -625,7 +600,7 @@ class DataCloudGCP(DataCloudBase): return data_item - def status(self, data_item): + def _status(self, data_item): """ status, gcp version """ bucket = self._get_bucket_gc(self.storage_bucket) @@ -634,20 +609,11 @@ class DataCloudGCP(DataCloudBase): remote_exists = blob is not None and blob.exists() local_exists = os.path.exists(data_item.resolved_cache.relative) - - if remote_exists and not local_exists: - return STATUS_DELETED - - if not remote_exists and local_exists: - return STATUS_NEW - + c = None if remote_exists and local_exists: - if self._cmp_checksum(blob, data_item.resolved_cache.dvc): - return STATUS_OK - else: - return STATUS_MODIFIED + c = self._cmp_checksum(blob, data_item.resolved_cache.dvc) - return STATUS_UNKNOWN + return (local_exists, remote_exists, c) def remove(self, item): raise Exception('NOT IMPLEMENTED YET') From e7bf8054da5d4ef399c26fbe0c2ed3afabe76823 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 19 Aug 2017 19:27:24 +0300 Subject: [PATCH 25/33] functest: fix existing scripts A few minor adjustments to make what we currently have work, while I prepare new func tests. Fixes #175 Signed-off-by: Ruslan Kuprieiev --- functest/clean.sh | 4 ++-- functest/{common.rc => common.sh} | 5 ++++- functest/init.sh | 3 ++- functest/test-repro-code.sh | 4 +++- functest/tutorial_end_to_end.sh | 3 +++ 5 files changed, 14 insertions(+), 5 deletions(-) rename functest/{common.rc => common.sh} (91%) diff --git a/functest/clean.sh b/functest/clean.sh index f75d4f11e..8e84c920e 100755 --- a/functest/clean.sh +++ b/functest/clean.sh @@ -1,5 +1,5 @@ +#!/bin/bash -source common.rc +source common.sh rm -rf $RAW_DATA_LOCAL $CODE_LOCAL $TEST_REPO - diff --git a/functest/common.rc b/functest/common.sh similarity index 91% rename from functest/common.rc rename to functest/common.sh index dda00a5f8..dfab8f296 100644 --- a/functest/common.rc +++ b/functest/common.sh @@ -1,3 +1,6 @@ +#!/bin/bash + +set -e BASE_DIR_S3=https://s3-us-west-2.amazonaws.com/dataversioncontrol/functests RAW_DATA_S3=$BASE_DIR_S3/stackoverflow_raw_small @@ -17,7 +20,7 @@ function _init_repo() { dvc_info 'import xml files' mkdir data/xml - dvc data-import ../$RAW_DATA_LOCAL/* data/xml + dvc import ../$RAW_DATA_LOCAL/* data/xml dvc_info 'copy code' mkdir code diff --git a/functest/init.sh b/functest/init.sh index 7ef46b650..0e88babad 100755 --- a/functest/init.sh +++ b/functest/init.sh @@ -1,5 +1,6 @@ +#!/bin/bash -source common.rc +source common.sh ./clean.sh diff --git a/functest/test-repro-code.sh b/functest/test-repro-code.sh index 479131f0a..18653d582 100755 --- a/functest/test-repro-code.sh +++ b/functest/test-repro-code.sh @@ -1,6 +1,8 @@ #!/bin/bash -source common.rc +set -e + +source common.sh function run_test() { mkdir data/tsv diff --git a/functest/tutorial_end_to_end.sh b/functest/tutorial_end_to_end.sh index 26ffdb58c..e209023d6 100755 --- a/functest/tutorial_end_to_end.sh +++ b/functest/tutorial_end_to_end.sh @@ -1,3 +1,6 @@ +#!/bin/bash + +set -e # 1. First ML model From 194e1ec419b12c94224c3df6204686cf84b47be1 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 20 Aug 2017 06:17:40 +0300 Subject: [PATCH 26/33] dvc: add 'config' command Fixes #177 Signed-off-by: Ruslan Kuprieiev --- dvc/cli.py | 14 ++++++++++++ dvc/command/config.py | 51 +++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 1 + setup.py | 1 + 4 files changed, 67 insertions(+) create mode 100644 dvc/command/config.py diff --git a/dvc/cli.py b/dvc/cli.py index 5711dd11d..1072d7272 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -15,6 +15,7 @@ from dvc.command.lock import CmdLock from dvc.command.gc import CmdGC from dvc.command.import_file import CmdImportFile from dvc.command.target import CmdTarget +from dvc.command.config import CmdConfig from dvc.command.test import CmdTest from dvc.config import Config from dvc import VERSION @@ -298,6 +299,19 @@ def parse_args(argv=None): help='Reset target.') target_parser.set_defaults(func=CmdTarget) + # Config + config_parser = subparsers.add_parser( + 'config', + parents=[parent_parser], + help='Get or set repository options') + config_parser.add_argument('name', + help='Option name') + config_parser.add_argument('value', + nargs='?', + default=None, + help='Option value') + config_parser.set_defaults(func=CmdConfig) + if isinstance(argv, str): argv = argv.split() diff --git a/dvc/command/config.py b/dvc/command/config.py new file mode 100644 index 000000000..340c40fbf --- /dev/null +++ b/dvc/command/config.py @@ -0,0 +1,51 @@ +import os +import configobj + +from dvc.command.base import CmdBase +from dvc.logger import Logger +from dvc.config import Config + +class CmdConfig(CmdBase): + def __init__(self, settings): + super(CmdConfig, self).__init__(settings) + + @property + def _config_path(self): + return os.path.join(self.git.git_dir, Config.CONFIG) + + def _find_key(self, l, name): + for k in l: + if k.lower() == name.lower(): + return k + + return None + + def run(self): + _section, _opt = self.parsed_args.name.strip().split('.') + + # Using configobj because it doesn't + # drop comments like configparser does. + config = configobj.ConfigObj(self._config_path) + + section = self._find_key(config.keys(), _section) + if not section: + return 1 + + opt = self._find_key(config[section].keys(), _opt) + if not opt: + return 1 + + if self.parsed_args.value == None: + Logger.info(config[section][opt]) + return 0 + + try: + config[section][opt] = self.parsed_args.value + config.write() + except Exception as exc: + Logger.error('Failed to set \'{}\' to \'{}\': {}'.format(self.parsed_args.name, + self.parsed_args.value, + exc)) + return 1 + + return 0 diff --git a/requirements.txt b/requirements.txt index 5ee7b96f0..8e3d4ad16 100644 --- a/requirements.txt +++ b/requirements.txt @@ -86,3 +86,4 @@ traitlets==4.3.2 wcwidth==0.1.7 colorama==0.3.9 google-compute-engine==2.4.1 +configobj==5.0.6 diff --git a/setup.py b/setup.py index faef14732..d47341ba2 100644 --- a/setup.py +++ b/setup.py @@ -92,6 +92,7 @@ install_requires = [ 'wcwidth==0.1.7', 'colorama==0.3.9', 'google-compute-engine==2.4.1', + 'configobj==5.0.6', ] if platform.system() == 'Darwin': From 0f45afb3a43805006831f0df5be061bb8537584a Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 20 Aug 2017 08:56:30 +0300 Subject: [PATCH 27/33] functest: rework test system and introduce new tests Reworked existing tests and test-system, added a few new tests(init, reworked repro, import). Fixes #175 Signed-off-by: Ruslan Kuprieiev --- functest/clean.sh | 2 +- functest/common.sh | 132 +++++++++++++++++++++++++++++------- functest/init.sh | 17 ----- functest/run.sh | 17 +++++ functest/test-import.sh | 26 +++++++ functest/test-init.sh | 17 +++++ functest/test-repro-code.sh | 28 ++++---- 7 files changed, 182 insertions(+), 57 deletions(-) delete mode 100755 functest/init.sh create mode 100755 functest/run.sh create mode 100755 functest/test-import.sh create mode 100755 functest/test-init.sh diff --git a/functest/clean.sh b/functest/clean.sh index 8e84c920e..8d39fe6d4 100755 --- a/functest/clean.sh +++ b/functest/clean.sh @@ -2,4 +2,4 @@ source common.sh -rm -rf $RAW_DATA_LOCAL $CODE_LOCAL $TEST_REPO +rm -rf $CACHE_DIR $TEST_REPO diff --git a/functest/common.sh b/functest/common.sh index dfab8f296..094296428 100644 --- a/functest/common.sh +++ b/functest/common.sh @@ -2,37 +2,119 @@ set -e -BASE_DIR_S3=https://s3-us-west-2.amazonaws.com/dataversioncontrol/functests -RAW_DATA_S3=$BASE_DIR_S3/stackoverflow_raw_small +SDIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +CACHE_DIR=$SDIR/cache +TEST_REPO=$SDIR/myrepo +CODE_CACHE=$CACHE_DIR/code +DATA_CACHE=$CACHE_DIR/data -RAW_DATA_LOCAL=sync_raw_data -CODE_LOCAL=sync_code -TEST_REPO=repo_test +DATA="Badges.xml Tags.xml PostLinks.xml" +CODE="xmltotsv.py" + +REGION='us-west-2' +BASE_DIR_S3=https://s3-$REGION.amazonaws.com/dataversioncontrol/functests +DATA_S3=$BASE_DIR_S3/stackoverflow_raw_small +CODE_S3=$BASE_DIR_S3/code + +function dvc_print() { + echo -e "$1" +} function dvc_info() { - echo "=============== DVC test: $1 ===============" + dvc_print "\e[34m$1\e[0m" } -function _init_repo() { - dvc_info 'init reporitory' - git init - dvc init - - dvc_info 'import xml files' - mkdir data/xml - dvc import ../$RAW_DATA_LOCAL/* data/xml - - dvc_info 'copy code' - mkdir code - cp ../$CODE_LOCAL/* code/ - git add code/ - git commit -m 'Add code' +function dvc_header() { + dvc_info "=============== DVC test: $1 ===============" } -function create_repo() { - dvc_info 'create reporitory' - rm -rf $TEST_REPO - mkdir $TEST_REPO - (cd $TEST_REPO && _init_repo) +function dvc_error() { + dvc_print "\e[31m$1\e[0m" } +function dvc_fail() { + dvc_error "FAIL" + exit 1 +} + +function dvc_fatal() { + dvc_error "$1" + dvc_fail +} + +function dvc_pass() { + dvc_print "\e[32mPASS\e[0m" + exit 0 +} + +trap 'dvc_fail;' ERR + +function dvc_prepare_cache() { + if [ -d $CACHE_DIR ]; then + return + fi + + mkdir -p $DATA_CACHE + pushd $DATA_CACHE + + for f in $DATA; do + wget $DATA_S3/$f + done + + popd + + mkdir -p $CODE_CACHE + pushd $CODE_CACHE + + for f in $CODE; do + wget $CODE_S3/$f + done + + popd +} + +function dvc_create_git_repo() { + dvc_info "Creating git repo" + + dvc_prepare_cache + + mkdir $TEST_REPO + cd $TEST_REPO + + git init + + git config user.name "DVC tester" + git config user.email "dvctester@dataversioncontrol.com" + + mkdir code + cp $CODE_CACHE/* code/ + git add code/ + git commit -m 'Add code' +} + +function dvc_create_repo() { + dvc_info "Creating dvc repo" + + dvc_create_git_repo + + dvc init + + mkdir data/xml + dvc import $DATA_CACHE/* data/xml +} + +function dvc_check_dirs() { + for d in $1; do + if [ ! -d "$d" ]; then + dvc_fatal "Directory $d doesn't exist" + fi + done +} + +function dvc_check_files() { + for f in $1; do + if [ ! -f "$f" ]; then + dvc_fatal "File $f doesn't exist" + fi + done +} diff --git a/functest/init.sh b/functest/init.sh deleted file mode 100755 index 0e88babad..000000000 --- a/functest/init.sh +++ /dev/null @@ -1,17 +0,0 @@ -#!/bin/bash - -source common.sh - -./clean.sh - -rm -rf $RAW_DATA_LOCAL -mkdir $RAW_DATA_LOCAL -for file in "Badges.xml" "Tags.xml" "PostLinks.xml" -do - (cd $RAW_DATA_LOCAL; wget $RAW_DATA_S3/$file) -done - -rm -rf $CODE_LOCAL -mkdir $CODE_LOCAL -(cd $CODE_LOCAL; wget "$BASE_DIR_S3/code/xmltotsv.py") - diff --git a/functest/run.sh b/functest/run.sh new file mode 100755 index 000000000..e6d3ef6db --- /dev/null +++ b/functest/run.sh @@ -0,0 +1,17 @@ +#!/bin/bash + +set -e + +source common.sh + +TESTS+=" test-init.sh" +TESTS+=" test-import.sh" +TESTS+=" test-repro-code.sh" + +for t in $TESTS; do + rm -rf $TEST_REPO + dvc_header "$t" + ./$t +done + +rm -rf $TEST_REPO diff --git a/functest/test-import.sh b/functest/test-import.sh new file mode 100755 index 000000000..d57e1294a --- /dev/null +++ b/functest/test-import.sh @@ -0,0 +1,26 @@ +#!/bin/bash + +set -e + +source common.sh + +dvc_create_repo + +dvc_info "Import from local file" +dvc import $DATA_CACHE/Tags.xml data/local +dvc_check_files "data/local" + +dvc_info "Import from http url" +dvc import $DATA_S3/Tags.xml data/http +dvc_check_files "data/http" + +# Temporarily disabled for travis builds, as +# they don't have aws credentials setup yet. +if [ -z $TRAVIS ]; then + dvc_info "Import from s3 url" + dvc config aws.region $REGION + dvc import s3://dataversioncontrol/functests/stackoverflow_raw_small/Tags.xml data/s3 + dvc_check_files "data/s3" +fi + +dvc_pass diff --git a/functest/test-init.sh b/functest/test-init.sh new file mode 100755 index 000000000..e291597db --- /dev/null +++ b/functest/test-init.sh @@ -0,0 +1,17 @@ +#!/bin/bash + +set -e + +source common.sh + +dvc_create_git_repo + +dvc init + +DIRS="data .cache .state" +FILES="dvc.conf" + +dvc_check_dirs $DIRS +dvc_check_files $FILES + +dvc_pass diff --git a/functest/test-repro-code.sh b/functest/test-repro-code.sh index 18653d582..58d071f3d 100755 --- a/functest/test-repro-code.sh +++ b/functest/test-repro-code.sh @@ -4,19 +4,19 @@ set -e source common.sh -function run_test() { - mkdir data/tsv +dvc_create_repo - dvc_info 'convert Budget.xml to tsv' - dvc run --code code/xmltotsv.py python code/xmltotsv.py --extract row/@Id,row/@UserId,row/@Name,row/@Class,row/@TagBased,row/@Date data/xml/Badges.xml data/tsv/Badges.tsv - - dvc_info 'modify code' - echo " " >> code/xmltotsv.py - git commit -am 'Change code' - - dvc_info 'reproduce tsv convertion code' - dvc repro data/tsv/Badges.tsv -} +mkdir data/tsv -create_repo -(cd $TEST_REPO; run_test) +dvc_info 'Convert Budget.xml to tsv' +pip2 install xmltodict +dvc run --code code/xmltotsv.py python2 code/xmltotsv.py --extract row/@Id,row/@UserId,row/@Name,row/@Class,row/@TagBased,row/@Date data/xml/Badges.xml data/tsv/Badges.tsv + +dvc_info 'Modify code' +echo " " >> code/xmltotsv.py +git commit -am 'Change code' + +dvc_info 'Reproduce tsv convertion code' +dvc repro data/tsv/Badges.tsv + +dvc_pass From dc77e03459db7ccba9999ad1a3fcea9594d3c918 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 20 Aug 2017 09:02:45 +0300 Subject: [PATCH 28/33] travis: run functests Fixes #175 Signed-off-by: Ruslan Kuprieiev --- .travis.yml | 1 + functests.sh | 11 +++++++++++ test-requirements.txt | 1 + 3 files changed, 13 insertions(+) create mode 100755 functests.sh diff --git a/.travis.yml b/.travis.yml index d20cf3f2e..10a0fbcc7 100644 --- a/.travis.yml +++ b/.travis.yml @@ -39,6 +39,7 @@ install: - mkdir -p ~/.aws && printf "[default]\naws_access_key_id = ABCDEFG\naws_secret_access_key = 123456789" > ~/.aws/credentials script: - ./unittests.sh + - ./functests.sh - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then ./build_macos.sh; else ./build_linux.sh; fi - ./build_package.sh after_sucess: diff --git a/functests.sh b/functests.sh new file mode 100755 index 000000000..4c55eee8c --- /dev/null +++ b/functests.sh @@ -0,0 +1,11 @@ +#!/bin/bash + +DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +BIN=$DIR/bin +DVC_HOME=$DIR + +export PATH=$BIN:$PATH +export DVC_HOME=$DIR + +cd functest +./run.sh diff --git a/test-requirements.txt b/test-requirements.txt index 4ebc8aea5..21329b8d1 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1 +1,2 @@ coverage +xmltodict From 78cad172c616215a069e1358186432922eef12a1 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 20 Aug 2017 14:24:31 +0300 Subject: [PATCH 29/33] aws: use multipart to push changes Fixes #100. --- dvc/data_cloud.py | 26 ++++++++++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index c79c0e186..0d5695843 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -7,6 +7,7 @@ import tempfile import requests import configparser import filecmp +import math from boto.s3.connection import S3Connection from google.cloud import storage as gc @@ -455,6 +456,28 @@ class DataCloudAWS(DataCloudBase): return data_item + def _push_multipart(self, b, aws_key, fname): + mp = b.initiate_multipart_upload(aws_key) + + source_size = os.stat(fname).st_size + chunk_size = 5000*1024*1024 + chunk_count = int(math.ceil(source_size / float(chunk_size))) + + with open(fname, 'r') as fp: + for i in range(chunk_count): + offset = i * chunk_size + left = source_size - offset + size = min([chunk_size, left]) + part_num = i + 1 + + fp.seek(offset) + mp.upload_part_from_file(fp=fp, part_num=part_num, size=size, cb=create_cb(fname)) + + if len(mp.get_all_parts()) != chunk_count: + raise Exception("Couldn't upload all file parts") + + mp.complete_upload() + def push(self, data_item): """ push, aws version """ @@ -473,8 +496,7 @@ class DataCloudAWS(DataCloudBase): key = bucket.new_key(aws_key) try: - key.set_contents_from_filename(data_item.resolved_cache.relative, - cb=create_cb(data_item.resolved_cache.relative)) + self._push_multipart(bucket, aws_key, data_item.resolved_cache.relative) except Exception as exc: Logger.error('Failed to upload "{}": {}'.format(data_item.resolved_cache.relative, exc)) return None From 163558c9c2ace433887eb3aaf04ae42b58b7784c Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 20 Aug 2017 17:55:23 +0300 Subject: [PATCH 30/33] Revert "aws: use multipart to push changes" --- dvc/data_cloud.py | 26 ++------------------------ 1 file changed, 2 insertions(+), 24 deletions(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 0d5695843..c79c0e186 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -7,7 +7,6 @@ import tempfile import requests import configparser import filecmp -import math from boto.s3.connection import S3Connection from google.cloud import storage as gc @@ -456,28 +455,6 @@ class DataCloudAWS(DataCloudBase): return data_item - def _push_multipart(self, b, aws_key, fname): - mp = b.initiate_multipart_upload(aws_key) - - source_size = os.stat(fname).st_size - chunk_size = 5000*1024*1024 - chunk_count = int(math.ceil(source_size / float(chunk_size))) - - with open(fname, 'r') as fp: - for i in range(chunk_count): - offset = i * chunk_size - left = source_size - offset - size = min([chunk_size, left]) - part_num = i + 1 - - fp.seek(offset) - mp.upload_part_from_file(fp=fp, part_num=part_num, size=size, cb=create_cb(fname)) - - if len(mp.get_all_parts()) != chunk_count: - raise Exception("Couldn't upload all file parts") - - mp.complete_upload() - def push(self, data_item): """ push, aws version """ @@ -496,7 +473,8 @@ class DataCloudAWS(DataCloudBase): key = bucket.new_key(aws_key) try: - self._push_multipart(bucket, aws_key, data_item.resolved_cache.relative) + key.set_contents_from_filename(data_item.resolved_cache.relative, + cb=create_cb(data_item.resolved_cache.relative)) except Exception as exc: Logger.error('Failed to upload "{}": {}'.format(data_item.resolved_cache.relative, exc)) return None From 90762da18f85104d92567c3e87542473e4a3fff5 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 20 Aug 2017 17:27:02 +0300 Subject: [PATCH 31/33] cloud: fix typo Signed-off-by: Ruslan Kuprieiev --- dvc/data_cloud.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index c79c0e186..a194e5871 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -138,7 +138,7 @@ class DataCloudBase(object): fname = item.resolved_cache.dvc key_name = self.cache_file_key(fname) - return self._import(self.storage_bucket, key_name, fname, data_item) + return self._import(self.storage_bucket, key_name, fname, item) def import_data(self, url, item): o = urlparse(url) From 04e66f94fca69a73ce18893fbc1760a6f4c2c68a Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 20 Aug 2017 19:04:34 +0300 Subject: [PATCH 32/33] aws: use multipart to push changes, v2 Fixes #100 Signed-off-by: Ruslan Kuprieiev --- dvc/data_cloud.py | 35 ++++++++++++++++++++++++++++++++--- 1 file changed, 32 insertions(+), 3 deletions(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index a194e5871..ea936e234 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -7,6 +7,7 @@ import tempfile import requests import configparser import filecmp +import math from boto.s3.connection import S3Connection from google.cloud import storage as gc @@ -418,9 +419,12 @@ class DataCloudAWS(DataCloudBase): return bucket def _cmp_checksum(self, key, fname): - md5_cloud = key.etag[1:-1] + 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 @@ -455,6 +459,32 @@ class DataCloudAWS(DataCloudBase): return data_item + def _push_multipart(self, key, fname): + # 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])} + + mp = key.bucket.initiate_multipart_upload(key.name, metadata=metadata) + + source_size = os.stat(fname).st_size + chunk_size = 5000*1024*1024 + chunk_count = int(math.ceil(source_size / float(chunk_size))) + + with open(fname, 'rb') as fp: + for i in range(chunk_count): + offset = i * chunk_size + left = source_size - offset + size = min([chunk_size, left]) + part_num = i + 1 + + fp.seek(offset) + mp.upload_part_from_file(fp=fp, part_num=part_num, size=size, cb=create_cb(fname)) + + if len(mp.get_all_parts()) != chunk_count: + raise Exception("Couldn't upload all file parts") + + mp.complete_upload() + def push(self, data_item): """ push, aws version """ @@ -473,8 +503,7 @@ class DataCloudAWS(DataCloudBase): key = bucket.new_key(aws_key) try: - key.set_contents_from_filename(data_item.resolved_cache.relative, - cb=create_cb(data_item.resolved_cache.relative)) + self._push_multipart(key, data_item.resolved_cache.relative) except Exception as exc: Logger.error('Failed to upload "{}": {}'.format(data_item.resolved_cache.relative, exc)) return None From 6430982ef0c503f05a4f3a0190c281e2ab92ac12 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 21 Aug 2017 08:30:06 +0300 Subject: [PATCH 33/33] aws: fix progress and chunk size for multipart upload Signed-off-by: Ruslan Kuprieiev --- dvc/data_cloud.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index ea936e234..3e61eeaa7 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -56,7 +56,10 @@ def sizeof_fmt(num, suffix='B'): return "%.1f%s%s" % (num, 'Y', suffix) -def percent_cb(name, complete, total): +def percent_cb(name, part_complete, part_total, offset=0, multipart_total=None): + complete = offset + part_complete + total = multipart_total if multipart_total != None else part_total + Logger.debug('{}: {} transferred out of {}'.format( name, sizeof_fmt(complete), @@ -64,8 +67,8 @@ def percent_cb(name, complete, total): progress.update_target(os.path.basename(name), complete, total) -def create_cb(name): - return (lambda cur,tot: percent_cb(name, cur, tot)) +def create_cb(name, offset=0, multipart_total=None): + return (lambda cur,tot: percent_cb(name, cur, tot, offset, multipart_total)) def file_md5(fname): @@ -467,7 +470,7 @@ class DataCloudAWS(DataCloudBase): mp = key.bucket.initiate_multipart_upload(key.name, metadata=metadata) source_size = os.stat(fname).st_size - chunk_size = 5000*1024*1024 + chunk_size = 50*1024*1024 chunk_count = int(math.ceil(source_size / float(chunk_size))) with open(fname, 'rb') as fp: @@ -478,7 +481,11 @@ class DataCloudAWS(DataCloudBase): part_num = i + 1 fp.seek(offset) - mp.upload_part_from_file(fp=fp, part_num=part_num, size=size, cb=create_cb(fname)) + mp.upload_part_from_file(fp=fp, + size=size, + num_cb=100, + part_num=part_num, + cb=create_cb(fname, offset, source_size)) if len(mp.get_all_parts()) != chunk_count: raise Exception("Couldn't upload all file parts")