gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-20 19:42:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feature/rustfs config (#396)

Browse Source 
* 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)
This commit is contained in:
houseme
2025-05-12 01:17:31 +08:00
committed by GitHub
parent 0c351965a2
commit dd7da015e3
64 changed files with 1283 additions and 1008 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/obs/src/config.rs
+100 -44
View File
@@ -1,6 +1,6 @@
use crate::global::{ENVIRONMENT, LOGGER_LEVEL, METER_INTERVAL, SAMPLE_RATIO, SERVICE_NAME, SERVICE_VERSION, USE_STDOUT};
use config::{Config, Environment, File, FileFormat};
use serde::Deserialize;
use config::{Config, File, FileFormat};
use serde::{Deserialize, Serialize};
use std::env;
/// OpenTelemetry Configuration
@@ -11,7 +11,7 @@ use std::env;
/// Add use_stdout for output to stdout
/// Add logger level for log level
/// Add local_logging_enabled for local logging enabled
#[derive(Debug, Deserialize, Clone)]
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct OtelConfig {
pub endpoint: String, // Endpoint for metric collection
pub use_stdout: Option<bool>, // Output to stdout
@@ -24,7 +24,7 @@ pub struct OtelConfig {
pub local_logging_enabled: Option<bool>, // Local logging enabled
}
// Helper function: Extract observable configuration from environment variables
/// Helper function: Extract observable configuration from environment variables
fn extract_otel_config_from_env() -> OtelConfig {
OtelConfig {
endpoint: env::var("RUSTFS_OBSERVABILITY_ENDPOINT").unwrap_or_else(|_| "".to_string()),
@@ -63,36 +63,89 @@ fn extract_otel_config_from_env() -> OtelConfig {
}
}
impl Default for OtelConfig {
fn default() -> Self {
impl OtelConfig {
/// Create a new instance of OtelConfig with default values
///
/// # Returns
/// A new instance of OtelConfig
pub fn new() -> Self {
extract_otel_config_from_env()
}
}
impl Default for OtelConfig {
fn default() -> Self {
Self::new()
}
}
/// Kafka Sink Configuration - Add batch parameters
#[derive(Debug, Deserialize, Clone, Default)]
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct KafkaSinkConfig {
pub enabled: bool,
pub bootstrap_servers: String,
pub brokers: String,
pub topic: String,
pub batch_size: Option<usize>, // Batch size, default 100
pub batch_timeout_ms: Option<u64>, // Batch timeout time, default 1000ms
}
impl KafkaSinkConfig {
pub fn new() -> Self {
Self {
brokers: env::var("RUSTFS__SINKS_0_KAFKA_BROKERS")
.ok()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| "localhost:9092".to_string()),
topic: env::var("RUSTFS__SINKS_0_KAFKA_TOPIC")
.ok()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| "default_topic".to_string()),
batch_size: Some(100),
batch_timeout_ms: Some(1000),
}
}
}
impl Default for KafkaSinkConfig {
fn default() -> Self {
Self::new()
}
}
/// Webhook Sink Configuration - Add Retry Parameters
#[derive(Debug, Deserialize, Clone, Default)]
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct WebhookSinkConfig {
pub enabled: bool,
pub endpoint: String,
pub auth_token: String,
pub max_retries: Option<usize>, // Maximum number of retry times, default 3
pub retry_delay_ms: Option<u64>, // Retry the delay cardinality, default 100ms
}
impl WebhookSinkConfig {
pub fn new() -> Self {
Self {
endpoint: env::var("RUSTFS__SINKS_0_WEBHOOK_ENDPOINT")
.ok()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| "http://localhost:8080".to_string()),
auth_token: env::var("RUSTFS__SINKS_0_WEBHOOK_AUTH_TOKEN")
.ok()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| "default_token".to_string()),
max_retries: Some(3),
retry_delay_ms: Some(100),
}
}
}
impl Default for WebhookSinkConfig {
fn default() -> Self {
Self::new()
}
}
/// File Sink Configuration - Add buffering parameters
#[derive(Debug, Deserialize, Clone)]
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct FileSinkConfig {
pub enabled: bool,
pub path: String,
pub buffer_size: Option<usize>, // Write buffer size, default 8192
pub flush_interval_ms: Option<u64>, // Refresh interval time, default 1000ms
@@ -114,13 +167,9 @@ impl FileSinkConfig {
.unwrap_or("rustfs/rustfs.log")
.to_string()
}
}
impl Default for FileSinkConfig {
fn default() -> Self {
FileSinkConfig {
enabled: true,
path: env::var("RUSTFS_SINKS_FILE_PATH")
pub fn new() -> Self {
Self {
path: env::var("RUSTFS__SINKS_0_FILE_PATH")
.ok()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(Self::get_default_log_path),
@@ -131,38 +180,53 @@ impl Default for FileSinkConfig {
}
}
impl Default for FileSinkConfig {
fn default() -> Self {
Self::new()
}
}
/// Sink configuration collection
#[derive(Debug, Deserialize, Clone)]
pub struct SinkConfig {
pub kafka: Option<KafkaSinkConfig>,
pub webhook: Option<WebhookSinkConfig>,
pub file: Option<FileSinkConfig>,
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum SinkConfig {
File(FileSinkConfig),
Kafka(KafkaSinkConfig),
Webhook(WebhookSinkConfig),
}
impl SinkConfig {
pub fn new() -> Self {
Self::File(FileSinkConfig::new())
}
}
impl Default for SinkConfig {
fn default() -> Self {
SinkConfig {
kafka: None,
webhook: None,
file: Some(FileSinkConfig::default()),
}
Self::new()
}
}
///Logger Configuration
#[derive(Debug, Deserialize, Clone)]
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct LoggerConfig {
pub queue_capacity: Option<usize>,
}
impl Default for LoggerConfig {
fn default() -> Self {
LoggerConfig {
impl LoggerConfig {
pub fn new() -> Self {
Self {
queue_capacity: Some(10000),
}
}
}
impl Default for LoggerConfig {
fn default() -> Self {
Self::new()
}
}
/// Overall application configuration
/// Add observability, sinks, and logger configuration
///
@@ -180,7 +244,7 @@ impl Default for LoggerConfig {
#[derive(Debug, Deserialize, Clone)]
pub struct AppConfig {
pub observability: OtelConfig,
pub sinks: SinkConfig,
pub sinks: Vec<SinkConfig>,
pub logger: Option<LoggerConfig>,
}
@@ -192,7 +256,7 @@ impl AppConfig {
pub fn new() -> Self {
Self {
observability: OtelConfig::default(),
sinks: SinkConfig::default(),
sinks: vec![SinkConfig::default()],
logger: Some(LoggerConfig::default()),
}
}
@@ -258,14 +322,6 @@ pub fn load_config(config_dir: Option<String>) -> AppConfig {
let app_config = Config::builder()
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false))
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false))
.add_source(
Environment::default()
.prefix("RUSTFS")
.prefix_separator("__")
.separator("__")
.with_list_parse_key("volumes")
.try_parsing(true),
)
.build()
.unwrap_or_default();
crates/obs/src/lib.rs
+2 -8
View File
@@ -32,7 +32,7 @@ mod config;
mod entry;
mod global;
mod logger;
mod sink;
mod sinks;
mod system;
mod telemetry;
mod utils;
@@ -40,12 +40,6 @@ mod worker;
use crate::logger::InitLogStatus;
pub use config::load_config;
#[cfg(feature = "file")]
pub use config::FileSinkConfig;
#[cfg(feature = "kafka")]
pub use config::KafkaSinkConfig;
#[cfg(feature = "webhook")]
pub use config::WebhookSinkConfig;
pub use config::{AppConfig, LoggerConfig, OtelConfig, SinkConfig};
pub use entry::args::Args;
pub use entry::audit::{ApiDetails, AuditLogEntry};
@@ -79,7 +73,7 @@ use tracing::{error, info};
/// ```
pub async fn init_obs(config: AppConfig) -> (Arc<Mutex<Logger>>, telemetry::OtelGuard) {
let guard = init_telemetry(&config.observability);
let sinks = sink::create_sinks(&config).await;
let sinks = sinks::create_sinks(&config).await;
let logger = init_global_logger(&config, sinks).await;
let obs_config = config.observability.clone();
tokio::spawn(async move {
crates/obs/src/logger.rs
+1 -1
View File
@@ -1,5 +1,5 @@
use crate::global::{ENVIRONMENT, SERVICE_NAME, SERVICE_VERSION};
use crate::sink::Sink;
use crate::sinks::Sink;
use crate::{AppConfig, AuditLogEntry, BaseLogEntry, ConsoleLogEntry, GlobalError, OtelConfig, ServerLogEntry, UnifiedLogEntry};
use std::sync::Arc;
use std::time::SystemTime;
crates/obs/src/sink.rs
-497
View File
@@ -1,497 +0,0 @@
use crate::{AppConfig, LogRecord, UnifiedLogEntry};
use async_trait::async_trait;
use std::sync::Arc;
use tokio::fs::OpenOptions;
use tokio::io;
use tokio::io::AsyncWriteExt;
/// Sink Trait definition, asynchronously write logs
#[async_trait]
pub trait Sink: Send + Sync {
async fn write(&self, entry: &UnifiedLogEntry);
}
#[cfg(feature = "kafka")]
/// Kafka Sink Implementation
pub struct KafkaSink {
producer: rdkafka::producer::FutureProducer,
topic: String,
batch_size: usize,
batch_timeout_ms: u64,
entries: Arc<tokio::sync::Mutex<Vec<UnifiedLogEntry>>>,
last_flush: Arc<std::sync::atomic::AtomicU64>,
}
#[cfg(feature = "kafka")]
impl KafkaSink {
/// Create a new KafkaSink instance
pub fn new(producer: rdkafka::producer::FutureProducer, topic: String, batch_size: usize, batch_timeout_ms: u64) -> Self {
// Create Arc-wrapped values first
let entries = Arc::new(tokio::sync::Mutex::new(Vec::with_capacity(batch_size)));
let last_flush = Arc::new(std::sync::atomic::AtomicU64::new(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64,
));
let sink = KafkaSink {
producer: producer.clone(),
topic: topic.clone(),
batch_size,
batch_timeout_ms,
entries: entries.clone(),
last_flush: last_flush.clone(),
};
// Start background flusher
tokio::spawn(Self::periodic_flush(producer, topic, entries, last_flush, batch_timeout_ms));
sink
}
/// Add a getter method to read the batch_timeout_ms field
#[allow(dead_code)]
pub fn batch_timeout(&self) -> u64 {
self.batch_timeout_ms
}
/// Add a method to dynamically adjust the timeout if needed
#[allow(dead_code)]
pub fn set_batch_timeout(&mut self, new_timeout_ms: u64) {
self.batch_timeout_ms = new_timeout_ms;
}
async fn periodic_flush(
producer: rdkafka::producer::FutureProducer,
topic: String,
entries: Arc<tokio::sync::Mutex<Vec<UnifiedLogEntry>>>,
last_flush: Arc<std::sync::atomic::AtomicU64>,
timeout_ms: u64,
) {
loop {
tokio::time::sleep(tokio::time::Duration::from_millis(timeout_ms / 2)).await;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let last = last_flush.load(std::sync::atomic::Ordering::Relaxed);
if now - last >= timeout_ms {
let mut batch = entries.lock().await;
if !batch.is_empty() {
Self::send_batch(&producer, &topic, batch.drain(..).collect()).await;
last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
}
}
}
}
async fn send_batch(producer: &rdkafka::producer::FutureProducer, topic: &str, entries: Vec<UnifiedLogEntry>) {
for entry in entries {
let payload = match serde_json::to_string(&entry) {
Ok(p) => p,
Err(e) => {
eprintln!("Failed to serialize log entry: {}", e);
continue;
}
};
let span_id = entry.get_timestamp().to_rfc3339();
let _ = producer
.send(
rdkafka::producer::FutureRecord::to(topic).payload(&payload).key(&span_id),
std::time::Duration::from_secs(5),
)
.await;
}
}
}
#[cfg(feature = "kafka")]
#[async_trait]
impl Sink for KafkaSink {
async fn write(&self, entry: &UnifiedLogEntry) {
let mut batch = self.entries.lock().await;
batch.push(entry.clone());
let should_flush_by_size = batch.len() >= self.batch_size;
let should_flush_by_time = {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
now - last >= self.batch_timeout_ms
};
if should_flush_by_size || should_flush_by_time {
// Existing flush logic
let entries_to_send: Vec<UnifiedLogEntry> = batch.drain(..).collect();
let producer = self.producer.clone();
let topic = self.topic.clone();
self.last_flush.store(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64,
std::sync::atomic::Ordering::Relaxed,
);
tokio::spawn(async move {
KafkaSink::send_batch(&producer, &topic, entries_to_send).await;
});
}
}
}
#[cfg(feature = "kafka")]
impl Drop for KafkaSink {
fn drop(&mut self) {
// Perform any necessary cleanup here
// For example, you might want to flush any remaining entries
let producer = self.producer.clone();
let topic = self.topic.clone();
let entries = self.entries.clone();
let last_flush = self.last_flush.clone();
tokio::spawn(async move {
let mut batch = entries.lock().await;
if !batch.is_empty() {
KafkaSink::send_batch(&producer, &topic, batch.drain(..).collect()).await;
last_flush.store(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64,
std::sync::atomic::Ordering::Relaxed,
);
}
});
eprintln!("Dropping KafkaSink with topic: {}", self.topic);
}
}
#[cfg(feature = "webhook")]
/// Webhook Sink Implementation
pub struct WebhookSink {
endpoint: String,
auth_token: String,
client: reqwest::Client,
max_retries: usize,
retry_delay_ms: u64,
}
#[cfg(feature = "webhook")]
impl WebhookSink {
pub fn new(endpoint: String, auth_token: String, max_retries: usize, retry_delay_ms: u64) -> Self {
WebhookSink {
endpoint,
auth_token,
client: reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(10))
.build()
.unwrap_or_else(|_| reqwest::Client::new()),
max_retries,
retry_delay_ms,
}
}
}
#[cfg(feature = "webhook")]
#[async_trait]
impl Sink for WebhookSink {
async fn write(&self, entry: &UnifiedLogEntry) {
let mut retries = 0;
let url = self.endpoint.clone();
let entry_clone = entry.clone();
let auth_value = reqwest::header::HeaderValue::from_str(format!("Bearer {}", self.auth_token.clone()).as_str()).unwrap();
while retries < self.max_retries {
match self
.client
.post(&url)
.header(reqwest::header::AUTHORIZATION, auth_value.clone())
.json(&entry_clone)
.send()
.await
{
Ok(response) if response.status().is_success() => {
return;
}
_ => {
retries += 1;
if retries < self.max_retries {
tokio::time::sleep(tokio::time::Duration::from_millis(
self.retry_delay_ms * (1 << retries), // Exponential backoff
))
.await;
}
}
}
}
eprintln!("Failed to send log to webhook after {} retries", self.max_retries);
}
}
#[cfg(feature = "webhook")]
impl Drop for WebhookSink {
fn drop(&mut self) {
// Perform any necessary cleanup here
// For example, you might want to log that the sink is being dropped
eprintln!("Dropping WebhookSink with URL: {}", self.endpoint);
}
}
#[cfg(feature = "file")]
/// File Sink Implementation
pub struct FileSink {
path: String,
buffer_size: usize,
writer: Arc<tokio::sync::Mutex<io::BufWriter<tokio::fs::File>>>,
entry_count: std::sync::atomic::AtomicUsize,
last_flush: std::sync::atomic::AtomicU64,
flush_interval_ms: u64, // Time between flushes
flush_threshold: usize, // Number of entries before flush
}
#[cfg(feature = "file")]
impl FileSink {
/// Create a new FileSink instance
pub async fn new(
path: String,
buffer_size: usize,
flush_interval_ms: u64,
flush_threshold: usize,
) -> Result<Self, io::Error> {
// check if the file exists
let file_exists = tokio::fs::metadata(&path).await.is_ok();
// if the file not exists, create it
if !file_exists {
tokio::fs::create_dir_all(std::path::Path::new(&path).parent().unwrap()).await?;
tracing::debug!("File does not exist, creating it. Path: {:?}", path)
}
let file = if file_exists {
// If the file exists, open it in append mode
tracing::debug!("FileSink: File exists, opening in append mode.");
OpenOptions::new().append(true).create(true).open(&path).await?
} else {
// If the file does not exist, create it
tracing::debug!("FileSink: File does not exist, creating a new file.");
// Create the file and write a header or initial content if needed
OpenOptions::new().create(true).truncate(true).write(true).open(&path).await?
};
let writer = io::BufWriter::with_capacity(buffer_size, file);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
Ok(FileSink {
path,
buffer_size,
writer: Arc::new(tokio::sync::Mutex::new(writer)),
entry_count: std::sync::atomic::AtomicUsize::new(0),
last_flush: std::sync::atomic::AtomicU64::new(now),
flush_interval_ms,
flush_threshold,
})
}
#[allow(dead_code)]
async fn initialize_writer(&mut self) -> io::Result<()> {
let file = tokio::fs::File::create(&self.path).await?;
// Use buffer_size to create a buffer writer with a specified capacity
let buf_writer = io::BufWriter::with_capacity(self.buffer_size, file);
// Replace the original writer with the new Mutex
self.writer = Arc::new(tokio::sync::Mutex::new(buf_writer));
Ok(())
}
// Get the current buffer size
#[allow(dead_code)]
pub fn buffer_size(&self) -> usize {
self.buffer_size
}
// How to dynamically adjust the buffer size
#[allow(dead_code)]
pub async fn set_buffer_size(&mut self, new_size: usize) -> io::Result<()> {
if self.buffer_size != new_size {
self.buffer_size = new_size;
// Reinitialize the writer directly, without checking is_some()
self.initialize_writer().await?;
}
Ok(())
}
// Check if flushing is needed based on count or time
fn should_flush(&self) -> bool {
// Check entry count threshold
if self.entry_count.load(std::sync::atomic::Ordering::Relaxed) >= self.flush_threshold {
return true;
}
// Check time threshold
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
now - last >= self.flush_interval_ms
}
}
#[cfg(feature = "file")]
#[async_trait]
impl Sink for FileSink {
async fn write(&self, entry: &UnifiedLogEntry) {
let line = format!("{:?}\n", entry);
let mut writer = self.writer.lock().await;
if let Err(e) = writer.write_all(line.as_bytes()).await {
eprintln!(
"Failed to write log to file {}: {},entry timestamp:{:?}",
self.path,
e,
entry.get_timestamp()
);
return;
}
// Only flush periodically to improve performance
// Logic to determine when to flush could be added here
// Increment the entry count
self.entry_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
// Check if we should flush
if self.should_flush() {
if let Err(e) = writer.flush().await {
eprintln!("Failed to flush log file {}: {}", self.path, e);
return;
}
// Reset counters
self.entry_count.store(0, std::sync::atomic::Ordering::Relaxed);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
self.last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
}
}
}
#[cfg(feature = "file")]
impl Drop for FileSink {
fn drop(&mut self) {
let writer = self.writer.clone();
let path = self.path.clone();
tokio::task::spawn_blocking(move || {
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut writer = writer.lock().await;
if let Err(e) = writer.flush().await {
eprintln!("Failed to flush log file {}: {}", path, e);
}
});
});
}
}
/// 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();
#[cfg(feature = "kafka")]
{
match &config.sinks.kafka {
Some(sink_kafka) => {
if sink_kafka.enabled {
match rdkafka::config::ClientConfig::new()
.set("bootstrap.servers", &sink_kafka.bootstrap_servers)
.set("message.timeout.ms", "5000")
.create()
{
Ok(producer) => {
sinks.push(Arc::new(KafkaSink::new(
producer,
sink_kafka.topic.clone(),
sink_kafka.batch_size.unwrap_or(100),
sink_kafka.batch_timeout_ms.unwrap_or(1000),
)));
}
Err(e) => {
tracing::error!("Failed to create Kafka producer: {}", e);
}
}
} else {
tracing::info!("Kafka sink is disabled in the configuration");
}
}
_ => {
tracing::info!("Kafka sink is not configured or disabled");
}
}
}
#[cfg(feature = "webhook")]
{
match &config.sinks.webhook {
Some(sink_webhook) => {
if sink_webhook.enabled {
sinks.push(Arc::new(WebhookSink::new(
sink_webhook.endpoint.clone(),
sink_webhook.auth_token.clone(),
sink_webhook.max_retries.unwrap_or(3),
sink_webhook.retry_delay_ms.unwrap_or(100),
)));
} else {
tracing::info!("Webhook sink is disabled in the configuration");
}
}
_ => {
tracing::info!("Webhook sink is not configured or disabled");
}
}
}
#[cfg(feature = "file")]
{
// let config = config.clone();
match &config.sinks.file {
Some(sink_file) => {
tracing::info!("File sink is enabled in the configuration");
let path = if sink_file.enabled {
sink_file.path.clone()
} else {
"rustfs.log".to_string()
};
tracing::debug!("FileSink: Using path: {}", path);
sinks.push(Arc::new(
FileSink::new(
path.clone(),
sink_file.buffer_size.unwrap_or(8192),
sink_file.flush_interval_ms.unwrap_or(1000),
sink_file.flush_threshold.unwrap_or(100),
)
.await
.unwrap(),
));
}
_ => {
tracing::info!("File sink is not configured or disabled");
}
}
}
sinks
}
crates/obs/src/sinks/file.rs
+164
View File
@@ -0,0 +1,164 @@
use crate::sinks::Sink;
use crate::{LogRecord, UnifiedLogEntry};
use async_trait::async_trait;
use std::sync::Arc;
use tokio::fs::OpenOptions;
use tokio::io;
use tokio::io::AsyncWriteExt;
/// File Sink Implementation
pub struct FileSink {
path: String,
buffer_size: usize,
writer: Arc<tokio::sync::Mutex<io::BufWriter<tokio::fs::File>>>,
entry_count: std::sync::atomic::AtomicUsize,
last_flush: std::sync::atomic::AtomicU64,
flush_interval_ms: u64, // Time between flushes
flush_threshold: usize, // Number of entries before flush
}
impl FileSink {
/// Create a new FileSink instance
pub async fn new(
path: String,
buffer_size: usize,
flush_interval_ms: u64,
flush_threshold: usize,
) -> Result<Self, io::Error> {
// check if the file exists
let file_exists = tokio::fs::metadata(&path).await.is_ok();
// if the file not exists, create it
if !file_exists {
tokio::fs::create_dir_all(std::path::Path::new(&path).parent().unwrap()).await?;
tracing::debug!("File does not exist, creating it. Path: {:?}", path)
}
let file = if file_exists {
// If the file exists, open it in append mode
tracing::debug!("FileSink: File exists, opening in append mode.");
OpenOptions::new().append(true).create(true).open(&path).await?
} else {
// If the file does not exist, create it
tracing::debug!("FileSink: File does not exist, creating a new file.");
// Create the file and write a header or initial content if needed
OpenOptions::new().create(true).truncate(true).write(true).open(&path).await?
};
let writer = io::BufWriter::with_capacity(buffer_size, file);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
Ok(FileSink {
path,
buffer_size,
writer: Arc::new(tokio::sync::Mutex::new(writer)),
entry_count: std::sync::atomic::AtomicUsize::new(0),
last_flush: std::sync::atomic::AtomicU64::new(now),
flush_interval_ms,
flush_threshold,
})
}
#[allow(dead_code)]
async fn initialize_writer(&mut self) -> io::Result<()> {
let file = tokio::fs::File::create(&self.path).await?;
// Use buffer_size to create a buffer writer with a specified capacity
let buf_writer = io::BufWriter::with_capacity(self.buffer_size, file);
// Replace the original writer with the new Mutex
self.writer = Arc::new(tokio::sync::Mutex::new(buf_writer));
Ok(())
}
// Get the current buffer size
#[allow(dead_code)]
pub fn buffer_size(&self) -> usize {
self.buffer_size
}
// How to dynamically adjust the buffer size
#[allow(dead_code)]
pub async fn set_buffer_size(&mut self, new_size: usize) -> io::Result<()> {
if self.buffer_size != new_size {
self.buffer_size = new_size;
// Reinitialize the writer directly, without checking is_some()
self.initialize_writer().await?;
}
Ok(())
}
// Check if flushing is needed based on count or time
fn should_flush(&self) -> bool {
// Check entry count threshold
if self.entry_count.load(std::sync::atomic::Ordering::Relaxed) >= self.flush_threshold {
return true;
}
// Check time threshold
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
now - last >= self.flush_interval_ms
}
}
#[async_trait]
impl Sink for FileSink {
async fn write(&self, entry: &UnifiedLogEntry) {
let line = format!("{:?}\n", entry);
let mut writer = self.writer.lock().await;
if let Err(e) = writer.write_all(line.as_bytes()).await {
eprintln!(
"Failed to write log to file {}: {},entry timestamp:{:?}",
self.path,
e,
entry.get_timestamp()
);
return;
}
// Only flush periodically to improve performance
// Logic to determine when to flush could be added here
// Increment the entry count
self.entry_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
// Check if we should flush
if self.should_flush() {
if let Err(e) = writer.flush().await {
eprintln!("Failed to flush log file {}: {}", self.path, e);
return;
}
// Reset counters
self.entry_count.store(0, std::sync::atomic::Ordering::Relaxed);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
self.last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
}
}
}
impl Drop for FileSink {
fn drop(&mut self) {
let writer = self.writer.clone();
let path = self.path.clone();
tokio::task::spawn_blocking(move || {
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut writer = writer.lock().await;
if let Err(e) = writer.flush().await {
eprintln!("Failed to flush log file {}: {}", path, e);
}
});
});
}
}
crates/obs/src/sinks/kafka.rs
+165
View File
@@ -0,0 +1,165 @@
use crate::sinks::Sink;
use crate::{LogRecord, UnifiedLogEntry};
use async_trait::async_trait;
use std::sync::Arc;
/// Kafka Sink Implementation
pub struct KafkaSink {
producer: rdkafka::producer::FutureProducer,
topic: String,
batch_size: usize,
batch_timeout_ms: u64,
entries: Arc<tokio::sync::Mutex<Vec<UnifiedLogEntry>>>,
last_flush: Arc<std::sync::atomic::AtomicU64>,
}
impl KafkaSink {
/// Create a new KafkaSink instance
pub fn new(producer: rdkafka::producer::FutureProducer, topic: String, batch_size: usize, batch_timeout_ms: u64) -> Self {
// Create Arc-wrapped values first
let entries = Arc::new(tokio::sync::Mutex::new(Vec::with_capacity(batch_size)));
let last_flush = Arc::new(std::sync::atomic::AtomicU64::new(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64,
));
let sink = KafkaSink {
producer: producer.clone(),
topic: topic.clone(),
batch_size,
batch_timeout_ms,
entries: entries.clone(),
last_flush: last_flush.clone(),
};
// Start background flusher
tokio::spawn(Self::periodic_flush(producer, topic, entries, last_flush, batch_timeout_ms));
sink
}
/// Add a getter method to read the batch_timeout_ms field
#[allow(dead_code)]
pub fn batch_timeout(&self) -> u64 {
self.batch_timeout_ms
}
/// Add a method to dynamically adjust the timeout if needed
#[allow(dead_code)]
pub fn set_batch_timeout(&mut self, new_timeout_ms: u64) {
self.batch_timeout_ms = new_timeout_ms;
}
async fn periodic_flush(
producer: rdkafka::producer::FutureProducer,
topic: String,
entries: Arc<tokio::sync::Mutex<Vec<UnifiedLogEntry>>>,
last_flush: Arc<std::sync::atomic::AtomicU64>,
timeout_ms: u64,
) {
loop {
tokio::time::sleep(tokio::time::Duration::from_millis(timeout_ms / 2)).await;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let last = last_flush.load(std::sync::atomic::Ordering::Relaxed);
if now - last >= timeout_ms {
let mut batch = entries.lock().await;
if !batch.is_empty() {
Self::send_batch(&producer, &topic, batch.drain(..).collect()).await;
last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
}
}
}
}
async fn send_batch(producer: &rdkafka::producer::FutureProducer, topic: &str, entries: Vec<UnifiedLogEntry>) {
for entry in entries {
let payload = match serde_json::to_string(&entry) {
Ok(p) => p,
Err(e) => {
eprintln!("Failed to serialize log entry: {}", e);
continue;
}
};
let span_id = entry.get_timestamp().to_rfc3339();
let _ = producer
.send(
rdkafka::producer::FutureRecord::to(topic).payload(&payload).key(&span_id),
std::time::Duration::from_secs(5),
)
.await;
}
}
}
#[async_trait]
impl Sink for KafkaSink {
async fn write(&self, entry: &UnifiedLogEntry) {
let mut batch = self.entries.lock().await;
batch.push(entry.clone());
let should_flush_by_size = batch.len() >= self.batch_size;
let should_flush_by_time = {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
now - last >= self.batch_timeout_ms
};
if should_flush_by_size || should_flush_by_time {
// Existing flush logic
let entries_to_send: Vec<UnifiedLogEntry> = batch.drain(..).collect();
let producer = self.producer.clone();
let topic = self.topic.clone();
self.last_flush.store(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64,
std::sync::atomic::Ordering::Relaxed,
);
tokio::spawn(async move {
KafkaSink::send_batch(&producer, &topic, entries_to_send).await;
});
}
}
}
impl Drop for KafkaSink {
fn drop(&mut self) {
// Perform any necessary cleanup here
// For example, you might want to flush any remaining entries
let producer = self.producer.clone();
let topic = self.topic.clone();
let entries = self.entries.clone();
let last_flush = self.last_flush.clone();
tokio::spawn(async move {
let mut batch = entries.lock().await;
if !batch.is_empty() {
KafkaSink::send_batch(&producer, &topic, batch.drain(..).collect()).await;
last_flush.store(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64,
std::sync::atomic::Ordering::Relaxed,
);
}
});
eprintln!("Dropping KafkaSink with topic: {}", self.topic);
}
}
crates/obs/src/sinks/mod.rs
+92
View File
@@ -0,0 +1,92 @@
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
}
crates/obs/src/sinks/webhook.rs
+70
View File
@@ -0,0 +1,70 @@
use crate::sinks::Sink;
use crate::UnifiedLogEntry;
use async_trait::async_trait;
/// Webhook Sink Implementation
pub struct WebhookSink {
endpoint: String,
auth_token: String,
client: reqwest::Client,
max_retries: usize,
retry_delay_ms: u64,
}
impl WebhookSink {
pub fn new(endpoint: String, auth_token: String, max_retries: usize, retry_delay_ms: u64) -> Self {
WebhookSink {
endpoint,
auth_token,
client: reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(10))
.build()
.unwrap_or_else(|_| reqwest::Client::new()),
max_retries,
retry_delay_ms,
}
}
}
#[async_trait]
impl Sink for WebhookSink {
async fn write(&self, entry: &UnifiedLogEntry) {
let mut retries = 0;
let url = self.endpoint.clone();
let entry_clone = entry.clone();
let auth_value = reqwest::header::HeaderValue::from_str(format!("Bearer {}", self.auth_token.clone()).as_str()).unwrap();
while retries < self.max_retries {
match self
.client
.post(&url)
.header(reqwest::header::AUTHORIZATION, auth_value.clone())
.json(&entry_clone)
.send()
.await
{
Ok(response) if response.status().is_success() => {
return;
}
_ => {
retries += 1;
if retries < self.max_retries {
tokio::time::sleep(tokio::time::Duration::from_millis(
self.retry_delay_ms * (1 << retries), // Exponential backoff
))
.await;
}
}
}
}
eprintln!("Failed to send log to webhook after {} retries", self.max_retries);
}
}
impl Drop for WebhookSink {
fn drop(&mut self) {
// Perform any necessary cleanup here
// For example, you might want to log that the sink is being dropped
eprintln!("Dropping WebhookSink with URL: {}", self.endpoint);
}
}
crates/obs/src/worker.rs
+1 -1
View File
@@ -1,4 +1,4 @@
use crate::{sink::Sink, UnifiedLogEntry};
use crate::{sinks::Sink, UnifiedLogEntry};
use std::sync::Arc;
use tokio::sync::mpsc::Receiver;