Restart Watcher programmatically after a timeout.

[mancioshell]

I have the need to download a new kubeconfig after a certain timeout.

I'm using with success kube.rs, to expose api rest through a reflector with a watcher.
But i need to restart the watcher, with a new client (and a new kubeconfig) after a certain timeout has expired.

Any suggest?

[clux]

as it stands, you have to recreate the watcher to do that, since it is passed the Client via the Api, and if you create a new client from a new kubeconfig then that needs to be passed in again. maybe that's something that could be improved in an issue.

[mancioshell]

Hi @clux , thank you for your reply.

Actually, I have the following function which starts the watcher, spawns the reflectors and return the store which is used in a axum route.

pub fn watch_pods(&self) -> Store<Pod> {
        let (store, writer) = reflector::store();

        let pods: Api<Pod> = Api::all(self.client.clone());
        let environment = self.environment.clone();

        let pods_stream = watcher(pods, Default::default())
            .modify(|pod| {
                pod.managed_fields_mut().clear();
                pod.annotations_mut().clear();
            })
            .filter({
                move |obj| match obj {
                    Ok(event) => _filter_events(event, environment.clone()),
                    Err(_) => future::ready(false),
                }
            });

        let watch = reflector(writer, pods_stream)
            .default_backoff()
            .touched_objects()
            .for_each(|r| {
                future::ready(match r {
                    Ok(o) => debug!(
                        "Pod with name {} in namespace {}",
                        o.name().unwrap(),
                        o.metadata.namespace.as_ref().unwrap()
                    ),
                    Err(e) => warn!("watcher error: {e}"),
                })
            });

        tokio::spawn(watch);

        return store;
    }

I have two pain points:

1. how to restarts the watcher after a timeout occured (in my case the timeout comes from the sts assume role token expiration time)
2. how to re-initialize the reflector without re-initialize the store, but instead continue to use the current store to avoid data loss.

Could i open an issue and link this discussion?

[mancioshell]

I have found a solution (i don't know is the most efficient, but it works):

pub fn watch_pods(&self) -> Store<Pod> {
        let (store, writer) = reflector::store();

        let environment = self.environment.clone();
        let client = self.k8s_client.clone();

        let pods_stream = futures::stream::unfold((), move |_| {
            let environment = environment.clone();
            let client: K8SClient = client.clone();

            async move {
                loop {

                    let (client, timeout_duration) = client.get_client().await;
                    let api: Api<Pod> = Api::all(client.clone());

                    println!("Downloading kubeconfig");

                    let stream = watcher(api, Default::default())
                        .modify(|obj| {
                            obj.managed_fields_mut().clear();
                            obj.annotations_mut().clear();
                        })
                        .filter({
                            move |obj| match obj {
                                Ok(event) => _filter_events(event, environment.clone()),
                                Err(_) => future::ready(false),
                            }
                        })
                        .take_until(tokio::time::sleep(timeout_duration));

                    println!("Stream: started");

                    return Some((stream, ()));
                }
            }
        })
        .flatten(); 

        let watch = reflector(writer, pods_stream)
            .default_backoff()
            .touched_objects()
            .for_each(|r| {
                future::ready(match r {
                    Ok(o) => debug!(
                        "Pod with name {} in namespace {}",
                        o.name().unwrap(),
                        o.metadata.namespace.as_ref().unwrap()
                    ),
                    Err(e) => warn!("watcher error: {e}"),
                })
            });

        tokio::spawn(watch);

        return store;
    }