From c04536b13249bfebdaee0992c51966e2edaa7906 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 21 Feb 2025 18:49:25 +0800 Subject: [PATCH] feat(logging): add Kafka and Webhook audit targets - Implemented `KafkaAuditTarget` to send audit entries to a Kafka topic. - Implemented `WebhookAuditTarget` to send audit entries to a specified webhook URL. - Updated `AuditLogger` to support multiple audit targets including Kafka and Webhook. - Added examples and documentation for the new audit targets. --- Cargo.lock | 33 ++++++++++++++ Cargo.toml | 1 + packages/logging/Cargo.toml | 8 ++++ packages/logging/src/audit.rs | 84 +++++++++++++++++++++++++++++++++++ packages/logging/src/lib.rs | 12 ++++- 5 files changed, 137 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index d6a65c06e..1df7969a7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5192,6 +5192,36 @@ version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20675572f6f24e9e76ef639bc5552774ed45f1c30e2951e1e99c59888861c539" +[[package]] +name = "rdkafka" +version = "0.37.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "14b52c81ac3cac39c9639b95c20452076e74b8d9a71bc6fc4d83407af2ea6fff" +dependencies = [ + "futures-channel", + "futures-util", + "libc", + "log", + "rdkafka-sys", + "serde", + "serde_derive", + "serde_json", + "slab", + "tokio", +] + +[[package]] +name = "rdkafka-sys" +version = "4.8.0+2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ced38182dc436b3d9df0c77976f37a67134df26b050df1f0006688e46fc4c8be" +dependencies = [ + "libc", + "libz-sys", + "num_enum", + "pkg-config", +] + [[package]] name = "reader" version = "0.0.1" @@ -5557,7 +5587,10 @@ dependencies = [ "opentelemetry-semantic-conventions", "opentelemetry-stdout", "opentelemetry_sdk", + "rdkafka", + "reqwest", "serde", + "serde_json", "tokio", "tracing", "tracing-opentelemetry", diff --git a/Cargo.toml b/Cargo.toml index 511155ef1..581799808 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,6 +74,7 @@ protobuf = "3.7" protos = { path = "./common/protos" } rand = "0.8.5" reqwest = { version = "0.12.12", default-features = false, features = ["rustls-tls", "charset", "http2", "macos-system-configuration", "stream"] } +rdkafka = { version = "0.37", features = ["tokio"] } rfd = "0.15.2" rmp = "0.8.14" rmp-serde = "1.3.0" diff --git a/packages/logging/Cargo.toml b/packages/logging/Cargo.toml index af37740d9..5b8f84493 100644 --- a/packages/logging/Cargo.toml +++ b/packages/logging/Cargo.toml @@ -9,6 +9,11 @@ version.workspace = true [lints] workspace = true +[features] +default = [] +audit-kafka = ["dep:rdkafka", "dep:serde_json"] +audit-webhook = ["dep:reqwest"] + [dependencies] opentelemetry = { workspace = true } opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] } @@ -20,6 +25,9 @@ tracing = { workspace = true } tracing-opentelemetry = { workspace = true } tracing-subscriber = { workspace = true, features = ["fmt", "env-filter", "tracing-log", "time", "local-time", "json"] } tokio = { workspace = true, features = ["full"] } +rdkafka = { workspace = true, features = ["tokio"], optional = true } +reqwest = { workspace = true, features = ["json"], optional = true } +serde_json = { workspace = true, optional = true } diff --git a/packages/logging/src/audit.rs b/packages/logging/src/audit.rs index 0f120db2c..311d31044 100644 --- a/packages/logging/src/audit.rs +++ b/packages/logging/src/audit.rs @@ -1,3 +1,12 @@ +#[cfg(feature = "audit-kafka")] +use rdkafka::{ + producer::{FutureProducer, FutureRecord}, + ClientConfig, +}; + +#[cfg(feature = "audit-webhook")] +use reqwest::Client; + use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; @@ -72,6 +81,80 @@ impl AuditTarget for FileAuditTarget { } } +#[cfg(feature = "audit-webhook")] +/// Webhook audit objectives +pub struct WebhookAuditTarget { + client: Client, + url: String, +} + +#[cfg(feature = "audit-webhook")] +impl WebhookAuditTarget { + pub fn new(url: &str) -> Self { + Self { + client: Client::new(), + url: url.to_string(), + } + } +} + +#[cfg(feature = "audit-webhook")] +impl AuditTarget for WebhookAuditTarget { + fn send(&self, entry: AuditEntry) { + let client = self.client.clone(); + let url = self.url.clone(); + tokio::spawn(async move { + if let Err(e) = client.post(&url).json(&entry).send().await { + eprintln!("Failed to send to Webhook: {:?}", e); + } + }); + } +} + +#[cfg(feature = "audit-kafka")] +/// Kafka audit objectives +pub struct KafkaAuditTarget { + producer: FutureProducer, + topic: String, +} + +#[cfg(feature = "audit-kafka")] +impl KafkaAuditTarget { + pub fn new(brokers: &str, topic: &str) -> Self { + let producer: FutureProducer = ClientConfig::new() + .set("bootstrap.servers", brokers) + .set("message.timeout.ms", "5000") + .create() + .expect("Kafka producer creation failed"); + Self { + producer, + topic: topic.to_string(), + } + } +} + +#[cfg(feature = "audit-kafka")] +impl AuditTarget for KafkaAuditTarget { + fn send(&self, entry: AuditEntry) { + let topic = self.topic.clone(); + let span_id = entry.span_id.clone(); + let payload = serde_json::to_string(&entry).unwrap(); + // let record = FutureRecord::to(&topic).payload(&payload).key(&span_id); + tokio::spawn({ + // 在异步闭包内部创建 record + let topic = topic; + let payload = payload; + let span_id = span_id; + let producer = self.producer.clone(); + async move { + let record = FutureRecord::to(&topic).payload(&payload).key(&span_id); + if let Err(e) = producer.send(record, std::time::Duration::from_secs(0)).await { + eprintln!("Failed to send to Kafka: {:?}", e); + } + } + }); + } +} /// AuditLogger is a logger that logs audit entries /// to multiple targets /// @@ -110,6 +193,7 @@ impl AuditLogger { /// # Example /// ``` /// use rustfs_logging::{AuditLogger, AuditEntry, FileAuditTarget}; + /// /// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]); /// ``` pub fn new(targets: Vec>) -> Self { diff --git a/packages/logging/src/lib.rs b/packages/logging/src/lib.rs index 3fdbb83d8..280d9e76a 100644 --- a/packages/logging/src/lib.rs +++ b/packages/logging/src/lib.rs @@ -9,6 +9,10 @@ /// log_info("This is an informational message"); /// log_error("This is an error message"); /// ``` +#[cfg(feature = "audit-kafka")] +pub use audit::KafkaAuditTarget; +#[cfg(feature = "audit-webhook")] +pub use audit::WebhookAuditTarget; pub use audit::{AuditEntry, AuditLogger, AuditTarget, FileAuditTarget}; pub use logger::{log_debug, log_error, log_info}; pub use telemetry::Telemetry; @@ -66,11 +70,17 @@ mod tests { } #[tokio::test] + // #[cfg(feature = "audit-webhook")] + // #[cfg(feature = "audit-kafka")] async fn test_main() { let telemetry = Telemetry::init(); // Initialize multiple audit objectives - let audit_targets: Vec> = vec![Box::new(FileAuditTarget)]; + let audit_targets: Vec> = vec![ + Box::new(FileAuditTarget), + // Box::new(KafkaAuditTarget::new("localhost:9092", "rustfs-audit")), + // Box::new(WebhookAuditTarget::new("http://localhost:8080/audit")), + ]; let audit_logger = AuditLogger::new(audit_targets); // Test the PUT operation