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, LogFormat};
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<LogFormat>,
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 `name ->
85    ///   "template"` (`{{namespace.path}}` references plus legacy `$var`
86    ///   interpolation).
87    /// - Batch tuning: `batch_max_size`, `inactive_timeout`,
88    ///   `buffer_duration`, `max_retry_count`, `retry_delay`,
89    ///   `max_pending_entries` (see [`BatchConfig`]).
90    ///
91    /// ```yaml
92    /// type: elasticsearch-logger
93    /// config:
94    ///   endpoint_addr: http://es:9200
95    ///   field:
96    ///     index: services
97    ///   auth:
98    ///     username: elastic
99    ///     password: ${ES_PASSWORD}
100    ///   ssl_verify: true
101    ///   timeout: 10
102    ///   batch_max_size: 1000
103    /// ```
104    pub fn from_config(
105        config: &HashMap<String, Value>,
106        resources: &Arc<PluginResources>,
107    ) -> Result<Self, String> {
108        let endpoints = collect_endpoints(config)?;
109
110        let field = config
111            .get("field")
112            .and_then(|v| v.as_object())
113            .ok_or_else(|| "elasticsearch-logger requires 'field'".to_string())?;
114        let index = field
115            .get("index")
116            .and_then(|v| v.as_str())
117            .filter(|s| !s.is_empty())
118            .ok_or_else(|| "elasticsearch-logger requires 'field.index'".to_string())?
119            .to_string();
120
121        let authorization = config
122            .get("auth")
123            .and_then(|v| v.as_object())
124            .and_then(|auth| {
125                let user = auth.get("username").and_then(|v| v.as_str())?;
126                let pass = auth.get("password").and_then(|v| v.as_str())?;
127                Some(format!(
128                    "Basic {}",
129                    STANDARD.encode(format!("{user}:{pass}"))
130                ))
131            });
132
133        let ssl_verify = config
134            .get("ssl_verify")
135            .and_then(|v| v.as_bool())
136            .unwrap_or(true);
137        let timeout =
138            Duration::from_secs(config.get("timeout").and_then(|v| v.as_u64()).unwrap_or(10));
139
140        let include_req_body = config
141            .get("include_req_body")
142            .and_then(|v| v.as_bool())
143            .unwrap_or(false);
144        let include_resp_body = config
145            .get("include_resp_body")
146            .and_then(|v| v.as_bool())
147            .unwrap_or(false);
148        let log_format = parse_log_format(config)?;
149
150        let batch_cfg = BatchConfig::from_config(config)?;
151        let flusher = Arc::new(EsFlusher {
152            client: resources.outbound.clone(),
153            endpoints,
154            cursor: AtomicUsize::new(0),
155            index,
156            authorization,
157            ssl_verify,
158            timeout,
159        });
160        let sink = BatchSink::spawn("elasticsearch-logger", batch_cfg, flusher);
161
162        Ok(Self {
163            sink,
164            log_format,
165            include_req_body,
166            include_resp_body,
167        })
168    }
169}
170
171/// Reads `endpoint_addr` (string) and/or `endpoint_addrs` (array), trims
172/// trailing slashes, and errors when none are present.
173fn collect_endpoints(config: &HashMap<String, Value>) -> Result<Vec<String>, String> {
174    let mut endpoints = Vec::new();
175    if let Some(s) = config.get("endpoint_addr").and_then(|v| v.as_str()) {
176        endpoints.push(s.trim_end_matches('/').to_string());
177    }
178    if let Some(arr) = config.get("endpoint_addrs").and_then(|v| v.as_array()) {
179        for v in arr {
180            if let Some(s) = v.as_str() {
181                endpoints.push(s.trim_end_matches('/').to_string());
182            }
183        }
184    }
185    if endpoints.is_empty() {
186        return Err(
187            "elasticsearch-logger requires 'endpoint_addr' or 'endpoint_addrs'".to_string(),
188        );
189    }
190    Ok(endpoints)
191}
192
193/// Builds the NDJSON `_bulk` body: for every entry an action line
194/// `{"index":{"_index":<index>}}` followed by the entry, each terminated by a
195/// newline (APISIX `get_logger_entry`). Pure and network-free for testing.
196fn build_bulk_body(index: &str, entries: &[Value]) -> String {
197    let action = serde_json::json!({ "index": { "_index": index } });
198    let action_line = serde_json::to_string(&action).unwrap_or_default();
199    let mut body = String::new();
200    for entry in entries {
201        body.push_str(&action_line);
202        body.push('\n');
203        body.push_str(&serde_json::to_string(entry).unwrap_or_else(|_| "{}".to_string()));
204        body.push('\n');
205    }
206    body
207}
208
209#[async_trait]
210impl BatchFlusher for EsFlusher {
211    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
212        let idx = self.cursor.fetch_add(1, Ordering::Relaxed) % self.endpoints.len();
213        let url = format!("{}/_bulk", self.endpoints[idx]);
214        let body = build_bulk_body(&self.index, entries);
215
216        let mut headers = vec![
217            (
218                "Content-Type".to_string(),
219                "application/x-ndjson".to_string(),
220            ),
221            (
222                "Accept".to_string(),
223                "application/vnd.elasticsearch+json".to_string(),
224            ),
225        ];
226        if let Some(auth) = &self.authorization {
227            headers.push(("Authorization".to_string(), auth.clone()));
228        }
229
230        let req = OutboundRequest {
231            method: http::Method::POST,
232            url,
233            headers,
234            body: Bytes::from(body),
235            timeout: self.timeout,
236            ssl_verify: self.ssl_verify,
237            tls: None,
238        };
239
240        match self.client.request(req).await {
241            Ok(resp) if resp.status == 200 => Ok(()),
242            Ok(resp) => Err(FlushError {
243                message: format!(
244                    "elasticsearch returned status {}: {}",
245                    resp.status,
246                    String::from_utf8_lossy(&resp.body)
247                ),
248                first_fail: None,
249            }),
250            Err(e) => Err(FlushError {
251                message: format!("elasticsearch callout failed: {e}"),
252                first_fail: None,
253            }),
254        }
255    }
256}
257
258#[async_trait]
259impl Plugin for ElasticsearchLoggerPlugin {
260    fn plugin_type(&self) -> &str {
261        "elasticsearch-logger"
262    }
263
264    fn reads_response_body(&self) -> bool {
265        crate::plugins::util::log_entry::reads_response_body(
266            self.log_format.as_ref(),
267            self.include_resp_body,
268        )
269    }
270
271    async fn execute(&self, ctx: Context) -> PluginResult {
272        let entry = build_entry(
273            &ctx,
274            self.log_format.as_ref(),
275            self.include_req_body,
276            self.include_resp_body,
277        );
278        self.sink.push(entry);
279        Ok(PluginOutput::success(ctx))
280    }
281}
282
283#[cfg(test)]
284mod tests {
285    use super::*;
286    use serde_json::json;
287
288    fn cfg(v: Value) -> HashMap<String, Value> {
289        serde_json::from_value(v).unwrap()
290    }
291
292    #[tokio::test]
293    async fn requires_endpoint_and_index() {
294        // no endpoint
295        assert!(ElasticsearchLoggerPlugin::from_config(
296            &cfg(json!({ "field": { "index": "svc" } })),
297            &PluginResources::empty()
298        )
299        .is_err());
300        // no field.index
301        assert!(ElasticsearchLoggerPlugin::from_config(
302            &cfg(json!({ "endpoint_addr": "http://es:9200" })),
303            &PluginResources::empty()
304        )
305        .is_err());
306        // both present
307        assert!(ElasticsearchLoggerPlugin::from_config(
308            &cfg(json!({ "endpoint_addr": "http://es:9200/", "field": { "index": "svc" } })),
309            &PluginResources::empty()
310        )
311        .is_ok());
312    }
313
314    #[test]
315    fn collects_and_trims_endpoints() {
316        let e = collect_endpoints(&cfg(json!({
317            "endpoint_addr": "http://a:9200/",
318            "endpoint_addrs": ["http://b:9200/", "http://c:9200"]
319        })))
320        .unwrap();
321        assert_eq!(e, vec!["http://a:9200", "http://b:9200", "http://c:9200"]);
322    }
323
324    #[tokio::test]
325    async fn basic_auth_header_encoded() {
326        let p = ElasticsearchLoggerPlugin::from_config(
327            &cfg(json!({
328                "endpoint_addr": "http://es:9200",
329                "field": { "index": "svc" },
330                "auth": { "username": "elastic", "password": "secret" }
331            })),
332            &PluginResources::empty(),
333        );
334        assert!(p.is_ok());
335    }
336
337    #[test]
338    fn bulk_body_ndjson_shape() {
339        let entries = vec![json!({ "a": 1 }), json!({ "b": 2 })];
340        let body = build_bulk_body("svc", &entries);
341        let lines: Vec<&str> = body.split('\n').collect();
342        // action\nentry\naction\nentry\n -> 5 parts (trailing empty)
343        assert_eq!(lines.len(), 5);
344        assert_eq!(lines[0], r#"{"index":{"_index":"svc"}}"#);
345        assert_eq!(lines[1], r#"{"a":1}"#);
346        assert_eq!(lines[2], r#"{"index":{"_index":"svc"}}"#);
347        assert_eq!(lines[3], r#"{"b":2}"#);
348        assert_eq!(lines[4], "");
349        assert!(body.ends_with('\n'));
350    }
351}
featherbit/plugins/native/elasticsearch_logger.rs

featherbit/plugins/native/
elasticsearch_logger.rs
