Encode the protocol version in the CRD

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/Cli.scala

@@ -16,9 +16,7 @@ object Cli {
   def defaultKubeClient(config: Option[File], logger: CliLogger) =
     new KubeClientFabric8(config)(logger)
 
-  val ProtocolVersion = "6"
-
-  val SupportedApplicationDescriptorVersion = 6
+  val ApplicationDescriptorVersion = 6
 
   val RequiredSparkVersion = "v1beta2"
 

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/ConfigurationExecution.scala

@@ -16,7 +16,7 @@ final case class ConfigurationExecution(c: Configuration, client: KubeClient, lo
   def run(): Try[ConfigurationResult] = {
     logger.info("Executing command Configuration")
     for {
-      _ <- validateProtocolVersion(client)
+      _ <- validateProtocolVersion(client, logger)
       res <- client.getAppInputSecret(c.cloudflowApp, c.namespace.getOrElse(c.cloudflowApp))
       config <- Try { ConfigFactory.parseString(res) }.recover {
         case ex => throw CliException("Failed to parse the current configuration", ex)

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/ConfigureExecution.scala

@@ -17,7 +17,7 @@ final case class ConfigureExecution(c: Configure, client: KubeClient, logger: Cl
   def run(): Try[ConfigureResult] = {
     logger.info("Executing command Configure")
     for {
-      _ <- validateProtocolVersion(client)
+      _ <- validateProtocolVersion(client, logger)
       namespace = c.namespace.getOrElse(c.cloudflowApp)
 
       currentCr <- client.readCloudflowApp(c.cloudflowApp, namespace).map {

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/DeployExecution.scala

@@ -53,10 +53,10 @@ final case class DeployExecution(d: Deploy, client: KubeClient, logger: CliLogge
           lazy val lvMsg = s"built with sbt-cloudflow version ${libraryVersion},"
 
           version match {
-            case v if Cli.SupportedApplicationDescriptorVersion > v =>
+            case v if Cli.ApplicationDescriptorVersion > v =>
               Failure(CliException(
                 s"Application ${lvMsg} is incompatible and requires a newer version of the kubectl cloudflow plugin. Please upgrade and try again"))
-            case v if Cli.SupportedApplicationDescriptorVersion < v =>
+            case v if Cli.ApplicationDescriptorVersion < v =>
               Failure(CliException(
                 s"Application ${lvMsg} is incompatible and no longer supported. Please upgrade sbt-cloudflow and rebuild the application with 'sbt buildApp'"))
             case _ => Success(())
@@ -146,7 +146,7 @@ final case class DeployExecution(d: Deploy, client: KubeClient, logger: CliLogge
     logger.info("Executing command Deploy")
     for {
       // Default protocol validation
-      _ <- validateProtocolVersion(client)
+      _ <- validateProtocolVersion(client, logger)
 
       // prepare the data
       baseApplicationCr <- loadCrFile(d.crFile)

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/ListExecution.scala

@@ -16,7 +16,7 @@ final case class ListExecution(l: List, client: KubeClient, logger: CliLogger)
   def run(): Try[ListResult] = {
     logger.info("Executing command List")
     for {
-      _ <- validateProtocolVersion(client)
+      _ <- validateProtocolVersion(client, logger)
       res <- client.listCloudflowApps(l.namespace)
     } yield {
       ListResult(res)

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/ScaleExecution.scala

@@ -17,7 +17,7 @@ final case class ScaleExecution(s: Scale, client: KubeClient, logger: CliLogger)
   def run(): Try[ScaleResult] = {
     logger.info("Executing command Status")
     for {
-      _ <- validateProtocolVersion(client)
+      _ <- validateProtocolVersion(client, logger)
       namespace = s.namespace.getOrElse(s.cloudflowApp)
 
       currentAppCrOpt <- client.readCloudflowApp(s.cloudflowApp, namespace)

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/StatusExecution.scala

@@ -16,7 +16,7 @@ final case class StatusExecution(s: Status, client: KubeClient, logger: CliLogge
   def run(): Try[StatusResult] = {
     logger.info("Executing command Status")
     for {
-      _ <- validateProtocolVersion(client)
+      _ <- validateProtocolVersion(client, logger)
       res <- client.getCloudflowAppStatus(s.cloudflowApp, s.namespace.getOrElse(s.cloudflowApp))
     } yield {
       StatusResult(res)

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/UndeployExecution.scala

@@ -16,7 +16,7 @@ final case class UndeployExecution(u: Undeploy, client: KubeClient, logger: CliL
   def run(): Try[UndeployResult] = {
     logger.info("Executing command Undeploy")
     for {
-      _ <- validateProtocolVersion(client)
+      _ <- validateProtocolVersion(client, logger)
       _ <- client.deleteCloudflowApp(u.cloudflowApp, u.namespace.getOrElse(u.cloudflowApp))
     } yield {
       UndeployResult()

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/UpdateCredentialsExecution.scala

@@ -16,7 +16,7 @@ final case class UpdateCredentialsExecution(u: UpdateCredentials, client: KubeCl
   def run(): Try[UpdateCredentialsResult] = {
     logger.info("Executing command UpdateCredentials")
     for {
-      _ <- validateProtocolVersion(client)
+      _ <- validateProtocolVersion(client, logger)
       namespace = u.namespace.getOrElse(u.cloudflowApp)
       _ <- client.createNamespace(namespace)
       _ <- client.createImagePullSecret(

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/WithProtocolVersion.scala

@@ -4,22 +4,25 @@
 
 package akka.cli.cloudflow.execution
 
-import akka.cli.cloudflow.{ Cli, CliException }
+import akka.datap.crd.App
+
+import akka.cli.cloudflow.{ CliException, CliLogger }
 import akka.cli.cloudflow.kubeclient.KubeClient
 
 import scala.util.{ Failure, Success, Try }
 
 trait WithProtocolVersion {
 
-  def validateProtocolVersion(client: KubeClient): Try[String] = {
+  def validateProtocolVersion(client: KubeClient, logger: CliLogger): Try[String] = {
     (for {
       version <- client.getOperatorProtocolVersion()
     } yield {
+      logger.info(s"Protocol version found: $version")
       version match {
-        case v if v == Cli.ProtocolVersion =>
+        case v if v == App.ProtocolVersion =>
           Success(version)
         case ver =>
-          val pVersion = Integer.parseInt(Cli.ProtocolVersion)
+          val pVersion = Integer.parseInt(App.ProtocolVersion)
           Integer.parseInt(ver) match {
             case v if pVersion > v =>
               Failure(CliException(

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/kubeclient/KubeClient.scala

@@ -13,10 +13,6 @@ object KubeClient {
 
   val CloudflowResource = App.ResourceName
 
-  val CloudflowProtocolVersionConfigMap = "cloudflow-protocol-version"
-
-  val ProtocolVersionKey = "protocol-version"
-
   val SparkResource = "sparkapplications.sparkoperator.k8s.io"
 
   val FlinkResource = "flinkapplications.flink.k8s.io"

core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/kubeclient/KubeClientFabric8.scala

@@ -211,25 +211,16 @@ class KubeClientFabric8(
 
   def getOperatorProtocolVersion(): Try[String] = withClient { client =>
     for {
-      protocolVersionCM <- Try {
+      crd <- Try {
         client
-          .configMaps()
-          .inAnyNamespace()
-          .withLabel(KubeClient.CloudflowProtocolVersionConfigMap)
-          .list()
-          .getItems()
-      }
-      protocolVersion <- {
-        protocolVersionCM.size() match {
-          case 1 => Success(protocolVersionCM.get(0))
-          case x if x > 1 =>
-            Failure(
-              CliException("Multiple Cloudflow operators detected in the cluster. This is not supported. Exiting"))
-          case x if x < 1 => Failure(CliException("No Cloudflow operators detected in the cluster. Exiting"))
-        }
+          .apiextensions()
+          .v1beta1()
+          .customResourceDefinitions()
+          .withName(App.ResourceName)
+          .get()
       }
-      version <- Option(protocolVersion.getData.get(KubeClient.ProtocolVersionKey))
-        .fold[Try[String]](Failure(CliException("Cannot find the protocol version in the config map")))(Success(_))
+      version <- Option(crd.getMetadata.getLabels.get(App.ProtocolVersionKey))
+        .fold[Try[String]](Failure(CliException("Cannot find the protocol version in the CRD")))(Success(_))
     } yield {
       version
     }

core/cloudflow-cli/src/test/scala/akka/cli/cloudflow/CliWorkflowSpec.scala

@@ -29,7 +29,7 @@ class CliWorkflowSpec extends AnyFlatSpec with Matchers with TryValues {
   val defaultProvidedKafkaClusters = Map("default" -> """bootstrap.servers = "localhost:9092"""")
 
   @nowarn def testingKubeClientFactory(
-      protocolVersion: String = Cli.ProtocolVersion,
+      protocolVersion: String = App.ProtocolVersion,
       sparkVersion: String = Cli.RequiredSparkVersion,
       flinkVersion: String = Cli.RequiredFlinkVersion,
       providedPvcs: List[String] = defaultPvcMounts,

core/cloudflow-crd/kubernetes/cloudflow-crd.yaml

@@ -2,6 +2,8 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   name: cloudflowapplications.cloudflow.lightbend.com
+  labels:
+    protocol-version: "7"
 spec:
   group: cloudflow.lightbend.com
   names:

core/cloudflow-crd/src/main/scala/akka/datap/crd/App.scala

@@ -46,6 +46,9 @@ object App {
 
   final val Scope = "Namespaced"
 
+  final val ProtocolVersionKey = "protocol-version"
+  final val ProtocolVersion = "7"
+
   val customResourceDefinitionContext: CustomResourceDefinitionContext =
     new CustomResourceDefinitionContext.Builder()
       .withVersion(GroupVersion)

core/cloudflow-operator/src/main/scala/cloudflow/operator/Main.scala

@@ -28,7 +28,7 @@ import io.fabric8.kubernetes.client.{ Config, DefaultKubernetesClient, Kubernete
 
 import java.lang.management.ManagementFactory
 import scala.jdk.CollectionConverters._
-import scala.util.Try
+import scala.util.{ Success, Try }
 
 object Main extends {
 
@@ -61,7 +61,6 @@ object Main extends {
       checkCRD(settings, client)
 
       val ownerReferences = getDeploymentOwnerReferences(settings, client)
-      installProtocolVersion(settings, client, ownerReferences)
 
       import cloudflow.operator.action.runner._
       val flinkRunner = {
@@ -152,24 +151,25 @@ object Main extends {
         .get()) match {
       case Some(crd) if crd.getSpec.getVersion == App.GroupVersion =>
         system.log.info(s"CRD found at version ${App.GroupVersion}")
+        Try(crd.getMetadata.getLabels.get(App.ProtocolVersionKey)) match {
+          case Success(protocolVersion) =>
+            if (protocolVersion != App.ProtocolVersion) {
+              system.log.error(
+                s"The CRD is incompatible the current version of the Cloudflow operator. (CRD protocol version: $protocolVersion), operator protocol version: ${App.ProtocolVersion}")
+              throw new Exception("Protocol version mismatch")
+            }
+          case _ =>
+            system.log.error(
+              "The installed CRD doesn't contain the protocol version label, please upgrade it to the same version as the operator")
+            throw new Exception("Protocol version not found")
+        }
       case _ =>
         system.log.error(
           s"Cloudflow CRD not found, please install it: 'kubectl apply -f https://raw.githubusercontent.com/lightbend/cloudflow/v${BuildInfo.version}/core/cloudflow-crd/kubernetes/cloudflow-crd.yaml'")
         throw new Exception("Cloudflow CRD not found")
     }
   }
 
-  private def installProtocolVersion(
-      settings: Settings,
-      client: KubernetesClient,
-      ownerReferences: List[OwnerReference]): Unit = {
-    client
-      .configMaps()
-      .inNamespace(settings.podNamespace)
-      .withName(Operator.ProtocolVersionConfigMapName)
-      .createOrReplace(Operator.ProtocolVersionConfigMap(ownerReferences))
-  }
-
   private def getGCInfo: List[(String, javax.management.ObjectName)] = {
     val gcMxBeans = ManagementFactory.getGarbageCollectorMXBeans()
     gcMxBeans.asScala.map(b => (b.getName, b.getObjectName)).toList

core/cloudflow-operator/src/main/scala/cloudflow/operator/Operator.scala

@@ -47,20 +47,6 @@ import scala.util._
 object Operator {
   lazy val log = LoggerFactory.getLogger("Operator")
 
-  val ProtocolVersion = "6"
-  val ProtocolVersionKey = "protocol-version"
-  val ProtocolVersionConfigMapName = "cloudflow-protocol-version"
-  def ProtocolVersionConfigMap(ownerReferences: List[OwnerReference]) = {
-    new ConfigMapBuilder()
-      .withNewMetadata()
-      .withName(ProtocolVersionConfigMapName)
-      .withLabels((Map(ProtocolVersionConfigMapName -> ProtocolVersionConfigMapName)).asJava)
-      .withOwnerReferences(ownerReferences: _*)
-      .endMetadata()
-      .withData(Map(ProtocolVersionKey -> ProtocolVersion).asJava)
-      .build()
-  }
-
   val AppIdLabel = "com.lightbend.cloudflow/app-id"
   val ConfigFormatLabel = "com.lightbend.cloudflow/config-format"
   val StreamletNameLabel = "com.lightbend.cloudflow/streamlet-name"