expose list/watch parameters - fixes ynqa#11

Author: clux

CHANGELOG.md

@@ -1,3 +1,8 @@
+0.7.0 / UNRELEASED
+==================
+  * Expose list/watch parameters #11
+
+
 0.6.0 / 2019-05-12
 ==================
   * Expose getter `Informer::version`

README.md

@@ -25,7 +25,9 @@ To use it, you just feed in `T` as a `Spec` struct and `U` as a `Status` struct,
 ```rust
 use k8s_openapi::api::core::v1::{PodSpec, PodStatus};
 let resource = ResourceType::Pods(Some("kube-system".into()));
-let rf : Reflector<PodSpec, PodStatus> = Reflector::new(client.clone(), resource.into())?;
+let rf : Reflector<PodSpec, PodStatus> = Reflector::new(client.clone(), resource.into())?
+    .timeout(10)
+    .init();
 ```
 
 then you can `poll()` the reflector, and `read()` to get the current cached state:
@@ -53,7 +55,8 @@ You tell it what type parameters correspond to; `T` should be a `Spec` struct, a
 ```rust
 use k8s_openapi::api::core::v1::{PodSpec, PodStatus};
 let resource = ResourceType::Pods(Some("kube-system".into()));
-let inf : Informer<PodSpec, PodStatus> = Informer::new(client.clone(), resource.into())?;
+let inf : Informer<PodSpec, PodStatus> = Informer::new(client.clone(), resource.into())
+    .init()?;
 ```
 
 The main feature of `Informer<T, U>` is that after calling `.poll()` you handle the events and decide what to do with them yourself:
@@ -120,7 +123,7 @@ cargo run --example crd_reflector
 then you can `kubectl apply -f crd-baz.yaml -n kube-system`, or `kubectl delete -f crd-baz.yaml -n kube-system`, or `kubectl edit foos baz -n kube-system` to verify that the events are being picked up.
 
 ## Timing
-All watch calls have timeouts set to `10` seconds as a default (and kube always waits that long regardless of activity). If you like to hammer the API less, call `.poll()` less often.
+All watch calls have timeouts set to `10` seconds as a default (and kube always waits that long regardless of activity). If you like to hammer the API less, you can call `.poll()` less often. But the kube api holds for the full timeout value anyway and you can set it with `.timeout(n)`.
 
 ## License
 Apache 2.0 licensed. See LICENSE for details.

examples/crd_reflector.rs

@@ -28,7 +28,8 @@ fn main() -> Result<(), failure::Error> {
         namespace: Some("kube-system".into()),
         ..Default::default()
     };
