Add service selector to fluentbit spec (anvil-verifier#437)

Signed-off-by: Wenjie Ma <[email protected]>

Author: euclidgame

deploy/fluent/crd.yaml

@@ -1840,6 +1840,11 @@ spec:
                     type: string
                   default: {}
                   type: object
+                serviceSelector:
+                  additionalProperties:
+                    type: string
+                  nullable: true
+                  type: object
                 tolerations:
                   items:
                     description: "The pod this Toleration is attached to tolerates any taint that matches the triple <key,value,effect> using the matching operator <operator>."

e2e/src/fluent_e2e.rs

@@ -1,6 +1,6 @@
 #![allow(unused_imports)]
 #![allow(unused_variables)]
-use k8s_openapi::api::core::v1::{Pod, ServiceAccount};
+use k8s_openapi::api::core::v1::{Pod, ServiceAccount, Service};
 use k8s_openapi::api::rbac::v1::RoleBinding;
 use k8s_openapi::api::{apps::v1::DaemonSet, rbac::v1::Role};
 use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
@@ -120,6 +120,7 @@ pub async fn desired_state_test(client: Client, fb_name: String) -> Result<(), E
         // Check daemon set
         let role_api: Api<Role> = Api::default_namespaced(client.clone());
         let rb_api: Api<RoleBinding> = Api::default_namespaced(client.clone());
+        let svc_api: Api<Service> = Api::default_namespaced(client.clone());
         let sa_api: Api<ServiceAccount> = Api::default_namespaced(client.clone());
         let ds_api: Api<DaemonSet> = Api::default_namespaced(client.clone());
 
@@ -150,6 +151,15 @@ pub async fn desired_state_test(client: Client, fb_name: String) -> Result<(), E
             _ => {}
         };
 
+        let svc = svc_api.get(&fb_name).await;
+        match svc {
+            Err(e) => {
+                println!("Get service failed with error {}.", e);
+                continue;
+            }
+            _ => {}
+        }
+
         let ds = ds_api.get(&fb_name).await;
         match ds {
             Err(e) => {
@@ -274,6 +284,61 @@ pub async fn relabel_test(client: Client, fb_name: String) -> Result<(), Error>
     Ok(())
 }
 
+pub async fn service_selector_test(client: Client, fb_name: String) -> Result<(), Error> {
+    let timeout = Duration::from_secs(360);
+    let start = Instant::now();
+    let svc_api: Api<Service> = Api::default_namespaced(client.clone());
+    run_command(
+        "kubectl",
+        vec![
+            "patch",
+            "fb",
+            "fluent-bit",
+            "--type=json",
+            "-p",
+            "[{\"op\": \"add\", \"path\": \"/spec/serviceSelector\", \"value\": {\"never-match-anything\": \"val\"}}]",
+        ],
+        "failed to set service selector to fb",
+    );
+
+    // Sleep for extra 5 seconds to ensure the upgrading has started
+    sleep(Duration::from_secs(5)).await;
+    loop {
+        sleep(Duration::from_secs(5)).await;
+        if start.elapsed() > timeout {
+            return Err(Error::Timeout);
+        }
+
+        // Check daemon set
+        let svc = svc_api.get(&fb_name).await;
+        match svc {
+            Err(e) => {
+                println!("Get service failed with error {}.", e);
+                continue;
+            }
+            Ok(svc) => {
+                if svc
+                    .spec
+                    .as_ref()
+                    .unwrap()
+                    .selector
+                    .as_ref()
+                    .unwrap()
+                    .contains_key("never-match-anything")
+                {
+                    println!("Selector for service is updated yet");
+                    break;
+                } else {
+                    println!("Selector for service is not updated yet");
+                }
+            }
+        };
+    }
+
+    println!("Service selector test passed.");
+    Ok(())
+}
+
 pub async fn fluent_e2e_test() -> Result<(), Error> {
     // check if the CRD is already registered
     let client = Client::try_default().await?;
@@ -296,6 +361,7 @@ pub async fn fluent_e2e_test() -> Result<(), Error> {
 
     desired_state_test(client.clone(), fb_name.clone()).await?;
     relabel_test(client.clone(), fb_name.clone()).await?;
+    service_selector_test(client.clone(), fb_name.clone()).await?;
 
     println!("E2e test passed.");
     Ok(())

src/controller_examples/fluent_controller/fluentbit/exec/resource/service.rs

@@ -158,7 +158,13 @@ pub fn make_service(fb: &FluentBit) -> (service: Service)
             }
             ports
         });
-        service_spec.set_selector(make_base_labels(fb));
+        service_spec.set_selector({
+            if fb.spec().service_selector().is_some() {
+                fb.spec().service_selector().unwrap()
+            } else {
+                make_base_labels(fb)
+            }
+        });
         service_spec
     });
     service

src/controller_examples/fluent_controller/fluentbit/model/resource/service.rs

@@ -117,7 +117,11 @@ pub open spec fn make_service(fb: FluentBitView) -> ServiceView {
                     Some(seq![metrics_port])
                 }
             },
-            selector: Some(make_base_labels(fb)),
+            selector: if fb.spec.service_selector.is_Some() {
+                    fb.spec.service_selector
+                } else {
+                    Some(make_base_labels(fb))
+                },
             ..ServiceSpecView::default()
         }),
         ..ServiceView::default()

src/controller_examples/fluent_controller/fluentbit/trusted/exec_types.rs

@@ -186,6 +186,18 @@ impl FluentBitSpec {
         StringMap::from_rust_map(self.inner.service_labels.clone())
     }
 
+    #[verifier(external_body)]
+    pub fn service_selector(&self) -> (service_selector: Option<StringMap>)
+        ensures 
+            service_selector.is_Some() == self@.service_selector.is_Some(),
+            service_selector.is_Some() ==> service_selector.get_Some_0()@ == self@.service_selector.get_Some_0(),
+    {
+        match &self.inner.service_selector {
+            Some(selector) => Some(StringMap::from_rust_map(selector.clone())),
+            None => None,
+        }
+    }
+
     #[verifier(external_body)]
     pub fn service_annotations(&self) -> (service_annotations: StringMap)
         ensures service_annotations@ == self@.service_annotations,

src/controller_examples/fluent_controller/fluentbit/trusted/spec_types.rs

@@ -150,6 +150,7 @@ pub struct FluentBitSpecView {
     pub annotations: Map<StringView, StringView>,
     pub service_account_annotations: Map<StringView, StringView>,
     pub service_labels: Map<StringView, StringView>,
+    pub service_selector: Option<Map<StringView, StringView>>,
     pub service_annotations: Map<StringView, StringView>,
     pub affinity: Option<AffinityView>,
     pub disable_log_volumes: bool,

src/deps_hack/src/lib.rs

@@ -211,6 +211,8 @@ pub struct FluentBitSpec {
     pub service_account_annotations: std::collections::BTreeMap<String, String>,
     #[serde(default, rename = "serviceLabels")]
     pub service_labels: std::collections::BTreeMap<String, String>,
+    #[serde(default, rename = "serviceSelector")]
+    pub service_selector: Option<std::collections::BTreeMap<String, String>>,
     #[serde(default, rename = "serviceAnnotations")]
     pub service_annotations: std::collections::BTreeMap<String, String>,
     pub affinity: Option<k8s_openapi::api::core::v1::Affinity>,