Add option to poll or watch resources

src/models/mod.rs

@@ -8,6 +8,6 @@ pub use cluster::NamespaceScope;
 pub use resources::{
     ConfigMapsModel, DaemonSetsModel, DeploymentsModel, IngressesModel, Item, JobsModel,
     NodesModel, ObjectKey, PersistentVolumeClaimsModel, PersistentVolumesModel, PodsModel,
-    ResourceDetail, ResourceDetailState, ResourceDetailUpdate, ResourceKind, ResourceSnapshot,
-    ResourceStore, SecretsModel, ServicesModel, SnapshotPhase,
+    PollRate, ResourceDetail, ResourceDetailState, ResourceDetailUpdate, ResourceKind,
+    ResourceSnapshot, ResourceStore, SecretsModel, ServicesModel, SnapshotPhase, WatchMode,
 };

src/models/resources/mod.rs

@@ -67,4 +67,4 @@ pub use service_accounts::ServiceAccountsModel;
 pub use services::ServicesModel;
 pub use stateful_sets::StatefulSetsModel;
 pub use storage_classes::StorageClassesModel;
-pub use store::{ObjectKey, ResourceSnapshot, ResourceStore, SnapshotPhase};
+pub use store::{ObjectKey, PollRate, ResourceSnapshot, ResourceStore, SnapshotPhase, WatchMode};

src/models/resources/store.rs

@@ -35,6 +35,7 @@ pub struct ResourceSnapshot {
     items: Arc<[Arc<Item>]>,
     revision: u64,
     phase: SnapshotPhase,
+    mode: WatchMode,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -44,12 +45,43 @@ pub enum SnapshotPhase {
     Error(String),
 }
 
+/// How the snapshot is kept up to date. `Poll` carries the cadence so the UI
+/// can show and adjust it.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum WatchMode {
+    /// Live watch stream.
+    Watch,
+    /// Watch is not permitted; the store is re-listed on this cadence.
+    Poll(PollRate),
+}
+
+/// Automatic re-list cadence while polling. `Off` disables the timer; a
+/// manual refresh still re-lists once. Persisted per resource kind.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum PollRate {
+    /// No automatic polling.
+    Off,
+    /// Poll every this many seconds.
+    Every(u64),
+}
+
+impl PollRate {
+    pub const DEFAULT: PollRate = PollRate::Every(5);
+}
+
+impl Default for PollRate {
+    fn default() -> Self {
+        Self::DEFAULT
+    }
+}
+
 impl ResourceSnapshot {
     pub fn loading() -> Self {
         Self {
             items: Arc::from(Vec::<Arc<Item>>::new()),
             revision: 0,
             phase: SnapshotPhase::Loading,
+            mode: WatchMode::Watch,
         }
     }
 
@@ -58,9 +90,15 @@ impl ResourceSnapshot {
             items: Arc::from(Vec::<Arc<Item>>::new()),
             revision: 0,
             phase: SnapshotPhase::Error(message.into()),
+            mode: WatchMode::Watch,
         }
     }
 
+    pub fn with_mode(mut self, mode: WatchMode) -> Self {
+        self.mode = mode;
+        self
+    }
+
     pub fn items(&self) -> &[Arc<Item>] {
         &self.items
     }
@@ -72,6 +110,10 @@ impl ResourceSnapshot {
     pub fn phase(&self) -> &SnapshotPhase {
         &self.phase
     }
+
+    pub fn mode(&self) -> WatchMode {
+        self.mode
+    }
 }
 
 /// Accumulates watch events; the repository flushes snapshots at a bounded rate.
@@ -148,6 +190,7 @@ impl ResourceStore {
             items: self.objects.values().cloned().collect(),
             revision: self.revision,
             phase,
+            mode: WatchMode::Watch,
         }
     }
 }
@@ -229,4 +272,15 @@ mod tests {
         assert_eq!(snapshot.items().len(), 1);
         assert_eq!(snapshot.items()[0].name, "b");
     }
+
+    #[test]
+    fn snapshots_carry_the_watch_mode() {
+        let mut store = ResourceStore::new();
+        let snapshot = store.snapshot(SnapshotPhase::Ready);
+        assert_eq!(snapshot.mode(), WatchMode::Watch);
+        let snapshot = snapshot.with_mode(WatchMode::Poll(PollRate::Every(5)));
+        assert_eq!(snapshot.mode(), WatchMode::Poll(PollRate::Every(5)));
+        assert_eq!(ResourceSnapshot::loading().mode(), WatchMode::Watch);
+        assert_eq!(ResourceSnapshot::failed("x").mode(), WatchMode::Watch);
+    }
 }

src/repository/mod.rs

@@ -3,8 +3,10 @@
 
 pub mod custom_resources;
 pub mod resources;
+pub mod retry;
+pub mod watch_prefs;
 
-pub use resources::namespace_names;
+pub use watch_prefs::{WatchPreference, WatchPrefs};
 
 use std::collections::HashMap;
 use std::sync::{Arc, Weak};
