COREML-3036 | Add Spark conf log and console progress config (#75)

* COREML-3036 | Add Spark conf log and console progress config

* COREML-3036 | Tests for Spark conf log and console progress config

* COREML-3036 | Version bump

Co-authored-by: Sameer Sharma <[email protected]>
Co-authored-by: Sameer Sharma <[email protected]>

Author: 3 people

service_configuration_lib/spark_config.py

@@ -206,6 +206,16 @@ def _get_k8s_volume_hostpath_dict(host_path: str, container_path: str, mode: str
     }
 
 
+def _append_spark_config(spark_opts: Dict[str, str], config_name: str, config_value: str) -> Dict[str, str]:
+    # config already defined by the user, don't modify
+    if config_name in spark_opts:
+        return spark_opts
+
+    # append the config
+    spark_opts[config_name] = config_value
+    return spark_opts
+
+
 def _append_sql_shuffle_partitions_conf(spark_opts: Dict[str, str]) -> Dict[str, str]:
     if 'spark.sql.shuffle.partitions' in spark_opts:
         return spark_opts
@@ -677,6 +687,12 @@ def get_spark_conf(
 
     # configure sql shuffle partitions
     spark_conf = _append_sql_shuffle_partitions_conf(spark_conf)
+
+    # configure spark conf log
+    spark_conf = _append_spark_config(spark_conf, 'spark.logConf', 'true')
+
+    # configure spark Console Progress
+    spark_conf = _append_spark_config(spark_conf, 'spark.ui.showConsoleProgress', 'true')
     return spark_conf
 
 

setup.py

@@ -17,7 +17,7 @@
 
 setup(
     name='service-configuration-lib',
-    version='2.10.0',
+    version='2.10.1',
     provides=['service_configuration_lib'],
     description='Start, stop, and inspect Yelp SOA services',
     url='https://github.com/Yelp/service_configuration_lib',

tests/spark_config_test.py

@@ -507,6 +507,54 @@ def test_append_sql_shuffle_partitions_conf(
         key = 'spark.sql.shuffle.partitions'
         assert output[key] == expected_output
 
+    @pytest.mark.parametrize(
+        'user_spark_opts,expected_output', [
+            # not configured by user
+            ({}, 'true'),
+            # configured by user
+            ({'spark.logConf': 'false'}, 'false'),
+        ],
+    )
+    def test_append_spark_conf_log(
+            self, user_spark_opts, expected_output,
+    ):
+        key = 'spark.logConf'
+        output = spark_config._append_spark_config(user_spark_opts, key, 'true')
+
+        assert output[key] == expected_output
+
+    @pytest.mark.parametrize(
+        'user_spark_opts,expected_output', [
+            # not configured by user
+            ({}, 'true'),
+            # configured by user
+            ({'spark.ui.showConsoleProgress': 'false'}, 'false'),
+        ],
+    )
+    def test_append_console_progress_conf(
+            self, user_spark_opts, expected_output,
+    ):
+        key = 'spark.ui.showConsoleProgress'
+        output = spark_config._append_spark_config(user_spark_opts, key, 'true')
+
+        assert output[key] == expected_output
+
+    @pytest.fixture
+    def mock_append_spark_conf_log(self):
+        return_value = {'spark.logConf': 'true'}
+        with MockConfigFunction(
+                '_append_spark_config', return_value,
+        ) as m:
+            yield m
+
+    @pytest.fixture
+    def mock_append_console_progress_conf(self):
+        return_value = {'spark.ui.showConsoleProgress': 'true'}
+        with MockConfigFunction(
+                '_append_spark_config', return_value,
+        ) as m:
+            yield m
+
     @pytest.fixture
     def mock_get_mesos_docker_volumes_conf(self):
         return_value = {'spark.mesos.executor.docker.volumes': '/tmp:/tmp:ro'}
@@ -768,6 +816,8 @@ def test_get_spark_conf_mesos(
         assert_ui_port,
         assert_app_name,
         mock_log,
+        mock_append_spark_conf_log,
+        mock_append_console_progress_conf,
     ):
         other_spark_opts = {'spark.driver.memory': '2g', 'spark.executor.memoryOverhead': '1024'}
         not_allowed_opts = {'spark.executorEnv.PAASTA_SERVICE': 'random-service'}
@@ -810,7 +860,9 @@ def test_get_spark_conf_mesos(
             list(mock_get_mesos_docker_volumes_conf.return_value.keys()) +
             list(mock_adjust_spark_requested_resources_mesos.return_value.keys()) +
             list(mock_append_event_log_conf.return_value.keys()) +
-            list(mock_append_sql_shuffle_partitions_conf.return_value.keys()),
+            list(mock_append_sql_shuffle_partitions_conf.return_value.keys()) +
+            list(mock_append_spark_conf_log.return_value.keys()) +
+            list(mock_append_console_progress_conf.return_value.keys()),
         )
         assert len(set(output.keys()) - verified_keys) == 0
         mock_get_mesos_docker_volumes_conf.mocker.assert_called_once_with(