mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-07 13:53:12 +00:00
houseme
dd7da015e3
* init rustfs config * improve code for rustfs-config crate * add * improve code for comment * fix: modify rustfs-config crate name * add default fn * improve error logger * fix: modify docker config yaml * improve code for config * feat: restrict kafka feature to Linux only - Add target-specific feature configuration in Cargo.toml for obs and event-notifier crates - Implement conditional compilation for kafka feature only on Linux systems - Add appropriate error handling for non-Linux platforms - Ensure backward compatibility with existing code * refactor(ci): optimize build workflow for better efficiency - Integrate GUI build steps into main build-rustfs job - Add conditional GUI build execution based on tag releases - Simplify workflow by removing redundant build-rustfs-gui job - Copy binary directly to embedded-rustfs directory without downloading artifacts - Update merge job dependency to only rely on build-rustfs - Improve cross-platform compatibility for Windows binary naming (.exe) - Streamline artifact uploading and OSS publishing process - Maintain consistent conditional logic for release operations * refactor(ci): optimize build workflow for better efficiency - Integrate GUI build steps into main build-rustfs job - Add conditional GUI build execution based on tag releases - Simplify workflow by removing redundant build-rustfs-gui job - Copy binary directly to embedded-rustfs directory without downloading artifacts - Update merge job dependency to only rely on build-rustfs - Improve cross-platform compatibility for Windows binary naming (.exe) - Streamline artifact uploading and OSS publishing process - Maintain consistent conditional logic for release operations * fix(ci): add repo-token to setup-protoc action for authentication - Add GITHUB_TOKEN parameter to arduino/setup-protoc@v3 action - Ensure proper authentication for Protoc installation in CI workflow - Maintain consistent setup across different CI environments * modify config * improve readme.md * remove env config relation * add allow(dead_code)
93 lines
3.6 KiB
Rust
93 lines
3.6 KiB
Rust
use crate::{AppConfig, SinkConfig, UnifiedLogEntry};
|
|
use async_trait::async_trait;
|
|
use std::sync::Arc;
|
|
|
|
#[cfg(feature = "file")]
|
|
mod file;
|
|
#[cfg(all(feature = "kafka", target_os = "linux"))]
|
|
mod kafka;
|
|
#[cfg(feature = "webhook")]
|
|
mod webhook;
|
|
|
|
/// Sink Trait definition, asynchronously write logs
|
|
#[async_trait]
|
|
pub trait Sink: Send + Sync {
|
|
async fn write(&self, entry: &UnifiedLogEntry);
|
|
}
|
|
|
|
/// Create a list of Sink instances
|
|
pub async fn create_sinks(config: &AppConfig) -> Vec<Arc<dyn Sink>> {
|
|
let mut sinks: Vec<Arc<dyn Sink>> = Vec::new();
|
|
|
|
for sink_config in &config.sinks {
|
|
match sink_config {
|
|
#[cfg(all(feature = "kafka", target_os = "linux"))]
|
|
SinkConfig::Kafka(kafka_config) => {
|
|
match rdkafka::config::ClientConfig::new()
|
|
.set("bootstrap.servers", &kafka_config.brokers)
|
|
.set("message.timeout.ms", "5000")
|
|
.create()
|
|
{
|
|
Ok(producer) => {
|
|
sinks.push(Arc::new(kafka::KafkaSink::new(
|
|
producer,
|
|
kafka_config.topic.clone(),
|
|
kafka_config.batch_size.unwrap_or(100),
|
|
kafka_config.batch_timeout_ms.unwrap_or(1000),
|
|
)));
|
|
tracing::info!("Kafka sink created for topic: {}", kafka_config.topic);
|
|
}
|
|
Err(e) => {
|
|
tracing::error!("Failed to create Kafka producer: {}", e);
|
|
}
|
|
}
|
|
}
|
|
#[cfg(feature = "webhook")]
|
|
SinkConfig::Webhook(webhook_config) => {
|
|
sinks.push(Arc::new(webhook::WebhookSink::new(
|
|
webhook_config.endpoint.clone(),
|
|
webhook_config.auth_token.clone(),
|
|
webhook_config.max_retries.unwrap_or(3),
|
|
webhook_config.retry_delay_ms.unwrap_or(100),
|
|
)));
|
|
tracing::info!("Webhook sink created for endpoint: {}", webhook_config.endpoint);
|
|
}
|
|
|
|
#[cfg(feature = "file")]
|
|
SinkConfig::File(file_config) => {
|
|
tracing::debug!("FileSink: Using path: {}", file_config.path);
|
|
match file::FileSink::new(
|
|
file_config.path.clone(),
|
|
file_config.buffer_size.unwrap_or(8192),
|
|
file_config.flush_interval_ms.unwrap_or(1000),
|
|
file_config.flush_threshold.unwrap_or(100),
|
|
)
|
|
.await
|
|
{
|
|
Ok(sink) => {
|
|
sinks.push(Arc::new(sink));
|
|
tracing::info!("File sink created for path: {}", file_config.path);
|
|
}
|
|
Err(e) => {
|
|
tracing::error!("Failed to create File sink: {}", e);
|
|
}
|
|
}
|
|
}
|
|
#[cfg(any(not(feature = "kafka"), not(target_os = "linux")))]
|
|
SinkConfig::Kafka(_) => {
|
|
tracing::warn!("Kafka sink is configured but the 'kafka' feature is not enabled");
|
|
}
|
|
#[cfg(not(feature = "webhook"))]
|
|
SinkConfig::Webhook(_) => {
|
|
tracing::warn!("Webhook sink is configured but the 'webhook' feature is not enabled");
|
|
}
|
|
#[cfg(not(feature = "file"))]
|
|
SinkConfig::File(_) => {
|
|
tracing::warn!("File sink is configured but the 'file' feature is not enabled");
|
|
}
|
|
}
|
|
}
|
|
|
|
sinks
|
|
}
|