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 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);