vector/sinks/loki/
config.rs

1use std::collections::HashMap;
2
3use vrl::value::Kind;
4
5use super::{healthcheck::healthcheck, sink::LokiSink};
6use crate::{
7    http::{Auth, HttpClient, MaybeAuth},
8    schema,
9    sinks::{prelude::*, util::UriSerde},
10    template::ConfinementConfig,
11};
12
13const fn default_compression() -> Compression {
14    Compression::Snappy
15}
16
17fn default_loki_path() -> String {
18    "/loki/api/v1/push".to_string()
19}
20
21/// Configuration for the `loki` sink.
22#[configurable_component(sink("loki", "Deliver log event data to the Loki aggregation system."))]
23#[derive(Clone, Debug)]
24#[serde(deny_unknown_fields)]
25pub struct LokiConfig {
26    /// The base URL of the Loki instance.
27    ///
28    /// The `path` value is appended to this.
29    #[configurable(metadata(docs::examples = "http://localhost:3100"))]
30    pub endpoint: UriSerde,
31
32    /// The path to use in the URL of the Loki instance.
33    #[serde(default = "default_loki_path")]
34    pub path: String,
35
36    #[configurable(derived)]
37    pub encoding: EncodingConfig,
38
39    /// The [tenant ID][tenant_id] to specify in requests to Loki.
40    ///
41    /// When running Loki locally, a tenant ID is not required.
42    ///
43    /// [tenant_id]: https://grafana.com/docs/loki/latest/operations/multi-tenancy/
44    #[configurable(metadata(
45        docs::examples = "some_tenant_id",
46        docs::examples = "{{ event_field }}",
47    ))]
48    pub tenant_id: Option<Template>,
49
50    /// A set of labels that are attached to each batch of events.
51    ///
52    /// Both keys and values are templateable, which enables you to attach dynamic labels to events.
53    ///
54    /// Valid label keys include `*`, and prefixes ending with `*`, to allow for the expansion of
55    /// objects into multiple labels. See [Label expansion][label_expansion] for more information.
56    ///
57    /// Note: If the set of labels has high cardinality, this can cause drastic performance issues
58    /// with Loki. To prevent this from happening, reduce the number of unique label keys and
59    /// values.
60    ///
61    /// [label_expansion]: https://vector.dev/docs/reference/configuration/sinks/loki/#label-expansion
62    #[configurable(metadata(docs::examples = "loki_labels_examples()"))]
63    #[configurable(metadata(docs::additional_props_description = "A Loki label."))]
64    pub labels: HashMap<Template, Template>,
65
66    /// Whether or not to delete fields from the event when they are used as labels.
67    #[serde(default = "crate::serde::default_false")]
68    pub remove_label_fields: bool,
69
70    /// Structured metadata that is attached to each batch of events.
71    ///
72    /// Both keys and values are templateable, which enables you to attach dynamic structured metadata to events.
73    ///
74    /// Valid metadata keys include `*`, and prefixes ending with `*`, to allow for the expansion of
75    /// objects into multiple metadata entries. This follows the same logic as [Label expansion][label_expansion].
76    ///
77    /// [label_expansion]: https://vector.dev/docs/reference/configuration/sinks/loki/#label-expansion
78    #[configurable(metadata(docs::examples = "loki_structured_metadata_examples()"))]
79    #[configurable(metadata(docs::additional_props_description = "Loki structured metadata."))]
80    #[serde(default)]
81    pub structured_metadata: HashMap<Template, Template>,
82
83    /// Whether or not to delete fields from the event when they are used in structured metadata.
84    #[serde(default = "crate::serde::default_false")]
85    pub remove_structured_metadata_fields: bool,
86
87    /// Whether or not to remove the timestamp from the event payload.
88    ///
89    /// The timestamp is still sent as event metadata for Loki to use for indexing.
90    #[serde(default = "crate::serde::default_true")]
91    pub remove_timestamp: bool,
92
93    /// Compression configuration.
94    /// Snappy compression implies sending push requests as Protocol Buffers.
95    #[serde(default = "default_compression")]
96    pub compression: Compression,
97
98    #[configurable(derived)]
99    #[serde(default)]
100    pub out_of_order_action: OutOfOrderAction,
101
102    #[configurable(derived)]
103    pub auth: Option<Auth>,
104
105    #[configurable(derived)]
106    #[serde(default)]
107    pub request: TowerRequestConfig,
108
109    #[configurable(derived)]
110    #[serde(default)]
111    pub batch: BatchConfig<LokiDefaultBatchSettings>,
112
113    #[configurable(derived)]
114    pub tls: Option<TlsConfig>,
115
116    #[configurable(derived)]
117    #[serde(
118        default,
119        deserialize_with = "crate::serde::bool_or_struct",
120        skip_serializing_if = "crate::serde::is_default"
121    )]
122    acknowledgements: AcknowledgementsConfig,
123
124    #[configurable(derived)]
125    #[serde(flatten)]
126    pub confinement: ConfinementConfig,
127}
128
129fn loki_labels_examples() -> HashMap<String, String> {
130    let mut examples = HashMap::new();
131    examples.insert("source".to_string(), "vector".to_string());
132    examples.insert(
133        "\"pod_labels_*\"".to_string(),
134        "{{ kubernetes.pod_labels }}".to_string(),
135    );
136    examples.insert("\"*\"".to_string(), "{{ metadata }}".to_string());
137    examples.insert(
138        "{{ event_field }}".to_string(),
139        "{{ some_other_event_field }}".to_string(),
140    );
141    examples
142}
143
144fn loki_structured_metadata_examples() -> HashMap<String, String> {
145    let mut examples = HashMap::new();
146    examples.insert("source".to_string(), "vector".to_string());
147    examples.insert(
148        "\"pod_labels_*\"".to_string(),
149        "{{ kubernetes.pod_labels }}".to_string(),
150    );
151    examples.insert("\"*\"".to_string(), "{{ metadata }}".to_string());
152    examples.insert(
153        "{{ event_field }}".to_string(),
154        "{{ some_other_event_field }}".to_string(),
155    );
156    examples
157}
158
159#[derive(Clone, Copy, Debug, Default)]
160pub struct LokiDefaultBatchSettings;
161
162impl SinkBatchSettings for LokiDefaultBatchSettings {
163    const MAX_EVENTS: Option<usize> = Some(100_000);
164    const MAX_BYTES: Option<usize> = Some(1_000_000);
165    const TIMEOUT_SECS: f64 = 1.0;
166}
167
168/// Out-of-order event behavior.
169///
170/// Some sources may generate events with timestamps that aren't in chronological order. Even though the
171/// sink sorts the events before sending them to Loki, there is a chance that another event could come in
172/// that is out of order with the latest events sent to Loki. Prior to Loki 2.4.0, this
173/// was not supported and would result in an error during the push request.
174///
175/// If you're using Loki 2.4.0 or newer, `Accept` is the preferred action, which lets Loki handle
176/// any necessary sorting/reordering. If you're using an earlier version, then you must use `Drop`
177/// or `RewriteTimestamp` depending on which option makes the most sense for your use case.
178#[configurable_component]
179#[derive(Copy, Clone, Debug, Default)]
180#[serde(rename_all = "snake_case")]
181pub enum OutOfOrderAction {
182    /// Accept the event.
183    ///
184    /// The event is not dropped and is sent without modification.
185    ///
186    /// Requires Loki 2.4.0 or newer.
187    #[default]
188    Accept,
189
190    /// Rewrite the timestamp of the event to the timestamp of the latest event seen by the sink.
191    RewriteTimestamp,
192
193    /// Drop the event.
194    Drop,
195}
196
197impl GenerateConfig for LokiConfig {
198    fn generate_config() -> toml::Value {
199        toml::from_str(
200            r#"endpoint = "http://localhost:3100"
201            encoding.codec = "json"
202            labels = {}"#,
203        )
204        .unwrap()
205    }
206}
207
208impl LokiConfig {
209    pub(super) fn build_client(&self, cx: SinkContext) -> crate::Result<HttpClient> {
210        let tls = TlsSettings::from_options(self.tls.as_ref())?;
211        let client = HttpClient::new(tls, cx.proxy())?;
212        Ok(client)
213    }
214}
215
216#[async_trait::async_trait]
217#[typetag::serde(name = "loki")]
218impl SinkConfig for LokiConfig {
219    async fn build(
220        &self,
221        cx: SinkContext,
222    ) -> crate::Result<(VectorSink, crate::sinks::Healthcheck)> {
223        if self.labels.is_empty() {
224            return Err("`labels` must include at least one label.".into());
225        }
226
227        for label in self.labels.keys() {
228            if !valid_label_name(label) {
229                return Err(format!("Invalid label name {:?}", label.get_ref()).into());
230            }
231        }
232
233        let client = self.build_client(cx)?;
234
235        let config = LokiConfig {
236            auth: self.auth.choose_one(&self.endpoint.auth)?,
237            ..self.clone()
238        };
239
240        let sink = LokiSink::new(config.clone(), client.clone())?;
241
242        let healthcheck = healthcheck(config, client).boxed();
243
244        self.confinement.set_confinement_gauge("sink", Self::NAME);
245        Ok((VectorSink::from_event_streamsink(sink), healthcheck))
246    }
247
248    fn input(&self) -> Input {
249        let requirement =
250            schema::Requirement::empty().optional_meaning("timestamp", Kind::timestamp());
251
252        Input::new(self.encoding.config().input_type() & DataType::Log)
253            .with_schema_requirement(requirement)
254    }
255
256    fn acknowledgements(&self) -> &AcknowledgementsConfig {
257        &self.acknowledgements
258    }
259}
260
261pub fn valid_label_name(label: &Template) -> bool {
262    label.is_dynamic() || {
263        // Loki follows prometheus on this https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels
264        // Although that isn't explicitly said anywhere besides what's in the code.
265        // The closest mention is in section about Parser Expression https://grafana.com/docs/loki/latest/logql/
266        //
267        // [a-zA-Z_][a-zA-Z0-9_]*
268        //
269        // '*' symbol at the end of the label name will be treated as a prefix for
270        // underlying object keys.
271        let mut label_trim = label.get_ref().trim();
272        if let Some(without_opening_end) = label_trim.strip_suffix('*') {
273            label_trim = without_opening_end
274        }
275
276        let mut label_chars = label_trim.chars();
277        if let Some(ch) = label_chars.next() {
278            (ch.is_ascii_alphabetic() || ch == '_')
279                && label_chars.all(|ch| ch.is_ascii_alphanumeric() || ch == '_')
280        } else {
281            label.get_ref().trim() == "*"
282        }
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use std::convert::TryInto;
289
290    use super::valid_label_name;
291    use crate::template::{ConfinementConfig, Template};
292
293    #[test]
294    fn valid_label_names() {
295        assert!(valid_label_name(&"name".try_into().unwrap()));
296        assert!(valid_label_name(&" name ".try_into().unwrap()));
297        assert!(valid_label_name(&"bee_bop".try_into().unwrap()));
298        assert!(valid_label_name(&"a09b".try_into().unwrap()));
299        assert!(valid_label_name(&"abc_*".try_into().unwrap()));
300        assert!(valid_label_name(&"_*".try_into().unwrap()));
301        assert!(valid_label_name(&"*".try_into().unwrap()));
302
303        assert!(!valid_label_name(&"0ab".try_into().unwrap()));
304        assert!(!valid_label_name(&"".try_into().unwrap()));
305        assert!(!valid_label_name(&" ".try_into().unwrap()));
306
307        assert!(valid_label_name(&"{{field}}".try_into().unwrap()));
308    }
309
310    #[test]
311    fn confinement_rejects_unconfined_tenant_id() {
312        let template = Template::try_from("{{ tenant }}").unwrap();
313        let config = ConfinementConfig::default();
314        let result = template.confine(&config, "loki", "tenant_id");
315        assert!(
316            result.is_err(),
317            "bare tenant_id template with no literal prefix must be rejected"
318        );
319    }
320
321    #[test]
322    fn confinement_opt_out_allows_unconfined_tenant_id() {
323        let template = Template::try_from("{{ tenant }}").unwrap();
324        let config = ConfinementConfig {
325            dangerously_allow_unconfined_template_resolution: true,
326        };
327        let result = template.confine(&config, "loki", "tenant_id");
328        assert!(result.is_ok(), "opt-out must allow bare tenant_id template");
329    }
330
331    #[test]
332    fn confinement_prefixed_tenant_id_locks_org_prefix() {
333        use crate::event::{Event, LogEvent};
334        use vrl::event_path;
335        // "team-{{ org }}" has literal prefix "team-"; an attacker controlling `org`
336        // cannot steer the rendered value to an org outside the "team-" namespace.
337        let template = Template::try_from("team-{{ org }}").unwrap();
338        let config = ConfinementConfig::default();
339        let confined = template.confine(&config, "loki", "tenant_id").unwrap();
340        let mut event = LogEvent::default();
341        event.insert(event_path!("org"), "other-tenant-entirely");
342        let rendered = confined.render_string(&Event::Log(event)).unwrap();
343        assert!(
344            rendered.starts_with("team-"),
345            "operator-controlled prefix must be preserved in rendered tenant_id"
346        );
347    }
348}
vector/sinks/loki/config.rs

vector/sinks/loki/
config.rs
