diff --git a/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md b/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md new file mode 100644 index 0000000000000..5b4fc0da9d2d9 --- /dev/null +++ b/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md @@ -0,0 +1,9 @@ +The `gcp_pubsub` source no longer floods the logs with misleading errors on idle, low-volume, or bursty subscriptions. GCP routinely ends a streaming pull with a transient error (such as an `Unavailable` status when there are simply no messages to deliver), which Vector retries automatically. These transient failures are now logged at `debug`, and a `failed_fetching_events` component error is only emitted once several failures occur in a row without any successful fetch in between. The threshold is configurable via the new `max_retry_errors` option (default `5`); a successful fetch resets the counter. + +The source is also more resilient to connections that silently break and stop delivering messages until Vector is restarted: + +- HTTP/2 keepalive pings are sent on the gRPC connection so a dead-but-open connection is detected and torn down instead of hanging. The ping interval and timeout are configurable via the new `keepalive` option. +- The stream keepalive uses a non-blocking send, so a backed-up request channel can no longer deadlock the task. +- A new `idle_timeout_secs` option (defaulting to `900`) restarts the stream if no activity is seen on an active connection within that window. + +authors: SamyDjemai diff --git a/lib/vector-config/src/external/serde_with.rs b/lib/vector-config/src/external/serde_with.rs index d671b8f48d530..da28a7c416f55 100644 --- a/lib/vector-config/src/external/serde_with.rs +++ b/lib/vector-config/src/external/serde_with.rs @@ -143,3 +143,12 @@ impl Configurable for Option> { + fn generate_schema(generator: &RefCell) -> Result + where + Self: Sized, + { + generate_optional_schema(&f64::as_configurable_ref(), generator) + } +} diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs index 1600c07dab339..8677b1c79a836 100644 --- a/src/sources/gcp_pubsub.rs +++ b/src/sources/gcp_pubsub.rs @@ -1,6 +1,7 @@ use std::{ error::Error as _, future::Future, + num::NonZeroU64, pin::Pin, sync::{ Arc, LazyLock, @@ -116,10 +117,61 @@ pub(crate) enum PubsubError { MAX_ACK_DEADLINE_SECS ))] InvalidAckDeadline, + #[snafu(display( + "`idle_timeout_secs` ({}) must be larger than `keepalive_secs` ({})", + idle_timeout_secs, + keepalive_secs + ))] + InvalidIdleTimeout { + idle_timeout_secs: f64, + keepalive_secs: f64, + }, } static CLIENT_ID: LazyLock = LazyLock::new(|| uuid::Uuid::new_v4().to_string()); +/// HTTP/2 keepalive configuration for the `gcp_pubsub` source's gRPC connection. +#[configurable_component] +#[derive(Clone, Copy, Debug)] +#[serde(deny_unknown_fields)] +pub struct PubsubKeepaliveConfig { + /// How often, in seconds, to send a keepalive PING on the connection. + /// + /// Shorter intervals detect dead connections faster at the cost of additional traffic. + /// Pub/Sub enforces gRPC's default minimum of one ping every 5 minutes, so lowering this + /// below `300` risks the server closing the stream with `too_many_pings`. + #[serde(default = "default_keepalive_interval_secs")] + #[configurable(metadata(docs::human_name = "Keepalive Interval"))] + pub interval_secs: NonZeroU64, + + /// How long, in seconds, to wait for a keepalive PING acknowledgement before treating + /// the connection as dead and closing it. + #[serde(default = "default_keepalive_timeout_secs")] + #[configurable(metadata(docs::human_name = "Keepalive Timeout"))] + pub timeout_secs: NonZeroU64, +} + +const fn default_keepalive_interval_secs() -> NonZeroU64 { + // Pub/Sub follows gRPC's default server enforcement, which closes streams pinged more often + // than once every 5 minutes with `too_many_pings`. Google's own Pub/Sub clients keepalive at + // this same interval. + NonZeroU64::new(300).expect("keepalive interval default must be nonzero") +} + +const fn default_keepalive_timeout_secs() -> NonZeroU64 { + // Matches hyper's default keepalive timeout. + NonZeroU64::new(20).expect("keepalive timeout default must be nonzero") +} + +impl Default for PubsubKeepaliveConfig { + fn default() -> Self { + Self { + interval_secs: default_keepalive_interval_secs(), + timeout_secs: default_keepalive_timeout_secs(), + } + } +} + /// Configuration for the `gcp_pubsub` source. #[serde_as] #[configurable_component(source( @@ -205,6 +257,43 @@ pub struct PubsubConfig { #[configurable(metadata(docs::human_name = "Keepalive"))] pub keepalive_secs: Duration, + /// The maximum amount of time, in seconds, to wait for a response on an + /// active stream before assuming the connection has stalled and + /// reconnecting. + /// + /// GCP Pub/Sub can leave a streaming pull in a state where the + /// connection looks healthy but no further responses are delivered and + /// the stream never errors. This bounds that inactivity. When set, it + /// must be larger than `keepalive_secs`; when unset, it defaults to `900` + /// or just above `keepalive_secs`, whichever is larger. + #[serde_as(as = "Option>")] + #[derivative(Default(value = "Option::None"))] + #[configurable(metadata(docs::human_name = "Idle Timeout"))] + pub idle_timeout_secs: Option, + + /// The number of consecutive stream failures, with no successful fetch in + /// between, before a fetch error is reported as a component error. + /// + /// GCP Pub/Sub routinely ends a streaming pull with a transient error + /// (for example, an `Unavailable` status when an idle subscription has no + /// messages). These are retried automatically and logged at `debug` + /// until this threshold is reached. A successful fetch resets the count. + /// Set to `1` to report every failure as an error. + #[serde(default = "default_max_retry_errors")] + #[derivative(Default(value = "default_max_retry_errors()"))] + #[configurable(metadata(docs::human_name = "Max Retry Errors Before Alerting"))] + pub max_retry_errors: usize, + + /// HTTP/2 keepalive settings for the source's gRPC connection. + /// + /// The source sends HTTP/2 PING frames on the connection so that one which has gone away + /// (for example, dropped by Pub/Sub or cut off by a network partition) is detected and + /// torn down instead of silently delivering no further messages. + #[configurable(derived)] + #[serde(default)] + #[derivative(Default(value = "PubsubKeepaliveConfig::default()"))] + pub keepalive: PubsubKeepaliveConfig, + /// The namespace to use for logs. This overrides the global setting. #[configurable(metadata(docs::hidden))] #[serde(default)] @@ -241,6 +330,18 @@ const fn default_keepalive() -> Duration { Duration::from_secs(60) } +const fn default_idle_timeout() -> Duration { + Duration::from_secs(900) +} + +/// Margin over `keepalive_secs` for the derived default idle timeout, keeping it +/// strictly larger when `keepalive_secs` is at or above `default_idle_timeout`. +const IDLE_TIMEOUT_KEEPALIVE_MARGIN: Duration = Duration::from_secs(60); + +const fn default_max_retry_errors() -> usize { + 5 +} + const fn default_max_concurrency() -> usize { 10 } @@ -271,6 +372,21 @@ impl SourceConfig for PubsubConfig { return Err(PubsubError::InvalidAckDeadline.into()); } + // Only an explicitly set idle timeout is validated; when unset, derive + // one that is always larger than the keepalive so upgrades with a long + // `keepalive_secs` keep starting. + let idle_timeout = match self.idle_timeout_secs { + Some(idle_timeout) if idle_timeout <= self.keepalive_secs => { + return Err(PubsubError::InvalidIdleTimeout { + idle_timeout_secs: idle_timeout.as_secs_f64(), + keepalive_secs: self.keepalive_secs.as_secs_f64(), + } + .into()); + } + Some(idle_timeout) => idle_timeout, + None => default_idle_timeout().max(self.keepalive_secs + IDLE_TIMEOUT_KEEPALIVE_MARGIN), + }; + let retry_delay_secs = match self.retry_delay_seconds { None => self.retry_delay_secs, Some(rds) => { @@ -301,6 +417,14 @@ impl SourceConfig for PubsubConfig { endpoint = endpoint.tls_config(tls_config).context(EndpointTlsSnafu)?; } + // Send HTTP/2 keepalive PINGs so a silently broken connection (for + // example, one dropped by Pub/Sub but left open) is detected and + // torn down instead of delivering no further messages. + endpoint = endpoint + .http2_keep_alive_interval(Duration::from_secs(self.keepalive.interval_secs.get())) + .keep_alive_timeout(Duration::from_secs(self.keepalive.timeout_secs.get())) + .keep_alive_while_idle(true); + let token_generator = auth.spawn_regenerate_token(); let protocol = uri @@ -328,6 +452,9 @@ impl SourceConfig for PubsubConfig { ack_deadline_secs, retry_delay: retry_delay_secs, keepalive: self.keepalive_secs, + idle_timeout, + max_retry_errors: self.max_retry_errors.max(1), + consecutive_failures: Default::default(), concurrency: Default::default(), full_response_size: self.full_response_size, log_namespace, @@ -393,6 +520,11 @@ struct PubsubSource { out: SourceSender, retry_delay: Duration, keepalive: Duration, + idle_timeout: Duration, + max_retry_errors: usize, + // Consecutive stream failures since the last successful fetch, shared across + // all concurrent streams so a success on any stream resets the count. + consecutive_failures: Arc, // The current concurrency is shared across all tasks. It is used // by the streams to avoid shutting down the last stream, which // would result in repeatedly re-opening the stream on idle. @@ -502,12 +634,23 @@ impl PubsubSource { let (ack_ids_sender, ack_ids_receiver) = mpsc::channel(ACK_QUEUE_SIZE); + // Reconnect if no response arrives within the idle timeout, to + // recover from a stream that stalls without ever erroring. Reset + // whenever a response arrives. Armed before the streaming-pull call so + // it also bounds a startup that Pub/Sub accepts but never resolves. + let idle_deadline = tokio::time::sleep(self.idle_timeout); + tokio::pin!(idle_deadline); + // Handle shutdown during startup, the streaming pull doesn't // start if there is no data in the subscription. let request_stream = self.request_stream(ack_ids_receiver); debug!("Starting streaming pull."); let stream = tokio::select! { _ = &mut self.shutdown => return State::Shutdown, + _ = &mut idle_deadline => { + debug!("Streaming pull did not start within the idle timeout, restarting stream."); + return State::RetryDelay; + } result = client.streaming_pull(request_stream) => match result { Ok(stream) => stream, Err(error) => { @@ -528,23 +671,39 @@ impl PubsubSource { receipts = ack_stream.next() => if let Some((status, receipts)) = receipts { pending_acks -= 1; if status == BatchStatus::Delivered { - ack_ids_sender - .send(receipts) - .await - .unwrap_or_else(|_| unreachable!("request stream never closes")); + // Non-blocking: awaiting a full channel would stall the + // select loop, preventing the idle timeout below from + // recovering a stalled request stream. Reconnect instead; + // dropped ack IDs are redelivered (at-least-once). + match ack_ids_sender.try_send(receipts) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + debug!("Request stream stalled sending acks, restarting stream."); + break State::RetryDelay; + } + Err(mpsc::error::TrySendError::Closed(_)) => { + unreachable!("request stream never closes") + } + } } }, response = stream.next() => match response { Some(Ok(response)) => { - self.handle_response( + self.consecutive_failures.store(0, Ordering::Relaxed); + idle_deadline + .as_mut() + .reset(tokio::time::Instant::now() + self.idle_timeout); + if let Some(state) = self.handle_response( response, &finalizer, &ack_ids_sender, &mut pending_acks, busy_flag, - ).await; + ).await { + break state; + } } - Some(Err(error)) => break translate_error(error), + Some(Err(error)) => break self.handle_stream_error(error), None => break State::RetryNow, }, _ = &mut self.shutdown, if pending_acks == 0 => return State::Shutdown, @@ -552,6 +711,16 @@ impl PubsubSource { debug!("New authentication token generated, restarting stream."); break State::RetryNow; }, + // The recovery path for a stalled stream: always fires, never + // gated on the request channel that may have stopped draining. + // Unsent ack IDs are dropped and redelivered (at-least-once). + _ = &mut idle_deadline => { + debug!( + message = "No activity on the stream within the idle timeout, restarting stream.", + timeout_secs = self.idle_timeout.as_secs_f64(), + ); + break State::RetryDelay; + }, _ = tokio::time::sleep(self.keepalive) => { if pending_acks == 0 { // No pending acks, and no new data, so drop @@ -564,17 +733,19 @@ impl PubsubSource { // Otherwise, mark this stream as idle. busy_flag.store(false, Ordering::Relaxed); } - // GCP Pub/Sub likes to time out connections after - // about 75 seconds of inactivity. To forestall - // the resulting error, send an empty array of - // acknowledgement IDs to the request stream if no - // other activity has happened. This will result - // in a new request with empty fields, effectively - // a keepalive. - ack_ids_sender - .send(Vec::new()) - .await - .unwrap_or_else(|_| unreachable!("request stream never closes")); + // GCP Pub/Sub times out connections after ~75s of + // inactivity. Send an empty ack ID array as a keepalive to + // forestall that. Non-blocking: a full channel already has + // acks queued, so the keepalive is redundant and skipped. + match ack_ids_sender.try_send(Vec::new()) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + debug!("Keepalive skipped: request stream has acks queued."); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + unreachable!("request stream never closes") + } + } } } } @@ -612,6 +783,7 @@ impl PubsubSource { })) } + // Returns `Some(state)` when the stream should be torn down and restarted. async fn handle_response( &mut self, response: proto::StreamingPullResponse, @@ -619,7 +791,7 @@ impl PubsubSource { ack_ids: &mpsc::Sender>, pending_acks: &mut usize, busy_flag: &Arc, - ) { + ) -> Option { if response.received_messages.len() >= self.full_response_size { busy_flag.store(true, Ordering::Relaxed); } @@ -632,10 +804,19 @@ impl PubsubSource { match self.out.send_batch(events).await { Err(_) => emit!(StreamClosedError { count }), Ok(()) => match notifier { - None => ack_ids - .send(ids) - .await - .unwrap_or_else(|_| unreachable!("request stream never closes")), + // Auto-ack: send without blocking so a stalled request stream + // can't park the task here and starve the idle timeout. On a + // full channel, reconnect; the unsent IDs are redelivered. + None => match ack_ids.try_send(ids) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + debug!("Request stream stalled sending acks, restarting stream."); + return Some(State::RetryDelay); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + unreachable!("request stream never closes") + } + }, Some(notifier) => { finalizer .as_ref() @@ -645,6 +826,37 @@ impl PubsubSource { } }, } + None + } + + // GCP frequently ends a streaming pull with a transient, auto-retried + // error (a reset, or `Unavailable` when an idle subscription has no + // messages). Only escalate to a component error once failures persist, + // to avoid flooding the logs on quiet subscriptions. + fn handle_stream_error(&self, error: tonic::Status) -> State { + // A server-side reset is not really an error, so retry immediately + // without counting it. + if is_reset(&error) { + debug!("Stream reset by server."); + return State::RetryNow; + } + + // Escalate once the streak reaches the threshold, and keep reporting + // while it persists so a genuinely broken stream stays visible (and + // `max_retry_errors: 1` reports every failure). A successful fetch + // resets the counter. + let consecutive_failures = self.consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1; + if consecutive_failures >= self.max_retry_errors { + emit!(GcpPubsubReceiveError { error }); + } else { + debug!( + message = "Transient error fetching events, retrying.", + %error, + consecutive_failures, + max_retry_errors = self.max_retry_errors, + ); + } + State::RetryDelay } async fn parse_messages( @@ -712,22 +924,6 @@ impl PubsubSource { } } -fn translate_error(error: tonic::Status) -> State { - // GCP occasionally issues a connection reset - // in the middle of the streaming pull. This - // reset is not technically an error, so we - // want to retry immediately, but it is - // reported to us as an error from the - // underlying library (`tonic`). - if is_reset(&error) { - debug!("Stream reset by server."); - State::RetryNow - } else { - emit!(GcpPubsubReceiveError { error }); - State::RetryDelay - } -} - fn is_reset(error: &Status) -> bool { error .source() @@ -762,6 +958,73 @@ mod tests { crate::test_util::test_generate_config::(); } + #[test] + fn default_idle_timeout_is_larger_than_keepalive() { + assert!( + default_idle_timeout() > default_keepalive(), + "the default idle timeout must be larger than the default keepalive" + ); + } + + #[test] + fn programmatic_default_preserves_max_retry_errors() { + // `Default` must match the serde default so generated/programmatic + // configs keep the retry threshold instead of collapsing to 0. + assert_eq!( + PubsubConfig::default().max_retry_errors, + default_max_retry_errors() + ); + } + + #[tokio::test] + async fn accepts_long_keepalive_without_explicit_idle_timeout() { + // A large `keepalive_secs` with no `idle_timeout_secs` must still build: + // the derived default stays larger than the keepalive. + let config: PubsubConfig = serde_yaml::from_str( + r#" + project: my-project + subscription: my-subscription + skip_authentication: true + keepalive_secs: 900 + "#, + ) + .unwrap(); + + let (tx, _rx) = SourceSender::new_test(); + let context = SourceContext::new_test(tx, None); + assert!( + config.build(context).await.is_ok(), + "build should accept a long keepalive when idle_timeout_secs is unset" + ); + } + + #[tokio::test] + async fn rejects_idle_timeout_not_larger_than_keepalive() { + let config: PubsubConfig = serde_yaml::from_str( + r#" + project: my-project + subscription: my-subscription + skip_authentication: true + keepalive_secs: 60 + idle_timeout_secs: 60 + "#, + ) + .unwrap(); + + let (tx, _rx) = SourceSender::new_test(); + let context = SourceContext::new_test(tx, None); + let error = match config.build(context).await { + Ok(_) => { + panic!("build should reject an idle timeout that is not larger than keepalive") + } + Err(error) => error.to_string(), + }; + assert!( + error.contains("idle_timeout_secs"), + "unexpected error: {error}" + ); + } + #[test] fn output_schema_definition_vector_namespace() { let config = PubsubConfig { diff --git a/website/cue/reference/components/sources/generated/gcp_pubsub.cue b/website/cue/reference/components/sources/generated/gcp_pubsub.cue index af645bf7384b3..f666ac907258e 100644 --- a/website/cue/reference/components/sources/generated/gcp_pubsub.cue +++ b/website/cue/reference/components/sources/generated/gcp_pubsub.cue @@ -597,6 +597,52 @@ generated: components: sources: gcp_pubsub: configuration: { required: false type: uint: default: 100 } + idle_timeout_secs: { + description: """ + The maximum amount of time, in seconds, to wait for a response on an + active stream before assuming the connection has stalled and + reconnecting. + + GCP Pub/Sub can leave a streaming pull in a state where the + connection looks healthy but no further responses are delivered and + the stream never errors. This bounds that inactivity. When set, it + must be larger than `keepalive_secs`; when unset, it defaults to `900` + or just above `keepalive_secs`, whichever is larger. + """ + required: false + type: float: {} + } + keepalive: { + description: """ + HTTP/2 keepalive settings for the source's gRPC connection. + + The source sends HTTP/2 PING frames on the connection so that one which has gone away + (for example, dropped by Pub/Sub or cut off by a network partition) is detected and + torn down instead of silently delivering no further messages. + """ + required: false + type: object: options: { + interval_secs: { + description: """ + How often, in seconds, to send a keepalive PING on the connection. + + Shorter intervals detect dead connections faster at the cost of additional traffic. + Pub/Sub enforces gRPC's default minimum of one ping every 5 minutes, so lowering this + below `300` risks the server closing the stream with `too_many_pings`. + """ + required: false + type: uint: default: 300 + } + timeout_secs: { + description: """ + How long, in seconds, to wait for a keepalive PING acknowledgement before treating + the connection as dead and closing it. + """ + required: false + type: uint: default: 20 + } + } + } keepalive_secs: { description: """ The amount of time, in seconds, with no received activity @@ -614,6 +660,20 @@ generated: components: sources: gcp_pubsub: configuration: { required: false type: uint: default: 10 } + max_retry_errors: { + description: """ + The number of consecutive stream failures, with no successful fetch in + between, before a fetch error is reported as a component error. + + GCP Pub/Sub routinely ends a streaming pull with a transient error + (for example, an `Unavailable` status when an idle subscription has no + messages). These are retried automatically and logged at `debug` + until this threshold is reached. A successful fetch resets the count. + Set to `1` to report every failure as an error. + """ + required: false + type: uint: default: 5 + } poll_time_seconds: { description: """ How often to poll the currently active streams to see if they