fix: Fail if too long lease name, #910

* for backwards compatibility it's possible to allow old behavior
  with config, e.g. to support rolling update

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

@@ -47,4 +47,8 @@ akka.coordination.lease.kubernetes {
     # server that are required. If this timeout is hit then the lease *may* be taken due to the response being lost
     # on the way back from the API server but will be reported as not taken and can be safely retried.
     lease-operation-timeout = 5s
+
+    # For backwards compatibility to support rolling update. Truncation of lease name may cause conflicting names
+    # of different lease resources.
+    allow-lease-name-truncation = off
 }

lease-kubernetes/src/main/scala/akka/coordination/lease/kubernetes/KubernetesLease.scala

@@ -60,10 +60,22 @@ object KubernetesLease {
    * 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, allowLeaseNameTruncation: Boolean): String = {
     val normalized =
       Normalizer.normalize(name, Normalizer.Form.NFKD).toLowerCase.replaceAll("[_.]", "-").replaceAll("[^-a-z0-9]", "")
-    trim(truncateTo63Characters(normalized), List('-'))
+
+    if (allowLeaseNameTruncation)
+      trim(truncateTo63Characters(normalized), List('-'))
+    else {
+      if (normalized.length > 63)
+        throw new IllegalArgumentException(
+          s"Too long lease resource name [$normalized]. At most 63 characters is accepted. " +
+          "A custom resource name can be defined in configuration `lease-name`, ClusterSingletonSettings, " +
+          "or ClusterShardingSettings. " +
+          "For backwards compatibility, lease name truncation can be allowed by enabling config " +
+          "`akka.coordination.lease.kubernetes.allow-lease-name-truncation`.")
+      trim(normalized, List('-'))
+    }
   }
 
 }
@@ -74,6 +86,8 @@ class KubernetesLease private[akka] (system: ExtendedActorSystem, leaseTaken: At
   private val log = Logging(system, classOf[KubernetesLease])
 
   private val k8sSettings = KubernetesSettings(settings.leaseConfig, settings.timeoutSettings)
+  private val leaseName = makeDNS1039Compatible(settings.leaseName, k8sSettings.allowLeaseNameTruncation)
+
   private val k8sApi: Future[KubernetesApi] = {
     implicit val blockingDispatcher: ExecutionContext = system.dispatchers.lookup(DefaultBlockingDispatcherId)
     for {
@@ -96,7 +110,6 @@ class KubernetesLease private[akka] (system: ExtendedActorSystem, leaseTaken: At
   def this(leaseSettings: LeaseSettings, system: ExtendedActorSystem) =
     this(system, new AtomicBoolean(false), leaseSettings)
 
-  private val leaseName = makeDNS1039Compatible(settings.leaseName)
   private val leaseActor: Future[ActorRef] = {
     k8sApi.map { api =>
       log.debug(

lease-kubernetes/src/main/scala/akka/coordination/lease/kubernetes/KubernetesSettings.scala

@@ -52,7 +52,8 @@ private[akka] object KubernetesSettings {
       config.getString("namespace-path"),
       apiServerRequestTimeout,
       secure = config.getBoolean("secure-api-server"),
-      apiServerRequestTimeout / 2)
+      apiServerRequestTimeout / 2,
+      config.getBoolean("allow-lease-name-truncation"))
 
   }
 }
@@ -70,4 +71,5 @@ private[akka] class KubernetesSettings(
     val namespacePath: String,
     val apiServerRequestTimeout: FiniteDuration,
     val secure: Boolean = true,
-    val bodyReadTimeout: FiniteDuration = 1.second)
+    val bodyReadTimeout: FiniteDuration = 1.second,
+    val allowLeaseNameTruncation: Boolean = false)