Specify the operator namespace to constrain CLI's requests

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,

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(

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