@@ -13,7 +15,7 @@ use gpui_kit::{App, BorrowAppContext as _, Global};
 use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
-use crate::models::{NamespaceScope, ResourceKind, ResourceSnapshot};
+use crate::models::{NamespaceScope, PollRate, ResourceKind, ResourceSnapshot, WatchMode};
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 struct SubscriptionKey {
@@ -26,17 +28,21 @@ struct Entry {
     snapshots: watch::Sender<Arc<ResourceSnapshot>>,
     _cancel: watch::Sender<bool>,
     refresh: watch::Sender<u64>,
+    prefs: watch::Sender<WatchPrefs>,
 }
 
-#[derive(Default)]
 pub struct Repository {
     entries: HashMap<SubscriptionKey, Weak<Entry>>,
+    watch_prefs: watch_prefs::WatchPrefsStore,
 }
 
 impl Global for Repository {}
 
 pub fn init(cx: &mut App) {
-    cx.set_global(Repository::default());
+    cx.set_global(Repository {
+        entries: HashMap::new(),
+        watch_prefs: watch_prefs::WatchPrefsStore::load(),
+    });
 }
 
 /// A live list/watch subscription. Dropping the last handle for a key cancels
@@ -68,7 +74,6 @@ impl Subscription {
 
 /// Subscribe to a kind in a namespace scope. Subscriptions with the same key
 /// share one list/watch task, so opening the same resource twice costs nothing.
-#[allow(dead_code)]
 pub fn subscribe(
     cx: &mut App,
     kind: ResourceKind,
@@ -108,18 +113,76 @@ pub(crate) fn subscribe_kind(
             };
         }
 
-        let (snapshots, rx) = watch::channel(Arc::new(ResourceSnapshot::loading()));
+        let mut initial_prefs = repository.watch_prefs.prefs_for(kind);
+        if initial_prefs.mode == WatchPreference::Watch
+            && resources::watcher::watch_denied(key.epoch, kind)
+        {
+            // Watch is known to be denied for this kind: go straight to
+            // polling instead of re-trying the doomed watch on every
+            // namespace filter change. The saved preference stays `Watch`.
+            initial_prefs.mode = WatchPreference::Poll;
+        }
+        let initial_mode = match initial_prefs.mode {
+            WatchPreference::Watch => WatchMode::Watch,
+            WatchPreference::Poll => WatchMode::Poll(initial_prefs.rate),
+        };
+
+        let (snapshots, rx) = watch::channel(Arc::new(
+            ResourceSnapshot::loading().with_mode(initial_mode),
+        ));
         let (cancel, cancel_rx) = watch::channel(false);
         let (refresh, refresh_rx) = watch::channel(0u64);
+        let (prefs, prefs_rx) = watch::channel(initial_prefs);
         let entry = Arc::new(Entry {
             snapshots: snapshots.clone(),
             _cancel: cancel,
             refresh,
+            prefs,
         });
-        let watch_task =
-            resources::watcher::run(client, kind, scope, snapshots, cancel_rx, refresh_rx);
+        let watch_task = resources::watcher::run(
+            client, kind, scope, key.epoch, snapshots, cancel_rx, refresh_rx, prefs_rx,
+        );
         tokio.spawn(watch_task);
         repository.entries.insert(key, Arc::downgrade(&entry));
         Subscription { rx, _entry: entry }
     }))
 }
+
+/// Set whether a resource kind is watched live or polled. The saved poll rate
+/// is kept. Applies to every running watch task for the kind and is saved.
+pub fn set_watch_preference(cx: &mut App, kind: ResourceKind, mode: WatchPreference) {
+    if mode == WatchPreference::Watch {
+        // Explicit request: forget any RBAC denial so the watch is retried.
+        let epoch = cx.global::<ClusterState>().context_epoch();
+        resources::watcher::clear_watch_denied(epoch, kind);
+    }
+    update_watch_prefs(cx, kind, |prefs| prefs.mode = mode);
+}
+
+/// Set a kind's poll rate and switch it to poll mode. Applies to every
+/// running watch task for the kind and is saved.
+pub fn set_poll_rate(cx: &mut App, kind: ResourceKind, rate: PollRate) {
+    update_watch_prefs(cx, kind, |prefs| {
+        prefs.mode = WatchPreference::Poll;
+        prefs.rate = rate;
+    });
+}
+
+fn update_watch_prefs(cx: &mut App, kind: ResourceKind, change: impl FnOnce(&mut WatchPrefs)) {
+    let epoch = cx.global::<ClusterState>().context_epoch();
+    cx.update_global::<Repository, _>(|repository, _| {
+        repository.watch_prefs.update(kind, change);
+        if let Err(error) = repository.watch_prefs.save() {
+            tracing::warn!(%error, "repository: failed to save watch prefs");
+        }
+        let prefs = repository.watch_prefs.prefs_for(kind);
+        for (key, entry) in &repository.entries {
+            if key.kind == kind
+                && key.epoch == epoch
+                && let Some(entry) = entry.upgrade()
+            {
+                entry.prefs.send_replace(prefs);
+            }
+        }
+    });
+}

