featherbit/plugins/native/
tencent_cloud_cls.rs

1//! Tencent Cloud CLS (Cloud Log Service) access-logger (`tencent-cloud-cls`).
2//!
3//! Ships access-log entries to Tencent Cloud's Log Service. Each request builds
4//! one log entry (the shared [`build_entry`] shape, or a custom `log_format`)
5//! that is handed to a fire-and-forget [`BatchSink`]; a background task POSTs
6//! batches to the CLS `/structuredlog` upload endpoint, signed with the
7//! CLS/COS-style `q-sign-algorithm=sha1` `Authorization` header.
8//!
9//! The signature is ported faithfully from APISIX's `cls-sdk.lua` `sign()`:
10//! `sign_key = hex(hmac_sha1(secret_key, sign_time))`, then
11//! `signature = hex(hmac_sha1(sign_key, string_to_sign))`, where
12//! `string_to_sign = "sha1\n<sign_time>\n<sha1(http_request_info)>\n"` and
13//! `http_request_info = "post\n/structuredlog\n\n\n"`. See
14//! <https://cloud.tencent.com/document/product/614/12445>.
15//!
16//! ## Deviations from APISIX
17//!
18//! - **JSON body, not protobuf.** The CLS SDK serializes the `LogGroupList`
19//!   with protobuf and sends `application/x-protobuf`. featherbit has no
20//!   protobuf codec, so it sends the equivalent structured-log payload as JSON
21//!   (`application/json`): each entry is normalized to a list of
22//!   `{key, value}` `contents` (non-string values JSON-encoded), grouped into
23//!   one `LogGroup`. The signature, endpoint, topic query parameter, and log
24//!   normalization are otherwise faithful. Against a live CLS endpoint the
25//!   protobuf content type would be required; this is documented as a subset.
26//! - **`source` omitted.** The SDK sets each `LogGroup.source` to the host IP;
27//!   featherbit does not resolve its own IP and leaves it empty.
28//!
29//! The signing helper and its primitives (SHA-1, HMAC-SHA1) are unit-tested
30//! against fixed vectors.
31
32use std::collections::HashMap;
33use std::sync::Arc;
34use std::time::{Duration, SystemTime, UNIX_EPOCH};
35
36use async_trait::async_trait;
37use bytes::Bytes;
38use serde_json::Value;
39
40use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
41use crate::context::Context;
42use crate::outbound::{OutboundClient, OutboundRequest};
43use crate::plugins::resources::PluginResources;
44use crate::plugins::util::log_entry::{build_entry, parse_log_format};
45use crate::plugins::{Plugin, PluginOutput, PluginResult};
46
47const CLS_API_PATH: &str = "/structuredlog";
48const AUTH_EXPIRE_SECS: u64 = 60;
49
50/// Node that batches access-log entries and ships them to Tencent Cloud CLS.
51pub struct TencentCloudClsPlugin {
52    sink: BatchSink,
53    log_format: Option<HashMap<String, Value>>,
54    include_req_body: bool,
55    include_resp_body: bool,
56    /// Static tags merged into every entry before batching (`global_tag`).
57    global_tag: HashMap<String, Value>,
58}
59
60/// Delivers batches to the CLS `/structuredlog` endpoint.
61struct ClsFlusher {
62    client: Arc<OutboundClient>,
63    scheme: String,
64    host: String,
65    topic: String,
66    secret_id: String,
67    secret_key: String,
68    ssl_verify: bool,
69    timeout: Duration,
70}
71
72impl TencentCloudClsPlugin {
73    /// Builds the plugin from node config.
74    ///
75    /// Accepted keys:
76    /// - `cls_host` (alias `endpoint`) (string, **required**): CLS upload host,
77    ///   e.g. `ap-guangzhou.cls.tencentcs.com`.
78    /// - `cls_topic` (alias `topic_id`) (string, **required**): destination
79    ///   topic id, sent as the `topic_id` query parameter.
80    /// - `secret_id` / `secret_key` (strings, **required**): API credentials.
81    /// - `scheme` (`http`|`https`, default `https`).
82    /// - `ssl_verify` (bool, default `true`): verify TLS certificates.
83    /// - `timeout` (integer ms, default `10000`): per-flush HTTP deadline.
84    /// - `global_tag` (object, optional): fields merged into every entry.
85    /// - `include_req_body` / `include_resp_body` (bool, default `false`).
86    /// - `log_format` (object, optional): custom flat entry.
87    /// - Batch tuning (see [`BatchConfig`]).
88    ///
89    /// ```yaml
90    /// type: tencent-cloud-cls
91    /// config:
92    ///   cls_host: ap-guangzhou.cls.tencentcs.com
93    ///   cls_topic: xxxxxxxx-xxxx-xxxx
94    ///   secret_id: ${CLS_SECRET_ID}
95    ///   secret_key: ${CLS_SECRET_KEY}
96    ///   scheme: https
97    /// ```
98    pub fn from_config(
99        config: &HashMap<String, Value>,
100        resources: &Arc<PluginResources>,
101    ) -> Result<Self, String> {
102        let host = string_alias(config, "cls_host", "endpoint")
103            .ok_or_else(|| "tencent-cloud-cls requires 'cls_host'".to_string())?;
104        let topic = string_alias(config, "cls_topic", "topic_id")
105            .ok_or_else(|| "tencent-cloud-cls requires 'cls_topic'".to_string())?;
106        let secret_id = required_string(config, "secret_id")?;
107        let secret_key = required_string(config, "secret_key")?;
108
109        let scheme = config
110            .get("scheme")
111            .and_then(|v| v.as_str())
112            .filter(|s| *s == "http" || *s == "https")
113            .unwrap_or("https")
114            .to_string();
115        let ssl_verify = config
116            .get("ssl_verify")
117            .and_then(|v| v.as_bool())
118            .unwrap_or(true);
119        let timeout = Duration::from_millis(
120            config
121                .get("timeout")
122                .and_then(|v| v.as_u64())
123                .unwrap_or(10000),
124        );
125
126        let global_tag = config
127            .get("global_tag")
128            .and_then(|v| v.as_object())
129            .map(|m| m.clone().into_iter().collect())
130            .unwrap_or_default();
131
132        let include_req_body = config
133            .get("include_req_body")
134            .and_then(|v| v.as_bool())
135            .unwrap_or(false);
136        let include_resp_body = config
137            .get("include_resp_body")
138            .and_then(|v| v.as_bool())
139            .unwrap_or(false);
140        let log_format = parse_log_format(config)?;
141
142        let batch_cfg = BatchConfig::from_config(config)?;
143        let flusher = Arc::new(ClsFlusher {
144            client: resources.outbound.clone(),
145            scheme,
146            host,
147            topic,
148            secret_id,
149            secret_key,
150            ssl_verify,
151            timeout,
152        });
153        let sink = BatchSink::spawn("tencent-cloud-cls", batch_cfg, flusher);
154
155        Ok(Self {
156            sink,
157            log_format,
158            include_req_body,
159            include_resp_body,
160            global_tag,
161        })
162    }
163}
164
165fn required_string(config: &HashMap<String, Value>, key: &str) -> Result<String, String> {
166    config
167        .get(key)
168        .and_then(|v| v.as_str())
169        .filter(|s| !s.is_empty())
170        .map(|s| s.to_string())
171        .ok_or_else(|| format!("tencent-cloud-cls requires '{key}'"))
172}
173
174fn string_alias(config: &HashMap<String, Value>, primary: &str, alias: &str) -> Option<String> {
175    config
176        .get(primary)
177        .or_else(|| config.get(alias))
178        .and_then(|v| v.as_str())
179        .filter(|s| !s.is_empty())
180        .map(|s| s.to_string())
181}
182
183/// Builds the CLS structured-log payload: one `LogGroup` whose `logs` carry
184/// the normalized `{key, value}` `contents` of each entry. Pure and
185/// network-free for testing.
186fn build_log_payload(entries: &[Value], now_ms: u64) -> Value {
187    let logs: Vec<Value> = entries
188        .iter()
189        .map(|entry| {
190            let contents: Vec<Value> = match entry.as_object() {
191                Some(map) => map
192                    .iter()
193                    .map(|(k, v)| serde_json::json!({ "key": k, "value": stringify(v) }))
194                    .collect(),
195                None => vec![serde_json::json!({ "key": "log", "value": stringify(entry) })],
196            };
197            serde_json::json!({ "time": now_ms, "contents": contents })
198        })
199        .collect();
200    serde_json::json!({ "logGroupList": [ { "logs": logs } ] })
201}
202
203/// CLS `content.value` is a string; keep strings as-is, JSON-encode the rest.
204fn stringify(v: &Value) -> String {
205    match v {
206        Value::String(s) => s.clone(),
207        other => other.to_string(),
208    }
209}
210
211/// Builds the CLS `Authorization` header value for the current time.
212/// Faithful port of `cls-sdk.lua`'s `sign()`. `cur_time` (Unix seconds) is
213/// injected so the signature is unit-testable.
214fn sign(secret_id: &str, secret_key: &str, cur_time: u64) -> String {
215    let http_request_info = format!("post\n{CLS_API_PATH}\n\n\n");
216    let sign_time = format!("{};{}", cur_time, cur_time + AUTH_EXPIRE_SECS);
217    let string_to_sign = format!(
218        "sha1\n{sign_time}\n{}\n",
219        sha1_hex(http_request_info.as_bytes())
220    );
221
222    let sign_key = hmac_sha1_hex(secret_key.as_bytes(), sign_time.as_bytes());
223    let signature = hmac_sha1_hex(sign_key.as_bytes(), string_to_sign.as_bytes());
224
225    [
226        "q-sign-algorithm=sha1".to_string(),
227        format!("q-ak={secret_id}"),
228        format!("q-sign-time={sign_time}"),
229        format!("q-key-time={sign_time}"),
230        "q-header-list=".to_string(),
231        "q-url-param-list=".to_string(),
232        format!("q-signature={signature}"),
233    ]
234    .join("&")
235}
236
237/// Lowercase hex SHA-1 (`str_util.to_hex(ngx_sha1_bin(msg))`).
238fn sha1_hex(msg: &[u8]) -> String {
239    let digest = ring::digest::digest(&ring::digest::SHA1_FOR_LEGACY_USE_ONLY, msg);
240    to_hex(digest.as_ref())
241}
242
243/// Lowercase hex HMAC-SHA1 (`str_util.to_hex(ngx_hmac_sha1(key, msg))`).
244fn hmac_sha1_hex(key: &[u8], msg: &[u8]) -> String {
245    let k = ring::hmac::Key::new(ring::hmac::HMAC_SHA1_FOR_LEGACY_USE_ONLY, key);
246    to_hex(ring::hmac::sign(&k, msg).as_ref())
247}
248
249fn to_hex(bytes: &[u8]) -> String {
250    let mut s = String::with_capacity(bytes.len() * 2);
251    for b in bytes {
252        s.push_str(&format!("{b:02x}"));
253    }
254    s
255}
256
257#[async_trait]
258impl BatchFlusher for ClsFlusher {
259    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
260        let now = SystemTime::now()
261            .duration_since(UNIX_EPOCH)
262            .unwrap_or_default();
263        let now_ms = now.as_millis() as u64;
264        let payload = build_log_payload(entries, now_ms);
265        let body = serde_json::to_vec(&payload).map_err(|e| FlushError {
266            message: format!("cls payload encode failed: {e}"),
267            first_fail: None,
268        })?;
269
270        let authorization = sign(&self.secret_id, &self.secret_key, now.as_secs());
271        let url = format!(
272            "{}://{}{}?topic_id={}",
273            self.scheme, self.host, CLS_API_PATH, self.topic
274        );
275        let headers = vec![
276            ("Host".to_string(), self.host.clone()),
277            ("Content-Type".to_string(), "application/json".to_string()),
278            ("Authorization".to_string(), authorization),
279        ];
280
281        let req = OutboundRequest {
282            method: http::Method::POST,
283            url,
284            headers,
285            body: Bytes::from(body),
286            timeout: self.timeout,
287            ssl_verify: self.ssl_verify,
288            tls: None,
289        };
290
291        match self.client.request(req).await {
292            Ok(resp) if resp.status == 200 => Ok(()),
293            // 413/404/401/403 are non-retryable per the SDK; treat as delivered.
294            Ok(resp) if matches!(resp.status, 401 | 403 | 404 | 413) => {
295                tracing::error!(
296                    status = resp.status,
297                    "tencent-cloud-cls non-retryable error, dropping batch"
298                );
299                Ok(())
300            }
301            Ok(resp) => Err(FlushError {
302                message: format!(
303                    "cls returned status {}: {}",
304                    resp.status,
305                    String::from_utf8_lossy(&resp.body)
306                ),
307                first_fail: None,
308            }),
309            Err(e) => Err(FlushError {
310                message: format!("cls callout failed: {e}"),
311                first_fail: None,
312            }),
313        }
314    }
315}
316
317#[async_trait]
318impl Plugin for TencentCloudClsPlugin {
319    fn plugin_type(&self) -> &str {
320        "tencent-cloud-cls"
321    }
322
323    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
324        let mut entry = build_entry(
325            &ctx,
326            self.log_format.as_ref(),
327            self.include_req_body,
328            self.include_resp_body,
329        );
330        if !self.global_tag.is_empty() {
331            if let Some(map) = entry.as_object_mut() {
332                for (k, v) in &self.global_tag {
333                    map.insert(k.clone(), v.clone());
334                }
335            }
336        }
337        self.sink.push(entry);
338        Ok(PluginOutput {
339            context: ctx,
340            named_outputs: HashMap::new(),
341        })
342    }
343}
344
345#[cfg(test)]
346mod tests {
347    use super::*;
348    use serde_json::json;
349
350    fn cfg(v: Value) -> HashMap<String, Value> {
351        serde_json::from_value(v).unwrap()
352    }
353
354    fn full_cfg() -> Value {
355        json!({
356            "cls_host": "ap-guangzhou.cls.tencentcs.com",
357            "cls_topic": "topic-123",
358            "secret_id": "id",
359            "secret_key": "secret"
360        })
361    }
362
363    #[tokio::test]
364    async fn requires_host_topic_and_keys() {
365        assert!(TencentCloudClsPlugin::from_config(
366            &cfg(json!({ "cls_topic": "t", "secret_id": "i", "secret_key": "k" })),
367            &PluginResources::empty()
368        )
369        .is_err());
370        assert!(
371            TencentCloudClsPlugin::from_config(&cfg(full_cfg()), &PluginResources::empty()).is_ok()
372        );
373    }
374
375    #[tokio::test]
376    async fn accepts_endpoint_and_topic_id_aliases() {
377        assert!(TencentCloudClsPlugin::from_config(
378            &cfg(json!({
379                "endpoint": "h", "topic_id": "t", "secret_id": "i", "secret_key": "k"
380            })),
381            &PluginResources::empty()
382        )
383        .is_ok());
384    }
385
386    #[test]
387    fn sha1_known_vector() {
388        assert_eq!(sha1_hex(b"abc"), "a9993e364706816aba3e25717850c26c9cd0d89d");
389        assert_eq!(sha1_hex(b""), "da39a3ee5e6b4b0d3255bfef95601890afd80709");
390    }
391
392    #[test]
393    fn hmac_sha1_rfc2202_vector() {
394        // RFC 2202 test case 2: key = "Jefe", data = "what do ya want ...".
395        let mac = hmac_sha1_hex(b"Jefe", b"what do ya want for nothing?");
396        assert_eq!(mac, "effcdf6ae5eb2fa2d27416d5f184df9c259a7c79");
397    }
398
399    #[test]
400    fn sign_is_stable_and_shaped() {
401        let a = sign("AKID", "AKSECRET", 1_600_000_000);
402        let b = sign("AKID", "AKSECRET", 1_600_000_000);
403        assert_eq!(a, b);
404        assert!(a.starts_with("q-sign-algorithm=sha1&q-ak=AKID"));
405        assert!(a.contains("q-sign-time=1600000000;1600000060"));
406        // Stable signature (recomputed vector).
407        assert!(
408            a.ends_with("&q-signature=690a6e12e797585ceb04a4f21fd5e3886f997972"),
409            "unexpected signature: {a}"
410        );
411    }
412
413    #[test]
414    fn log_payload_contents_shape() {
415        let entries = vec![json!({ "status": 200, "path": "/x" })];
416        let payload = build_log_payload(&entries, 1234);
417        let log = &payload["logGroupList"][0]["logs"][0];
418        assert_eq!(log["time"], json!(1234));
419        let contents = log["contents"].as_array().unwrap();
420        assert_eq!(contents.len(), 2);
421        // values are all strings
422        for c in contents {
423            assert!(c["value"].is_string());
424        }
425    }
426}
featherbit/plugins/native/tencent_cloud_cls.rs

featherbit/plugins/native/
tencent_cloud_cls.rs
