Further reduce memory usage

foriequal0 · Aug 12, 2024 · e42b998 · e42b998
1 parent ebcb50e
commit e42b998
Show file tree

Hide file tree

Showing 4 changed files with 99 additions and 14 deletions.
diff --git a/charts/pod-graceful-drain/values.yaml b/charts/pod-graceful-drain/values.yaml
@@ -39,7 +39,16 @@ securityContext:
   #   drop:
   #   - ALL
 
-resources: { }
+resources:
+  requests:
+    cpu: "100m"
+    # 10MB at rest.
+    memory: "50Mi"
+  limits:
+    # main runtime is single-threaded.
+    cpu: "1000m"
+    memory: "500Mi"
+
 nodeSelector: { }
 tolerations: [ ]
 affinity: { }

diff --git a/src/reflector.rs b/src/reflector.rs
@@ -4,7 +4,8 @@ use std::hash::Hash;
 use std::sync::Arc;
 
 use eyre::Result;
-use futures::{Stream, StreamExt};
+use futures::{Stream, StreamExt, TryStreamExt};
+use k8s_openapi::api::core::v1::{PodSpec, PodStatus};
 use k8s_openapi::api::{
     core::v1::{Pod, Service},
     networking::v1::Ingress,
@@ -21,7 +22,7 @@ use crate::elbv2::apis::TargetGroupBinding;
 use crate::service_registry::ServiceSignal;
 use crate::shutdown::Shutdown;
 use crate::spawn_service::spawn_service;
-use crate::{instrumented, Config, ServiceRegistry};
+use crate::{instrumented, try_some, Config, ServiceRegistry};
 
 #[derive(Clone)]
 pub struct Stores {
@@ -66,23 +67,50 @@ pub fn start_reflectors(
     let (pod_reader, pod_writer) = store();
     spawn_service(shutdown, "reflector:Pod", {
         let api: Api<Pod> = api_proivder.all();
-        let stream = watcher(api, Default::default());
+        let stream = watcher(api, Default::default()).map_ok(|event| {
+            event.modify(|pod| {
+                if let Some(spec) = try_some!(mut pod.spec?) {
+                    *spec = PodSpec {
+                        readiness_gates: spec.readiness_gates.clone(),
+                        ..PodSpec::default()
+                    }
+                }
+                if let Some(spec) = try_some!(mut pod.status?) {
+                    *spec = PodStatus {
+                        conditions: spec.conditions.clone(),
+                        ..PodStatus::default()
+                    }
+                }
+            })
+        });
         let signal = service_registry.register("reflector:Pod");
         run_reflector(shutdown, pod_writer, stream, signal)
     })?;
 
     let (service_reader, service_writer) = store();
     spawn_service(shutdown, "reflector:Service", {
         let api: Api<Service> = api_proivder.all();
-        let stream = watcher(api, Default::default());
+        let stream = watcher(api, Default::default()).map_ok(|ev| {
+            ev.modify(|service| {
+                service.metadata.annotations = None;
+                service.metadata.labels = None;
+                service.status = None;
+            })
+        });
         let signal = service_registry.register("reflector:Service");
         run_reflector(shutdown, service_writer, stream, signal)
     })?;
 
     let (ingress_reader, ingress_writer) = store();
     spawn_service(shutdown, "reflector:Ingress", {
         let api: Api<Ingress> = api_proivder.all();
-        let stream = watcher(api, Default::default());
+        let stream = watcher(api, Default::default()).map_ok(|ev| {
+            ev.modify(|ingress| {
+                ingress.metadata.annotations = None;
+                ingress.metadata.labels = None;
+                ingress.status = None;
+            })
+        });
         let signal = service_registry.register("reflector:Ingress");
         run_reflector(shutdown, ingress_writer, stream, signal)
     })?;
@@ -91,7 +119,13 @@ pub fn start_reflectors(
     if !config.experimental_general_ingress {
         spawn_service(shutdown, "reflector:TargetGroupBinding", {
             let api: Api<TargetGroupBinding> = api_proivder.all();
-            let stream = watcher(api, Default::default());
+            let stream = watcher(api, Default::default()).map_ok(|ev| {
+                ev.modify(|tgb| {
+                    tgb.metadata.annotations = None;
+                    tgb.metadata.labels = None;
+                    tgb.status = None;
+                })
+            });
             let signal = service_registry.register("reflector:TargetGroupBinding");
             run_reflector(shutdown, tgb_writer, stream, signal)
         })?;

diff --git a/src/utils.rs b/src/utils.rs
@@ -93,14 +93,55 @@ macro_rules! try_some {
         $crate::try_some!(@coalesce ($($h)* $m) $($t)*)
     };
 
+    (@coalesce_mut ($($h:tt)*)) => {
+        $($h)*
+    };
+    (@coalesce_mut ($($h:tt)*) ? $($t:tt)*) => {
+        $crate::try_some!(@coalesce_mut ($($h)*.as_mut()?) $($t)*)
+    };
+    (@coalesce_mut ($($h:tt)*) $m:tt $($t:tt)*) => {
+        $crate::try_some!(@coalesce_mut ($($h)* $m) $($t)*)
+    };
+
+    (mut $($tt:tt)*) => {
+        {
+            fn call<R>(f: impl FnOnce() ->::std::option::Option<R>) -> ::std::option::Option <R>{
+                f()
+            }
+            call(|| {
+                ::std::option::Option::Some($crate::try_some!(@coalesce_mut () $($tt)*))
+            })
+        }
+    };
+    (&mut $($tt:tt)*) => {
+        {
+            fn call<R>(f: impl FnOnce() ->::std::option::Option<R>) -> ::std::option::Option<R>
+            {
+                f()
+            }
+            call(|| {
+                ::std::option::Option::Some(&mut $crate::try_some!(@coalesce_mut () $($tt)*))
+            })
+        }
+    };
     (& $($tt:tt)*) => {
-        (|| -> ::std::option::Option<_> {
-            ::std::option::Option::Some(& $crate::try_some!(@coalesce () $($tt)*))
-        })()
+        {
+            fn call<R>(f: impl FnOnce() ->::std::option::Option<R>) -> ::std::option::Option <R>{
+                f()
+            }
+            call(|| {
+                ::std::option::Option::Some(&$crate::try_some!(@coalesce () $($tt)*))
+            })
+        }
     };
     ($($tt:tt)*) => {
-        (|| -> ::std::option::Option<_> {
-            ::std::option::Option::Some($crate::try_some!(@coalesce () $($tt)*))
-        })()
+        {
+            fn call<R>(f: impl FnOnce() ->::std::option::Option<R>) -> ::std::option::Option <R>{
+                f()
+            }
+            call(|| {
+                ::std::option::Option::Some($crate::try_some!(@coalesce () $($tt)*))
+            })
+        }
     };
 }
diff --git a/src/webhooks/mod.rs b/src/webhooks/mod.rs
@@ -193,8 +193,9 @@ where
 
     let object_ref: ObjectRef<K> =
         get_object_ref_from_name(&request.name, request.namespace.as_ref());
+    let request_id: u32 = rand::random();
     instrumented!(
-        span!(Level::ERROR, "admission", %object_ref, operation = ?request.operation),
+        span!(Level::ERROR, "admission", %object_ref, operation = ?request.operation, request_id),
         async move {
             trace!(user_info=?request.user_info);