src/repository/resources/mod.rs

@@ -30,5 +30,3 @@ pub mod services;
 pub mod stateful_sets;
 pub mod storage_classes;
 pub mod watcher;
-
-pub use watcher::namespace_names;

src/repository/resources/pods/mod.rs

@@ -21,6 +21,7 @@ use crate::models::resources::pod::detail::{
 use crate::models::{NamespaceScope, ResourceKind};
 use gpui_kit::App;
 
+use crate::repository::retry::{Retry, RetryAction};
 use crate::repository::{Subscription, subscribe_kind};
 
 const METRICS_POLL_INTERVAL: Duration = Duration::from_secs(10);
@@ -113,41 +114,106 @@ async fn watch_logs(
     mut cancel: watch::Receiver<bool>,
 ) {
     let api: Api<Pod> = Api::namespaced(client, &namespace);
-    let params = LogParams {
-        container: Some(container),
-        follow,
-        tail_lines: Some(2_000),
-        ..LogParams::default()
-    };
-    let stream = tokio::select! {
-        _ = cancel.changed() => return,
-        result = api.log_stream(&name, &params) => result,
-    };
-    let stream = match stream {
-        Ok(stream) => stream,
-        Err(error) => {
-            let _ = updates.send(Err(format!("Could not stream logs: {error}")));
-            return;
-        }
-    };
-    let mut lines = stream.lines();
+    let mut retry = Retry::new();
+    // Timestamp of the last streamed line; reconnects resume after it instead
+    // of replaying the tail, so a dropped follow stream does not duplicate
+    // lines. kubectl does the same with --since-time.
+    let mut last_timestamp: Option<jiff::Timestamp> = None;
+
     loop {
-        tokio::select! {
+        let params = LogParams {
+            container: Some(container.clone()),
+            follow,
+            // Timestamps are only needed to resume a dropped follow stream.
+            timestamps: follow,
+            tail_lines: if last_timestamp.is_some() {
+                None
+            } else {
+                Some(2_000)
+            },
+            since_time: last_timestamp,
+            ..LogParams::default()
+        };
+        let timestamps = params.timestamps;
+        let stream = tokio::select! {
             _ = cancel.changed() => return,
-            line = lines.try_next() => {
-                match line {
-                    Ok(Some(line)) => {
-                        if updates.send(Ok(line.trim_end_matches('\r').to_string())).is_err() {
-                            return;
-                        }
+            result = api.log_stream(&name, &params) => result,
+        };
+        let mut lines = match stream {
+            Ok(stream) => {
+                retry.list_succeeded();
+                stream.lines()
+            }
+            Err(error) => {
+                if !follow {
+                    let _ = updates.send(Err(format!("Could not stream logs: {error}")));
+                    return;
+                }
+                match retry.failed(&error) {
+                    RetryAction::Retry(delay) => {
+                        tracing::warn!(%error, pod = %name, delay_ms = delay.as_millis() as u64, "pods: log stream setup failed, retrying");
+                        tokio::select! {
+                            _ = cancel.changed() => return,
+                            _ = tokio::time::sleep(delay) => continue,
+                        };
                     }
-                    Ok(None) => return,
-                    Err(error) => {
-                        let _ = updates.send(Err(format!("Log stream error: {error}")));
+                    RetryAction::GiveUp => {
+                        let _ = updates.send(Err(format!("Could not stream logs: {error}")));
                         return;
                     }
                 }
             }
+        };
+
+        loop {
+            let line = tokio::select! {
+                _ = cancel.changed() => return,
+                line = lines.try_next() => line,
+            };
+            match line {
+                Ok(Some(raw)) => {
+                    retry.watch_succeeded();
+                    let raw = raw.trim_end_matches('\r');
+                    let text = if timestamps {
+                        match raw.split_once(' ') {
+                            Some((stamp, text)) => {
+                                if let Ok(timestamp) = stamp.parse::<jiff::Timestamp>() {
+                                    last_timestamp = Some(timestamp);
+                                }
+                                text
+                            }
+                            None => raw,
+                        }
+                    } else {
+                        raw
+                    };
+                    if updates.send(Ok(text.to_string())).is_err() {
+                        return;
+                    }
+                }
+                Ok(None) => {
+                    tracing::debug!(pod = %name, "pods: log stream ended");
+                    break;
+                }
+                Err(error) => {
+                    tracing::warn!(%error, pod = %name, "pods: log stream read error");
+                    break;
+                }
+            }
+        }
+
+        if !follow {
+            return;
+        }
+        // The follow stream dropped: reconnect after a soft backoff, resuming
+        // after the last received line.
+        let RetryAction::Retry(delay) = retry.failed_soft() else {
+            return;
+        };
+        tracing::warn!(pod = %name, delay_ms = delay.as_millis() as u64, "pods: log stream interrupted, reconnecting");
+        tokio::select! {
+            _ = cancel.changed() => return,
+            _ = tokio::time::sleep(delay) => {}
         }
     }
 }