feat: configurable choice of annotations for rolling update revision (#…

…1303)

.gitignore

@@ -12,8 +12,12 @@ target/
 *.class
 *.log
 
-.metals
 .vscode
 
 # attachments created by scripts/prepare-downloads.sh
 docs/src/main/paradox/attachments
+
+# Metals detritus
+.metals
+metals.sbt
+**/project/project

docs/src/main/paradox/rolling-updates.md

@@ -237,7 +237,7 @@ Java
 
 #### Configuration
 
-The following configuration is required, more details for each and additional configurations can be found in [reference.conf](https://github.com/akka/akka-management/blob/main/rolling-updates-kubernetes/src/main/resources/reference.conf):
+The following configuration is required, more details for each and additional configurations can be found in [reference.conf](https://github.com/akka/akka-management/blob/main/rolling-update-kubernetes/src/main/resources/reference.conf):
 
 * `akka.rollingupdate.kubernetes.pod-name`: this can be provided by setting `KUBERNETES_POD_NAME` environment variable to `metadata.name` on the Kubernetes container spec.
 

integration-test/rolling-update-kubernetes/src/test/scala/akka/rollingupdate/kubernetes/KubernetesApiIntegrationTest.scala

@@ -63,7 +63,8 @@ class KubernetesApiIntegrationTest
       enabled = true,
       crName = None,
       cleanupAfter = 60.seconds
-    )
+    ),
+    revisionAnnotation = "deployment.kubernetes.io/revision"
   )
 
   private val underTest =

rolling-update-kubernetes/src/main/resources/reference.conf

@@ -34,6 +34,10 @@ akka.rollingupdate.kubernetes {
     pod-name = ""
     pod-name = ${?KUBERNETES_POD_NAME}
 
+    # Annotations to check to determine the revision.  The default is suitable for "vanilla"
+    # Kubernetes Deployments, but other CI/CD systems may set a different annotation.
+    revision-annotation = "deployment.kubernetes.io/revision"
+
     secure-api-server = true
 
     # Configuration for the Pod Deletion Cost extension

rolling-update-kubernetes/src/main/scala/akka/rollingupdate/kubernetes/KubernetesApiImpl.scala

@@ -55,6 +55,8 @@ import scala.util.control.NonFatal
 
   import system.dispatcher
 
+  override val revisionAnnotation = settings.revisionAnnotation
+
   private implicit val sys: ActorSystem = system
   private val log = Logging(system, classOf[KubernetesApiImpl])
   private val http = Http()(system)
@@ -93,6 +95,7 @@ import scala.util.control.NonFatal
       case HttpResponse(status, _, e, _) =>
         e.discardBytes()
         throw new PodCostException(s"Request failed with status=$status")
+      // Can we make this exhaustive?
     }
   }
 
@@ -213,7 +216,7 @@ PUTs must contain resourceVersions. Response:
         case None              => Future.failed(new ReadRevisionException("No replica name found"))
       }
 
-      val revision = replicaSet.map(_.metadata.annotations.`deployment.kubernetes.io/revision`)
+      val revision = replicaSet.map(_.metadata.annotations.revision)
       revision.map(Some(_)).recoverWith {
         case ex =>
           if (tries >= maxTries) {

rolling-update-kubernetes/src/main/scala/akka/rollingupdate/kubernetes/KubernetesJsonSupport.scala

@@ -10,13 +10,16 @@ import akka.annotation.InternalApi
 import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
 import spray.json.DefaultJsonProtocol
 import spray.json.JsonFormat
+import spray.json.JsObject
 import spray.json.RootJsonFormat
+import spray.json.JsString
+import spray.json.JsValue
 
 /**
  * INTERNAL API
  */
 @InternalApi
-case class ReplicaAnnotation(`deployment.kubernetes.io/revision`: String)
+case class ReplicaAnnotation(revision: String, source: String, otherAnnotations: Map[String, JsValue])
 
 /**
  * INTERNAL API
@@ -75,6 +78,8 @@ case class Spec(pods: immutable.Seq[PodCost])
  */
 @InternalApi
 trait KubernetesJsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
+  val revisionAnnotation: String
+
   // If adding more formats here, remember to also add in META-INF/native-image reflect config
   implicit val metadataFormat: JsonFormat[Metadata] = jsonFormat2(Metadata.apply)
   implicit val podCostFormat: JsonFormat[PodCost] = jsonFormat5(PodCost.apply)
@@ -86,7 +91,30 @@ trait KubernetesJsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
   implicit val podMetadataFormat: JsonFormat[PodMetadata] = jsonFormat1(PodMetadata.apply)
   implicit val podFormat: RootJsonFormat[Pod] = jsonFormat1(Pod.apply)
 
-  implicit val replicaAnnotationFormat: JsonFormat[ReplicaAnnotation] = jsonFormat1(ReplicaAnnotation.apply)
+  implicit val replicaAnnotationFormat: RootJsonFormat[ReplicaAnnotation] =
+    new RootJsonFormat[ReplicaAnnotation] {
+      // Not sure if we ever write this out, but if we do, this will let us write out exactly what we took in
+      def write(ra: ReplicaAnnotation): JsValue =
+        if (ra.revision.nonEmpty && ra.source.nonEmpty) {
+          JsObject(ra.otherAnnotations + (ra.source -> JsString(ra.revision)))
+        } else JsObject(ra.otherAnnotations)
+
+      def read(json: JsValue): ReplicaAnnotation = {
+        json match {
+          case JsObject(fields) =>
+            fields.get(revisionAnnotation) match {
+              case Some(JsString(foundRevision)) =>
+                ReplicaAnnotation(foundRevision, revisionAnnotation, fields - revisionAnnotation)
+
+              case _ =>
+                ReplicaAnnotation("", "", fields)
+            }
+
+          case _ => spray.json.deserializationError("expected an object")
+        }
+      }
+    }
+
   implicit val replicaSetMedatataFormat: JsonFormat[ReplicaSetMetadata] = jsonFormat1(ReplicaSetMetadata.apply)
   implicit val podReplicaSetFormat: RootJsonFormat[ReplicaSet] = jsonFormat1(ReplicaSet.apply)
 }

rolling-update-kubernetes/src/main/scala/akka/rollingupdate/kubernetes/KubernetesSettings.scala

@@ -4,11 +4,13 @@
 
 package akka.rollingupdate.kubernetes
 
-import scala.concurrent.duration._
 import akka.util.JavaDurationConverters._
 import akka.annotation.InternalApi
 import com.typesafe.config.Config
 
+import scala.concurrent.duration._
+import scala.jdk.CollectionConverters.ListHasAsScala
+
 /**
  * INTERNAL API
  */
@@ -47,7 +49,8 @@ private[kubernetes] object KubernetesSettings {
       config.getString("pod-name"),
       config.getBoolean("secure-api-server"),
       config.getDuration("api-service-request-timeout").asScala,
-      customResourceSettings
+      customResourceSettings,
+      config.getString("revision-annotation")
     )
   }
 }
