vector/sinks/webhdfs/
config.rs

1use opendal::{Operator, layers::LoggingLayer, services::Webhdfs};
2use tower::ServiceBuilder;
3use vector_lib::{
4    codecs::{JsonSerializerConfig, NewlineDelimitedEncoderConfig, encoding::Framer},
5    config::{AcknowledgementsConfig, DataType, Input},
6    configurable::configurable_component,
7    sink::VectorSink,
8};
9
10use crate::{
11    codecs::{Encoder, EncodingConfigWithFraming, SinkType},
12    config::{GenerateConfig, SinkConfig, SinkContext},
13    sinks::{
14        Healthcheck,
15        opendal_common::*,
16        util::{
17            BatchConfig, BulkSizeBasedDefaultBatchSettings, Compression,
18            partitioner::KeyPartitioner,
19        },
20    },
21    template::{ConfinementConfig, Template},
22};
23
24/// Configuration for the `webhdfs` sink.
25#[configurable_component(sink("webhdfs", "WebHDFS."))]
26#[derive(Clone, Debug)]
27#[serde(deny_unknown_fields)]
28pub struct WebHdfsConfig {
29    /// The root path for WebHDFS.
30    ///
31    /// Must be a valid directory.
32    ///
33    /// The final file path is in the format of `{root}/{prefix}{suffix}`.
34    #[serde(default)]
35    pub root: String,
36
37    /// A prefix to apply to all keys.
38    ///
39    /// Prefixes are useful for partitioning objects, such as by creating a blob key that
40    /// stores blobs under a particular directory. If using a prefix for this purpose, it must end
41    /// in `/` to act as a directory path. A trailing `/` is **not** automatically added.
42    ///
43    /// The final file path is in the format of `{root}/{prefix}{suffix}`.
44    #[serde(default)]
45    #[configurable(metadata(docs::templateable))]
46    pub prefix: String,
47
48    /// An HDFS cluster consists of a single NameNode, a master server that manages the file system namespace and regulates access to files by clients.
49    ///
50    /// The endpoint is the HDFS's web restful HTTP API endpoint.
51    ///
52    /// For more information, see the [HDFS Architecture][hdfs_arch] documentation.
53    ///
54    /// [hdfs_arch]: https://hadoop.apache.org/docs/r3.3.4/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html#NameNode_and_DataNodes
55    #[serde(default)]
56    #[configurable(metadata(docs::examples = "http://127.0.0.1:9870"))]
57    pub endpoint: String,
58
59    #[serde(flatten)]
60    pub encoding: EncodingConfigWithFraming,
61
62    #[configurable(derived)]
63    #[serde(default = "Compression::gzip_default")]
64    pub compression: Compression,
65
66    #[configurable(derived)]
67    #[serde(default)]
68    pub batch: BatchConfig<BulkSizeBasedDefaultBatchSettings>,
69
70    #[configurable(derived)]
71    #[serde(
72        default,
73        deserialize_with = "crate::serde::bool_or_struct",
74        skip_serializing_if = "crate::serde::is_default"
75    )]
76    pub acknowledgements: AcknowledgementsConfig,
77
78    #[serde(flatten)]
79    pub confinement: ConfinementConfig,
80}
81
82impl GenerateConfig for WebHdfsConfig {
83    fn generate_config() -> toml::Value {
84        toml::Value::try_from(Self {
85            root: "/".to_string(),
86            prefix: "%F/".to_string(),
87            endpoint: "http://127.0.0.1:9870".to_string(),
88
89            encoding: (
90                Some(NewlineDelimitedEncoderConfig::new()),
91                JsonSerializerConfig::default(),
92            )
93                .into(),
94            compression: Compression::gzip_default(),
95            batch: BatchConfig::default(),
96
97            acknowledgements: Default::default(),
98            confinement: ConfinementConfig::default(),
99        })
100        .unwrap()
101    }
102}
103
104#[async_trait::async_trait]
105#[typetag::serde(name = "webhdfs")]
106impl SinkConfig for WebHdfsConfig {
107    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
108        let op = self.build_operator()?;
109
110        let check_op = op.clone();
111        let healthcheck = Box::pin(async move { Ok(check_op.check().await?) });
112
113        let sink = self.build_processor(op)?;
114        self.confinement.set_confinement_gauge("sink", Self::NAME);
115        Ok((sink, healthcheck))
116    }
117
118    fn input(&self) -> Input {
119        Input::new(self.encoding.config().1.input_type() & DataType::Log)
120    }
121
122    fn acknowledgements(&self) -> &AcknowledgementsConfig {
123        &self.acknowledgements
124    }
125}
126
127impl WebHdfsConfig {
128    pub fn build_operator(&self) -> crate::Result<Operator> {
129        // Build OpenDal Operator
130        let mut builder = Webhdfs::default();
131        // Prefix logic will be handled by key_partitioner.
132        builder = builder.root(&self.root);
133        builder = builder.endpoint(&self.endpoint);
134
135        let op = Operator::new(builder)?
136            .layer(LoggingLayer::default())
137            .finish();
138        Ok(op)
139    }
140
141    pub fn build_processor(&self, op: Operator) -> crate::Result<VectorSink> {
142        // Configure our partitioning/batching.
143        let batcher_settings = self.batch.into_batcher_settings()?;
144
145        let transformer = self.encoding.transformer();
146        let (framer, serializer) = self.encoding.build(SinkType::MessageBased)?;
147        let encoder = Encoder::<Framer>::new(framer, serializer);
148
149        let request_builder = OpenDalRequestBuilder {
150            encoder: (transformer, encoder),
151            compression: self.compression,
152        };
153
154        // TODO: we can add tower middleware here.
155        let svc = ServiceBuilder::new().service(OpenDalService::new(op));
156
157        let sink = OpenDalSink::new(
158            svc,
159            request_builder,
160            self.key_partitioner()?,
161            batcher_settings,
162        );
163
164        Ok(VectorSink::from_event_streamsink(sink))
165    }
166
167    pub fn key_partitioner(&self) -> crate::Result<KeyPartitioner> {
168        let prefix: Template = self.prefix.clone().try_into()?;
169        let prefix = prefix.confine(&self.confinement, Self::NAME, "prefix")?;
170        Ok(KeyPartitioner::new(prefix, None))
171    }
172}
173
174#[cfg(test)]
175mod tests {
176    use super::*;
177
178    #[test]
179    fn generate_config() {
180        crate::test_util::test_generate_config::<WebHdfsConfig>();
181    }
182
183    fn base_config() -> WebHdfsConfig {
184        WebHdfsConfig {
185            root: "/tmp/test/".into(),
186            prefix: String::new(),
187            endpoint: "http://127.0.0.1:9870".into(),
188            encoding: (
189                None::<vector_lib::codecs::encoding::FramingConfig>,
190                vector_lib::codecs::TextSerializerConfig::default(),
191            )
192                .into(),
193            compression: crate::sinks::util::Compression::None,
194            batch: Default::default(),
195            acknowledgements: Default::default(),
196            confinement: ConfinementConfig::default(),
197        }
198    }
199
200    #[test]
201    fn confinement_rejects_unconfined_prefix() {
202        let config = WebHdfsConfig {
203            prefix: "{{ tenant }}".into(),
204            ..base_config()
205        };
206        match config.key_partitioner() {
207            Err(err) => assert!(
208                err.to_string().contains("no literal string prefix"),
209                "unexpected error: {err}"
210            ),
211            Ok(_) => panic!("expected confinement error"),
212        }
213    }
214
215    #[test]
216    fn confinement_opt_out_allows_unconfined_prefix() {
217        let config = WebHdfsConfig {
218            prefix: "{{ tenant }}".into(),
219            confinement: ConfinementConfig {
220                dangerously_allow_unconfined_template_resolution: true,
221            },
222            ..base_config()
223        };
224        assert!(config.key_partitioner().is_ok());
225    }
226
227    #[test]
228    fn confinement_blocks_dotdot_escape_at_render() {
229        use crate::event::Event;
230        use vector_lib::event::LogEvent;
231        use vector_lib::partition::Partitioner;
232        use vrl::event_path;
233
234        let config = WebHdfsConfig {
235            prefix: "safe/{{ tenant }}/".into(),
236            ..base_config()
237        };
238        let partitioner = config.key_partitioner().unwrap();
239        let mut event = Event::Log(LogEvent::from("x"));
240        event
241            .as_mut_log()
242            .insert(event_path!("tenant"), "../../escape");
243        assert!(partitioner.partition(&event).is_none());
244    }
245}
vector/sinks/webhdfs/config.rs

vector/sinks/webhdfs/
config.rs
