Merge pull request #70 from Yelp/gcoll_COREML-2591_fix_issues_with_long_label_names

[COREML-2591] Fix issue with long label names in k8s

Author: gcollura

service_configuration_lib/spark_config.py

@@ -1,3 +1,4 @@
+import base64
 import functools
 import hashlib
 import itertools
@@ -34,6 +35,7 @@
 DEFAULT_EXECUTOR_CORES = 2
 DEFAULT_EXECUTOR_INSTANCES = 2
 DEFAULT_EXECUTOR_MEMORY = '4g'
+DEFAULT_K8S_LABEL_LENGTH = 63
 
 
 NON_CONFIGURABLE_SPARK_OPTS = {
@@ -446,6 +448,12 @@ def _get_k8s_spark_env(
     volumes: Optional[List[Mapping[str, str]]],
     paasta_pool: str,
 ) -> Dict[str, str]:
+    # RFC 1123: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#dns-label-names
+    # technically only paasta instance can be longer than 63 chars. But we apply the normalization regardless.
+    # NOTE: this affects only k8s labels, not the pod names.
+    _paasta_cluster = _get_k8s_resource_name_limit_size_with_hash(paasta_cluster)
+    _paasta_service = _get_k8s_resource_name_limit_size_with_hash(paasta_service)
+    _paasta_instance = _get_k8s_resource_name_limit_size_with_hash(paasta_instance)
     spark_env = {
         'spark.master': f'k8s://https://k8s.{paasta_cluster}.paasta:6443',
         'spark.executorEnv.PAASTA_SERVICE': paasta_service,
@@ -460,19 +468,37 @@ def _get_k8s_spark_env(
         'spark.kubernetes.authenticate.clientKeyFile': f'{K8S_AUTH_FOLDER}/{paasta_cluster}-client.key',
         'spark.kubernetes.authenticate.clientCertFile': f'{K8S_AUTH_FOLDER}/{paasta_cluster}-client.crt',
         'spark.kubernetes.container.image.pullPolicy': 'Always',
-        'spark.kubernetes.executor.label.yelp.com/paasta_service': paasta_service,
-        'spark.kubernetes.executor.label.yelp.com/paasta_instance': paasta_instance,
-        'spark.kubernetes.executor.label.yelp.com/paasta_cluster': paasta_cluster,
-        'spark.kubernetes.executor.label.paasta.yelp.com/service': paasta_service,
-        'spark.kubernetes.executor.label.paasta.yelp.com/instance': paasta_instance,
-        'spark.kubernetes.executor.label.paasta.yelp.com/cluster': paasta_cluster,
+        'spark.kubernetes.executor.label.yelp.com/paasta_service': _paasta_service,
+        'spark.kubernetes.executor.label.yelp.com/paasta_instance': _paasta_instance,
+        'spark.kubernetes.executor.label.yelp.com/paasta_cluster': _paasta_cluster,
+        'spark.kubernetes.executor.label.paasta.yelp.com/service': _paasta_service,
+        'spark.kubernetes.executor.label.paasta.yelp.com/instance': _paasta_instance,
+        'spark.kubernetes.executor.label.paasta.yelp.com/cluster': _paasta_cluster,
         'spark.kubernetes.node.selector.yelp.com/pool': paasta_pool,
         'spark.kubernetes.executor.label.yelp.com/pool': paasta_pool,
         **_get_k8s_docker_volumes_conf(volumes),
     }
     return spark_env
 
 
+def _get_k8s_resource_name_limit_size_with_hash(name: str, limit: int = 63, suffix: int = 4) -> str:
+    """ Returns `name` unchanged if it's length does not exceed the `limit`.
+        Otherwise, returns truncated `name` with it's hash of size `suffix`
+        appended.
+
+        base32 encoding is chosen as it satisfies the common requirement in
+        various k8s names to be alphanumeric.
+
+        NOTE: This function is the same as paasta/paasta_tools/kubernetes_tools.py
+    """
+    if len(name) > limit:
+        digest = hashlib.md5(name.encode()).digest()
+        hash = base64.b32encode(digest).decode().replace('=', '').lower()
+        return f'{name[:(limit-suffix-1)]}-{hash[:suffix]}'
+    else:
+        return name
+
+
 def stringify_spark_env(spark_env: Mapping[str, str]) -> str:
     return ' '.join([f'--conf {k}={v}' for k, v in spark_env.items()])
 

setup.py

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

@@ -1137,3 +1137,17 @@ def test_send_and_calculate_resources_cost(
     mock_clusterman_metrics.util.costs.estimate_cost_per_hour.assert_called_once_with(
         cluster='test-cluster', pool='test-pool', cpus=10, mem=2048,
     )
+
+
+@pytest.mark.parametrize(
+    'instance_name,expected_instance_label',
+    (
+        ('my_job.do_something', 'my_job.do_something'),
+        (
+            f"my_job.{'a'* 100}",
+            'my_job.aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-6xhe',
+        ),
+    ),
+)
+def test_get_k8s_resource_name_limit_size_with_hash(instance_name, expected_instance_label):
+    assert expected_instance_label == spark_config._get_k8s_resource_name_limit_size_with_hash(instance_name)