rsearch-ingest 0.2.0

Ingest path for rSearch: bulk parsing, append-before-ack WAL, and the indexer pipeline
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

//! GELF message parsing (Graylog Extended Log Format), TCP null-framed.

use serde_json::{Value, json};

/// Convert a GELF JSON payload into a log document: `short_message`
/// becomes `message`, additional `_fields` lose their underscore, and
/// the epoch-seconds `timestamp` passes through for the converter.
pub fn parse_gelf(payload: &[u8]) -> Option<Value> {
    let gelf: Value = serde_json::from_slice(payload).ok()?;
    let obj = gelf.as_object()?;
    let mut doc = serde_json::Map::new();
    for (key, value) in obj {
        match key.as_str() {
            "version" => {}
            "short_message" => {
                doc.insert("message".to_string(), value.clone());
            }
            "full_message" => {
                doc.insert("full_message".to_string(), value.clone());
            }
            "timestamp" => {
                doc.insert("@timestamp".to_string(), value.clone());
            }
            "level" => {
                doc.insert("severity".to_string(), value.clone());
            }
            other => {
                let name = other.strip_prefix('_').unwrap_or(other);
                doc.insert(name.to_string(), value.clone());
            }
        }
    }
    doc.insert("source".to_string(), json!("gelf"));
    Some(Value::Object(doc))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_gelf_message() {
        let payload = serde_json::to_vec(&json!({
            "version": "1.1",
            "host": "app01",
            "short_message": "job failed",
            "timestamp": 1753333200.123,
            "level": 3,
            "_service": "billing",
            "_attempt": 4,
        }))
        .unwrap();
        let doc = parse_gelf(&payload).unwrap();
        assert_eq!(doc["message"], "job failed");
        assert_eq!(doc["severity"], 3);
        assert_eq!(doc["service"], "billing");
        assert_eq!(doc["attempt"], 4);
        assert_eq!(doc["host"], "app01");
        assert_eq!(doc["@timestamp"], 1753333200.123);
        assert!(doc.get("version").is_none());
    }

    #[test]
    fn rejects_non_json() {
        assert!(parse_gelf(b"nope").is_none());
    }
}