Change forbidden watcher to drop, not degrade (#178)

Signed-off-by: Daniel Guns <danbguns@gmail.com>

Author: dgunzy

src/kube/api.rs

@@ -387,6 +387,20 @@ pub fn is_version_missing_error(error: &str) -> bool {
         || lower.contains("page not found")
 }
 
+/// Returns true if an error string indicates the user is forbidden (RBAC 403)
+/// from watching this resource type.
+///
+/// A 403 means the credentials are valid but lack permission to list/watch the
+/// resource — e.g. RBAC denies access to a particular CRD. Unlike a transient
+/// outage, this won't recover by retrying within the session, so watchers treat
+/// it like a missing CRD: log it once and stop, rather than flagging the
+/// "watch degraded" banner. kube-rs surfaces these as
+/// "... is forbidden: User \"...\" cannot list resource ..." (HTTP 403).
+pub fn is_forbidden_error(error: &str) -> bool {
+    let lower = error.to_lowercase();
+    lower.contains("forbidden") || lower.contains("403")
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -483,6 +497,28 @@ mod tests {
         assert!(!is_version_missing_error("unauthorized: token expired"));
     }
 
+    #[test]
+    fn test_is_forbidden_error_rbac_message() {
+        // The message kube-rs surfaces when RBAC denies list/watch on a CRD
+        assert!(is_forbidden_error(
+            "failed to perform initial watch: Api error: artifactgenerators.swp.fluxcd.io is forbidden: User \"u\" cannot list resource \"artifactgenerators\" in API group \"swp.fluxcd.io\" at the cluster scope"
+        ));
+    }
+
+    #[test]
+    fn test_is_forbidden_error_numeric_code() {
+        assert!(is_forbidden_error("error 403 from server"));
+    }
+
+    #[test]
+    fn test_is_forbidden_error_unrelated_error() {
+        assert!(!is_forbidden_error("connection refused"));
+        assert!(!is_forbidden_error(
+            "the server could not find the requested resource"
+        ));
+        assert!(!is_forbidden_error("timeout waiting for response"));
+    }
+
     fn api_error(code: u16, message: &str) -> kube::Error {
         kube::Error::Api(Box::new(
             kube::core::Status::failure(message, "NotFound").with_code(code),

src/watcher/mod.rs

@@ -23,7 +23,9 @@ use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 
-use crate::kube::api::{get_flux_api_resources_with_fallback, is_version_missing_error};
+use crate::kube::api::{
+    get_flux_api_resources_with_fallback, is_forbidden_error, is_version_missing_error,
+};
 use crate::models::FluxResourceKind;
 
 /// Maximum interval between watch-reconnect attempts.
@@ -217,7 +219,8 @@ impl ResourceWatcher {
                 let ev = match event {
                     Ok(ev) => ev,
                     Err(e) => {
-                        if is_version_missing_error(&format!("{}", e)) {
+                        let err_str = format!("{}", e);
+                        if is_version_missing_error(&err_str) {
                             // CRD not installed or version not served — stop, don't retry.
                             // Clear any degraded state: this watcher is intentionally
                             // stopping, not reconnecting.
@@ -236,6 +239,26 @@ impl ResourceWatcher {
                             break;
                         }
 
+                        if is_forbidden_error(&err_str) {
+                            // RBAC denies access to this resource — retrying won't help
+                            // within the session, so stop rather than flag the watch
+                            // as degraded. Clear any degraded state from earlier errors.
+                            if error_count > 0 {
+                                let _ = event_tx
+                                    .send(WatchEvent::WatcherRecovered(display_name.clone()));
+                            }
+                            tracing::info!(
+                                "{} watch forbidden by RBAC, stopping watcher: {}",
+                                display_name,
+                                e
+                            );
+                            let _ = event_tx.send(WatchEvent::Error(format!(
+                                "{} watch forbidden by RBAC: {}",
+                                display_name, e
+                            )));
+                            break;
+                        }
+
                         error_count += 1;
                         if error_count == 1 {
                             // First error after healthy operation: flag as degraded
@@ -331,6 +354,16 @@ impl ResourceWatcher {
                 let ev = match event {
                     Ok(ev) => ev,
                     Err(e) => {
+                        if is_forbidden_error(&format!("{}", e)) {
+                            // RBAC denies access — retrying won't help, so stop rather
+                            // than flag the watch as degraded. Clear earlier degraded state.
+                            if error_count > 0 {
+                                let _ = event_tx
+                                    .send(WatchEvent::WatcherRecovered(WATCHER_NAME.to_string()));
+                            }
+                            tracing::info!("Pod watcher forbidden by RBAC, stopping: {}", e);
+                            break;
+                        }
                         error_count += 1;
                         if error_count == 1 {
                             let _ = event_tx
@@ -394,6 +427,16 @@ impl ResourceWatcher {
                 let ev = match event {
                     Ok(ev) => ev,
                     Err(e) => {
+                        if is_forbidden_error(&format!("{}", e)) {
+                            // RBAC denies access — retrying won't help, so stop rather
+                            // than flag the watch as degraded. Clear earlier degraded state.
+                            if error_count > 0 {
+                                let _ = event_tx
+                                    .send(WatchEvent::WatcherRecovered(WATCHER_NAME.to_string()));
+                            }
+                            tracing::info!("Deployment watcher forbidden by RBAC, stopping: {}", e);
+                            break;
+                        }
                         error_count += 1;
                         if error_count == 1 {
                             let _ = event_tx
@@ -490,7 +533,8 @@ impl ResourceWatcher {
                     let ev = match w.next().await {
                         Some(Ok(ev)) => ev,
                         Some(Err(e)) => {
-                            if is_version_missing_error(&format!("{}", e)) && !version_working {
+                            let err_str = format!("{}", e);
+                            if is_version_missing_error(&err_str) && !version_working {
                                 tracing::debug!(
                                     "{} version {} not available, trying next version",
                                     display_name,
@@ -499,6 +543,26 @@ impl ResourceWatcher {
                                 break; // Try next version
                             }
 
+                            if is_forbidden_error(&err_str) {
+                                // RBAC denies access to this resource — every version
+                                // would be forbidden too, so stop rather than flag the
+                                // watch as degraded. Clear any degraded state first.
+                                if degraded_sent {
+                                    let _ = event_tx
+                                        .send(WatchEvent::WatcherRecovered(resource_type.clone()));
+                                }
+                                tracing::info!(
+                                    "{} watch forbidden by RBAC, stopping watcher: {}",
+                                    display_name,
+                                    e
+                                );
+                                let _ = event_tx.send(WatchEvent::Error(format!(
+                                    "{} watch forbidden by RBAC: {}",
+                                    display_name, e
+                                )));
+                                return;
+                            }
+
                             error_count += 1;
                             if !degraded_sent {
                                 degraded_sent = true;