diff --git a/Cargo.lock b/Cargo.lock index fb04b7c42..ca81fe744 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -176,6 +176,12 @@ dependencies = [ "password-hash", ] +[[package]] +name = "arraydeque" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236" + [[package]] name = "arrayvec" version = "0.7.6" @@ -352,9 +358,9 @@ checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" [[package]] name = "async-trait" -version = "0.1.86" +version = "0.1.87" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "644dd749086bf3771a2fbc5f256fdb982d53f011c7d5d560304eafeecebce79d" +checksum = "d556ec1359574147ec0c4fc5eb525f3f23263a592b1a9c07e0a75b427de55c97" dependencies = [ "proc-macro2", "quote", @@ -486,6 +492,12 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "base64" +version = "0.21.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567" + [[package]] name = "base64" version = "0.22.1" @@ -907,6 +919,25 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "config" +version = "0.15.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb07d21d12f9f0bc5e7c3e97ccc78b2341b9b4a4604eac3ed7c1d0d6e2c3b23e" +dependencies = [ + "async-trait", + "convert_case 0.6.0", + "json5", + "pathdiff", + "ron", + "rust-ini", + "serde", + "serde_json", + "toml", + "winnow 0.7.3", + "yaml-rust2", +] + [[package]] name = "console_error_panic_hook" version = "0.1.7" @@ -923,6 +954,26 @@ version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1cb3c4a0d3776f7535c32793be81d6d5fec0d48ac70955d9834e643aa249a52f" +[[package]] +name = "const-random" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" +dependencies = [ + "const-random-macro", +] + +[[package]] +name = "const-random-macro" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" +dependencies = [ + "getrandom 0.2.15", + "once_cell", + "tiny-keccak", +] + [[package]] name = "const-serialize" version = "0.6.2" @@ -1135,7 +1186,7 @@ dependencies = [ "serde_json", "sha2 0.10.8", "test-case", - "thiserror 2.0.11", + "thiserror 2.0.12", "time", ] @@ -1415,7 +1466,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b5b0cca3e7a10a4a3df37ea52c4cc7a53e5c9233489e03ee3f2829471fc3099a" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "cocoa 0.25.0", "core-foundation 0.9.4", "dioxus-cli-config", @@ -1509,7 +1560,7 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fe99b48a1348eec385b5c4bd3e80fd863b0d3b47257d34e2ddc58754dec5d128" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "ciborium", "dioxus-desktop", @@ -1850,6 +1901,15 @@ dependencies = [ "syn 2.0.98", ] +[[package]] +name = "dlv-list" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "442039f5147480ba31067cb00ada1adae6892028e40e45fc5de7b7df6dcc1b5f" +dependencies = [ + "const-random", +] + [[package]] name = "downcast-rs" version = "1.2.1" @@ -1948,7 +2008,7 @@ dependencies = [ "sha2 0.11.0-pre.4", "siphasher 1.0.1", "tempfile", - "thiserror 2.0.11", + "thiserror 2.0.12", "time", "tokio", "tokio-stream", @@ -2132,6 +2192,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a0d2fde1f7b3d48b8395d5f2de76c18a528bd6a9cdde438df747bfcba3e05d6f" + [[package]] name = "foreign-types" version = "0.5.0" @@ -2720,6 +2786,18 @@ name = "hashbrown" version = "0.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289" +dependencies = [ + "foldhash", +] + +[[package]] +name = "hashlink" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" +dependencies = [ + "hashbrown 0.15.2", +] [[package]] name = "heck" @@ -2962,7 +3040,7 @@ dependencies = [ "serde_json", "strum", "test-case", - "thiserror 2.0.11", + "thiserror 2.0.12", "time", "tokio", "tracing", @@ -3287,13 +3365,24 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "json5" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96b0db21af676c1ce64250b5f40f3ce2cf27e4e47cb91ed91eb6fe9350b430c1" +dependencies = [ + "pest", + "pest_derive", + "serde", +] + [[package]] name = "jsonwebtoken" version = "9.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5a87cc7a48537badeae96744432de36f4be2b4a34a05a5ef32e9dd8a1c169dde" dependencies = [ - "base64", + "base64 0.22.1", "js-sys", "pem", "ring", @@ -4218,7 +4307,7 @@ dependencies = [ "futures-sink", "js-sys", "pin-project-lite", - "thiserror 2.0.11", + "thiserror 2.0.12", "tracing", ] @@ -4265,7 +4354,7 @@ dependencies = [ "opentelemetry_sdk", "prost", "reqwest", - "thiserror 2.0.11", + "thiserror 2.0.12", "tokio", "tonic", "tracing", @@ -4301,7 +4390,7 @@ dependencies = [ "opentelemetry", "opentelemetry_sdk", "serde", - "thiserror 2.0.11", + "thiserror 2.0.12", ] [[package]] @@ -4319,7 +4408,7 @@ dependencies = [ "percent-encoding", "rand 0.8.5", "serde_json", - "thiserror 2.0.11", + "thiserror 2.0.12", "tokio", "tokio-stream", "tracing", @@ -4331,6 +4420,16 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" +[[package]] +name = "ordered-multimap" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49203cdcae0030493bad186b28da2fa25645fa276a51b6fec8010d281e02ef79" +dependencies = [ + "dlv-list", + "hashbrown 0.14.5", +] + [[package]] name = "ordered-stream" version = "0.2.0" @@ -4473,6 +4572,12 @@ dependencies = [ "once_cell", ] +[[package]] +name = "pathdiff" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" + [[package]] name = "pbkdf2" version = "0.12.2" @@ -4489,7 +4594,7 @@ version = "3.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38af38e8470ac9dee3ce1bae1af9c1671fffc44ddfd8bd1d0a3445bf349a8ef3" dependencies = [ - "base64", + "base64 0.22.1", "serde", ] @@ -4499,6 +4604,51 @@ version = "2.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e" +[[package]] +name = "pest" +version = "2.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b7cafe60d6cf8e62e1b9b2ea516a089c008945bb5a275416789e7db0bc199dc" +dependencies = [ + "memchr", + "thiserror 2.0.12", + "ucd-trie", +] + +[[package]] +name = "pest_derive" +version = "2.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "816518421cfc6887a0d62bf441b6ffb4536fcc926395a69e1a85852d4363f57e" +dependencies = [ + "pest", + "pest_generator", +] + +[[package]] +name = "pest_generator" +version = "2.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d1396fd3a870fc7838768d171b4616d5c91f6cc25e377b673d714567d99377b" +dependencies = [ + "pest", + "pest_meta", + "proc-macro2", + "quote", + "syn 2.0.98", +] + +[[package]] +name = "pest_meta" +version = "2.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1e58089ea25d717bfd31fb534e4f3afcc2cc569c70de3e239778991ea3b7dea" +dependencies = [ + "once_cell", + "pest", + "sha2 0.10.8", +] + [[package]] name = "petgraph" version = "0.7.1" @@ -4952,7 +5102,7 @@ dependencies = [ "rustc-hash 2.1.1", "rustls", "socket2", - "thiserror 2.0.11", + "thiserror 2.0.12", "tokio", "tracing", ] @@ -4971,7 +5121,7 @@ dependencies = [ "rustls", "rustls-pki-types", "slab", - "thiserror 2.0.11", + "thiserror 2.0.12", "tinyvec", "tracing", "web-time", @@ -4988,7 +5138,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -5164,7 +5314,7 @@ dependencies = [ "pin-project-lite", "s3s", "sha2 0.11.0-pre.4", - "thiserror 2.0.11", + "thiserror 2.0.12", "tokio", "tracing", ] @@ -5206,7 +5356,7 @@ checksum = "dd6f9d3d47bdd2ad6945c5015a226ec6155d0bcdfd8f7cd29f86b71f8de99d2b" dependencies = [ "getrandom 0.2.15", "libredox", - "thiserror 2.0.11", + "thiserror 2.0.12", ] [[package]] @@ -5274,7 +5424,7 @@ version = "0.12.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "43e734407157c3c2034e0258f5e4473ddb361b1e85f95a66690d67264d7cd1da" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "encoding_rs", "futures-channel", @@ -5401,6 +5551,18 @@ dependencies = [ "serde", ] +[[package]] +name = "ron" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b91f7eff05f748767f183df4320a63d6936e9c6107d97c9e6bdd9784f4289c94" +dependencies = [ + "base64 0.21.7", + "bitflags 2.9.0", + "serde", + "serde_derive", +] + [[package]] name = "rust-embed" version = "8.6.0" @@ -5436,6 +5598,17 @@ dependencies = [ "walkdir", ] +[[package]] +name = "rust-ini" +version = "0.21.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e310ef0e1b6eeb79169a1171daf9abcb87a2e17c03bee2c4bb100b55c75409f" +dependencies = [ + "cfg-if", + "ordered-multimap", + "trim-in-place", +] + [[package]] name = "rustc-demangle" version = "0.1.24" @@ -5553,7 +5726,10 @@ dependencies = [ name = "rustfs-obs" version = "0.0.1" dependencies = [ + "async-trait", "chrono", + "config", + "local-ip-address", "opentelemetry", "opentelemetry-appender-tracing", "opentelemetry-otlp", @@ -5564,8 +5740,10 @@ dependencies = [ "reqwest", "serde", "serde_json", + "thiserror 2.0.12", "tokio", "tracing", + "tracing-core", "tracing-opentelemetry", "tracing-subscriber", ] @@ -5676,7 +5854,7 @@ dependencies = [ "smallvec", "std-next", "sync_wrapper", - "thiserror 2.0.11", + "thiserror 2.0.12", "time", "tokio", "tower 0.5.2", @@ -5694,7 +5872,7 @@ dependencies = [ "indexmap 2.7.1", "serde", "serde_json", - "thiserror 2.0.11", + "thiserror 2.0.12", ] [[package]] @@ -6066,7 +6244,7 @@ checksum = "297f631f50729c8c99b84667867963997ec0b50f32b2a7dbcab828ef0541e8bb" dependencies = [ "num-bigint", "num-traits", - "thiserror 2.0.11", + "thiserror 2.0.12", "time", ] @@ -6197,7 +6375,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bafdb55260d9b29c04fa52351e0db2a4aaeadc462cd884ccd7771c5a31aaf1aa" dependencies = [ "simdutf8", - "thiserror 2.0.11", + "thiserror 2.0.12", ] [[package]] @@ -6467,11 +6645,11 @@ dependencies = [ [[package]] name = "thiserror" -version = "2.0.11" +version = "2.0.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d452f284b73e6d76dd36758a0c8684b1d5be31f92b89d07fd5822175732206fc" +checksum = "567b8a2dae586314f7be2a752ec7474332959c6460e02bde30d702a66d488708" dependencies = [ - "thiserror-impl 2.0.11", + "thiserror-impl 2.0.12", ] [[package]] @@ -6487,9 +6665,9 @@ dependencies = [ [[package]] name = "thiserror-impl" -version = "2.0.11" +version = "2.0.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26afc1baea8a989337eeb52b6e72a039780ce45c3edfcc9c5b9d112feeb173c2" +checksum = "7f7cf42b4507d8ea322120659672cf1b9dbb93f8f2d4ecfd6e51350ff5b17a1d" dependencies = [ "proc-macro2", "quote", @@ -6539,6 +6717,15 @@ dependencies = [ "time-core", ] +[[package]] +name = "tiny-keccak" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237" +dependencies = [ + "crunchy", +] + [[package]] name = "tinystr" version = "0.7.6" @@ -6566,9 +6753,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.43.0" +version = "1.44.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d61fa4ffa3de412bfea335c6ecff681de2b609ba3c77ef3e00e521813a9ed9e" +checksum = "9975ea0f48b5aa3972bf2d888c238182458437cc2a19374b81b25cdf1023fb3a" dependencies = [ "backtrace", "bytes", @@ -6694,7 +6881,7 @@ dependencies = [ "async-stream", "async-trait", "axum", - "base64", + "base64 0.22.1", "bytes", "flate2", "h2", @@ -6960,10 +7147,16 @@ dependencies = [ "objc2-foundation 0.3.0", "once_cell", "png", - "thiserror 2.0.11", + "thiserror 2.0.12", "windows-sys 0.59.0", ] +[[package]] +name = "trim-in-place" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "343e926fc669bc8cde4fa3129ab681c63671bae288b1f1081ceee6d9d37904fc" + [[package]] name = "try-lock" version = "0.2.5" @@ -6994,6 +7187,12 @@ version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1dccffe3ce07af9386bfd29e80c0ab1a8205a2fc34e4bcd40364df902cfa8f3f" +[[package]] +name = "ucd-trie" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" + [[package]] name = "uds_windows" version = "1.1.0" @@ -7870,7 +8069,7 @@ version = "0.45.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac0099a336829fbf54c26b5f620c68980ebbe37196772aeaf6118df4931b5cb0" dependencies = [ - "base64", + "base64 0.22.1", "block", "cocoa 0.26.0", "core-graphics 0.24.0", @@ -7941,6 +8140,17 @@ version = "0.8.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fdd20c5420375476fbd4394763288da7eb0cc0b8c11deed431a91562af7335d3" +[[package]] +name = "yaml-rust2" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "232bdb534d65520716bef0bbb205ff8f2db72d807b19c0bc3020853b92a0cd4b" +dependencies = [ + "arraydeque", + "encoding_rs", + "hashlink", +] + [[package]] name = "yoke" version = "0.7.5" diff --git a/Cargo.toml b/Cargo.toml index 70e5e3888..fff723ded 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,12 +32,13 @@ all = "warn" [workspace.dependencies] madmin = { path = "./madmin" } -async-trait = "0.1.86" +async-trait = "0.1.87" backon = "1.3.0" bytes = "1.9.0" bytesize = "1.3.0" chrono = { version = "0.4.40", features = ["serde"] } clap = { version = "4.5.31", features = ["derive", "env"] } +config = "0.15.9" dioxus = { version = "0.6.3", features = ["router"] } dirs = "6.0.0" ecstore = { path = "./ecstore" } @@ -63,7 +64,7 @@ local-ip-address = "0.6.3" mime = "0.3.17" netif = "0.1.6" opentelemetry = { version = "0.28" } -opentelemetry-appender-tracing = { version = "0.28.1" } +opentelemetry-appender-tracing = { version = "0.28.1", features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] } opentelemetry_sdk = { version = "0.28" } opentelemetry-stdout = { version = "0.28.0" } opentelemetry-otlp = { version = "0.28" } @@ -76,7 +77,7 @@ prost-types = "0.13.4" protobuf = "3.7" protos = { path = "./common/protos" } rand = "0.8.5" -reqwest = { version = "0.12.12", default-features = false, features = ["rustls-tls", "charset", "http2", "macos-system-configuration", "stream"] } +reqwest = { version = "0.12.12", default-features = false, features = ["json", "rustls-tls", "charset", "http2", "macos-system-configuration", "stream"] } rdkafka = { version = "0.37", features = ["tokio"] } rfd = { version = "0.15.2", default-features = false, features = ["xdg-portal", "tokio"] } rmp = "0.8.14" @@ -92,7 +93,7 @@ serde = { version = "1.0.217", features = ["derive"] } serde_json = "1.0.138" sha2 = "0.10.8" tempfile = "3.16.0" -thiserror = "2.0.11" +thiserror = "2.0.12" time = { version = "0.3.37", features = [ "std", "parsing", @@ -100,13 +101,14 @@ time = { version = "0.3.37", features = [ "macros", "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-build = "0.12.3" tonic-reflection = "0.12" tokio-stream = "0.1.17" tower = { version = "0.5.2", features = ["timeout"] } tracing = "0.1.41" +tracing-core = "0.1.33" tracing-error = "0.2.1" tracing-subscriber = { version = "0.3.19", features = ["env-filter", "time"] } tracing-appender = "0.2.3" diff --git a/packages/obs/Cargo.toml b/packages/obs/Cargo.toml index 0e5b3407f..8753d63c7 100644 --- a/packages/obs/Cargo.toml +++ b/packages/obs/Cargo.toml @@ -10,11 +10,15 @@ version.workspace = true workspace = true [features] -default = [] -audit-kafka = ["dep:rdkafka", "dep:serde_json"] -audit-webhook = ["dep:reqwest"] +default = ["file"] +kafka = ["dep:rdkafka", "dep:serde_json"] +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"] } @@ -23,13 +27,15 @@ opentelemetry-otlp = { workspace = true, features = ["grpc-tonic", "metrics"] } opentelemetry-semantic-conventions = { workspace = true, features = ["semconv_experimental"] } serde = { workspace = true } tracing = { workspace = true } +tracing-core = { workspace = true } tracing-opentelemetry = { workspace = true } tracing-subscriber = { workspace = true, features = ["fmt", "env-filter", "tracing-log", "time", "local-time", "json"] } -tokio = { workspace = true, features = ["full"] } +tokio = { workspace = true, features = ["sync", "fs"] } rdkafka = { workspace = true, features = ["tokio"], optional = true } -reqwest = { workspace = true, features = ["json"], optional = true } +reqwest = { workspace = true, optional = true, default-features = false } serde_json = { workspace = true, optional = true } - +thiserror = { workspace = true } +local-ip-address = { workspace = true } [dev-dependencies] diff --git a/packages/obs/examples/config.toml b/packages/obs/examples/config.toml new file mode 100644 index 000000000..9c3c68a26 --- /dev/null +++ b/packages/obs/examples/config.toml @@ -0,0 +1,31 @@ +[observability] +endpoint = "http://localhost:4317" +use_stdout = true +sample_ratio = 0.5 +meter_interval = 30 +service_name = "logging_service" +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 = "app.log" +batch_size = 100 +batch_timeout_ms = 1000 # Default is 8192 bytes if not specified + +[logger] +queue_capacity = 10000 \ No newline at end of file diff --git a/packages/obs/examples/server.rs b/packages/obs/examples/server.rs new file mode 100644 index 000000000..b2c104dd3 --- /dev/null +++ b/packages/obs/examples/server.rs @@ -0,0 +1,78 @@ +use opentelemetry::global; +use opentelemetry::trace::TraceContextExt; +use rustfs_obs::{init_logging, load_config, LogEntry}; +use std::time::{Duration, SystemTime}; +use tracing::{info, instrument, Span}; +use tracing_core::Level; +use tracing_opentelemetry::OpenTelemetrySpanExt; + +#[tokio::main] +async fn main() { + let start_time = SystemTime::now(); + let config = load_config(Some("packages/obs/examples/config".to_string())); + info!("Configuration file loading is complete {:?}", config.clone()); + let (logger, _guard) = init_logging(config); + info!("Log module initialization is completed"); + // Simulate the operation + tokio::time::sleep(Duration::from_millis(100)).await; + + // Record Metrics + let meter = global::meter("rustfs.rs"); + 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")], + ); + + // Gets the current span + let span = Span::current(); + // Use 'OpenTelemetrySpanExt' to get 'SpanContext' + let span_context = span.context(); // Get context via OpenTelemetrySpanExt + let span_id = span_context.span().span_context().span_id().to_string(); // Get the SpanId + let trace_id = span_context.span().span_context().trace_id().to_string(); // Get the TraceId + let result = logger + .log(LogEntry::new( + Level::INFO, + "Process user requests".to_string(), + "api_handler".to_string(), + Some("req-12345".to_string()), + Some("user-6789".to_string()), + vec![ + ("endpoint".to_string(), "/api/v1/data".to_string()), + ("method".to_string(), "GET".to_string()), + ("span_id".to_string(), span_id), + ("trace_id".to_string(), trace_id), + ], + )) + .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 ends"); +} + +#[instrument(fields(bucket, object, user))] +async fn put_object(bucket: String, object: String, user: String) { + let start_time = SystemTime::now(); + info!("Starting PUT operation"); + // Gets the current span + let span = Span::current(); + // Use 'OpenTelemetrySpanExt' to get 'SpanContext' + let span_context = span.context(); // Get context via OpenTelemetrySpanExt + let span_id = span_context.span().span_context().span_id().to_string(); // Get the SpanId + let trace_id = span_context.span().span_context().trace_id().to_string(); // Get the TraceId + info!( + "Starting PUT operation content: bucket = {}, object = {}, user = {},span_id = {},trace_id = {},start_time = {}", + bucket, + object, + user, + span_id, + trace_id, + start_time.elapsed().unwrap().as_secs_f64() + ); + // Simulate the operation + tokio::time::sleep(Duration::from_millis(100)).await; + + info!("PUT operation completed"); +} diff --git a/packages/obs/src/audit.rs b/packages/obs/src/audit.rs deleted file mode 100644 index b353d83fb..000000000 --- a/packages/obs/src/audit.rs +++ /dev/null @@ -1,319 +0,0 @@ -#[cfg(feature = "audit-kafka")] -use rdkafka::{ - producer::{FutureProducer, FutureRecord}, - ClientConfig, -}; - -#[cfg(feature = "audit-webhook")] -use reqwest::Client; - -use serde::{Deserialize, Serialize}; -use tokio::sync::mpsc; - -/// AuditEntry is a struct that represents an audit entry -/// that can be logged -/// # Fields -/// * `version` - The version of the audit entry -/// * `event_type` - The type of event that occurred -/// * `bucket` - The bucket that was accessed -/// * `object` - The object that was accessed -/// * `user` - The user that accessed the object -/// * `time` - The time the event occurred -/// * `user_agent` - The user agent that accessed the object -/// * `span_id` - The span ID of the event -/// # Example -/// ``` -/// use rustfs_obs::AuditEntry; -/// let entry = AuditEntry { -/// version: "1.0".to_string(), -/// event_type: "read".to_string(), -/// bucket: "bucket".to_string(), -/// object: "object".to_string(), -/// user: "user".to_string(), -/// time: "time".to_string(), -/// user_agent: "user_agent".to_string(), -/// span_id: "span_id".to_string(), -/// }; -/// ``` -#[derive(Serialize, Deserialize, Debug, Clone)] -pub struct AuditEntry { - pub version: String, - pub event_type: String, - pub bucket: String, - pub object: String, - pub user: String, - pub time: String, - pub user_agent: String, - pub span_id: String, -} - -/// AuditTarget is a trait that defines the interface for audit targets -/// that can receive audit entries -pub trait AuditTarget: Send + Sync { - fn send(&self, entry: AuditEntry); -} - -/// FileAuditTarget is an audit target that logs audit entries to a file -pub struct FileAuditTarget; - -impl AuditTarget for FileAuditTarget { - /// Send an audit entry to a file - /// # Arguments - /// * `entry` - The audit entry to send - /// - /// # Example - /// ``` - /// use rustfs_obs::{AuditEntry, AuditTarget, FileAuditTarget}; - /// let entry = AuditEntry { - /// version: "1.0".to_string(), - /// event_type: "read".to_string(), - /// bucket: "bucket".to_string(), - /// object: "object".to_string(), - /// user: "user".to_string(), - /// time: "time".to_string(), - /// user_agent: "user_agent".to_string(), - /// span_id: "span_id".to_string(), - /// }; - /// FileAuditTarget.send(entry); - /// ``` - fn send(&self, entry: AuditEntry) { - println!("File audit: {:?}", entry); - } -} - -#[cfg(feature = "audit-webhook")] -/// Webhook audit objectives -/// #Arguments -/// * `client` - The reqwest client -/// * `url` - The URL of the webhook -/// # Example -/// ``` -/// use rustfs_obs::WebhookAuditTarget; -/// let target = WebhookAuditTarget::new("http://localhost:8080"); -/// ``` -pub struct WebhookAuditTarget { - client: Client, - url: String, -} - -#[cfg(feature = "audit-webhook")] -impl WebhookAuditTarget { - pub fn new(url: &str) -> Self { - Self { - client: Client::new(), - url: url.to_string(), - } - } -} - -#[cfg(feature = "audit-webhook")] -impl AuditTarget for WebhookAuditTarget { - fn send(&self, entry: AuditEntry) { - let client = self.client.clone(); - let url = self.url.clone(); - tokio::spawn(async move { - if let Err(e) = client.post(&url).json(&entry).send().await { - eprintln!("Failed to send to Webhook: {:?}", e); - } - }); - } -} - -#[cfg(feature = "audit-kafka")] -/// Kafka audit objectives -/// # Arguments -/// * `producer` - The Kafka producer -/// * `topic` - The Kafka topic -/// # Example -/// ``` -/// use rustfs_obs::KafkaAuditTarget; -/// let target = KafkaAuditTarget::new("localhost:9092", "rustfs-audit"); -/// ``` -/// # Note -/// This feature requires the `rdkafka` crate -/// # Example -/// ```toml -/// [dependencies] -/// rdkafka = "0.26.0" -/// rustfs_obs = { version = "0.1.0", features = ["audit-kafka"] } -/// ``` -/// # Note -/// The `rdkafka` crate requires the `librdkafka` library to be installed -/// # Example -/// ```sh -/// sudo apt-get install librdkafka-dev -/// ``` -/// # Note -/// The `rdkafka` crate requires the `libssl-dev` and `pkg-config` packages to be installed -/// # Example -/// ```sh -/// sudo apt-get install libssl-dev pkg-config -/// ``` -/// # Note -/// The `rdkafka` crate requires the `zlib1g-dev` package to be installed -/// # Example -/// ```sh -/// sudo apt-get install zlib1g-dev -/// ``` -/// # Note -/// The `rdkafka` crate requires the `zstd` package to be installed -/// # Example -/// ```sh -/// sudo apt-get install zstd -/// ``` -/// # Note -/// The `rdkafka` crate requires the `lz4` package to be installed -/// # Example -/// ```sh -/// sudo apt-get install lz4 -/// ``` -pub struct KafkaAuditTarget { - producer: FutureProducer, - topic: String, -} - -#[cfg(feature = "audit-kafka")] -impl KafkaAuditTarget { - pub fn new(brokers: &str, topic: &str) -> Self { - let producer: FutureProducer = ClientConfig::new() - .set("bootstrap.servers", brokers) - .set("message.timeout.ms", "5000") - .create() - .expect("Kafka producer creation failed"); - Self { - producer, - topic: topic.to_string(), - } - } -} - -#[cfg(feature = "audit-kafka")] -impl AuditTarget for KafkaAuditTarget { - fn send(&self, entry: AuditEntry) { - let topic = self.topic.clone(); - let span_id = entry.span_id.clone(); - let payload = serde_json::to_string(&entry).unwrap(); - // let record = FutureRecord::to(&topic).payload(&payload).key(&span_id); - tokio::spawn({ - // 在异步闭包内部创建 record - let topic = topic; - let payload = payload; - let span_id = span_id; - let producer = self.producer.clone(); - async move { - let record = FutureRecord::to(&topic).payload(&payload).key(&span_id); - if let Err(e) = producer.send(record, std::time::Duration::from_secs(0)).await { - eprintln!("Failed to send to Kafka: {:?}", e); - } - } - }); - } -} -/// AuditLogger is a logger that logs audit entries -/// to multiple targets -/// -/// # Example -/// ``` -/// use rustfs_obs::{AuditEntry, AuditLogger, FileAuditTarget}; -/// -/// #[tokio::main] -/// async fn main() { -/// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]); -/// let entry = AuditEntry { -/// version: "1.0".to_string(), -/// event_type: "read".to_string(), -/// bucket: "bucket".to_string(), -/// object: "object".to_string(), -/// user: "user".to_string(), -/// time: "time".to_string(), -/// user_agent: "user_agent".to_string(), -/// span_id: "span_id".to_string(), -/// }; -/// logger.log(entry).await; -/// } -/// ``` - -#[derive(Debug)] -/// AuditLogger is a logger that logs audit entries -/// to multiple targets -/// # Example -/// ``` -/// use rustfs_obs::{AuditEntry, AuditLogger, FileAuditTarget}; -/// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]); -/// ``` -/// # Note -/// This feature requires the `tokio` crate -/// # Example -/// ```toml -/// [dependencies] -/// tokio = { version = "1", features = ["full"] } -/// rustfs_obs = { version = "0.1.0"} -/// ``` -/// # Note -/// This feature requires the `serde` crate -/// # Example -/// ```toml -/// [dependencies] -/// serde = { version = "1", features = ["derive"] } -/// rustfs_obs = { version = "0.1.0"} -/// ``` -pub struct AuditLogger { - tx: mpsc::Sender, -} - -impl AuditLogger { - /// Create a new AuditLogger with the given targets - /// that will receive audit entries - /// # Arguments - /// * `targets` - A vector of audit targets - /// # Returns - /// * An AuditLogger - /// # Example - /// ``` - /// use rustfs_obs::{AuditLogger, AuditEntry, FileAuditTarget}; - /// - /// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]); - /// ``` - pub fn new(targets: Vec>) -> Self { - let (tx, mut rx) = mpsc::channel::(1000); - tokio::spawn(async move { - while let Some(entry) = rx.recv().await { - for target in &targets { - target.send(entry.clone()); - } - } - }); - Self { tx } - } - - /// Log an audit entry - /// # Arguments - /// * `entry` - The audit entry to log - /// # Example - /// ``` - /// use rustfs_obs::{AuditEntry, AuditLogger, FileAuditTarget}; - /// - /// #[tokio::main] - /// async fn main() { - /// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]); - /// let entry = AuditEntry { - /// version: "1.0".to_string(), - /// event_type: "read".to_string(), - /// bucket: "bucket".to_string(), - /// object: "object".to_string(), - /// user: "user".to_string(), - /// time: "time".to_string(), - /// user_agent: "user_agent".to_string(), - /// span_id: "span_id".to_string(), - /// }; - /// logger.log(entry).await; - /// } - /// ``` - pub async fn log(&self, entry: AuditEntry) { - // 将日志消息记录到当前 Span - tracing::Span::current() - .record("log_message", &entry.bucket) - .record("source", &entry.event_type); - let _ = self.tx.send(entry).await; - } -} diff --git a/packages/obs/src/config.rs b/packages/obs/src/config.rs new file mode 100644 index 000000000..8b14f7d0d --- /dev/null +++ b/packages/obs/src/config.rs @@ -0,0 +1,90 @@ +use config::{Config, File, FileFormat}; +use serde::Deserialize; +use std::env; + +/// OpenTelemetry Configuration +#[derive(Debug, Deserialize, Clone, Default)] +pub struct OtelConfig { + pub endpoint: String, + pub use_stdout: bool, + pub sample_ratio: f64, + pub meter_interval: u64, + pub service_name: String, + pub service_version: String, + pub deployment_environment: String, +} + +/// Kafka Sink Configuration - Add batch parameters +#[derive(Debug, Deserialize, Clone)] +pub struct KafkaSinkConfig { + pub enabled: bool, + pub bootstrap_servers: String, + pub topic: String, + pub batch_size: Option, // Batch size, default 100 + pub batch_timeout_ms: Option, // Batch timeout time, default 1000ms +} + +/// Webhook Sink Configuration - Add Retry Parameters +#[derive(Debug, Deserialize, Clone)] +pub struct WebhookSinkConfig { + pub enabled: bool, + pub url: String, + pub max_retries: Option, // Maximum number of retry times, default 3 + pub retry_delay_ms: Option, // 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, // Write buffer size, default 8192 + pub flush_interval_ms: Option, // Refresh interval time, default 1000ms + pub flush_threshold: Option, // Refresh threshold, default 100 logs +} + +/// Sink configuration collection +#[derive(Debug, Deserialize, Clone)] +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, +} + +/// Overall application configuration +#[derive(Debug, Deserialize, Clone)] +pub struct AppConfig { + pub observability: OtelConfig, + pub sinks: SinkConfig, + pub logger: LoggerConfig, +} + +/// Loading the configuration file +/// Supports TOML, YAML and .env formats, read in order by priority +pub fn load_config(config_dir: Option) -> AppConfig { + let config_dir = config_dir.unwrap_or_else(|| { + env::current_dir() + .map(|path| path.to_string_lossy().to_string()) + .unwrap_or_else(|_| { + eprintln!("Warning: Failed to get current directory, using empty path"); + String::new() + }) + }); + + println!("config_dir: {}", config_dir); + + let config = Config::builder() + .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml)) + .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false)) + .add_source(config::Environment::with_prefix("")) + .build() + .unwrap(); + + config.try_deserialize().unwrap() +} diff --git a/packages/obs/src/entry.rs b/packages/obs/src/entry.rs new file mode 100644 index 000000000..a8cef168a --- /dev/null +++ b/packages/obs/src/entry.rs @@ -0,0 +1,80 @@ +use chrono::{DateTime, Utc}; +use serde::de::Error; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use tracing_core::Level; + +/// Wrapper for `tracing_core::Level` to implement `Serialize` and `Deserialize` +#[derive(Debug, Clone)] +pub struct SerializableLevel(pub Level); + +impl From for SerializableLevel { + fn from(level: Level) -> Self { + SerializableLevel(level) + } +} + +impl From for Level { + fn from(serializable_level: SerializableLevel) -> Self { + serializable_level.0 + } +} + +impl Serialize for SerializableLevel { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(self.0.as_str()) + } +} + +impl<'de> Deserialize<'de> for SerializableLevel { + fn deserialize(deserializer: D) -> Result + 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")), + } + } +} + +/// Server log entry structure +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct LogEntry { + pub timestamp: DateTime, // Log timestamp + pub level: SerializableLevel, // Log Level + pub message: String, // Log messages + pub source: String, // Log source (such as module name) + pub request_id: Option, // Request ID (Common Server Fields) + pub user_id: Option, // User ID (Common Server Fields) + pub fields: Vec<(String, String)>, // Attached fields (key value pairs) +} + +impl LogEntry { + /// Create a new LogEntry + pub fn new( + level: Level, + message: String, + source: String, + request_id: Option, + user_id: Option, + fields: Vec<(String, String)>, + ) -> Self { + LogEntry { + timestamp: Utc::now(), + level: SerializableLevel::from(level), + message, + source, + request_id, + user_id, + fields, + } + } +} diff --git a/packages/obs/src/lib.rs b/packages/obs/src/lib.rs index caf993220..469547d61 100644 --- a/packages/obs/src/lib.rs +++ b/packages/obs/src/lib.rs @@ -1,190 +1,32 @@ -//! Logging utilities - +/// # obs /// -/// This crate provides utilities for logging. -/// -/// # Examples -/// ``` -/// use rustfs_obs::{log_info, log_error}; -/// -/// log_info("This is an informational message"); -/// log_error("This is an error message"); -/// ``` -#[cfg(feature = "audit-kafka")] -pub use audit::KafkaAuditTarget; -#[cfg(feature = "audit-webhook")] -pub use audit::WebhookAuditTarget; -pub use audit::{AuditEntry, AuditLogger, AuditTarget, FileAuditTarget}; -pub use logger::{log_debug, log_error, log_info}; -pub use telemetry::Telemetry; - -mod audit; +/// `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. +mod config; +mod entry; mod logger; +mod sink; mod telemetry; +mod utils; +mod worker; -#[cfg(test)] -mod tests { - use crate::{log_info, AuditEntry, AuditLogger, AuditTarget, FileAuditTarget, Telemetry}; - use chrono::Utc; - use opentelemetry::global; - use opentelemetry::trace::{TraceContextExt, Tracer}; - use std::time::{Duration, SystemTime}; - use tracing::{instrument, Span}; - use tracing_opentelemetry::OpenTelemetrySpanExt; +pub use config::load_config; +pub use config::{AppConfig, OtelConfig}; +pub use entry::{LogEntry, SerializableLevel}; +pub use logger::start_logger; +pub use logger::{LogError, Logger}; +pub use sink::Sink; +pub use telemetry::init_telemetry; +pub use utils::{get_local_ip, get_local_ip_with_default}; +pub use worker::start_worker; - #[instrument(fields(bucket, object, user))] - async fn put_object(audit_logger: &AuditLogger, bucket: String, object: String, user: String) { - let start_time = SystemTime::now(); - log_info("Starting PUT operation"); - - // Simulate the operation - tokio::time::sleep(Duration::from_millis(100)).await; - - // Record Metrics - let meter = global::meter("rustfs.rs"); - 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")], - ); - - // Gets the current span - let span = Span::current(); - - // Use 'OpenTelemetrySpanExt' to get 'SpanContext' - let span_context = span.context(); // Get context via OpenTelemetrySpanExt - let span_id = span_context.span().span_context().span_id().to_string(); // Get the SpanId - - // Audit events are logged - let audit_entry = AuditEntry { - version: "1.0".to_string(), - event_type: "s3_put_object".to_string(), - bucket, - object, - user, - time: Utc::now().to_rfc3339(), - user_agent: "rustfs.rs-client".to_string(), - span_id, - }; - audit_logger.log(audit_entry).await; - - log_info("PUT operation completed"); - } - - #[tokio::test] - // #[cfg(feature = "audit-webhook")] - // #[cfg(feature = "audit-kafka")] - async fn test_main() { - let telemetry = Telemetry::init(); - - // Initialize multiple audit objectives - let audit_targets: Vec> = vec![ - Box::new(FileAuditTarget), - // Box::new(KafkaAuditTarget::new("localhost:9092", "rustfs-audit")), - // Box::new(WebhookAuditTarget::new("http://localhost:8080/audit")), - ]; - let audit_logger = AuditLogger::new(audit_targets); - - // 创建根 Span 并执行操作 - // let tracer = global::tracer("main"); - // tracer.in_span("main_operation", |cx| { - // Span::current().set_parent(cx); - // log_info("Starting test async"); - // tokio::runtime::Runtime::new().unwrap().block_on(async { - log_info("Starting test"); - // Test the PUT operation - put_object(&audit_logger, "my-bucket".to_string(), "my-object.txt".to_string(), "user123".to_string()).await; - tokio::time::sleep(Duration::from_millis(100)).await; - query_object(&audit_logger, "my-bucket".to_string(), "my-object.txt".to_string(), "user123".to_string()).await; - tokio::time::sleep(Duration::from_millis(100)).await; - for i in 0..100 { - put_object( - &audit_logger, - format!("my-bucket-{}", i), - format!("my-object-{}", i), - "user123".to_string(), - ) - .await; - tokio::time::sleep(Duration::from_millis(100)).await; - query_object( - &audit_logger, - format!("my-bucket-{}", i), - format!("my-object-{}", i), - "user123".to_string(), - ) - .await; - tokio::time::sleep(Duration::from_millis(100)).await; - } - - // Wait for the export to complete - tokio::time::sleep(Duration::from_secs(2)).await; - log_info("Test completed"); - // }); - // }); - drop(telemetry); // Make sure to clean up - } - - #[instrument(fields(bucket, object, user))] - async fn query_object(audit_logger: &AuditLogger, bucket: String, object: String, user: String) { - let start_time = SystemTime::now(); - log_info("Starting query operation"); - - // Simulate the operation - tokio::time::sleep(Duration::from_millis(100)).await; - - // Record Metrics - let meter = global::meter("rustfs.rs"); - 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", "query_object")], - ); - - // Gets the current span - let span = Span::current(); - // Use 'OpenTelemetrySpanExt' to get 'SpanContext' - let span_context = span.context(); // Get context via OpenTelemetrySpanExt - let span_id = span_context.span().span_context().span_id().to_string(); // Get the SpanId - query_one(user.clone()); - query_two(user.clone()); - query_three(user.clone()); - // Audit events are logged - let audit_entry = AuditEntry { - version: "1.0".to_string(), - event_type: "s3_query_object".to_string(), - bucket, - object, - user, - time: Utc::now().to_rfc3339(), - user_agent: "rustfs.rs-client".to_string(), - span_id, - }; - audit_logger.log(audit_entry).await; - - log_info("query operation completed"); - } - #[instrument(fields(user))] - fn query_one(user: String) { - // 初始化 OpenTelemetry Tracer - let tracer = global::tracer("query_one"); - tracer.in_span("doing_work", |cx| { - // Traced app logic here... - Span::current().set_parent(cx); - log_info("Doing work..."); - let current_span = Span::current(); - let span_context = current_span.context(); - let trace_id = span_context.clone().span().span_context().trace_id().to_string(); - let span_id = span_context.clone().span().span_context().span_id().to_string(); - log_info(format!("trace_id: {}, span_id: {}", trace_id, span_id).as_str()); - }); - log_info(format!("Starting query_one operation user:{}", user).as_str()); - } - #[instrument(fields(user))] - fn query_two(user: String) { - log_info(format!("Starting query_two operation user:{}", user).as_str()); - } - #[instrument(fields(user))] - fn query_three(user: String) { - log_info(format!("Starting query_three operation user: {}", user).as_str()); - } +/// Log module initialization function +/// +/// Return to Logger and Clean Guard +pub fn init_logging(config: AppConfig) -> (Logger, telemetry::OtelGuard) { + let guard = init_telemetry(&config.observability); + let sinks = sink::create_sinks(&config); + let logger = start_logger(&config, sinks); + (logger, guard) } diff --git a/packages/obs/src/logger.rs b/packages/obs/src/logger.rs index 023419592..f2297c9ef 100644 --- a/packages/obs/src/logger.rs +++ b/packages/obs/src/logger.rs @@ -1,46 +1,238 @@ -use tracing::{debug, error, info}; +use crate::{AppConfig, LogEntry, SerializableLevel, Sink}; +use std::sync::Arc; +use tokio::sync::mpsc::{self, Receiver, Sender}; -/// Log an info message -/// -/// # Arguments -/// msg: &str - The message to log -/// -/// # Example -/// ``` -/// use rustfs_obs::log_info; -/// -/// log_info("This is an info message"); -/// ``` -pub fn log_info(msg: &str) { - info!("{}", msg); +/// Server log processor +pub struct Logger { + sender: Sender, // Log sending channel + queue_capacity: usize, } -/// Log an error message -/// -/// # Arguments -/// msg: &str - The message to log -/// -/// # Example -/// ``` -/// use rustfs_obs::log_error; -/// -/// log_error("This is an error message"); -/// ``` -pub fn log_error(msg: &str) { - error!("{}", msg); +impl Logger { + /// Create a new Logger instance + /// Returns Logger and corresponding Receiver + pub fn new(config: &AppConfig) -> (Self, Receiver) { + // 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, + ) + } + + // Add a method to get queue capacity + pub fn queue_capacity(&self) -> usize { + self.queue_capacity + } + + /// Asynchronous logging of server logs + /// Attach the log to the current Span and generate a separate Tracing Event + #[tracing::instrument(skip(self), fields(log_source = "logger"))] + pub async fn log(&self, entry: LogEntry) -> Result<(), LogError> { + // Log messages to the current Span + tracing::Span::current() + .record("log_message", &entry.message) + .record("source", &entry.source); + + // Record queue utilization (if a certain threshold is exceeded) + let queue_len = self.sender.capacity(); + let utilization = queue_len as f64 / self.queue_capacity as f64; + if utilization > 0.8 { + tracing::warn!("Log queue utilization high: {:.1}%", utilization * 100.0); + } + + // Generate independent Tracing Events with full LogEntry information + // Generate corresponding events according to level + match entry.level { + SerializableLevel(tracing::Level::ERROR) => { + tracing::error!( + target: "server_logs", + timestamp = %entry.timestamp, + message = %entry.message, + source = %entry.source, + request_id = ?entry.request_id, + user_id = ?entry.user_id, + fields = ?entry.fields + ); + } + SerializableLevel(tracing::Level::WARN) => { + tracing::warn!( + target: "server_logs", + timestamp = %entry.timestamp, + message = %entry.message, + source = %entry.source, + request_id = ?entry.request_id, + user_id = ?entry.user_id, + fields = ?entry.fields + ); + } + SerializableLevel(tracing::Level::INFO) => { + tracing::info!( + target: "server_logs", + timestamp = %entry.timestamp, + message = %entry.message, + source = %entry.source, + request_id = ?entry.request_id, + user_id = ?entry.user_id, + fields = ?entry.fields + ); + } + SerializableLevel(tracing::Level::DEBUG) => { + tracing::debug!( + target: "server_logs", + timestamp = %entry.timestamp, + message = %entry.message, + source = %entry.source, + request_id = ?entry.request_id, + user_id = ?entry.user_id, + fields = ?entry.fields + ); + } + SerializableLevel(tracing::Level::TRACE) => { + tracing::trace!( + target: "server_logs", + timestamp = %entry.timestamp, + message = %entry.message, + source = %entry.source, + request_id = ?entry.request_id, + user_id = ?entry.user_id, + fields = ?entry.fields + ); + } + } + + // Send logs to asynchronous queues to improve 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")) + } + } + } + + // Add convenient methods to simplify logging + // Fix the info() method, replacing None with an empty vector instead of the Option type + pub async fn info(&self, message: &str, source: &str) -> Result<(), LogError> { + self.log(LogEntry::new( + tracing::Level::INFO, + message.to_string(), + source.to_string(), + None, + None, + Vec::new(), // 使用空向量代替 None + )) + .await + } + + /// Add warn() method + pub async fn error(&self, message: &str, source: &str) -> Result<(), LogError> { + self.log(LogEntry::new( + tracing::Level::ERROR, + message.to_string(), + source.to_string(), + None, + None, + Vec::new(), + )) + .await + } + + /// Add warn() method + pub async fn warn(&self, message: &str, source: &str) -> Result<(), LogError> { + self.log(LogEntry::new( + tracing::Level::WARN, + message.to_string(), + source.to_string(), + None, + None, + Vec::new(), + )) + .await + } + + /// Add debug() method + pub async fn debug(&self, message: &str, source: &str) -> Result<(), LogError> { + self.log(LogEntry::new( + tracing::Level::DEBUG, + message.to_string(), + source.to_string(), + None, + None, + Vec::new(), + )) + .await + } + + /// Add trace() method + pub async fn trace(&self, message: &str, source: &str) -> Result<(), LogError> { + self.log(LogEntry::new( + tracing::Level::TRACE, + message.to_string(), + source.to_string(), + None, + None, + Vec::new(), + )) + .await + } + + // Add extension methods with context information for more flexibility + pub async fn info_with_context( + &self, + message: &str, + source: &str, + request_id: Option, + user_id: Option, + fields: Vec<(String, String)>, + ) -> Result<(), LogError> { + self.log(LogEntry::new( + tracing::Level::INFO, + message.to_string(), + source.to_string(), + request_id, + user_id, + fields, + )) + .await + } + + // Add elegant closing method + 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 a debug message -/// -/// # Arguments -/// msg: &str - The message to log -/// -/// # Example -/// ``` -/// use rustfs_obs::log_debug; -/// -/// log_debug("This is a debug message"); -/// ``` -pub fn log_debug(msg: &str) { - debug!("{}", msg); +// Define custom error type +#[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 +pub fn start_logger(config: &AppConfig, sinks: Vec>) -> Logger { + let (logger, receiver) = Logger::new(config); + tokio::spawn(crate::worker::start_worker(receiver, sinks)); + logger } diff --git a/packages/obs/src/sink.rs b/packages/obs/src/sink.rs new file mode 100644 index 000000000..37785cfba --- /dev/null +++ b/packages/obs/src/sink.rs @@ -0,0 +1,450 @@ +use crate::{AppConfig, LogEntry}; +use async_trait::async_trait; +use std::sync::Arc; +use tokio::fs::OpenOptions; +use tokio::io; +use tokio::io::AsyncWriteExt; + +/// Sink Trait definition, asynchronously write logs +#[async_trait] +pub trait Sink: Send + Sync { + async fn write(&self, entry: &LogEntry); +} + +#[cfg(feature = "kafka")] +/// Kafka Sink Implementation +pub struct KafkaSink { + producer: rdkafka::producer::FutureProducer, + topic: String, + batch_size: usize, + batch_timeout_ms: u64, + entries: Arc>>, + last_flush: Arc, +} + +#[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>>, + last_flush: Arc, + 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) { + 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.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: &LogEntry) { + 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 = 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 { + url: String, + client: reqwest::Client, + max_retries: usize, + retry_delay_ms: u64, +} + +#[cfg(feature = "webhook")] +impl WebhookSink { + pub fn new(url: String, max_retries: usize, retry_delay_ms: u64) -> Self { + WebhookSink { + url, + 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: &LogEntry) { + let mut retries = 0; + let url = self.url.clone(); + let entry_clone = entry.clone(); + + while retries < self.max_retries { + match self.client.post(&url).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.url); + } +} + +#[cfg(feature = "file")] +/// File Sink Implementation +pub struct FileSink { + path: String, + buffer_size: usize, + writer: Arc>>, + 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 { + #[allow(dead_code)] + pub async fn new( + path: String, + buffer_size: usize, + flush_interval_ms: u64, + flush_threshold: usize, + ) -> Result { + let file = OpenOptions::new().append(true).create(true).open(&path).await?; + + let writer = tokio::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: &LogEntry) { + 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 {}: {}", self.path, e); + 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 fn create_sinks(config: &AppConfig) -> Vec> { + let mut sinks: Vec> = 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.url.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() + }; + + // Use synchronous file operations + let file_result = std::fs::OpenOptions::new().append(true).create(true).open(&path); + + match file_result { + Ok(file) => { + let buffer_size = config.sinks.file.buffer_size.unwrap_or(8192); + let writer = tokio::io::BufWriter::with_capacity(buffer_size, tokio::fs::File::from_std(file)); + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + + sinks.push(Arc::new(FileSink { + path: path.clone(), + 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: config.sinks.file.flush_interval_ms.unwrap_or(1000), + flush_threshold: config.sinks.file.flush_threshold.unwrap_or(100), + })); + } + Err(e) => eprintln!("Failed to create file sink: {}", e), + } + } + + sinks +} diff --git a/packages/obs/src/telemetry.rs b/packages/obs/src/telemetry.rs index 0eed5466d..fb1111303 100644 --- a/packages/obs/src/telemetry.rs +++ b/packages/obs/src/telemetry.rs @@ -1,149 +1,222 @@ +use crate::{get_local_ip_with_default, OtelConfig}; use opentelemetry::trace::TracerProvider; use opentelemetry::{global, KeyValue}; use opentelemetry_appender_tracing::layer; -use opentelemetry_otlp::{self, WithExportConfig}; +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::NETWORK_LOCAL_ADDRESS; use opentelemetry_semantic_conventions::{ - attribute::{DEPLOYMENT_ENVIRONMENT_NAME, SERVICE_NAME, SERVICE_VERSION}, + attribute::{DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_NAME, SERVICE_VERSION}, SCHEMA_URL, }; -use std::time::Duration; -use tracing::{info, Level}; use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer}; -/// Telemetry is a wrapper around the OpenTelemetry SDK Tracer and Meter providers. -/// It initializes the global Tracer and Meter providers, and sets up the tracing subscriber. -/// The Tracer and Meter providers are shut down when the Telemetry instance is dropped. -/// This is a convenience struct to ensure that the global providers are properly initialized and shut down. +/// 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::Telemetry; /// -/// let _telemetry = Telemetry::init(); /// ``` -pub struct Telemetry { +/// 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); +/// ``` +pub struct OtelGuard { tracer_provider: SdkTracerProvider, meter_provider: SdkMeterProvider, + logger_provider: SdkLoggerProvider, } -impl Telemetry { - pub fn init() -> Self { - // Define service resource information - let resource = Resource::builder() - .with_service_name("rustfs-service") - .with_schema_url( - [ - KeyValue::new(SERVICE_NAME, "rustfs-service"), - KeyValue::new(SERVICE_VERSION, "0.1.0"), - KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, "develop"), - KeyValue::new(NETWORK_LOCAL_ADDRESS, "127.0.0.1"), - ], - SCHEMA_URL, - ) - .build(); +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); + } + } +} - let tracer_exporter = opentelemetry_otlp::SpanExporter::builder() +/// create OpenTelemetry Resource +fn resource(config: &OtelConfig) -> Resource { + Resource::builder() + .with_service_name(config.service_name.clone()) + .with_schema_url( + [ + KeyValue::new(SERVICE_NAME, config.service_name.clone()), + KeyValue::new(SERVICE_VERSION, config.service_version.clone()), + KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, config.deployment_environment.clone()), + KeyValue::new(NETWORK_LOCAL_ADDRESS, get_local_ip_with_default()), + ], + SCHEMA_URL, + ) + .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( + PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default()) + .with_interval(std::time::Duration::from_secs(config.meter_interval)) + .build(), + ); + } else { + // If endpoint is not empty, use otlp output + let exporter = opentelemetry_otlp::MetricExporter::builder() .with_tonic() - .with_endpoint("http://localhost:4317") - .with_protocol(opentelemetry_otlp::Protocol::Grpc) - .with_timeout(Duration::from_secs(3)) - .build() - .unwrap(); - - // Configure Tracer Provider - let tracer_provider = SdkTracerProvider::builder() - .with_sampler(Sampler::AlwaysOn) - .with_id_generator(RandomIdGenerator::default()) - .with_resource(resource.clone()) - .with_batch_exporter(tracer_exporter) - // .with_simple_exporter(opentelemetry_stdout::SpanExporter::default()) - .build(); - - let meter_exporter = opentelemetry_otlp::MetricExporter::builder() - .with_tonic() - .with_endpoint("http://localhost:4317") - .with_protocol(opentelemetry_otlp::Protocol::Grpc) - .with_timeout(Duration::from_secs(3)) + .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)) + .build(), + ); + // If use_stdout is true, output to stdout at the same time + if config.use_stdout { + builder = builder.with_reader( + PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default()) + .with_interval(std::time::Duration::from_secs(config.meter_interval)) + .build(), + ); + } + } - let meter_reader = PeriodicReader::builder(meter_exporter) - .with_interval(Duration::from_secs(30)) - .build(); + let meter_provider = builder.build(); + global::set_meter_provider(meter_provider.clone()); + meter_provider +} - // For debugging in development - // let meter_stdout_reader = PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default()).build(); +/// Initialize Tracer Provider +fn init_tracer_provider(config: &OtelConfig) -> SdkTracerProvider { + let sampler = if config.sample_ratio > 0.0 && config.sample_ratio < 1.0 { + Sampler::TraceIdRatioBased(config.sample_ratio) + } else { + Sampler::AlwaysOn + }; + let builder = SdkTracerProvider::builder() + .with_sampler(sampler) + .with_id_generator(RandomIdGenerator::default()) + .with_resource(resource(config)); - // Configure Meter Provider - let meter_provider = MeterProviderBuilder::default() - .with_resource(resource.clone()) - .with_reader(meter_reader) - // .with_reader(meter_stdout_reader) - .build(); + let tracer_provider = if config.endpoint.is_empty() { + builder + .with_simple_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 { + builder + .with_batch_exporter(exporter) + .with_batch_exporter(opentelemetry_stdout::SpanExporter::default()) + } else { + builder.with_batch_exporter(exporter) + } + .build() + }; - // Set global Tracer and Meter providers - global::set_tracer_provider(tracer_provider.clone()); - global::set_meter_provider(meter_provider.clone()); + global::set_tracer_provider(tracer_provider.clone()); + tracer_provider +} - let tracer = tracer_provider.tracer("rustfs-service"); +/// Initialize Telemetry +pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { + let tracer_provider = init_tracer_provider(config); + let meter_provider = init_meter_provider(config); + let tracer = tracer_provider.tracer(config.service_name.clone()); - // // let _stdout_exporter = opentelemetry_stdout::LogExporter::default(); - // let otlp_exporter = opentelemetry_otlp::LogExporter::builder() - // .with_tonic() - // .with_endpoint("http://localhost:4317") - // // .with_timeout(Duration::from_secs(3)) - // // .with_protocol(opentelemetry_otlp::Protocol::Grpc) - // .build() - // .unwrap(); - // let provider: SdkLoggerProvider = SdkLoggerProvider::builder() - // .with_resource(resource.clone()) - // .with_simple_exporter(otlp_exporter) - // .build(); - // let filter_otel = EnvFilter::new("debug") - // // .add_directive("hyper=off".parse().unwrap()) - // // .add_directive("opentelemetry=off".parse().unwrap()) - // // .add_directive("tonic=off".parse().unwrap()) - // // .add_directive("h2=off".parse().unwrap()) - // .add_directive("reqwest=off".parse().unwrap()); - // let otel_layer = layer::OpenTelemetryTracingBridge::new(&provider).with_filter(filter_otel); + let logger_provider = if config.endpoint.is_empty() { + SdkLoggerProvider::builder() + .with_resource(resource(config)) + .with_simple_exporter(opentelemetry_stdout::LogExporter::default()) + .build() + } else { + let exporter = opentelemetry_otlp::LogExporter::builder() + .with_tonic() + .with_endpoint(&config.endpoint) + .build() + .unwrap(); + SdkLoggerProvider::builder() + .with_resource(resource(config)) + .with_batch_exporter(exporter) + .with_batch_exporter(opentelemetry_stdout::LogExporter::default()) + .build() + }; + + let otel_layer = layer::OpenTelemetryTracingBridge::new(&logger_provider); + // For the OpenTelemetry layer, add a tracing filter to filter events from + // OpenTelemetry and its dependent crates (opentelemetry-otlp uses crates + // like reqwest/tonic etc.) from being sent back to OTel itself, thus + // preventing infinite telemetry generation. The filter levels are set as + // follows: + // - Allow `info` level and above by default. + // - Restrict `opentelemetry`, `hyper`, `tonic`, and `reqwest` completely. + // Note: This will also drop events from crates like `tonic` etc. even when + // they are used outside the OTLP Exporter. For more details, see: + // https://github.com/open-telemetry/opentelemetry-rust/issues/761 + let filter_otel = EnvFilter::new("info") + .add_directive("hyper=off".parse().unwrap()) + .add_directive("opentelemetry=off".parse().unwrap()) + .add_directive("tonic=off".parse().unwrap()) + .add_directive("h2=off".parse().unwrap()) + .add_directive("reqwest=off".parse().unwrap()); + let otel_layer = otel_layer.with_filter(filter_otel); + let registry = tracing_subscriber::registry() + .with(tracing_subscriber::filter::LevelFilter::INFO) + .with(OpenTelemetryLayer::new(tracer)) + .with(MetricsLayer::new(meter_provider.clone())) + .with(otel_layer); + if config.endpoint.is_empty() { + // Create a new tracing::Fmt layer to print the logs to stdout. It has a + // default filter of `info` level and above, and `debug` and above for logs + // from OpenTelemetry crates. The filter levels can be customized as needed. let filter_fmt = EnvFilter::new("info").add_directive("opentelemetry=debug".parse().unwrap()); let fmt_layer = tracing_subscriber::fmt::layer() .with_thread_names(true) .with_filter(filter_fmt); - // Configure `tracing subscriber` - tracing_subscriber::registry() - .with(tracing_subscriber::filter::LevelFilter::from_level(Level::DEBUG)) + registry .with(tracing_subscriber::fmt::layer().with_ansi(true)) - .with(MetricsLayer::new(meter_provider.clone())) - .with(OpenTelemetryLayer::new(tracer)) - // .with(otel_layer) .with(fmt_layer) .init(); - info!(name: "my-event-name", target: "my-system", event_id = 20, user_name = "otel", user_email = "otel@opentelemetry.io", message = "This is an example message"); - Self { - tracer_provider, - meter_provider, - } + } else { + registry.with(tracing_subscriber::fmt::layer().with_ansi(false)).init(); + println!("Logs and meter,tracer enabled"); } -} -impl Drop for Telemetry { - fn drop(&mut self) { - if let Err(err) = self.tracer_provider.shutdown() { - eprintln!("{err:?}"); - } - if let Err(err) = self.meter_provider.shutdown() { - eprintln!("{err:?}"); - } + OtelGuard { + tracer_provider, + meter_provider, + logger_provider, } } diff --git a/packages/obs/src/utils.rs b/packages/obs/src/utils.rs new file mode 100644 index 000000000..774594577 --- /dev/null +++ b/packages/obs/src/utils.rs @@ -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 { + 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()); + } +} diff --git a/packages/obs/src/worker.rs b/packages/obs/src/worker.rs new file mode 100644 index 000000000..a04327945 --- /dev/null +++ b/packages/obs/src/worker.rs @@ -0,0 +1,13 @@ +use crate::{entry::LogEntry, sink::Sink}; +use std::sync::Arc; +use tokio::sync::mpsc::Receiver; + +/// Start the log processing worker thread +pub async fn start_worker(receiver: Receiver, sinks: Vec>) { + let mut receiver = receiver; + while let Some(entry) = receiver.recv().await { + for sink in &sinks { + sink.write(&entry).await; + } + } +}