feat(kubernetes_logs source): Use resource_version of 0 to use cache (#…

…9974) * feat(kubernetes_logs source): Use resource_version of 0 to use cache Signed-off-by: Spencer Gilbert <spencer.gilbert@datadoghq.com> * Update reflector unit tests, use to_owned over to_string Signed-off-by: Spencer Gilbert <spencer.gilbert@datadoghq.com> * make fmt Signed-off-by: Spencer Gilbert <spencer.gilbert@datadoghq.com>
vectordotdev · Nov 11, 2021 · 6cfb28d · 6cfb28d
1 parent 1b91d6b
commit 6cfb28d
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 12 deletions.
diff --git a/src/kubernetes/reflector.rs b/src/kubernetes/reflector.rs
@@ -359,15 +359,15 @@ mod tests {
         drop(reflector);
     }
 
-    // Test the properties of the normal  execution flow.
+    // Test the properties of the normal execution flow.
     #[tokio::test]
     async fn flow_test() {
         trace_init();
 
         let invocations = vec![
             (
                 vec![],
-                None,
+                Some("0".to_owned()),
                 ExpInvRes::Stream(vec![
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid0", "10"))),
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid1", "15"))),
@@ -419,7 +419,7 @@ mod tests {
         let invocations = vec![
             (
                 vec![],
-                None,
+                Some("0".to_owned()),
                 ExpInvRes::Stream(vec![
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid0", "10"))),
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid1", "15"))),
@@ -460,7 +460,7 @@ mod tests {
         let invocations = vec![
             (
                 vec![],
-                None,
+                Some("0".to_owned()),
                 ExpInvRes::Stream(vec![
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid0", "10"))),
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid1", "15"))),
@@ -499,7 +499,7 @@ mod tests {
         let invocations = vec![
             (
                 vec![],
-                None,
+                Some("0".to_owned()),
                 ExpInvRes::Stream(vec![
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid0", "10"))),
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid1", "15"))),
@@ -543,7 +543,7 @@ mod tests {
         let invocations = vec![
             (
                 vec![],
-                None,
+                Some("0".to_owned()),
                 ExpInvRes::Stream(vec![
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid0", "10"))),
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid1", "15"))),
@@ -584,7 +584,7 @@ mod tests {
         let invocations = vec![
             (
                 vec![],
-                None,
+                Some("0".to_owned()),
                 ExpInvRes::Stream(vec![
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid0", "10"))),
                     ExpStmRes::Item(WatchEvent::Added(make_pod("uid1", "15"))),
@@ -681,7 +681,7 @@ mod tests {
                     field_selector: Some("fields".to_owned()),
                     label_selector: Some("labels".to_owned()),
                     pretty: None,
-                    resource_version: None,
+                    resource_version: Some("0".to_owned()),
                     timeout_seconds: Some(290),
                 }
             );
@@ -1083,8 +1083,13 @@ mod tests {
 
         // Prepare reflector.
         let pause_between_requests = Duration::from_secs(60 * 60); // 1 hour
-        let mut reflector =
-            Reflector::new(watcher, state_writer, None, None, pause_between_requests);
+        let mut reflector = Reflector::new(
+            watcher,
+            state_writer,
+            None,
+            Some("0".to_owned()),
+            pause_between_requests,
+        );
 
         // Run test logic.
         let logic = tokio::spawn(async move {

diff --git a/src/kubernetes/resource_version.rs b/src/kubernetes/resource_version.rs
@@ -9,8 +9,8 @@ pub struct State(Option<String>);
 
 impl State {
     /// Create a new resource version [`State`].
-    pub const fn new() -> Self {
-        Self(None)
+    pub fn new() -> Self {
+        Self(Some("0".to_owned()))
     }
 
     /// Update the resource version from a candidate obtained earlier.