From 51dd60abc80c2f2c6e976cbd7c7ba09fd040c3a9 Mon Sep 17 00:00:00 2001 From: Dmitry Petrov Date: Sun, 5 Mar 2017 23:08:48 -0800 Subject: [PATCH] Sync command --- bin/nlx-data-sync | 3 + neatlynx/{base_cmd.py => cmd_base.py} | 17 +++- neatlynx/cmd_data_import.py | 65 ++++++------- neatlynx/cmd_data_sync.py | 129 ++++++++++++++++++++++++++ neatlynx/cmd_init.py | 17 ++-- neatlynx/config.py | 47 ++++++++-- neatlynx/data_file_obj.py | 27 ++---- neatlynx/git_wrapper.py | 4 +- 8 files changed, 230 insertions(+), 79 deletions(-) rename neatlynx/{base_cmd.py => cmd_base.py} (86%) create mode 100644 neatlynx/cmd_data_sync.py diff --git a/bin/nlx-data-sync b/bin/nlx-data-sync index e69de29bb..9c60357fa 100755 --- a/bin/nlx-data-sync +++ b/bin/nlx-data-sync @@ -0,0 +1,3 @@ +#!/bin/bash + +PYTHONPATH=$NEATLYNX_HOME python $NEATLYNX_HOME/neatlynx/cmd_data_sync.py $@ diff --git a/neatlynx/base_cmd.py b/neatlynx/cmd_base.py similarity index 86% rename from neatlynx/base_cmd.py rename to neatlynx/cmd_base.py index c1cce3bf4..f56ad65c9 100644 --- a/neatlynx/base_cmd.py +++ b/neatlynx/cmd_base.py @@ -18,13 +18,20 @@ class Logger(object): def error(msg): print('{}'.format(msg)) + @staticmethod + def verbose(msg): + print('{}'.format(msg)) + + +class CmdBase(object): + CONFIG = 'neatlynx.conf' -class BaseCmd(object): def __init__(self, parse_config=True): self._git = GitWrapper() self._args = None self._lnx_home = None + self._config = None if parse_config: self._config = Config(os.path.join(self.git.git_dir, self.CONFIG)) @@ -40,6 +47,10 @@ class BaseCmd(object): raise ConfigError("NEATLYNX_HOME directory doesn't exists") pass + @property + def config(self): + return self._config + @property def lnx_home(self): return self._lnx_home @@ -58,7 +69,7 @@ class BaseCmd(object): def add_string_arg(self, parser, name, message, default = None, conf_section=None, conf_name=None): if conf_section and conf_name: - section = self._config[conf_section] + section = self.config[conf_section] if not section: raise ConfigError("") default_value = section.get(conf_section, default) @@ -71,4 +82,4 @@ class BaseCmd(object): help=message) def run(self): - pass + pass \ No newline at end of file diff --git a/neatlynx/cmd_data_import.py b/neatlynx/cmd_data_import.py index 3c7727720..655711b2b 100644 --- a/neatlynx/cmd_data_import.py +++ b/neatlynx/cmd_data_import.py @@ -3,10 +3,9 @@ import os from pathlib import Path from shutil import copyfile -from neatlynx.base_cmd import BaseCmd, Logger +from neatlynx.cmd_base import CmdBase, Logger from neatlynx.data_file_obj import DataFileObj from neatlynx.exceptions import NeatLynxException -from neatlynx.git_wrapper import GitWrapper class DataImportError(NeatLynxException): @@ -14,9 +13,9 @@ class DataImportError(NeatLynxException): NeatLynxException.__init__(self, 'Import error: {}'.format(msg)) -class DataImport(BaseCmd): +class CmdDataImport(CmdBase): def __init__(self): - BaseCmd.__init__(self) + CmdBase.__init__(self) pass def define_args(self, parser): @@ -31,51 +30,39 @@ class DataImport(BaseCmd): if not input.is_file(): raise DataImportError('Input file "{}" has to be a regular file'.format(input)) - dataFileObj = DataFileObj(self.args.output, self._config, GitWrapper.curr_commit()) + output = self.args.output + if os.path.isdir(self.args.output): + output = os.path.join(output, os.path.basename(self.args.input)) - #output = Path(self.args.output) - if dataFileObj.data_file_relative.exists(): - raise DataImportError('Output file "{}" already exists'.format(dataFileObj.data_file_relative)) - if os.path.isdir(dataFileObj.data_dir_relative): - raise DataImportError('Output file directory "{}" does not exists'.format(dataFileObj.data_dir_relative)) + dobj = DataFileObj(output, self.git, self.config) - data_dir_path = Path(self.config.DataDir) - if output.parent < data_dir_path: - raise DataImportError('Output file has to be in data dir - {}'.format(data_dir_path)) + if os.path.exists(dobj.data_file_relative): + raise DataImportError('Output file "{}" already exists'.format(dobj.data_file_relative)) + if not os.path.isdir(os.path.dirname(dobj.data_file_abs)): + raise DataImportError('Output file directory "{}" does not exists'.format( + os.path.dirname(dobj.data_file_relative))) - # data_dir_path_str = str(data_dir_path) - # output_dir_str = str(output.parent) - # relative_dir = output_dir_str[len(data_dir_path_str):].strip(os.path.sep) - # - # cache_file_dir = os.path.join(self.config.CachDir, relative_dir) - # cache_file_dir_path = Path(cache_file_dir) - # - # state_file_dir = os.path.join(self.config.StateDir, relative_dir) - # state_file_dir_path = Path(state_file_dir) - # - # commit = GitWrapper.curr_commit() - # cache_file_name = output.name + '_' + commit - # cache_file = cache_file_dir_path / cache_file_name - # state_file = state_file_dir_path / output.name + os.makedirs(os.path.dirname(dobj.cache_file_relative), exist_ok=True) + copyfile(self.args.input, dobj.cache_file_relative) + Logger.verbose('Input file "{}" was copied to cache "{}"'.format( + self.args.input, dobj.cache_file_relative)) - # Perform actions - cache_file.parent.mkdir(parents=True, exist_ok=True) - state_file.parent.mkdir(parents=True, exist_ok=True) + cache_relative_to_data = os.path.relpath(dobj.cache_file_relative, os.path.dirname(dobj.data_file_relative)) + os.symlink(cache_relative_to_data, dobj.data_file_relative) + Logger.verbose('Symlink from data file "{}" to the cache file "{}" was created'. + format(dobj.data_file_relative, cache_relative_to_data)) - copyfile(self.args.input, str(cache_file)) - Logger.verbose('Input file "{}" was copied to cache "{}"'.format(self.args.input, cache_file)) - - output.symlink_to(cache_file) - Logger.verbose('Symlink from data file "{}" the cache file "{}" was created'. - format(output, cache_file)) - - StateFile.create(state_file, input, output.absolute(), cache_file.absolute()) + os.makedirs(os.path.dirname(dobj.state_file_relative), exist_ok=True) + with open(dobj.state_file_relative, 'w') as fd: + fd.write('NLX_state. v0.1\n') + fd.write('Args: {}\n'.format(sys.argv)) + Logger.verbose('State file "{}" was created'.format(dobj.state_file_relative)) pass if __name__ == '__main__': try: - sys.exit(DataImport().run()) + sys.exit(CmdDataImport().run()) except NeatLynxException as e: Logger.error(e) sys.exit(1) diff --git a/neatlynx/cmd_data_sync.py b/neatlynx/cmd_data_sync.py new file mode 100644 index 000000000..43a879f22 --- /dev/null +++ b/neatlynx/cmd_data_sync.py @@ -0,0 +1,129 @@ +import os +import hashlib +from boto.s3.connection import S3Connection + +from neatlynx.cmd_base import CmdBase, Logger +from neatlynx.exceptions import NeatLynxException + + +class DataSyncError(NeatLynxException): + def __init__(self, msg): + NeatLynxException.__init__(self, 'Data sync error: {}'.format(msg)) + + +def sizeof_fmt(num, suffix='B'): + for unit in ['','K','M','G','T','P','E','Z']: + if abs(num) < 1024.0: + return "%3.1f%s%s" % (num, unit, suffix) + num /= 1024.0 + return "%.1f%s%s" % (num, 'Y', suffix) + + +def percent_cb(complete, total): + Logger.verbose('{} transferred out of {}'.format(sizeof_fmt(complete), sizeof_fmt(total))) + + +def file_md5(fname): + hash_md5 = hashlib.md5() + with open(fname, "rb") as f: + for chunk in iter(lambda: f.read(1024*100), b""): + hash_md5.update(chunk) + return hash_md5.hexdigest() + + +class CmdDataSync(CmdBase): + def __init__(self): + CmdBase.__init__(self) + + conn = S3Connection(self.config.aws_access_key_id, self.config.aws_secret_access_key) + + bucket_name = self.config.aws_storage_bucket + self._bucket = conn.lookup(bucket_name) + if not self._bucket: + self._bucket = conn.create_bucket(bucket_name) + Logger.info('S3 bucket "{}" was created'.format(bucket_name)) + pass + + def define_args(self, parser): + self.add_string_arg(parser, 'target', 'Target to sync - file or directory') + pass + + def run(self): + target = self.args.target + if os.path.islink(target): + return self.sync_symlink(target) + + if os.path.isdir(target): + return self.sync_dir(target) + + raise DataSyncError('File "{}" does not exit'.format(target)) + + def sync_dir(self, dir): + for f in os.listdir(dir): + fname = os.path.join(dir, f) + if os.path.isdir(fname): + self.sync_dir(fname) + elif os.path.islink(fname): + self.sync_symlink(fname) + else: + raise DataSyncError('Unsupported file type "{}"'.format(fname)) + pass + + def sync_symlink(self, file): + cache_file_rel_data = os.path.join(os.path.dirname(file), os.readlink(file)) + cache_file = os.path.relpath(os.path.relpath(cache_file_rel_data), os.path.realpath(os.curdir)) + + if os.path.isfile(cache_file): + self.sync_to_cloud(cache_file) + else: + self.sync_from_cloud(cache_file) + pass + pass + + def sync_from_cloud(self, cache_file): + s3_file = self._get_target_s3_name(cache_file) + key = self._bucket.get_key(s3_file) + if not key: + raise DataSyncError('File "{}" is not exist in the cloud'.format(cache_file)) + + Logger.info('Downloading cache file "{}" from S3 {}/{}'.format(cache_file, self._bucket.name, s3_file)) + key.get_contents_to_filename(cache_file, cb=percent_cb) + Logger.info('Downloading completed') + pass + + def _get_target_s3_name(self, cache_file): + cache_file_rel = os.path.relpath(cache_file, self.config.cache_dir) + cache_file_rel = cache_file_rel.replace(os.sep, '/').strip('/') + + target_file = self.config.aws_storage_prefix + '/' + cache_file_rel + return target_file + + def sync_to_cloud(self, cache_file): + target_file = self._get_target_s3_name(cache_file) + + key = self._bucket.get_key(target_file) + if key: + Logger.verbose('File already uploaded to the cloud. Checking checksum...') + + md5_cloud = key.etag[1:-1] + md5_local = file_md5(cache_file) + if md5_cloud == md5_local: + Logger.verbose('File checksum matches. No uploading is needed.') + return + + Logger.info('Checksum miss-match. Re-uploading is required.') + + Logger.info('Uploading cache file "{}" to S3 {}/{}'.format(cache_file, self._bucket.name, target_file)) + key = self._bucket.new_key(target_file) + key.set_contents_from_filename(cache_file, cb=percent_cb) + Logger.info('Uploading completed') + pass + + +if __name__ == '__main__': + import sys + try: + sys.exit(CmdDataSync().run()) + except NeatLynxException as e: + Logger.error(e) + sys.exit(1) diff --git a/neatlynx/cmd_init.py b/neatlynx/cmd_init.py index 07256e3ca..3a2c60bc2 100644 --- a/neatlynx/cmd_init.py +++ b/neatlynx/cmd_init.py @@ -2,7 +2,7 @@ import os import sys from pathlib import Path -from neatlynx.base_cmd import BaseCmd, Logger +from neatlynx.cmd_base import CmdBase, Logger from neatlynx.config import Config from neatlynx.exceptions import NeatLynxException @@ -12,22 +12,23 @@ class InitError(NeatLynxException): NeatLynxException.__init__(self, 'Init error: {}'.format(msg)) -class CmdInit(BaseCmd): - DEFAULT_CONFIG = 'neatlynx.conf.default' - +class CmdInit(CmdBase): CONFIG_TEMPLATE = '''[Global] DataDir = {} -StateDir = {} CacheDir = {} +StateDir = {} Cloud = AWS [AWS] -AccessKeyID = +AccessKeyId = SecretAccessKey = + +StoragePath = neatlynx/tutorial + + Region = us-east-1 Zone = us-east-1a -StorageBucket = neatlynx-tutorial Image = ami-2d39803a InstanceType = t2.nano @@ -43,7 +44,7 @@ KeyName = neatlynx-key SecurityGroup = neatlynx-group''' def __init__(self): - BaseCmd.__init__(self, parse_config=False) + CmdBase.__init__(self, parse_config=False) pass def define_args(self, parser): diff --git a/neatlynx/config.py b/neatlynx/config.py index ca426cd15..ac42f9d2f 100644 --- a/neatlynx/config.py +++ b/neatlynx/config.py @@ -32,15 +32,18 @@ class ConfigI(object): class Config(ConfigI): CONFIG = 'neatlynx.conf' - def __init__(self, git_dir, file=None, data_dir=None, cache_dir=None, state_dir=None): + def __init__(self, conf_file): self._config = configparser.ConfigParser() - self._config.read(os.path.join(git_dir, self.CONFIG)) - self._global = self._config['Global'] + if not os.path.isfile(conf_file): + raise ConfigError('Config file "{}" does not exist'.format(conf_file)) + self._config.read(conf_file) - self._data_dir = self._global['DataDir'] - self._cache_dir = self._global['CacheDir'] - self._state_dir = self._global['StateDir'] + # self._global = self._config['Global'] + # + # self._data_dir = self._global['DataDir'] + # self._cache_dir = self._global['CacheDir'] + # self._state_dir = self._global['StateDir'] pass # def get_data_file_obj(self): @@ -48,12 +51,38 @@ class Config(ConfigI): @property def data_dir(self): - return self._data_dir + return self._config['Global']['DataDir'] @property def cache_dir(self): - return self._cache_dir + return self._config['Global']['CacheDir'] @property def state_dir(self): - return self._state_dir \ No newline at end of file + return self._config['Global']['StateDir'] + + @property + def aws_access_key_id(self): + return self._config['AWS']['AccessKeyId'] + + @property + def aws_secret_access_key(self): + return self._config['AWS']['SecretAccessKey'] + + @property + def aws_storage_path(self): + return self._config['AWS']['StoragePath'] + + def _aws_storage_path_parts(self): + return self.aws_storage_path.strip('/').split('/', 1) + + @property + def aws_storage_bucket(self): + return self._aws_storage_path_parts()[0] + + @property + def aws_storage_prefix(self): + parts = self._aws_storage_path_parts() + if len(parts) > 1: + return parts[1] + return '' diff --git a/neatlynx/data_file_obj.py b/neatlynx/data_file_obj.py index 36bb45e25..3d8829541 100644 --- a/neatlynx/data_file_obj.py +++ b/neatlynx/data_file_obj.py @@ -9,8 +9,10 @@ class DataFilePathError(NeatLynxException): class NotInDataDirError(NeatLynxException): - def __init__(self): - NeatLynxException.__init__(self, 'Data file location error: the file has to be in the data directory') + def __init__(self, data_dir): + NeatLynxException.__init__(self, + 'Data file location error: the file has to be in the data directory "{}"'. + format(data_dir)) class DataFileObj(object): @@ -20,6 +22,7 @@ class DataFileObj(object): def __init__(self, data_file, git, config): self._git = git self._config = config + self._curr_dir_abs = os.path.realpath(os.curdir) data_file = os.path.realpath(data_file) @@ -30,7 +33,7 @@ class DataFileObj(object): self._rel_path = os.path.relpath(data_file_dir, data_dir) if not data_file_dir.startswith(data_dir): - raise NotInDataDirError() + raise NotInDataDirError(config.data_dir) if self._rel_path == '.': self._rel_path = '' @@ -51,11 +54,7 @@ class DataFileObj(object): @property def data_file_relative(self): - return os.path.relpath(self.data_file_abs, self._git.git_dir_abs()) - - @property - def data_dir_relative(self): - return os.path.basename(self.data_file_relative) + return os.path.relpath(self.data_file_abs, self._curr_dir_abs) # Cache file properties @property @@ -76,11 +75,7 @@ class DataFileObj(object): @property def cache_file_relative(self): - return os.path.relpath(self.cache_file_abs, self._git.git_dir_abs) - - @property - def cache_dir_relative(self): - return os.path.basename(self.cache_file_relative) + return os.path.relpath(self.cache_file_abs, self._curr_dir_abs) # State file properties @property @@ -97,8 +92,4 @@ class DataFileObj(object): @property def state_file_relative(self): - return os.path.relpath(self.state_file_abs, self._git.git_dir_abs) - - @property - def state_dir_relative(self): - return os.path.basename(self.state_file_relative) + return os.path.relpath(self.state_file_abs, self._curr_dir_abs) diff --git a/neatlynx/git_wrapper.py b/neatlynx/git_wrapper.py index 277c39064..7ece88531 100644 --- a/neatlynx/git_wrapper.py +++ b/neatlynx/git_wrapper.py @@ -53,7 +53,7 @@ class GitWrapper(GitWrapperI): self._git_dir = out return self._git_dir - except GitCmdError as e: + except GitCmdError: raise except Exception as e: raise GitCmdError('Unable to run git command: {}'.format(e)) @@ -77,7 +77,7 @@ class GitWrapper(GitWrapperI): @property def curr_commit(self): if self._commit is None: - code, out, err = GitWrapper._exec_cmd(['git', 'rev-parse' 'HEAD']) + code, out, err = GitWrapper._exec_cmd(['git', 'rev-parse', 'HEAD']) if code != 0: raise GitCmdError('Git command error - {}'.format(err)) self._commit = out