async fn start_pod_informer<P: 'static + Provider + Sync + Send>(
    client: kube::Client,
    node_name: String,
    mut queue: Queue<P>,
    signal: Arc<AtomicBool>,
) {
    let node_selector = format!("spec.nodeName={}", node_name);
    let params = ListParams {
        field_selector: Some(node_selector),
        ..Default::default()
    };
    let api = Api::<KubePod>::all(client);
    let mut informer = watcher(api, params).boxed();
    loop {
        match informer.try_next().await {
            Ok(Some(event)) => {
                debug!("Handling Kubernetes pod event: {:?}", event);
                if matches!(event, kube_runtime::watcher::Event::Applied(_))
                    && signal.load(Ordering::Relaxed)
                {
                    warn!("Node is shutting down and unschedulable. Dropping Add Pod event.");
                    continue;
                }
                if let kube_runtime::watcher::Event::Restarted(pods) = event {
                    info!("Got a pod watch restart. Resyncing queue...");
                    // If we got a restart, we need to requeue an applied event for all pods
                    match queue.resync(pods).await {
                        Ok(()) => info!("Finished resync of pods"),
                        Err(e) => warn!("Error resyncing pods: {}", e),
                    };
                } else {
                    match queue.enqueue(event).await {
                        Ok(()) => debug!("Enqueued event for processing"),
                        Err(e) => warn!("Error enqueuing pod event: {}", e),
                    };
                }
            }
            Ok(None) => break,
            Err(e) => warn!("Error streaming pod events: {:?}", e),
        }
    }
}
/// Listens for updates to pods on this node and forwards them to queue.
Listens for updates to pods on this node and forwards them to queue.
Listens for updates to pods on this node and forwards them to queue .
async fn start_pod_informer<P: 'static + Provider + Sync + Send>(
    client: kube::Client,
    node_name: String,
    mut queue: Queue<P>,
    signal: Arc<AtomicBool>,
) {
    let node_selector = format!("spec.nodeName={}", node_name);
    let params = ListParams {
        field_selector: Some(node_selector),
        ..Default::default()
    };
    let api = Api::<KubePod>::all(client);
    let mut informer = watcher(api, params).boxed();
    loop {
        match informer.try_next().await {
            Ok(Some(event)) => {
                debug!("Handling Kubernetes pod event: {:?}", event);
                if matches!(event, kube_runtime::watcher::Event::Applied(_))
                    && signal.load(Ordering::Relaxed)
                {
                    warn!("Node is shutting down and unschedulable. Dropping Add Pod event.");
                    continue;
                }
                if let kube_runtime::watcher::Event::Restarted(pods) = event {
                    info!("Got a pod watch restart. Resyncing queue...");
                    // If we got a restart, we need to requeue an applied event for all pods
                    match queue.resync(pods).await {
                        Ok(()) => info!("Finished resync of pods"),
                        Err(e) => warn!("Error resyncing pods: {}", e),
                    };
                } else {
                    match queue.enqueue(event).await {
                        Ok(()) => debug!("Enqueued event for processing"),
                        Err(e) => warn!("Error enqueuing pod event: {}", e),
                    };
                }
            }
            Ok(None) => break,
            Err(e) => warn!("Error streaming pod events: {:?}", e),
        }
    }
}
async fn start_pod_informer < P : ' static + Provider + Sync + Send > ( client : kube :: Client , node_name : String , mut queue : Queue < P > , signal : Arc < AtomicBool > , ) { let node_selector = format ! ( " " node_name ) ; let params = ListParams { field_selector : Some ( node_selector ) , .. Default :: default ( ) } ; let api = Api :: < KubePod > :: all ( client ) ; let mut informer = watcher ( api , params ) . boxed ( ) ; loop { match informer . try_next ( ) . await { Ok ( Some ( event ) ) => { debug ! ( " " event ) ; if matches ! ( event kube_runtime watcher Event Applied ( ) ) && signal . load ( Ordering :: Relaxed ) { warn ! ( " " ) ; continue ; } if let kube_runtime :: watcher :: Event :: Restarted ( pods ) = event { info ! ( " " ) ; // If we got a restart, we need to requeue an applied event for all pods match queue . resync ( pods ) . await { Ok ( ( ) ) => info ! ( " " ) , Err ( e ) => warn ! ( " " e ) , } ; } else { match queue . enqueue ( event ) . await { Ok ( ( ) ) => debug ! ( " " ) , Err ( e ) => warn ! ( " " e ) , } ; } } Ok ( None ) => break , Err ( e ) => warn ! ( " " e ) , } } }
Listens for updates to pods on this node and forwards them to queue.
Listens for updates to pods on this node and forwards them to queue .
client
kube::Client
node_name
String
queue
Queue<P>
signal
Arc<AtomicBool>
client
kube::Client
node_name
String
queue
Queue<P>
signal
Arc<AtomicBool>