Repository: nchammas/flintrock
Pull request #249 (2018-05-15 02:42:40)

Download Hadoop and Spark from Apache mirrors by default
This addresses the issue described in #238, which is that the Spark project have stopped publishing releases to S3. This means that we now have to download Spark from the Apache mirror network.

While doing this work I also:
* Unified how we download Hadoop and Spark since they are both coming from the same place now.
* Bumped the default version of Hadoop to 2.8.4 and Spark to 2.3.0.
* Added a test for the download script to make sure it works on both Python 2 and 3 (since it runs on the cluster nodes).

TODO:
- [x] Add warnings about problems with Apache mirrors -- slow, unreliable, only latest releases -- and recommend users set custom `download-source`.
- [x] Proactively check for package availability before launching cluster.
- [x] Confirm building Spark from source still works when Hadoop 2.8.4. is configured.
- [ ] ~Add wiki page explaining how to host Hadoop or Spark packages on a public S3 bucket.~

For a future PR:
- [ ] Allow `download-source` to be set to an `s3://` URL, and use the AWS CLI to download in that case. That will let users serve packages from private buckets which the cluster can access via IAM roles.


Patch:
diff --git a/README.md b/README.md
index f17db06..4e53a6b 100644
--- a/README.md
+++ b/README.md
@@ -32,7 +32,7 @@ Here's a quick way to launch a cluster on EC2, assuming you already have an [AWS
 ```sh
 flintrock launch test-cluster \
     --num-slaves 1 \
-    --spark-version 2.2.0 \
+    --spark-version 2.3.0 \
     --ec2-key-name key_name \
     --ec2-identity-file /path/to/key.pem \
     --ec2-ami ami-97785bed \
@@ -89,7 +89,7 @@ these steps:
    better performance.
 3. Make sure Flintrock is configured to use Hadoop/HDFS 2.7+. Earlier
    versions of Hadoop do not have solid implementations of `s3a://`.
-   Flintrock's default is Hadoop 2.7.5, so you don't need to do anything
+   Flintrock's default is Hadoop 2.8.4, so you don't need to do anything
    here if you're using a vanilla configuration.
 
 With this approach you don't need to copy around your AWS credentials
@@ -243,7 +243,7 @@ provider: ec2
 
 services:
   spark:
-    version: 2.2.0
+    version: 2.3.0
 
 launch:
   num-slaves: 1
diff --git a/flintrock/config.yaml.template b/flintrock/config.yaml.template
index a134790..2987d96 100644
--- a/flintrock/config.yaml.template
+++ b/flintrock/config.yaml.template
@@ -1,6 +1,6 @@
 services:
   spark:
-    version: 2.2.0
+    version: 2.3.0
     # git-commit: latest  # if not 'latest', provide a full commit SHA; e.g. d6dc12ef0146ae409834c78737c116050961f350
     # git-repository:  # optional; defaults to https://github.com/apache/spark
     # optional; defaults to download from from the official Spark S3 bucket
@@ -10,7 +10,7 @@ services:
     # download-source: "https://www.example.com/files/spark/{v}/spark-{v}.tar.gz"
     # executor-instances: 1
   hdfs:
-    version: 2.7.5
+    version: 2.8.4
     # optional; defaults to download from a dynamically selected Apache mirror
     #   - must contain a {v} template corresponding to the version
     #   - must be a .tar.gz file
