diff --git a/.travis.yml b/.travis.yml index 7711e54ff..41163d7b5 100644 --- a/.travis.yml +++ b/.travis.yml @@ -53,13 +53,6 @@ before_install: - which python - which pip - python --version - - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then - docker pull harisekhon/hadoop; - export HADOOP_CONTAINER_ID=$(docker run -dt harisekhon/hadoop); - echo $HADOOP_CONTAINER_ID; - export HADOOP_CONTAINER_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' $HADOOP_CONTAINER_ID); - echo $HADOOP_CONTAINER_IP; - fi install: - bash ./scripts/ci/install.sh script: diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 9290389d1..815143396 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -35,6 +35,12 @@ class DependencyBase(object): def match(cls, url): return re.match(cls.REGEX, url) + def group(self, name): + match = self.match(self.path) + if not match: + return None + return match.group(name) + @classmethod def supported(cls, url): return cls.match(url) != None diff --git a/dvc/dependency/hdfs.py b/dvc/dependency/hdfs.py index 9e0e6e5c8..aab808ff8 100644 --- a/dvc/dependency/hdfs.py +++ b/dvc/dependency/hdfs.py @@ -10,7 +10,7 @@ from dvc.remote.hdfs import RemoteHDFS class DependencyHDFS(DependencyBase): - REGEX = r'^hdfs://.*$' + REGEX = RemoteHDFS.REGEX def __init__(self, stage, path, info=None, remote=None): super(DependencyHDFS, self).__init__(stage, path) @@ -18,7 +18,9 @@ class DependencyHDFS(DependencyBase): self.remote = remote if remote else RemoteHDFS(stage.project, {}) if remote: path = posixpath.join(remote.url, urlparse(path).path.lstrip('/')) + user = remote.user if remote else self.group('user') self.path_info = {'scheme': 'hdfs', + 'user': user, 'url': path} def changed(self): diff --git a/dvc/dependency/ssh.py b/dvc/dependency/ssh.py index 2be6b1d46..4be409082 100644 --- a/dvc/dependency/ssh.py +++ b/dvc/dependency/ssh.py @@ -29,12 +29,6 @@ class DependencySSH(DependencyBase): 'user': user, 'path': path} - def group(self, name): - match = self.match(self.path) - if not match: - return None - return match.group(name) - def changed(self): return self.info != self.remote.save_info(self.path_info) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 151e9f9db..11e6fbd43 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -46,6 +46,12 @@ class RemoteBase(object): def match(cls, url): return re.match(cls.REGEX, url) + def group(self, name): + m = self.match(self.url) + if not m: + return None + return m.group(name) + @staticmethod def tmp_file(fname): """ Temporary name for a partial download """ diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 445297747..6dc98cea6 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -1,5 +1,6 @@ import os import re +import getpass import posixpath from subprocess import Popen, PIPE @@ -15,23 +16,23 @@ from dvc.exceptions import DvcException class RemoteHDFS(RemoteBase): scheme='hdfs' - REGEX = r'^hdfs://(?P.*)$' + REGEX = r'^hdfs://((?P.*)@)?.*$' PARAM_CHECKSUM = 'checksum' def __init__(self, project, config): self.project = project self.url = config.get(Config.SECTION_REMOTE_URL, '/') - self.user = config.get(Config.SECTION_REMOTE_USER, None) + self.user = self.group('user') + if not self.user: + self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser()) - def hadoop_fs(self, cmd): + def hadoop_fs(self, cmd, user=None): cmd = 'hadoop fs -' + cmd - env = os.environ.copy() - if self.user: - env['HADOOP_USER_NAME'] = self.user + if user: + cmd = 'HADOOP_USER_NAME={} '.format(user) + cmd p = Popen(cmd, shell=True, close_fds=True, - env=env, executable=os.getenv('SHELL'), stdin=PIPE, stdout=PIPE, @@ -39,25 +40,25 @@ class RemoteHDFS(RemoteBase): out, err = p.communicate() if p.returncode != 0: raise DvcException('HDFS command failed: {}: {}'.format(cmd, err)) - return out + return out.decode('utf-8') @staticmethod - def group(regex, s, gname): + def _group(regex, s, gname): match = re.match(regex, s) assert match != None return match.group(gname) - def checksum(self, url): + def checksum(self, path_info): regex = r'.*\t.*\t(?P.*)' - stdout = self.hadoop_fs('checksum {}'.format(url)) - return self.group(regex, stdout, 'checksum') + stdout = self.hadoop_fs('checksum {}'.format(path_info['url']), user=path_info['user']) + return self._group(regex, stdout, 'checksum') - def cp(self, from_url, to_url): - self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_url))) - self.hadoop_fs('cp {} {}'.format(from_url, to_url)) + def cp(self, from_info, to_info): + self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])), user=to_info['user']) + self.hadoop_fs('cp {} {}'.format(from_info['url'], to_info['url']), user=to_info['user']) - def rm(self, url): - self.hadoop_fs('rm {}'.format(url)) + def rm(self, path_info): + self.hadoop_fs('rm {}'.format(path_info['url']), user=path_info['user']) def save_info(self, path_info): if path_info['scheme'] != 'hdfs': @@ -65,7 +66,7 @@ class RemoteHDFS(RemoteBase): assert path_info.get('url') - return {self.PARAM_CHECKSUM: self.checksum(path_info['url'])} + return {self.PARAM_CHECKSUM: self.checksum(path_info)} def save(self, path_info): if path_info['scheme'] != 'hdfs': @@ -73,10 +74,11 @@ class RemoteHDFS(RemoteBase): assert path_info.get('url') - checksum = self.checksum(path_info['url']) - dest_url = posixpath.join(self.url, checksum[0:2], checksum[2:]) + checksum = self.checksum(path_info) + dest = path_info.copy() + dest['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) - self.cp(path_info['url'], dest_url) + self.cp(path_info, dest) return {self.PARAM_CHECKSUM: checksum} @@ -90,9 +92,10 @@ class RemoteHDFS(RemoteBase): if not checksum: return - url = posixpath.join(self.url, checksum[0:2], checksum[2:]) + src = path_info.copy() + src['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) - self.cp(url, path_info['url']) + self.cp(src, path_info) def remove(self, path_info): if path_info['scheme'] != 'hdfs': @@ -100,14 +103,14 @@ class RemoteHDFS(RemoteBase): assert path_info.get('url') - self.rm(path_info['url']) + self.rm(path_info) def upload(self, path, path_info): if path_info['scheme'] != 'hdfs': raise NotImplementedError - self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(path_info['url']))) - self.hadoop_fs('copyFromLocal {} {}'.format(path, path_info['url'])) + self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(path_info['url'])), user=path_info['user']) + self.hadoop_fs('copyFromLocal {} {}'.format(path, path_info['url']), user=path_info['user']) def download(self, path_info, path): if path_info['scheme'] != 'hdfs': @@ -117,4 +120,4 @@ class RemoteHDFS(RemoteBase): if not os.path.exists(dname): os.makedirs(dname) - self.hadoop_fs('copyToLocal {} {}'.format(path_info['url'], path)) + self.hadoop_fs('copyToLocal {} {}'.format(path_info['url'], path), user=path_info['user']) diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 77686c86a..6becc191f 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -52,12 +52,6 @@ class RemoteSSH(RemoteBase): self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser()) self.prefix = self.group('path') - def group(self, group): - m = self.match(self.url) - if not m: - return None - return m.group(group) - def cache_file_key(self, path): relpath = os.path.relpath(os.path.abspath(path), self.project.cache.local.cache_dir) return posixpath.join(self.prefix, relpath.replace('\\', '/')) diff --git a/tests/basic_env.py b/tests/basic_env.py index 2445f5afd..c1c90e392 100644 --- a/tests/basic_env.py +++ b/tests/basic_env.py @@ -16,9 +16,9 @@ class TestDir(TestCase): DATA_CONTENTS = DATA DATA_SUB_CONTENTS = DATA_SUB FOO = 'foo' - FOO_CONTENTS = FOO + FOO_CONTENTS = FOO + '\n' BAR = 'bar' - BAR_CONTENTS = BAR + BAR_CONTENTS = BAR + '\n' CODE = 'code.py' CODE_CONTENTS = 'import sys\nimport shutil\nshutil.copyfile(sys.argv[1], sys.argv[2])' diff --git a/tests/log.log b/tests/log.log new file mode 100644 index 000000000..7c3affb79 --- /dev/null +++ b/tests/log.log @@ -0,0 +1 @@ +/home/efiop/virtenvs/python3-dvc/bin/python: No module named tests.__main__; 'tests' is a package and cannot be directly executed diff --git a/tests/test_repro.py b/tests/test_repro.py index f4e15d8a7..6356e6c76 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -293,7 +293,7 @@ class TestReproExternalBase(TestDvc): out_foo_path = 'remote://' + remote_name + '/' + self.FOO out_bar_path = bar_path - self.write(self.bucket, foo_key, 'foo') + self.write(self.bucket, foo_key, self.FOO_CONTENTS) import_stage = self.dvc.imp(out_foo_path, 'import') self.assertTrue(os.path.exists('import')) @@ -303,7 +303,7 @@ class TestReproExternalBase(TestDvc): deps=[out_foo_path], cmd=self.cmd(foo_path, bar_path)) - self.write(self.bucket, foo_key, 'bar') + self.write(self.bucket, foo_key, self.BAR_CONTENTS) stages = self.dvc.reproduce(import_stage.path) self.assertEqual(len(stages), 1) @@ -372,14 +372,14 @@ class TestReproExternalHDFS(TestReproExternalBase): @property def bucket(self): - return os.getenv('HADOOP_CONTAINER_IP') + return 'root@' + os.getenv('HADOOP_CONTAINER_IP') def cmd(self, i, o): return 'HADOOP_USER_NAME=root hadoop fs -cp {} {}'.format(i, o) def write(self, bucket, key, body): url = self.scheme + '://' + bucket + '/' + key - p = Popen('hadoop fs -rm -f {}'.format(url), + p = Popen('HADOOP_USER_NAME=root hadoop fs -rm -f {}'.format(url), shell=True, executable=os.getenv('SHELL'), stdin=PIPE, @@ -387,13 +387,16 @@ class TestReproExternalHDFS(TestReproExternalBase): stderr=PIPE) p.communicate() - p = Popen('echo "{}" | hadoop fs -put - {}'.format(body, url), + p = Popen('echo "{}" | HADOOP_USER_NAME=root hadoop fs -put - {}'.format(body.strip(), url), shell=True, executable=os.getenv('SHELL'), stdin=PIPE, stdout=PIPE, stderr=PIPE) - p.communicate() + out, err = p.communicate() + if p.returncode != 0: + print(out) + print(err) self.assertEqual(p.returncode, 0) @@ -438,7 +441,7 @@ class TestReproExternalSSH(TestReproExternalBase): p.communicate() self.assertEqual(p.returncode, 0) - p = Popen('echo "{}" | ssh {} "tr -d \'\\n\' > {}"'.format(body, dest, path), + p = Popen('echo "{}" | ssh {} "cat > {}"'.format(body.strip(), dest, path), shell=True, executable=os.getenv('SHELL'), stdin=PIPE,