Add a flag to disable Flink in the operator

tools/cloudflow-operator/src/main/resources/application.conf

@@ -11,6 +11,8 @@ cloudflow {
     pod-namespace = "cloudflow"
     pod-namespace = ${?POD_NAMESPACE}
 
+    flink-enabled = true
+
     api {
       # API version for client compatibility checks
       compatibility-version = "1"

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

@@ -64,10 +64,18 @@ object Main extends {
       installProtocolVersion(settings, client, ownerReferences)
 
       import cloudflow.operator.action.runner._
+      val flinkRunner = {
+        if (settings.flinkEnabled) {
+          Map(FlinkRunner.Runtime -> new FlinkRunner(ctx.flinkRunnerDefaults))
+        } else {
+          Map.empty
+        }
+      }
+
       val runners = Map(
-        AkkaRunner.Runtime -> new AkkaRunner(ctx.akkaRunnerDefaults),
-        SparkRunner.Runtime -> new SparkRunner(ctx.sparkRunnerDefaults),
-        FlinkRunner.Runtime -> new FlinkRunner(ctx.flinkRunnerDefaults))
+          AkkaRunner.Runtime -> new AkkaRunner(ctx.akkaRunnerDefaults),
+          SparkRunner.Runtime -> new SparkRunner(ctx.sparkRunnerDefaults)) ++ flinkRunner
+
       Operator.handleEvents(client, runners, ctx.podName, ctx.podNamespace)
     } catch {
       case t: Throwable =>

tools/cloudflow-operator/src/main/scala/cloudflow/operator/Settings.scala

@@ -150,6 +150,8 @@ final case class Settings(config: Config) extends Extension {
   val sparkRunnerSettings = getSparkRunnerDefaults(config, root, SparkRunner.Runtime)
   val flinkRunnerSettings = getFlinkRunnerDefaults(config, root, FlinkRunner.Runtime)
 
+  val flinkEnabled = config.getBoolean(s"$root.flink-enabled")
+
   val api = ApiSettings(getNonEmptyString(config, s"$root.api.bind-interface"), getPort(config, s"$root.api.bind-port"))
 
   val deploymentContext = {

tools/cloudflow-operator/src/main/scala/cloudflow/operator/action/CloudflowStatus.scala

@@ -18,7 +18,7 @@ package cloudflow.operator.action
 
 import akka.datap.crd.App
 import akka.kube.actions.{ Action, CustomResourceAdapter }
-import cloudflow.operator.action.runner.Runner
+import cloudflow.operator.action.runner.{ FlinkRunner, Runner }
 import io.fabric8.kubernetes.api.model.{ ContainerState, Pod }
 import io.fabric8.kubernetes.api.{ model => fabric8 }
 import io.fabric8.kubernetes.client.KubernetesClient
@@ -46,8 +46,10 @@ object CloudflowStatus {
     val ReadyFalse = "False"
   }
 
-  private def podReady(ps: App.PodStatus) = {
-    ps.status == PodStatus.Running && ps.nrOfContainersReady == ps.nrOfContainers && ps.nrOfContainers > 0
+  private def podReady(ps: App.PodStatus, expectedPodCount: Option[Int]) = {
+    (ps.status == PodStatus.Running && ps.nrOfContainersReady == ps.nrOfContainers && ps.nrOfContainers > 0) || (expectedPodCount
+      .map(_ == 0)
+      .getOrElse(false) && ps.nrOfContainers == 0 && ps.nrOfContainersReady == 0)
   }
 
   private def podStatus(
@@ -102,7 +104,7 @@ object CloudflowStatus {
       nrOfContainers = nrOfContainers)
   }
 
-  private def getStatusFromContainerStates(containerStates: List[ContainerState], nrOfContainers: Int): String =
+  private def getStatusFromContainerStates(containerStates: List[ContainerState], nrOfContainers: Int): String = {
     if (containerStates.nonEmpty) {
       // - Running if all containers running;
       // - Terminated if all containers terminated;
@@ -118,9 +120,12 @@ object CloudflowStatus {
           .getOrElse(PodStatus.Pending)
       }
     } else PodStatus.Pending
+  }
 
   private def hasExpectedPods(streamlet: App.StreamletStatus)(nrOfPodsDetected: Int) =
-    streamlet.expectedPodCount.getOrElse(0) == nrOfPodsDetected
+    streamlet.expectedPodCount.getOrElse(0) == nrOfPodsDetected || streamlet.expectedPodCount
+      .map(_ == 0)
+      .getOrElse(false)
 
   private def updatePod(streamlet: App.StreamletStatus)(pod: Pod) = {
     val podStatus = fromPod(pod)
@@ -170,17 +175,31 @@ object CloudflowStatus {
 
   private def createStreamletStatuses(spec: App.Spec, runners: Map[String, Runner[_]]) =
     spec.deployments.map { deployment =>
-      val expectedPodCount = runners.get(deployment.runtime).map(_.expectedPodCount(deployment)).getOrElse(1)
-      App.StreamletStatus(
-        streamletName = deployment.streamletName,
-        expectedPodCount = Some(expectedPodCount),
-        podStatuses = Nil)
+      if (deployment.runtime == FlinkRunner.Runtime && !runners.contains(FlinkRunner.Runtime)) {
+        App.StreamletStatus(
+          streamletName = deployment.streamletName,
+          expectedPodCount = Some(0),
+          podStatuses = Seq(
+            App.PodStatus(
+              name = "<external>",
+              ready = PodStatus.ReadyTrue,
+              nrOfContainersReady = 0,
+              nrOfContainers = 0,
+              restarts = 0,
+              status = PodStatus.Unknown)))
+      } else {
+        val expectedPodCount = runners.get(deployment.runtime).map(_.expectedPodCount(deployment)).getOrElse(1)
+        App.StreamletStatus(
+          streamletName = deployment.streamletName,
+          expectedPodCount = Some(expectedPodCount),
+          podStatuses = Nil)
+      }
     }.toVector
 
   private def calcAppStatus(streamletStatuses: Seq[App.StreamletStatus]): String = {
     if (streamletStatuses.forall { streamletStatus =>
           hasExpectedPods(streamletStatus)(streamletStatus.podStatuses.size) &&
-          streamletStatus.podStatuses.forall(podReady)
+          streamletStatus.podStatuses.forall(p => podReady(p, streamletStatus.expectedPodCount))
         }) {
       Status.Running
     } else if (streamletStatuses.flatMap(_.podStatuses).exists(_.status == PodStatus.CrashLoopBackOff)) {

tools/cloudflow-operator/src/test/scala/cloudflow/operator/action/AppCrSpec.scala

@@ -189,6 +189,20 @@ class AppCrSpec
       CloudflowStatus.aggregatedStatus(status) mustBe CloudflowStatus.Status.Running
     }
 
+    "report its status as Running when all streamlet pods but flink are running and ready in a mixed app" in {
+      var status = mkTestStatusExternalFlinkApp()
+      status = CloudflowStatus.updatePod(status)("ingress", mkRunningReadyPod("ingress"))
+
+      CloudflowStatus.aggregatedStatus(status) mustBe CloudflowStatus.Status.Pending
+      (1 to SparkRunner.DefaultNrOfExecutorInstances).foreach { _ =>
+        status = CloudflowStatus.updatePod(status)("spark-egress", mkRunningReadyPod("spark-egress"))
+        CloudflowStatus.aggregatedStatus(status) mustBe CloudflowStatus.Status.Pending
+      }
+      status = CloudflowStatus.updatePod(status)("spark-egress", mkRunningReadyPod("spark-egress"))
+
+      CloudflowStatus.aggregatedStatus(status) mustBe CloudflowStatus.Status.Running
+    }
+
     "report pod status as Running" in {
       val podStatus = CloudflowStatus.fromPod(mkRunningReadyPod("s1"))
       podStatus.status mustBe CloudflowStatus.PodStatus.Running
@@ -272,6 +286,30 @@ class AppCrSpec
     CloudflowStatus.freshStatus(newApp, runners)
   }
 
+  def mkTestStatusExternalFlinkApp() = {
+    val ingress = randomStreamlet("akka").asIngress[Foo].withServerAttribute
+    val sparkEgress = randomStreamlet("spark").asEgress[Foo]
+    val flinkEgress = randomStreamlet("flink").asEgress[Foo]
+
+    val ingressRef = ingress.ref("ingress")
+    val sparkEgressRef = sparkEgress.ref("spark-egress")
+    val flinkEgressRef = flinkEgress.ref("flink-egress")
+
+    val verifiedBlueprint = Blueprint()
+      .define(Vector(ingress, sparkEgress, flinkEgress))
+      .use(ingressRef)
+      .use(sparkEgressRef)
+      .use(flinkEgressRef)
+      .connect(Topic("foos1"), ingressRef.out, sparkEgressRef.in)
+      .connect(Topic("foos2"), flinkEgressRef.in)
+      .verified
+      .right
+      .value
+
+    val newApp = mkApp(verifiedBlueprint)
+    CloudflowStatus.freshStatus(newApp, runners.filter { case (k, _) => k != "flink" })
+  }
+
   def mkApp(verifiedBlueprint: VerifiedBlueprint) =
     CloudflowApplicationSpecBuilder.create(appId, appVersion, image, verifiedBlueprint, agentPaths)