[Merged by Bors] - Upgrade fix

.github/workflows/cd_dev.yaml

@@ -123,19 +123,10 @@ jobs:
         with:
           minikube version: "v1.22.0"
           kubernetes version: "v1.21.2"
-      - name: Setup Minikube for Linux
-        if: startsWith(matrix.os, 'infinyon-ubuntu')
-        run: |
-          minikube delete
-          minikube start --driver=docker --kubernetes-version 1.21.2
       - name: Test minikube
         run: |
           minikube profile list
           minikube status
-      - name: Setup for upgrade test
-        run: |
-          curl -fsS https://packages.fluvio.io/v1/install.sh | bash
-
       - name: Run upgrade test
         env:
           TEST_DATA_BYTES: 10000
@@ -144,13 +135,10 @@ jobs:
           timeout_minutes: 45
           max_attempts: 3
           command: |
+            ./k8-util/cluster/reset-minikube.sh 
             export PATH=~/.fluvio/bin:$PATH
             USE_LATEST=true make upgrade-test
 
-      - name: Clean minikube
-        run: |
-          minikube delete
-
       - name: Slack Notification
         uses: 8398a7/action-slack@v3
         if: failure()

.github/workflows/ci.yml

@@ -489,7 +489,7 @@ jobs:
 
   k8_upgrade_test:
     name: Upgrade cluster test
-    if: ${{ false }} # Disable job for CI
+    if: ${{ false }}
     needs: build_image
     runs-on: ${{ matrix.os }}
     strategy:
@@ -533,10 +533,6 @@ jobs:
       - name: Print artifacts and mark executable
         run: ls -la . && chmod +x ./fluvio && ./fluvio version
 
-      - name: Setup for upgrade test
-        run: |
-          curl -fsS https://packages.fluvio.io/v1/install.sh | bash
-
       - name: Run upgrade test with CI artifacts
         env:
           TEST_DATA_BYTES: 10000
@@ -545,6 +541,7 @@ jobs:
           timeout_minutes: 15
           max_attempts: 3
           command: |
+            ./k8-util/cluster/reset-minikube.sh 
             export PATH=~/.fluvio/bin:$PATH
             make FLUVIO_BIN=./fluvio upgrade-test
 
@@ -589,7 +586,7 @@ jobs:
       - build_binaries
       - local_cluster_test
       - k8_cluster_test
-      #- k8_upgrade_test
+      # - k8_upgrade_test
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2

CHANGELOG.md