@@ -67,6 +70,7 @@ private[kubernetes] class KubernetesSettings(
     val secure: Boolean,
     val apiServiceRequestTimeout: FiniteDuration,
     val customResourceSettings: CustomResourceSettings,
+    val revisionAnnotation: String,
     val bodyReadTimeout: FiniteDuration = 1.second
 )
 

rolling-update-kubernetes/src/test/scala/akka/rollingupdate/kubernetes/KubernetesApiSpec.scala

@@ -83,7 +83,8 @@ class KubernetesApiSpec
       podName = podName,
       secure = false,
       apiServiceRequestTimeout = 2.seconds,
-      customResourceSettings = new CustomResourceSettings(enabled = false, crName = None, 60.seconds)
+      customResourceSettings = new CustomResourceSettings(enabled = false, crName = None, 60.seconds),
+      revisionAnnotation = "deployment.kubernetes.io/revision"
     )
   }
 
@@ -167,6 +168,42 @@ class KubernetesApiSpec
         }
     }
 
+    "parse pod and replica responses to get the revision from custom annotations" in {
+      stubPodResponse()
+      stubReplicaResponse(defaultReplicaResponseJson.replaceAllLiterally("deployment.kubernetes.io", "custom.akka.io"))
+
+      val customSettings = {
+        val base = settings(podName1)
+        new KubernetesSettings(
+          apiCaPath = base.apiCaPath,
+          apiTokenPath = base.apiTokenPath,
+          apiServiceHost = base.apiServiceHost,
+          apiServicePort = base.apiServicePort,
+          namespace = base.namespace,
+          namespacePath = base.namespacePath,
+          podName = base.podName,
+          secure = base.secure,
+          apiServiceRequestTimeout = base.apiServiceRequestTimeout,
+          customResourceSettings = base.customResourceSettings,
+          revisionAnnotation = "custom.akka.io/revision"
+        )
+      }
+
+      val customKubernetesApi =
+        new KubernetesApiImpl(
+          system,
+          customSettings,
+          namespace,
+          apiToken = "apiToken",
+          clientHttpsConnectionContext = None)
+
+      EventFilter
+        .info(pattern = "Reading revision from Kubernetes: akka.cluster.app-version was set to 1", occurrences = 1)
+        .intercept {
+          customKubernetesApi.readRevision().futureValue should be("1")
+        }
+    }
+
     "retry and then fail when pod not found" in {
       stubFor(getPod(podName1).willReturn(aResponse().withStatus(404)))
       EventFilter

rolling-update-kubernetes/src/test/scala/akka/rollingupdate/kubernetes/PodDeletionCostAnnotatorCrSpec.scala

@@ -115,7 +115,8 @@ class PodDeletionCostAnnotatorCrSpec
       podName = podName,
       secure = false,
       apiServiceRequestTimeout = 2.seconds,
-      customResourceSettings = new CustomResourceSettings(enabled = false, crName = None, 60.seconds)
+      customResourceSettings = new CustomResourceSettings(enabled = false, crName = None, 60.seconds),
+      revisionAnnotation = "deployment.kubernetes.io/revision"
     )
   }
 

rolling-update-kubernetes/src/test/scala/akka/rollingupdate/kubernetes/PodDeletionCostAnnotatorSpec.scala

@@ -89,7 +89,8 @@ class PodDeletionCostAnnotatorSpec
       podName = podName,
       secure = false,
       apiServiceRequestTimeout = 2.seconds,
-      customResourceSettings = new CustomResourceSettings(enabled = false, crName = None, 60.seconds)
+      customResourceSettings = new CustomResourceSettings(enabled = false, crName = None, 60.seconds),
+      revisionAnnotation = "deployment.kubernetes.io/revision"
     )
   }