1123 строки
45 KiB
Python
1123 строки
45 KiB
Python
# Copyright (c) Microsoft Corporation
|
|
#
|
|
# All rights reserved.
|
|
#
|
|
# MIT License
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining a
|
|
# copy of this software and associated documentation files (the "Software"),
|
|
# to deal in the Software without restriction, including without limitation
|
|
# the rights to use, copy, modify, merge, publish, distribute, sublicense,
|
|
# and/or sell copies of the Software, and to permit persons to whom the
|
|
# Software is furnished to do so, subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be included in
|
|
# all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
|
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
|
# DEALINGS IN THE SOFTWARE.
|
|
|
|
# compat imports
|
|
from __future__ import (
|
|
absolute_import, division, print_function, unicode_literals
|
|
)
|
|
from builtins import ( # noqa
|
|
bytes, dict, int, list, object, range, str, ascii, chr, hex, input,
|
|
next, oct, open, pow, round, super, filter, map, zip)
|
|
# stdlib imports
|
|
import datetime
|
|
import fnmatch
|
|
import logging
|
|
import math
|
|
import os
|
|
try:
|
|
import pathlib2 as pathlib
|
|
except ImportError:
|
|
import pathlib
|
|
try:
|
|
from shlex import quote as shellquote
|
|
except ImportError:
|
|
from pipes import quote as shellquote
|
|
import threading
|
|
import time
|
|
# non-stdlib imports
|
|
import azure.batch.models as batchmodels
|
|
# local imports
|
|
from . import crypto
|
|
from . import resource
|
|
from . import settings
|
|
from . import storage
|
|
from . import util
|
|
from .version import __version__
|
|
|
|
# create logger
|
|
logger = logging.getLogger(__name__)
|
|
util.setup_logger(logger)
|
|
# global defines
|
|
_BLOBXFER_VERSION = '1.7.1'
|
|
_MEGABYTE = 1048576
|
|
_MAX_READ_BLOCKSIZE_BYTES = 4194304
|
|
_FILE_SPLIT_PREFIX = '_shipyard-'
|
|
|
|
|
|
def _get_gluster_paths(config):
|
|
# type: (dict) -> Tuple[str, str]
|
|
"""Get Gluster paths
|
|
:param dict config: configuration dict
|
|
:rtype: tuple
|
|
:return: (gluster host path, gluster container path)
|
|
"""
|
|
gluster_host = None
|
|
gluster_container = None
|
|
sdv = settings.global_resources_shared_data_volumes(config)
|
|
for sdvkey in sdv:
|
|
if settings.is_shared_data_volume_gluster_on_compute(sdv, sdvkey):
|
|
gluster_host = '{}/{}'.format(
|
|
settings.get_host_mounts_path(False),
|
|
settings.get_gluster_on_compute_volume())
|
|
gluster_container = settings.shared_data_volume_container_path(
|
|
sdv, sdvkey).rstrip('/')
|
|
break
|
|
return (gluster_host, gluster_container)
|
|
|
|
|
|
def _convert_filter_to_blobxfer_option(includes, excludes):
|
|
# type: (list, list) -> str
|
|
"""Converts filters to blobxfer options
|
|
:param list includes: includes
|
|
:param list excludes: excludes
|
|
:rtype: str
|
|
:return: blobxfer options
|
|
"""
|
|
if util.is_not_empty(includes):
|
|
src_incl = []
|
|
for include in includes:
|
|
src_incl.append('--include \"{}\"'.format(include))
|
|
else:
|
|
src_incl = None
|
|
if util.is_not_empty(excludes):
|
|
src_excl = []
|
|
for exclude in excludes:
|
|
src_excl.append('--exclude \"{}\"'.format(exclude))
|
|
else:
|
|
src_excl = None
|
|
return '{} {}'.format(
|
|
' '.join(src_incl) if src_incl is not None else '',
|
|
' '.join(src_excl) if src_excl is not None else '',
|
|
).rstrip()
|
|
|
|
|
|
def _process_storage_input_data(config, input_data, on_task):
|
|
# type: (dict, dict, bool) -> str
|
|
"""Process Azure storage input data to ingress
|
|
:param dict config: configuration dict
|
|
:param dict input_data: config spec with input_data
|
|
:param bool on_task: if this is originating from a task spec
|
|
:rtype: list
|
|
:return: args to pass to blobxfer script
|
|
"""
|
|
# get gluster host/container paths
|
|
gluster_host, gluster_container = _get_gluster_paths(config)
|
|
# parse storage input data blocks
|
|
encrypt = settings.batch_shipyard_encryption_enabled(config)
|
|
args = []
|
|
for xfer in input_data:
|
|
storage_settings = settings.credentials_storage(
|
|
config, settings.data_storage_account_settings(xfer))
|
|
remote_path = settings.data_remote_path(xfer)
|
|
# derive container from remote_path
|
|
container = settings.data_container_from_remote_path(
|
|
xfer, rp=remote_path)
|
|
eo = settings.data_blobxfer_extra_options(xfer)
|
|
# append appropriate option for fshare
|
|
if settings.data_is_file_share(xfer) and '--mode file' not in eo:
|
|
eo = '--mode file {}'.format(eo)
|
|
if '--mode file' in eo:
|
|
# create saskey for file share with rl perm
|
|
saskey = storage.create_file_share_saskey(
|
|
storage_settings, container, 'ingress')
|
|
else:
|
|
# create saskey for container with rl perm
|
|
saskey = storage.create_blob_container_saskey(
|
|
storage_settings, container, 'ingress')
|
|
includes = settings.data_include(xfer)
|
|
excludes = settings.data_exclude(xfer)
|
|
# convert include/excludes into extra options
|
|
filters = _convert_filter_to_blobxfer_option(includes, excludes)
|
|
local_path = settings.data_local_path(xfer, on_task)
|
|
# auto replace container path for gluster with host path
|
|
if (util.is_not_empty(gluster_container) and
|
|
local_path.startswith(gluster_container)):
|
|
local_path = local_path.replace(gluster_container, gluster_host, 1)
|
|
# construct argument
|
|
# kind:encrypted:<sa:ep:saskey:remote_path>:local_path:eo
|
|
creds = crypto.encrypt_string(
|
|
encrypt,
|
|
'{},{},{},{}'.format(
|
|
storage_settings.account, storage_settings.endpoint,
|
|
saskey, remote_path),
|
|
config)
|
|
args.append('"{bxver},i,{enc},{creds},{lp},{eo}"'.format(
|
|
bxver=_BLOBXFER_VERSION,
|
|
enc=encrypt,
|
|
creds=creds,
|
|
lp=local_path,
|
|
eo=' '.join((filters, eo)).lstrip(),
|
|
))
|
|
return args
|
|
|
|
|
|
def _process_batch_input_data(config, input_data, on_task):
|
|
# type: (dict, dict, bool) -> str
|
|
"""Process Azure batch input data to ingress
|
|
:param dict config: configuration dict
|
|
:param dict input_data: config spec with input_data
|
|
:param bool on_task: if this is originating from a task spec
|
|
:rtype: list
|
|
:return: args to pass to task file mover
|
|
"""
|
|
# get batch creds
|
|
bc = settings.credentials_batch(config)
|
|
# fail (for now) if aad is being used
|
|
if util.is_none_or_empty(bc.account_key):
|
|
raise RuntimeError(
|
|
'cannot move Azure Batch task input data without an account key')
|
|
# construct arg
|
|
encrypt = settings.batch_shipyard_encryption_enabled(config)
|
|
args = []
|
|
for xfer in input_data:
|
|
jobid = settings.input_data_job_id(xfer)
|
|
taskid = settings.input_data_task_id(xfer)
|
|
include = settings.data_include(xfer)
|
|
if util.is_not_empty(include):
|
|
include = ';'.join(include)
|
|
else:
|
|
include = ''
|
|
exclude = settings.data_exclude(xfer)
|
|
if util.is_not_empty(exclude):
|
|
exclude = ';'.join(exclude)
|
|
else:
|
|
exclude = ''
|
|
local_path = settings.data_local_path(xfer, on_task)
|
|
creds = crypto.encrypt_string(
|
|
encrypt,
|
|
'{};{};{}'.format(
|
|
bc.account, bc.account_service_url, bc.account_key),
|
|
config)
|
|
# construct argument
|
|
# encrypt,creds,jobid,taskid,incl,excl,lp
|
|
args.append('"{},{},{},{},{},{},{}"'.format(
|
|
encrypt, creds, jobid, taskid, include, exclude, local_path))
|
|
return args
|
|
|
|
|
|
def process_input_data(config, bxfile, spec, on_task=False):
|
|
# type: (dict, tuple, dict, bool) -> str
|
|
"""Process input data to ingress
|
|
:param dict config: configuration dict
|
|
:param tuple bxfile: blobxfer script
|
|
:param dict spec: config spec with input_data
|
|
:param bool on_task: if this is originating from a task spec
|
|
:rtype: str
|
|
:return: additonal command
|
|
"""
|
|
tfmimage = 'alfpark/batch-shipyard:{}-cargo'.format(__version__)
|
|
is_windows = settings.is_windows_pool(config)
|
|
if is_windows:
|
|
bxcmd = ('powershell -ExecutionPolicy Unrestricted -command '
|
|
'%AZ_BATCH_NODE_STARTUP_DIR%\\wd\\{} {{}}').format(bxfile[0])
|
|
tfmimage = '{}-windows'.format(tfmimage)
|
|
tfmbind = (
|
|
'-v %AZ_BATCH_NODE_ROOT_DIR%:%AZ_BATCH_NODE_ROOT_DIR% '
|
|
'-w %AZ_BATCH_TASK_WORKING_DIR% '
|
|
'-e "AZ_BATCH_NODE_STARTUP_DIR='
|
|
'%AZ_BATCH_NODE_STARTUP_DIR%" '
|
|
)
|
|
tfmcmd = 'C:\\batch-shipyard\\task_file_mover.cmd'
|
|
tfmpre = ''
|
|
tfmpost = ''
|
|
else:
|
|
bxcmd = 'set -f; $AZ_BATCH_NODE_STARTUP_DIR/wd/{} {{}}; set +f'.format(
|
|
bxfile[0])
|
|
tfmbind = (
|
|
'-v $AZ_BATCH_NODE_ROOT_DIR:$AZ_BATCH_NODE_ROOT_DIR '
|
|
'-w $AZ_BATCH_TASK_WORKING_DIR '
|
|
'-e "AZ_BATCH_NODE_STARTUP_DIR='
|
|
'$AZ_BATCH_NODE_STARTUP_DIR" '
|
|
)
|
|
tfmcmd = '/opt/batch-shipyard/task_file_mover.sh'
|
|
tfmpre = 'set -f; '
|
|
tfmpost = '; set +f'
|
|
ret = []
|
|
input_data = settings.input_data(spec)
|
|
if util.is_not_empty(input_data):
|
|
for key in input_data:
|
|
if key == 'azure_storage':
|
|
args = _process_storage_input_data(
|
|
config, input_data[key], on_task)
|
|
if is_windows:
|
|
cmds = []
|
|
for arg in args:
|
|
cmds.append('""{}""'.format(arg))
|
|
args = cmds
|
|
ret.append(bxcmd.format(' '.join(args)))
|
|
elif key == 'azure_batch':
|
|
args = _process_batch_input_data(
|
|
config, input_data[key], on_task)
|
|
if is_windows:
|
|
cmds = []
|
|
for arg in args:
|
|
cmds.append('""{}""'.format(arg))
|
|
args = cmds
|
|
ret.append(
|
|
('{tfmpre}docker run --rm -t {tfmbind} {tfmimage} '
|
|
'{tfmcmd} {args}{tfmpost}').format(
|
|
tfmpre=tfmpre, tfmbind=tfmbind, tfmimage=tfmimage,
|
|
tfmcmd=tfmcmd, tfmpost=tfmpost,
|
|
args=' '.join(args))
|
|
)
|
|
else:
|
|
raise ValueError(
|
|
'unknown input_data method: {}'.format(key))
|
|
if len(ret) > 0:
|
|
return ';'.join(ret)
|
|
else:
|
|
return None
|
|
|
|
|
|
def _process_storage_output_data(config, native, is_windows, output_data):
|
|
# type: (dict, bool, bool, dict) -> str
|
|
"""Process output data to egress to Azure storage
|
|
:param dict config: configuration dict
|
|
:param bool native: is native container pool
|
|
:param bool is_windows: is windows pool
|
|
:param dict output_data: config spec with output_data
|
|
:rtype: list
|
|
:return: OutputFiles or args to pass to blobxfer script
|
|
"""
|
|
# get gluster host/container paths and encryption settings
|
|
gluster_host, gluster_container = _get_gluster_paths(config)
|
|
encrypt = settings.batch_shipyard_encryption_enabled(config)
|
|
# parse storage output data blocks
|
|
args = []
|
|
for xfer in output_data:
|
|
storage_settings = settings.credentials_storage(
|
|
config, settings.data_storage_account_settings(xfer))
|
|
remote_path = settings.data_remote_path(xfer)
|
|
# derive container from remote_path
|
|
container = settings.data_container_from_remote_path(
|
|
xfer, rp=remote_path)
|
|
eo = settings.data_blobxfer_extra_options(xfer)
|
|
if native and util.is_not_empty(eo):
|
|
raise ValueError(
|
|
'native container pool does not support '
|
|
'blobxfer_extra_options')
|
|
# append appropriate option for fshare
|
|
if settings.data_is_file_share(xfer) and '--mode file' not in eo:
|
|
eo = '--mode file {}'.format(eo)
|
|
if '--mode file' in eo:
|
|
if native:
|
|
raise ValueError(
|
|
'native container pool does not support fileshares')
|
|
# create saskey for file share with rwdl perm
|
|
saskey = storage.create_file_share_saskey(
|
|
storage_settings, container, 'egress', create_share=True)
|
|
else:
|
|
# create saskey for container with rwdl perm
|
|
saskey = storage.create_blob_container_saskey(
|
|
storage_settings, container, 'egress', create_container=True)
|
|
includes = settings.data_include(xfer)
|
|
excludes = settings.data_exclude(xfer)
|
|
# convert include/excludes into extra options
|
|
filters = _convert_filter_to_blobxfer_option(includes, excludes)
|
|
local_path = settings.data_local_path(xfer, True, task_wd=False)
|
|
# auto replace container path for gluster with host path
|
|
if (util.is_not_empty(gluster_container) and
|
|
local_path.startswith(gluster_container)):
|
|
local_path = local_path.replace(gluster_container, gluster_host, 1)
|
|
if native:
|
|
if util.is_not_empty(excludes):
|
|
raise ValueError(
|
|
'native container pool does not support excludes')
|
|
if is_windows:
|
|
sep = '\\'
|
|
else:
|
|
sep = '/'
|
|
if util.is_none_or_empty(includes):
|
|
include = '**{}*'.format(sep)
|
|
if not local_path.endswith(sep):
|
|
fp = sep.join((local_path, include))
|
|
else:
|
|
fp = ''.join((local_path, include))
|
|
of = batchmodels.OutputFile(
|
|
file_pattern=fp,
|
|
destination=batchmodels.OutputFileDestination(
|
|
container=batchmodels.OutputFileBlobContainerDestination(
|
|
path='',
|
|
container_url='{}?{}'.format(
|
|
storage.generate_blob_container_uri(
|
|
storage_settings, container),
|
|
saskey)
|
|
)
|
|
),
|
|
upload_options=batchmodels.OutputFileUploadOptions(
|
|
upload_condition=batchmodels.
|
|
OutputFileUploadCondition.task_success
|
|
),
|
|
)
|
|
args.append(of)
|
|
else:
|
|
# construct argument
|
|
# kind:encrypted:<sa:ep:saskey:remote_path>:local_path:eo
|
|
creds = crypto.encrypt_string(
|
|
encrypt,
|
|
'{},{},{},{}'.format(
|
|
storage_settings.account, storage_settings.endpoint,
|
|
saskey, remote_path),
|
|
config)
|
|
args.append('"{bxver},e,{enc},{creds},{lp},{eo}"'.format(
|
|
bxver=_BLOBXFER_VERSION,
|
|
enc=encrypt,
|
|
creds=creds,
|
|
lp=local_path,
|
|
eo=' '.join((filters, eo)).lstrip(),
|
|
))
|
|
return args
|
|
|
|
|
|
def process_output_data(config, bxfile, spec):
|
|
# type: (dict, tuple, dict) -> str
|
|
"""Process output data to egress
|
|
:param dict config: configuration dict
|
|
:param tuple bxfile: blobxfer script
|
|
:param dict spec: config spec with input_data
|
|
:rtype: str or list
|
|
:return: additonal commands or list of OutputFiles
|
|
"""
|
|
native = settings.is_native_docker_pool(config)
|
|
is_windows = settings.is_windows_pool(config)
|
|
if is_windows:
|
|
bxcmd = ('powershell -ExecutionPolicy Unrestricted -command '
|
|
'%AZ_BATCH_NODE_STARTUP_DIR%\\wd\\{} {{}}').format(bxfile[0])
|
|
else:
|
|
bxcmd = 'set -f; $AZ_BATCH_NODE_STARTUP_DIR/wd/{} {{}}; set +f'.format(
|
|
bxfile[0])
|
|
ret = []
|
|
output_data = settings.output_data(spec)
|
|
if util.is_not_empty(output_data):
|
|
for key in output_data:
|
|
if key == 'azure_storage':
|
|
args = _process_storage_output_data(
|
|
config, native, is_windows, output_data[key])
|
|
if native:
|
|
ret.extend(args)
|
|
else:
|
|
if is_windows:
|
|
cmds = []
|
|
for arg in args:
|
|
cmds.append('""{}""'.format(arg))
|
|
args = cmds
|
|
ret.append(bxcmd.format(' '.join(args)))
|
|
else:
|
|
raise ValueError(
|
|
'unknown output_data method: {}'.format(key))
|
|
if len(ret) > 0:
|
|
if native:
|
|
return ret
|
|
else:
|
|
return ';'.join(ret)
|
|
else:
|
|
return None
|
|
|
|
|
|
def _singlenode_transfer(dest, src, dst, username, ssh_private_key, rls):
|
|
# type: (DestinationSettings, str, str, pathlib.Path, dict) -> None
|
|
"""Transfer data to a single node
|
|
:param DestinationSettings dest: destination settings
|
|
:param str src: source path
|
|
:param str dst: destination path
|
|
:param str username: username
|
|
:param pathlib.Path: ssh private key
|
|
:param dict rls: remote login settings
|
|
"""
|
|
# get remote settings
|
|
_rls = next(iter(rls.values()))
|
|
ip = _rls.remote_login_ip_address
|
|
port = _rls.remote_login_port
|
|
del _rls
|
|
# modify dst with relative dest
|
|
if util.is_not_empty(dest.relative_destination_path):
|
|
dst = '{}{}'.format(dst, dest.relative_destination_path)
|
|
# create relative path on host
|
|
logger.debug('creating remote directory: {}'.format(dst))
|
|
dirs = ['mkdir -p {}'.format(dst)]
|
|
mkdircmd = ('ssh -T -x -o StrictHostKeyChecking=no '
|
|
'-o UserKnownHostsFile={} -i {} -p {} {}@{} {}'.format(
|
|
os.devnull, ssh_private_key, port, username, ip,
|
|
util.wrap_commands_in_shell(dirs)))
|
|
rc = util.subprocess_with_output(
|
|
mkdircmd, shell=True, suppress_output=True)
|
|
if rc == 0:
|
|
logger.info('remote directories created on {}'.format(dst))
|
|
else:
|
|
logger.error('remote directory creation failed')
|
|
return
|
|
del dirs
|
|
# determine if recursive flag must be set
|
|
psrc = pathlib.Path(src)
|
|
recursive = '-r' if psrc.is_dir() else ''
|
|
# set command source path and adjust dst path
|
|
if recursive:
|
|
cmdsrc = '.'
|
|
else:
|
|
cmdsrc = shellquote(src)
|
|
# transfer data
|
|
if dest.data_transfer.method == 'scp':
|
|
cmd = ('scp -o StrictHostKeyChecking=no '
|
|
'-o UserKnownHostsFile={} -p {} {} -i {} '
|
|
'-P {} {} {}@{}:"{}"'.format(
|
|
os.devnull, dest.data_transfer.scp_ssh_extra_options,
|
|
recursive, ssh_private_key.resolve(), port, cmdsrc,
|
|
username, ip, shellquote(dst)))
|
|
elif dest.data_transfer.method == 'rsync+ssh':
|
|
cmd = ('rsync {} {} -e "ssh -T -x -o StrictHostKeyChecking=no '
|
|
'-o UserKnownHostsFile={} {} -i {} -p {}" {} {}@{}:"{}"'.format(
|
|
dest.data_transfer.rsync_extra_options, recursive,
|
|
os.devnull, dest.data_transfer.scp_ssh_extra_options,
|
|
ssh_private_key.resolve(), port, cmdsrc, username, ip,
|
|
shellquote(dst)))
|
|
else:
|
|
raise ValueError('Unknown transfer method: {}'.format(
|
|
dest.data_transfer.method))
|
|
logger.info('begin ingressing data from {} to {}'.format(
|
|
src, dst))
|
|
start = datetime.datetime.now()
|
|
rc = util.subprocess_with_output(
|
|
cmd, shell=True, cwd=src if recursive else None)
|
|
diff = datetime.datetime.now() - start
|
|
if rc == 0:
|
|
logger.info(
|
|
'finished ingressing data from {0} to {1} in {2:.2f} sec'.format(
|
|
src, dst, diff.total_seconds()))
|
|
else:
|
|
logger.error(
|
|
'data ingress from {} to {} failed with return code: {}'.format(
|
|
src, dst, rc))
|
|
|
|
|
|
def _multinode_transfer(
|
|
method, dest, source, dst, username, ssh_private_key, rls, mpt):
|
|
# type: (str, DestinationSettings, SourceSettings, str, str,
|
|
# pathlib.Path, dict, int) -> None
|
|
"""Transfer data to multiple destination nodes simultaneously
|
|
:param str method: transfer method
|
|
:param DestinationSettings dest: destination settings
|
|
:param SourceSettings source: source settings
|
|
:param str dst: destination path
|
|
:param str username: username
|
|
:param pathlib.Path: ssh private key
|
|
:param dict rls: remote login settings
|
|
:param int mpt: max parallel transfers per node
|
|
"""
|
|
src = source.path
|
|
src_incl = source.include
|
|
src_excl = source.exclude
|
|
psrc = pathlib.Path(src)
|
|
# if source isn't a directory, convert it using src_incl
|
|
if not psrc.is_dir():
|
|
src_excl = None
|
|
src_incl = [src]
|
|
src = str(psrc.parent)
|
|
psrc = psrc.parent
|
|
# if split is specified, force to multinode_scp
|
|
if (dest.data_transfer.split_files_megabytes is not None and
|
|
method != 'multinode_scp'):
|
|
logger.warning('forcing transfer method to multinode_scp with split')
|
|
method = 'multinode_scp'
|
|
buckets = {}
|
|
files = {}
|
|
rcodes = {}
|
|
spfiles = []
|
|
spfiles_count = {}
|
|
spfiles_count_lock = threading.Lock()
|
|
for rkey in rls:
|
|
buckets[rkey] = 0
|
|
files[rkey] = []
|
|
rcodes[rkey] = None
|
|
# walk the directory structure
|
|
# 1. construct a set of dirs to create on the remote side
|
|
# 2. binpack files to different nodes
|
|
total_files = 0
|
|
dirs = set()
|
|
if dest.relative_destination_path is not None:
|
|
dirs.add(dest.relative_destination_path)
|
|
for entry in util.scantree(src):
|
|
rel = pathlib.Path(entry.path).relative_to(psrc)
|
|
sparent = str(pathlib.Path(entry.path).relative_to(psrc).parent)
|
|
if entry.is_file():
|
|
srel = str(rel)
|
|
# check filters
|
|
if src_excl is not None:
|
|
inc = not any([fnmatch.fnmatch(srel, x) for x in src_excl])
|
|
else:
|
|
inc = True
|
|
if src_incl is not None:
|
|
inc = any([fnmatch.fnmatch(srel, x) for x in src_incl])
|
|
if not inc:
|
|
logger.debug('skipping file {} due to filters'.format(
|
|
entry.path))
|
|
continue
|
|
if dest.relative_destination_path is None:
|
|
dstpath = '{}{}'.format(dst, rel)
|
|
else:
|
|
dstpath = '{}{}/{}'.format(
|
|
dst, dest.relative_destination_path, rel)
|
|
# get key of min bucket values
|
|
fsize = entry.stat().st_size
|
|
if (dest.data_transfer.split_files_megabytes is not None and
|
|
fsize > dest.data_transfer.split_files_megabytes):
|
|
nsplits = int(math.ceil(
|
|
fsize / dest.data_transfer.split_files_megabytes))
|
|
lpad = int(math.log10(nsplits)) + 1
|
|
spfiles.append(dstpath)
|
|
spfiles_count[dstpath] = nsplits
|
|
n = 0
|
|
curr = 0
|
|
while True:
|
|
end = curr + dest.data_transfer.split_files_megabytes
|
|
if end > fsize:
|
|
end = fsize
|
|
key = min(buckets, key=buckets.get)
|
|
buckets[key] += (end - curr)
|
|
if n == 0:
|
|
dstfname = dstpath
|
|
else:
|
|
dstfname = '{}.{}{}'.format(
|
|
dstpath, _FILE_SPLIT_PREFIX, str(n).zfill(lpad))
|
|
files[key].append((entry.path, dstfname, curr, end))
|
|
if end == fsize:
|
|
break
|
|
curr = end
|
|
n += 1
|
|
else:
|
|
key = min(buckets, key=buckets.get)
|
|
buckets[key] += fsize
|
|
files[key].append((entry.path, dstpath, None, None))
|
|
total_files += 1
|
|
# add directory to create
|
|
if sparent != '.':
|
|
if dest.relative_destination_path is None:
|
|
dirs.add(sparent)
|
|
else:
|
|
dirs.add('{}/{}'.format(
|
|
dest.relative_destination_path, sparent))
|
|
total_size = sum(buckets.values())
|
|
if total_files == 0:
|
|
logger.error('no files to ingress')
|
|
return
|
|
# create remote directories via ssh
|
|
if len(dirs) == 0:
|
|
logger.debug('no remote directories to create')
|
|
else:
|
|
logger.debug('creating remote directories: {}'.format(dirs))
|
|
dirs = ['mkdir -p {}'.format(x) for x in list(dirs)]
|
|
dirs.insert(0, 'cd {}'.format(dst))
|
|
_rls = next(iter(rls.values()))
|
|
ip = _rls.remote_login_ip_address
|
|
port = _rls.remote_login_port
|
|
del _rls
|
|
mkdircmd = ('ssh -T -x -o StrictHostKeyChecking=no '
|
|
'-o UserKnownHostsFile={} -i {} -p {} {}@{} {}'.format(
|
|
os.devnull, ssh_private_key, port, username, ip,
|
|
util.wrap_commands_in_shell(dirs)))
|
|
rc = util.subprocess_with_output(
|
|
mkdircmd, shell=True, suppress_output=True)
|
|
if rc == 0:
|
|
logger.info('remote directories created on {}'.format(dst))
|
|
else:
|
|
logger.error('remote directory creation failed')
|
|
return
|
|
del ip
|
|
del port
|
|
logger.info(
|
|
'ingress data: {0:.4f} MiB in {1} files to transfer, using {2} max '
|
|
'parallel transfers per node'.format(
|
|
total_size / _MEGABYTE, total_files, mpt))
|
|
logger.info('begin ingressing data from {} to {}'.format(src, dst))
|
|
nodekeys = list(buckets.keys())
|
|
threads = []
|
|
start = datetime.datetime.now()
|
|
for i in range(0, len(buckets)):
|
|
nkey = nodekeys[i]
|
|
thr = threading.Thread(
|
|
target=_multinode_thread_worker,
|
|
args=(method, mpt, nkey, rcodes, files[nkey],
|
|
spfiles_count, spfiles_count_lock,
|
|
rls[nkey].remote_login_ip_address,
|
|
rls[nkey].remote_login_port, username, ssh_private_key,
|
|
dest.data_transfer.scp_ssh_extra_options,
|
|
dest.data_transfer.rsync_extra_options)
|
|
)
|
|
threads.append(thr)
|
|
thr.start()
|
|
for i in range(0, len(buckets)):
|
|
threads[i].join()
|
|
diff = datetime.datetime.now() - start
|
|
del threads
|
|
success = True
|
|
for nkey in rcodes:
|
|
if rcodes[nkey] != 0:
|
|
logger.error('data ingress failed to node: {}'.format(nkey))
|
|
success = False
|
|
if success:
|
|
logger.info(
|
|
'finished ingressing {0:.4f} MB of data in {1} files from {2} to '
|
|
'{3} in {4:.2f} sec ({5:.3f} Mbit/s)'.format(
|
|
total_size / _MEGABYTE, total_files, src, dst,
|
|
diff.total_seconds(),
|
|
(total_size * 8 / 1e6) / diff.total_seconds()))
|
|
|
|
|
|
def _spawn_next_transfer(
|
|
method, file, ip, port, username, ssh_private_key, eo, reo,
|
|
procs, psprocs, psdst):
|
|
# type: (str, tuple, str, int, str, pathlib.Path, str, str, list,
|
|
# list, list) -> None
|
|
"""Spawn the next transfer given a file tuple
|
|
:param str method: transfer method
|
|
:param tuple file: file tuple
|
|
:param str ip: ip address
|
|
:param int port: port
|
|
:param str username: username
|
|
:param pathlib.Path: ssh private key
|
|
:param str eo: extra options
|
|
:param str reo: rsync extra options
|
|
:param list procs: process list
|
|
:param list psprocs: split files process list
|
|
:param list psdst: split files dstpath list
|
|
"""
|
|
src = file[0]
|
|
dst = file[1]
|
|
begin = file[2]
|
|
end = file[3]
|
|
if method == 'multinode_scp':
|
|
if begin is None and end is None:
|
|
cmd = ('scp -o StrictHostKeyChecking=no '
|
|
'-o UserKnownHostsFile={} -p {} -i {} '
|
|
'-P {} {} {}@{}:"{}"'.format(
|
|
os.devnull, eo, ssh_private_key.resolve(), port,
|
|
shellquote(src), username, ip, shellquote(dst)))
|
|
else:
|
|
cmd = ('ssh -T -x -o StrictHostKeyChecking=no '
|
|
'-o UserKnownHostsFile={} {} -i {} '
|
|
'-p {} {}@{} \'cat > "{}"\''.format(
|
|
os.devnull, eo, ssh_private_key.resolve(), port,
|
|
username, ip, shellquote(dst)))
|
|
elif method == 'multinode_rsync+ssh':
|
|
if begin is not None or end is not None:
|
|
raise RuntimeError('cannot rsync with file offsets')
|
|
cmd = ('rsync {} -e "ssh -T -x -o StrictHostKeyChecking=no '
|
|
'-o UserKnownHostsFile={} {} -i {} -p {}" {} {}@{}:"{}"'.format(
|
|
reo, os.devnull, eo, ssh_private_key.resolve(), port,
|
|
shellquote(src), username, ip, shellquote(dst)))
|
|
else:
|
|
raise ValueError('Unknown transfer method: {}'.format(method))
|
|
if begin is None and end is None:
|
|
procs.append(util.subprocess_nowait(cmd, shell=True))
|
|
else:
|
|
proc = util.subprocess_attach_stdin(cmd, shell=True)
|
|
with open(src, 'rb') as f:
|
|
f.seek(begin, 0)
|
|
curr = begin
|
|
while curr < end:
|
|
buf = f.read(_MAX_READ_BLOCKSIZE_BYTES)
|
|
if buf is None or len(buf) == 0:
|
|
break
|
|
proc.stdin.write(buf)
|
|
curr += len(buf)
|
|
proc.stdin.close()
|
|
psprocs.append(proc)
|
|
dstsp = dst.split('.')
|
|
if dstsp[-1].startswith(_FILE_SPLIT_PREFIX):
|
|
dstpath = '.'.join(dstsp[:-1])
|
|
else:
|
|
dstpath = dst
|
|
psdst.append(dstpath)
|
|
|
|
|
|
def _multinode_thread_worker(
|
|
method, mpt, node_id, rcodes, files, spfiles_count,
|
|
spfiles_count_lock, ip, port, username, ssh_private_key, eo, reo):
|
|
# type: (str, int, str, dict, list, dict, threading.Lock, str, int, str,
|
|
# pathlib.Path, str, str) -> None
|
|
"""Worker thread code for data transfer to a node with a file list
|
|
:param str method: transfer method
|
|
:param int mpt: max parallel transfers per node
|
|
:param str node_id: node id
|
|
:param dict rcodes: return codes dict
|
|
:param list files: list of files to copy
|
|
:param dict spfiles_count: split files count dict
|
|
:param threading.Lock spfiles_count_lock: split files count lock
|
|
:param str ip: ip address
|
|
:param int port: port
|
|
:param str username: username
|
|
:param pathlib.Path: ssh private key
|
|
:param str eo: extra options
|
|
:param str reo: rsync extra options
|
|
"""
|
|
procs = []
|
|
psprocs = []
|
|
psdst = []
|
|
completed = 0
|
|
i = 0
|
|
while completed != len(files):
|
|
xfers = len(procs) + len(psprocs)
|
|
while xfers < mpt and i < len(files):
|
|
file = files[i]
|
|
_spawn_next_transfer(
|
|
method, file, ip, port, username, ssh_private_key, eo, reo,
|
|
procs, psprocs, psdst)
|
|
xfers = len(procs) + len(psprocs)
|
|
i += 1
|
|
plist, n, rc = util.subprocess_wait_multi(psprocs, procs)
|
|
if rc != 0:
|
|
logger.error(
|
|
'data ingress to {} failed with return code: {}'.format(
|
|
node_id, rc))
|
|
rcodes[node_id] = rc
|
|
return
|
|
if plist == psprocs:
|
|
dstpath = psdst[n]
|
|
del psdst[n]
|
|
del psprocs[n]
|
|
join = False
|
|
with spfiles_count_lock:
|
|
spfiles_count[dstpath] = spfiles_count[dstpath] - 1
|
|
if spfiles_count[dstpath] == 0:
|
|
join = True
|
|
if join:
|
|
logger.debug('joining files on compute node to {}'.format(
|
|
dstpath))
|
|
cmds = [
|
|
'cat {}.* >> {}'.format(dstpath, dstpath),
|
|
'rm -f {}.*'.format(dstpath)
|
|
]
|
|
joincmd = ('ssh -T -x -o StrictHostKeyChecking=no '
|
|
'-o UserKnownHostsFile={} -i {} '
|
|
'-p {} {}@{} {}'.format(
|
|
os.devnull, ssh_private_key, port, username,
|
|
ip, util.wrap_commands_in_shell(cmds)))
|
|
procs.append(
|
|
util.subprocess_nowait(joincmd, shell=True))
|
|
else:
|
|
completed += 1
|
|
else:
|
|
del procs[n]
|
|
completed += 1
|
|
rcodes[node_id] = 0
|
|
|
|
|
|
def _azure_blob_storage_transfer(storage_settings, data_transfer, source):
|
|
# type: (settings.StorageCredentialsSettings,
|
|
# settings.DataTransferSettings,
|
|
# settings.SourceSettings) -> None
|
|
"""Initiate an azure blob storage transfer
|
|
:param settings.StorageCredentialsSettings storage_settings:
|
|
storage settings
|
|
:param settings.DataTransferSettings data_transfer: data transfer settings
|
|
:param settings.SourceSettings source: source settings
|
|
"""
|
|
eo = data_transfer.blobxfer_extra_options
|
|
# append appropriate option for fshare
|
|
if data_transfer.is_file_share and '--mode file' not in eo:
|
|
eo = '--mode file {}'.format(eo)
|
|
thr = threading.Thread(
|
|
target=_wrap_blobxfer_subprocess,
|
|
args=(
|
|
storage_settings,
|
|
data_transfer.remote_path,
|
|
source,
|
|
eo,
|
|
)
|
|
)
|
|
thr.start()
|
|
return thr
|
|
|
|
|
|
def _wrap_blobxfer_subprocess(storage_settings, remote_path, source, eo):
|
|
# type: (StorageCredentialsSettings, str, SourceSettings, str) -> None
|
|
"""Wrapper function for blobxfer
|
|
:param StorageCredentialsSettings storage_settings: storage settings
|
|
:param str remote_path: remote path to transfer to
|
|
:param SourceSettings source: source settings
|
|
:param str eo: blobxfer extra options
|
|
"""
|
|
# generate include/exclude options
|
|
filters = _convert_filter_to_blobxfer_option(
|
|
source.include, source.exclude)
|
|
# get correct path
|
|
psrc = pathlib.Path(source.path)
|
|
cwd = str(psrc.parent)
|
|
rsrc = psrc.relative_to(psrc.parent)
|
|
# generate env
|
|
env = os.environ.copy()
|
|
env['BLOBXFER_STORAGE_ACCOUNT_KEY'] = storage_settings.account_key
|
|
# set cmd
|
|
cmd = [
|
|
('blobxfer upload --storage-account {sa} --remote-path {rp} '
|
|
'--local-path {lp} --endpoint {ep} --no-progress-bar '
|
|
'{filters} {eo}').format(
|
|
sa=storage_settings.account,
|
|
rp=remote_path,
|
|
lp=rsrc,
|
|
ep=storage_settings.endpoint,
|
|
filters=filters,
|
|
eo=eo)
|
|
]
|
|
logger.info('begin ingressing data from {} to remote path {}'.format(
|
|
source.path, remote_path))
|
|
proc = util.subprocess_nowait_pipe_stdout(
|
|
util.wrap_local_commands_in_shell(cmd), shell=True, cwd=cwd, env=env)
|
|
stdout, stderr = proc.communicate()
|
|
if proc.returncode != 0:
|
|
if stderr is not None:
|
|
logger.error(stderr.decode('utf8'))
|
|
if stdout is not None:
|
|
logger.error(stdout.decode('utf8'))
|
|
logger.error('data ingress failed from {} to remote path {}'.format(
|
|
source.path, remote_path))
|
|
else:
|
|
if stdout is not None:
|
|
logger.debug(stdout.decode('utf8'))
|
|
|
|
|
|
def wait_for_storage_threads(storage_threads):
|
|
# type: (list) -> None
|
|
"""Wait for storage processes to complete
|
|
:param list storage_threads: list of storage threads
|
|
"""
|
|
if storage_threads is None:
|
|
return
|
|
i = 0
|
|
nthreads = len(storage_threads)
|
|
while nthreads > 0:
|
|
alive = sum(thr.is_alive() for thr in storage_threads)
|
|
if alive > 0:
|
|
i += 1
|
|
if i % 10 == 0:
|
|
i = 0
|
|
logger.debug(
|
|
'waiting for Azure Blob Storage transfer processes '
|
|
'to complete: {} active, {} completed'.format(
|
|
alive, nthreads - alive))
|
|
time.sleep(1)
|
|
else:
|
|
for thr in storage_threads:
|
|
thr.join()
|
|
if nthreads > 0:
|
|
logger.info('Azure Blob/File Storage transfer completed')
|
|
break
|
|
|
|
|
|
def ingress_data(
|
|
batch_client, compute_client, network_client, config, rls=None,
|
|
kind=None, total_vm_count=None, to_fs=None):
|
|
# type: (batch.BatchServiceClient,
|
|
# azure.mgmt.compute.ComputeManagementClient, dict, dict, str,
|
|
# int, str) -> list
|
|
"""Ingresses data into Azure
|
|
:param batch_client: The batch client to use.
|
|
:type batch_client: `batchserviceclient.BatchServiceClient`
|
|
:param azure.mgmt.compute.ComputeManagementClient compute_client:
|
|
compute client
|
|
:param azure.mgmt.network.NetworkManagementClient network_client:
|
|
network client
|
|
:param dict config: configuration dict
|
|
:param dict rls: remote login settings
|
|
:param str kind: 'all', 'shared', 'storage', or 'remotefs'
|
|
:param int total_vm_count: total current vm count
|
|
:param str to_fs: to remote filesystem
|
|
:rtype: list
|
|
:return: list of storage threads
|
|
"""
|
|
storage_threads = []
|
|
files = settings.global_resources_files(config)
|
|
if util.is_none_or_empty(files):
|
|
logger.info('no files to ingress detected')
|
|
return storage_threads
|
|
pool = settings.pool_settings(config)
|
|
is_windows = settings.is_windows_pool(config)
|
|
for fdict in files:
|
|
source = settings.files_source_settings(fdict)
|
|
dest = settings.files_destination_settings(fdict)
|
|
if (dest.shared_data_volume is not None and
|
|
dest.storage_account_settings is not None):
|
|
raise RuntimeError(
|
|
'cannot specify both shared data volume and storage for the '
|
|
'destination for source: {}'.format(source.path))
|
|
direct_single_node = False
|
|
if dest.relative_destination_path is not None:
|
|
if dest.storage_account_settings is not None:
|
|
raise RuntimeError(
|
|
'cannot specify a relative destination path for ingress '
|
|
'to storage; use the --collate option in blobxfer '
|
|
'instead.')
|
|
# check if this is going to a single vm
|
|
if dest.shared_data_volume is None:
|
|
if total_vm_count == 1:
|
|
direct_single_node = True
|
|
elif kind == 'storage':
|
|
# this is to prevent total_vm_count check below for
|
|
# non shared/all targets and will force continuation
|
|
# of the loop below
|
|
direct_single_node = True
|
|
elif total_vm_count is None:
|
|
raise ValueError('total_vm_count is not set')
|
|
else:
|
|
raise RuntimeError(
|
|
'Cannot ingress data directly into compute node '
|
|
'host for pools with more than one node. Please use '
|
|
'a shared data volume as the ingress destination '
|
|
'instead.')
|
|
if dest.shared_data_volume is not None or direct_single_node:
|
|
if kind == 'storage':
|
|
logger.warning(
|
|
'skipping data ingress from {} to {} for pool as ingress '
|
|
'to shared file system not specified'.format(
|
|
source.path, dest.shared_data_volume))
|
|
continue
|
|
if is_windows:
|
|
logger.error(
|
|
('cannot data ingress from {} to pool {} with windows '
|
|
'compute nodes').format(source.path, pool.id))
|
|
continue
|
|
# get rfs settings
|
|
rfs = None
|
|
dst_rfs = False
|
|
# set base dst path
|
|
dst = '{}/batch/tasks/mounts'.format(
|
|
settings.temp_disk_mountpoint(config))
|
|
# convert shared to actual path
|
|
if not direct_single_node:
|
|
sdv = settings.global_resources_shared_data_volumes(config)
|
|
for sdvkey in sdv:
|
|
if sdvkey == dest.shared_data_volume:
|
|
if settings.is_shared_data_volume_gluster_on_compute(
|
|
sdv, sdvkey):
|
|
if kind == 'remotefs':
|
|
continue
|
|
dst = '{}/{}/'.format(
|
|
dst, settings.get_gluster_on_compute_volume())
|
|
elif settings.is_shared_data_volume_storage_cluster(
|
|
sdv, sdvkey):
|
|
if kind != 'remotefs' or sdvkey != to_fs:
|
|
continue
|
|
if rfs is None:
|
|
rfs = settings.remotefs_settings(config, to_fs)
|
|
dst = rfs.storage_cluster.file_server.mountpoint
|
|
# add trailing directory separator if needed
|
|
if dst[-1] != '/':
|
|
dst = dst + '/'
|
|
dst_rfs = True
|
|
else:
|
|
raise RuntimeError(
|
|
'data ingress to {} not supported'.format(
|
|
sdvkey))
|
|
break
|
|
# skip entries that are a mismatch if remotefs transfer
|
|
# is selected
|
|
if kind == 'remotefs':
|
|
if not dst_rfs:
|
|
continue
|
|
else:
|
|
if dst_rfs:
|
|
continue
|
|
# set ssh info
|
|
if dst_rfs:
|
|
username = rfs.storage_cluster.ssh.username
|
|
# retrieve public ips from all vms in named storage cluster
|
|
rls = {}
|
|
for i in range(rfs.storage_cluster.vm_count):
|
|
vm_name = '{}-vm{}'.format(
|
|
rfs.storage_cluster.hostname_prefix, i)
|
|
vm = compute_client.virtual_machines.get(
|
|
resource_group_name=rfs.storage_cluster.resource_group,
|
|
vm_name=vm_name,
|
|
)
|
|
_, pip = resource.get_nic_and_pip_from_virtual_machine(
|
|
network_client, rfs.storage_cluster.resource_group, vm)
|
|
# create compute node rls settings with sc vm ip/port
|
|
rls[vm_name] = \
|
|
batchmodels.ComputeNodeGetRemoteLoginSettingsResult(
|
|
remote_login_ip_address=pip.ip_address,
|
|
remote_login_port=22)
|
|
else:
|
|
username = pool.ssh.username
|
|
if rls is None:
|
|
logger.warning(
|
|
'skipping data ingress from {} to {} for pool with no '
|
|
'remote login settings or non-existent pool'.format(
|
|
source.path, dest.shared_data_volume))
|
|
continue
|
|
if username is None:
|
|
raise RuntimeError(
|
|
'cannot ingress data to shared data volume without a '
|
|
'valid SSH user')
|
|
# try to get valid ssh private key (from various config blocks)
|
|
ssh_private_key = dest.data_transfer.ssh_private_key
|
|
if ssh_private_key is None:
|
|
ssh_private_key = pool.ssh.ssh_private_key
|
|
if ssh_private_key is None:
|
|
ssh_private_key = pathlib.Path(crypto.get_ssh_key_prefix())
|
|
if not ssh_private_key.exists():
|
|
raise RuntimeError(
|
|
'specified SSH private key is invalid or does not '
|
|
'exist')
|
|
logger.debug('using ssh_private_key from: {}'.format(
|
|
ssh_private_key))
|
|
if (dest.data_transfer.method == 'scp' or
|
|
dest.data_transfer.method == 'rsync+ssh'):
|
|
# split/source include/exclude will force multinode
|
|
# transfer with mpt=1
|
|
if (dest.data_transfer.split_files_megabytes is not None or
|
|
source.include is not None or
|
|
source.exclude is not None):
|
|
_multinode_transfer(
|
|
'multinode_' + dest.data_transfer.method, dest,
|
|
source, dst, username, ssh_private_key, rls, 1)
|
|
else:
|
|
_singlenode_transfer(
|
|
dest, source.path, dst, username, ssh_private_key,
|
|
rls)
|
|
elif (dest.data_transfer.method == 'multinode_scp' or
|
|
dest.data_transfer.method == 'multinode_rsync+ssh'):
|
|
_multinode_transfer(
|
|
dest.data_transfer.method, dest, source, dst,
|
|
username, ssh_private_key, rls,
|
|
dest.data_transfer.max_parallel_transfers_per_node)
|
|
else:
|
|
raise RuntimeError(
|
|
'unknown transfer method: {}'.format(
|
|
dest.data_transfer.method))
|
|
elif dest.storage_account_settings is not None:
|
|
if kind == 'shared':
|
|
logger.warning(
|
|
'skipping data ingress from {} for pool as ingress '
|
|
'to Azure Blob/File Storage not specified'.format(
|
|
source.path))
|
|
continue
|
|
thr = _azure_blob_storage_transfer(
|
|
settings.credentials_storage(
|
|
config, dest.storage_account_settings),
|
|
dest.data_transfer, source)
|
|
storage_threads.append(thr)
|
|
else:
|
|
raise RuntimeError(
|
|
'invalid file transfer configuration: {}'.format(fdict))
|
|
return storage_threads
|