Merge pull request #68 from Yelp/gcoll_COREML-2570_paasta_spark_run_check_k8s_volumes

[COREML-2570] fix k8s volumes being mounted twice

Author: gcollura

service_configuration_lib/spark_config.py

@@ -1,4 +1,6 @@
+import functools
 import hashlib
+import itertools
 import json
 import logging
 import os
@@ -63,11 +65,6 @@
     'spark.kubernetes.executor.label.paasta.yelp.com/cluster',
 }
 K8S_AUTH_FOLDER = '/etc/pki/spark'
-DEFAULT_SPARK_K8S_SECRET_VOLUME = {
-    'hostPath': K8S_AUTH_FOLDER,
-    'containerPath': K8S_AUTH_FOLDER,
-    'mode': 'RO',
-}
 
 log = logging.Logger(__name__)
 
@@ -173,19 +170,37 @@ def _get_k8s_docker_volumes_conf(
     volumes: Optional[List[Mapping[str, str]]] = None,
 ):
     env = {}
+    mounted_volumes = set()
     k8s_volumes = volumes or []
-    k8s_volumes.append(DEFAULT_SPARK_K8S_SECRET_VOLUME)
+    k8s_volumes.append({'containerPath': K8S_AUTH_FOLDER, 'hostPath': K8S_AUTH_FOLDER, 'mode': 'RO'})
     k8s_volumes.append({'containerPath': '/etc/passwd', 'hostPath': '/etc/passwd', 'mode': 'RO'})
     k8s_volumes.append({'containerPath': '/etc/group', 'hostPath': '/etc/group', 'mode': 'RO'})
-    for volume_name, volume in enumerate(k8s_volumes):
-        env[f'spark.kubernetes.executor.volumes.hostPath.{volume_name}.mount.path'] = volume['containerPath']
-        env[f'spark.kubernetes.executor.volumes.hostPath.{volume_name}.mount.readOnly'] = (
-            'true' if volume['mode'].lower() == 'ro' else 'false'
-        )
-        env[f'spark.kubernetes.executor.volumes.hostPath.{volume_name}.options.path'] = volume['hostPath']
+    _get_k8s_volume = functools.partial(_get_k8s_volume_hostpath_dict, count=itertools.count())
+
+    for volume in k8s_volumes:
+        host_path, container_path, mode = volume['hostPath'], volume['containerPath'], volume['mode']
+        if os.path.exists(host_path) and host_path not in mounted_volumes:
+            env.update(_get_k8s_volume(host_path, container_path, mode))
+            mounted_volumes.add(host_path)
+        else:
+            log.warning(
+                f'Path {host_path} does not exist on this host or it has already been mounted.'
+                ' Skipping this bindings.',
+            )
     return env
 
 
+def _get_k8s_volume_hostpath_dict(host_path: str, container_path: str, mode: str, count: itertools.count):
+    volume_name = next(count)
+    return {
+        f'spark.kubernetes.executor.volumes.hostPath.{volume_name}.mount.path': container_path,
+        f'spark.kubernetes.executor.volumes.hostPath.{volume_name}.options.path': host_path,
+        f'spark.kubernetes.executor.volumes.hostPath.{volume_name}.mount.readOnly': (
+            'true' if mode.lower() == 'ro' else 'false'
+        ),
+    }
+
+
 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

setup.py

@@ -17,7 +17,7 @@
 
 setup(
     name='service-configuration-lib',
-    version='2.5.7',
+    version='2.5.8',
     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

@@ -1,3 +1,5 @@
+import functools
+import itertools
 import json
 import os
 from unittest import mock
@@ -165,7 +167,9 @@ def mock_paasta_volumes(self, monkeypatch, tmpdir):
 
     @pytest.fixture
     def mock_existed_files(self, mock_paasta_volumes):
-        existed_files = [v.split(':')[0] for v in mock_paasta_volumes] + ['/host/file1', '/host/file2', '/host/file3']
+        existed_files = [v.split(':')[0] for v in mock_paasta_volumes] + [
+            '/host/file1', '/host/file2', '/host/file3', '/etc/pki/spark', '/etc/group', '/etc/passwd',
+        ]
         with mock.patch('os.path.exists', side_effect=lambda f: f in existed_files):
             yield existed_files
 
@@ -220,6 +224,45 @@ def test_get_mesos_docker_volumes_conf(
         )
         assert sorted(output[validate_key].split(',')) == sorted(set(expected_volumes))
 
+    def test_get_k8s_volume_hostpath_dict(self):
+        assert spark_config._get_k8s_volume_hostpath_dict(
+            '/host/file1', '/container/file1', 'RO', itertools.count(),
+        ) == {
+            'spark.kubernetes.executor.volumes.hostPath.0.mount.path': '/container/file1',
+            'spark.kubernetes.executor.volumes.hostPath.0.options.path': '/host/file1',
+            'spark.kubernetes.executor.volumes.hostPath.0.mount.readOnly': 'true',
+        }
+
+    @pytest.mark.parametrize(
+        'volumes', [
+            None,
+            [
+                {'hostPath': '/host/file1', 'containerPath': '/containter/file1', 'mode': 'RO'},
+                {'hostPath': '/host/file2', 'containerPath': '/containter/file2', 'mode': 'RO'},
+                {'hostPath': '/host/paasta1', 'containerPath': '/container/paasta1', 'mode': 'RO'},
+            ],
+        ],
+    )
+    @pytest.mark.usefixtures('mock_existed_files')
+    def test_get_k8s_docker_volumes_conf(self, volumes):
+        expected_volumes = {}
+
+        _get_k8s_volume = functools.partial(spark_config._get_k8s_volume_hostpath_dict, count=itertools.count())
+        if volumes:
+            for volume in volumes:
+                expected_volumes.update(
+                    _get_k8s_volume(volume['hostPath'], volume['containerPath'], volume['mode']),
+                )
+
+        expected_volumes.update({
+            **_get_k8s_volume('/etc/pki/spark', '/etc/pki/spark', 'ro'),
+            **_get_k8s_volume('/etc/passwd', '/etc/passwd', 'ro'),
+            **_get_k8s_volume('/etc/group', '/etc/group', 'ro'),
+        })
+
+        output = spark_config._get_k8s_docker_volumes_conf(volumes)
+        assert output == expected_volumes
+
     @pytest.fixture
     def mock_account_id(self, tmpdir, monkeypatch):
         def get_client(service_name, **kwargs):
@@ -744,7 +787,7 @@ def test_get_spark_conf_mesos(
     @pytest.fixture
     def assert_kubernetes_conf(self):
         expected_output = {
-            'spark.master': f'k8s://https://k8s.paasta-{self.cluster}.yelp:16443',
+            'spark.master': f'k8s://https://k8s.{self.cluster}.paasta:6443',
             'spark.executorEnv.PAASTA_SERVICE': self.service,
             'spark.executorEnv.PAASTA_INSTANCE': self.instance,
             'spark.executorEnv.PAASTA_CLUSTER': self.cluster,