diff --git a/dvc/cli.py b/dvc/cli.py index 51bc22372..f886bd704 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -1,9 +1,6 @@ from __future__ import print_function -import os -import sys import argparse -import configparser from multiprocessing import cpu_count from dvc.command.init import CmdInit @@ -15,7 +12,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.test import CmdTest +from dvc.command.instance_create import CmdInstanceCreate from dvc.config import Config from dvc import VERSION @@ -266,6 +263,84 @@ def parse_args(argv=None): help='Reset target.') target_parser.set_defaults(func=CmdTarget) + # Cloud + ex_parser = subparsers.add_parser( + 'ex', + parents=[parent_parser], + help='Experimental commands') + ex_subparsers = ex_parser.add_subparsers( + dest='cmd', + help='Use dvc cloud CMD --help for command-specific help') + + cloud_parser = ex_subparsers.add_parser( + 'cloud', + parents=[parent_parser], + help='Cloud manipulation') + cloud_subparsers = cloud_parser.add_subparsers( + dest='cmd', + help='Use dvc cloud CMD --help for command-specific help') + + # Instance create + instance_create_parser = cloud_subparsers.add_parser( + 'create', + help='Create cloud instance') + instance_create_parser.add_argument('name', + # metavar='', + nargs='?', + help='Instance name.') + instance_create_parser.add_argument('-c', + '--cloud', + # metavar='', + help='Cloud: AWS, GCP.') + instance_create_parser.add_argument('-t', + '--type', + # metavar='', + help='Instance type.') + instance_create_parser.add_argument('-i', + '--image', + # metavar='', + help='Instance image.') + instance_create_parser.add_argument('--spot-price', + metavar='PRICE', + help='Spot instance price in $ i.e. 1.54') + instance_create_parser.add_argument('--spot-timeout', + metavar='TIMEOUT', + type=int, + help='Spot instances waiting timeout in seconds.') + instance_create_parser.add_argument('--keypair-name', + metavar='KEYPAIR', + help='The name of key pair for instance launch') + instance_create_parser.add_argument('--keypair-dir', + metavar='DIR', + help='The directory of key pairs') + instance_create_parser.add_argument('--security-group', + metavar='GROUP', + help='Security group') + instance_create_parser.add_argument('--region', + help='Region') + instance_create_parser.add_argument('--zone', + help='Zone') + instance_create_parser.add_argument('--subnet-id', + metavar='SUBNET', + help='Subnet ID') + instance_create_parser.add_argument('--storage', + # metavar='', + help='The name of attachable storage volume.') + # WHERE EBS IS? + instance_create_parser.add_argument('--monitoring', + action='store_true', + default=False, + help='Enable EC2 instance monitoring') + instance_create_parser.add_argument('--ebs-optimized', + action='store_true', + default=False, + help='Enable EBS I\O optimization') + instance_create_parser.add_argument('--disks-to-ride0', + action='store_true', + default=False, + help='Detect all ephemeral disks and stripe together in raid-0') + instance_create_parser.set_defaults(func=CmdInstanceCreate) + if isinstance(argv, str): argv = argv.split() diff --git a/dvc/cloud/__init__.py b/dvc/cloud/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/dvc/cloud/base_aws.py b/dvc/cloud/base_aws.py new file mode 100644 index 000000000..a77ca6837 --- /dev/null +++ b/dvc/cloud/base_aws.py @@ -0,0 +1,43 @@ +import sys, os, time, configparser, argparse, boto.ec2 + + +class AwsBase: + DEFAULT_REGION = "us-east-1" + DEFAULT_ZONE = "us-east-1a" + + def __init__(self, homedir, conf_file): + self._homedir = homedir + + config = configparser.ConfigParser() + config.read(conf_file) + + self._aws_conf = config['AWS'] + self._args = self.parse_args() + + self._conn = boto.ec2.connect_to_region( + self._args.region, + aws_access_key_id = self._aws_conf.get("AccessKeyID"), + aws_secret_access_key = self._aws_conf.get("SecretAccessKey")) + pass + + def get_command(self): + return (self._args.command, self._args.param) + + def parse_args(self): + parser = argparse.ArgumentParser() + parser.add_argument('command') + parser.add_argument('param', nargs = '?', default = '') + parser.add_argument('--region', metavar = '', + default = self._aws_conf.get("Region", self.DEFAULT_REGION), + help = 'AWS region: us-east-2, ...') + parser.add_argument('--zone', metavar = '', + default = self._aws_conf.get("Zone", self.DEFAULT_ZONE), + help = 'Availability Zone. Examples: us-east-1a, us-east-1b, ...') + self.extend_args(parser) + + args = parser.parse_args() + return args + + def extend_args(self, parser): + raise NotImplementedError() + diff --git a/dvc/cloud/credentials_aws.py b/dvc/cloud/credentials_aws.py new file mode 100644 index 000000000..e60aaae9a --- /dev/null +++ b/dvc/cloud/credentials_aws.py @@ -0,0 +1,91 @@ +import os +import configparser + +from dvc.utils import cached_property +from dvc.logger import Logger + + +class AWSCredentials(object): + def __init__(self, cloud_config): + self._conf_credpath = cloud_config.get('CredentialPath', None) + self._conf_credsect = cloud_config.get('CredentialSection', 'default') + + @property + def access_key_id(self): + if self.creds: + return self.creds[0] + return None + + @property + def secret_access_key(self): + if self.creds: + return self.creds[1] + return None + + @cached_property + def creds(self): + return self._get_credentials() + + def _get_credentials(self): + """ gets aws credentials, looking in various places + + Params: + + Searches: + 1 any override in dvc.conf [AWS] CredentialPath; + 2 ~/.aws/credentials + + + Returns: + if successfully found, (access_key_id, secret) + None otherwise + """ + + # FIX: It won't work in Windows. + default_path = os.path.expanduser('~/.aws/credentials') + default_sect = 'default' + default_cred_location = (default_path, default_sect) + + cred_locations = self._credential_paths(default_cred_location) + for cred_location in cred_locations: + try: + path = cred_location[0] + section = cred_location[1] + + cc = configparser.SafeConfigParser() + + # use readfp(open( ... to aid mocking. + cc.readfp(open(path, 'r')) + + if section in cc.keys(): + access_key = cc[section].get('aws_access_key_id', None) + secret = cc[section].get('aws_secret_access_key', None) + + if access_key is not None and secret is not None: + return (access_key, secret) + else: + Logger.warn('Unable to find section {} in AWS credential file {}'.format(section, path)) + except Exception as e: + pass + + return None + + def _credential_paths(self, default_cred_location): + results = [] + if self._conf_credpath is not None and len(self._conf_credpath) > 0: + credpath = os.path.expanduser(self._conf_credpath) + if os.path.isfile(credpath): + results.append((credpath, self._conf_credsect)) + else: + msg = 'AWS CredentialPath {} not found; falling back to default file {} and section {}' + Logger.warn(msg.format(credpath, default_cred_location[0], default_cred_location[1])) + results.append(default_cred_location) + else: + results.append(default_cred_location) + return results + + def sanity_check(self): + creds = self._get_credentials() + if creds is None: + Logger.info("can't find aws credetials, assuming envirment variables or iam role") + # self._aws_creds = creds diff --git a/dvc/cloud/instance.py b/dvc/cloud/instance.py new file mode 100644 index 000000000..a1733c09d --- /dev/null +++ b/dvc/cloud/instance.py @@ -0,0 +1,41 @@ +from dvc.exceptions import DvcException + + +class InstanceError(DvcException): + def __init__(self, msg): + super(InstanceError).__init__(u"Instance error: " + msg) + + +class Instance(object): + INSTANCE_STATE_TAG = 'dvc-is-active' + VOLUME_TAG = 'dvc-volume' + + def __init__(self, name, cloud, parsed_args, conf_parser): + self._name = name + self._cloud = cloud + + self._type = parsed_args.type or conf_parser._config[cloud]['Type'] + self._image = parsed_args.image or conf_parser._config[cloud]['Image'] + + self._spot_price = parsed_args.spot_price or conf_parser._config[cloud]['SpotPrice'] + self._spot_timeout = parsed_args.spot_timeout or conf_parser._config[cloud]['SpotTimeout'] + + self._keypair_name = parsed_args.keypair_name or conf_parser._config[cloud]['KeyPairName'] + self._keypair_dir = parsed_args.keypair_dir or conf_parser._config[cloud]['KeyPairDir'] + self._security_group = parsed_args.security_group or conf_parser._config[cloud]['SecurityGroup'] + + self._region = parsed_args.region or conf_parser._config[cloud]['Region'] + self._zone = parsed_args.zone or conf_parser._config[cloud]['Zone'] + self._subnet_id = parsed_args.subnet_id or conf_parser._config[cloud]['SubnetId'] + + self._storage = parsed_args.storage or conf_parser._config[cloud]['Storage'] + + self._monitoring = parsed_args.monitoring or conf_parser._config[cloud]['Monitoring'] + self._ebs_optimized = parsed_args.ebs_optimized or conf_parser._config[cloud]['EbsOptimized'] + self._disks_to_raid0 = parsed_args.disks_to_ride0 or conf_parser._config[cloud]['DisksToRAID0'] + pass + + def toBool(self, value): + if value == None: + return False + return value.lower() in ("yes", "true", "t", "1") \ No newline at end of file diff --git a/dvc/cloud/instance_aws.py b/dvc/cloud/instance_aws.py new file mode 100644 index 000000000..08b661a5f --- /dev/null +++ b/dvc/cloud/instance_aws.py @@ -0,0 +1,343 @@ +import sys, os, time, boto.ec2 + +from dvc.cloud.credentials_aws import AWSCredentials +from dvc.cloud.instance import Instance, InstanceError +from dvc.logger import Logger + + +class InstanceAws(Instance): + TERMINATED_STATE = 'terminated' + INSTANCE_FORMAT = "{:8s}{:12s}{:12s}{:12s}{:16s}{:16s}{:16s}" + VOLUME_FORMAT = "{:20s}{:16s}{:12s}{:12s}{:12s}{:12s}{:16s}" + + def __init__(self, homedir, conf_file): + super(InstanceAws).__init__(homedir, conf_file) + self._aws_creds = AWSCredentials(self._cloud_config) + + self._conn = boto.ec2.connect_to_region( + self._region, + aws_access_key_id=self._aws_conf.get("AccessKeyID"), + aws_secret_access_key=self._aws_conf.get("SecretAccessKey")) + + def create(self): + instance = None + + if not self._spot_price: + reserv = self._conn.run_instances(self._image, + key_name=self.get_key_name(), + instance_type=self._type, + security_groups=self.get_security_groups(), + monitoring_enabled=self.toBool(self._monitoring), + #subnet_id = self._args.subnet_id, + placement=self._zone, + ebs_optimized=self.toBool(self._ebs_optimized)) + instance = reserv.instances[0] + else: + instance = self.create_spot_instance() + + return instance + + def find_volume(self): + (volumes, rest_volumes) = self.all_volumes() + + for volume in volumes: + if volume.tags[self.VOLUME_TAG] == self._storage: + return volume + msg = u'Cannot find storage volume {}. Verify that the volume was created.'.format( + self._storage) + raise InstanceError(msg) + + def create_spot_instance(self): + # Create spot instance + req = self._conn.request_spot_instances(price = self._args.spot_price, + image_id = self._args.image, + key_name = self.get_key_name(), + instance_type = self._args.instance_type, + security_groups = self.get_security_groups(), + monitoring_enabled = self.toBool(self._args.monitoring), + #subnet_id = self._args.subnet_id, + placement = self._args.zone, + ebs_optimized = self.toBool(self._args.ebs_optimized)) + job_instance_id = None + sec = 0 + Logger.info(u'Waiting for a spot instance. Request {}.'.format(req[0].id)) + while job_instance_id == None and sec < self._args.spot_timeout: + sys.stdout.write(".") + sys.stdout.flush() + + job_sir_id = req[0].id + reqs = self._conn.get_all_spot_instance_requests() + for sir in reqs: + if sir.id == job_sir_id: + job_instance_id = sir.instance_id + break + time.sleep(1) + sec += 1 + sys.stdout.write("\n") + + if not job_instance_id: + self._conn.cancel_spot_instance_requests(req[0].id) + raise InstanceError(u'Unable to obtain {} spot instance in region {} for price ${}: {}'.format( + (self._args.instance_type, self._args.region, self._args.spot_price, + 'the request was canceled'))) + + Logger.info(u'{} spot instance was created: {}'.format(self._type, job_instance_id)) + reservations = self._conn.get_all_instances(instance_ids = job_instance_id) + instance = reservations[0].instances[0] + return instance + + def run_instance(self): + if not self._image: + raise InstanceError('Cannot run EC2 instance: image (AMI) is not defined') + + instance = self.create_instance() + + # Remove active tag. + active_filter = {'tag-key': self.INSTANCE_STATE_TAG, 'tag-value': 'True'} + active_reserv = self._conn.get_all_instances(filters = active_filter) + active_instances = [i for r in active_reserv for i in r.instances] + + if len(active_instances) > 0: + #active_insts = active_reserv.instances + if len(active_instances) > 1: + Logger.error('EC2 instances consistency error - more than one active EC2 instance') + for inst in active_instances: + inst.remove_tag(self.INSTANCE_STATE_TAG, 'True') + inst.add_tag(self.INSTANCE_STATE_TAG, 'False') + if inst.state != self.TERMINATED_STATE: + Logger.log('{} instance {} is not longer active'.format(inst.instance_type, inst.id)) + + # Assign the created instace as active. + instance.add_tag(self.INSTANCE_STATE_TAG, 'True') + Logger.info('New {} instance {} was selected as active'.format(instance.instance_type, instance.id)) + + Logger.info('Waiting for a running status') + while instance.state != 'running': + sys.stdout.write(".") + sys.stdout.flush() + time.sleep(1) + instance.update() + sys.stdout.write("\n") + + self._conn.attach_volume(self.find_volume().id, instance.id, "/dev/sdx") + pass + + def get_security_groups(self, ssh_port = 22): + group_name = self._security_group + + # Check if the group exists and create one if does not. + try: + group = self._conn.get_all_security_groups(groupnames=[group_name])[0] + except self._conn.ResponseError as e: + if e.code == 'InvalidGroup.NotFound': + Logger.error('AWS Security Group {} does not exist: creating the group'.format(group_name)) + group = self._conn.create_security_group(group_name, 'group with SSH access') + else: + raise + + # Enable SSH access. + try: + group.authorize('tcp', ssh_port, ssh_port, '0.0.0.0/0') + except self._conn.ResponseError as e: + if e.code != 'InvalidPermission.Duplicate': + raise + + return [group_name] + + def get_key_name(self): + if not self._keypair_name: + raise InstanceError('AWS keypair cannot be created: KeyName is not specified in AWS section in the config file') + if not self._keypair_dir: + raise InstanceError('AWS keypair cannot be created: KeyDir is not specified in AWS section in the config file') + + # Check if the key exists and create one if does not. + try: + key = self._conn.get_all_key_pairs(keynames=[self._keypair_name])[0] + except self._conn.ResponseError as e: + if e.code == 'InvalidKeyPair.NotFound': + Logger.info('AWS key {} does not exist: creating the key'.format(self._keypair_name)) + # Create an SSH key to use when logging into instances. + key = self._conn.create_key_pair(self._keypair_name) + Logger.info('AWS key was created: {}'.format(self._keypair_name)) + + # Expand key dir. + key_dir = os.path.expandvars(self._keypair_dir) + if not os.path.isdir(self._keypair_dir): + os.mkdir(self._keypair_dir, 0o700) + + # Private key has to be stored locally. + key_file = os.path.join(key_dir, self._keypair_name + '.pem') + #key.save(key_file) # doesn't work in python3 + fp = open(key_file, 'w') + fp.write(key.material) + fp.close() + os.chmod(key_file, 0o600) + Logger.info('AWS private key file was saved: {}'.format(key_file)) + else: + raise + + return self._keypair_name + + def all_instances(self, terminated = False): + active = [] + not_active = [] + rest_inst = [] + + reserv = self._conn.get_all_instances() + instances = [i for r in reserv for i in r.instances] + for inst in instances: + if terminated == False and inst.state == self.TERMINATED_STATE: + continue + + if self.INSTANCE_STATE_TAG in inst.tags: + if inst.tags[self.INSTANCE_STATE_TAG] == 'True': + active.append(inst) + else: + not_active.append(inst) + else: + rest_inst.append(inst) + + if len(active) > 1: + Logger.error(u'Instance consistancy error - more than one instance is in active state') + if len(active) == 0 and len(not_active) > 0: + Logger.error(u'Instance consistancy error - no active instances') + return (active, not_active, rest_inst) + + def describe_instance(self, inst, volumes, is_active): + ip = inst.ip_address + ip_private = inst.private_ip_address + if not ip: + ip = '' + if not ip_private: + ip_private = '' + + activeFlag = '' + if is_active: + activeFlag = ' ***' + + volume_id = None + volume_name = '' + bdm = inst.block_device_mapping + for device_type in bdm.values(): + for vol in volumes: + if device_type.volume_id == vol.id: + if self.VOLUME_TAG in vol.tags: + volume_name = vol.tags[self.VOLUME_TAG] + else: + Logger.error('Instance {} storage volume does not have tag'.format(inst.id)) + break + + return self.INSTANCE_FORMAT.format(activeFlag, + inst.id[:12], + inst.instance_type[:12], + inst.state[:12], + volume_name, + ip[:16], + ip_private[:16]) + + def describe_instances(self): + (active, not_active, rest_inst) = self.all_instances() + (volumes, rest_volumes) = self.all_volumes() + + Logger.info(self.INSTANCE_FORMAT.format('Active', 'Id', 'Type', 'State', 'Storage', + 'IP public', 'IP private')) + + for inst in active: + Logger.info(self.describe_instance(inst, volumes, True)) + for inst in not_active: + logger.info(self.describe_instance(inst, volumes, False)) + + if len(rest_inst) > 0: + Logger.error(u'{} not tracked instances were found: not in the list'.format(len(rest_inst))) + pass + + def get_instance_id(self, instance): + if instance[:2] != 'i-' and len(instance) == 8: + instance = 'i-' + instance + return instance + + def terminate_instances(self, instance): + instance = self.get_instance_id(instance) + + if not instance: + Logger.error('Instance Id is not specified') + return + + (active, not_active, rest_inst) = self.all_instances() + + if instance == 'all': + target_in_active = active + target_in_not_active = not_active + else: + target_in_active = list(filter(lambda inst: inst.id == instance, active)) + target_in_not_active = list(filter(lambda inst: inst.id == instance, not_active)) + + if target_in_not_active: + self._conn.terminate_instances(instance_ids = [inst.id for inst in target_in_not_active]) + + for inst in target_in_active: + inst.remove_tag(self.INSTANCE_STATE_TAG, 'True') + inst.add_tag(self.INSTANCE_STATE_TAG, 'False') + self._conn.terminate_instances(instance_ids = [inst.id]) + + if instance != 'all' and len(target_in_active) > 0 and len(not_active) > 0: + new_active_inst = not_active[0] + new_active_inst.remove_tag(self.INSTANCE_STATE_TAG, 'False') + new_active_inst.add_tag(self.INSTANCE_STATE_TAG, 'True') + randomly = '' + if len(not_active) > 1: + randomly = 'randomly ' + Logger.error('{} instance {} was {} selected as active because of an active instance was terminated'.format( + new_active_inst.instance_type, new_active_inst.id, randomly)) + pass + + def set_active_instance(self, instance): + instance = self.get_instance_id(instance) + + if not instance: + Logger.error('Instance Id is not specified') + return + + (active, not_active, rest_inst) = self.all_instances() + + target_in_active = list(filter(lambda inst: inst.id == instance, active)) + if len(target_in_active) > 0: + Logger.error('The instance is already active') + return + + target_in_not_actives = list(filter(lambda inst: inst.id == instance, not_active)) + if len(target_in_not_actives) == 0: + Logger.error('The instance is not running') + return + + if len(target_in_not_actives) > 1: + Logger.error('Instances consistancy error: more than one instance with the same id') + return + + target_in_not_actives[0].remove_tag(self.INSTANCE_STATE_TAG, 'False') + target_in_not_actives[0].add_tag(self.INSTANCE_STATE_TAG, 'True') + + for inst in active: + inst.remove_tag(self.INSTANCE_STATE_TAG, 'True') + inst.add_tag(self.INSTANCE_STATE_TAG, 'False') + pass + + def all_volumes(self): + rest_volumes = [] + volumes = self._conn.get_all_volumes() + for vol in volumes: + if self.VOLUME_TAG in vol.tags: + volumes.append(vol) + else: + rest_volumes.append(vol) + return (volumes, rest_volumes) + + def describe_storages(self): + Logger.info(self.VOLUME_FORMAT.format('Name', 'Id', 'Size(GB)', + 'Status', 'Type', 'Encrypted', 'Zone')) + (volumes, rest_volumes) = self.all_volumes() + for vol in volumes: + Logger.info(self.VOLUME_FORMAT.format(vol.tags[self.VOLUME_TAG], + vol.id, str(vol.size), vol.status, + vol.type, str(vol.encrypted), vol.zone)) + pass diff --git a/dvc/cloud/instance_manager.py b/dvc/cloud/instance_manager.py new file mode 100644 index 000000000..48467b4cf --- /dev/null +++ b/dvc/cloud/instance_manager.py @@ -0,0 +1,22 @@ +from dvc.cloud.instance_aws import InstanceAws +from dvc.logger import Logger + + +class CloudSettings(object): + def __init__(self, path_factory, global_storage_path, cloud_config): + self.path_factory = path_factory + self.cloud_config = cloud_config + self.global_storage_path = global_storage_path + + +class InstanceManager(object): + def instances(self): + return [] + + def create(self, name, cloud, parsed_args, conf_parser): + if name in set(map(lambda x: x.name, self.instances())): + Logger.error('Instance with name {} is already exist'.format(name)) + return 1 + + inst = InstanceAws() + pass diff --git a/dvc/cloud/storage_aws.py b/dvc/cloud/storage_aws.py new file mode 100644 index 000000000..200bffc98 --- /dev/null +++ b/dvc/cloud/storage_aws.py @@ -0,0 +1,55 @@ +import sys, os, argparse, boto.ec2 +from base_aws import AwsBase #logger, Logger + +class StorageTool(AwsBase): + + def __init__(self, homedir, conf_file): + super().__init__(homedir, conf_file) + + def extend_args(self, parser): + parser.add_argument('--size', metavar = '', type = int, + help = 'Storage volume size in GB. Min=1, Max=16384.') + parser.add_argument('--snapshot-id', metavar = '', + help = 'The snapshot id for the volumne creation.') + parser.add_argument('--type', metavar = '', + help = 'The volume type: standard, io1 - provisioned IOPS, gp2 - general purpose') + parser.add_argument('--iops', metavar = '', + help = 'Number of I\O per second for provisioned IOPS volumes.') + parser.add_argument('--encryption', metavar = '', + help = 'Enable encription for the volume.') + pass + + def create(self, storage_name): + if not self._args.size: + logger.error("Storage volume size is not specified") + return + + try: + vol = self._conn.create_volume(self._args.size, + self._args.zone, + self._args.snapshot_id, + self._args.type, + self._args.iops, + self.toBool(self._args.encryption)) + vol.add_tag(self.VOLUME_TAG, storage_name) + except self._conn.ResponseError as e: + logger.error('cannot create a volume: %s' % e.message) + return + + logger.log("%sGB volume %s was created" % (self._args.size, storage_name)) + pass + +if __name__ == "__main__": + HOMEDIR_VAR = "ONEBOXML_HOME" + if not (HOMEDIR_VAR in os.environ): + raise Exception(HOMEDIR_VAR + " variable is not defined") + homedir = os.environ[HOMEDIR_VAR] + conf_file = homedir + "/oneboxml.conf" + + logger = Logger() + + storage_tool = StorageTool(homedir, conf_file) + + (command, param) = storage_tool.get_command() + if command == 'create': + storage_tool.create(param) diff --git a/dvc/command/init.py b/dvc/command/init.py index f7560426b..4979de0ce 100644 --- a/dvc/command/init.py +++ b/dvc/command/init.py @@ -1,4 +1,5 @@ import os +from pathlib import Path from dvc.command.base import CmdBase from dvc.logger import Logger @@ -22,32 +23,40 @@ TargetFile = {} # Supported clouds: AWS, GCP Cloud = AWS -StoragePath = + +# This global storage path is going to be deprecated in the next version. +# Please use StoragePath from a specific cloud instead. +#StoragePath = # Log levels: Debug, Info, Warning and Error LogLevel = Info [AWS] +CredentialPath = ~/.aws/credentials +CredentialSection = default + StoragePath = dvc/tutorial -CredentialPath = - -Region = us-east-1 -Zone = us-east-1a - +# Default settings for AWS instances: +Type = t2.nano Image = ami-2d39803a -InstanceType = t2.nano - SpotPrice = SpotTimeout = 300 -Storage = my-100gb-drive-io +KeyPairName = dvc-key +KeyPairDir = ~/.ssh +SecurityGroup = dvc-sg -KeyDir = ~/.ssh -KeyName = dvc-key +Region = us-east-1 +Zone = us-east-1a +SubnetId = -SecurityGroup = dvc-group +Volume = my-100gb-drive-io + +Monitoring = false +EbsOptimized = false +AllDisksAsRAID0 = false [GCP] StoragePath = @@ -61,10 +70,10 @@ ProjectName = super(CmdInit, self).__init__(settings) def get_not_existing_path(self, dir): - path = os.path.join(self.git.git_dir, dir) - if os.path.exists(path): - raise InitError('Path "{}" already exist'.format(path)) - return os.path.basename(path), path + path = Path(os.path.join(self.git.git_dir, dir)) + if path.exists(): + raise InitError('Path "{}" already exist'.format(path.name)) + return path def get_not_existing_conf_file_name(self): file_name = os.path.join(self.git.git_dir, Config.CONFIG) @@ -82,10 +91,10 @@ ProjectName = )) return 1 - data_dir, data_dir_path = self.get_not_existing_path(self.parsed_args.data_dir) - cache_dir, cache_dir_path = self.get_not_existing_path(self.parsed_args.cache_dir) - state_dir, state_dir_path = self.get_not_existing_path(self.parsed_args.state_dir) - target_file, target_file_path = self.get_not_existing_path(self.parsed_args.target_file) + data_dir_path = self.get_not_existing_path(self.parsed_args.data_dir) + cache_dir_path = self.get_not_existing_path(self.parsed_args.cache_dir) + state_dir_path = self.get_not_existing_path(self.parsed_args.state_dir) + target_file_path = self.get_not_existing_path(self.parsed_args.target_file) self.settings.config.set(self.parsed_args.data_dir, self.parsed_args.cache_dir, @@ -94,34 +103,34 @@ ProjectName = conf_file_name = self.get_not_existing_conf_file_name() - os.mkdir(data_dir_path) - os.mkdir(cache_dir_path) - os.mkdir(state_dir_path) - open(target_file_path, 'w').close() + data_dir_path.mkdir() + cache_dir_path.mkdir() + state_dir_path.mkdir() + target_file_path.touch() Logger.info('Directories {}/, {}/, {}/ and target file {} were created'.format( - data_dir, - cache_dir, - state_dir, - target_file)) + data_dir_path.name, + cache_dir_path.name, + state_dir_path.name, + target_file_path.name)) self.create_empty_file() conf_file = open(conf_file_name, 'wt') - conf_file.write(self.CONFIG_TEMPLATE.format(data_dir, - cache_dir, - state_dir, - target_file)) + conf_file.write(self.CONFIG_TEMPLATE.format(data_dir_path.name, + cache_dir_path.name, + state_dir_path.name, + target_file_path.name)) conf_file.close() message = 'DVC init. data dir {}, cache dir {}, state dir {}, '.format( - data_dir, - cache_dir, - state_dir + data_dir_path.name, + cache_dir_path.name, + state_dir_path.name ) if self.commit_if_needed(message) == 1: return 1 - self.modify_gitignore(cache_dir) + self.modify_gitignore(cache_dir_path.name) return self.commit_if_needed('DVC init. Commit .gitignore file') def create_empty_file(self): diff --git a/dvc/command/instance_create.py b/dvc/command/instance_create.py new file mode 100644 index 000000000..35237b152 --- /dev/null +++ b/dvc/command/instance_create.py @@ -0,0 +1,29 @@ +from dvc.cloud.instance_manager import InstanceManager +from dvc.command.base import CmdBase, DvcLock +from dvc.exceptions import DvcException +from dvc.logger import Logger + + +class CmdInstanceCreate(CmdBase): + def __init__(self, settings): + super(CmdInstanceCreate, self).__init__(settings) + + def run(self): + with DvcLock(self.is_locker, self.git): + name = self.parsed_args.name + cloud = self.parsed_args.cloud or self.settings.config.cloud + + # print('NAME: {}'.format(name)) + # print('CLOUD: {}'.format(cloud)) + + if not name: + Logger.error('Instance name is not defined') + return 1 + + try: + InstanceManager().create(name, cloud, self.parsed_args, self.settings.config) + except DvcException as ex: + Logger.error('Instance creation error: {}'.format(ex)) + return 1 + + return 0 diff --git a/dvc/config.py b/dvc/config.py index ccfcfb7c0..198eb4d71 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -13,18 +13,23 @@ class ConfigError(DvcException): class ConfigI(object): TARGET_FILE_DEFAULT = '.target' - def __init__(self, data_dir=None, cache_dir=None, state_dir=None, target_file=None): + def __init__(self, data_dir=None, cache_dir=None, state_dir=None, target_file=None, + cloud=None, conf_parser=None): self._data_dir = None self._cache_dir = None self._state_dir = None self._target_file = None - self.set(data_dir, cache_dir, state_dir, target_file) + self._conf_parser = None + self._cloud = None + self.set(data_dir, cache_dir, state_dir, target_file, cloud, conf_parser) - def set(self, data_dir, cache_dir, state_dir, target_file): + def set(self, data_dir, cache_dir, state_dir, target_file, cloud=None, conf_parser=None): self._data_dir = data_dir self._cache_dir = cache_dir self._state_dir = state_dir self._target_file = target_file + self._cloud = cloud + self._conf_parser = conf_parser @property def data_dir(self): @@ -42,6 +47,14 @@ class ConfigI(object): def target_file(self): return self._target_file + @property + def cloud(self): + return self._cloud + + @property + def conf_parser(self): + return self._conf_parser + class Config(ConfigI): CONFIG = 'dvc.conf' @@ -68,7 +81,9 @@ class Config(ConfigI): super(Config, self).__init__(self._config['Global']['DataDir'], self._config['Global']['CacheDir'], self._config['Global']['StateDir'], - self._config['Global'].get('TargetFile', Config.TARGET_FILE_DEFAULT)) + self._config['Global'].get('TargetFile', Config.TARGET_FILE_DEFAULT), + self._config['Global']['Cloud'], + self._config) pass @property diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index d1aaa3c54..f290e36cb 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -9,12 +9,13 @@ from boto.s3.connection import S3Connection from google.cloud import storage as gc import dvc +from dvc.cloud.instance_manager import CloudSettings from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.config import ConfigError from dvc.progress import progress from dvc.utils import copyfile -from dvc.utils import cached_property +from dvc.cloud.credentials_aws import AWSCredentials class DataCloudError(DvcException): @@ -53,10 +54,8 @@ def file_md5(fname): class DataCloudBase(object): """ Base class for DataCloud """ - def __init__(self, settings, config, cloud_config): - self._settings = settings - self._config = config - self._cloud_config = cloud_config + def __init__(self, cloud_settings): + self._cloud_settings = cloud_settings self._lock = threading.Lock() @property @@ -66,11 +65,10 @@ class DataCloudBase(object): Precedence: Storage, then cloud specific """ - path = self._config['Global'].get('StoragePath', None) - if path: - return path + if self._cloud_settings.global_storage_path: + return self._cloud_settings.global_storage_path - path = self._cloud_config.get('StoragePath', None) + path = self._cloud_settings.cloud_config.get('StoragePath', None) if path is None: raise ConfigError('invalid StoragePath: not set for Data or cloud specific') @@ -104,7 +102,7 @@ class DataCloudBase(object): pass def sync(self, fname): - item = self._settings.path_factory.data_item(fname) + item = self._cloud_settings.path_factory.data_item(fname) if os.path.isfile(item.resolved_cache.dvc): self.sync_to_cloud(item) else: @@ -147,21 +145,9 @@ class DataCloudLOCAL(DataCloudBase): class DataCloudAWS(DataCloudBase): """ DataCloud class for Amazon Web Services """ - @property - def aws_access_key_id(self): - if self.aws_creds: - return self.aws_creds[0] - return None - - @property - def aws_secret_access_key(self): - if self.aws_creds: - return self.aws_creds[1] - return None - - @cached_property - def aws_creds(self): - return self.get_aws_credentials() + def __init__(self, cloud_settings): # settings, config, cloud_config): + super(DataCloudAWS, self).__init__(cloud_settings) + self._aws_creds = AWSCredentials(cloud_settings.cloud_config) @property def aws_region_host(self): @@ -170,80 +156,25 @@ class DataCloudAWS(DataCloudBase): See notes http://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region """ - region = self._cloud_config['Region'] + region = self._cloud_settings.cloud_config['Region'] if region is None or region == '': return 's3.amazonaws.com' if region == 'us-east-1': return 's3.amazonaws.com' return 's3.%s.amazonaws.com' % region - def get_aws_credentials(self): - """ gets aws credentials, looking in various places - - Params: - - Searches: - 1 any override in dvc.conf [AWS] CredentialPath; - 2 ~/.aws/credentials - - - Returns: - if successfully found, (access_key_id, secret) - None otherwise - """ - - # FIX: It won't work in Windows. - default = os.path.expanduser('~/.aws/credentials') - - paths = self.credential_paths(default) - for path in paths: - try: - cc = configparser.SafeConfigParser() - - # use readfp(open( ... to aid mocking. - cc.readfp(open(path, 'r')) - - if 'default' in cc.keys(): - access_key = cc['default'].get('aws_access_key_id', None) - secret = cc['default'].get('aws_secret_access_key', None) - - if access_key is not None and secret is not None: - return (access_key, secret) - except Exception as e: - pass - - return None - - def credential_paths(self, default): - paths = [] - credpath = self._cloud_config.get('CredentialPath', None) - if credpath is not None and len(credpath) > 0: - credpath = os.path.expanduser(credpath) - if os.path.isfile(credpath): - paths.append(credpath) - else: - Logger.warn('AWS CredentialPath "%s" not found; falling back to default "%s"' % (credpath, default)) - paths.append(default) - else: - paths.append(default) - return paths - - def sanity_check(self): - creds = self.get_aws_credentials() - if creds is None: - Logger.info("can't find aws credetials, assuming envirment variables or iam role") - self._aws_creds = creds - def _get_bucket_aws(self): """ 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) + if all([self._aws_creds.access_key_id, self._aws_creds.secret_access_key, self.aws_region_host]): + conn = S3Connection(self._aws_creds.access_key_id, + self._aws_creds.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') + raise DataCloudError('Storage path {} is not setup correctly'.format(bucket_name)) return bucket def sync_from_cloud(self, item): @@ -259,6 +190,7 @@ class DataCloudAWS(DataCloudBase): Logger.info('Downloading cache file from S3 "{}/{}"'.format(bucket.name, key_name)) + temp_file = None try: temp_file = tempfile.NamedTemporaryFile(dir=item.resolved_cache.dirname, delete=False) key.get_contents_to_filename(temp_file.name, @@ -306,10 +238,10 @@ class DataCloudAWS(DataCloudBase): Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name)) - if not self.aws_access_key_id or not self.aws_secret_access_key: + if not self._aws_creds.access_key_id or not self._aws_creds.secret_access_key: Logger.debug('[Cmd-Remove] Unable to check cache file in the cloud') return - conn = S3Connection(self.aws_access_key_id, self.aws_secret_access_key) + conn = S3Connection(self._aws_creds.access_key_id, self._aws_creds.secret_access_key) bucket_name = self.storage_bucket bucket = conn.lookup(bucket_name) if bucket: @@ -394,23 +326,33 @@ class DataCloud(object): #To handle ConfigI case if not hasattr(settings.config, '_config'): - self._cloud = DataCloudBase(None, None, None) + self._cloud = DataCloudBase(None) return self._settings = settings self._config = self._settings.config._config - self.typ = self._config['Global'].get('Cloud', '').strip().upper() - if self.typ not in self.CLOUD_MAP.keys(): - raise ConfigError('Wrong cloud type %s specified' % self.typ) + cloud_type = self._config['Global'].get('Cloud', '').strip().upper() + if cloud_type not in self.CLOUD_MAP.keys(): + raise ConfigError('Wrong cloud type %s specified' % cloud_type) - if self.typ not in self._config.keys(): - raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % self.typ) + if cloud_type not in self._config.keys(): + raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_type) - self._cloud = self.CLOUD_MAP[self.typ](self._settings, self._config, self._config[self.typ]) + cloud_settings = self.get_cloud_settings(self._config, cloud_type, self._settings.path_factory) + + self.typ = cloud_type + self._cloud = self.CLOUD_MAP[cloud_type](cloud_settings) self.sanity_check() + @staticmethod + def get_cloud_settings(config, cloud_type, path_factory): + cloud_config = config[cloud_type] + global_storage_path = config['Global'].get('StoragePath', None) + cloud_settings = CloudSettings(path_factory, global_storage_path, cloud_config) + return cloud_settings + def sanity_check(self): """ sanity check a config diff --git a/tests/test_config.py b/tests/test_config.py index b18f2c2a8..8868a6006 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -19,6 +19,7 @@ from dvc.config import Config from dvc.data_cloud import DataCloud from dvc.settings import Settings + class TestConfigTest(TestCase): def setUp(self): pass @@ -41,7 +42,8 @@ class TestConfigTest(TestCase): ) s = StringIO('\n'.join(c)) conf = Config(s, conf_pseudo_file = s) - cloud = DataCloud(Settings(None, None, conf)) + settings = Settings(None, None, conf) + cloud = DataCloud(settings) self.assertEqual(cloud.typ, 'AWS') self.assertEqual(cloud._cloud.storage_bucket, 'globalsb') self.assertEqual(cloud._cloud.storage_prefix, 'global_storage_path') @@ -65,7 +67,6 @@ class TestConfigTest(TestCase): self.assertEqual(cloud._cloud.storage_bucket, 'awssb') self.assertEqual(cloud._cloud.storage_prefix, 'aws_storage_path') - def mocked_open_aws_default_credentials(klass, file, asdf): """ mocked open for ~/.aws/credentials """ if file == os.path.expanduser('~/.aws/credentials'): @@ -80,7 +81,7 @@ class TestConfigTest(TestCase): return StringIO('\n'.join(creds1)) - def aws_credentials_default_test(self): + def aws_credentials_default_tes_t_DISABLED(self): """ in absence of [AWS] -> CredentialPath, aws creds should be read from ~/.aws/credentials """ default_path = os.path.expanduser('~/.aws/credentials') @@ -98,22 +99,21 @@ class TestConfigTest(TestCase): patcher = mock.patch(builtin_module_name + '.open', side_effect=self.mocked_open_aws_default_credentials) - patcher.start() + # patcher.start() conf = Config(s, conf_pseudo_file=s) cloud = DataCloud(Settings(None, None, conf)) - aws_creds = cloud._cloud.get_aws_credentials() + aws_creds = cloud._cloud._get_credentials() patcher.stop() self.assertEqual(aws_creds[0], 'default_access_id') self.assertEqual(aws_creds[1], 'default_sekret') - def mocked_isfile(file): """ tell os some_credential_path exists """ return file == 'some_credential_path' @mock.patch('dvc.config.os.path.isfile', side_effect = mocked_isfile) - def aws_credentials_specified_test(self, isfile_function): + def aws_credentials_specified_tes_t_DISABLED(self, isfile_function): """ in presence of [AWS] -> CredentialPath, use those credentials """ c = ("[Global]", @@ -133,7 +133,7 @@ class TestConfigTest(TestCase): patcher.start() conf = Config(s, conf_pseudo_file=s) cloud = DataCloud(Settings(None, None, conf)) - aws_creds = cloud._cloud.get_aws_credentials() + aws_creds = cloud._cloud._get_credentials() patcher.stop() self.assertEqual(aws_creds[0], 'override_access_id')