diff --git a/flintrock/core.py b/flintrock/core.py
index 4dcac0a..a0f8edd 100644
--- a/flintrock/core.py
+++ b/flintrock/core.py
@@ -593,9 +593,16 @@ def setup_node(
     ensure_java8(ssh_client)
 
     for service in services:
-        service.install(
-            ssh_client=ssh_client,
-            cluster=cluster)
+        try:
+            service.install(
+                ssh_client=ssh_client,
+                cluster=cluster,
+            )
+        except Exception as e:
+            raise Exception(
+                "Failed to install {}."
+                .format(type(service).__name__)
+            ) from e
 
 
 def provision_cluster(
diff --git a/flintrock/flintrock.py b/flintrock/flintrock.py
index 8cc3af6..9ff212b 100644
--- a/flintrock/flintrock.py
+++ b/flintrock/flintrock.py
@@ -173,6 +173,52 @@ def configure_log(debug: bool):
     root_logger.addHandler(handler)
 
 
+def build_hdfs_download_url(ctx, param, value):
+    hdfs_download_url = value.format(v=ctx.params['hdfs_version'])
+    return hdfs_download_url
+
+
+def build_spark_download_url(ctx, param, value):
+    spark_download_url = value.format(v=ctx.params['spark_version'])
+    return spark_download_url
+
+
+def validate_download_source(url):
+    if 'spark' in url:
+        software = 'Spark'
+    elif 'hadoop' in url:
+        software = 'Hadoop'
+    else:
+        software = 'software'
+
+    parsed_url = urllib.parse.urlparse(url)
+
+    if parsed_url.netloc == 'www.apache.org' and parsed_url.path == '/dyn/closer.lua':
+        logger.warning(
+            "Warning: "
+            "Downloading {software} from an Apache mirror. Apache mirrors are "
+            "often slow and unreliable, and typically only serve the most recent releases. "
+            "We strongly recommend you specify a custom download source. "
+            "For more background on this issue, please see: https://github.com/nchammas/flintrock/issues/238"
+            .format(
+                software=software,
+            )
+        )
+        try:
+            urllib.request.urlopen(url)
+        except urllib.error.HTTPError as e:
+            raise Error(
+                "Error: Could not access {software} download. Maybe try a more recent release?\n"
+                "  - Automatically redirected to: {url}\n"
+                "  - HTTP error: {code}"
+                .format(
+                    software=software,
+                    url=e.url,
+                    code=e.code,
+                )
+            )
+
+
 @click.group()
 @click.option(
     '--config',
@@ -208,11 +254,12 @@ def cli(cli_context, config, provider, debug):
 @click.argument('cluster-name')
 @click.option('--num-slaves', type=click.IntRange(min=1), required=True)
 @click.option('--install-hdfs/--no-install-hdfs', default=False)
-@click.option('--hdfs-version', default='2.7.5')
+@click.option('--hdfs-version', default='2.8.4')
 @click.option('--hdfs-download-source',
               help="URL to download Hadoop from.",
-              default='http://www.apache.org/dyn/closer.lua/hadoop/common/hadoop-{v}/hadoop-{v}.tar.gz?as_json',
-              show_default=True)
+              default='https://www.apache.org/dyn/closer.lua?action=download&filename=hadoop/common/hadoop-{v}/hadoop-{v}.tar.gz',
+              show_default=True,
+              callback=build_hdfs_download_url)
 @click.option('--install-spark/--no-install-spark', default=True)
 @click.option('--spark-executor-instances', default=1,
               help="How many executor instances per worker.")
@@ -224,8 +271,9 @@ def cli(cli_context, config, provider, debug):
               help="Spark release version to install.")
 @click.option('--spark-download-source',
               help="URL to download a release of Spark from.",
-              default='https://s3.amazonaws.com/spark-related-packages/spark-{v}-bin-hadoop2.6.tgz',
-              show_default=True)
+              default='https://www.apache.org/dyn/closer.lua?action=download&filename=spark/spark-{v}/spark-{v}-bin-hadoop2.7.tgz',
+              show_default=True,
+              callback=build_spark_download_url)
 @click.option('--spark-git-commit',
               help="Git commit to build Spark from. "
                    "Set to 'latest' to build Spark from the latest commit on the "
@@ -351,10 +399,15 @@ def launch(
     check_external_dependency('ssh-keygen')
 
     if install_hdfs:
-        hdfs = HDFS(version=hdfs_version, download_source=hdfs_download_source)
+        validate_download_source(hdfs_download_source)
+        hdfs = HDFS(
+            version=hdfs_version,
+            download_source=hdfs_download_source,
+        )
         services += [hdfs]
     if install_spark:
         if spark_version:
+            validate_download_source(spark_download_source)
             spark = Spark(
                 spark_executor_instances=spark_executor_instances,
                 version=spark_version,
diff --git a/flintrock/scripts/download-hadoop.py b/flintrock/scripts/download-hadoop.py
deleted file mode 100644
index bdab8e1..0000000
--- a/flintrock/scripts/download-hadoop.py
+++ /dev/null
@@ -1,44 +0,0 @@
-"""
-Download Hadoop from the best available Apache mirror or a custom location.
-"""
-
-from __future__ import print_function
-
-import json
-import sys
-import subprocess
-
-if sys.version_info < (3, 0):
-    from urllib import urlretrieve
-    from urllib2 import urlopen
-else:
-    from urllib.request import urlopen, urlretrieve
-
-
-if __name__ == '__main__':
-    hadoop_version = sys.argv[1]
-    mirror_url = sys.argv[2].format(v=hadoop_version)
-    default_mirror_url = \
-        'http://www.apache.org/dyn/closer.lua/hadoop/common/hadoop-{v}/hadoop-{v}.tar.gz?as_json'\
-        .format(v=hadoop_version)
-
-    tries = 0
-    while tries < 3:
-        if mirror_url == default_mirror_url:
-            mirror_info = json.loads(urlopen(mirror_url).read().decode('utf-8'))
-            file_url = mirror_info['preferred'] + mirror_info['path_info']
-        else:
-            file_url = mirror_url
-
-        print("Downloading file at:", file_url)
-        file_path, _ = \
-            urlretrieve(url=file_url, filename="hadoop-{v}.tar.gz".format(v=hadoop_version))
-        ret = subprocess.call(['gzip', '--test', file_path])
-
-        if ret == 0:
-            break
-        else:
-            tries += 1
-            print("gzip check failed. Retrying download...")
-
-    sys.exit(ret)
diff --git a/flintrock/scripts/download-package.py b/flintrock/scripts/download-package.py
new file mode 100644
index 0000000..64416ed
--- /dev/null
+++ b/flintrock/scripts/download-package.py
@@ -0,0 +1,57 @@
+from __future__ import print_function
+
+import argparse
+import errno
+import os.path
+import sys
+import subprocess
+import time
+
+MAX_TRIES = 5
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('url')
+    parser.add_argument('destination_dir')
+    args = parser.parse_args()
+    return (args.url, args.destination_dir)
+
+
+if __name__ == '__main__':
+    url, destination_dir = parse_args()
+
+    try:
+        os.makedirs(destination_dir, mode=0o755)
+    except OSError as e:
+        if e.errno == errno.EEXIST:
+            pass
+        else:
+            raise
+
+    download_path = '{}.download'.format(os.path.basename(destination_dir))
+
+    tries = 0
+    while True:
+        try:
+            subprocess.check_call(['curl', '--location', '--output', download_path, url])
+            subprocess.check_call(['gzip', '--test', download_path])
+            subprocess.check_call(['tar', 'xzf', download_path, '-C', destination_dir, '--strip-components=1'])
+            subprocess.check_call(['rm', download_path])
+        except subprocess.CalledProcessError as e:
+            print(e, file=sys.stderr)
+            if tries < MAX_TRIES:
+                tries += 1
+                time.sleep(1)
+            else:
+                print(
+                    "Failed to download and unpack '{url}' after {tries} tries."
+                    .format(
+                        url=url,
+                        tries=MAX_TRIES,
+                    ),
+                    file=sys.stderr,
+                )
+                sys.exit(1)
+        else:
+            break
diff --git a/flintrock/scripts/install-spark.sh b/flintrock/scripts/install-spark.sh
deleted file mode 100644
index bee28d7..0000000
--- a/flintrock/scripts/install-spark.sh
+++ /dev/null
@@ -1,37 +0,0 @@
-#!/bin/bash
-
-set -e
-
-url="$1"
-
-echo "Installing Spark..."
-echo "  from: ${url}"
-
-file="$(basename ${url})"
-
-# S3 is generally reliable, but sometimes when launching really large
-# clusters it can hiccup on us, in which case we'll need to retry the
-# download.
-set +e
-tries=1
-while true; do
-    curl --remote-name "${url}"
-    curl_ret=$?
-
-    if ((curl_ret == 0)); then
-        break
-    elif ((tries >= 3)); then
-        exit 1
-    else
-        tries=$((tries + 1))
-        sleep 1
-    fi
-done
-set -e
-
-gzip -t "$file"
-
-mkdir "spark"
-# strip-components puts the files in the root of spark/
-tar xzf "$file" -C "spark" --strip-components=1
-rm "$file"
diff --git a/flintrock/services.py b/flintrock/services.py
index 433b1f2..7c861e2 100644
--- a/flintrock/services.py
+++ b/flintrock/services.py
@@ -122,27 +122,25 @@ class HDFS(FlintrockService):
 
         with ssh_client.open_sftp() as sftp:
             sftp.put(
-                localpath=os.path.join(SCRIPTS_DIR, 'download-hadoop.py'),
-                remotepath='/tmp/download-hadoop.py')
+                localpath=os.path.join(SCRIPTS_DIR, 'download-package.py'),
+                remotepath='/tmp/download-package.py')
 
         ssh_check_output(
             client=ssh_client,
             command="""
                 set -e
 
-                python /tmp/download-hadoop.py "{version}" "{download_source}"
-
-                mkdir "hadoop"
-                mkdir "hadoop/conf"
-
-                tar xzf "hadoop-{version}.tar.gz" -C "hadoop" --strip-components=1
-                rm "hadoop-{version}.tar.gz"
+                python /tmp/download-package.py "{download_source}" "hadoop"
 
                 for f in $(find hadoop/bin -type f -executable -not -name '*.cmd'); do
                     sudo ln -s "$(pwd)/$f" "/usr/local/bin/$(basename $f)"
                 done
+
                 echo "export HADOOP_LIBEXEC_DIR='$(pwd)/hadoop/libexec'" >> .bashrc
-            """.format(version=self.version, download_source=self.download_source))
+            """.format(
+                version=self.version,
+                download_source=self.download_source.format(v=self.version),
+            ))
 
     def configure(
             self,
@@ -157,6 +155,11 @@ class HDFS(FlintrockService):
             'hadoop/conf/hdfs-site.xml',
         ]
 
+        ssh_check_output(
+            client=ssh_client,
+            command="mkdir -p hadoop/conf",
+        )
+
         for template_path in template_paths:
             ssh_check_output(
                 client=ssh_client,
@@ -274,64 +277,58 @@ class Spark(FlintrockService):
             self,
             ssh_client: paramiko.client.SSHClient,
             cluster: FlintrockCluster):
-
         logger.info("[{h}] Installing Spark...".format(
             h=ssh_client.get_transport().getpeername()[0]))
 
-        try:
-            if self.version:
-                with ssh_client.open_sftp() as sftp:
-                    sftp.put(
-                        localpath=os.path.join(SCRIPTS_DIR, 'install-spark.sh'),
-                        remotepath='/tmp/install-spark.sh')
-                    sftp.chmod(path='/tmp/install-spark.sh', mode=0o755)
-                url = self.download_source.format(v=self.version)
-                ssh_check_output(
-                    client=ssh_client,
-                    command="""
-                        set -e
-                        /tmp/install-spark.sh {url}
-                        rm -f /tmp/install-spark.sh
-                    """.format(url=shlex.quote(url)))
-            else:
-                ssh_check_output(
-                    client=ssh_client,
-                    command="""
-                        set -e
-                        sudo yum install -y git
-                        sudo yum install -y java-devel
-                        """)
-                ssh_check_output(
-                    client=ssh_client,
-                    command="""
-                        set -e
-                        git clone {repo} spark
-                        cd spark
-                        git reset --hard {commit}
-                        if [ -e "make-distribution.sh" ]; then
-                            ./make-distribution.sh -Phadoop-{hadoop_short_version}
-                        else
-                            ./dev/make-distribution.sh -Phadoop-{hadoop_short_version}
-                        fi
-                    """.format(
-                        repo=shlex.quote(self.git_repository),
-                        commit=shlex.quote(self.git_commit),
-                        hadoop_short_version='.'.join(self.hadoop_version.split('.')[:2]),
-                    ))
+        if self.version:
+            with ssh_client.open_sftp() as sftp:
+                sftp.put(
+                    localpath=os.path.join(SCRIPTS_DIR, 'download-package.py'),
+                    remotepath='/tmp/download-package.py')
+
+            ssh_check_output(
+                client=ssh_client,
+                command="""
+                    python /tmp/download-package.py "{download_source}" "spark"
+                """.format(
+                    version=self.version,
+                    download_source=self.download_source.format(v=self.version),
+                ))
+
+        else:
             ssh_check_output(
                 client=ssh_client,
                 command="""
                     set -e
-                    for f in $(find spark/bin -type f -executable -not -name '*.cmd'); do
-                        sudo ln -s "$(pwd)/$f" "/usr/local/bin/$(basename $f)"
-                    done
-                    echo "export SPARK_HOME='$(pwd)/spark'" >> .bashrc
-                """)
-        except Exception as e:
-            # TODO: This should be a more specific exception.
-            print("Error: Failed to install Spark.", file=sys.stderr)
-            print(e, file=sys.stderr)
-            raise
+                    sudo yum install -y git
+                    sudo yum install -y java-devel
+                    """)
+            ssh_check_output(
+                client=ssh_client,
+                command="""
+                    set -e
+                    git clone {repo} spark
+                    cd spark
+                    git reset --hard {commit}
+                    if [ -e "make-distribution.sh" ]; then
+                        ./make-distribution.sh -Phadoop-{hadoop_short_version}
+                    else
+                        ./dev/make-distribution.sh -Phadoop-{hadoop_short_version}
+                    fi
+                """.format(
+                    repo=shlex.quote(self.git_repository),
+                    commit=shlex.quote(self.git_commit),
+                    hadoop_short_version='.'.join(self.hadoop_version.split('.')[:2]),
+                ))
+        ssh_check_output(
+            client=ssh_client,
+            command="""
+                set -e
+                for f in $(find spark/bin -type f -executable -not -name '*.cmd'); do
+                    sudo ln -s "$(pwd)/$f" "/usr/local/bin/$(basename $f)"
+                done
+                echo "export SPARK_HOME='$(pwd)/spark'" >> .bashrc
+            """)
 
     def configure(
             self,
@@ -343,6 +340,12 @@ class Spark(FlintrockService):
             'spark/conf/slaves',
             'spark/conf/spark-defaults.conf',
         ]
+
+        ssh_check_output(
+            client=ssh_client,
+            command="mkdir -p spark/conf",
+        )
+
         for template_path in template_paths:
             ssh_check_output(
                 client=ssh_client,


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index a3df8b8..de07e68 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -10,8 +10,8 @@ from flintrock.core import StorageDirs
 # External
 import pytest
 
-HADOOP_VERSION = '2.7.5'
-SPARK_VERSION = '2.2.0'
+HADOOP_VERSION = '2.8.4'
+SPARK_VERSION = '2.3.0'
 SPARK_GIT_COMMIT = '584354eaac02531c9584188b143367ba694b0c34'  # 2.0.2
 
 
@@ -25,6 +25,15 @@ aws_credentials_required = (
         reason="USE_AWS_CREDENTIALS not set"))
 
 
+@pytest.fixture(scope='session')
+def project_root_dir():
+    return os.path.dirname(
+        os.path.dirname(
+            os.path.realpath(__file__)
+        )
+    )
+
+
 @pytest.fixture(scope='session')
 def dummy_cluster():
     storage_dirs = StorageDirs(
diff --git a/tests/test_core.py b/tests/test_core.py
index 8f9af0d..9d9dfaf 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -16,7 +16,7 @@ FLINTROCK_ROOT_DIR = (
 @pytest.mark.parametrize(
     'spark_version', [
         (''),
-        ('2.2.0'),
+        ('2.3.0'),
         ('0626b11147133b67b26a04b4819f61a33dd958d3'),
     ])
 def test_templates(dummy_cluster, spark_version):
diff --git a/tests/test_flintrock.py b/tests/test_flintrock.py
index 474e09c..5b379e7 100644
--- a/tests/test_flintrock.py
+++ b/tests/test_flintrock.py
@@ -3,14 +3,16 @@ import pytest
 
 # Flintrock modules
 from flintrock.exceptions import (
-    UsageError
+    Error,
+    UsageError,
 )
 from flintrock.flintrock import (
     option_name_to_variable_name,
     variable_name_to_option_name,
     option_requires,
     mutually_exclusive,
-    get_latest_commit
+    get_latest_commit,
+    validate_download_source,
 )
 
 
@@ -138,3 +140,20 @@ def test_get_latest_commit():
 
     with pytest.raises(Exception):
         get_latest_commit("https://github.com/apache/nonexistent-repo")
+
+
+@pytest.mark.xfail(
+    reason=(
+        "This test will fail whenever a new Hadoop or Spark "
+        "release is made, which is out of our control."
+    ),
+    raises=Error,
+)
+def test_validate_valid_download_source():
+    validate_download_source("https://www.apache.org/dyn/closer.lua?action=download&filename=hadoop/common/hadoop-2.8.4/hadoop-2.8.4.tar.gz")
+    validate_download_source("https://www.apache.org/dyn/closer.lua?action=download&filename=spark/spark-2.3.0/spark-2.3.0-bin-hadoop2.7.tgz")
+
+
+def test_validate_invalid_download_source():
+    with pytest.raises(Error):
+        validate_download_source("https://www.apache.org/dyn/closer.lua?action=download&filename=hadoop/common/hadoop-2.8.3/hadoop-2.8.3.tar.gz")
diff --git a/tests/test_scripts.py b/tests/test_scripts.py
new file mode 100644
index 0000000..f464df2
--- /dev/null
+++ b/tests/test_scripts.py
@@ -0,0 +1,39 @@
+import os
+import subprocess
+import tempfile
+
+import pytest
+
+
+@pytest.fixture(scope='module')
+def tgz_file(request):
+    with tempfile.NamedTemporaryFile() as source_file:
+        source_file.file.write('Hi!'.encode('utf-8'))
+        tgz_file_name = source_file.name + '.tgz'
+        subprocess.run(
+            ['tar', 'czf', tgz_file_name, source_file.name],
+            check=True,
+        )
+
+    def destroy():
+        subprocess.run(
+            ['rm', tgz_file_name],
+            check=True,
+        )
+    request.addfinalizer(destroy)
+
+    return tgz_file_name
+
+
+@pytest.mark.parametrize('python', ['python', 'python2'])
+def test_download_package(python, project_root_dir, tgz_file):
+    with tempfile.TemporaryDirectory() as temp_dir:
+        subprocess.run(
+            [
+                python,
+                os.path.join(project_root_dir, 'flintrock/scripts/download-package.py'),
+                'file://' + tgz_file,
+                temp_dir,
+            ],
+            check=True,
+        )