@@ -14,6 +14,7 @@
 * Remove `#[deprecated]` items from crates ([#1299](https://github.com/infinyon/fluvio/issues/1299))
 * Bump `MINIMUM_PLATFORM_VERSION` to `0.9.0` ([#1310](https://github.com/infinyon/fluvio/issues/1310))
 * Fix owner reference type to work delete in K 1.20.0 ([#1342](https://github.com/infinyon/fluvio/issues/1342))
+* Fix Upgrading K8 Cluster ([#1347](https://github.com/infinyon/fluvio/issues/1347))
 
 ## Platform Version 0.8.5 - 2021-07-14
 * Add unstable Admin Watch API for topics, partitions, and SPUs ([#1136](https://github.com/infinyon/fluvio/pull/1136))

Cargo.toml

@@ -37,6 +37,7 @@ resolver = "2"
 [patch.crates-io]
 k8-types = { git = "https://github.com/infinyon/k8-api" }
 
+
 # Used to make eyre faster on debug builds
 # See https://github.com/yaahc/color-eyre#improving-perf-on-debug-builds
 [profile.dev.package.backtrace]

DEVELOPER.md

@@ -165,6 +165,10 @@ We highly recommend using the `flvd cluster start --local --develop` command for
 
 However, in the following cases, we run `sc` and `spu` individually, allowing individual testing.
 
+## Filtering Log
+
+Please see [filering tracing log](https://tracing.rs/tracing_subscriber/filter/struct.envfilter).
+
 ## Starting SC
 
 The Streaming Controller (SC) is the controller for a Fluvio cluster.
@@ -308,6 +312,8 @@ Note that when you uninstall cluster, CLI will remove all related objects such a
 
 ## Running SC in locally
 
+
+
 First install fluvio k8 cluster as normally.
 
 Then delete deployment:

k8-util/DEVELOP.md

@@ -0,0 +1,31 @@
+## Trace K8 API 
+
+Set up Local Proxy to minikube.  It's easy to confingure for other K8 cluster as well:
+
+```
+kubectl proxy
+Starting to serve on 127.0.0.1:8001
+```
+Add proxy cluster configuration to ~/.kube/config
+```
+- cluster:
+    server: http://localhost:8001
+  name: proxy
+- context:
+    cluster: proxy
+    user: minikube
+  name: proxy
+```
+
+Change kubectl to use proxy:
+```
+kubectl config use-context proxy
+```
+
+Use tcpdump to inspect:
+```
+ sudo tcpdump -i any  port 8001 -A
+```
+
+
+

k8-util/cluster/reset-minikube.sh

@@ -5,4 +5,5 @@ set -e
 ARG1=${1:-docker}
 K8_VERSION=${2:-1.21.2}
 minikube delete
-minikube start --driver $ARG1 --kubernetes-version=$K8_VERSION
+minikube start --driver $ARG1 --kubernetes-version=$K8_VERSION
+# minikube start --extra-config=apiserver.v=10

src/sc/Cargo.toml

@@ -36,7 +36,7 @@ async-lock = "2.3.0"
 async-channel = "1.4.0"
 event-listener = "2.2.0"
 tokio = { version = "1.3.0", features = ["macros"] }
-structopt = "0.3.17"
+structopt = "0.3.22"
 semver = "1.0.0"
 once_cell = "1.5"
 cfg-if = { version = "1.0.0" }

src/sc/src/cli.rs

@@ -64,6 +64,10 @@ pub struct ScOpt {
         env
     )]
     auth_policy: Option<PathBuf>,
+
+    /// only allow white list of controllers
+    #[structopt(long)]
+    white_list: Vec<String>,
 }
 
 impl ScOpt {
@@ -110,6 +114,7 @@ impl ScOpt {
 
         config.namespace = self.namespace.unwrap();
         config.x509_auth_scopes = self.x509_auth_scopes;
+        config.white_list = self.white_list.into_iter().collect();
 
         // Set Configuration Authorzation Policy
         let policy = match self.auth_policy {

src/sc/src/config/mod.rs

@@ -1,4 +1,13 @@
 mod sc_config;
 
 pub use self::sc_config::ScConfig;
+
 pub use self::sc_config::ScConfigBuilder;
+
+macro_rules! whitelist {
+    ($config:expr,$name:expr,$start:expr) => {
+        if $config.enabled($name) {
+            $start;
+        }
+    };
+}

src/sc/src/config/sc_config.rs

@@ -3,6 +3,7 @@
 //!
 //! Stores configuration parameter used by Streaming Controller module.
 //!
+use std::collections::HashSet;
 use std::{io::Error as IoError, path::PathBuf};
 
 use fluvio_types::defaults::SC_PUBLIC_PORT;
@@ -25,6 +26,7 @@ pub struct ScConfig {
     pub run_k8_dispatchers: bool,
     pub namespace: String,
     pub x509_auth_scopes: Option<PathBuf>,
+    pub white_list: HashSet<String>,
 }
 
 impl ::std::default::Default for ScConfig {
@@ -35,6 +37,20 @@ impl ::std::default::Default for ScConfig {
             run_k8_dispatchers: true,
             namespace: "default".to_owned(),
             x509_auth_scopes: None,
+            white_list: HashSet::new(),
+        }
+    }
+}
+
+impl ScConfig {
+    /// check if white list is enabled for controller name
+    /// if white list is empty then everything is enabled,
+    /// otherwise only included
+    pub fn enabled(&self, name: &str) -> bool {
+        if self.white_list.is_empty() {
+            true
+        } else {
+            self.white_list.contains(name)
         }
     }
 }

src/sc/src/init.rs

@@ -14,7 +14,7 @@ use crate::core::SharedContext;
 use crate::controllers::spus::SpuController;
 use crate::controllers::topics::TopicController;
 use crate::controllers::partitions::PartitionController;
-use crate::config::ScConfig;
+use crate::config::{ScConfig};
 use crate::services::start_internal_server;
 use crate::dispatcher::dispatcher::K8ClusterStateDispatcher;
 use crate::services::auth::basic::BasicRbacPolicy;
@@ -37,6 +37,7 @@ where
 
     let namespace = sc_config.namespace.clone();
     let ctx = Context::shared_metadata(sc_config);
+    let config = ctx.config();
 
     K8ClusterStateDispatcher::<SpuSpec, C>::start(
         namespace.clone(),
@@ -62,13 +63,16 @@ where
         ctx.spgs().clone(),
     );
 
-    SpuController::start(ctx.clone());
-    TopicController::start(ctx.clone());
-    PartitionController::start(ctx.clone());
+    whitelist!(config, "spu", SpuController::start(ctx.clone()));
+    whitelist!(config, "topic", TopicController::start(ctx.clone()));
+    whitelist!(config, "partition", PartitionController::start(ctx.clone()));
 
-    start_internal_server(ctx.clone());
-
-    pub_server::start(ctx.clone(), auth_policy);
+    whitelist!(config, "internal", start_internal_server(ctx.clone()));
+    whitelist!(
+        config,
+        "public",
+        pub_server::start(ctx.clone(), auth_policy)
+    );
 
     mod pub_server {
 

src/sc/src/k8/controllers/mod.rs

@@ -1,4 +1,4 @@
-pub mod spg;
+pub mod spg_stateful;
 pub mod spu_service;
 pub mod spu;
 
@@ -17,7 +17,7 @@ mod k8_operator {
     use crate::k8::objects::spu_service::SpuServiceSpec;
     use crate::k8::objects::statefulset::StatefulsetSpec;
     use crate::k8::objects::spg_service::SpgServiceSpec;
-    use crate::k8::controllers::spg::SpgStatefulSetController;
+    use crate::k8::controllers::spg_stateful::SpgStatefulSetController;
     use crate::k8::controllers::spu_service::SpuServiceController;
     use crate::k8::controllers::spu::SpuController;
 
@@ -27,6 +27,8 @@ mod k8_operator {
         global_ctx: SharedContext,
         tls: Option<TlsConfig>,
     ) {
+        let config = global_ctx.config();
+
         let spu_service_ctx: StoreContext<SpuServiceSpec> = StoreContext::new();
         let statefulset_ctx: StoreContext<StatefulsetSpec> = StoreContext::new();
         let spg_service_ctx: StoreContext<SpgServiceSpec> = StoreContext::new();
@@ -51,27 +53,33 @@ mod k8_operator {
             spg_service_ctx.clone(),
         );
 
-        SpgStatefulSetController::start(
-            k8_client.clone(),
-            namespace.clone(),
-            global_ctx.spgs().clone(),
-            statefulset_ctx,
-            global_ctx.spus().clone(),
-            spg_service_ctx,
-            tls,
-        );
+        whitelist!(config, "k8_spg", {
+            SpgStatefulSetController::start(
+                k8_client.clone(),
+                namespace.clone(),
+                global_ctx.spgs().clone(),
+                statefulset_ctx,
+                global_ctx.spus().clone(),
+                spg_service_ctx,
+                tls,
+            );
+        });
 
-        SpuController::start(
-            global_ctx.spus().clone(),
-            spu_service_ctx.clone(),
-            global_ctx.spgs().clone(),
-        );
+        whitelist!(config, "k8_spu", {
+            SpuController::start(
+                global_ctx.spus().clone(),
+                spu_service_ctx.clone(),
+                global_ctx.spgs().clone(),
+            );
+        });
 
-        SpuServiceController::start(
-            k8_client,
-            namespace,
-            spu_service_ctx,
-            global_ctx.spgs().clone(),
-        );
+        whitelist!(config, "k8_spu_service", {
+            SpuServiceController::start(
+                k8_client,
+                namespace,
+                spu_service_ctx,
+                global_ctx.spgs().clone(),
+            );
+        });
     }
 }

src/sc/src/k8/controllers/spg.rs → src/sc/src/k8/controllers/spg_stateful.rs

@@ -118,7 +118,7 @@ impl SpgStatefulSetController {
         Ok(())
     }
 
-    #[instrument(skip(self, spu_k8_config))]
+    #[instrument(skip(self, spu_k8_config, spu_group))]
     async fn sync_spg_to_statefulset(
         &mut self,
         spu_group: SpuGroupObj,
@@ -137,13 +137,15 @@ impl SpgStatefulSetController {
         } else {
             // if we pass this stage, then we reserved id.
             if !spu_group.is_already_valid() {
+                debug!("not valid");
                 let status = SpuGroupStatus::reserved();
                 self.groups
                     .update_status(spg_name.to_owned(), status)
                     .await?;
                 return Ok(());
             }
 
+            debug!("continue");
             let (spg_service_key, spg_service_action) = spu_group.generate_service();
 
             trace!("spg_service_actions: {:#?}", spg_service_action);