From 3d18f9b48ae7102246a678376b54b6c66db9a767 Mon Sep 17 00:00:00 2001 From: akshayakumar-t Date: Sun, 13 Sep 2026 11:53:39 +0530 Subject: [PATCH 1/2] fix(gcp auth): OBE-12220 lazy token fetch to prevent startup crash loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously GcpAuthenticator::from_file() and new_implicit() fetched an OAuth token eagerly at build time. If credentials were expired when vector (re)started, auth.build() would fail, the entire topology would fail to build, and the process manager would restart vector into a crash loop — taking all sources down, not just the GCP one. With this change, build() only parses/validates credentials (cheap, local). The first token fetch is deferred to the token_regenerator background task which fires immediately on startup and retries on failure. GCP components start in a token-less state and begin working as soon as the background fetch succeeds, without affecting the rest of the topology. Co-Authored-By: Akshaya's Agent --- lib/observo/private | 2 +- lib/vector-core/src/gcp.rs | 70 ++++++++++--------- src/sinks/gcp/cloud_storage.rs | 2 +- src/sinks/gcp/pubsub.rs | 2 +- src/sinks/gcp/stackdriver/logs/config.rs | 3 +- src/sinks/gcp/stackdriver/metrics/config.rs | 2 +- .../gcp_chronicle/chronicle_unstructured.rs | 2 +- src/sources/gcp_pubsub.rs | 4 +- 8 files changed, 46 insertions(+), 41 deletions(-) diff --git a/lib/observo/private b/lib/observo/private index 5a2922f00..247fb1eb3 160000 --- a/lib/observo/private +++ b/lib/observo/private @@ -1 +1 @@ -Subproject commit 5a2922f00853924b2915712a8ea191d95a03343b +Subproject commit 247fb1eb3cd521cf7bf66ebc472d81188c02a6dc diff --git a/lib/vector-core/src/gcp.rs b/lib/vector-core/src/gcp.rs index a3d50b9f1..094d661e9 100644 --- a/lib/vector-core/src/gcp.rs +++ b/lib/vector-core/src/gcp.rs @@ -103,7 +103,7 @@ pub struct GcpAuthConfig { } impl GcpAuthConfig { - pub async fn build(&self, scope: Scope, app_info: &AppInfo) -> crate::Result { + pub async fn build(&self, scope: Scope) -> crate::Result { Ok(if self.skip_authentication { GcpAuthenticator::None } else { @@ -112,7 +112,7 @@ impl GcpAuthConfig { match (&creds_path, &self.api_key) { (Some(path), _) => GcpAuthenticator::from_file(path, scope).await?, (None, Some(api_key)) => GcpAuthenticator::from_api_key(api_key.inner())?, - (None, None) => GcpAuthenticator::new_implicit(app_info).await?, + (None, None) => GcpAuthenticator::new_implicit().await?, } }) } @@ -128,21 +128,24 @@ pub enum GcpAuthenticator { #[derive(Debug)] pub struct InnerCreds { creds: Option<(Credentials, Scope)>, - token: RwLock, + token: RwLock>, } impl GcpAuthenticator { async fn from_file(path: &str, scope: Scope) -> crate::Result { let creds = Credentials::from_file(path).context(InvalidCredentialsSnafu)?; - let token = RwLock::new(fetch_token(&creds, &scope).await?); let creds = Some((creds, scope)); - Ok(Self::Credentials(Arc::new(InnerCreds { creds, token }))) + Ok(Self::Credentials(Arc::new(InnerCreds { + creds, + token: RwLock::new(None), + }))) } - async fn new_implicit(app_info: &AppInfo) -> crate::Result { - let token = RwLock::new(get_token_implicit(app_info).await?); - let creds = None; - Ok(Self::Credentials(Arc::new(InnerCreds { creds, token }))) + async fn new_implicit() -> crate::Result { + Ok(Self::Credentials(Arc::new(InnerCreds { + creds: None, + token: RwLock::new(None), + }))) } fn from_api_key(api_key: &str) -> crate::Result { @@ -154,7 +157,7 @@ impl GcpAuthenticator { pub fn make_token(&self) -> Option { match self { - Self::Credentials(inner) => Some(inner.make_token()), + Self::Credentials(inner) => inner.make_token(), Self::ApiKey(_) | Self::None => None, } } @@ -198,33 +201,35 @@ impl GcpAuthenticator { async fn token_regenerator(self, sender: watch::Sender<()>, app_info: &'static AppInfo) { match self { Self::Credentials(inner) => { - let expires_in = inner.token.read().unwrap().expires_in() as u64; - let mut deadline = - Duration::from_secs(expires_in - METADATA_TOKEN_EXPIRY_MARGIN_SECS); loop { - tokio::time::sleep(deadline).await; - debug!("Renewing GCP authentication token."); + debug!("Fetching GCP authentication token."); match inner.regenerate_token(app_info).await { Ok(()) => { sender.send_replace(()); - let expires_in = inner.token.read().unwrap().expires_in() as u64; + let expires_in = inner + .token + .read() + .unwrap() + .as_ref() + .map_or(METADATA_TOKEN_ERROR_RETRY_SECS, |t| t.expires_in() as u64); // Rather than an expected fresh token, the Metadata Server may return // the same (cached) token during the last 300 seconds of its lifetime. // This scenario is handled by retrying the token refresh after the // METADATA_TOKEN_ERROR_RETRY_SECS period when a fresh token is expected - let new_deadline = if expires_in <= METADATA_TOKEN_EXPIRY_MARGIN_SECS { + let deadline = if expires_in <= METADATA_TOKEN_EXPIRY_MARGIN_SECS { METADATA_TOKEN_ERROR_RETRY_SECS } else { expires_in - METADATA_TOKEN_EXPIRY_MARGIN_SECS }; - deadline = Duration::from_secs(new_deadline); + tokio::time::sleep(Duration::from_secs(deadline)).await; } Err(error) => { error!( - message = "Failed to update GCP authentication token.", + message = "Failed to fetch GCP authentication token.", %error ); - deadline = Duration::from_secs(METADATA_TOKEN_ERROR_RETRY_SECS); + tokio::time::sleep(Duration::from_secs(METADATA_TOKEN_ERROR_RETRY_SECS)) + .await; } } } @@ -245,13 +250,15 @@ impl InnerCreds { Some((creds, scope)) => fetch_token(creds, scope).await?, None => get_token_implicit(app_info).await?, }; - *self.token.write().unwrap() = token; + *self.token.write().unwrap() = Some(token); Ok(()) } - fn make_token(&self) -> String { + fn make_token(&self) -> Option { let token = self.token.read().unwrap(); - format!("{} {}", token.token_type(), token.access_token()) + token + .as_ref() + .map(|t| format!("{} {}", t.token_type(), t.access_token())) } } @@ -307,10 +314,13 @@ mod tests { use crate::assert_downcast_matches; #[tokio::test] - async fn fails_missing_creds() { - let error = build_auth("").await.expect_err("build failed to error"); - assert_downcast_matches!(error, GcpError, GcpError::GetImplicitToken { .. }); - // This should be a more relevant error + async fn defers_implicit_auth_when_no_creds() { + // With lazy token fetching, building with no credentials succeeds immediately. + // The first token fetch attempt happens in the background via token_regenerator. + let auth = build_auth("").await.expect("build should succeed with deferred auth"); + assert!(matches!(auth, GcpAuthenticator::Credentials(..))); + // No token yet — make_token returns None until the background fetch succeeds. + assert!(auth.make_token().is_none()); } #[tokio::test] @@ -369,10 +379,6 @@ mod tests { async fn build_auth(toml: &str) -> crate::Result { let config: GcpAuthConfig = toml::from_str(toml).expect("Invalid TOML"); - let app_info = vector_common::AppInfo { - name: "vector", - version: String::from("0.44.5"), - }; - config.build(Scope::Compute, &app_info).await + config.build(Scope::Compute).await } } diff --git a/src/sinks/gcp/cloud_storage.rs b/src/sinks/gcp/cloud_storage.rs index c42947b3d..0b39f8d0f 100644 --- a/src/sinks/gcp/cloud_storage.rs +++ b/src/sinks/gcp/cloud_storage.rs @@ -238,7 +238,7 @@ impl SinkConfig for GcsSinkConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { let auth = self .auth - .build(Scope::DevStorageReadWrite, &APP_INFO) + .build(Scope::DevStorageReadWrite) .await?; let base_url = format!("{}/{}/", self.endpoint, self.bucket); let tls = TlsSettings::from_options(self.tls.as_ref())?; diff --git a/src/sinks/gcp/pubsub.rs b/src/sinks/gcp/pubsub.rs index 5703e270e..320474120 100644 --- a/src/sinks/gcp/pubsub.rs +++ b/src/sinks/gcp/pubsub.rs @@ -164,7 +164,7 @@ struct PubsubSink { impl PubsubSink { async fn from_config(config: &PubsubConfig) -> crate::Result { // We only need to load the credentials if we are not targeting an emulator. - let auth = config.auth.build(Scope::PubSub, &APP_INFO).await?; + let auth = config.auth.build(Scope::PubSub).await?; let uri_base = format!( "{}/v1/projects/{}/topics/{}", diff --git a/src/sinks/gcp/stackdriver/logs/config.rs b/src/sinks/gcp/stackdriver/logs/config.rs index 399dea3a3..f244018b7 100644 --- a/src/sinks/gcp/stackdriver/logs/config.rs +++ b/src/sinks/gcp/stackdriver/logs/config.rs @@ -202,8 +202,7 @@ impl_generate_config_from_default!(StackdriverConfig); #[typetag::serde(name = "gcp_stackdriver_logs")] impl SinkConfig for StackdriverConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { - let app_info = crate::app_info(); - let auth = self.auth.build(Scope::LoggingWrite, &app_info).await?; + let auth = self.auth.build(Scope::LoggingWrite).await?; let request_builder = StackdriverLogsRequestBuilder { encoder: StackdriverLogsEncoder::new( diff --git a/src/sinks/gcp/stackdriver/metrics/config.rs b/src/sinks/gcp/stackdriver/metrics/config.rs index 26f576fa7..577256d18 100644 --- a/src/sinks/gcp/stackdriver/metrics/config.rs +++ b/src/sinks/gcp/stackdriver/metrics/config.rs @@ -95,7 +95,7 @@ impl_generate_config_from_default!(StackdriverConfig); #[typetag::serde(name = "gcp_stackdriver_metrics")] impl SinkConfig for StackdriverConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { - let auth = self.auth.build(Scope::MonitoringWrite, &APP_INFO).await?; + let auth = self.auth.build(Scope::MonitoringWrite).await?; let healthcheck = healthcheck().boxed(); let started = chrono::Utc::now(); diff --git a/src/sinks/gcp_chronicle/chronicle_unstructured.rs b/src/sinks/gcp_chronicle/chronicle_unstructured.rs index 71815b213..77e1db174 100644 --- a/src/sinks/gcp_chronicle/chronicle_unstructured.rs +++ b/src/sinks/gcp_chronicle/chronicle_unstructured.rs @@ -241,7 +241,7 @@ impl SinkConfig for ChronicleUnstructuredConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { let creds = self .auth - .build(Scope::MalachiteIngestion, &APP_INFO) + .build(Scope::MalachiteIngestion) .await?; let tls = TlsSettings::from_options(self.tls.as_ref())?; diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs index 2c0afa50a..3991f459b 100644 --- a/src/sources/gcp_pubsub.rs +++ b/src/sources/gcp_pubsub.rs @@ -95,7 +95,7 @@ mod proto { pub(crate) enum PubsubError { #[snafu(display("Invalid endpoint URI: {}", source))] Uri { source: InvalidUri }, - #[snafu(display("Could not create endpoint: {}", source))] +#[snafu(display("Could not create endpoint: {}", source))] Endpoint { source: tonic::transport::Error }, #[snafu(display("Could not set up endpoint TLS settings: {}", source))] EndpointTls { source: tonic::transport::Error }, @@ -266,7 +266,7 @@ impl SourceConfig for PubsubConfig { } }; - let auth = self.auth.build(Scope::PubSub, &APP_INFO).await?; + let auth = self.auth.build(Scope::PubSub).await?; let mut uri: Uri = self.endpoint.parse().context(UriSnafu)?; auth.apply_uri(&mut uri); From 3b9adccc8a295414bbb9e5d41e796dcb2c750af5 Mon Sep 17 00:00:00 2001 From: akshayakumar-t Date: Tue, 15 Sep 2026 20:38:28 +0530 Subject: [PATCH 2/2] docs: OBE-12220 add review findings for lazy GCP auth token fetch Captures Opus code review findings and AWS auth comparison. Key issues: sink healthcheck races token fetch (HIGH), silent 401 on first HTTP sink request (MEDIUM), dead map_or fallback (MEDIUM). Co-Authored-By: Akshaya's Agent --- OBE-12220_review_findings.md | 148 +++++++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 OBE-12220_review_findings.md diff --git a/OBE-12220_review_findings.md b/OBE-12220_review_findings.md new file mode 100644 index 000000000..ea11c519d --- /dev/null +++ b/OBE-12220_review_findings.md @@ -0,0 +1,148 @@ +# OBE-12220 Review Findings + +## Context + +This change makes GCP OAuth token fetching lazy — previously tokens were fetched eagerly at +`build()` time, causing crash loops when credentials were expired at restart. Tokens are now +fetched by a background `token_regenerator` task that fires immediately on startup and retries +on failure. + +Reviewed with Opus. AWS auth patterns studied for comparison. + +--- + +## Issues + +### HIGH — Health checks race the background token fetch (sinks) + +Every GCP sink constructs the healthcheck future and then spawns the token regenerator: + +```rust +// same pattern in gcp/pubsub.rs, stackdriver/logs, stackdriver/metrics, +// gcs cloud_storage, gcp_chronicle +let healthcheck = healthcheck(client, uri, sink.auth.clone()).boxed(); +sink.auth.spawn_regenerate_token(&APP_INFO); // token not yet fetched +``` + +The healthcheck calls `auth.apply()` → `make_token()` → `None` (no token yet) → no +`Authorization` header → GCP returns 401 → healthcheck fails. With healthchecks enabled +(the default), the crash loop becomes a healthcheck failure at startup instead of being +fully resolved. + +The `gcp_pubsub` **source** is handled correctly — the watch channel + stream restart +covers the race. **Sinks are the weak spot.** + +Fix options: +- **Option A**: Await the first watch signal (with timeout) inside the healthcheck before + sending the request, treating a missing token as "pending" rather than fatal. +- **Option B**: Gate sink readiness on the first `watch::Receiver::changed()` signal. +- **Option C** (lowest touch): Treat 401 as a retryable/non-fatal response in the + healthcheck instead of a hard failure. + +### MEDIUM — First real sink request sends no auth header + +For HTTP sinks, `apply()` silently omits the `Authorization` header when `make_token()` +returns `None`. Recovery depends on each sink's retry logic classifying 401 as retryable. +That is not uniform across the affected sinks — needs verification: + +- `GcsRetryLogic` (`src/sinks/gcs_common/`) +- Stackdriver logs/metrics retry logic +- Chronicle retry logic + +If any classify 401 as non-retryable, the first batch of events is dropped on startup. + +### MEDIUM — `map_or` guard is dead code masking an invariant + +```rust +// gcp.rs — after Ok(()) from regenerate_token: +let expires_in = inner.token.read().unwrap() + .as_ref() + .map_or(METADATA_TOKEN_ERROR_RETRY_SECS, |t| t.expires_in() as u64); +``` + +`regenerate_token` unconditionally writes `Some(token)` before returning `Ok`, so the +`None` fallback arm is unreachable under the current design. If a future change broke this +invariant, the fallback (retry every 2s) would silently hammer the token endpoint rather +than surfacing the bug. + +Prefer: +```rust +let expires_in = inner.token.read().unwrap() + .as_ref() + .expect("token present after successful regenerate") + .expires_in() as u64; +``` + +### LOW — No exponential backoff on repeated fetch failures + +With invalid static credentials (e.g. bad service account JSON), the loop retries every +`METADATA_TOKEN_ERROR_RETRY_SECS` (2s) forever against `oauth2.googleapis.com`. The old +code failed fast at build time. Consider exponential backoff with a cap for persistent +auth failures. + +### LOW — `from_file` and `new_implicit` are needlessly `async` + +Both constructors do zero async work after the token fetch was removed. Drop `async` for +accuracy. `GcpAuthConfig::build` will also no longer need to be `async` unless another +reason requires it. + +### Formatting regression (unrelated) + +`src/sources/gcp_pubsub.rs` — the `#[snafu(display(...))]` attribute on the `Endpoint` +variant lost its indentation: + +```rust +// broken +#[snafu(display("Could not create endpoint: {}", source))] +``` + +Fix with `make fmt` before merge. + +--- + +## Test Coverage Gaps + +1. **No test that the background task delivers a token and fires the watch signal.** + The core async behavior (token transitions `None → Some`, watch receiver observes + `changed()`) is completely untested. + +2. **No test for the sink healthcheck race.** The HIGH finding above has no regression + test. + +3. **Invalid credentials path is untested.** `fails_missing_creds` was the only test + covering this. Its deletion leaves the error path for bad/expired credentials with + zero test coverage. + +--- + +## AWS Comparison + +AWS uses `IdentityCache::lazy()` across all auth variants (`src/aws/auth.rs:205-225`). +Credentials are never fetched at build time — only on the first request. There is no +background refresh task; the AWS SDK refreshes inline via `provide_credentials()` on +every signing call, with expiry managed inside the identity cache. + +The key difference: AWS delegates "no token yet" to the SDK, which blocks the first +request for up to `load_timeout` (default 5s) waiting for credentials. The first request +does not go out unauthenticated. GCP's new approach sends immediately with no header, +which is what creates the sink healthcheck and first-request 401 problems. + +The GCP source path (gcp_pubsub) converges on the same safety property via a different +mechanism — the stream fails, `changed()` fires when the token arrives, and the stream +restarts authenticated. The sink path lacks an equivalent gate. + +--- + +## Summary + +| Finding | Severity | Affects | +|---------|----------|---------| +| Healthcheck races token fetch | HIGH | All GCP sinks | +| First request sends no auth header | MEDIUM | All GCP HTTP sinks | +| `map_or` dead fallback masks invariant | MEDIUM | `token_regenerator` | +| No backoff on auth failures | LOW | `token_regenerator` | +| Needless `async` on constructors | LOW | `from_file`, `new_implicit` | +| Formatting regression | LOW | `gcp_pubsub.rs` | +| Missing async-token test | Gap | `gcp.rs` tests | +| Missing healthcheck race test | Gap | sink integration tests | +| Invalid creds path untested | Gap | `gcp.rs` tests | \ No newline at end of file