featherbit/plugins/native/
elasticsearch_logger.rs

1//! Elasticsearch access-logger (`elasticsearch-logger`).
2//!
3//! Ships access-log entries to Elasticsearch via its [`_bulk`] API. Each
4//! request builds one log entry (the shared [`build_entry`] shape, or a custom
5//! `log_format`) that is handed to a fire-and-forget [`BatchSink`]; a
6//! background task POSTs batches as newline-delimited JSON (NDJSON) to
7//! `<endpoint>/_bulk`.
8//!
9//! Ports APISIX's `elasticsearch-logger`: the bulk envelope pairs an
10//! `{"index":{"_index":<name>}}` action line with each entry line, exactly as
11//! `get_logger_entry` does in the Lua plugin.
12//!
13//! ## Deviations from APISIX
14//!
15//! - **No Elasticsearch version probe.** APISIX issues a `GET /` to detect the
16//!   ES major version and, for ES 5/6, adds `_type: "_doc"` to the action
17//!   line. featherbit targets ES 7+ and never emits `_type`, so it performs no
18//!   version-probe callout.
19//! - **Static index name.** APISIX resolves `{time}` strftime tokens and
20//!   `$var` references in `field.index` per request against `ctx.var`. Because
21//!   entries are flushed in batches without a request context, featherbit uses
22//!   `field.index` as a literal string.
23//!
24//! [`_bulk`]: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
25//! [`build_entry`]: crate::plugins::util::log_entry::build_entry
26
27use std::collections::HashMap;
28use std::sync::atomic::{AtomicUsize, Ordering};
29use std::sync::Arc;
30use std::time::Duration;
31
32use async_trait::async_trait;
33use base64::{engine::general_purpose::STANDARD, Engine};
34use bytes::Bytes;
35use serde_json::Value;
36
37use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
38use crate::context::Context;
39use crate::outbound::{OutboundClient, OutboundRequest};
40use crate::plugins::resources::PluginResources;
41use crate::plugins::util::log_entry::{build_entry, parse_log_format};
42use crate::plugins::{Plugin, PluginOutput, PluginResult};
43
44/// Node that batches access-log entries and ships them to Elasticsearch.
45pub struct ElasticsearchLoggerPlugin {
46    sink: BatchSink,
47    log_format: Option<HashMap<String, Value>>,
48    include_req_body: bool,
49    include_resp_body: bool,
50}
51
52/// Delivers batches to Elasticsearch's `_bulk` endpoint.
53struct EsFlusher {
54    client: Arc<OutboundClient>,
55    /// Endpoint base URLs (no trailing slash); one is chosen per flush.
56    endpoints: Vec<String>,
57    /// Rotating cursor for endpoint selection across flushes.
58    cursor: AtomicUsize,
59    /// Destination index name (`field.index`).
60    index: String,
61    /// Pre-built `Basic <base64>` header value, when auth is configured.
62    authorization: Option<String>,
63    ssl_verify: bool,
64    timeout: Duration,
65}
66
67impl ElasticsearchLoggerPlugin {
68    /// Builds the plugin from node config.
69    ///
70    /// Accepted keys:
71    /// - `endpoint_addr` (string) / `endpoint_addrs` (array of strings):
72    ///   Elasticsearch base URL(s), e.g. `http://es:9200`. At least one is
73    ///   **required**. Trailing slashes are trimmed.
74    /// - `field.index` (string, **required**): destination index name.
75    /// - `field.type` (string, optional): accepted for compatibility; not
76    ///   emitted (see module deviations).
77    /// - `auth.username` / `auth.password` (strings, optional): when both are
78    ///   present, sent as HTTP Basic auth.
79    /// - `ssl_verify` (bool, default `true`): verify TLS certificates for
80    ///   `https` endpoints.
81    /// - `timeout` (integer seconds, default `10`): per-flush HTTP deadline.
82    /// - `include_req_body` / `include_resp_body` (bool, default `false`):
83    ///   include the request/response body in the default entry.
84    /// - `log_format` (object, optional): custom flat entry of
85    ///   `name -> "$var template"`.
86    /// - Batch tuning: `batch_max_size`, `inactive_timeout`,
87    ///   `buffer_duration`, `max_retry_count`, `retry_delay`,
88    ///   `max_pending_entries` (see [`BatchConfig`]).
89    ///
90    /// ```yaml
91    /// type: elasticsearch-logger
92    /// config:
93    ///   endpoint_addr: http://es:9200
94    ///   field:
95    ///     index: services
96    ///   auth:
97    ///     username: elastic
98    ///     password: ${ES_PASSWORD}
99    ///   ssl_verify: true
100    ///   timeout: 10
101    ///   batch_max_size: 1000
102    /// ```
103    pub fn from_config(
104        config: &HashMap<String, Value>,
105        resources: &Arc<PluginResources>,
106    ) -> Result<Self, String> {
107        let endpoints = collect_endpoints(config)?;
108
109        let field = config
110            .get("field")
111            .and_then(|v| v.as_object())
112            .ok_or_else(|| "elasticsearch-logger requires 'field'".to_string())?;
113        let index = field
114            .get("index")
115            .and_then(|v| v.as_str())
116            .filter(|s| !s.is_empty())
117            .ok_or_else(|| "elasticsearch-logger requires 'field.index'".to_string())?
118            .to_string();
119
120        let authorization = config
121            .get("auth")
122            .and_then(|v| v.as_object())
123            .and_then(|auth| {
124                let user = auth.get("username").and_then(|v| v.as_str())?;
125                let pass = auth.get("password").and_then(|v| v.as_str())?;
126                Some(format!(
127                    "Basic {}",
128                    STANDARD.encode(format!("{user}:{pass}"))
129                ))
130            });
131
132        let ssl_verify = config
133            .get("ssl_verify")
134            .and_then(|v| v.as_bool())
135            .unwrap_or(true);
136        let timeout =
137            Duration::from_secs(config.get("timeout").and_then(|v| v.as_u64()).unwrap_or(10));
138
139        let include_req_body = config
140            .get("include_req_body")
141            .and_then(|v| v.as_bool())
142            .unwrap_or(false);
143        let include_resp_body = config
144            .get("include_resp_body")
145            .and_then(|v| v.as_bool())
146            .unwrap_or(false);
147        let log_format = parse_log_format(config)?;
148
149        let batch_cfg = BatchConfig::from_config(config)?;
150        let flusher = Arc::new(EsFlusher {
151            client: resources.outbound.clone(),
152            endpoints,
153            cursor: AtomicUsize::new(0),
154            index,
155            authorization,
156            ssl_verify,
157            timeout,
158        });
159        let sink = BatchSink::spawn("elasticsearch-logger", batch_cfg, flusher);
160
161        Ok(Self {
162            sink,
163            log_format,
164            include_req_body,
165            include_resp_body,
166        })
167    }
168}
169
170/// Reads `endpoint_addr` (string) and/or `endpoint_addrs` (array), trims
171/// trailing slashes, and errors when none are present.
172fn collect_endpoints(config: &HashMap<String, Value>) -> Result<Vec<String>, String> {
173    let mut endpoints = Vec::new();
174    if let Some(s) = config.get("endpoint_addr").and_then(|v| v.as_str()) {
175        endpoints.push(s.trim_end_matches('/').to_string());
176    }
177    if let Some(arr) = config.get("endpoint_addrs").and_then(|v| v.as_array()) {
178        for v in arr {
179            if let Some(s) = v.as_str() {
180                endpoints.push(s.trim_end_matches('/').to_string());
181            }
182        }
183    }
184    if endpoints.is_empty() {
185        return Err(
186            "elasticsearch-logger requires 'endpoint_addr' or 'endpoint_addrs'".to_string(),
187        );
188    }
189    Ok(endpoints)
190}
191
192/// Builds the NDJSON `_bulk` body: for every entry an action line
193/// `{"index":{"_index":<index>}}` followed by the entry, each terminated by a
194/// newline (APISIX `get_logger_entry`). Pure and network-free for testing.
195fn build_bulk_body(index: &str, entries: &[Value]) -> String {
196    let action = serde_json::json!({ "index": { "_index": index } });
197    let action_line = serde_json::to_string(&action).unwrap_or_default();
198    let mut body = String::new();
199    for entry in entries {
200        body.push_str(&action_line);
201        body.push('\n');
202        body.push_str(&serde_json::to_string(entry).unwrap_or_else(|_| "{}".to_string()));
203        body.push('\n');
204    }
205    body
206}
207
208#[async_trait]
209impl BatchFlusher for EsFlusher {
210    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
211        let idx = self.cursor.fetch_add(1, Ordering::Relaxed) % self.endpoints.len();
212        let url = format!("{}/_bulk", self.endpoints[idx]);
213        let body = build_bulk_body(&self.index, entries);
214
215        let mut headers = vec![
216            (
217                "Content-Type".to_string(),
218                "application/x-ndjson".to_string(),
219            ),
220            (
221                "Accept".to_string(),
222                "application/vnd.elasticsearch+json".to_string(),
223            ),
224        ];
225        if let Some(auth) = &self.authorization {
226            headers.push(("Authorization".to_string(), auth.clone()));
227        }
228
229        let req = OutboundRequest {
230            method: http::Method::POST,
231            url,
232            headers,
233            body: Bytes::from(body),
234            timeout: self.timeout,
235            ssl_verify: self.ssl_verify,
236            tls: None,
237        };
238
239        match self.client.request(req).await {
240            Ok(resp) if resp.status == 200 => Ok(()),
241            Ok(resp) => Err(FlushError {
242                message: format!(
243                    "elasticsearch returned status {}: {}",
244                    resp.status,
245                    String::from_utf8_lossy(&resp.body)
246                ),
247                first_fail: None,
248            }),
249            Err(e) => Err(FlushError {
250                message: format!("elasticsearch callout failed: {e}"),
251                first_fail: None,
252            }),
253        }
254    }
255}
256
257#[async_trait]
258impl Plugin for ElasticsearchLoggerPlugin {
259    fn plugin_type(&self) -> &str {
260        "elasticsearch-logger"
261    }
262
263    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
264        let entry = build_entry(
265            &ctx,
266            self.log_format.as_ref(),
267            self.include_req_body,
268            self.include_resp_body,
269        );
270        self.sink.push(entry);
271        Ok(PluginOutput {
272            context: ctx,
273            named_outputs: HashMap::new(),
274        })
275    }
276}
277
278#[cfg(test)]
279mod tests {
280    use super::*;
281    use serde_json::json;
282
283    fn cfg(v: Value) -> HashMap<String, Value> {
284        serde_json::from_value(v).unwrap()
285    }
286
287    #[tokio::test]
288    async fn requires_endpoint_and_index() {
289        // no endpoint
290        assert!(ElasticsearchLoggerPlugin::from_config(
291            &cfg(json!({ "field": { "index": "svc" } })),
292            &PluginResources::empty()
293        )
294        .is_err());
295        // no field.index
296        assert!(ElasticsearchLoggerPlugin::from_config(
297            &cfg(json!({ "endpoint_addr": "http://es:9200" })),
298            &PluginResources::empty()
299        )
300        .is_err());
301        // both present
302        assert!(ElasticsearchLoggerPlugin::from_config(
303            &cfg(json!({ "endpoint_addr": "http://es:9200/", "field": { "index": "svc" } })),
304            &PluginResources::empty()
305        )
306        .is_ok());
307    }
308
309    #[test]
310    fn collects_and_trims_endpoints() {
311        let e = collect_endpoints(&cfg(json!({
312            "endpoint_addr": "http://a:9200/",
313            "endpoint_addrs": ["http://b:9200/", "http://c:9200"]
314        })))
315        .unwrap();
316        assert_eq!(e, vec!["http://a:9200", "http://b:9200", "http://c:9200"]);
317    }
318
319    #[tokio::test]
320    async fn basic_auth_header_encoded() {
321        let p = ElasticsearchLoggerPlugin::from_config(
322            &cfg(json!({
323                "endpoint_addr": "http://es:9200",
324                "field": { "index": "svc" },
325                "auth": { "username": "elastic", "password": "secret" }
326            })),
327            &PluginResources::empty(),
328        );
329        assert!(p.is_ok());
330    }
331
332    #[test]
333    fn bulk_body_ndjson_shape() {
334        let entries = vec![json!({ "a": 1 }), json!({ "b": 2 })];
335        let body = build_bulk_body("svc", &entries);
336        let lines: Vec<&str> = body.split('\n').collect();
337        // action\nentry\naction\nentry\n -> 5 parts (trailing empty)
338        assert_eq!(lines.len(), 5);
339        assert_eq!(lines[0], r#"{"index":{"_index":"svc"}}"#);
340        assert_eq!(lines[1], r#"{"a":1}"#);
341        assert_eq!(lines[2], r#"{"index":{"_index":"svc"}}"#);
342        assert_eq!(lines[3], r#"{"b":2}"#);
343        assert_eq!(lines[4], "");
344        assert!(body.ends_with('\n'));
345    }
346}
featherbit/plugins/native/elasticsearch_logger.rs

featherbit/plugins/native/
elasticsearch_logger.rs
