make amqp max_channels configurable; default to 4; added to docs and changelog

Author: aramperes

changelog.d/22971_reconnect_broken_amqp_sink.fix.md

@@ -1,3 +1,3 @@
-The `amqp` sink now attempts to re-connect to the AMQP broker when the channel has been disconnected.
+The `amqp` sink now attempts to re-connect to the AMQP broker when the channel has been disconnected. It will also create up to 4 channels in a pool (configurable with the `max_channels` configuration) to improve throughput.
 
 authors: aramperes

src/sinks/amqp/channel.rs

@@ -7,11 +7,22 @@ use lapin::options::ConfirmSelectOptions;
 pub type AmqpSinkChannels = Pool<AmqpSinkChannelManager>;
 
 pub(super) fn new_channel_pool(config: &AmqpSinkConfig) -> crate::Result<AmqpSinkChannels> {
+    let max_channels = config.max_channels.try_into().map_err(|_| {
+        Box::new(AmqpError::PoolError {
+            error: "max_channels must fit into usize".into(),
+        })
+    })?;
+    if max_channels == 0 {
+        return Err(Box::new(AmqpError::PoolError {
+            error: "max_channels must be positive".into(),
+        }));
+    }
     let channel_manager = AmqpSinkChannelManager::new(&config.connection);
     let channels = Pool::builder(channel_manager)
-        .max_size(4)
+        .max_size(max_channels)
         .runtime(deadpool::Runtime::Tokio1)
         .build()?;
+    debug!("AMQP channel pool created with max size: {}", max_channels);
     Ok(channels)
 }
 

src/sinks/amqp/config.rs

@@ -90,6 +90,14 @@ pub struct AmqpSinkConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     pub(crate) acknowledgements: AcknowledgementsConfig,
+
+    /// Maximum number of AMQP channels to keep active (channels are created as needed).
+    #[serde(default = "default_max_channels")]
+    pub(crate) max_channels: u32,
+}
+
+const fn default_max_channels() -> u32 {
+    4
 }
 
 impl Default for AmqpSinkConfig {
@@ -101,6 +109,7 @@ impl Default for AmqpSinkConfig {
             encoding: TextSerializerConfig::default().into(),
             connection: AmqpConfig::default(),
             acknowledgements: AcknowledgementsConfig::default(),
+            max_channels: default_max_channels(),
         }
     }
 }
@@ -111,7 +120,8 @@ impl GenerateConfig for AmqpSinkConfig {
             r#"connection_string = "amqp://localhost:5672/%2f"
             routing_key = "user_id"
             exchange = "test"
-            encoding.codec = "json""#,
+            encoding.codec = "json"
+            max_channels = 4"#,
         )
         .unwrap()
     }

website/cue/reference/components/sinks/base/amqp.cue

@@ -392,6 +392,11 @@ base: components: sinks: amqp: configuration: {
 		required:    true
 		type: string: syntax: "template"
 	}
+	max_channels: {
+		description: "Maximum number of AMQP channels to keep active (channels are created as needed)."
+		required: 	 false
+		type: uint: default: 4
+	}
 	properties: {
 		description: """
 			Configure the AMQP message properties.