From 9b2ed1931001bf7ed171af2b42d505fe5ac20551 Mon Sep 17 00:00:00 2001
From: Andrea Peruffo <andrea.peruffo1982@gmail.com>
Date: Tue, 31 Aug 2021 17:25:32 +0100
Subject: [PATCH] Specify the operator namespace to constrain CLI's requests

---
 .../src/main/scala/akka/cli/cloudflow/Commands.scala   | 10 ++++++++++
 .../cli/cloudflow/execution/ConfigureExecution.scala   |  2 +-
 .../akka/cli/cloudflow/execution/DeployExecution.scala |  4 ++--
 3 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/Commands.scala b/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/Commands.scala
index 311aef0ce..6018a4447 100644
--- a/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/Commands.scala
+++ b/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/Commands.scala
@@ -215,6 +215,10 @@ object OptionsParser {
         commandParse[commands.Deploy, String](opt("serviceaccount"))((c, sa) => c.copy(serviceAccount = Some(sa)))
           .optional()
           .text("the serviceaccount to be used"),
+        commandParse[commands.Deploy, String](opt("operator-namespace"))((c, ons) =>
+          c.copy(operatorNamespace = Some(ons)))
+          .optional()
+          .text("the namespace where the operator is deployed"),
         commandCheck[commands.Deploy](d => {
           if (d.logbackConfig.isDefined && !d.logbackConfig.get.exists()) {
             failure("the provided logback configuration file doesn't exist")
@@ -349,6 +353,10 @@ object OptionsParser {
         commandParse[commands.Configure, File](opt("logback-config"))((c, f) => c.copy(logbackConfig = Some(f)))
           .optional()
           .text("the logback configuration to be applied"),
+        commandParse[commands.Configure, String](opt("operator-namespace"))((c, ons) =>
+          c.copy(operatorNamespace = Some(ons)))
+          .optional()
+          .text("the namespace where the operator is deployed"),
         commandCheck[commands.Configure](c => {
           if (c.logbackConfig.isDefined && !c.logbackConfig.get.exists()) {
             failure("the provided logback configuration file doesn't exist")
@@ -487,6 +495,7 @@ object commands {
   case class Deploy(
       crFile: File = new File(""),
       namespace: Option[String] = None,
+      operatorNamespace: Option[String] = None,
       dockerUsername: String = "",
       dockerPassword: String = "",
       noRegistryCredentials: Boolean = false,
@@ -537,6 +546,7 @@ object commands {
   case class Configure(
       cloudflowApp: String = "",
       namespace: Option[String] = None,
+      operatorNamespace: Option[String] = None,
       confs: Seq[File] = Seq(),
       configKeys: Map[String, String] = Map(),
       logbackConfig: Option[File] = None,
diff --git a/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/ConfigureExecution.scala b/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/ConfigureExecution.scala
index 647cf182a..8f3ee9637 100644
--- a/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/ConfigureExecution.scala
+++ b/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/ConfigureExecution.scala
@@ -36,7 +36,7 @@ final case class ConfigureExecution(c: Configure, client: KubeClient, logger: Cl
       streamletsConfigs <- streamletsConfigs(
         currentCr,
         cloudflowConfig,
-        () => client.getKafkaClusters(None).map(parseValues))
+        () => client.getKafkaClusters(namespace = c.operatorNamespace).map(parseValues))
 
       uid <- client.uidCloudflowApp(currentCr.spec.appId, namespace)
       _ <- client.configureCloudflowApp(
diff --git a/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/DeployExecution.scala b/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/DeployExecution.scala
index 1e6a19194..5ab846181 100644
--- a/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/DeployExecution.scala
+++ b/core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/DeployExecution.scala
@@ -190,11 +190,11 @@ final case class DeployExecution(d: Deploy, client: KubeClient, logger: CliLogge
       // validate the Cr against the cluster
       _ <- referencedKafkaSecretExists(
         applicationCr,
-        () => client.getKafkaClusters(namespace = Some(namespace)).map(_.keys.toList))
+        () => client.getKafkaClusters(namespace = d.operatorNamespace).map(_.keys.toList))
 
       // streamlets configurations
       streamletsConfigs <- streamletsConfigs(applicationCr, cloudflowConfig, () => {
-        client.getKafkaClusters(None).map(parseValues)
+        client.getKafkaClusters(namespace = d.operatorNamespace).map(parseValues)
       })
 
       // Operations on the cluster