diff --git a/Cargo.lock b/Cargo.lock index 4633cdf62e..bbfcc3e409 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4899,12 +4899,15 @@ dependencies = [ "openshell-core", "openshell-sdk", "rand 0.9.4", + "serde", + "serde_json", "sha2 0.10.9", "thiserror 2.0.20", "tokio", "tonic", "tracing", "tracing-subscriber", + "uuid", ] [[package]] diff --git a/crates/openshell-provider-sync/Cargo.toml b/crates/openshell-provider-sync/Cargo.toml index 6cb67f75d7..70a82656ea 100644 --- a/crates/openshell-provider-sync/Cargo.toml +++ b/crates/openshell-provider-sync/Cargo.toml @@ -24,15 +24,18 @@ k8s-openapi = { workspace = true } kube = { workspace = true } miette = { workspace = true } rand = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } sha2 = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tonic = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +uuid = { workspace = true } [dev-dependencies] -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] } [lints] workspace = true diff --git a/crates/openshell-provider-sync/README.md b/crates/openshell-provider-sync/README.md index 68443d688d..fa328f12e2 100644 --- a/crates/openshell-provider-sync/README.md +++ b/crates/openshell-provider-sync/README.md @@ -43,9 +43,38 @@ accident, and one Secret key may be listed only once. changed keys of that provider as a value-only `UpdateProvider`. The gateway merges them into the provider, and attached sandboxes refetch on their next provider poll, every 10s by default. -- **Running processes.** A process that is already running gets the new value - only when the credential's profile sets `stable_placeholder: true`. Without it, - only processes started after the update see the new value. +- **Restarting affected agents.** After a successful rotation the controller + restarts every sandbox the gateway reports attached to that provider, with + `StopSandbox` then `StartSandbox`, so the agent starts fresh and no open + connection keeps using the old value. Sandboxes of other providers are not + touched, and a sandbox restarts once even when several of its providers + rotate together. A rotation that lands while a sandbox is already restarting + restarts it once more afterwards. Agents must therefore tolerate a restart at + any time: keep state outside the sandbox or resume from it. `--no-restart` + turns this off. +- **Never leaving an agent stopped.** Each restart is a saved state machine: + requested, then stopping, then awaiting running. The request is saved in the + same write as the rotation's fingerprints, and the controller saves that it + owns the stop before it stops anything, so after a controller restart it + resumes every restart where it was. Every gateway failure and every failed + save is retried, backing off from 10 seconds to 5 minutes. A restart is done + only when the sandbox is ready again. This holds with `--state-secret`; + without it a restart in progress is lost when the controller restarts, which + can leave an agent stopped, and the controller warns about it at startup. + At startup a state Secret that cannot be read is retried with backoff, never + replaced. One that does not parse is copied to + `-corrupt-` and reported with a `StateSecretCorrupt` + Warning; every restart entry that still parses is kept and the rest starts + over as a baseline. An agent whose restart entry was lost stays stopped, + because a stop by this controller cannot be told from one by a person. +- **What a restart leaves to others.** A sandbox that is stopped, stopping, + completed, in Error or being deleted when the restart begins is left alone, + and so is one someone else stops after the restart started it. A sandbox that + ends in Error is left to the gateway's own recovery, with an + `AgentRestartFailed` event. One race remains: if someone stops a sandbox in + the moment between the controller reading its phase and stopping it, the + controller starts it again afterwards. The gateway has no conditional stop + to close that window. - **What it leaves alone.** It never creates or deletes providers, never clears a credential when its Secret is deleted or unlabelled, and never touches credential keys that no Secret lists. Providers are still created by whatever @@ -65,10 +94,17 @@ accident, and one Secret key may be listed only once. - **One Secret per credential.** A second live Secret that maps the same workspace, provider and credential key is refused with a `ProviderSyncConflict` event. +- **Rotations, not baselines.** Only a value that replaces one this controller + applied before restarts agents. The first values it sees for a provider are a + baseline: they are sent, and nothing restarts. - **Startup and restarts.** The watch starts with a full list, so every labelled - Secret is synced at startup. State is in memory: a salted fingerprint of each - value sent, never the value. After a restart every Secret is sent once more. - That is a value-only update and does not disturb running sandboxes. + Secret is synced at startup. The controller keeps a salted fingerprint of each + value sent, never the value. With `--state-secret`, the salt and fingerprints + are kept in that Secret, so after a restart unchanged values are recognized: + nothing is re-sent and no agent restarts. Without it, every Secret is sent + once more after a restart as a baseline that restarts nothing; a rotation + that happened while the controller was down then reaches the providers but + restarts no agents. - **Retries.** - A missing provider is retried every 30s. - Transient gateway failures are retried every 10s. @@ -77,7 +113,11 @@ accident, and one Secret key may be listed only once. - An absent provider is not retried; the next change to the Secret is seen. - **Audit.** Each outcome is published as a Kubernetes Event on the Secret (`ProviderSynced`, `ProviderSyncSkipped`, `ProviderNotFound`, - `ProviderSyncConflict`, `InvalidProviderSync`, `ProviderSyncFailed`). The + `ProviderSyncConflict`, `InvalidProviderSync`, `ProviderSyncFailed`, + `AgentRestarted`, `AgentRestartSkipped`, `AgentRestartPending`, + `AgentRestartFailed`). `AgentRestartPending` is a Warning at each retry after + a failure, and when a sandbox takes over 2 minutes to stop or start; repeats + of the same failure aggregate into one Event with a count. The gateway separately audits each `UpdateProvider` with the controller's principal. Neither carries values. @@ -102,15 +142,22 @@ openshell-provider-sync \ | `--default-workspace` | `OPENSHELL_PROVIDER_SYNC_DEFAULT_WORKSPACE` | Workspace when a Secret has no workspace annotation | | `--reporter`, `--reporter-instance` | `OPENSHELL_PROVIDER_SYNC_REPORTER`, `POD_NAME` | Event reporter | | `--no-events` | | Disable Kubernetes Events | +| `--no-restart` | `OPENSHELL_PROVIDER_SYNC_NO_RESTART` | Do not restart agents after a rotation | +| `--state-secret` | `OPENSHELL_PROVIDER_SYNC_STATE_SECRET` | Secret that keeps applied-value fingerprints across restarts | +| `--state-namespace` | `POD_NAMESPACE` | Namespace of the state Secret, usually the controller's own | | `--log-json` | `OPENSHELL_LOG_JSON` | Log as JSON lines | On a gateway that requires authentication, the principal needs `provider:write` -with the workspace `admin` role in every workspace it updates. The token file +with the workspace `admin` role in every workspace it updates, plus +`sandbox:read` and `sandbox:write` there to restart agents. The token file suits a projected ServiceAccount token or a Secret-mounted token that something else rotates on disk. The ServiceAccount needs `get`, `list` and `watch` on Secrets in the watched -namespace, and `create` and `patch` on `events.k8s.io` Events. +namespace, and `create` and `patch` on `events.k8s.io` Events. With +`--state-secret` it also needs `get`, `create` and `patch` on Secrets in its own +namespace, which a `resourceNames` rule can limit to the state Secret for `get` +and `patch`. The gateway client is the Rust SDK, which speaks bearer-token auth only. A gateway that requires mTLS client certificates is not supported yet. @@ -120,5 +167,9 @@ gateway that requires mTLS client certificates is not supported yet. - **One replica.** There is no leader election. Two replicas would send the same idempotent updates twice; that is wasteful but not wrong. - **No activation wait.** The controller does not wait for supervisors to report - that they activated the update. The gateway's mutation id is logged for - correlation. + that they activated the update before restarting. A restarted sandbox fetches + the provider's current values when it starts. The gateway's mutation id is + logged for correlation. +- **Attachment at update time.** The restart set is the gateway's target set for + the update. A sandbox attached to the provider afterwards already starts with + the new values. diff --git a/crates/openshell-provider-sync/src/controller.rs b/crates/openshell-provider-sync/src/controller.rs index b0c2fb026e..9e6bdbe6e4 100644 --- a/crates/openshell-provider-sync/src/controller.rs +++ b/crates/openshell-provider-sync/src/controller.rs @@ -16,9 +16,11 @@ use kube::{Api, Client, Resource, ResourceExt}; use tokio::sync::Mutex; use tracing::{debug, info, warn}; -use crate::gateway::{ProviderUpdater, UpdateError}; +use crate::gateway::{ProviderUpdater, SandboxRestarter, UpdateError}; use crate::mapping::{self, Extracted, MappingError}; use crate::state::{Conflict, SecretRef, SyncState}; +use crate::state::{PendingRestart, RestartStage}; +use crate::state_store::StateStore; /// How long to wait before retrying a provider that does not exist yet. const PROVIDER_MISSING_RETRY: Duration = Duration::from_secs(30); @@ -27,7 +29,8 @@ const TRANSIENT_RETRY: Duration = Duration::from_secs(10); /// How long to wait before retrying a Secret that can only succeed once it, /// the provider or the gateway's configuration changes. const INVALID_RETRY: Duration = Duration::from_mins(5); - +/// How often applied fingerprints of deleted Secrets are dropped. +const PRUNE_INTERVAL: Duration = Duration::from_mins(10); /// The result of syncing one provider. #[derive(Debug, PartialEq, Eq)] pub enum Outcome { @@ -35,6 +38,11 @@ pub enum Outcome { Updated { keys: Vec, mutation_id: String, + /// Sandboxes attached to the provider whose restart this rotation + /// requested. Empty for a baseline: values this controller had not + /// applied before, such as at first sight. A sandbox already + /// restarting is marked to restart once more instead. + restart: Vec, }, /// Nothing changed since the last push. Unchanged, @@ -92,10 +100,31 @@ impl SyncReport { } } +impl SyncReport { + /// Every sandbox to restart, once each even when several of its + /// providers rotated. + pub fn sandboxes_to_restart(&self) -> Vec { + let mut sandboxes: Vec = self + .providers + .iter() + .filter_map(|(_, result)| match result { + Ok(Outcome::Updated { restart, .. }) => Some(restart.iter().cloned()), + _ => None, + }) + .flatten() + .collect(); + sandboxes.sort(); + sandboxes.dedup(); + sandboxes + } +} + /// Shared by every reconcile. pub struct Context { updater: Arc, + restarter: Option>, state: Mutex, + store: Option>, default_workspace: String, } @@ -103,11 +132,29 @@ impl Context { pub fn new(updater: Arc, default_workspace: impl Into) -> Self { Self { updater, + restarter: None, state: Mutex::new(SyncState::new()), + store: None, default_workspace: default_workspace.into(), } } + /// Restart the agents attached to a provider whose values rotated. + #[must_use] + pub fn with_restarter(mut self, restarter: Arc) -> Self { + self.restarter = Some(restarter); + self + } + + /// Resume from persisted state and keep it current, so a restart of this + /// controller does not mistake values it already applied for rotations. + #[must_use] + pub fn with_state(mut self, state: SyncState, store: Arc) -> Self { + self.state = Mutex::new(state); + self.store = Some(store); + self + } + /// Sync one Secret: parse its annotations, then for each provider read /// the mapped keys and push the ones that changed. /// @@ -153,19 +200,45 @@ impl Context { if changed.is_empty() { return Ok(Outcome::Unchanged); } - let mutation_id = self + let rotation = state.has_applied(secret_ref, &target.workspace, provider); + let updated = self .updater .update_credentials(&target.workspace, provider, &changed) .await?; state.commit(secret_ref, &target.workspace, provider, &values); + // Requested in the same state as the commit, so one save + // records both: a controller restart can never see the + // rotation as applied and its restarts as never asked for. + let restart = if rotation && self.restarter.is_some() { + updated + .sandboxes + .into_iter() + .filter(|sandbox| { + state.request_restart(secret_ref, &target.workspace, sandbox) + }) + .collect() + } else { + Vec::new() + }; Ok(Outcome::Updated { keys: changed.keys().cloned().collect(), - mutation_id, + mutation_id: updated.mutation_id, + restart, }) } .await; providers.push((provider.clone(), result)); } + if providers + .iter() + .any(|(_, result)| matches!(result, Ok(Outcome::Updated { .. }))) + && let Err(err) = self.persist(&state).await + { + // In-memory state is still right. If this controller restarts + // before a later save, the rotation is seen again and requested + // again; a restart never stops anything before it is saved. + warn!(error = %err, "failed to persist provider sync state"); + } Ok(SyncReport { workspace: target.workspace, providers, @@ -175,12 +248,84 @@ impl Context { async fn forget(&self, secret_ref: &SecretRef) { self.state.lock().await.forget(secret_ref); } + + async fn persist(&self, state: &SyncState) -> Result<(), String> { + match &self.store { + Some(store) => store.save(&state.to_persisted()).await, + None => Ok(()), + } + } + + pub(crate) async fn restart_stage( + &self, + workspace: &str, + sandbox: &str, + ) -> Option { + self.state.lock().await.restart_stage(workspace, sandbox) + } + + /// Move a restart to `stage` and save it. Fails when the save fails, so + /// a caller never acts on a stage a controller restart would not see. + pub(crate) async fn set_restart_stage( + &self, + workspace: &str, + sandbox: &str, + next: RestartStage, + ) -> Result<(), String> { + let mut state = self.state.lock().await; + let previous = state.restart_stage(workspace, sandbox); + state.set_restart_stage(workspace, sandbox, next); + if let Err(err) = self.persist(&state).await { + if let Some(previous) = previous { + state.set_restart_stage(workspace, sandbox, previous); + } + return Err(err); + } + Ok(()) + } + + /// The sandbox runs again; true when it must restart once more. + pub(crate) async fn finish_restart(&self, workspace: &str, sandbox: &str) -> bool { + let mut state = self.state.lock().await; + let again = state.finish_restart(workspace, sandbox); + if let Err(err) = self.persist(&state).await { + // A lost save leaves a finished restart recorded: after a + // controller restart it sees the sandbox ready and finishes again. + warn!(error = %err, "failed to persist provider sync state"); + } + again + } + + pub(crate) async fn abandon_restart(&self, workspace: &str, sandbox: &str) { + let mut state = self.state.lock().await; + state.abandon_restart(workspace, sandbox); + if let Err(err) = self.persist(&state).await { + warn!(error = %err, "failed to persist provider sync state"); + } + } + + /// Restarts a previous run left in progress. + pub async fn pending_restarts(&self) -> Vec { + self.state.lock().await.pending_restarts() + } + + async fn prune(&self, exists: impl Fn(&SecretRef) -> bool) { + let mut state = self.state.lock().await; + if state.prune_applied(exists) + && let Err(err) = self.persist(&state).await + { + warn!(error = %err, "failed to persist provider sync state"); + } + } } -struct Runtime { - ctx: Context, +pub(crate) struct Runtime { + pub(crate) ctx: Context, store: Store, recorder: Option, + /// Event reasons published, kept only by tests. + #[cfg(test)] + events: std::sync::Mutex>, } fn secret_ref(secret: &Secret) -> SecretRef { @@ -213,7 +358,7 @@ async fn reconcile(secret: Arc, runtime: Arc) -> Result, runtime: Arc) -> Result { + Ok(Outcome::Updated { + keys, + mutation_id, + restart, + }) => { info!( secret = %secret_ref, workspace, provider, keys = ?keys, mutation_id, + restart = ?restart, "provider credentials updated from Secret" ); let note = format!( "Updated {} on provider {workspace}/{provider}", keys.join(", ") ); - publish(&runtime, &secret, EventType::Normal, "ProviderSynced", note).await; + publish( + &runtime, + &secret_ref, + EventType::Normal, + "ProviderSynced", + note, + ) + .await; } Ok(Outcome::Unchanged) => { debug!(secret = %secret_ref, workspace, provider, "provider credentials unchanged"); @@ -257,7 +414,7 @@ async fn reconcile(secret: Arc, runtime: Arc) -> Result, runtime: Arc) -> Result, err: &SyncError, _runtime: Arc) - Action::requeue(err.retry_after()) } -async fn publish(runtime: &Runtime, secret: &Secret, type_: EventType, reason: &str, note: String) { +pub(crate) async fn publish( + runtime: &Runtime, + secret: &SecretRef, + type_: EventType, + reason: &str, + note: String, +) { + #[cfg(test)] + runtime.events.lock().unwrap().push(reason.to_string()); let Some(recorder) = &runtime.recorder else { return; }; @@ -297,7 +480,14 @@ async fn publish(runtime: &Runtime, secret: &Secret, type_: EventType, reason: & action: "UpdateProviderCredentials".to_string(), secondary: None, }; - if let Err(err) = recorder.publish(&event, &secret.object_ref(&())).await { + let reference = k8s_openapi::api::core::v1::ObjectReference { + api_version: Some("v1".to_string()), + kind: Some("Secret".to_string()), + namespace: Some(secret.namespace.clone()), + name: Some(secret.name.clone()), + ..Default::default() + }; + if let Err(err) = recorder.publish(&event, &reference).await { debug!(error = %err, "failed to publish Kubernetes event"); } } @@ -329,6 +519,42 @@ pub async fn run(client: Client, ctx: Context, options: Options) { recorder: options .reporter .map(|reporter| Recorder::new(client, reporter)), + #[cfg(test)] + events: std::sync::Mutex::default(), + }); + // An earlier run may have left restarts in progress, including agents + // it stopped and had not started yet. + if let Some(restarter) = &runtime.ctx.restarter { + for pending in runtime.ctx.pending_restarts().await { + info!( + workspace = pending.workspace, + sandbox = pending.sandbox, + stage = ?pending.stage, + "resuming an agent restart" + ); + tokio::spawn(crate::restart::drive( + Arc::clone(restarter), + Arc::clone(&runtime), + pending.secret, + pending.workspace, + pending.sandbox, + )); + } + } + // Fingerprints of deleted Secrets would otherwise be kept forever: the + // watch does not see a Secret that is gone or no longer labelled. + let pruner = Arc::clone(&runtime); + tokio::spawn(async move { + if pruner.store.wait_until_ready().await.is_err() { + return; + } + loop { + pruner + .ctx + .prune(|secret| is_live(&pruner.store, secret)) + .await; + tokio::time::sleep(PRUNE_INTERVAL).await; + } }); controller .shutdown_on_signal() @@ -349,6 +575,7 @@ mod tests { }; use k8s_openapi::ByteString; use kube::api::ObjectMeta; + use openshell_core::proto::SandboxPhase; use std::collections::BTreeMap; use std::sync::Mutex as StdMutex; @@ -363,6 +590,24 @@ mod tests { struct FakeUpdater { calls: StdMutex>, fail_with: StdMutex>, + /// Sandboxes every provider reports attached; `-agent` + /// when unset. + attached: StdMutex>>, + } + + /// Keeps every save, newest last. + #[derive(Default)] + struct MemoryStore(StdMutex>); + + #[async_trait::async_trait] + impl StateStore for MemoryStore { + async fn load(&self) -> Result, String> { + Ok(self.0.lock().unwrap().last().cloned()) + } + async fn save(&self, state: &crate::state::PersistedState) -> Result<(), String> { + self.0.lock().unwrap().push(state.clone()); + Ok(()) + } } #[async_trait::async_trait] @@ -372,7 +617,7 @@ mod tests { workspace: &str, provider: &str, values: &CredentialValues, - ) -> Result { + ) -> Result { let fail_with = *self.fail_with.lock().unwrap(); if let Some((only, fail)) = fail_with && only.is_none_or(|only| only == provider) @@ -384,7 +629,16 @@ mod tests { provider.to_string(), values.0.clone(), )); - Ok("mutation-1".to_string()) + let sandboxes = self + .attached + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| vec![format!("{provider}-agent")]); + Ok(crate::gateway::Updated { + mutation_id: "mutation-1".to_string(), + sandboxes, + }) } } @@ -549,6 +803,462 @@ mod tests { ); } + const PM_CREDENTIALS: [(&str, &str); 4] = [ + ("SLACK_BOT_TOKEN", "bot"), + ("SLACK_APP_TOKEN", "app"), + ("LINEAR_API_KEY", "lin"), + ("ANTHROPIC_API_KEY", "ant"), + ]; + + fn with( + base: [(&'static str, &'static str); 4], + key: &str, + value: &'static str, + ) -> [(&'static str, &'static str); 4] { + base.map(|(name, current)| { + if name == key { + (name, value) + } else { + (name, current) + } + }) + } + + #[tokio::test] + async fn a_rotation_restarts_only_the_agents_attached_to_its_provider() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default") + .with_restarter(FakeSandbox::new(SandboxPhase::Ready)); + let (r, s) = agent(&PM_CREDENTIALS); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!( + report.sandboxes_to_restart().is_empty(), + "first sight is a baseline, not a rotation" + ); + + let (r, s) = agent(&with(PM_CREDENTIALS, "SLACK_BOT_TOKEN", "bot-2")); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.sandboxes_to_restart(), ["slack-pm-agent"]); + assert!(matches!( + outcome(&report, "slack-pm"), + Ok(Outcome::Updated { restart, .. }) if restart == &["slack-pm-agent"] + )); + assert!(matches!(outcome(&report, "linear"), Ok(Outcome::Unchanged))); + + // Unchanged values restart nothing. + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!(report.sandboxes_to_restart().is_empty()); + } + + #[tokio::test] + async fn an_agent_restarts_once_when_several_of_its_providers_rotate() { + let updater = Arc::new(FakeUpdater::default()); + *updater.attached.lock().unwrap() = Some(vec!["pm".to_string()]); + let ctx = Context::new(updater.clone(), "default") + .with_restarter(FakeSandbox::new(SandboxPhase::Ready)); + let (r, s) = agent(&PM_CREDENTIALS); + ctx.sync(&r, &s, |_| true).await.unwrap(); + + let rotated = with( + with(PM_CREDENTIALS, "SLACK_BOT_TOKEN", "bot-2"), + "LINEAR_API_KEY", + "lin-2", + ); + let (r, s) = agent(&rotated); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.sandboxes_to_restart(), ["pm"]); + } + + #[tokio::test] + async fn persisted_state_keeps_a_controller_restart_from_restarting_agents() { + let store = Arc::new(MemoryStore::default()); + let updater = Arc::new(FakeUpdater::default()); + let first = Context::new(updater.clone(), "default") + .with_restarter(FakeSandbox::new(SandboxPhase::Ready)) + .with_state(SyncState::new(), store.clone()); + let (r, s) = agent(&PM_CREDENTIALS); + first.sync(&r, &s, |_| true).await.unwrap(); + updater.calls.lock().unwrap().clear(); + + // A new controller process resumes from the saved fingerprints. + let persisted = store.load().await.unwrap().expect("state saved"); + let second = Context::new(updater.clone(), "default") + .with_restarter(FakeSandbox::new(SandboxPhase::Ready)) + .with_state(SyncState::from_persisted(persisted), store.clone()); + let report = second.sync(&r, &s, |_| true).await.unwrap(); + assert!(report.sandboxes_to_restart().is_empty()); + assert!( + updater.calls.lock().unwrap().is_empty(), + "nothing re-pushed" + ); + + let (r, s) = agent(&with(PM_CREDENTIALS, "LINEAR_API_KEY", "lin-2")); + let report = second.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.sandboxes_to_restart(), ["linear-agent"]); + } + + /// A sandbox whose phase moves like the gateway's: a stop goes through + /// Stopping to Stopped, a start through Starting to `after_start`. + struct FakeSandbox { + phase: StdMutex>, + after_start: StdMutex, + phase_failures: StdMutex, + stop_failures: StdMutex, + start_failures: StdMutex, + stops: StdMutex, + starts: StdMutex, + } + + impl FakeSandbox { + fn new(phase: SandboxPhase) -> Arc { + Arc::new(Self { + phase: StdMutex::new(Some(phase)), + after_start: StdMutex::new(SandboxPhase::Ready), + phase_failures: StdMutex::new(0), + stop_failures: StdMutex::new(0), + start_failures: StdMutex::new(0), + stops: StdMutex::new(0), + starts: StdMutex::new(0), + }) + } + fn counts(&self) -> (u32, u32) { + (*self.stops.lock().unwrap(), *self.starts.lock().unwrap()) + } + } + + fn fail_once(counter: &StdMutex) -> bool { + let mut remaining = counter.lock().unwrap(); + if *remaining > 0 { + *remaining -= 1; + return true; + } + false + } + + #[async_trait::async_trait] + impl SandboxRestarter for FakeSandbox { + async fn phase(&self, _: &str, _: &str) -> Result, UpdateError> { + if fail_once(&self.phase_failures) { + return Err(UpdateError::Transient( + "Unavailable: gateway down".to_string(), + )); + } + let mut phase = self.phase.lock().unwrap(); + let current = *phase; + *phase = match current { + Some(SandboxPhase::Stopping) => Some(SandboxPhase::Stopped), + Some(SandboxPhase::Starting) => Some(*self.after_start.lock().unwrap()), + other => other, + }; + Ok(current) + } + async fn stop(&self, _: &str, _: &str) -> Result<(), UpdateError> { + if fail_once(&self.stop_failures) { + return Err(UpdateError::Transient( + "Unavailable: gateway down".to_string(), + )); + } + *self.stops.lock().unwrap() += 1; + *self.phase.lock().unwrap() = Some(SandboxPhase::Stopping); + Ok(()) + } + async fn start(&self, _: &str, _: &str) -> Result<(), UpdateError> { + if fail_once(&self.start_failures) { + return Err(UpdateError::Transient( + "Unavailable: gateway down".to_string(), + )); + } + *self.starts.lock().unwrap() += 1; + *self.phase.lock().unwrap() = Some(SandboxPhase::Starting); + Ok(()) + } + } + + /// A store whose saves can be made to fail. + #[derive(Default)] + struct FlakyStore { + inner: MemoryStore, + failing: StdMutex, + } + + #[async_trait::async_trait] + impl StateStore for FlakyStore { + async fn load(&self) -> Result, String> { + self.inner.load().await + } + async fn save(&self, state: &crate::state::PersistedState) -> Result<(), String> { + if *self.failing.lock().unwrap() { + return Err("apiserver unavailable".to_string()); + } + self.inner.save(state).await + } + } + + fn runtime(ctx: Context) -> Arc { + let (store, _writer) = kube::runtime::reflector::store(); + Arc::new(Runtime { + ctx, + store, + recorder: None, + events: StdMutex::default(), + }) + } + + fn restarting_runtime( + sandbox: &Arc, + store: Arc, + state: SyncState, + ) -> Arc { + runtime( + Context::new(Arc::new(FakeUpdater::default()), "default") + .with_restarter(sandbox.clone()) + .with_state(state, store), + ) + } + + fn events(runtime: &Runtime) -> Vec { + runtime.events.lock().unwrap().clone() + } + + async fn drive(sandbox: &Arc, runtime: &Arc) { + let (secret, _) = github("v1"); + crate::restart::drive( + sandbox.clone(), + Arc::clone(runtime), + secret, + "oncall".to_string(), + "triage".to_string(), + ) + .await; + } + + fn requested(at: RestartStage) -> SyncState { + let mut state = SyncState::new(); + let (secret, _) = github("v1"); + assert!(state.request_restart(&secret, "oncall", "triage")); + state.set_restart_stage("oncall", "triage", at); + state + } + + #[tokio::test(start_paused = true)] + async fn a_rotation_records_its_restart_in_the_same_save_and_restarts_the_agent() { + let store = Arc::new(MemoryStore::default()); + let updater = Arc::new(FakeUpdater::default()); + *updater.attached.lock().unwrap() = Some(vec!["triage".to_string()]); + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + let ctx = Context::new(updater, "default") + .with_restarter(sandbox.clone()) + .with_state(SyncState::new(), store.clone()); + let (r, s) = github("v1"); + ctx.sync(&r, &s, |_| true).await.unwrap(); + let (r, s) = github("v2"); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.sandboxes_to_restart(), ["triage"]); + + // The save that records the new fingerprint also records the + // restart, so no crash can apply the rotation and forget the restart. + let saved = SyncState::from_persisted(store.load().await.unwrap().unwrap()); + assert_eq!( + saved.restart_stage("oncall", "triage"), + Some(RestartStage::Requested) + ); + assert!(saved.has_applied(&r, "oncall", "github")); + assert!( + saved + .changed( + &r, + "oncall", + "github", + &CredentialValues(BTreeMap::from([( + "GITHUB_TOKEN".to_string(), + "v2".to_string() + )])) + ) + .0 + .is_empty() + ); + + let runtime = runtime(ctx); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (1, 1)); + assert_eq!(*sandbox.phase.lock().unwrap(), Some(SandboxPhase::Ready)); + assert!(runtime.ctx.pending_restarts().await.is_empty()); + assert_eq!(events(&runtime), ["AgentRestarted"]); + } + + #[tokio::test(start_paused = true)] + async fn a_long_gateway_outage_delays_the_restart_but_never_leaves_the_agent_stopped() { + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + *sandbox.stop_failures.lock().unwrap() = 3; + *sandbox.start_failures.lock().unwrap() = 30; + *sandbox.phase_failures.lock().unwrap() = 2; + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (1, 1)); + assert_eq!(*sandbox.phase.lock().unwrap(), Some(SandboxPhase::Ready)); + assert!(runtime.ctx.pending_restarts().await.is_empty()); + let events = events(&runtime); + assert_eq!(events.last().map(String::as_str), Some("AgentRestarted")); + assert!( + events + .iter() + .filter(|reason| *reason == "AgentRestartPending") + .count() + >= 35 + ); + } + + #[tokio::test(start_paused = true)] + async fn a_requested_restart_resumed_after_a_crash_still_stops_and_starts() { + // Crash after the request was saved but before the stop reached the + // gateway: the agent still runs on old connections. + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!( + sandbox.counts(), + (1, 1), + "a ready sandbox is not mistaken for restarted" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_resumed_stop_starts_the_sandbox_without_stopping_it_again() { + let sandbox = FakeSandbox::new(SandboxPhase::Stopped); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Stopping), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (0, 1)); + assert_eq!(events(&runtime), ["AgentRestarted"]); + } + + #[tokio::test(start_paused = true)] + async fn nothing_is_stopped_until_the_restart_is_saved() { + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + let store = Arc::new(FlakyStore::default()); + *store.failing.lock().unwrap() = true; + let runtime = + restarting_runtime(&sandbox, store.clone(), requested(RestartStage::Requested)); + let task = tokio::spawn({ + let sandbox = sandbox.clone(); + let runtime = runtime.clone(); + async move { drive(&sandbox, &runtime).await } + }); + tokio::time::sleep(Duration::from_mins(30)).await; + assert_eq!(sandbox.counts(), (0, 0), "no stop while the save fails"); + assert_eq!( + runtime.ctx.restart_stage("oncall", "triage").await, + Some(RestartStage::Requested) + ); + + *store.failing.lock().unwrap() = false; + task.await.unwrap(); + assert_eq!(sandbox.counts(), (1, 1)); + } + + #[tokio::test(start_paused = true)] + async fn an_agent_in_error_after_its_start_is_not_reported_restarted() { + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + *sandbox.after_start.lock().unwrap() = SandboxPhase::Error; + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (1, 1)); + assert_eq!(events(&runtime), ["AgentRestartFailed"]); + assert!(runtime.ctx.pending_restarts().await.is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_starting_agent_is_restarted_once_it_runs() { + let sandbox = FakeSandbox::new(SandboxPhase::Starting); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (1, 1)); + assert_eq!(events(&runtime), ["AgentRestarted"]); + } + + #[tokio::test(start_paused = true)] + async fn agents_someone_else_stopped_or_failed_are_left_alone() { + for phase in [ + SandboxPhase::Stopped, + SandboxPhase::Stopping, + SandboxPhase::Completed, + SandboxPhase::Error, + SandboxPhase::Deleting, + ] { + let sandbox = FakeSandbox::new(phase); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (0, 0), "{phase:?}"); + assert_eq!(events(&runtime), ["AgentRestartSkipped"], "{phase:?}"); + } + // Stopped by someone else after this controller started it. + let sandbox = FakeSandbox::new(SandboxPhase::Stopped); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::AwaitingRunning), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (0, 0)); + assert_eq!(events(&runtime), ["AgentRestartSkipped"]); + } + + #[tokio::test(start_paused = true)] + async fn a_rotation_during_a_restart_restarts_the_agent_once_more() { + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + let mut state = requested(RestartStage::Requested); + let (secret, _) = github("v1"); + assert!(!state.request_restart(&secret, "oncall", "triage")); + let runtime = restarting_runtime(&sandbox, Arc::new(MemoryStore::default()), state); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (2, 2)); + assert_eq!(events(&runtime), ["AgentRestarted", "AgentRestarted"]); + } + + #[tokio::test(start_paused = true)] + async fn a_controller_restart_resumes_restarts_from_saved_state() { + let store = Arc::new(MemoryStore::default()); + store + .save(&requested(RestartStage::Stopping).to_persisted()) + .await + .unwrap(); + let persisted = store.load().await.unwrap().unwrap(); + let sandbox = FakeSandbox::new(SandboxPhase::Stopped); + let runtime = restarting_runtime( + &sandbox, + store.clone(), + SyncState::from_persisted(persisted), + ); + assert_eq!(runtime.ctx.pending_restarts().await.len(), 1); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (0, 1)); + let saved = SyncState::from_persisted(store.load().await.unwrap().unwrap()); + assert!(saved.pending_restarts().is_empty()); + } + #[tokio::test] async fn a_failing_provider_does_not_hold_back_the_others() { let updater = Arc::new(FakeUpdater::default()); diff --git a/crates/openshell-provider-sync/src/gateway.rs b/crates/openshell-provider-sync/src/gateway.rs index 106bbaf243..009e9a1ed0 100644 --- a/crates/openshell-provider-sync/src/gateway.rs +++ b/crates/openshell-provider-sync/src/gateway.rs @@ -7,7 +7,10 @@ use std::collections::HashMap; use std::path::PathBuf; use openshell_core::proto::datamodel::v1::{ObjectMeta, Provider}; -use openshell_core::proto::{UpdateProviderRequest, workspace_selector}; +use openshell_core::proto::{ + GetSandboxRequest, ProviderResponse, SandboxPhase, StartSandboxRequest, StopSandboxRequest, + UpdateProviderRequest, workspace_selector, +}; use openshell_sdk::{OpenShellClient, Refresh, RefreshError, RefreshedToken}; use tonic::Code; @@ -43,17 +46,44 @@ impl UpdateError { } } +/// What one provider update reached. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Updated { + pub mutation_id: String, + /// Sandboxes the gateway found attached to the provider when it applied + /// the update, by name. + pub sandboxes: Vec, +} + +impl Updated { + fn from_response(response: ProviderResponse) -> Self { + let mut sandboxes: Vec = response + .target_receipts + .iter() + .filter_map(|receipt| receipt.desired.as_ref()) + .map(|desired| desired.sandbox.clone()) + .filter(|sandbox| !sandbox.is_empty()) + .collect(); + sandboxes.sort(); + sandboxes.dedup(); + Self { + mutation_id: response.mutation_id, + sandboxes, + } + } +} + /// Writes credential values to a provider. #[async_trait::async_trait] pub trait ProviderUpdater: Send + Sync + 'static { /// Merge `values` into the provider's credentials. Keys not in `values` - /// are left unchanged. Returns the gateway's mutation receipt id. + /// are left unchanged. async fn update_credentials( &self, workspace: &str, provider: &str, values: &CredentialValues, - ) -> Result; + ) -> Result; } /// [`ProviderUpdater`] over the gateway's `UpdateProvider` RPC. @@ -101,7 +131,7 @@ impl ProviderUpdater for GatewayUpdater { workspace: &str, provider: &str, values: &CredentialValues, - ) -> Result { + ) -> Result { let mut retried_auth = false; loop { let mut grpc = self @@ -113,7 +143,7 @@ impl ProviderUpdater for GatewayUpdater { .update_provider(update_request(workspace, provider, values)) .await { - Ok(response) => return Ok(response.into_inner().mutation_id), + Ok(response) => return Ok(Updated::from_response(response.into_inner())), Err(status) if status.code() == Code::Unauthenticated && !retried_auth => { retried_auth = true; let refreshed = self @@ -131,6 +161,130 @@ impl ProviderUpdater for GatewayUpdater { } } +/// Gateway calls a restart needs. Each is one attempt; deciding what to do +/// next and retrying is the caller's job (see [`crate::restart`]). +#[async_trait::async_trait] +pub trait SandboxRestarter: Send + Sync + 'static { + /// The sandbox's phase, or `None` when it no longer exists. + async fn phase( + &self, + workspace: &str, + sandbox: &str, + ) -> Result, UpdateError>; + /// Ask the gateway to stop the sandbox. A sandbox that no longer exists + /// is not an error; the next phase read reports it. + async fn stop(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError>; + /// Ask the gateway to start the sandbox, with the same treatment of a + /// sandbox that no longer exists. + async fn start(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError>; +} + +/// Call one gateway RPC, refreshing the bearer token once if it is rejected. +macro_rules! call_refreshing { + ($client:expr, $method:ident, $request:expr) => {{ + let client = $client; + let first = client + .raw_grpc_fresh() + .await + .map_err(|err| UpdateError::Transient(err.to_string()))? + .$method($request.clone()) + .await; + match first { + Err(status) if status.code() == Code::Unauthenticated => { + client + .force_refresh() + .await + .map_err(|err| UpdateError::Transient(err.to_string()))?; + client + .raw_grpc_fresh() + .await + .map_err(|err| UpdateError::Transient(err.to_string()))? + .$method($request) + .await + } + other => other, + } + }}; +} + +/// [`SandboxRestarter`] over the gateway's `GetSandbox`, `StopSandbox` and +/// `StartSandbox`. +pub struct GatewayRestarter { + client: OpenShellClient, +} + +impl GatewayRestarter { + pub fn new(client: OpenShellClient) -> Self { + Self { client } + } +} + +fn sandbox_error(status: &tonic::Status, workspace: &str, sandbox: &str) -> UpdateError { + let message = format!( + "sandbox {workspace}/{sandbox}: {:?}: {}", + status.code(), + status.message() + ); + match status.code() { + Code::InvalidArgument | Code::FailedPrecondition | Code::PermissionDenied => { + UpdateError::Rejected(message) + } + _ => UpdateError::Transient(message), + } +} + +#[async_trait::async_trait] +impl SandboxRestarter for GatewayRestarter { + async fn phase( + &self, + workspace: &str, + sandbox: &str, + ) -> Result, UpdateError> { + let request = GetSandboxRequest { + workspace_scope: Some(workspace_selector(workspace)), + name: sandbox.to_string(), + }; + match call_refreshing!(&self.client, get_sandbox, request) { + Ok(response) => Ok(Some( + response + .into_inner() + .sandbox + .and_then(|sandbox| sandbox.status) + .map_or(SandboxPhase::Unspecified, |status| status.phase()), + )), + Err(status) if status.code() == Code::NotFound => Ok(None), + Err(status) => Err(sandbox_error(&status, workspace, sandbox)), + } + } + + async fn stop(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError> { + // A request id makes a retried call a replay, not a repeat. + let stop = StopSandboxRequest { + workspace_scope: Some(workspace_selector(workspace)), + name: sandbox.to_string(), + request_id: uuid::Uuid::new_v4().to_string(), + }; + match call_refreshing!(&self.client, stop_sandbox, stop) { + Ok(_) => Ok(()), + Err(status) if status.code() == Code::NotFound => Ok(()), + Err(status) => Err(sandbox_error(&status, workspace, sandbox)), + } + } + + async fn start(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError> { + let start = StartSandboxRequest { + workspace_scope: Some(workspace_selector(workspace)), + name: sandbox.to_string(), + request_id: uuid::Uuid::new_v4().to_string(), + }; + match call_refreshing!(&self.client, start_sandbox, start) { + Ok(_) => Ok(()), + Err(status) if status.code() == Code::NotFound => Ok(()), + Err(status) => Err(sandbox_error(&status, workspace, sandbox)), + } + } +} + /// Bearer token read from a file, re-read whenever the client refreshes. /// /// Suits a projected `ServiceAccount` token or a Secret-mounted token that @@ -189,6 +343,40 @@ mod tests { assert_eq!(request.workspace_scope, Some(workspace_selector("oncall"))); } + #[test] + fn update_reports_the_attached_sandboxes_once_each() { + use openshell_core::proto::{ProviderDesiredIdentity, ProviderMutationReceipt}; + let receipt = |sandbox: &str| ProviderMutationReceipt { + desired: Some(ProviderDesiredIdentity { + sandbox: sandbox.to_string(), + ..Default::default() + }), + ..Default::default() + }; + let updated = Updated::from_response(ProviderResponse { + mutation_id: "m".to_string(), + target_receipts: vec![ + receipt("pm"), + receipt("gtm-bot"), + receipt("pm"), + receipt(""), + ], + ..Default::default() + }); + assert_eq!(updated.mutation_id, "m"); + assert_eq!(updated.sandboxes, ["gtm-bot", "pm"]); + } + + #[test] + fn sandbox_errors_name_the_sandbox() { + let error = sandbox_error(&tonic::Status::failed_precondition("starting"), "w", "pm"); + assert!( + matches!(&error, UpdateError::Rejected(message) if message.contains("sandbox w/pm")) + ); + let error = sandbox_error(&tonic::Status::unavailable("down"), "w", "pm"); + assert!(matches!(error, UpdateError::Transient(_))); + } + #[test] fn classifies_gateway_errors() { let not_found = UpdateError::from_status(&tonic::Status::not_found("x"), "w", "p"); diff --git a/crates/openshell-provider-sync/src/lib.rs b/crates/openshell-provider-sync/src/lib.rs index 5eeeac09f0..dab72e615b 100644 --- a/crates/openshell-provider-sync/src/lib.rs +++ b/crates/openshell-provider-sync/src/lib.rs @@ -6,10 +6,9 @@ //! Secrets managed outside `OpenShell` (for example by External Secrets //! Operator) opt in with a label and name the providers and credential keys //! they back. When a Secret's mapped values change, the controller sends the -//! changed keys to the gateway as a value-only `UpdateProvider`. Running -//! sandboxes pick the new values up on their next provider poll, and -//! credentials with `stable_placeholder: true` reach already-running -//! processes too. +//! changed keys to the gateway as a value-only `UpdateProvider`, then +//! restarts the sandboxes attached to that provider so their agents start +//! fresh with the new values. Agents of other providers are not touched. //! //! The controller never creates or deletes providers, and never clears a //! credential when its Secret goes away. @@ -17,4 +16,6 @@ pub mod controller; pub mod gateway; pub mod mapping; +pub mod restart; pub mod state; +pub mod state_store; diff --git a/crates/openshell-provider-sync/src/main.rs b/crates/openshell-provider-sync/src/main.rs index ddb8a0eea3..29b07ff263 100644 --- a/crates/openshell-provider-sync/src/main.rs +++ b/crates/openshell-provider-sync/src/main.rs @@ -7,10 +7,12 @@ use std::path::PathBuf; use std::sync::Arc; use clap::Parser; -use kube::runtime::events::Reporter; +use kube::runtime::events::{Event, EventType, Recorder, Reporter}; use miette::{IntoDiagnostic, Result, WrapErr}; use openshell_provider_sync::controller::{self, Context, Options}; -use openshell_provider_sync::gateway::{GatewayUpdater, TokenFile}; +use openshell_provider_sync::gateway::{GatewayRestarter, GatewayUpdater, TokenFile}; +use openshell_provider_sync::state::{PersistedState, SyncState}; +use openshell_provider_sync::state_store::{LoadError, SecretStateStore}; use openshell_sdk::{AuthConfig, ClientConfig, OpenShellClient}; use tracing_subscriber::EnvFilter; @@ -63,6 +65,21 @@ struct Cli { #[arg(long)] no_events: bool, + /// Do not restart the agents attached to a provider whose values rotated. + /// They then keep running with connections opened on the old values. + #[arg(long, env = "OPENSHELL_PROVIDER_SYNC_NO_RESTART")] + no_restart: bool, + + /// Secret, in this controller's own namespace, that keeps fingerprints of + /// applied values across restarts. Without it, the first sync after a + /// restart is a baseline that restarts no agents. + #[arg(long, env = "OPENSHELL_PROVIDER_SYNC_STATE_SECRET")] + state_secret: Option, + + /// Namespace of the state Secret; usually this pod's namespace. + #[arg(long, env = "POD_NAMESPACE", requires = "state_secret")] + state_namespace: Option, + /// Log as JSON lines. #[arg(long, env = "OPENSHELL_LOG_JSON")] log_json: bool, @@ -100,21 +117,42 @@ async fn main() -> Result<()> { .wrap_err("build Kubernetes client")?; let reporter = (!cli.no_events).then_some(Reporter { - controller: cli.reporter, - instance: cli.reporter_instance, + controller: cli.reporter.clone(), + instance: cli.reporter_instance.clone(), }); + let mut context = Context::new( + Arc::new(GatewayUpdater::new(gateway.clone())), + cli.default_workspace.clone(), + ); + if !cli.no_restart { + context = context.with_restarter(Arc::new(GatewayRestarter::new(gateway))); + if cli.state_secret.is_none() { + tracing::warn!( + "agent restarts are on without --state-secret: a restart in progress is \ + forgotten if this controller restarts, which can leave an agent stopped" + ); + } + } + if let Some(name) = &cli.state_secret { + let namespace = cli.state_namespace.clone().ok_or_else(|| { + miette::miette!("--state-secret needs --state-namespace or POD_NAMESPACE") + })?; + let store = SecretStateStore::new(kube_client.clone(), &namespace, name.clone()); + let state = load_state(&store, &kube_client, &namespace, reporter.as_ref()).await; + context = context.with_state(state, Arc::new(store)); + } + tracing::info!( gateway = %cli.gateway, namespace = cli.namespace.as_deref().unwrap_or(""), default_workspace = %cli.default_workspace, + restart_agents = !cli.no_restart, + state_secret = cli.state_secret.as_deref().unwrap_or(""), "starting provider sync" ); controller::run( kube_client, - Context::new( - Arc::new(GatewayUpdater::new(gateway)), - cli.default_workspace, - ), + context, Options { namespace: cli.namespace, reporter, @@ -124,6 +162,67 @@ async fn main() -> Result<()> { Ok(()) } +/// Load saved state, never starting over state that is only unreadable for +/// now, and never stopping all syncing over state that is corrupt. +async fn load_state( + store: &SecretStateStore, + client: &kube::Client, + namespace: &str, + reporter: Option<&Reporter>, +) -> SyncState { + let mut backoff = std::time::Duration::from_secs(5); + loop { + match store.load_checked().await { + Ok(Some(persisted)) => return SyncState::from_persisted(persisted), + Ok(None) => return SyncState::new(), + Err(LoadError::Unavailable(err)) => { + // Starting fresh now would overwrite the saved state, and + // with it any restart in progress. + tracing::warn!(error = %err, retry_in_secs = backoff.as_secs(), "provider sync state unavailable; retrying"); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(std::time::Duration::from_mins(5)); + } + Err(LoadError::Corrupt { raw, error }) => { + let (salvaged, recovered) = PersistedState::salvage(&raw); + let copy = store.quarantine(&raw).await; + let kept = match © { + Ok(name) => format!("kept in Secret {name}"), + Err(err) => format!("not kept ({err})"), + }; + let note = format!( + "State in Secret {} does not parse ({error}); {kept}. Starting over: the next values are a \ + baseline that restarts no agents. Recovered {recovered} restart(s) in progress; any \ + other agent this controller had stopped stays stopped, because a stop by this \ + controller cannot be told from one by a person", + store.name() + ); + tracing::warn!(note, "provider sync state is corrupt"); + if let Some(reporter) = reporter { + let recorder = Recorder::new(client.clone(), reporter.clone()); + let reference = k8s_openapi::api::core::v1::ObjectReference { + api_version: Some("v1".to_string()), + kind: Some("Secret".to_string()), + namespace: Some(namespace.to_string()), + name: Some(store.name().to_string()), + ..Default::default() + }; + let event = Event { + type_: EventType::Warning, + reason: "StateSecretCorrupt".to_string(), + note: Some(note), + action: "LoadState".to_string(), + secondary: None, + }; + if let Err(err) = recorder.publish(&event, &reference).await { + tracing::debug!(error = %err, "failed to publish Kubernetes event"); + } + } + return SyncState::from_persisted(salvaged); + } + } + } +} + fn init_tracing(json: bool) { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let builder = tracing_subscriber::fmt().with_env_filter(filter); diff --git a/crates/openshell-provider-sync/src/restart.rs b/crates/openshell-provider-sync/src/restart.rs new file mode 100644 index 0000000000..79b5772380 --- /dev/null +++ b/crates/openshell-provider-sync/src/restart.rs @@ -0,0 +1,284 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Restarting an agent after its credentials rotate, without ever leaving it +//! stopped. +//! +//! A restart is a persisted state machine, one per sandbox: +//! +//! - **Requested**: a rotation needs the agent restarted. Nothing is stopped +//! until the next stage has been saved. +//! - **Stopping**: this controller stops the sandbox, then starts it. +//! - **`AwaitingRunning`**: the start was accepted; it is done when the sandbox +//! is ready. +//! +//! Every gateway failure is retried with capped backoff. The stage is read +//! from the sandbox's phase on every step, so a reply lost in a crash or a +//! network failure never makes a step repeat wrongly, and a controller +//! restart resumes each restart where it was. A sandbox that someone else +//! stops or deletes, or that fails on its own, is left to them. + +use std::sync::Arc; +use std::time::Duration; + +use kube::runtime::events::EventType; +use openshell_core::proto::SandboxPhase; +use tracing::{info, warn}; + +use crate::controller::{Runtime, publish}; +use crate::gateway::SandboxRestarter; +use crate::state::{RestartStage, SecretRef}; + +/// First and longest wait between attempts after a failure. Retried until +/// it succeeds. +const BACKOFF_FIRST: Duration = Duration::from_secs(10); +const BACKOFF_MAX: Duration = Duration::from_mins(5); +/// How often to look again while the sandbox changes phase, and how long a +/// phase may take before each look is reported. +const POLL: Duration = Duration::from_secs(5); +const SLOW: Duration = Duration::from_mins(2); + +/// What one look at the sandbox decided. +enum Step { + /// Moved to the next stage; look again now. + Advanced, + /// Waiting for the sandbox to change phase. + Waiting(&'static str), + /// A gateway call or a save failed; retry after backoff. + Failed(String), + /// The sandbox runs again. + Done, + /// The restart ends without this controller starting the sandbox. + Abandoned { + type_: EventType, + reason: &'static str, + note: String, + }, +} + +/// Drive the restart of one sandbox to its end. Started for each new request +/// and, at startup, for each restart a previous run left in progress. +pub(crate) async fn drive( + restarter: Arc, + runtime: Arc, + secret: SecretRef, + workspace: String, + sandbox: String, +) { + let ctx = &runtime.ctx; + let mut backoff = BACKOFF_FIRST; + let mut waiting_since: Option<(&'static str, tokio::time::Instant)> = None; + loop { + let Some(stage) = ctx.restart_stage(&workspace, &sandbox).await else { + return; + }; + let step = match restarter.phase(&workspace, &sandbox).await { + Err(err) => Step::Failed(err.to_string()), + Ok(phase) => { + step( + restarter.as_ref(), + &runtime, + stage, + phase, + &workspace, + &sandbox, + ) + .await + } + }; + let delay = match step { + Step::Advanced => { + backoff = BACKOFF_FIRST; + waiting_since = None; + continue; + } + Step::Waiting(what) => { + let since = match waiting_since { + Some((previous, since)) if previous == what => since, + _ => { + let now = tokio::time::Instant::now(); + waiting_since = Some((what, now)); + now + } + }; + if since.elapsed() >= SLOW { + let note = format!( + "Restart of sandbox {workspace}/{sandbox} is waiting: it is still {what}" + ); + publish( + &runtime, + &secret, + EventType::Warning, + "AgentRestartPending", + note, + ) + .await; + waiting_since = Some((what, tokio::time::Instant::now())); + } + POLL + } + Step::Failed(error) => { + warn!( + secret = %secret, + workspace, + sandbox, + ?stage, + retry_in_secs = backoff.as_secs(), + error, + "agent restart step failed; retrying" + ); + // No counter or delay in the note, so repeats aggregate into + // one Event with a count. + let note = format!( + "Restart of sandbox {workspace}/{sandbox} is retrying after a failure: {error}" + ); + publish( + &runtime, + &secret, + EventType::Warning, + "AgentRestartPending", + note, + ) + .await; + let delay = backoff; + backoff = (backoff * 2).min(BACKOFF_MAX); + delay + } + Step::Done => { + let again = ctx.finish_restart(&workspace, &sandbox).await; + info!(secret = %secret, workspace, sandbox, again, "restarted agent after credential rotation"); + let note = if again { + format!( + "Restarted sandbox {workspace}/{sandbox}; restarting once more for a later rotation" + ) + } else { + format!("Restarted sandbox {workspace}/{sandbox} after its credentials rotated") + }; + publish(&runtime, &secret, EventType::Normal, "AgentRestarted", note).await; + if !again { + return; + } + backoff = BACKOFF_FIRST; + waiting_since = None; + continue; + } + Step::Abandoned { + type_, + reason, + note, + } => { + ctx.abandon_restart(&workspace, &sandbox).await; + info!(secret = %secret, workspace, sandbox, reason, note, "agent restart ended"); + publish(&runtime, &secret, type_, reason, note).await; + return; + } + }; + tokio::time::sleep(delay).await; + } +} + +fn skipped(workspace: &str, sandbox: &str, why: &str) -> Step { + Step::Abandoned { + type_: EventType::Normal, + reason: "AgentRestartSkipped", + note: format!("Did not restart sandbox {workspace}/{sandbox}: {why}"), + } +} + +fn failed(workspace: &str, sandbox: &str, why: &str) -> Step { + Step::Abandoned { + type_: EventType::Warning, + reason: "AgentRestartFailed", + note: format!("Restart of sandbox {workspace}/{sandbox} ended: {why}"), + } +} + +async fn step( + restarter: &dyn SandboxRestarter, + runtime: &Runtime, + stage: RestartStage, + phase: Option, + workspace: &str, + sandbox: &str, +) -> Step { + use SandboxPhase as P; + let ctx = &runtime.ctx; + match stage { + RestartStage::Requested => match phase { + None | Some(P::Deleting) => skipped(workspace, sandbox, "it was deleted"), + Some(P::Stopped | P::Stopping | P::Completed) => skipped( + workspace, + sandbox, + "it is not running, and was stopped by someone else", + ), + Some(P::Error) => skipped( + workspace, + sandbox, + "it is in Error; it fetches the new values when it next starts", + ), + // It may have fetched values from before the rotation; restart it + // once it runs. + Some(P::Provisioning | P::Starting) => Step::Waiting("starting"), + Some(P::Ready | P::Unknown | P::Unspecified) => { + // Nothing is stopped until this controller has saved that it + // owns the stop, so it can never forget to start it. + match ctx + .set_restart_stage(workspace, sandbox, RestartStage::Stopping) + .await + { + Ok(()) => Step::Advanced, + Err(err) => Step::Failed(format!("could not record the restart: {err}")), + } + } + }, + RestartStage::Stopping => match phase { + None | Some(P::Deleting) => failed(workspace, sandbox, "it was deleted while stopped"), + Some(P::Stopping) => Step::Waiting("stopping"), + Some(P::Stopped | P::Completed) => match restarter.start(workspace, sandbox).await { + Ok(()) => { + // A lost save only means a resumed restart sees the + // sandbox starting or ready in this stage and stops it + // once more: an extra restart, never a stopped agent. + if let Err(err) = ctx + .set_restart_stage(workspace, sandbox, RestartStage::AwaitingRunning) + .await + { + warn!(workspace, sandbox, error = %err, "could not record the start"); + } + Step::Advanced + } + Err(err) => Step::Failed(err.to_string()), + }, + // The gateway refuses to stop a sandbox that is still starting. + Some(P::Provisioning | P::Starting) => Step::Waiting("starting"), + Some(P::Error) => failed( + workspace, + sandbox, + "it is in Error after being stopped; the gateway's own recovery owns it now", + ), + Some(P::Ready | P::Unknown | P::Unspecified) => { + match restarter.stop(workspace, sandbox).await { + Ok(()) => Step::Waiting("stopping"), + Err(err) => Step::Failed(err.to_string()), + } + } + }, + RestartStage::AwaitingRunning => match phase { + Some(P::Ready) => Step::Done, + Some(P::Provisioning | P::Starting | P::Unknown | P::Unspecified) => { + Step::Waiting("starting") + } + None | Some(P::Deleting) => skipped(workspace, sandbox, "it was deleted"), + Some(P::Stopped | P::Stopping | P::Completed) => skipped( + workspace, + sandbox, + "someone else stopped it after it started", + ), + Some(P::Error) => failed( + workspace, + sandbox, + "it is in Error after starting; the gateway's own recovery owns it now", + ), + }, + } +} diff --git a/crates/openshell-provider-sync/src/state.rs b/crates/openshell-provider-sync/src/state.rs index fece7bbf6a..2ee508769f 100644 --- a/crates/openshell-provider-sync/src/state.rs +++ b/crates/openshell-provider-sync/src/state.rs @@ -3,10 +3,12 @@ //! What has been pushed to the gateway, so only changed keys are sent. //! -//! State is in memory only. Values are never stored: each applied value is -//! kept as a salted SHA-256 fingerprint, and the salt is random per process so -//! a fingerprint is useless outside it. After a restart every Secret is pushed -//! once more, which is a value-only update and harmless to running sandboxes. +//! Values are never stored: each applied value is kept as a salted SHA-256 +//! fingerprint. The salt and fingerprints can be persisted (see +//! [`PersistedState`]) so a restart of this controller recognizes values it +//! already applied, and does not treat them as rotations that restart agents. +//! Without persisted state every Secret is pushed once more after a restart, +//! as a baseline that restarts nothing. use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt; @@ -16,7 +18,9 @@ use sha2::{Digest, Sha256}; use crate::mapping::{CredentialValues, KeyMap, SyncTarget}; /// `namespace/name` of a Secret. -#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] +#[derive( + Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, serde::Serialize, serde::Deserialize, +)] pub struct SecretRef { pub namespace: String, pub name: String, @@ -67,6 +71,109 @@ pub struct SyncState { salt: [u8; 32], applied: HashMap, owners: HashMap, + restarts: BTreeMap<(String, String), Restart>, +} + +/// How far a restart has got. Persisted before each step, so a controller +/// restart resumes it and never forgets a sandbox it stopped. +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RestartStage { + /// A rotation needs this agent restarted; nothing has been stopped yet. + Requested, + /// This controller is stopping the sandbox, or has stopped it. + Stopping, + /// The start was accepted; waiting for the sandbox to run. + AwaitingRunning, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Restart { + secret: SecretRef, + stage: RestartStage, + /// Another rotation landed while this restart was under way, so the + /// agent may have started on the earlier value: restart once more. + rerun: bool, +} + +/// One sandbox restart in progress. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct PendingRestart { + pub workspace: String, + pub sandbox: String, + /// The Secret whose rotation caused the restart, for its Events. + pub secret: SecretRef, + pub stage: RestartStage, + #[serde(default)] + pub rerun: bool, +} + +/// Fingerprints of applied values, kept across controller restarts. +/// +/// Holds no values. It is stored in a Secret because the salt and the +/// fingerprints together would let anyone who reads them test guesses. +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct PersistedState { + salt: String, + applied: Vec, + #[serde(default)] + restarts: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +struct PersistedApplied { + namespace: String, + secret: String, + provider: String, + workspace: String, + fingerprints: BTreeMap, +} + +impl PersistedState { + /// Recover what can be recovered from state that does not parse: every + /// restart entry that still parses on its own. Fingerprints are dropped, + /// which only makes the next values a baseline; restarts are kept because + /// losing one could leave an agent stopped. Returns the state and how + /// many restarts were recovered. + pub fn salvage(raw: &[u8]) -> (Self, usize) { + let restarts: Vec = serde_json::from_slice::(raw) + .ok() + .and_then(|value| value.get("restarts").cloned()) + .and_then(|restarts| restarts.as_array().cloned()) + .unwrap_or_default() + .into_iter() + .filter_map(|entry| serde_json::from_value(entry).ok()) + .collect(); + let recovered = restarts.len(); + ( + Self { + restarts, + ..Self::default() + }, + recovered, + ) + } +} + +fn to_hex(bytes: &[u8; 32]) -> String { + use std::fmt::Write as _; + bytes + .iter() + .fold(String::with_capacity(64), |mut hex, byte| { + let _ = write!(hex, "{byte:02x}"); + hex + }) +} + +fn from_hex(text: &str) -> Option<[u8; 32]> { + if text.len() != 64 { + return None; + } + let mut bytes = [0u8; 32]; + for (index, byte) in bytes.iter_mut().enumerate() { + *byte = u8::from_str_radix(text.get(index * 2..index * 2 + 2)?, 16).ok()?; + } + Some(bytes) } impl Default for SyncState { @@ -91,9 +198,189 @@ impl SyncState { salt: rand::random(), applied: HashMap::new(), owners: HashMap::new(), + restarts: BTreeMap::new(), + } + } + + /// Resume from persisted fingerprints. Entries that do not parse are + /// dropped, which only makes their next value a baseline. + pub fn from_persisted(persisted: PersistedState) -> Self { + // Restarts in progress are kept whatever else is unreadable: losing + // one could leave an agent stopped. + let restarts = persisted + .restarts + .into_iter() + .map(|pending| { + ( + (pending.workspace, pending.sandbox), + Restart { + secret: pending.secret, + stage: pending.stage, + rerun: pending.rerun, + }, + ) + }) + .collect(); + let Some(salt) = from_hex(&persisted.salt) else { + return Self { + restarts, + ..Self::new() + }; + }; + let applied = persisted + .applied + .into_iter() + .filter_map(|entry| { + let fingerprints = entry + .fingerprints + .into_iter() + .map(|(key, fingerprint)| Some((key, from_hex(&fingerprint)?))) + .collect::>>()?; + Some(( + ( + SecretRef { + namespace: entry.namespace, + name: entry.secret, + }, + entry.provider, + ), + Applied { + workspace: entry.workspace, + fingerprints, + }, + )) + }) + .collect(); + Self { + salt, + applied, + owners: HashMap::new(), + restarts, + } + } + + /// The fingerprints to persist. Claims are rebuilt from the watch. + pub fn to_persisted(&self) -> PersistedState { + let mut applied: Vec = self + .applied + .iter() + .map(|((secret, provider), applied)| PersistedApplied { + namespace: secret.namespace.clone(), + secret: secret.name.clone(), + provider: provider.clone(), + workspace: applied.workspace.clone(), + fingerprints: applied + .fingerprints + .iter() + .map(|(key, fingerprint)| (key.clone(), to_hex(fingerprint))) + .collect(), + }) + .collect(); + applied.sort_by(|left, right| { + (&left.namespace, &left.secret, &left.provider).cmp(&( + &right.namespace, + &right.secret, + &right.provider, + )) + }); + PersistedState { + salt: to_hex(&self.salt), + applied, + restarts: self.pending_restarts(), + } + } + + /// Ask for `sandbox` to be restarted. Returns true when this starts a + /// new restart; when one is already under way it is marked to run once + /// more, so the agent cannot keep a value from before this rotation. + pub fn request_restart(&mut self, secret: &SecretRef, workspace: &str, sandbox: &str) -> bool { + let key = (workspace.to_string(), sandbox.to_string()); + if let Some(restart) = self.restarts.get_mut(&key) { + restart.rerun = true; + return false; + } + self.restarts.insert( + key, + Restart { + secret: secret.clone(), + stage: RestartStage::Requested, + rerun: false, + }, + ); + true + } + + pub fn restart_stage(&self, workspace: &str, sandbox: &str) -> Option { + self.restarts + .get(&(workspace.to_string(), sandbox.to_string())) + .map(|restart| restart.stage) + } + + pub fn set_restart_stage(&mut self, workspace: &str, sandbox: &str, stage: RestartStage) { + if let Some(restart) = self + .restarts + .get_mut(&(workspace.to_string(), sandbox.to_string())) + { + restart.stage = stage; } } + /// The sandbox runs again. Returns true when another rotation asked for + /// one more restart, which is then requested afresh. + pub fn finish_restart(&mut self, workspace: &str, sandbox: &str) -> bool { + let key = (workspace.to_string(), sandbox.to_string()); + match self.restarts.get_mut(&key) { + Some(restart) if restart.rerun => { + restart.rerun = false; + restart.stage = RestartStage::Requested; + true + } + Some(_) => { + self.restarts.remove(&key); + false + } + None => false, + } + } + + /// The restart ends without the sandbox running, as when someone else + /// stopped or deleted it. + pub fn abandon_restart(&mut self, workspace: &str, sandbox: &str) { + self.restarts + .remove(&(workspace.to_string(), sandbox.to_string())); + } + + /// Restarts in progress. + pub fn pending_restarts(&self) -> Vec { + self.restarts + .iter() + .map(|((workspace, sandbox), restart)| PendingRestart { + workspace: workspace.clone(), + sandbox: sandbox.clone(), + secret: restart.secret.clone(), + stage: restart.stage, + rerun: restart.rerun, + }) + .collect() + } + + /// Drop applied fingerprints of Secrets that no longer exist, so + /// persisted state does not grow without bound. Returns whether any were + /// dropped. Claims are left to the watch. + pub fn prune_applied(&mut self, exists: impl Fn(&SecretRef) -> bool) -> bool { + let before = self.applied.len(); + self.applied.retain(|(secret, _), _| exists(secret)); + self.applied.len() != before + } + + /// Whether `secret` has applied values to `provider` before, so a change + /// now is a rotation rather than a baseline. + pub fn has_applied(&self, secret: &SecretRef, workspace: &str, provider: &str) -> bool { + self.applied + .get(&(secret.clone(), provider.to_string())) + .is_some_and(|applied| applied.workspace == workspace) + } + fn fingerprint(&self, value: &str) -> [u8; 32] { let mut hasher = Sha256::new(); hasher.update(self.salt); @@ -408,6 +695,120 @@ mod tests { ); } + #[test] + fn persisted_state_recognizes_applied_values_after_a_restart() { + let mut state = SyncState::new(); + let secret = secret("a"); + let applied = values(&[("TOKEN", "v1")]); + state.commit(&secret, "w", "p", &applied); + assert!(state.has_applied(&secret, "w", "p")); + + let persisted = state.to_persisted(); + let serialized = serde_json::to_string(&persisted).unwrap(); + assert!(!serialized.contains("v1"), "values are never persisted"); + let restarted = SyncState::from_persisted(serde_json::from_str(&serialized).unwrap()); + + assert!(restarted.has_applied(&secret, "w", "p")); + assert!(changed_keys(&restarted.changed(&secret, "w", "p", &applied)).is_empty()); + assert_eq!( + changed_keys(&restarted.changed(&secret, "w", "p", &values(&[("TOKEN", "v2")]))), + ["TOKEN"] + ); + assert!(!restarted.has_applied(&secret, "other", "p")); + } + + #[test] + fn a_restart_survives_a_controller_restart_at_every_stage() { + let mut state = SyncState::new(); + assert!(state.request_restart(&secret("pm"), "pm", "pm")); + for stage in [ + RestartStage::Requested, + RestartStage::Stopping, + RestartStage::AwaitingRunning, + ] { + state.set_restart_stage("pm", "pm", stage); + let restarted = SyncState::from_persisted(state.to_persisted()); + assert_eq!(restarted.restart_stage("pm", "pm"), Some(stage)); + } + // Even with an unreadable salt, restarts in progress are kept. + let mut persisted = state.to_persisted(); + persisted.salt = "not hex".to_string(); + assert_eq!( + SyncState::from_persisted(persisted).restart_stage("pm", "pm"), + Some(RestartStage::AwaitingRunning) + ); + } + + #[test] + fn a_rotation_during_a_restart_restarts_once_more() { + let mut state = SyncState::new(); + assert!(state.request_restart(&secret("pm"), "pm", "pm")); + state.set_restart_stage("pm", "pm", RestartStage::AwaitingRunning); + assert!( + !state.request_restart(&secret("pm"), "pm", "pm"), + "one restart at a time" + ); + assert!(state.finish_restart("pm", "pm"), "runs once more"); + assert_eq!( + state.restart_stage("pm", "pm"), + Some(RestartStage::Requested) + ); + assert!(!state.finish_restart("pm", "pm")); + assert!(state.pending_restarts().is_empty()); + } + + #[test] + fn prune_drops_fingerprints_of_deleted_secrets() { + let mut state = SyncState::new(); + state.commit(&secret("kept"), "w", "p", &values(&[("A", "1")])); + state.commit(&secret("gone"), "w", "p", &values(&[("A", "1")])); + assert!(state.prune_applied(|secret| secret.name == "kept")); + assert!(state.has_applied(&secret("kept"), "w", "p")); + assert!(!state.has_applied(&secret("gone"), "w", "p")); + assert!(!state.prune_applied(|_| true)); + } + + #[test] + fn corrupt_state_keeps_every_restart_that_still_parses() { + let mut state = SyncState::new(); + state.commit(&secret("a"), "w", "p", &values(&[("A", "1")])); + assert!(state.request_restart(&secret("a"), "w", "pm")); + state.set_restart_stage("w", "pm", RestartStage::Stopping); + let mut raw = serde_json::to_value(state.to_persisted()).unwrap(); + // Corrupt everything but the one restart. + raw["salt"] = serde_json::json!(42); + raw["applied"] = serde_json::json!("garbage"); + raw["restarts"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({"workspace": "w"})); + let raw = serde_json::to_vec(&raw).unwrap(); + assert!(serde_json::from_slice::(&raw).is_err()); + + let (salvaged, recovered) = PersistedState::salvage(&raw); + assert_eq!(recovered, 1); + let restored = SyncState::from_persisted(salvaged); + assert_eq!( + restored.restart_stage("w", "pm"), + Some(RestartStage::Stopping) + ); + assert!( + !restored.has_applied(&secret("a"), "w", "p"), + "fingerprints become a baseline" + ); + + assert_eq!(PersistedState::salvage(b"not json").1, 0); + } + + #[test] + fn unreadable_persisted_state_starts_fresh() { + let restarted = SyncState::from_persisted(PersistedState { + salt: "not hex".to_string(), + ..Default::default() + }); + assert!(!restarted.has_applied(&secret("a"), "w", "p")); + } + #[test] fn forget_releases_claims() { let mut state = SyncState::new(); diff --git a/crates/openshell-provider-sync/src/state_store.rs b/crates/openshell-provider-sync/src/state_store.rs new file mode 100644 index 0000000000..8c50eb867a --- /dev/null +++ b/crates/openshell-provider-sync/src/state_store.rs @@ -0,0 +1,132 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Keeping applied-value fingerprints across controller restarts. +//! +//! Without them every Secret looks new after a restart, and a rotation could +//! not be told from a baseline. Restarting every agent whenever this +//! controller restarts would be wrong, so the fingerprints live in a Secret +//! this controller owns. + +use k8s_openapi::ByteString; +use k8s_openapi::api::core::v1::Secret; +use kube::Api; +use kube::api::{ObjectMeta, Patch, PatchParams}; + +use crate::state::PersistedState; + +const STATE_KEY: &str = "state.json"; +const FIELD_MANAGER: &str = "openshell-provider-sync"; + +/// Why saved state could not be loaded. +#[derive(Debug, thiserror::Error)] +pub enum LoadError { + /// It could not be read, as when the API server is unavailable. Retry: + /// starting without it would overwrite it. + #[error("read state: {0}")] + Unavailable(String), + /// It was read but does not parse. Retrying cannot help. + #[error("state does not parse: {error}")] + Corrupt { raw: Vec, error: String }, +} + +/// Where persisted state is kept. +#[async_trait::async_trait] +pub trait StateStore: Send + Sync + 'static { + /// The saved state, or `None` when nothing has been saved yet. + async fn load(&self) -> Result, String>; + async fn save(&self, state: &PersistedState) -> Result<(), String>; +} + +/// [`StateStore`] in one Kubernetes Secret. +pub struct SecretStateStore { + api: Api, + name: String, +} + +impl SecretStateStore { + pub fn new(client: kube::Client, namespace: &str, name: impl Into) -> Self { + Self { + api: Api::namespaced(client, namespace), + name: name.into(), + } + } + + pub fn name(&self) -> &str { + &self.name + } + + /// Load, telling a read failure from state that does not parse. + pub async fn load_checked(&self) -> Result, LoadError> { + let Some(secret) = self + .api + .get_opt(&self.name) + .await + .map_err(|err| LoadError::Unavailable(format!("Secret {}: {err}", self.name)))? + else { + return Ok(None); + }; + let Some(raw) = secret.data.as_ref().and_then(|data| data.get(STATE_KEY)) else { + return Ok(None); + }; + serde_json::from_slice(&raw.0) + .map(Some) + .map_err(|err| LoadError::Corrupt { + raw: raw.0.clone(), + error: err.to_string(), + }) + } + + /// Keep unparseable state for diagnosis in a new Secret beside this one, + /// before it is overwritten. Returns the copy's name. + pub async fn quarantine(&self, raw: &[u8]) -> Result { + let stamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |elapsed| elapsed.as_secs()); + let name = format!("{}-corrupt-{stamp}", self.name); + let copy = Secret { + metadata: ObjectMeta { + name: Some(name.clone()), + ..Default::default() + }, + data: Some([(STATE_KEY.to_string(), ByteString(raw.to_vec()))].into()), + ..Default::default() + }; + match self + .api + .create(&kube::api::PostParams::default(), ©) + .await + { + Ok(_) => Ok(name), + Err(err) => Err(format!("create Secret {name}: {err}")), + } + } +} + +#[async_trait::async_trait] +impl StateStore for SecretStateStore { + async fn load(&self) -> Result, String> { + self.load_checked().await.map_err(|err| err.to_string()) + } + + async fn save(&self, state: &PersistedState) -> Result<(), String> { + let raw = serde_json::to_vec(state).map_err(|err| err.to_string())?; + let secret = Secret { + metadata: ObjectMeta { + name: Some(self.name.clone()), + ..Default::default() + }, + data: Some([(STATE_KEY.to_string(), ByteString(raw))].into()), + ..Default::default() + }; + self.api + .patch( + &self.name, + &PatchParams::apply(FIELD_MANAGER).force(), + &Patch::Apply(&secret), + ) + .await + .map(|_| ()) + .map_err(|err| format!("write Secret {}: {err}", self.name)) + } +} diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 2d8289347e..e61e6b0b3f 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -966,13 +966,30 @@ kubectl -n logs deployment/ --tail=200 | `ProviderSyncConflict` | Another live Secret already feeds the same workspace, provider and key | | `InvalidProviderSync` | Bad annotations, or only some of a provider's keys are set, or a value is not UTF-8 | | `ProviderSyncFailed` | The gateway rejected the update or was unreachable | +| `AgentRestarted` | A rotation restarted a sandbox attached to that provider, and it is ready again | +| `AgentRestartSkipped` | The sandbox was stopped, stopping, completed, in Error or deleted, or someone else stopped it after the restart started it; left alone | +| `AgentRestartPending` | A stop, start, phase read or state save failed and is being retried (backing off to every 5 minutes), or the sandbox is taking over 2 minutes to stop or start. Retried until it succeeds, and resumed after a provider-sync restart | +| `AgentRestartFailed` | The sandbox was deleted while stopped for the restart, or ended in Error; the gateway's own recovery owns it | No Events at all means the controller is not watching the Secret: check the label, its `OPENSHELL_PROVIDER_SYNC_NAMESPACE`, and its RBAC (`get`, `list` and `watch` on Secrets, `create` and `patch` on `events.k8s.io` Events). A -`ProviderSynced` event without the change reaching a running process usually -means the credential's profile lacks `stable_placeholder: true`; only processes -started after the update see the new value. Avoid rotating while gateways are +`ProviderSynced` event with no `AgentRestarted` after it means the values were a +baseline (the first values the controller saw, or the first sync after it +restarted without `--state-secret`), restarts are off (`--no-restart`), or no +sandbox was attached to the provider. Agents restarting right after a +provider-sync restart mean values changed while it was down, or restarts it had +in progress being resumed; a missing state Secret restarts nothing. If the state Secret +cannot be read, provider-sync waits and retries rather than starting over. If it +does not parse, provider-sync copies it to `-corrupt-`, +records a `StateSecretCorrupt` Warning on it, keeps every restart in progress it +can still parse, and starts over; an agent it had stopped whose restart entry +was lost stays stopped and must be started by hand, as the Event says. +Restarts in progress are in that Secret (`state.json`, `restarts`). One ends +when its sandbox is ready again or is deleted. Starting the sandbox by hand does +not cancel it: provider-sync stops and starts it once more, so the agent still +ends up running. +Avoid rotating while gateways are rolling: a supervisor whose provider refetch fails revokes its static credentials.