-    let rf : Reflector<FooResource, Void> = Reflector::new(client, resource)?;
+    let rf : Reflector<FooResource, Void> = Reflector::new(client, resource)
+        .init()?;
 
     loop {
         // Update internal state by calling watch (blocks):

examples/deployment_reflector.rs

@@ -13,7 +13,9 @@ fn main() -> Result<(), failure::Error> {
     let client = APIClient::new(config);
 
     let resource = ResourceType::Deploys(Some("kube-system".into()));
-    let rf : Reflector<DeploymentSpec, DeploymentStatus> = Reflector::new(client, resource.into())?;
+    let rf : Reflector<DeploymentSpec, DeploymentStatus> =
+        Reflector::new(client, resource.into())
+        .init()?;
 
     // rf is initialized with full state, which can be extracted on demand.
     // Output is Map of name -> Deployment

examples/node_informer.rs

@@ -16,7 +16,9 @@ fn main() -> Result<(), failure::Error> {
     let client = APIClient::new(config);
 
     let nodes = ResourceType::Nodes;
-    let ni : Informer<NodeSpec, NodeStatus> = Informer::new(client.clone(), nodes.into())?;
+    let ni = Informer::new(client.clone(), nodes.into())
+        .labels("role=worker")
+        .init()?;
 
     loop {
         ni.poll()?;
@@ -34,6 +36,7 @@ fn handle_nodes(client: &APIClient, ev: WatchEvent<NodeSpec, NodeStatus>) -> Res
             info!("New Node: {}", o.spec.provider_id.unwrap());
         },
         WatchEvent::Modified(o) => {
+            // Nodes often modify a lot - only print broken nodes
             if let Some(true) = o.spec.unschedulable {
                 let failed = o.status.conditions.unwrap().into_iter().filter(|c| {
                     // In a failed state either some of the extra conditions are not False
@@ -51,6 +54,9 @@ fn handle_nodes(client: &APIClient, ev: WatchEvent<NodeSpec, NodeStatus>) -> Res
                 let req = Event::list_event_for_all_namespaces(opts)?.0;
                 let res = client.request::<Event>(req)?;
                 warn!("Node events: {:?}", res);
+            } else {
+                // Turn up logging above to see
+                debug!("Normal node: {}", o.metadata.name);
             }
         },
         WatchEvent::Deleted(o) => {

examples/node_reflector.rs

@@ -13,7 +13,9 @@ fn main() -> Result<(), failure::Error> {
     let client = APIClient::new(config);
 
     let resource = ResourceType::Nodes;
-    let rf : Reflector<NodeSpec, NodeStatus> = Reflector::new(client, resource.into())?;
+    let rf : Reflector<NodeSpec, NodeStatus> = Reflector::new(client, resource.into())
+        .labels("role=master")
+        .init()?;
 
     // rf is initialized with full state, which can be extracted on demand.
     // Output is Map of name -> Node

examples/pod_informer.rs

@@ -15,7 +15,8 @@ fn main() -> Result<(), failure::Error> {
     let namespace = Some(env::var("NAMESPACE").unwrap_or("kube-system".into()));
 
     let resource = ResourceType::Pods(namespace);
-    let inf : Informer<PodSpec, PodStatus> = Informer::new(client.clone(), resource.into())?;
+    let inf : Informer<PodSpec, PodStatus> = Informer::new(client.clone(), resource.into())
+        .init()?;
 
     // Here we both poll and reconcile based on events from the main thread
     // If you run this next to actix-web (say), spawn a thread and pass `inf` as app state

examples/pod_reflector.rs

@@ -13,7 +13,8 @@ fn main() -> Result<(), failure::Error> {
     let client = APIClient::new(config);
 
     let resource = ResourceType::Pods(Some("kube-system".into()));
-    let rf : Reflector<PodSpec, PodStatus> = Reflector::new(client.clone(), resource.into())?;
+    let rf : Reflector<PodSpec, PodStatus> = Reflector::new(client.clone(), resource.into())
+        .init()?;
 
     // Can read initial state now:
     rf.read()?.into_iter().for_each(|(name, p)| {

src/api/informer.rs

@@ -2,6 +2,7 @@ use crate::api::resource::{
     ResourceList,
     WatchEvent,
     ApiResource,
+    QueryParams,
 };
 use crate::client::APIClient;
 use crate::{Result};
@@ -31,48 +32,85 @@ pub struct Informer<T, U> where
     version: Arc<RwLock<String>>,
     client: APIClient,
     resource: ApiResource,
+    params: QueryParams,
 }
 
 impl<T, U> Informer<T, U> where
     T: Clone + DeserializeOwned,
     U: Clone + DeserializeOwned,
 {
     /// Create a reflector with a kube client on a kube resource
-    ///
-    /// Initializes resourceVersion with a 1 limit list call
-    pub fn new(client: APIClient, r: ApiResource) -> Result<Self> {
-        info!("Creating Informer for {:?}", r);
-        let initial = get_resource_version(&client, &r)?;
-        Ok(Informer {
+    pub fn new(client: APIClient, r: ApiResource) -> Self {
+        Informer {
             client,
             resource: r,
+            params: QueryParams::default(),
             events: Arc::new(RwLock::new(VecDeque::new())),
-            version: Arc::new(RwLock::new(initial)),
-        })
+            version: Arc::new(RwLock::new(0.to_string())),
+        }
     }
 
-    /// Create a reflector with a kube client on a kube resource
+    /// Configure the timeout for the list/watch call.
     ///
-    /// Initializes resourceVersion from a passed in value
-    pub fn from_version(client: APIClient, r: ApiResource, v: String) -> Result<Self> {
-        info!("Creating Informer for {:?}", r);
-        Ok(Informer {
-            client,
-            resource: r,
-            events: Arc::new(RwLock::new(VecDeque::new())),
-            version: Arc::new(RwLock::new(v)),
-        })
+    /// This limits the duration of the call, regardless of any activity or inactivity.
+    /// Defaults to 10s
+    pub fn timeout(mut self, timeout_secs: u32) -> Self {
+        self.params.timeout = Some(timeout_secs);
+        self
+    }
+
+    /// Configure the selector to restrict the list of returned objects by their fields.
+    ///
+    /// Defaults to everything.
+    /// Supports '=', '==', and '!=', and can comma separate: key1=value1,key2=value2
+    /// The server only supports a limited number of field queries per type.
+    pub fn fields(mut self, field_selector: &str) -> Self {
+        self.params.field_selector = Some(field_selector.to_string());
+        self
+    }
+
+    /// Configure the selector to restrict the list of returned objects by their labels.
+    ///
+    /// Defaults to everything.
+    /// Supports '=', '==', and '!=', and can comma separate: key1=value1,key2=value2
+    pub fn labels(mut self, label_selector: &str) -> Self {
+        self.params.label_selector = Some(label_selector.to_string());
+        self
+    }
+
+    /// If called, partially initialized resources are included in watch/list responses.
+    pub fn include_uninitialized(mut self) -> Self {
+        self.params.include_uninitialized = true;
+        self
+    }
+
+
+    /// Initialize without a prior version
+    ///
+    /// Will seed resourceVersion with a 1 limit list call to the resource
+    pub fn init(self) -> Result<Self> {
+        let initial = self.get_resource_version()?;
+        info!("Starting Informer for {:?}", self.resource);
+        *self.version.write().unwrap() = initial;
+        Ok(self)
+    }
+
+    /// Initialize from a prior version
+    pub fn init_from(self, v: String) -> Self {
+        info!("Recreating Informer for {:?} at {}", self.resource, v);
+        *self.version.write().unwrap() = v;
+        self
     }
 
+
     /// Run a single watch poll
     ///
     /// If this returns an error, it resets the resourceVersion.
     /// This is meant to be run continually and events are meant to be handled between.
     /// If handling all the events is too time consuming, you probably need a queue.
     pub fn poll(&self) -> Result<()> {
         trace!("Watching {:?}", self.resource);
-        let oldver = self.version();
-        match watch_for_resource_updates(&self.client, &self.resource, &oldver) {
+        match self.single_watch() {
             Ok((events, newver)) => {
                 *self.version.write().unwrap() = newver;
                 for e in events {
@@ -97,7 +135,7 @@ impl<T, U> Informer<T, U> where
     /// Reset the resourceVersion to current and clear the event queue
     pub fn reset(&self) -> Result<()> {
         // Fetch a new initial version:
-        let initial = get_resource_version(&self.client, &self.resource)?;
+        let initial = self.get_resource_version()?;
         *self.version.write().unwrap() = initial;
         self.events.write().unwrap().clear();
         Ok(())
@@ -107,42 +145,40 @@ impl<T, U> Informer<T, U> where
     pub fn version(&self) -> String {
         self.version.read().unwrap().clone()
     }
-}
 
-fn get_resource_version(client: &APIClient, rg: &ApiResource) -> Result<String>
-{
-    let req = rg.list_zero_resource_entries()?;
 
-    // parse to void a ResourceList into void except for Metadata
-    #[derive(Clone, Deserialize)]
-    struct Discard {} // ffs
-    let res = client.request::<ResourceList<Option<Discard>>>(req)?;
+    /// Init helper
+    fn get_resource_version(&self) -> Result<String> {
+        let req = self.resource.list_zero_resource_entries(&self.params)?;
 
-    let version = res.metadata.resourceVersion.unwrap_or_else(|| "0".into());
-    debug!("Got fresh resourceVersion={} for {}", version, rg.resource);
-    Ok( version )
-}
+        // parse to void a ResourceList into void except for Metadata
+        #[derive(Clone, Deserialize)]
+        struct Discard {} // ffs
+        let res = self.client.request::<ResourceList<Option<Discard>>>(req)?;
 
+        let version = res.metadata.resourceVersion.unwrap_or_else(|| "0".into());
+        debug!("Got fresh resourceVersion={} for {}", version, self.resource.resource);
+        Ok( version )
+    }
 
-fn watch_for_resource_updates<T, U>(client: &APIClient, rg: &ApiResource, ver: &str)
-    -> Result<(Vec<WatchEvent<T, U>>, String)> where
-  T: Clone + DeserializeOwned,
-  U: Clone + DeserializeOwned,
-{
-    let req = rg.watch_resource_entries_after(ver)?;
-    let events = client.request_events::<WatchEvent<T, U>>(req)?;
-
-    // Follow docs conventions and store the last resourceVersion
-    // https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes
-    let newver = events.iter().filter_map(|e| {
-        match e {
-            WatchEvent::Added(o) => o.metadata.resourceVersion.clone(),
-            WatchEvent::Modified(o) => o.metadata.resourceVersion.clone(),
-            WatchEvent::Deleted(o) => o.metadata.resourceVersion.clone(),
-            _ => None
-        }
-    }).last().unwrap_or_else(|| ver.into());
-    debug!("Got {} {} events, resourceVersion={}", events.len(), rg.resource, newver);
+    /// Watch helper
+    fn single_watch(&self) -> Result<(Vec<WatchEvent<T, U>>, String)> {
+        let oldver = self.version();
+        let req = self.resource.watch_resource_entries_after(&self.params, &oldver)?;
+        let events = self.client.request_events::<WatchEvent<T, U>>(req)?;
+
+        // Follow docs conventions and store the last resourceVersion
+        // https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes
+        let newver = events.iter().filter_map(|e| {
+            match e {
+                WatchEvent::Added(o) => o.metadata.resourceVersion.clone(),
+                WatchEvent::Modified(o) => o.metadata.resourceVersion.clone(),
+                WatchEvent::Deleted(o) => o.metadata.resourceVersion.clone(),
+                _ => None
+            }
+        }).last().unwrap_or_else(|| oldver.into());
+        debug!("Got {} {} events, resourceVersion={}", events.len(), self.resource.resource, newver);
 
-    Ok((events, newver))
+        Ok((events, newver))
+    }
 }

src/api/mod.rs

@@ -19,6 +19,7 @@ pub use self::resource::{
     Resource,
     ApiResource,
     ResourceType,
+    QueryParams,
     WatchEvent,
     ApiError,
 };