gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-28 07:57:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'feature/observability'

Browse Source 
* feature/observability: (27 commits)
  modify default value
  TryInto cover
  upgrade reqwest version from 0.12.12 to 0.12.15
  improve code for FileSink
  improve code for config and FileSink
  webhook add auth_token
  upgrade docker images
  feat(obs): enhance OpenTelemetry configuration and logging
  merge main
  feat: add metrics_handler
  add prometheus
  upgrade opentelemetry create from 0.28.0 to 0.29.0
  update .gitignore
  replace log to tracing
  improve code for observability
  improve code for main
  feat(observability): add obs_config option and document stdout export
  improve logger entry for Observability
  improve log struct
  improve code
  ...
This commit is contained in:
houseme
2025-04-02 16:19:38 +08:00
parent 8d4c3dfa0e c57ee8e471
commit b2194758dc
36 changed files with 4581 additions and 614 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.docker/observability/README.md
+101
View File
@@ -0,0 +1,101 @@
# Observability
This directory contains the observability stack for the application. The stack is composed of the following components:
- Prometheus v3.2.1
- Grafana 11.6.0
- Loki 3.4.2
- Jaeger 2.4.0
- Otel Collector 0.120.0 #0.121.0 remove loki
## Prometheus
Prometheus is a monitoring and alerting toolkit. It scrapes metrics from instrumented jobs, either directly or via an
intermediary push gateway for short-lived jobs. It stores all scraped samples locally and runs rules over this data to
either aggregate and record new time series from existing data or generate alerts. Grafana or other API consumers can be
used to visualize the collected data.
## Grafana
Grafana is a multi-platform open-source analytics and interactive visualization web application. It provides charts,
graphs, and alerts for the web when connected to supported data sources.
## Loki
Loki is a horizontally-scalable, highly-available, multi-tenant log aggregation system inspired by Prometheus. It is
designed to be very cost-effective and easy to operate. It does not index the contents of the logs, but rather a set of
labels for each log stream.
## Jaeger
Jaeger is a distributed tracing system released as open source by Uber Technologies. It is used for monitoring and
troubleshooting microservices-based distributed systems, including:
- Distributed context propagation
- Distributed transaction monitoring
- Root cause analysis
- Service dependency analysis
- Performance / latency optimization
## Otel Collector
The OpenTelemetry Collector offers a vendor-agnostic implementation on how to receive, process, and export telemetry
data. It removes the need to run, operate, and maintain multiple agents/collectors in order to support open-source
observability data formats (e.g. Jaeger, Prometheus, etc.) sending to one or more open-source or commercial back-ends.
## How to use
To deploy the observability stack, run the following command:
```bash
docker-compose -f docker-compose.yml -f docker-compose.override.yml up -d
```
To access the Grafana dashboard, navigate to `http://localhost:3000` in your browser. The default username and password
are `admin` and `admin`, respectively.
To access the Jaeger dashboard, navigate to `http://localhost:16686` in your browser.
To access the Prometheus dashboard, navigate to `http://localhost:9090` in your browser.
## How to stop
To stop the observability stack, run the following command:
```bash
docker-compose -f docker-compose.yml -f docker-compose.override.yml down
```
## How to remove data
To remove the data generated by the observability stack, run the following command:
```bash
docker-compose -f docker-compose.yml -f docker-compose.override.yml down -v
```
## How to configure
To configure the observability stack, modify the `docker-compose.override.yml` file. The file contains the following
```yaml
services:
prometheus:
environment:
- PROMETHEUS_CONFIG_FILE=/etc/prometheus/prometheus.yml
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
grafana:
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- ./grafana/provisioning:/etc/grafana/provisioning
```
The `prometheus` service mounts the `prometheus.yml` file to `/etc/prometheus/prometheus.yml`. The `grafana` service
mounts the `grafana/provisioning` directory to `/etc/grafana/provisioning`. You can modify these files to configure the
observability stack.
.docker/observability/docker-compose.yml
+65
View File
@@ -0,0 +1,65 @@
services:
otel-collector:
image: otel/opentelemetry-collector-contrib:0.120.0
environment:
- TZ=Asia/Shanghai
volumes:
- ./otel-collector-config.yaml:/etc/otelcol-contrib/config.yaml
ports:
- 1888:1888
- 8888:8888
- 8889:8889
- 13133:13133
- 4317:4317
- 4318:4318
- 55679:55679
networks:
- otel-network
jaeger:
image: jaegertracing/jaeger:2.4.0
environment:
- TZ=Asia/Shanghai
ports:
- "16686:16686"
- "14317:4317"
- "14318:4318"
networks:
- otel-network
prometheus:
image: prom/prometheus:v3.2.1
environment:
- TZ=Asia/Shanghai
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
ports:
- "9090:9090"
networks:
- otel-network
loki:
image: grafana/loki:3.4.2
environment:
- TZ=Asia/Shanghai
volumes:
- ./loki-config.yaml:/etc/loki/local-config.yaml
ports:
- "3100:3100"
command: -config.file=/etc/loki/local-config.yaml
networks:
- otel-network
grafana:
image: grafana/grafana:11.6.0
ports:
- "3000:3000" # Web UI
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
- TZ=Asia/Shanghai
networks:
- otel-network
networks:
otel-network:
driver: bridge
name: "network_otel_config"
driver_opts:
com.docker.network.enable_ipv6: "true"
.docker/observability/jaeger-config.yaml
+87
View File
@@ -0,0 +1,87 @@
service:
extensions: [ jaeger_storage, jaeger_query, remote_sampling, healthcheckv2 ]
pipelines:
traces:
receivers: [ otlp, jaeger, zipkin ]
processors: [ batch, adaptive_sampling ]
exporters: [ jaeger_storage_exporter ]
telemetry:
resource:
service.name: jaeger
metrics:
level: detailed
readers:
- pull:
exporter:
prometheus:
host: 0.0.0.0
port: 8888
logs:
level: debug
# TODO Initialize telemetry tracer once OTEL released new feature.
# https://github.com/open-telemetry/opentelemetry-collector/issues/10663
extensions:
healthcheckv2:
use_v2: true
http:
# pprof:
# endpoint: 0.0.0.0:1777
# zpages:
# endpoint: 0.0.0.0:55679
jaeger_query:
storage:
traces: some_store
traces_archive: another_store
ui:
config_file: ./cmd/jaeger/config-ui.json
# The maximum duration that is considered for clock skew adjustments.
# Defaults to 0 seconds, which means it's disabled.
max_clock_skew_adjust: 0s
jaeger_storage:
backends:
some_store:
memory:
max_traces: 100000
another_store:
memory:
max_traces: 100000
remote_sampling:
# You can either use file or adaptive sampling strategy in remote_sampling
# file:
# path: ./cmd/jaeger/sampling-strategies.json
adaptive:
sampling_store: some_store
initial_sampling_probability: 0.1
http:
grpc:
receivers:
otlp:
protocols:
grpc:
http:
jaeger:
protocols:
grpc:
thrift_binary:
thrift_compact:
thrift_http:
zipkin:
processors:
batch:
# Adaptive Sampling Processor is required to support adaptive sampling.
# It expects remote_sampling extension with `adaptive:` config to be enabled.
adaptive_sampling:
exporters:
jaeger_storage_exporter:
trace_storage: some_store
.docker/observability/loki-config.yaml
+63
View File
@@ -0,0 +1,63 @@
auth_enabled: false
server:
http_listen_port: 3100
grpc_listen_port: 9096
log_level: debug
grpc_server_max_concurrent_streams: 1000
common:
instance_addr: 127.0.0.1
path_prefix: /tmp/loki
storage:
filesystem:
chunks_directory: /tmp/loki/chunks
rules_directory: /tmp/loki/rules
replication_factor: 1
ring:
kvstore:
store: inmemory
query_range:
results_cache:
cache:
embedded_cache:
enabled: true
max_size_mb: 100
limits_config:
metric_aggregation_enabled: true
schema_config:
configs:
- from: 2020-10-24
store: tsdb
object_store: filesystem
schema: v13
index:
prefix: index_
period: 24h
pattern_ingester:
enabled: true
metric_aggregation:
loki_address: localhost:3100
ruler:
alertmanager_url: http://localhost:9093
frontend:
encoding: protobuf
# By default, Loki will send anonymous, but uniquely-identifiable usage and configuration
# analytics to Grafana Labs. These statistics are sent to https://stats.grafana.org/
#
# Statistics help us better understand how Loki is used, and they show us performance
# levels for most users. This helps us prioritize features and documentation.
# For more information on what's sent, look at
# https://github.com/grafana/loki/blob/main/pkg/analytics/stats.go
# Refer to the buildReport method to see what goes into a report.
#
# If you would like to disable reporting, uncomment the following lines:
#analytics:
# reporting_enabled: false
.docker/observability/otel-collector-config.yaml
+57
View File
@@ -0,0 +1,57 @@
receivers:
otlp:
protocols:
grpc: # OTLP gRPC 接收器
endpoint: 0.0.0.0:4317
http: # OTLP HTTP 接收器
endpoint: 0.0.0.0:4318
processors:
batch: # 批处理处理器，提升吞吐量
timeout: 5s
send_batch_size: 1000
memory_limiter:
check_interval: 1s
limit_mib: 512
exporters:
otlp/traces: # OTLP 导出器，用于跟踪数据
endpoint: "jaeger:4317" # Jaeger 的 OTLP gRPC 端点
tls:
insecure: true # 开发环境禁用 TLS，生产环境需配置证书
prometheus: # Prometheus 导出器，用于指标数据
endpoint: "0.0.0.0:8889" # Prometheus 刮取端点
namespace: "rustfs" # 指标前缀
send_timestamps: true # 发送时间戳
# enable_open_metrics: true
loki: # Loki 导出器，用于日志数据
# endpoint: "http://loki:3100/otlp/v1/logs"
endpoint: "http://loki:3100/loki/api/v1/push"
tls:
insecure: true
extensions:
health_check:
pprof:
zpages:
service:
extensions: [ health_check, pprof, zpages ] # 启用扩展
pipelines:
traces:
receivers: [ otlp ]
processors: [ memory_limiter,batch ]
exporters: [ otlp/traces ]
metrics:
receivers: [ otlp ]
processors: [ batch ]
exporters: [ prometheus ]
logs:
receivers: [ otlp ]
processors: [ batch ]
exporters: [ loki ]
telemetry:
logs:
level: "info" # Collector 日志级别
metrics:
address: "0.0.0.0:8888" # Collector 自身指标暴露
.docker/observability/prometheus.yml
+11
View File
@@ -0,0 +1,11 @@
global:
scrape_interval: 5s # 刮取间隔
scrape_configs:
- job_name: 'otel-collector'
static_configs:
- targets: ['otel-collector:8888'] # 从 Collector 刮取指标
- job_name: 'otel-metrics'
static_configs:
- targets: ['otel-collector:8889'] # 应用指标
.gitignore
+2 -2
View File
@@ -9,6 +9,6 @@
rustfs/static/* rustfs/static/*
vendor vendor
cli/rustfs-gui/embedded-rustfs/rustfs cli/rustfs-gui/embedded-rustfs/rustfs
.log
config/obs.toml config/obs.toml
config/certs/* *.log
config/certs/*
Cargo.lock
Generated
+1064 -569
View File
File diff suppressed because it is too large Load Diff
Cargo.toml
+21 -8
View File
@@ -11,6 +11,7 @@ members = [
"iam", # Identity and Access Management "iam", # Identity and Access Management
"crypto", # Cryptography and security features "crypto", # Cryptography and security features
"cli/rustfs-gui", # Graphical user interface client "cli/rustfs-gui", # Graphical user interface client
"packages/obs", # Observability utilities
"s3select/api", "s3select/api",
"s3select/query", "s3select/query",
] ]
@@ -32,12 +33,13 @@ all = "warn"
[workspace.dependencies] [workspace.dependencies]
madmin = { path = "./madmin" } madmin = { path = "./madmin" }
async-recursion = "1.0.5" async-recursion = "1.0.5"
async-trait = "0.1.86" async-trait = "0.1.87"
backon = "1.3.0" backon = "1.3.0"
bytes = "1.9.0" bytes = "1.9.0"
bytesize = "1.3.0" bytesize = "1.3.0"
chrono = { version = "0.4.39", features = ["serde"] } chrono = { version = "0.4.40", features = ["serde"] }
clap = { version = "4.5.31", features = ["derive", "env"] } clap = { version = "4.5.31", features = ["derive", "env"] }
config = "0.15.9"
datafusion = "46.0.0" datafusion = "46.0.0"
derive_builder = "0.20.2" derive_builder = "0.20.2"
dioxus = { version = "0.6.3", features = ["router"] } dioxus = { version = "0.6.3", features = ["router"] }
@@ -50,7 +52,6 @@ common = { path = "./common/common" }
policy = { path = "./policy" } policy = { path = "./policy" }
hex = "0.4.3" hex = "0.4.3"
hyper = "1.6.0" hyper = "1.6.0"
#hyper-rustls = { version = "0.27.5", features = ["http2"] }
hyper-util = { version = "0.1.10", features = [ hyper-util = { version = "0.1.10", features = [
"tokio", "tokio",
"server-auto", "server-auto",
@@ -65,7 +66,16 @@ lazy_static = "1.5.0"
local-ip-address = "0.6.3" local-ip-address = "0.6.3"
mime = "0.3.17" mime = "0.3.17"
netif = "0.1.6" netif = "0.1.6"
once_cell = "1.21.1"
opentelemetry = { version = "0.29" }
opentelemetry-appender-tracing = { version = "0.29.1", features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] }
opentelemetry_sdk = { version = "0.29" }
opentelemetry-stdout = { version = "0.29.0" }
opentelemetry-otlp = { version = "0.29" }
opentelemetry-prometheus = { version = "0.29" }
opentelemetry-semantic-conventions = { version = "0.29.0", features = ["semconv_experimental"] }
pin-project-lite = "0.2" pin-project-lite = "0.2"
prometheus = "0.13.4"
# pin-utils = "0.1.0" # pin-utils = "0.1.0"
prost = "0.13.4" prost = "0.13.4"
prost-build = "0.13.4" prost-build = "0.13.4"
@@ -73,10 +83,12 @@ prost-types = "0.13.4"
protobuf = "3.7" protobuf = "3.7"
protos = { path = "./common/protos" } protos = { path = "./common/protos" }
rand = "0.8.5" rand = "0.8.5"
reqwest = { version = "0.12.12", default-features = false, features = ["rustls-tls", "charset", "http2", "macos-system-configuration", "stream", "blocking"] } rdkafka = { version = "0.37", features = ["tokio"] }
reqwest = { version = "0.12.15", default-features = false, features = ["rustls-tls", "charset", "http2", "macos-system-configuration", "stream", "json", "blocking"] }
rfd = { version = "0.15.2", default-features = false, features = ["xdg-portal", "tokio"] } rfd = { version = "0.15.2", default-features = false, features = ["xdg-portal", "tokio"] }
rmp = "0.8.14" rmp = "0.8.14"
rmp-serde = "1.3.0" rmp-serde = "1.3.0"
rustfs-obs = { path = "packages/obs", version = "0.0.1" }
rust-embed = "8.6.0" rust-embed = "8.6.0"
rustls = { version = "0.23" } rustls = { version = "0.23" }
rustls-pki-types = "1.11.0" rustls-pki-types = "1.11.0"
@@ -90,15 +102,15 @@ serde = { version = "1.0.217", features = ["derive"] }
serde_json = "1.0.138" serde_json = "1.0.138"
sha2 = "0.10.8" sha2 = "0.10.8"
tempfile = "3.16.0" tempfile = "3.16.0"
thiserror = "2.0.11" thiserror = "2.0.12"
time = { version = "0.3.37", features = [ time = { version = "0.3.41", features = [
"std", "std",
"parsing", "parsing",
"formatting", "formatting",
"macros", "macros",
"serde", "serde",
] } ] }
tokio = { version = "1.43.0", features = ["fs", "rt-multi-thread"] } tokio = { version = "1.44.0", features = ["fs", "rt-multi-thread"] }
tonic = { version = "0.12.3", features = ["gzip"] } tonic = { version = "0.12.3", features = ["gzip"] }
tonic-build = "0.12.3" tonic-build = "0.12.3"
tonic-reflection = "0.12" tonic-reflection = "0.12"
@@ -108,9 +120,11 @@ tokio-util = { version = "0.7.13", features = ["io", "compat"] }
tower = { version = "0.5.2", features = ["timeout"] } tower = { version = "0.5.2", features = ["timeout"] }
tower-http = { version = "0.6.2", features = ["cors"] } tower-http = { version = "0.6.2", features = ["cors"] }
tracing = "0.1.41" tracing = "0.1.41"
tracing-core = "0.1.33"
tracing-error = "0.2.1" tracing-error = "0.2.1"
tracing-subscriber = { version = "0.3.19", features = ["env-filter", "time"] } tracing-subscriber = { version = "0.3.19", features = ["env-filter", "time"] }
tracing-appender = "0.2.3" tracing-appender = "0.2.3"
tracing-opentelemetry = "0.30"
transform-stream = "0.3.1" transform-stream = "0.3.1"
url = "2.5.4" url = "2.5.4"
uuid = { version = "1.15.1", features = [ uuid = { version = "1.15.1", features = [
@@ -125,7 +139,6 @@ workers = { path = "./common/workers" }
test-case = "3.3.1" test-case = "3.3.1"
snafu = "0.8.5" snafu = "0.8.5"
[profile.wasm-dev] [profile.wasm-dev]
inherits = "dev" inherits = "dev"
opt-level = 1 opt-level = 1
README.md
+45
View File
@@ -40,6 +40,7 @@ export RUSTFS_VOLUMES="./target/volume/test"
export RUSTFS_ADDRESS="0.0.0.0:9000" export RUSTFS_ADDRESS="0.0.0.0:9000"
export RUSTFS_CONSOLE_ENABLE=true export RUSTFS_CONSOLE_ENABLE=true
export RUSTFS_CONSOLE_ADDRESS="0.0.0.0:9001" export RUSTFS_CONSOLE_ADDRESS="0.0.0.0:9001"
export RUSTFS_OBS_CONFIG="config/obs.toml"
``` ```
You need replace your real data folder: You need replace your real data folder:
@@ -47,3 +48,47 @@ You need replace your real data folder:
``` ```
./rustfs /data/rustfs ./rustfs /data/rustfs
``` ```
## How to deploy the observability stack
The OpenTelemetry Collector offers a vendor-agnostic implementation on how to receive, process, and export telemetry
data. It removes the need to run, operate, and maintain multiple agents/collectors in order to support open-source
observability data formats (e.g. Jaeger, Prometheus, etc.) sending to one or more open-source or commercial back-ends.
1. Enter the `.docker/observability` directory,
2. Run the following command:
```bash
docker-compose -f docker-compose.yml up -d
```
3. Access the Grafana dashboard by navigating to `http://localhost:3000` in your browser. The default username and
password are `admin` and `admin`, respectively.
4. Access the Jaeger dashboard by navigating to `http://localhost:16686` in your browser.
5. Access the Prometheus dashboard by navigating to `http://localhost:9090` in your browser.
## Create a new Observability configuration file
#### 1. Enter the `config` directory,
#### 2. Copy `obs.toml.example` to `obs.toml`
#### 3. Modify the `obs.toml` configuration file
##### 3.1. Modify the `endpoint` value to the address of the OpenTelemetry Collector
##### 3.2. Modify the `service_name` value to the name of the service
##### 3.3. Modify the `service_version` value to the version of the service
##### 3.4. Modify the `deployment_environment` value to the environment of the service
##### 3.5. Modify the `meter_interval` value to export interval
##### 3.6. Modify the `sample_ratio` value to the sample ratio
##### 3.7. Modify the `use_stdout` value to export to stdout
config/obs.example.toml
+33
View File
@@ -0,0 +1,33 @@
[observability]
endpoint = "http://localhost:4317" # Default is "http://localhost:4317" if not specified
use_stdout = false # Output with stdout, true output, false no output
sample_ratio = 2.0
meter_interval = 30
service_name = "rustfs"
service_version = "0.1.0"
environment = "develop"
looger_level = "info"
[sinks]
[sinks.kafka] # Kafka sink is disabled by default
enabled = false
bootstrap_servers = "localhost:9092"
topic = "logs"
batch_size = 100 # Default is 100 if not specified
batch_timeout_ms = 1000 # Default is 1000ms if not specified
[sinks.webhook]
enabled = false
endpoint = "http://localhost:8080/webhook"
auth_token = ""
batch_size = 100 # Default is 3 if not specified
batch_timeout_ms = 1000 # Default is 100ms if not specified
[sinks.file]
enabled = true
path = "logs/app.log"
batch_size = 100
batch_timeout_ms = 1000 # Default is 8192 bytes if not specified
[logger]
queue_capacity = 10000
ecstore/src/endpoints.rs
+4 -3
View File
@@ -1,4 +1,4 @@
use tracing::warn; use tracing::{instrument, warn};
use crate::{ use crate::{
disk::endpoint::{Endpoint, EndpointType}, disk::endpoint::{Endpoint, EndpointType},
@@ -407,7 +407,7 @@ pub struct PoolEndpoints {
pub platform: String, pub platform: String,
} }
/// list of list of endpoints /// list of endpoints
#[derive(Debug, Clone, Default)] #[derive(Debug, Clone, Default)]
pub struct EndpointServerPools(pub Vec<PoolEndpoints>); pub struct EndpointServerPools(pub Vec<PoolEndpoints>);
@@ -532,6 +532,8 @@ impl EndpointServerPools {
nodes nodes
} }
#[instrument]
pub fn hosts_sorted(&self) -> Vec<Option<XHost>> { pub fn hosts_sorted(&self) -> Vec<Option<XHost>> {
let (mut peers, local) = self.peers(); let (mut peers, local) = self.peers();
@@ -604,7 +606,6 @@ impl EndpointServerPools {
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use super::*; use super::*;
use std::path::Path; use std::path::Path;
ecstore/src/global.rs
+1 -1
View File
@@ -61,7 +61,7 @@ pub fn get_global_deployment_id() -> Option<String> {
pub fn set_global_endpoints(eps: Vec<PoolEndpoints>) { pub fn set_global_endpoints(eps: Vec<PoolEndpoints>) {
GLOBAL_Endpoints GLOBAL_Endpoints
.set(EndpointServerPools::from(eps)) .set(EndpointServerPools::from(eps))
.expect("GLOBAL_Endpoints set faild") .expect("GLOBAL_Endpoints set failed")
} }
pub fn get_global_endpoints() -> EndpointServerPools { pub fn get_global_endpoints() -> EndpointServerPools {
ecstore/src/heal/data_scanner.rs
+9 -15
View File
@@ -352,7 +352,7 @@ impl CurrentScannerCycle {
let str_len = rmp::decode::read_str_len(&mut cur)?; let str_len = rmp::decode::read_str_len(&mut cur)?;
// ！！！ Vec::with_capacity(str_len) 失败，vec!正常 // ！！！Vec::with_capacity(str_len) 失败，vec! 正常
let mut field_buff = vec![0u8; str_len as usize]; let mut field_buff = vec![0u8; str_len as usize];
cur.read_exact(&mut field_buff)?; cur.read_exact(&mut field_buff)?;
@@ -435,7 +435,7 @@ impl ScannerItem {
pub async fn apply_versions_actions(&self, fivs: &[FileInfo]) -> Result<Vec<ObjectInfo>> { pub async fn apply_versions_actions(&self, fivs: &[FileInfo]) -> Result<Vec<ObjectInfo>> {
let obj_infos = self.apply_newer_noncurrent_version_limit(fivs).await?; let obj_infos = self.apply_newer_noncurrent_version_limit(fivs).await?;
if obj_infos.len() >= SCANNER_EXCESS_OBJECT_VERSIONS.load(Ordering::SeqCst).try_into().unwrap() { if obj_infos.len() >= <u64 as TryInto<usize>>::try_into(SCANNER_EXCESS_OBJECT_VERSIONS.load(Ordering::SeqCst)).unwrap() {
// todo // todo
} }
@@ -445,10 +445,7 @@ impl ScannerItem {
} }
if cumulative_size if cumulative_size
>= SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL_SIZE >= <u64 as TryInto<usize>>::try_into(SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL_SIZE.load(Ordering::SeqCst)).unwrap()
.load(Ordering::SeqCst)
.try_into()
.unwrap()
{ {
//todo //todo
} }
@@ -684,16 +681,13 @@ impl FolderScanner {
} }
let should_compact = self.new_cache.info.name != folder.name let should_compact = self.new_cache.info.name != folder.name
&& existing_folders.len() + new_folders.len() >= DATA_SCANNER_COMPACT_AT_FOLDERS.try_into().unwrap() && existing_folders.len() + new_folders.len()
|| existing_folders.len() + new_folders.len() >= DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS.try_into().unwrap(); >= <u64 as TryInto<usize>>::try_into(DATA_SCANNER_COMPACT_AT_FOLDERS).unwrap()
|| existing_folders.len() + new_folders.len()
>= <u64 as TryInto<usize>>::try_into(DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS).unwrap();
let total_folders = existing_folders.len() + new_folders.len(); let total_folders = existing_folders.len() + new_folders.len();
if total_folders if total_folders > <u64 as TryInto<usize>>::try_into(SCANNER_EXCESS_FOLDERS.load(Ordering::SeqCst)).unwrap() {
> SCANNER_EXCESS_FOLDERS
.load(std::sync::atomic::Ordering::SeqCst)
.try_into()
.unwrap()
{
let _prefix_name = format!("{}/", folder.name.trim_end_matches('/')); let _prefix_name = format!("{}/", folder.name.trim_end_matches('/'));
// todo: notification // todo: notification
} }
@@ -957,7 +951,7 @@ impl FolderScanner {
if !into.compacted && self.new_cache.info.name != folder.name { if !into.compacted && self.new_cache.info.name != folder.name {
let mut flat = self.new_cache.size_recursive(&this_hash.key()).unwrap_or_default(); let mut flat = self.new_cache.size_recursive(&this_hash.key()).unwrap_or_default();
flat.compacted = true; flat.compacted = true;
let compact = if flat.objects < DATA_SCANNER_COMPACT_LEAST_OBJECT.try_into().unwrap() { let compact = if flat.objects < <u64 as TryInto<usize>>::try_into(DATA_SCANNER_COMPACT_LEAST_OBJECT).unwrap() {
true true
} else { } else {
// Compact if we only have objects as children... // Compact if we only have objects as children...
ecstore/src/heal/data_usage_cache.rs
+1 -1
View File
@@ -588,7 +588,7 @@ impl DataUsageCache {
Some(e) => e, Some(e) => e,
None => return, None => return,
}; };
if top_e.children.len() > DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS.try_into().unwrap() { if top_e.children.len() > <u64 as TryInto<usize>>::try_into(DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS).unwrap() {
self.reduce_children_of(&hash_path(&self.info.name), limit, true); self.reduce_children_of(&hash_path(&self.info.name), limit, true);
} }
if self.cache.len() <= limit { if self.cache.len() <= limit {
iam/src/lib.rs
+2 -1
View File
@@ -6,7 +6,7 @@ use policy::auth::Credentials;
use std::sync::{Arc, OnceLock}; use std::sync::{Arc, OnceLock};
use store::object::ObjectStore; use store::object::ObjectStore;
use sys::IamSys; use sys::IamSys;
use tracing::debug; use tracing::{debug, instrument};
pub mod cache; pub mod cache;
pub mod error; pub mod error;
@@ -51,6 +51,7 @@ pub fn get_global_action_cred() -> Option<Credentials> {
GLOBAL_ACTIVE_CRED.get().cloned() GLOBAL_ACTIVE_CRED.get().cloned()
} }
#[instrument]
pub async fn init_iam_sys(ecstore: Arc<ECStore>) -> Result<()> { pub async fn init_iam_sys(ecstore: Arc<ECStore>) -> Result<()> {
debug!("init iam system"); debug!("init iam system");
let s = IamCache::new(ObjectStore::new(ecstore)).await; let s = IamCache::new(ObjectStore::new(ecstore)).await;
packages/obs/Cargo.toml
+53
View File
@@ -0,0 +1,53 @@
[package]
name = "rustfs-obs"
edition.workspace = true
license.workspace = true
repository.workspace = true
rust-version.workspace = true
version.workspace = true
[lints]
workspace = true
[features]
default = ["file"]
kafka = ["dep:rdkafka"]
webhook = ["dep:reqwest"]
file = []
[dependencies]
async-trait = { workspace = true }
chrono = { workspace = true }
config = { workspace = true }
opentelemetry = { workspace = true }
opentelemetry-appender-tracing = { workspace = true, features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] }
opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
opentelemetry-stdout = { workspace = true }
opentelemetry-otlp = { workspace = true, features = ["grpc-tonic", "gzip-tonic"] }
opentelemetry-prometheus = { workspace = true }
opentelemetry-semantic-conventions = { workspace = true, features = ["semconv_experimental"] }
prometheus = { workspace = true }
serde = { workspace = true }
tracing = { workspace = true, features = ["std", "attributes"] }
tracing-core = { workspace = true }
tracing-error = { workspace = true }
tracing-opentelemetry = { workspace = true }
tracing-subscriber = { workspace = true, features = ["registry", "std", "fmt", "env-filter", "tracing-log", "time", "local-time", "json"] }
tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread"] }
rdkafka = { workspace = true, features = ["tokio"], optional = true }
reqwest = { workspace = true, optional = true, default-features = false }
serde_json = { workspace = true }
thiserror = { workspace = true }
local-ip-address = { workspace = true }
[dev-dependencies]
chrono = { workspace = true }
opentelemetry = { workspace = true }
opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
opentelemetry-stdout = { workspace = true }
opentelemetry-otlp = { workspace = true, features = ["grpc-tonic"] }
opentelemetry-semantic-conventions = { workspace = true, features = ["semconv_experimental"] }
tokio = { workspace = true, features = ["full"] }
tracing = { workspace = true, features = ["std", "attributes"] }
tracing-subscriber = { workspace = true, features = ["registry", "std", "fmt"] }
packages/obs/examples/config.toml
+31
View File
@@ -0,0 +1,31 @@
[observability]
endpoint = "http://localhost:4317"
use_stdout = true
sample_ratio = 1
meter_interval = 30
service_name = "rustfs_obs"
service_version = "0.1.0"
deployment_environment = "develop"
[sinks]
[sinks.kafka]
enabled = false
bootstrap_servers = "localhost:9092"
topic = "logs"
batch_size = 100 # Default is 100 if not specified
batch_timeout_ms = 1000 # Default is 1000ms if not specified
[sinks.webhook]
enabled = false
url = "http://localhost:8080/webhook"
batch_size = 100 # Default is 3 if not specified
batch_timeout_ms = 1000 # Default is 100ms if not specified
[sinks.file]
enabled = true
path = "logs/app.log"
batch_size = 100
batch_timeout_ms = 1000 # Default is 8192 bytes if not specified
[logger]
queue_capacity = 10000
packages/obs/examples/server.rs
+88
View File
@@ -0,0 +1,88 @@
use opentelemetry::global;
use rustfs_obs::{get_logger, init_obs, load_config, log_info, BaseLogEntry, ServerLogEntry};
use std::collections::HashMap;
use std::time::{Duration, SystemTime};
use tracing::{info, instrument};
use tracing_core::Level;
#[tokio::main]
async fn main() {
let obs_conf = Some("packages/obs/examples/config.toml".to_string());
let config = load_config(obs_conf);
let (_logger, _guard) = init_obs(config.clone()).await;
let span = tracing::span!(Level::INFO, "main");
let _enter = span.enter();
info!("Program starts");
// Simulate the operation
tokio::time::sleep(Duration::from_millis(100)).await;
run(
"service-demo".to_string(),
"object-demo".to_string(),
"user-demo".to_string(),
"service-demo".to_string(),
)
.await;
info!("Program ends");
}
#[instrument(fields(bucket, object, user))]
async fn run(bucket: String, object: String, user: String, service_name: String) {
let start_time = SystemTime::now();
info!("Log module initialization is completed service_name: {:?}", service_name);
// Record Metrics
let meter = global::meter("rustfs");
let request_duration = meter.f64_histogram("s3_request_duration_seconds").build();
request_duration.record(
start_time.elapsed().unwrap().as_secs_f64(),
&[opentelemetry::KeyValue::new("operation", "run")],
);
let base_entry = BaseLogEntry::new()
.message(Some("run logger api_handler info".to_string()))
.request_id(Some("request_id".to_string()))
.timestamp(chrono::DateTime::from(start_time))
.tags(Some(HashMap::default()));
let server_entry = ServerLogEntry::new(Level::INFO, "api_handler".to_string())
.with_base(base_entry)
.user_id(Some(user.clone()))
.add_field("operation".to_string(), "login".to_string())
.add_field("bucket".to_string(), bucket.clone())
.add_field("object".to_string(), object.clone());
let result = get_logger().lock().await.log_server_entry(server_entry).await;
info!("Logging is completed {:?}", result);
put_object("bucket".to_string(), "object".to_string(), "user".to_string()).await;
info!("Logging is completed");
tokio::time::sleep(Duration::from_secs(2)).await;
info!("Program run ends");
}
#[instrument(fields(bucket, object, user))]
async fn put_object(bucket: String, object: String, user: String) {
let start_time = SystemTime::now();
info!("Starting put_object operation time: {:?}", start_time);
let meter = global::meter("rustfs");
let request_duration = meter.f64_histogram("s3_request_duration_seconds").build();
request_duration.record(
start_time.elapsed().unwrap().as_secs_f64(),
&[opentelemetry::KeyValue::new("operation", "put_object")],
);
info!(
"Starting PUT operation content: bucket = {}, object = {}, user = {},start_time = {}",
bucket,
object,
user,
start_time.elapsed().unwrap().as_secs_f64()
);
let result = log_info("put_object logger info", "put_object").await;
info!("put_object is completed {:?}", result);
// Simulate the operation
tokio::time::sleep(Duration::from_millis(100)).await;
info!("PUT operation completed");
}
packages/obs/src/config.rs
+182
View File
@@ -0,0 +1,182 @@
use crate::global::{ENVIRONMENT, LOGGER_LEVEL, METER_INTERVAL, SAMPLE_RATIO, SERVICE_NAME, SERVICE_VERSION};
use config::{Config, File, FileFormat};
use serde::Deserialize;
use std::env;
/// OpenTelemetry Configuration
/// Add service name, service version, environment
/// Add interval time for metric collection
/// Add sample ratio for trace sampling
/// Add endpoint for metric collection
/// Add use_stdout for output to stdout
/// Add logger level for log level
#[derive(Debug, Deserialize, Clone)]
pub struct OtelConfig {
pub endpoint: String,
pub use_stdout: Option<bool>,
pub sample_ratio: Option<f64>,
pub meter_interval: Option<u64>,
pub service_name: Option<String>,
pub service_version: Option<String>,
pub environment: Option<String>,
pub logger_level: Option<String>,
}
impl Default for OtelConfig {
fn default() -> Self {
OtelConfig {
endpoint: "".to_string(),
use_stdout: Some(true),
sample_ratio: Some(SAMPLE_RATIO),
meter_interval: Some(METER_INTERVAL),
service_name: Some(SERVICE_NAME.to_string()),
service_version: Some(SERVICE_VERSION.to_string()),
environment: Some(ENVIRONMENT.to_string()),
logger_level: Some(LOGGER_LEVEL.to_string()),
}
}
}
/// Kafka Sink Configuration - Add batch parameters
#[derive(Debug, Deserialize, Clone, Default)]
pub struct KafkaSinkConfig {
pub enabled: bool,
pub bootstrap_servers: String,
pub topic: String,
pub batch_size: Option<usize>, // Batch size, default 100
pub batch_timeout_ms: Option<u64>, // Batch timeout time, default 1000ms
}
/// Webhook Sink Configuration - Add Retry Parameters
#[derive(Debug, Deserialize, Clone, Default)]
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
}
/// File Sink Configuration - Add buffering parameters
#[derive(Debug, Deserialize, 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
pub flush_threshold: Option<usize>, // Refresh threshold, default 100 logs
}
impl Default for FileSinkConfig {
fn default() -> Self {
FileSinkConfig {
enabled: true,
path: "logs/app.log".to_string(),
buffer_size: Some(8192),
flush_interval_ms: Some(1000),
flush_threshold: Some(100),
}
}
}
/// Sink configuration collection
#[derive(Debug, Deserialize, Clone, Default)]
pub struct SinkConfig {
pub kafka: KafkaSinkConfig,
pub webhook: WebhookSinkConfig,
pub file: FileSinkConfig,
}
///Logger Configuration
#[derive(Debug, Deserialize, Clone)]
pub struct LoggerConfig {
pub queue_capacity: Option<usize>,
}
impl Default for LoggerConfig {
fn default() -> Self {
LoggerConfig {
queue_capacity: Some(1000),
}
}
}
/// Overall application configuration
/// Add observability, sinks, and logger configuration
///
/// Observability: OpenTelemetry configuration
/// Sinks: Kafka, Webhook, File sink configuration
/// Logger: Logger configuration
///
/// # Example
/// ```
/// use rustfs_obs::AppConfig;
/// use rustfs_obs::load_config;
///
/// let config = load_config(None);
/// ```
#[derive(Debug, Deserialize, Clone, Default)]
pub struct AppConfig {
pub observability: OtelConfig,
pub sinks: SinkConfig,
pub logger: LoggerConfig,
}
const DEFAULT_CONFIG_FILE: &str = "obs";
/// Loading the configuration file
/// Supports TOML, YAML and .env formats, read in order by priority
///
/// # Parameters
/// - `config_dir`: Configuration file path
///
/// # Returns
/// Configuration information
///
/// # Example
/// ```
/// use rustfs_obs::AppConfig;
/// use rustfs_obs::load_config;
///
/// let config = load_config(None);
/// ```
pub fn load_config(config_dir: Option<String>) -> AppConfig {
let config_dir = if let Some(path) = config_dir {
// If a path is provided, check if it's empty
if path.is_empty() {
// If empty, use the default config file name
DEFAULT_CONFIG_FILE.to_string()
} else {
// Use the provided path
let path = std::path::Path::new(&path);
if path.extension().is_some() {
// If path has extension, use it as is (extension will be added by Config::builder)
path.with_extension("").to_string_lossy().into_owned()
} else {
// If path is a directory, append the default config file name
path.to_string_lossy().into_owned()
}
}
} else {
// If no path provided, use current directory + default config file
match env::current_dir() {
Ok(dir) => dir.join(DEFAULT_CONFIG_FILE).to_string_lossy().into_owned(),
Err(_) => {
eprintln!("Warning: Failed to get current directory, using default config file");
DEFAULT_CONFIG_FILE.to_string()
}
}
};
// Log using proper logging instead of println when possible
println!("Using config file base: {}", config_dir);
let 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(config::Environment::with_prefix(""))
.build()
.unwrap_or_default();
config.try_deserialize().unwrap_or_default()
}
packages/obs/src/entry/args.rs
+74
View File
@@ -0,0 +1,74 @@
use crate::entry::ObjectVersion;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
/// Args - defines the arguments for API operations
/// Args is used to define the arguments for API operations.
///
/// # Example
/// ```
/// use rustfs_obs::Args;
/// use std::collections::HashMap;
///
/// let args = Args::new()
/// .set_bucket(Some("my-bucket".to_string()))
/// .set_object(Some("my-object".to_string()))
/// .set_version_id(Some("123".to_string()))
/// .set_metadata(Some(HashMap::new()));
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, Default, Eq, PartialEq)]
pub struct Args {
#[serde(rename = "bucket", skip_serializing_if = "Option::is_none")]
pub bucket: Option<String>,
#[serde(rename = "object", skip_serializing_if = "Option::is_none")]
pub object: Option<String>,
#[serde(rename = "versionId", skip_serializing_if = "Option::is_none")]
pub version_id: Option<String>,
#[serde(rename = "objects", skip_serializing_if = "Option::is_none")]
pub objects: Option<Vec<ObjectVersion>>,
#[serde(rename = "metadata", skip_serializing_if = "Option::is_none")]
pub metadata: Option<HashMap<String, String>>,
}
impl Args {
/// Create a new Args object
pub fn new() -> Self {
Args {
bucket: None,
object: None,
version_id: None,
objects: None,
metadata: None,
}
}
/// Set the bucket
pub fn set_bucket(mut self, bucket: Option<String>) -> Self {
self.bucket = bucket;
self
}
/// Set the object
pub fn set_object(mut self, object: Option<String>) -> Self {
self.object = object;
self
}
/// Set the version ID
pub fn set_version_id(mut self, version_id: Option<String>) -> Self {
self.version_id = version_id;
self
}
/// Set the objects
pub fn set_objects(mut self, objects: Option<Vec<ObjectVersion>>) -> Self {
self.objects = objects;
self
}
/// Set the metadata
pub fn set_metadata(mut self, metadata: Option<HashMap<String, String>>) -> Self {
self.metadata = metadata;
self
}
}
packages/obs/src/entry/audit.rs
+453
View File
@@ -0,0 +1,453 @@
use crate::{BaseLogEntry, LogRecord, ObjectVersion};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
/// API details structure
/// ApiDetails is used to define the details of an API operation
///
/// The `ApiDetails` structure contains the following fields:
/// - `name` - the name of the API operation
/// - `bucket` - the bucket name
/// - `object` - the object name
/// - `objects` - the list of objects
/// - `status` - the status of the API operation
/// - `status_code` - the status code of the API operation
/// - `input_bytes` - the input bytes
/// - `output_bytes` - the output bytes
/// - `header_bytes` - the header bytes
/// - `time_to_first_byte` - the time to first byte
/// - `time_to_first_byte_in_ns` - the time to first byte in nanoseconds
/// - `time_to_response` - the time to response
/// - `time_to_response_in_ns` - the time to response in nanoseconds
///
/// The `ApiDetails` structure contains the following methods:
/// - `new` - create a new `ApiDetails` with default values
/// - `set_name` - set the name
/// - `set_bucket` - set the bucket
/// - `set_object` - set the object
/// - `set_objects` - set the objects
/// - `set_status` - set the status
/// - `set_status_code` - set the status code
/// - `set_input_bytes` - set the input bytes
/// - `set_output_bytes` - set the output bytes
/// - `set_header_bytes` - set the header bytes
/// - `set_time_to_first_byte` - set the time to first byte
/// - `set_time_to_first_byte_in_ns` - set the time to first byte in nanoseconds
/// - `set_time_to_response` - set the time to response
/// - `set_time_to_response_in_ns` - set the time to response in nanoseconds
///
/// # Example
/// ```
/// use rustfs_obs::ApiDetails;
/// use rustfs_obs::ObjectVersion;
///
/// let api = ApiDetails::new()
/// .set_name(Some("GET".to_string()))
/// .set_bucket(Some("my-bucket".to_string()))
/// .set_object(Some("my-object".to_string()))
/// .set_objects(vec![ObjectVersion::new_with_object_name("my-object".to_string())])
/// .set_status(Some("OK".to_string()))
/// .set_status_code(Some(200))
/// .set_input_bytes(100)
/// .set_output_bytes(200)
/// .set_header_bytes(Some(50))
/// .set_time_to_first_byte(Some("100ms".to_string()))
/// .set_time_to_first_byte_in_ns(Some("100000000ns".to_string()))
/// .set_time_to_response(Some("200ms".to_string()))
/// .set_time_to_response_in_ns(Some("200000000ns".to_string()));
/// ```
#[derive(Debug, Serialize, Deserialize, Clone, Default, PartialEq, Eq)]
pub struct ApiDetails {
#[serde(rename = "name", skip_serializing_if = "Option::is_none")]
pub name: Option<String>,
#[serde(rename = "bucket", skip_serializing_if = "Option::is_none")]
pub bucket: Option<String>,
#[serde(rename = "object", skip_serializing_if = "Option::is_none")]
pub object: Option<String>,
#[serde(rename = "objects", skip_serializing_if = "Vec::is_empty", default)]
pub objects: Vec<ObjectVersion>,
#[serde(rename = "status", skip_serializing_if = "Option::is_none")]
pub status: Option<String>,
#[serde(rename = "statusCode", skip_serializing_if = "Option::is_none")]
pub status_code: Option<i32>,
#[serde(rename = "rx")]
pub input_bytes: i64,
#[serde(rename = "tx")]
pub output_bytes: i64,
#[serde(rename = "txHeaders", skip_serializing_if = "Option::is_none")]
pub header_bytes: Option<i64>,
#[serde(rename = "timeToFirstByte", skip_serializing_if = "Option::is_none")]
pub time_to_first_byte: Option<String>,
#[serde(rename = "timeToFirstByteInNS", skip_serializing_if = "Option::is_none")]
pub time_to_first_byte_in_ns: Option<String>,
#[serde(rename = "timeToResponse", skip_serializing_if = "Option::is_none")]
pub time_to_response: Option<String>,
#[serde(rename = "timeToResponseInNS", skip_serializing_if = "Option::is_none")]
pub time_to_response_in_ns: Option<String>,
}
impl ApiDetails {
/// Create a new `ApiDetails` with default values
pub fn new() -> Self {
ApiDetails {
name: None,
bucket: None,
object: None,
objects: Vec::new(),
status: None,
status_code: None,
input_bytes: 0,
output_bytes: 0,
header_bytes: None,
time_to_first_byte: None,
time_to_first_byte_in_ns: None,
time_to_response: None,
time_to_response_in_ns: None,
}
}
/// Set the name
pub fn set_name(mut self, name: Option<String>) -> Self {
self.name = name;
self
}
/// Set the bucket
pub fn set_bucket(mut self, bucket: Option<String>) -> Self {
self.bucket = bucket;
self
}
/// Set the object
pub fn set_object(mut self, object: Option<String>) -> Self {
self.object = object;
self
}
/// Set the objects
pub fn set_objects(mut self, objects: Vec<ObjectVersion>) -> Self {
self.objects = objects;
self
}
/// Set the status
pub fn set_status(mut self, status: Option<String>) -> Self {
self.status = status;
self
}
/// Set the status code
pub fn set_status_code(mut self, status_code: Option<i32>) -> Self {
self.status_code = status_code;
self
}
/// Set the input bytes
pub fn set_input_bytes(mut self, input_bytes: i64) -> Self {
self.input_bytes = input_bytes;
self
}
/// Set the output bytes
pub fn set_output_bytes(mut self, output_bytes: i64) -> Self {
self.output_bytes = output_bytes;
self
}
/// Set the header bytes
pub fn set_header_bytes(mut self, header_bytes: Option<i64>) -> Self {
self.header_bytes = header_bytes;
self
}
/// Set the time to first byte
pub fn set_time_to_first_byte(mut self, time_to_first_byte: Option<String>) -> Self {
self.time_to_first_byte = time_to_first_byte;
self
}
/// Set the time to first byte in nanoseconds
pub fn set_time_to_first_byte_in_ns(mut self, time_to_first_byte_in_ns: Option<String>) -> Self {
self.time_to_first_byte_in_ns = time_to_first_byte_in_ns;
self
}
/// Set the time to response
pub fn set_time_to_response(mut self, time_to_response: Option<String>) -> Self {
self.time_to_response = time_to_response;
self
}
/// Set the time to response in nanoseconds
pub fn set_time_to_response_in_ns(mut self, time_to_response_in_ns: Option<String>) -> Self {
self.time_to_response_in_ns = time_to_response_in_ns;
self
}
}
/// Entry - audit entry logs
/// AuditLogEntry is used to define the structure of an audit log entry
///
/// The `AuditLogEntry` structure contains the following fields:
/// - `base` - the base log entry
/// - `version` - the version of the audit log entry
/// - `deployment_id` - the deployment ID
/// - `event` - the event
/// - `entry_type` - the type of audit message
/// - `api` - the API details
/// - `remote_host` - the remote host
/// - `user_agent` - the user agent
/// - `req_path` - the request path
/// - `req_host` - the request host
/// - `req_claims` - the request claims
/// - `req_query` - the request query
/// - `req_header` - the request header
/// - `resp_header` - the response header
/// - `access_key` - the access key
/// - `parent_user` - the parent user
/// - `error` - the error
///
/// The `AuditLogEntry` structure contains the following methods:
/// - `new` - create a new `AuditEntry` with default values
/// - `new_with_values` - create a new `AuditEntry` with version, time, event and api details
/// - `with_base` - set the base log entry
/// - `set_version` - set the version
/// - `set_deployment_id` - set the deployment ID
/// - `set_event` - set the event
/// - `set_entry_type` - set the entry type
/// - `set_api` - set the API details
/// - `set_remote_host` - set the remote host
/// - `set_user_agent` - set the user agent
/// - `set_req_path` - set the request path
/// - `set_req_host` - set the request host
/// - `set_req_claims` - set the request claims
/// - `set_req_query` - set the request query
/// - `set_req_header` - set the request header
/// - `set_resp_header` - set the response header
/// - `set_access_key` - set the access key
/// - `set_parent_user` - set the parent user
/// - `set_error` - set the error
///
/// # Example
/// ```
/// use rustfs_obs::AuditLogEntry;
/// use rustfs_obs::ApiDetails;
/// use std::collections::HashMap;
///
/// let entry = AuditLogEntry::new()
/// .set_version("1.0".to_string())
/// .set_deployment_id(Some("123".to_string()))
/// .set_event("event".to_string())
/// .set_entry_type(Some("type".to_string()))
/// .set_api(ApiDetails::new())
/// .set_remote_host(Some("remote-host".to_string()))
/// .set_user_agent(Some("user-agent".to_string()))
/// .set_req_path(Some("req-path".to_string()))
/// .set_req_host(Some("req-host".to_string()))
/// .set_req_claims(Some(HashMap::new()))
/// .set_req_query(Some(HashMap::new()))
/// .set_req_header(Some(HashMap::new()))
/// .set_resp_header(Some(HashMap::new()))
/// .set_access_key(Some("access-key".to_string()))
/// .set_parent_user(Some("parent-user".to_string()))
/// .set_error(Some("error".to_string()));
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
pub struct AuditLogEntry {
#[serde(flatten)]
pub base: BaseLogEntry,
pub version: String,
#[serde(rename = "deploymentid", skip_serializing_if = "Option::is_none")]
pub deployment_id: Option<String>,
pub event: String,
// Class of audit message - S3, admin ops, bucket management
#[serde(rename = "type", skip_serializing_if = "Option::is_none")]
pub entry_type: Option<String>,
pub api: ApiDetails,
#[serde(rename = "remotehost", skip_serializing_if = "Option::is_none")]
pub remote_host: Option<String>,
#[serde(rename = "userAgent", skip_serializing_if = "Option::is_none")]
pub user_agent: Option<String>,
#[serde(rename = "requestPath", skip_serializing_if = "Option::is_none")]
pub req_path: Option<String>,
#[serde(rename = "requestHost", skip_serializing_if = "Option::is_none")]
pub req_host: Option<String>,
#[serde(rename = "requestClaims", skip_serializing_if = "Option::is_none")]
pub req_claims: Option<HashMap<String, Value>>,
#[serde(rename = "requestQuery", skip_serializing_if = "Option::is_none")]
pub req_query: Option<HashMap<String, String>>,
#[serde(rename = "requestHeader", skip_serializing_if = "Option::is_none")]
pub req_header: Option<HashMap<String, String>>,
#[serde(rename = "responseHeader", skip_serializing_if = "Option::is_none")]
pub resp_header: Option<HashMap<String, String>>,
#[serde(rename = "accessKey", skip_serializing_if = "Option::is_none")]
pub access_key: Option<String>,
#[serde(rename = "parentUser", skip_serializing_if = "Option::is_none")]
pub parent_user: Option<String>,
#[serde(rename = "error", skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
impl AuditLogEntry {
/// Create a new `AuditEntry` with default values
pub fn new() -> Self {
AuditLogEntry {
base: BaseLogEntry::new(),
version: String::new(),
deployment_id: None,
event: String::new(),
entry_type: None,
api: ApiDetails::new(),
remote_host: None,
user_agent: None,
req_path: None,
req_host: None,
req_claims: None,
req_query: None,
req_header: None,
resp_header: None,
access_key: None,
parent_user: None,
error: None,
}
}
/// Create a new `AuditEntry` with version, time, event and api details
pub fn new_with_values(version: String, time: DateTime<Utc>, event: String, api: ApiDetails) -> Self {
let mut base = BaseLogEntry::new();
base.timestamp = time;
AuditLogEntry {
base,
version,
deployment_id: None,
event,
entry_type: None,
api,
remote_host: None,
user_agent: None,
req_path: None,
req_host: None,
req_claims: None,
req_query: None,
req_header: None,
resp_header: None,
access_key: None,
parent_user: None,
error: None,
}
}
/// Set the base log entry
pub fn with_base(mut self, base: BaseLogEntry) -> Self {
self.base = base;
self
}
/// Set the version
pub fn set_version(mut self, version: String) -> Self {
self.version = version;
self
}
/// Set the deployment ID
pub fn set_deployment_id(mut self, deployment_id: Option<String>) -> Self {
self.deployment_id = deployment_id;
self
}
/// Set the event
pub fn set_event(mut self, event: String) -> Self {
self.event = event;
self
}
/// Set the entry type
pub fn set_entry_type(mut self, entry_type: Option<String>) -> Self {
self.entry_type = entry_type;
self
}
/// Set the API details
pub fn set_api(mut self, api: ApiDetails) -> Self {
self.api = api;
self
}
/// Set the remote host
pub fn set_remote_host(mut self, remote_host: Option<String>) -> Self {
self.remote_host = remote_host;
self
}
/// Set the user agent
pub fn set_user_agent(mut self, user_agent: Option<String>) -> Self {
self.user_agent = user_agent;
self
}
/// Set the request path
pub fn set_req_path(mut self, req_path: Option<String>) -> Self {
self.req_path = req_path;
self
}
/// Set the request host
pub fn set_req_host(mut self, req_host: Option<String>) -> Self {
self.req_host = req_host;
self
}
/// Set the request claims
pub fn set_req_claims(mut self, req_claims: Option<HashMap<String, Value>>) -> Self {
self.req_claims = req_claims;
self
}
/// Set the request query
pub fn set_req_query(mut self, req_query: Option<HashMap<String, String>>) -> Self {
self.req_query = req_query;
self
}
/// Set the request header
pub fn set_req_header(mut self, req_header: Option<HashMap<String, String>>) -> Self {
self.req_header = req_header;
self
}
/// Set the response header
pub fn set_resp_header(mut self, resp_header: Option<HashMap<String, String>>) -> Self {
self.resp_header = resp_header;
self
}
/// Set the access key
pub fn set_access_key(mut self, access_key: Option<String>) -> Self {
self.access_key = access_key;
self
}
/// Set the parent user
pub fn set_parent_user(mut self, parent_user: Option<String>) -> Self {
self.parent_user = parent_user;
self
}
/// Set the error
pub fn set_error(mut self, error: Option<String>) -> Self {
self.error = error;
self
}
}
impl LogRecord for AuditLogEntry {
fn to_json(&self) -> String {
serde_json::to_string(self).unwrap_or_else(|_| String::from("{}"))
}
fn get_timestamp(&self) -> DateTime<Utc> {
self.base.timestamp
}
}
packages/obs/src/entry/base.rs
+92
View File
@@ -0,0 +1,92 @@
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
/// Base log entry structure shared by all log types
/// This structure is used to serialize log entries to JSON
/// and send them to the log sinks
/// This structure is also used to deserialize log entries from JSON
/// This structure is also used to store log entries in the database
/// This structure is also used to query log entries from the database
///
/// The `BaseLogEntry` structure contains the following fields:
/// - `timestamp` - the timestamp of the log entry
/// - `request_id` - the request ID of the log entry
/// - `message` - the message of the log entry
/// - `tags` - the tags of the log entry
///
/// The `BaseLogEntry` structure contains the following methods:
/// - `new` - create a new `BaseLogEntry` with default values
/// - `message` - set the message
/// - `request_id` - set the request ID
/// - `tags` - set the tags
/// - `timestamp` - set the timestamp
///
/// # Example
/// ```
/// use rustfs_obs::BaseLogEntry;
/// use chrono::{DateTime, Utc};
/// use std::collections::HashMap;
///
/// let timestamp = Utc::now();
/// let request = Some("req-123".to_string());
/// let message = Some("This is a log message".to_string());
/// let tags = Some(HashMap::new());
///
/// let entry = BaseLogEntry::new()
/// .timestamp(timestamp)
/// .request_id(request)
/// .message(message)
/// .tags(tags);
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq, Default)]
pub struct BaseLogEntry {
#[serde(rename = "time")]
pub timestamp: DateTime<Utc>,
#[serde(rename = "requestID", skip_serializing_if = "Option::is_none")]
pub request_id: Option<String>,
#[serde(rename = "message", skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
#[serde(rename = "tags", skip_serializing_if = "Option::is_none")]
pub tags: Option<HashMap<String, Value>>,
}
impl BaseLogEntry {
/// Create a new BaseLogEntry with default values
pub fn new() -> Self {
BaseLogEntry {
timestamp: Utc::now(),
request_id: None,
message: None,
tags: None,
}
}
/// Set the message
pub fn message(mut self, message: Option<String>) -> Self {
self.message = message;
self
}
/// Set the request ID
pub fn request_id(mut self, request_id: Option<String>) -> Self {
self.request_id = request_id;
self
}
/// Set the tags
pub fn tags(mut self, tags: Option<HashMap<String, Value>>) -> Self {
self.tags = tags;
self
}
/// Set the timestamp
pub fn timestamp(mut self, timestamp: DateTime<Utc>) -> Self {
self.timestamp = timestamp;
self
}
}
packages/obs/src/entry/mod.rs
+143
View File
@@ -0,0 +1,143 @@
pub(crate) mod args;
pub(crate) mod audit;
pub(crate) mod base;
pub(crate) mod unified;
use serde::de::Error;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use tracing_core::Level;
/// ObjectVersion is used across multiple modules
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
pub struct ObjectVersion {
#[serde(rename = "name")]
pub object_name: String,
#[serde(rename = "versionId", skip_serializing_if = "Option::is_none")]
pub version_id: Option<String>,
}
impl ObjectVersion {
/// Create a new ObjectVersion object
pub fn new() -> Self {
ObjectVersion {
object_name: String::new(),
version_id: None,
}
}
/// Create a new ObjectVersion with object name
pub fn new_with_object_name(object_name: String) -> Self {
ObjectVersion {
object_name,
version_id: None,
}
}
/// Set the object name
pub fn set_object_name(mut self, object_name: String) -> Self {
self.object_name = object_name;
self
}
/// Set the version ID
pub fn set_version_id(mut self, version_id: Option<String>) -> Self {
self.version_id = version_id;
self
}
}
/// Log kind/level enum
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum LogKind {
#[serde(rename = "INFO")]
Info,
#[serde(rename = "WARNING")]
Warning,
#[serde(rename = "ERROR")]
Error,
#[serde(rename = "FATAL")]
Fatal,
}
impl Default for LogKind {
fn default() -> Self {
LogKind::Info
}
}
/// Trait for types that can be serialized to JSON and have a timestamp
/// This trait is used by `ServerLogEntry` to convert the log entry to JSON
/// and get the timestamp of the log entry
/// This trait is implemented by `ServerLogEntry`
///
/// # Example
/// ```
/// use rustfs_obs::LogRecord;
/// use chrono::{DateTime, Utc};
/// use rustfs_obs::ServerLogEntry;
/// use tracing_core::Level;
///
/// let log_entry = ServerLogEntry::new(Level::INFO, "api_handler".to_string());
/// let json = log_entry.to_json();
/// let timestamp = log_entry.get_timestamp();
/// ```
pub trait LogRecord {
fn to_json(&self) -> String;
fn get_timestamp(&self) -> chrono::DateTime<chrono::Utc>;
}
/// Wrapper for `tracing_core::Level` to implement `Serialize` and `Deserialize`
/// for `ServerLogEntry`
/// This is necessary because `tracing_core::Level` does not implement `Serialize`
/// and `Deserialize`
/// This is a workaround to allow `ServerLogEntry` to be serialized and deserialized
/// using `serde`
///
/// # Example
/// ```
/// use rustfs_obs::SerializableLevel;
/// use tracing_core::Level;
///
/// let level = Level::INFO;
/// let serializable_level = SerializableLevel::from(level);
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SerializableLevel(pub Level);
impl From<Level> for SerializableLevel {
fn from(level: Level) -> Self {
SerializableLevel(level)
}
}
impl From<SerializableLevel> for Level {
fn from(serializable_level: SerializableLevel) -> Self {
serializable_level.0
}
}
impl Serialize for SerializableLevel {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(self.0.as_str())
}
}
impl<'de> Deserialize<'de> for SerializableLevel {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let s = String::deserialize(deserializer)?;
match s.as_str() {
"TRACE" => Ok(SerializableLevel(Level::TRACE)),
"DEBUG" => Ok(SerializableLevel(Level::DEBUG)),
"INFO" => Ok(SerializableLevel(Level::INFO)),
"WARN" => Ok(SerializableLevel(Level::WARN)),
"ERROR" => Ok(SerializableLevel(Level::ERROR)),
_ => Err(D::Error::custom("unknown log level")),
}
}
}
packages/obs/src/entry/unified.rs
+279
View File
@@ -0,0 +1,279 @@
use crate::{AuditLogEntry, BaseLogEntry, LogKind, LogRecord, SerializableLevel};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use tracing_core::Level;
/// Server log entry with structured fields
/// ServerLogEntry is used to log structured log entries from the server
///
/// The `ServerLogEntry` structure contains the following fields:
/// - `base` - the base log entry
/// - `level` - the log level
/// - `source` - the source of the log entry
/// - `user_id` - the user ID
/// - `fields` - the structured fields of the log entry
///
/// The `ServerLogEntry` structure contains the following methods:
/// - `new` - create a new `ServerLogEntry` with specified level and source
/// - `with_base` - set the base log entry
/// - `user_id` - set the user ID
/// - `fields` - set the fields
/// - `add_field` - add a field
///
/// # Example
/// ```
/// use rustfs_obs::ServerLogEntry;
/// use tracing_core::Level;
///
/// let entry = ServerLogEntry::new(Level::INFO, "test_module".to_string())
/// .user_id(Some("user-456".to_string()))
/// .add_field("operation".to_string(), "login".to_string());
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ServerLogEntry {
#[serde(flatten)]
pub base: BaseLogEntry,
pub level: SerializableLevel,
pub source: String,
#[serde(rename = "userId", skip_serializing_if = "Option::is_none")]
pub user_id: Option<String>,
#[serde(skip_serializing_if = "Vec::is_empty", default)]
pub fields: Vec<(String, String)>,
}
impl ServerLogEntry {
/// Create a new ServerLogEntry with specified level and source
pub fn new(level: Level, source: String) -> Self {
ServerLogEntry {
base: BaseLogEntry::new(),
level: SerializableLevel(level),
source,
user_id: None,
fields: Vec::new(),
}
}
/// Set the base log entry
pub fn with_base(mut self, base: BaseLogEntry) -> Self {
self.base = base;
self
}
/// Set the user ID
pub fn user_id(mut self, user_id: Option<String>) -> Self {
self.user_id = user_id;
self
}
/// Set fields
pub fn fields(mut self, fields: Vec<(String, String)>) -> Self {
self.fields = fields;
self
}
/// Add a field
pub fn add_field(mut self, key: String, value: String) -> Self {
self.fields.push((key, value));
self
}
}
impl LogRecord for ServerLogEntry {
fn to_json(&self) -> String {
serde_json::to_string(self).unwrap_or_else(|_| String::from("{}"))
}
fn get_timestamp(&self) -> DateTime<Utc> {
self.base.timestamp
}
}
/// Console log entry structure
/// ConsoleLogEntry is used to log console log entries
/// The `ConsoleLogEntry` structure contains the following fields:
/// - `base` - the base log entry
/// - `level` - the log level
/// - `console_msg` - the console message
/// - `node_name` - the node name
/// - `err` - the error message
/// The `ConsoleLogEntry` structure contains the following methods:
/// - `new` - create a new `ConsoleLogEntry`
/// - `new_with_console_msg` - create a new `ConsoleLogEntry` with console message and node name
/// - `with_base` - set the base log entry
/// - `set_level` - set the log level
/// - `set_node_name` - set the node name
/// - `set_console_msg` - set the console message
/// - `set_err` - set the error message
/// # Example
/// ```
/// use rustfs_obs::ConsoleLogEntry;
///
/// let entry = ConsoleLogEntry::new_with_console_msg("Test message".to_string(), "node-123".to_string());
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConsoleLogEntry {
#[serde(flatten)]
pub base: BaseLogEntry,
pub level: LogKind,
pub console_msg: String,
pub node_name: String,
#[serde(skip)]
pub err: Option<String>,
}
impl ConsoleLogEntry {
/// Create a new ConsoleLogEntry
pub fn new() -> Self {
ConsoleLogEntry {
base: BaseLogEntry::new(),
level: LogKind::Info,
console_msg: String::new(),
node_name: String::new(),
err: None,
}
}
/// Create a new ConsoleLogEntry with console message and node name
pub fn new_with_console_msg(console_msg: String, node_name: String) -> Self {
ConsoleLogEntry {
base: BaseLogEntry::new(),
level: LogKind::Info,
console_msg,
node_name,
err: None,
}
}
/// Set the base log entry
pub fn with_base(mut self, base: BaseLogEntry) -> Self {
self.base = base;
self
}
/// Set the log level
pub fn set_level(mut self, level: LogKind) -> Self {
self.level = level;
self
}
/// Set the node name
pub fn set_node_name(mut self, node_name: String) -> Self {
self.node_name = node_name;
self
}
/// Set the console message
pub fn set_console_msg(mut self, console_msg: String) -> Self {
self.console_msg = console_msg;
self
}
/// Set the error message
pub fn set_err(mut self, err: Option<String>) -> Self {
self.err = err;
self
}
}
impl LogRecord for ConsoleLogEntry {
fn to_json(&self) -> String {
serde_json::to_string(self).unwrap_or_else(|_| String::from("{}"))
}
fn get_timestamp(&self) -> DateTime<Utc> {
self.base.timestamp
}
}
/// Unified log entry type
/// UnifiedLogEntry is used to log different types of log entries
///
/// The `UnifiedLogEntry` enum contains the following variants:
/// - `Server` - a server log entry
/// - `Audit` - an audit log entry
/// - `Console` - a console log entry
///
/// The `UnifiedLogEntry` enum contains the following methods:
/// - `to_json` - convert the log entry to JSON
/// - `get_timestamp` - get the timestamp of the log entry
///
/// # Example
/// ```
/// use rustfs_obs::{UnifiedLogEntry, ServerLogEntry};
/// use tracing_core::Level;
///
/// let server_entry = ServerLogEntry::new(Level::INFO, "test_module".to_string());
/// let unified = UnifiedLogEntry::Server(server_entry);
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum UnifiedLogEntry {
#[serde(rename = "server")]
Server(ServerLogEntry),
#[serde(rename = "audit")]
Audit(AuditLogEntry),
#[serde(rename = "console")]
Console(ConsoleLogEntry),
}
impl LogRecord for UnifiedLogEntry {
fn to_json(&self) -> String {
match self {
UnifiedLogEntry::Server(entry) => entry.to_json(),
UnifiedLogEntry::Audit(entry) => entry.to_json(),
UnifiedLogEntry::Console(entry) => entry.to_json(),
}
}
fn get_timestamp(&self) -> DateTime<Utc> {
match self {
UnifiedLogEntry::Server(entry) => entry.get_timestamp(),
UnifiedLogEntry::Audit(entry) => entry.get_timestamp(),
UnifiedLogEntry::Console(entry) => entry.get_timestamp(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_base_log_entry() {
let base = BaseLogEntry::new()
.request_id(Some("req-123".to_string()))
.message(Some("Test message".to_string()));
assert_eq!(base.request_id, Some("req-123".to_string()));
assert_eq!(base.message, Some("Test message".to_string()));
}
#[test]
fn test_server_log_entry() {
let entry = ServerLogEntry::new(Level::INFO, "test_module".to_string())
.user_id(Some("user-456".to_string()))
.add_field("operation".to_string(), "login".to_string());
assert_eq!(entry.level.0, Level::INFO);
assert_eq!(entry.source, "test_module");
assert_eq!(entry.user_id, Some("user-456".to_string()));
assert_eq!(entry.fields.len(), 1);
assert_eq!(entry.fields[0], ("operation".to_string(), "login".to_string()));
}
#[test]
fn test_unified_log_entry_json() {
let server_entry = ServerLogEntry::new(Level::INFO, "test_source".to_string());
let unified = UnifiedLogEntry::Server(server_entry);
let json = unified.to_json();
assert!(json.contains("test_source"));
}
}
packages/obs/src/global.rs
+89
View File
@@ -0,0 +1,89 @@
use crate::telemetry::OtelGuard;
use std::sync::{Arc, Mutex};
use tokio::sync::{OnceCell, SetError};
use tracing::{error, info};
pub(crate) const USE_STDOUT: bool = true;
pub(crate) const SERVICE_NAME: &str = "RustFS";
pub(crate) const SAMPLE_RATIO: f64 = 1.0;
pub(crate) const METER_INTERVAL: u64 = 60;
pub(crate) const SERVICE_VERSION: &str = "0.1.0";
pub(crate) const ENVIRONMENT: &str = "production";
pub(crate) const LOGGER_LEVEL: &str = "info";
/// Global guard for OpenTelemetry tracing
static GLOBAL_GUARD: OnceCell<Arc<Mutex<OtelGuard>>> = OnceCell::const_new();
/// Error type for global guard operations
#[derive(Debug, thiserror::Error)]
pub enum GuardError {
#[error("Failed to set global guard: {0}")]
SetError(#[from] SetError<Arc<Mutex<OtelGuard>>>),
#[error("Global guard not initialized")]
NotInitialized,
}
/// Set the global guard for OpenTelemetry
///
/// # Arguments
/// * `guard` - The OtelGuard instance to set globally
///
/// # Returns
/// * `Ok(())` if successful
/// * `Err(GuardError)` if setting fails
///
/// # Example
/// ```rust
/// use rustfs_obs::{init_telemetry, load_config, set_global_guard};
///
/// async fn init() -> Result<(), Box<dyn std::error::Error>> {
/// let config = load_config(None);
/// let guard = init_telemetry(&config.observability).await?;
/// set_global_guard(guard)?;
/// Ok(())
/// }
/// ```
pub fn set_global_guard(guard: OtelGuard) -> Result<(), GuardError> {
info!("Initializing global OpenTelemetry guard");
GLOBAL_GUARD.set(Arc::new(Mutex::new(guard))).map_err(GuardError::SetError)
}
/// Get the global guard for OpenTelemetry
///
/// # Returns
/// * `Ok(Arc<Mutex<OtelGuard>>)` if guard exists
/// * `Err(GuardError)` if guard not initialized
///
/// # Example
/// ```rust
/// use rustfs_obs::get_global_guard;
///
/// async fn trace_operation() -> Result<(), Box<dyn std::error::Error>> {
/// let guard = get_global_guard()?;
/// let _lock = guard.lock().unwrap();
/// // Perform traced operation
/// Ok(())
/// }
/// ```
pub fn get_global_guard() -> Result<Arc<Mutex<OtelGuard>>, GuardError> {
GLOBAL_GUARD.get().cloned().ok_or(GuardError::NotInitialized)
}
/// Try to get the global guard for OpenTelemetry
///
/// # Returns
/// * `Some(Arc<Mutex<OtelGuard>>)` if guard exists
/// * `None` if guard not initialized
pub fn try_get_global_guard() -> Option<Arc<Mutex<OtelGuard>>> {
GLOBAL_GUARD.get().cloned()
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_get_uninitialized_guard() {
let result = get_global_guard();
assert!(matches!(result, Err(GuardError::NotInitialized)));
}
}
packages/obs/src/lib.rs
+94
View File
@@ -0,0 +1,94 @@
/// # obs
///
/// `obs` is a logging and observability library for Rust.
/// It provides a simple and easy-to-use interface for logging and observability.
/// It is built on top of the `log` crate and `opentelemetry` crate.
///
/// ## Features
/// - Structured logging
/// - Distributed tracing
/// - Metrics collection
/// - Log processing worker
/// - Multiple sinks
/// - Configuration-based setup
/// - Telemetry guard
/// - Global logger
/// - Log levels
/// - Log entry types
/// - Log record
/// - Object version
/// - Local IP address
///
/// ## Usage
///
/// ```rust
/// use rustfs_obs::{AppConfig, init_obs};
///
/// let config = AppConfig::default();
/// let (logger, guard) = init_obs(config);
/// ```
mod config;
mod entry;
mod global;
mod logger;
mod sink;
mod telemetry;
mod utils;
mod worker;
pub use config::load_config;
pub use config::{AppConfig, OtelConfig};
pub use entry::args::Args;
pub use entry::audit::{ApiDetails, AuditLogEntry};
pub use entry::base::BaseLogEntry;
pub use entry::unified::{ConsoleLogEntry, ServerLogEntry, UnifiedLogEntry};
pub use entry::{LogKind, LogRecord, ObjectVersion, SerializableLevel};
pub use global::{get_global_guard, set_global_guard, try_get_global_guard, GuardError};
pub use logger::{ensure_logger_initialized, log_debug, log_error, log_info, log_trace, log_warn, log_with_context};
pub use logger::{get_global_logger, init_global_logger, locked_logger, start_logger};
pub use logger::{log_init_state, InitLogStatus};
pub use logger::{LogError, Logger};
pub use sink::Sink;
use std::sync::Arc;
pub use telemetry::init_telemetry;
pub use telemetry::{get_global_registry, metrics};
use tokio::sync::Mutex;
pub use utils::{get_local_ip, get_local_ip_with_default};
pub use worker::start_worker;
/// Initialize the observability module
///
/// # Parameters
/// - `config`: Configuration information
///
/// # Returns
/// A tuple containing the logger and the telemetry guard
///
/// # Example
/// ```
/// use rustfs_obs::{AppConfig, init_obs};
///
/// let config = AppConfig::default();
/// let (logger, guard) = init_obs(config);
/// ```
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 logger = init_global_logger(&config, sinks).await;
(logger, guard)
}
/// Get the global logger instance
/// This function returns a reference to the global logger instance.
///
/// # Returns
/// A reference to the global logger instance
///
/// # Example
/// ```
/// use rustfs_obs::get_logger;
///
/// let logger = get_logger();
/// ```
pub fn get_logger() -> &'static Arc<Mutex<Logger>> {
get_global_logger()
}
packages/obs/src/logger.rs
+534
View File
@@ -0,0 +1,534 @@
use crate::global::{ENVIRONMENT, SERVICE_NAME, SERVICE_VERSION};
use crate::{AppConfig, AuditLogEntry, BaseLogEntry, ConsoleLogEntry, OtelConfig, ServerLogEntry, Sink, UnifiedLogEntry};
use std::sync::Arc;
use std::time::SystemTime;
use tokio::sync::mpsc::{self, Receiver, Sender};
use tokio::sync::{Mutex, OnceCell};
use tracing_core::Level;
// Add the global instance at the module level
static GLOBAL_LOGGER: OnceCell<Arc<Mutex<Logger>>> = OnceCell::const_new();
/// Server log processor
#[derive(Debug)]
pub struct Logger {
sender: Sender<UnifiedLogEntry>, // Log sending channel
queue_capacity: usize,
}
impl Logger {
/// Create a new Logger instance
/// Returns Logger and corresponding Receiver
pub fn new(config: &AppConfig) -> (Self, Receiver<UnifiedLogEntry>) {
// Get queue capacity from configuration, or use default values 10000
let queue_capacity = config.logger.queue_capacity.unwrap_or(10000);
let (sender, receiver) = mpsc::channel(queue_capacity);
(Logger { sender, queue_capacity }, receiver)
}
/// get the queue capacity
/// This function returns the queue capacity.
/// # Returns
/// The queue capacity
/// # Example
/// ```
/// use rustfs_obs::Logger;
/// async fn example(logger: &Logger) {
/// let _ = logger.get_queue_capacity();
/// }
/// ```
pub fn get_queue_capacity(&self) -> usize {
self.queue_capacity
}
/// Log a server entry
#[tracing::instrument(skip(self), fields(log_source = "logger_server"))]
pub async fn log_server_entry(&self, entry: ServerLogEntry) -> Result<(), LogError> {
self.log_entry(UnifiedLogEntry::Server(entry)).await
}
/// Log an audit entry
#[tracing::instrument(skip(self), fields(log_source = "logger_audit"))]
pub async fn log_audit_entry(&self, entry: AuditLogEntry) -> Result<(), LogError> {
self.log_entry(UnifiedLogEntry::Audit(entry)).await
}
/// Log a console entry
#[tracing::instrument(skip(self), fields(log_source = "logger_console"))]
pub async fn log_console_entry(&self, entry: ConsoleLogEntry) -> Result<(), LogError> {
self.log_entry(UnifiedLogEntry::Console(entry)).await
}
/// Asynchronous logging of unified log entries
#[tracing::instrument(skip(self), fields(log_source = "logger"))]
pub async fn log_entry(&self, entry: UnifiedLogEntry) -> Result<(), LogError> {
// Extract information for tracing based on entry type
match &entry {
UnifiedLogEntry::Server(server) => {
tracing::Span::current()
.record("log_level", &server.level.0.as_str())
.record("log_message", &server.base.message.as_deref().unwrap_or(""))
.record("source", &server.source);
// Generate tracing event based on log level
match server.level.0 {
Level::ERROR => {
tracing::error!(target: "server_logs", message = %server.base.message.as_deref().unwrap_or(""));
}
Level::WARN => {
tracing::warn!(target: "server_logs", message = %server.base.message.as_deref().unwrap_or(""));
}
Level::INFO => {
tracing::info!(target: "server_logs", message = %server.base.message.as_deref().unwrap_or(""));
}
Level::DEBUG => {
tracing::debug!(target: "server_logs", message = %server.base.message.as_deref().unwrap_or(""));
}
Level::TRACE => {
tracing::trace!(target: "server_logs", message = %server.base.message.as_deref().unwrap_or(""));
}
}
}
UnifiedLogEntry::Audit(audit) => {
tracing::info!(
target: "audit_logs",
event = %audit.event,
api = %audit.api.name.as_deref().unwrap_or("unknown"),
message = %audit.base.message.as_deref().unwrap_or("")
);
}
UnifiedLogEntry::Console(console) => {
let level_str = match console.level {
crate::LogKind::Info => "INFO",
crate::LogKind::Warning => "WARN",
crate::LogKind::Error => "ERROR",
crate::LogKind::Fatal => "FATAL",
};
tracing::info!(
target: "console_logs",
level = %level_str,
node = %console.node_name,
message = %console.console_msg
);
}
}
// Send logs to async queue with improved error handling
match self.sender.try_send(entry) {
Ok(_) => Ok(()),
Err(mpsc::error::TrySendError::Full(entry)) => {
// Processing strategy when queue is full
tracing::warn!("Log queue full, applying backpressure");
match tokio::time::timeout(std::time::Duration::from_millis(500), self.sender.send(entry)).await {
Ok(Ok(_)) => Ok(()),
Ok(Err(_)) => Err(LogError::SendFailed("Channel closed")),
Err(_) => Err(LogError::Timeout("Queue backpressure timeout")),
}
}
Err(mpsc::error::TrySendError::Closed(_)) => Err(LogError::SendFailed("Logger channel closed")),
}
}
/// Write log with context information
/// This function writes log messages with context information.
///
/// # Parameters
/// - `message`: Message to be logged
/// - `source`: Source of the log
/// - `request_id`: Request ID
/// - `user_id`: User ID
/// - `fields`: Additional fields
///
/// # Returns
/// Result indicating whether the operation was successful
///
/// # Example
/// ```
/// use tracing_core::Level;
/// use rustfs_obs::Logger;
///
/// async fn example(logger: &Logger) {
/// let _ = logger.write_with_context("This is an information message", "example",Level::INFO, Some("req-12345".to_string()), Some("user-6789".to_string()), vec![("endpoint".to_string(), "/api/v1/data".to_string())]).await;
/// }
pub async fn write_with_context(
&self,
message: &str,
source: &str,
level: Level,
request_id: Option<String>,
user_id: Option<String>,
fields: Vec<(String, String)>,
) -> Result<(), LogError> {
let base = BaseLogEntry::new().message(Some(message.to_string())).request_id(request_id);
let server_entry = ServerLogEntry::new(level, source.to_string())
.user_id(user_id)
.fields(fields)
.with_base(base);
self.log_server_entry(server_entry).await
}
/// Write log
/// This function writes log messages.
/// # Parameters
/// - `message`: Message to be logged
/// - `source`: Source of the log
/// - `level`: Log level
///
/// # Returns
/// Result indicating whether the operation was successful
///
/// # Example
/// ```
/// use rustfs_obs::Logger;
/// use tracing_core::Level;
///
/// async fn example(logger: &Logger) {
/// let _ = logger.write("This is an information message", "example", Level::INFO).await;
/// }
/// ```
pub async fn write(&self, message: &str, source: &str, level: Level) -> Result<(), LogError> {
self.write_with_context(message, source, level, None, None, Vec::new()).await
}
/// Shutdown the logger
/// This function shuts down the logger.
///
/// # Returns
/// Result indicating whether the operation was successful
///
/// # Example
/// ```
/// use rustfs_obs::Logger;
///
/// async fn example(logger: Logger) {
/// let _ = logger.shutdown().await;
/// }
/// ```
pub async fn shutdown(self) -> Result<(), LogError> {
drop(self.sender); //Close the sending end so that the receiver knows that there is no new message
Ok(())
}
}
/// Log error type
/// This enum defines the error types that can occur when logging.
/// It is used to provide more detailed error information.
/// # Example
/// ```
/// use rustfs_obs::LogError;
/// use thiserror::Error;
///
/// LogError::SendFailed("Failed to send log");
/// LogError::Timeout("Operation timed out");
/// ```
#[derive(Debug, thiserror::Error)]
pub enum LogError {
#[error("Failed to send log: {0}")]
SendFailed(&'static str),
#[error("Operation timed out: {0}")]
Timeout(&'static str),
}
/// Start the log module
/// This function starts the log module.
/// It initializes the logger and starts the worker to process logs.
/// # Parameters
/// - `config`: Configuration information
/// - `sinks`: A vector of Sink instances
/// # Returns
/// The global logger instance
/// # Example
/// ```
/// use rustfs_obs::{AppConfig, start_logger};
///
/// let config = AppConfig::default();
/// let sinks = vec![];
/// let logger = start_logger(&config, sinks);
/// ```
pub fn start_logger(config: &AppConfig, sinks: Vec<Arc<dyn Sink>>) -> Logger {
let (logger, receiver) = Logger::new(config);
tokio::spawn(crate::worker::start_worker(receiver, sinks));
logger
}
/// Initialize the global logger instance
/// This function initializes the global logger instance and returns a reference to it.
/// If the logger has been initialized before, it will return the existing logger instance.
///
/// # Parameters
/// - `config`: Configuration information
/// - `sinks`: A vector of Sink instances
///
/// # Returns
/// A reference to the global logger instance
///
/// # Example
/// ```
/// use rustfs_obs::{AppConfig,init_global_logger};
///
/// let config = AppConfig::default();
/// let sinks = vec![];
/// let logger = init_global_logger(&config, sinks);
/// ```
pub async fn init_global_logger(config: &AppConfig, sinks: Vec<Arc<dyn Sink>>) -> Arc<Mutex<Logger>> {
let logger = Arc::new(Mutex::new(start_logger(config, sinks)));
GLOBAL_LOGGER.set(logger.clone()).expect("Logger already initialized");
logger
}
/// Get the global logger instance
///
/// This function returns a reference to the global logger instance.
///
/// # Returns
/// A reference to the global logger instance
///
/// # Example
/// ```
/// use rustfs_obs::get_global_logger;
///
/// let logger = get_global_logger();
/// ```
pub fn get_global_logger() -> &'static Arc<Mutex<Logger>> {
GLOBAL_LOGGER.get().expect("Logger not initialized")
}
/// Get the global logger instance with a lock
/// This function returns a reference to the global logger instance with a lock.
/// It is used to ensure that the logger is thread-safe.
///
/// # Returns
/// A reference to the global logger instance with a lock
///
/// # Example
/// ```
/// use rustfs_obs::locked_logger;
///
/// async fn example() {
/// let logger = locked_logger().await;
/// }
/// ```
pub async fn locked_logger() -> tokio::sync::MutexGuard<'static, Logger> {
get_global_logger().lock().await
}
/// Initialize with default empty logger if needed (optional)
/// This function initializes the logger with a default empty logger if needed.
/// It is used to ensure that the logger is initialized before logging.
///
/// # Returns
/// A reference to the global logger instance
///
/// # Example
/// ```
/// use rustfs_obs::ensure_logger_initialized;
///
/// let logger = ensure_logger_initialized();
/// ```
pub fn ensure_logger_initialized() -> &'static Arc<Mutex<Logger>> {
if GLOBAL_LOGGER.get().is_none() {
let config = AppConfig::default();
let sinks = vec![];
let logger = Arc::new(Mutex::new(start_logger(&config, sinks)));
let _ = GLOBAL_LOGGER.set(logger);
}
GLOBAL_LOGGER.get().unwrap()
}
/// Log information
/// This function logs information messages.
///
/// # Parameters
/// - `message`: Message to be logged
/// - `source`: Source of the log
///
/// # Returns
/// Result indicating whether the operation was successful
///
/// # Example
/// ```
/// use rustfs_obs::log_info;
///
/// async fn example() {
/// let _ = log_info("This is an information message", "example").await;
/// }
/// ```
pub async fn log_info(message: &str, source: &str) -> Result<(), LogError> {
get_global_logger().lock().await.write(message, source, Level::INFO).await
}
/// Log error
/// This function logs error messages.
/// # Parameters
/// - `message`: Message to be logged
/// - `source`: Source of the log
/// # Returns
/// Result indicating whether the operation was successful
/// # Example
/// ```
/// use rustfs_obs::log_error;
///
/// async fn example() {
/// let _ = log_error("This is an error message", "example").await;
/// }
pub async fn log_error(message: &str, source: &str) -> Result<(), LogError> {
get_global_logger().lock().await.write(message, source, Level::ERROR).await
}
/// Log warning
/// This function logs warning messages.
/// # Parameters
/// - `message`: Message to be logged
/// - `source`: Source of the log
/// # Returns
/// Result indicating whether the operation was successful
///
/// # Example
/// ```
/// use rustfs_obs::log_warn;
///
/// async fn example() {
/// let _ = log_warn("This is a warning message", "example").await;
/// }
/// ```
pub async fn log_warn(message: &str, source: &str) -> Result<(), LogError> {
get_global_logger().lock().await.write(message, source, Level::WARN).await
}
/// Log debug
/// This function logs debug messages.
/// # Parameters
/// - `message`: Message to be logged
/// - `source`: Source of the log
/// # Returns
/// Result indicating whether the operation was successful
///
/// # Example
/// ```
/// use rustfs_obs::log_debug;
///
/// async fn example() {
/// let _ = log_debug("This is a debug message", "example").await;
/// }
/// ```
pub async fn log_debug(message: &str, source: &str) -> Result<(), LogError> {
get_global_logger().lock().await.write(message, source, Level::DEBUG).await
}
/// Log trace
/// This function logs trace messages.
/// # Parameters
/// - `message`: Message to be logged
/// - `source`: Source of the log
///
/// # Returns
/// Result indicating whether the operation was successful
///
/// # Example
/// ```
/// use rustfs_obs::log_trace;
///
/// async fn example() {
/// let _ = log_trace("This is a trace message", "example").await;
/// }
/// ```
pub async fn log_trace(message: &str, source: &str) -> Result<(), LogError> {
get_global_logger().lock().await.write(message, source, Level::TRACE).await
}
/// Log with context information
/// This function logs messages with context information.
/// # Parameters
/// - `message`: Message to be logged
/// - `source`: Source of the log
/// - `level`: Log level
/// - `request_id`: Request ID
/// - `user_id`: User ID
/// - `fields`: Additional fields
/// # Returns
/// Result indicating whether the operation was successful
/// # Example
/// ```
/// use tracing_core::Level;
/// use rustfs_obs::log_with_context;
///
/// async fn example() {
/// let _ = log_with_context("This is an information message", "example", Level::INFO, Some("req-12345".to_string()), Some("user-6789".to_string()), vec![("endpoint".to_string(), "/api/v1/data".to_string())]).await;
/// }
/// ```
pub async fn log_with_context(
message: &str,
source: &str,
level: Level,
request_id: Option<String>,
user_id: Option<String>,
fields: Vec<(String, String)>,
) -> Result<(), LogError> {
get_global_logger()
.lock()
.await
.write_with_context(message, source, level, request_id, user_id, fields)
.await
}
/// Log initialization status
#[derive(Debug)]
pub struct InitLogStatus {
pub timestamp: SystemTime,
pub service_name: String,
pub version: String,
pub environment: String,
}
impl Default for InitLogStatus {
fn default() -> Self {
Self {
timestamp: SystemTime::now(),
service_name: String::from(SERVICE_NAME),
version: SERVICE_VERSION.to_string(),
environment: ENVIRONMENT.to_string(),
}
}
}
impl InitLogStatus {
pub fn new_config(config: &OtelConfig) -> Self {
let config = config.clone();
let environment = config.environment.unwrap_or(ENVIRONMENT.to_string());
let version = config.service_version.unwrap_or(SERVICE_VERSION.to_string());
Self {
timestamp: SystemTime::now(),
service_name: String::from(SERVICE_NAME),
version,
environment,
}
}
pub async fn init_start_log(config: &OtelConfig) -> Result<(), LogError> {
let status = Self::new_config(config);
log_init_state(Some(status)).await
}
}
/// Log initialization details during system startup
pub async fn log_init_state(status: Option<InitLogStatus>) -> Result<(), LogError> {
let status = status.unwrap_or_default();
let base_entry = BaseLogEntry::new()
.timestamp(chrono::DateTime::from(status.timestamp))
.message(Some(format!(
"Service initialization started - {} v{} in {}",
status.service_name, status.version, status.environment
)))
.request_id(Some("system_init".to_string()));
let server_entry = ServerLogEntry::new(Level::INFO, "system_initialization".to_string())
.with_base(base_entry)
.user_id(Some("system".to_string()));
get_global_logger().lock().await.log_server_entry(server_entry).await?;
Ok(())
}
packages/obs/src/sink.rs
+466
View File
@@ -0,0 +1,466 @@
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;
use tracing::debug;
/// 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?;
debug!("the file not exists,create if. path: {:?}", path)
}
let file = if file_exists {
// If the file exists, open it in append mode
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
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).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")]
if config.sinks.kafka.enabled {
match rdkafka::config::ClientConfig::new()
.set("bootstrap.servers", &config.sinks.kafka.bootstrap_servers)
.set("message.timeout.ms", "5000")
.create()
{
Ok(producer) => {
sinks.push(Arc::new(KafkaSink::new(
producer,
config.sinks.kafka.topic.clone(),
config.sinks.kafka.batch_size.unwrap_or(100),
config.sinks.kafka.batch_timeout_ms.unwrap_or(1000),
)));
}
Err(e) => eprintln!("Failed to create Kafka producer: {}", e),
}
}
#[cfg(feature = "webhook")]
if config.sinks.webhook.enabled {
sinks.push(Arc::new(WebhookSink::new(
config.sinks.webhook.endpoint.clone(),
config.sinks.webhook.auth_token.clone(),
config.sinks.webhook.max_retries.unwrap_or(3),
config.sinks.webhook.retry_delay_ms.unwrap_or(100),
)));
}
#[cfg(feature = "file")]
{
let path = if config.sinks.file.enabled {
config.sinks.file.path.clone()
} else {
"default.log".to_string()
};
debug!("FileSink: Using path: {}", path);
sinks.push(Arc::new(
FileSink::new(
path.clone(),
config.sinks.file.buffer_size.unwrap_or(8192),
config.sinks.file.flush_interval_ms.unwrap_or(1000),
config.sinks.file.flush_threshold.unwrap_or(100),
)
.await
.unwrap(),
));
}
sinks
}
packages/obs/src/telemetry.rs
+316
View File
@@ -0,0 +1,316 @@
use crate::global::{ENVIRONMENT, LOGGER_LEVEL, METER_INTERVAL, SAMPLE_RATIO, SERVICE_NAME, SERVICE_VERSION, USE_STDOUT};
use crate::{get_local_ip_with_default, OtelConfig};
use opentelemetry::trace::TracerProvider;
use opentelemetry::{global, KeyValue};
use opentelemetry_appender_tracing::layer;
use opentelemetry_otlp::WithExportConfig;
use opentelemetry_sdk::logs::SdkLoggerProvider;
use opentelemetry_sdk::{
metrics::{MeterProviderBuilder, PeriodicReader, SdkMeterProvider},
trace::{RandomIdGenerator, Sampler, SdkTracerProvider},
Resource,
};
use opentelemetry_semantic_conventions::{
attribute::{DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_VERSION as OTEL_SERVICE_VERSION},
SCHEMA_URL,
};
use prometheus::{Encoder, Registry, TextEncoder};
use std::io::IsTerminal;
use std::sync::Arc;
use tokio::sync::{Mutex, OnceCell};
use tracing::{info, warn};
use tracing_error::ErrorLayer;
use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer};
/// A guard object that manages the lifecycle of OpenTelemetry components.
///
/// This struct holds references to the created OpenTelemetry providers and ensures
/// they are properly shut down when the guard is dropped. It implements the RAII
/// (Resource Acquisition Is Initialization) pattern for managing telemetry resources.
///
/// When this guard goes out of scope, it will automatically shut down:
/// - The tracer provider (for distributed tracing)
/// - The meter provider (for metrics collection)
/// - The logger provider (for structured logging)
///
/// # Example
///
/// ```
/// use rustfs_obs::{init_telemetry, OtelConfig};
///
/// let config = OtelConfig::default();
/// let otel_guard = init_telemetry(&config);
///
/// // The guard is kept alive for the duration of the application
/// // When it's dropped, all telemetry components are properly shut down
/// drop(otel_guard);
/// ```
#[derive(Debug)]
pub struct OtelGuard {
tracer_provider: SdkTracerProvider,
meter_provider: SdkMeterProvider,
logger_provider: SdkLoggerProvider,
}
impl Drop for OtelGuard {
fn drop(&mut self) {
if let Err(err) = self.tracer_provider.shutdown() {
eprintln!("Tracer shutdown error: {:?}", err);
}
if let Err(err) = self.meter_provider.shutdown() {
eprintln!("Meter shutdown error: {:?}", err);
}
if let Err(err) = self.logger_provider.shutdown() {
eprintln!("Logger shutdown error: {:?}", err);
}
}
}
/// Global registry for Prometheus metrics
static GLOBAL_REGISTRY: OnceCell<Arc<Mutex<Registry>>> = OnceCell::const_new();
/// Get the global registry instance
/// This function returns a reference to the global registry instance.
///
/// # Returns
/// A reference to the global registry instance
///
/// # Example
/// ```
/// use rustfs_obs::get_global_registry;
///
/// let registry = get_global_registry();
/// ```
pub fn get_global_registry() -> Arc<Mutex<Registry>> {
GLOBAL_REGISTRY.get().unwrap().clone()
}
/// Prometheus metric endpoints
/// This function returns a string containing the Prometheus metrics.
/// The metrics are collected from the global registry.
/// The function is used to expose the metrics via an HTTP endpoint.
///
/// # Returns
/// A string containing the Prometheus metrics
///
/// # Example
/// ```
/// use rustfs_obs::metrics;
///
/// async fn main() {
/// let metrics = metrics().await;
/// println!("{}", metrics);
/// }
/// ```
pub async fn metrics() -> String {
let encoder = TextEncoder::new();
// Get a reference to the registry for reading metrics
let registry = get_global_registry().lock().await.to_owned();
let metric_families = registry.gather();
if metric_families.is_empty() {
warn!("No metrics available in Prometheus registry");
} else {
info!("Metrics collected: {} families", metric_families.len());
}
let mut buffer = Vec::new();
encoder.encode(&metric_families, &mut buffer).unwrap();
String::from_utf8(buffer).unwrap_or_else(|_| "Error encoding metrics".to_string())
}
/// create OpenTelemetry Resource
fn resource(config: &OtelConfig) -> Resource {
let config = config.clone();
Resource::builder()
.with_service_name(config.service_name.unwrap_or(SERVICE_NAME.to_string()))
.with_schema_url(
[
KeyValue::new(OTEL_SERVICE_VERSION, config.service_version.unwrap_or(SERVICE_VERSION.to_string())),
KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, config.environment.unwrap_or(ENVIRONMENT.to_string())),
KeyValue::new(NETWORK_LOCAL_ADDRESS, get_local_ip_with_default()),
],
SCHEMA_URL,
)
.build()
}
/// Creates a periodic reader for stdout metrics
fn create_periodic_reader(interval: u64) -> PeriodicReader<opentelemetry_stdout::MetricExporter> {
PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default())
.with_interval(std::time::Duration::from_secs(interval))
.build()
}
/// Initialize Meter Provider
fn init_meter_provider(config: &OtelConfig) -> SdkMeterProvider {
let mut builder = MeterProviderBuilder::default().with_resource(resource(config));
// If endpoint is empty, use stdout output
if config.endpoint.is_empty() {
builder = builder.with_reader(create_periodic_reader(config.meter_interval.unwrap_or(METER_INTERVAL)));
} else {
// If endpoint is not empty, use otlp output
let exporter = opentelemetry_otlp::MetricExporter::builder()
.with_tonic()
.with_endpoint(&config.endpoint)
.with_temporality(opentelemetry_sdk::metrics::Temporality::default())
.build()
.unwrap();
builder = builder.with_reader(
PeriodicReader::builder(exporter)
.with_interval(std::time::Duration::from_secs(config.meter_interval.unwrap_or(METER_INTERVAL)))
.build(),
);
// If use_stdout is true, output to stdout at the same time
if config.use_stdout.unwrap_or(USE_STDOUT) {
builder = builder.with_reader(create_periodic_reader(config.meter_interval.unwrap_or(METER_INTERVAL)));
}
}
let registry = Registry::new();
// Set global registry
GLOBAL_REGISTRY.set(Arc::new(Mutex::new(registry.clone()))).unwrap();
// Create Prometheus exporter
let prometheus_exporter = opentelemetry_prometheus::exporter().with_registry(registry).build().unwrap();
// Build meter provider
let meter_provider = builder.with_reader(prometheus_exporter).build();
global::set_meter_provider(meter_provider.clone());
meter_provider
}
/// Initialize Tracer Provider
fn init_tracer_provider(config: &OtelConfig) -> SdkTracerProvider {
let sample_ratio = config.sample_ratio.unwrap_or(SAMPLE_RATIO);
let sampler = if sample_ratio > 0.0 && sample_ratio < 1.0 {
Sampler::TraceIdRatioBased(sample_ratio)
} else {
Sampler::AlwaysOn
};
let builder = SdkTracerProvider::builder()
.with_sampler(sampler)
.with_id_generator(RandomIdGenerator::default())
.with_resource(resource(config));
let tracer_provider = if config.endpoint.is_empty() {
builder
.with_batch_exporter(opentelemetry_stdout::SpanExporter::default())
.build()
} else {
let exporter = opentelemetry_otlp::SpanExporter::builder()
.with_tonic()
.with_endpoint(&config.endpoint)
.build()
.unwrap();
if config.use_stdout.unwrap_or(USE_STDOUT) {
builder
.with_batch_exporter(exporter)
.with_batch_exporter(opentelemetry_stdout::SpanExporter::default())
} else {
builder.with_batch_exporter(exporter)
}
.build()
};
global::set_tracer_provider(tracer_provider.clone());
tracer_provider
}
/// Initialize Telemetry
pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
let tracer_provider = init_tracer_provider(config);
let meter_provider = init_meter_provider(config);
// Initialize logger provider based on configuration
let logger_provider = {
let mut builder = SdkLoggerProvider::builder().with_resource(resource(config));
if config.endpoint.is_empty() {
// Use stdout exporter when no endpoint is configured
builder = builder.with_simple_exporter(opentelemetry_stdout::LogExporter::default());
} else {
// Configure OTLP exporter when endpoint is provided
let exporter = opentelemetry_otlp::LogExporter::builder()
.with_tonic()
.with_endpoint(&config.endpoint)
.build()
.unwrap();
builder = builder.with_batch_exporter(exporter);
// Add stdout exporter if requested
if config.use_stdout.unwrap_or(USE_STDOUT) {
builder = builder.with_batch_exporter(opentelemetry_stdout::LogExporter::default());
}
}
builder.build()
};
let config = config.clone();
let logger_level = config.logger_level.unwrap_or(LOGGER_LEVEL.to_string());
let logger_level = logger_level.as_str();
// Setup OpenTelemetryTracingBridge layer
let otel_layer = {
// Filter to prevent infinite telemetry loops
// This blocks events from OpenTelemetry and its dependent libraries (tonic, reqwest, etc.)
// from being sent back to OpenTelemetry itself
let filter_otel = match logger_level {
"trace" | "debug" => {
info!("OpenTelemetry tracing initialized with level: {}", logger_level);
EnvFilter::new(logger_level)
}
_ => {
let mut filter = EnvFilter::new(logger_level);
for directive in ["hyper", "opentelemetry", "tonic", "h2", "reqwest"] {
filter = filter.add_directive(format!("{}=off", directive).parse().unwrap());
}
filter
}
};
layer::OpenTelemetryTracingBridge::new(&logger_provider).with_filter(filter_otel)
};
let tracer = tracer_provider.tracer(config.service_name.unwrap_or(SERVICE_NAME.to_string()));
let registry = tracing_subscriber::registry()
.with(switch_level(logger_level))
.with(OpenTelemetryLayer::new(tracer))
.with(MetricsLayer::new(meter_provider.clone()))
.with(otel_layer);
// Configure formatting layer
let enable_color = std::io::stdout().is_terminal();
let fmt_layer = tracing_subscriber::fmt::layer()
.with_ansi(enable_color)
.with_thread_names(true)
.with_file(true)
.with_line_number(true);
// Creating a formatting layer with explicit type to avoid type mismatches
let fmt_layer = fmt_layer.with_filter(
EnvFilter::new(logger_level).add_directive(
format!("opentelemetry={}", if config.endpoint.is_empty() { logger_level } else { "off" })
.parse()
.unwrap(),
),
);
registry.with(ErrorLayer::default()).with(fmt_layer).init();
if !config.endpoint.is_empty() {
info!("OpenTelemetry telemetry initialized with OTLP endpoint: {}", config.endpoint);
}
OtelGuard {
tracer_provider,
meter_provider,
logger_provider,
}
}
/// Switch log level
fn switch_level(logger_level: &str) -> tracing_subscriber::filter::LevelFilter {
match logger_level {
"error" => tracing_subscriber::filter::LevelFilter::ERROR,
"warn" => tracing_subscriber::filter::LevelFilter::WARN,
"info" => tracing_subscriber::filter::LevelFilter::INFO,
"debug" => tracing_subscriber::filter::LevelFilter::DEBUG,
"trace" => tracing_subscriber::filter::LevelFilter::TRACE,
_ => tracing_subscriber::filter::LevelFilter::OFF,
}
}
packages/obs/src/utils.rs
+42
View File
@@ -0,0 +1,42 @@
use local_ip_address::{local_ip, local_ipv6};
use std::net::{IpAddr, Ipv4Addr};
/// Get the IP address of the machine
///
/// Priority is given to trying to get the IPv4 address, and if it fails, try to get the IPv6 address.
/// If both fail to retrieve, None is returned.
///
/// # Returns
///
/// * `Some(IpAddr)` - Native IP address (IPv4 or IPv6)
/// * `None` - Unable to obtain any native IP address
pub fn get_local_ip() -> Option<IpAddr> {
local_ip().ok().or_else(|| local_ipv6().ok())
}
/// Get the IP address of the machine as a string
///
/// If the IP address cannot be obtained, returns "127.0.0.1" as the default value.
///
/// # Returns
///
/// * `String` - Native IP address (IPv4 or IPv6) as a string, or the default value
pub fn get_local_ip_with_default() -> String {
get_local_ip()
.unwrap_or_else(|| IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))) // Provide a safe default value
.to_string()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_get_local_ip() {
match get_local_ip() {
Some(ip) => println!("the ip address of this machine:{}", ip),
None => println!("Unable to obtain the IP address of the machine"),
}
assert!(get_local_ip().is_some());
}
}
packages/obs/src/worker.rs
+13
View File
@@ -0,0 +1,13 @@
use crate::{sink::Sink, UnifiedLogEntry};
use std::sync::Arc;
use tokio::sync::mpsc::Receiver;
/// Start the log processing worker thread
pub async fn start_worker(receiver: Receiver<UnifiedLogEntry>, sinks: Vec<Arc<dyn Sink>>) {
let mut receiver = receiver;
while let Some(entry) = receiver.recv().await {
for sink in &sinks {
sink.write(&entry).await;
}
}
}
rustfs/Cargo.toml
+4 -1
View File
@@ -27,7 +27,7 @@ policy.workspace = true
flatbuffers.workspace = true flatbuffers.workspace = true
futures.workspace = true futures.workspace = true
futures-util.workspace = true futures-util.workspace = true
h2 = "0.4.7" #h2 = "0.4.7"
hyper.workspace = true hyper.workspace = true
hyper-util.workspace = true hyper-util.workspace = true
http.workspace = true http.workspace = true
@@ -35,6 +35,7 @@ http-body.workspace = true
lock.workspace = true lock.workspace = true
mime.workspace = true mime.workspace = true
netif.workspace = true netif.workspace = true
once_cell.workspace = true
pin-project-lite.workspace = true pin-project-lite.workspace = true
prost.workspace = true prost.workspace = true
prost-types.workspace = true prost-types.workspace = true
@@ -62,6 +63,7 @@ tokio-stream.workspace = true
tonic.workspace = true tonic.workspace = true
tonic-reflection.workspace = true tonic-reflection.workspace = true
tower.workspace = true tower.workspace = true
tracing-core = { workspace = true }
tracing-error.workspace = true tracing-error.workspace = true
tracing-subscriber.workspace = true tracing-subscriber.workspace = true
transform-stream.workspace = true transform-stream.workspace = true
@@ -84,6 +86,7 @@ mime_guess = "2.0.5"
rust-embed = { workspace = true, features = ["interpolate-folder-path"] } rust-embed = { workspace = true, features = ["interpolate-folder-path"] }
local-ip-address = { workspace = true } local-ip-address = { workspace = true }
chrono = { workspace = true } chrono = { workspace = true }
rustfs-obs = { workspace = true }
[build-dependencies] [build-dependencies]
prost-build.workspace = true prost-build.workspace = true
rustfs/src/config/mod.rs
+27 -1
View File
@@ -1,11 +1,32 @@
use clap::Parser; use clap::Parser;
use const_str::concat; use const_str::concat;
use ecstore::global::DEFAULT_PORT; use ecstore::global::DEFAULT_PORT;
use std::string::ToString;
shadow_rs::shadow!(build); shadow_rs::shadow!(build);
/// Default Access Key
/// Default value: rustfsadmin
/// Environment variable: RUSTFS_ACCESS_KEY
/// Command line argument: --access-key
/// Example: RUSTFS_ACCESS_KEY=rustfsadmin
/// Example: --access-key rustfsadmin
pub const DEFAULT_ACCESS_KEY: &str = "rustfsadmin"; pub const DEFAULT_ACCESS_KEY: &str = "rustfsadmin";
/// Default Secret Key
/// Default value: rustfsadmin
/// Environment variable: RUSTFS_SECRET_KEY
/// Command line argument: --secret-key
/// Example: RUSTFS_SECRET_KEY=rustfsadmin
/// Example: --secret-key rustfsadmin
pub const DEFAULT_SECRET_KEY: &str = "rustfsadmin"; pub const DEFAULT_SECRET_KEY: &str = "rustfsadmin";
/// Default configuration file for observability
/// Default value: config/obs.toml
/// Environment variable: RUSTFS_OBS_CONFIG
/// Command line argument: --obs-config
/// Example: RUSTFS_OBS_CONFIG=config/obs.toml
/// Example: --obs-config config/obs.toml
/// Example: --obs-config /etc/rustfs/obs.toml
pub const DEFAULT_OBS_CONFIG: &str = "config/obs.toml";
/// Default TLS key for rustfs /// Default TLS key for rustfs
/// This is the default key for TLS. /// This is the default key for TLS.
@@ -39,7 +60,7 @@ const LONG_VERSION: &str = concat!(
concat!("git status :\n", build::GIT_STATUS_FILE), concat!("git status :\n", build::GIT_STATUS_FILE),
); );
#[derive(Debug, Parser)] #[derive(Debug, Parser, Clone)]
#[command(version = SHORT_VERSION, long_version = LONG_VERSION)] #[command(version = SHORT_VERSION, long_version = LONG_VERSION)]
pub struct Opt { pub struct Opt {
/// DIR points to a directory on a filesystem. /// DIR points to a directory on a filesystem.
@@ -71,6 +92,11 @@ pub struct Opt {
#[arg(long, default_value_t = format!("127.0.0.1:{}", 9002), env = "RUSTFS_CONSOLE_ADDRESS")] #[arg(long, default_value_t = format!("127.0.0.1:{}", 9002), env = "RUSTFS_CONSOLE_ADDRESS")]
pub console_address: String, pub console_address: String,
/// Observability configuration file
/// Default value: config/obs.toml
#[arg(long, default_value_t = DEFAULT_OBS_CONFIG.to_string(), env = "RUSTFS_OBS_CONFIG")]
pub obs_config: String,
/// tls path for rustfs api and console. /// tls path for rustfs api and console.
#[arg(long, env = "RUSTFS_TLS_PATH")] #[arg(long, env = "RUSTFS_TLS_PATH")]
pub tls_path: Option<String>, pub tls_path: Option<String>,
rustfs/src/logging/mod.rs
View File
rustfs/src/main.rs
+35 -12
View File
@@ -3,6 +3,7 @@ mod auth;
mod config; mod config;
mod console; mod console;
mod grpc; mod grpc;
mod logging;
mod service; mod service;
mod storage; mod storage;
mod utils; mod utils;
@@ -35,6 +36,7 @@ use hyper_util::{
}; };
use iam::init_iam_sys; use iam::init_iam_sys;
use protos::proto_gen::node_service::node_service_server::NodeServiceServer; use protos::proto_gen::node_service::node_service_server::NodeServiceServer;
use rustfs_obs::{init_obs, load_config, set_global_guard, InitLogStatus};
use rustls::ServerConfig; use rustls::ServerConfig;
use s3s::{host::MultiDomain, service::S3ServiceBuilder}; use s3s::{host::MultiDomain, service::S3ServiceBuilder};
use service::hybrid; use service::hybrid;
@@ -44,10 +46,11 @@ use tokio::net::TcpListener;
use tokio_rustls::TlsAcceptor; use tokio_rustls::TlsAcceptor;
use tonic::{metadata::MetadataValue, Request, Status}; use tonic::{metadata::MetadataValue, Request, Status};
use tower_http::cors::CorsLayer; use tower_http::cors::CorsLayer;
use tracing::{debug, error, info, warn}; use tracing::{debug, error, info, info_span, warn};
use tracing_error::ErrorLayer; use tracing_error::ErrorLayer;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[allow(dead_code)]
fn setup_tracing() { fn setup_tracing() {
use tracing_subscriber::EnvFilter; use tracing_subscriber::EnvFilter;
@@ -87,22 +90,35 @@ fn print_server_info() {
info!("Docs: {}", cfg.doc()); info!("Docs: {}", cfg.doc());
} }
fn main() -> Result<()> { #[tokio::main]
//解析获得到的参数 async fn main() -> Result<()> {
// Parse the obtained parameters
let opt = config::Opt::parse(); let opt = config::Opt::parse();
//设置 trace // Load the configuration file
setup_tracing(); let config = load_config(Some(opt.clone().obs_config));
//运行参数 // Initialize Observability
run(opt) let (_logger, guard) = init_obs(config.clone()).await;
// Store in global storage
set_global_guard(guard)?;
// Log initialization status
InitLogStatus::init_start_log(&config.observability).await?;
// Run parameters
run(opt).await
} }
#[tokio::main] // #[tokio::main]
async fn run(opt: config::Opt) -> Result<()> { async fn run(opt: config::Opt) -> Result<()> {
let span = info_span!("trace-main-run");
let _enter = span.enter();
debug!("opt: {:?}", &opt); debug!("opt: {:?}", &opt);
let mut server_addr = net::check_local_server_addr(opt.address.as_str()).unwrap(); let mut server_addr = net::check_local_server_addr(opt.address.as_str())?;
if server_addr.port() == 0 { if server_addr.port() == 0 {
server_addr.set_port(get_available_port()); server_addr.set_port(get_available_port());
@@ -115,7 +131,8 @@ async fn run(opt: config::Opt) -> Result<()> {
debug!("server_address {}", &server_address); debug!("server_address {}", &server_address);
//设置 AK 和 SK //设置 AK 和 SK
iam::init_global_action_cred(Some(opt.access_key.clone()), Some(opt.secret_key.clone())).unwrap(); iam::init_global_action_cred(Some(opt.access_key.clone()), Some(opt.secret_key.clone()))?;
set_global_rustfs_port(server_port); set_global_rustfs_port(server_port);
//监听地址，端口从参数中获取 //监听地址，端口从参数中获取
@@ -256,7 +273,7 @@ async fn run(opt: config::Opt) -> Result<()> {
match res { match res {
Ok(conn) => conn, Ok(conn) => conn,
Err(err) => { Err(err) => {
tracing::error!("error accepting connection: {err}"); error!("error accepting connection: {err}");
continue; continue;
} }
} }
@@ -267,7 +284,13 @@ async fn run(opt: config::Opt) -> Result<()> {
}; };
if has_tls_certs { if has_tls_certs {
debug!("TLS certificates found, starting with SIGINT"); debug!("TLS certificates found, starting with SIGINT");
let tls_socket = match tls_acceptor.as_ref().ok_or_else(|| error("TLS not configured".to_string())).unwrap().accept(socket).await { let tls_socket = match tls_acceptor
.as_ref()
.ok_or_else(|| error("TLS not configured".to_string()))
.unwrap()
.accept(socket)
.await
{
Ok(tls_socket) => tls_socket, Ok(tls_socket) => tls_socket,
Err(err) => { Err(err) => {
error!("TLS handshake failed {}", err); error!("TLS handshake failed {}", err);