Fix Kubernetes lease name uniqueness #702

lease-kubernetes/src/main/resources/reference.conf

@@ -53,6 +53,12 @@ pekko.coordination.lease.kubernetes {
     # on the way back from the API server but will be reported as not taken and can be safely retried.
     lease-operation-timeout = 5s
 
+    # If true, lease names longer than 253 characters are hashed with SHA-256 to guarantee uniqueness
+    # while remaining DNS subdomain compatible. Names up to 253 characters are kept as-is.
+    # If false, lease names are truncated to 63 characters, which is safe for all Kubernetes subsystems.
+    # See https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#dns-subdomain-names
+    allow-lease-name-hash = false
+
     # Settings that are specific to retrying requests with 401 responses due to possible token rotation
     token-rotation-retry {
       # Number of total attempts to make

lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/AbstractKubernetesLease.scala

@@ -13,6 +13,8 @@
 
 package org.apache.pekko.coordination.lease.kubernetes
 
+import java.nio.charset.StandardCharsets
+import java.security.MessageDigest
 import java.text.Normalizer
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
@@ -31,6 +33,7 @@ import pekko.coordination.lease.LeaseTimeoutException
 import pekko.pattern.AskTimeoutException
 import pekko.util.ConstantFun
 import pekko.util.Timeout
+import org.apache.commons.codec.binary.Base32
 import org.slf4j.LoggerFactory
 
 object AbstractKubernetesLease {
@@ -49,15 +52,42 @@ object AbstractKubernetesLease {
   private def trim(name: String, characters: List[Char]): String =
     name.dropWhile(characters.contains(_)).reverse.dropWhile(characters.contains(_)).reverse
 
+  /**
+   * Hashes a name with SHA-256.
+   * Outputs the base32 unpadded lowercase encoding of the hash.
+   */
+  private def sha256Hash(name: String): String = {
+    val digest = MessageDigest.getInstance("SHA-256")
+    val bytes = digest.digest(name.getBytes(StandardCharsets.UTF_8))
+
+    val base32WithPadding = new Base32().encodeAsString(bytes).toLowerCase
+
+    val paddingIndex = base32WithPadding.indexOf('=')
+
+    if (paddingIndex > 0) base32WithPadding.substring(0, paddingIndex) else base32WithPadding
+  }
+
   /**
    * Make a name compatible with DNS 1039 standard: like a single domain name segment.
    * Regex to follow: [a-z]([-a-z0-9]*[a-z0-9])
-   * Limit the resulting name to 63 characters
    */
-  private def makeDNS1039Compatible(name: String): String = {
+  private def makeDNS1039Compatible(name: String, allowHash: Boolean): String = {
     val normalized =
       Normalizer.normalize(name, Normalizer.Form.NFKD).toLowerCase.replaceAll("[_.]", "-").replaceAll("[^-a-z0-9]", "")
-    trim(truncateTo63Characters(normalized), List('-'))
+
+    if (allowHash) {
+      // Here we allow for 253 characters on this behavior, as it is opt in
+      if (normalized.length > 253) {
+        val hash = sha256Hash(name)
+        val prefix = trim(normalized.take(253 - hash.length - 1), List('-'))
+
+        s"$prefix-$hash"
+      } else {
+        trim(normalized, List('-'))
+      }
+    } else {
+      trim(truncateTo63Characters(normalized), List('-'))
+    }
   }
 }
 
@@ -74,7 +104,7 @@ abstract class AbstractKubernetesLease(system: ExtendedActorSystem, leaseTaken:
 
   private implicit val timeout: Timeout = Timeout(settings.timeoutSettings.operationTimeout)
 
-  private val leaseName = makeDNS1039Compatible(settings.leaseName)
+  private val leaseName = makeDNS1039Compatible(settings.leaseName, k8sSettings.allowLeaseNameHash)
   private val leaseActor = system.systemActorOf(
     LeaseActor.props(k8sApi, settings, leaseName, leaseTaken),
     s"kubernetesLease${AbstractKubernetesLease.leaseCounter.incrementAndGet}")

lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettings.scala

@@ -73,7 +73,9 @@ private[pekko] object KubernetesSettings {
       secure = config.getBoolean("secure-api-server"),
       tlsVersion = config.getString("tls-version"),
       bodyReadTimeout = apiServerRequestTimeout / 2,
-      tokenRetrySettings = tokenRetrySettings)
+      tokenRetrySettings = tokenRetrySettings,
+      allowLeaseNameHash = config.getBoolean("allow-lease-name-hash")
+    )
   }
 }
 
@@ -107,4 +109,5 @@ private[pekko] class KubernetesSettings(
       10.millis,
       1.minute,
       0.3
-    ))
+    ),
+    val allowLeaseNameHash: Boolean = false)

lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/AbstractKubernetesLeaseSpec.scala

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.coordination.lease.kubernetes
+
+import org.scalatest.PrivateMethodTester
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class AbstractKubernetesLeaseSpec extends AnyWordSpec with Matchers with PrivateMethodTester {
+
+  private val makeDNS1039CompatibleMethod = PrivateMethod[String](Symbol("makeDNS1039Compatible"))
+
+  private def makeDNS1039Compatible(leaseName: String, allowLeaseHash: Boolean): String =
+    AbstractKubernetesLease.invokePrivate(makeDNS1039CompatibleMethod(leaseName, allowLeaseHash))
+
+  "AbstractKubernetesLease" should {
+    "normalize a lease name shorter than 63 characters" when {
+      "lease hash is allowed" in {
+        val leaseName = "test-system-singleton-pekko://test-system/path/to/actor"
+        makeDNS1039Compatible(leaseName, allowLeaseHash = true) shouldEqual
+        "test-system-singleton-pekkotest-systempathtoactor"
+      }
+      "lease hash is not allowed" in {
+        val leaseName = "test-system-singleton-pekko://test-system/path/to/actor"
+        makeDNS1039Compatible(leaseName, allowLeaseHash = false) shouldEqual
+        "test-system-singleton-pekkotest-systempathtoactor"
+      }
+    }
+    "normalize and truncate a lease name longer than 63 characters when lease hash is not allowed" in {
+      val leaseName = "test-system-bit-too-long-singleton-pekko://test-system-bit-too-long/path/to/actor"
+      makeDNS1039Compatible(leaseName, allowLeaseHash = false) shouldEqual
+      "test-system-bit-too-long-singleton-pekkotest-system-bit-too-lon"
+    }
+    "normalize a lease name shorter than 253 characters when lease hash is allowed" in {
+      val leaseName = "test-system-bit-too-long-singleton-pekko://test-system-bit-too-long/path/to/actor"
+      makeDNS1039Compatible(leaseName, allowLeaseHash = true) shouldEqual
+      "test-system-bit-too-long-singleton-pekkotest-system-bit-too-longpathtoactor"
+    }
+    "hash a lease name longer than 253 characters when lease hash is allowed" in {
+      val leaseName =
+        "test-with-long-system-name-that-has-more-than-the-expected-characters-count-and-is-very-long-that-will-for-sure-break-singleton-pekko://test-with-long-system-name-that-has-more-than-the-expected-characters-count-and-is-very-long-that-will-for-sure-break/path/to/actor"
+      val result = makeDNS1039Compatible(leaseName, allowLeaseHash = true)
+      result.length shouldEqual 253
+      result shouldEqual
+      "test-with-long-system-name-that-has-more-than-the-expected-characters-count-and-is-very-long-that-will-for-sure-break-singleton-pekkotest-with-long-system-name-that-has-more-than-the-expected-characte-q3oh6h2gujhpm5rk7zzhrluufpyq37duqoa35wiwgc2tyikv5jkq"
+    }
+  }
+
+}

project/Dependencies.scala

@@ -31,6 +31,8 @@ object Dependencies {
   val logbackVersion = "1.5.32"
   val slf4jVersion = "2.0.17"
 
+  val commonsCodecVersion = "1.21.0"
+
   // often called-in transitively with insecure versions of databind / core
   private val jacksonDatabind = Seq(
     "com.fasterxml.jackson.core" % "jackson-databind" % jacksonVersion)
@@ -158,6 +160,7 @@ object Dependencies {
     "org.apache.pekko" %% "pekko-http" % pekkoHttpVersion,
     "org.apache.pekko" %% "pekko-slf4j" % pekkoVersion,
     "org.apache.pekko" %% "pekko-http-spray-json" % pekkoHttpVersion,
+    "commons-codec" % "commons-codec" % commonsCodecVersion,
     "org.scalatest" %% "scalatest" % scalaTestVersion % Test,
     "org.scalatestplus" %% "junit-4-13" % scalaTestPlusJUnitVersion % Test,
     "org.apache.pekko" %% "pekko-testkit" % pekkoVersion % Test) ++