fix(gcp_pubsub source): recover from stalled streams and reduce error noise

changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md

@@ -0,0 +1,7 @@
+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.

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,60 @@ 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<String> = 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.
+    /// gRPC guidance recommends no less than 60 seconds to avoid tripping `too_many_pings`
+    /// policies on servers or proxies between the source and Pub/Sub.
+    #[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 {
+    // Aligned with gRPC keepalive guidance, which recommends no less than one minute to avoid
+    // tripping `too_many_pings` policies on proxies between the source and Pub/Sub.
+    NonZeroU64::new(60).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 +256,41 @@ 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. It must be
+    /// larger than `keepalive_secs`.
+    #[serde(default = "default_idle_timeout")]
+    #[serde_as(as = "serde_with::DurationSecondsWithFrac<f64>")]
+    #[configurable(metadata(docs::human_name = "Idle Timeout"))]
+    pub idle_timeout_secs: Duration,
+
+    /// 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")]
+    #[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 +327,14 @@ const fn default_keepalive() -> Duration {
     Duration::from_secs(60)
 }
 
+const fn default_idle_timeout() -> Duration {
+    Duration::from_secs(900)
+}
+
+const fn default_max_retry_errors() -> usize {
+    5
+}
+
 const fn default_max_concurrency() -> usize {
     10
 }
@@ -271,6 +365,14 @@ impl SourceConfig for PubsubConfig {
             return Err(PubsubError::InvalidAckDeadline.into());
         }
 
+        if self.idle_timeout_secs <= self.keepalive_secs {
+            return Err(PubsubError::InvalidIdleTimeout {
+                idle_timeout_secs: self.idle_timeout_secs.as_secs_f64(),
+                keepalive_secs: self.keepalive_secs.as_secs_f64(),
+            }
+            .into());
+        }
+
         let retry_delay_secs = match self.retry_delay_seconds {
             None => self.retry_delay_secs,
             Some(rds) => {
@@ -301,6 +403,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 +438,9 @@ impl SourceConfig for PubsubConfig {
             ack_deadline_secs,
             retry_delay: retry_delay_secs,
             keepalive: self.keepalive_secs,
+            idle_timeout: self.idle_timeout_secs,
+            max_retry_errors: self.max_retry_errors.max(1),
+            consecutive_failures: 0,
             concurrency: Default::default(),
             full_response_size: self.full_response_size,
             log_namespace,
@@ -393,6 +506,10 @@ struct PubsubSource {
     out: SourceSender,
     retry_delay: Duration,
     keepalive: Duration,
+    idle_timeout: Duration,
+    max_retry_errors: usize,
+    // Consecutive stream failures since the last successful fetch.
+    consecutive_failures: usize,
     // 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.
@@ -522,6 +639,12 @@ impl PubsubSource {
             Finalizer::maybe_new(self.acknowledgements, Some(self.shutdown.clone()));
         let mut pending_acks = 0;
 
+        // Reconnect if no response arrives within the idle timeout, to
+        // recover from a stream that stalls without ever erroring. Reset
+        // whenever a response arrives.
+        let idle_deadline = tokio::time::sleep(self.idle_timeout);
+        tokio::pin!(idle_deadline);
+
         loop {
             tokio::select! {
                 biased;
@@ -536,6 +659,10 @@ impl PubsubSource {
                 },
                 response = stream.next() => match response {
                     Some(Ok(response)) => {
+                        self.consecutive_failures = 0;
+                        idle_deadline
+                            .as_mut()
+                            .reset(tokio::time::Instant::now() + self.idle_timeout);
                         self.handle_response(
                             response,
                             &finalizer,
@@ -544,14 +671,21 @@ impl PubsubSource {
                             busy_flag,
                         ).await;
                     }
-                    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,
                 _ = self.token_generator.changed() => {
                     debug!("New authentication token generated, restarting stream.");
                     break State::RetryNow;
                 },
+                _ = &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
@@ -571,10 +705,20 @@ impl PubsubSource {
                     // 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"));
+                    //
+                    // Send without blocking: a full channel means the
+                    // request stream stalled, so reconnect rather than
+                    // deadlock.
+                    match ack_ids_sender.try_send(Vec::new()) {
+                        Ok(()) => {}
+                        Err(mpsc::error::TrySendError::Full(_)) => {
+                            debug!("Keepalive request stream stalled, restarting stream.");
+                            break State::RetryDelay;
+                        }
+                        Err(mpsc::error::TrySendError::Closed(_)) => {
+                            unreachable!("request stream never closes")
+                        }
+                    }
                 }
             }
         }
@@ -647,6 +791,32 @@ impl PubsubSource {
         }
     }
 
+    // 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(&mut 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;
+        }
+
+        self.consecutive_failures += 1;
+        if self.consecutive_failures >= self.max_retry_errors {
+            emit!(GcpPubsubReceiveError { error });
+        } else {
+            debug!(
+                message = "Transient error fetching events, retrying.",
+                %error,
+                consecutive_failures = self.consecutive_failures,
+                max_retry_errors = self.max_retry_errors,
+            );
+        }
+        State::RetryDelay
+    }
+
     async fn parse_messages(
         &self,
         response: Vec<proto::ReceivedMessage>,
@@ -712,22 +882,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 +916,49 @@ mod tests {
         crate::test_util::test_generate_config::<PubsubConfig>();
     }
 
+    #[test]
+    fn default_idle_timeout_is_larger_than_keepalive() {
+        // Defaults come from serde, not `Default::default()`.
+        let config: PubsubConfig = serde_yaml::from_str(
+            r#"
+            project: my-project
+            subscription: my-subscription
+            "#,
+        )
+        .unwrap();
+        assert!(
+            config.idle_timeout_secs > config.keepalive_secs,
+            "the default idle timeout must be larger than the default keepalive"
+        );
+    }
+
+    #[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 {