vector/sinks/amqp/
config.rs

1//! Configuration functionality for the `AMQP` sink.
2use lapin::{BasicProperties, types::ShortString};
3use vector_lib::{
4    codecs::TextSerializerConfig,
5    internal_event::{error_stage, error_type},
6};
7
8use super::{channel::AmqpSinkChannels, sink::AmqpSink};
9use crate::{amqp::AmqpConfig, sinks::prelude::*, template::ConfinementConfig};
10
11/// AMQP properties configuration.
12#[configurable_component]
13#[configurable(title = "Configure the AMQP message properties.")]
14#[derive(Clone, Debug, Default)]
15pub struct AmqpPropertiesConfig {
16    /// Content-Type for the AMQP messages.
17    pub(crate) content_type: Option<String>,
18
19    /// Content-Encoding for the AMQP messages.
20    pub(crate) content_encoding: Option<String>,
21
22    /// Expiration for AMQP messages (in milliseconds).
23    pub(crate) expiration_ms: Option<u64>,
24
25    /// Priority for AMQP messages. It can be templated to an integer between 0 and 255 inclusive.
26    pub(crate) priority: Option<UnsignedIntTemplate>,
27}
28
29impl AmqpPropertiesConfig {
30    pub(super) fn build(&self, event: &Event) -> Option<BasicProperties> {
31        let mut prop = BasicProperties::default();
32        if let Some(content_type) = &self.content_type {
33            prop = prop.with_content_type(ShortString::from(content_type.clone()));
34        }
35        if let Some(content_encoding) = &self.content_encoding {
36            prop = prop.with_content_encoding(ShortString::from(content_encoding.clone()));
37        }
38        if let Some(expiration_ms) = &self.expiration_ms {
39            prop = prop.with_expiration(ShortString::from(expiration_ms.to_string()));
40        }
41        if let Some(priority_template) = &self.priority {
42            let priority = priority_template.render(event).unwrap_or_else(|error| {
43                warn!(
44                    message = "Failed to render numeric template for \"properties.priority\".",
45                    error = %error,
46                    error_type = error_type::TEMPLATE_FAILED,
47                    stage = error_stage::PROCESSING,
48                    internal_log_rate_limit = false,
49                );
50                Default::default()
51            });
52
53            // Clamp the value to the range of 0-255, as AMQP priority is a u8.
54            let priority = priority.clamp(0, u8::MAX.into()) as u8;
55            prop = prop.with_priority(priority);
56        }
57        Some(prop)
58    }
59}
60
61/// Configuration for the `amqp` sink.
62///
63/// Supports AMQP version 0.9.1
64#[configurable_component(sink(
65    "amqp",
66    "Send events to AMQP 0.9.1 compatible brokers like RabbitMQ."
67))]
68#[derive(Clone, Debug)]
69pub struct AmqpSinkConfig {
70    /// The exchange to publish messages to.
71    pub(crate) exchange: Template,
72
73    /// Template used to generate a routing key which corresponds to a queue binding.
74    pub(crate) routing_key: Option<Template>,
75
76    /// AMQP message properties.
77    pub(crate) properties: Option<AmqpPropertiesConfig>,
78
79    #[serde(flatten)]
80    pub(crate) connection: AmqpConfig,
81
82    #[configurable(derived)]
83    pub(crate) encoding: EncodingConfig,
84
85    #[configurable(derived)]
86    #[serde(
87        default,
88        deserialize_with = "crate::serde::bool_or_struct",
89        skip_serializing_if = "crate::serde::is_default"
90    )]
91    pub(crate) acknowledgements: AcknowledgementsConfig,
92
93    /// Maximum number of AMQP channels to keep active (channels are created as needed).
94    #[serde(default = "default_max_channels")]
95    pub(crate) max_channels: u32,
96
97    #[configurable(derived)]
98    #[serde(flatten)]
99    pub confinement: ConfinementConfig,
100}
101
102const fn default_max_channels() -> u32 {
103    4
104}
105
106impl Default for AmqpSinkConfig {
107    fn default() -> Self {
108        Self {
109            exchange: Template::try_from("vector").unwrap(),
110            routing_key: None,
111            properties: None,
112            encoding: TextSerializerConfig::default().into(),
113            connection: AmqpConfig::default(),
114            acknowledgements: AcknowledgementsConfig::default(),
115            max_channels: default_max_channels(),
116            confinement: ConfinementConfig::default(),
117        }
118    }
119}
120
121impl GenerateConfig for AmqpSinkConfig {
122    fn generate_config() -> toml::Value {
123        toml::from_str(
124            r#"connection_string = "amqp://localhost:5672/%2f"
125            routing_key = "user_id"
126            exchange = "test"
127            encoding.codec = "json"
128            max_channels = 4"#,
129        )
130        .unwrap()
131    }
132}
133
134#[async_trait::async_trait]
135#[typetag::serde(name = "amqp")]
136impl SinkConfig for AmqpSinkConfig {
137    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
138        let mut config = self.clone();
139        config.exchange = config
140            .exchange
141            .confine(&self.confinement, Self::NAME, "exchange")?;
142        config.routing_key = config
143            .routing_key
144            .map(|t| t.confine(&self.confinement, Self::NAME, "routing_key"))
145            .transpose()?;
146        let sink = AmqpSink::new(config).await?;
147        let hc = healthcheck(sink.channels.clone()).boxed();
148        self.confinement.set_confinement_gauge("sink", Self::NAME);
149        Ok((VectorSink::from_event_streamsink(sink), hc))
150    }
151
152    fn input(&self) -> Input {
153        Input::new(DataType::Log)
154    }
155
156    fn acknowledgements(&self) -> &AcknowledgementsConfig {
157        &self.acknowledgements
158    }
159}
160
161pub(super) async fn healthcheck(channels: AmqpSinkChannels) -> crate::Result<()> {
162    trace!("Healthcheck started.");
163
164    let channel = channels.get().await?;
165
166    if !channel.status().connected() {
167        return Err(Box::new(std::io::Error::new(
168            std::io::ErrorKind::BrokenPipe,
169            "Not Connected",
170        )));
171    }
172
173    trace!("Healthcheck completed.");
174    Ok(())
175}
176
177#[cfg(test)]
178mod tests {
179    use super::*;
180    use crate::config::format::{Format, deserialize};
181    use crate::template::{ConfinementConfig, Template};
182    use vrl::event_path;
183
184    #[test]
185    pub fn generate_config() {
186        crate::test_util::test_generate_config::<AmqpSinkConfig>();
187    }
188
189    #[test]
190    fn confinement_rejects_unconfined_exchange() {
191        let template = Template::try_from("{{ exchange }}").unwrap();
192        let config = ConfinementConfig::default();
193        let result = template.confine(&config, "amqp", "exchange");
194        assert!(result.is_err());
195    }
196
197    #[test]
198    fn confinement_opt_out_allows_unconfined_exchange() {
199        let template = Template::try_from("{{ exchange }}").unwrap();
200        let config = ConfinementConfig {
201            dangerously_allow_unconfined_template_resolution: true,
202        };
203        let result = template.confine(&config, "amqp", "exchange");
204        assert!(result.is_ok());
205    }
206
207    #[test]
208    fn confinement_allows_prefixed_exchange() {
209        let template = Template::try_from("events-{{ env }}").unwrap();
210        let config = ConfinementConfig::default();
211        let result = template.confine(&config, "amqp", "exchange");
212        assert!(result.is_ok());
213    }
214
215    fn assert_config_priority_eq(config: AmqpSinkConfig, event: &LogEvent, priority: u8) {
216        assert_eq!(
217            config
218                .properties
219                .unwrap()
220                .priority
221                .unwrap()
222                .render(event)
223                .unwrap(),
224            priority as u64
225        );
226    }
227
228    #[test]
229    pub fn parse_config_priority_static() {
230        for (format, config) in [
231            (
232                Format::Yaml,
233                r#"
234            exchange: "test"
235            routing_key: "user_id"
236            encoding:
237                codec: "json"
238            connection_string: "amqp://user:password@127.0.0.1:5672/"
239            properties:
240                priority: 1
241            "#,
242            ),
243            (
244                Format::Toml,
245                r#"
246            exchange = "test"
247            routing_key = "user_id"
248            encoding.codec = "json"
249            connection_string = "amqp://user:password@127.0.0.1:5672/"
250            properties = { priority = 1 }
251            "#,
252            ),
253            (
254                Format::Json,
255                r#"
256            {
257                "exchange": "test",
258                "routing_key": "user_id",
259                "encoding": {
260                    "codec": "json"
261                },
262                "connection_string": "amqp://user:password@127.0.0.1:5672/",
263                "properties": {
264                    "priority": 1
265                }
266            }
267            "#,
268            ),
269        ] {
270            let config: AmqpSinkConfig = deserialize(config, format).unwrap();
271            let event = LogEvent::from_str_legacy("message");
272            assert_config_priority_eq(config, &event, 1);
273        }
274    }
275
276    #[test]
277    pub fn parse_config_priority_templated() {
278        for (format, config) in [
279            (
280                Format::Yaml,
281                r#"
282            exchange: "test"
283            routing_key: "user_id"
284            encoding:
285                codec: "json"
286            connection_string: "amqp://user:password@127.0.0.1:5672/"
287            properties:
288                priority: "{{ .priority }}"
289            "#,
290            ),
291            (
292                Format::Toml,
293                r#"
294            exchange = "test"
295            routing_key = "user_id"
296            encoding.codec = "json"
297            connection_string = "amqp://user:password@127.0.0.1:5672/"
298            properties = { priority = "{{ .priority }}" }
299            "#,
300            ),
301            (
302                Format::Json,
303                r#"
304            {
305                "exchange": "test",
306                "routing_key": "user_id",
307                "encoding": {
308                    "codec": "json"
309                },
310                "connection_string": "amqp://user:password@127.0.0.1:5672/",
311                "properties": {
312                    "priority": "{{ .priority }}"
313                }
314            }
315            "#,
316            ),
317        ] {
318            let config: AmqpSinkConfig = deserialize(config, format).unwrap();
319            let event = {
320                let mut event = LogEvent::from_str_legacy("message");
321                event.insert(event_path!("priority"), 2);
322                event
323            };
324            assert_config_priority_eq(config, &event, 2);
325        }
326    }
327}
vector/sinks/amqp/config.rs

vector/sinks/amqp/
config.rs
