diff --git a/.gitignore b/.gitignore index 9b8266092..8f19d5fca 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,4 @@ cli/rustfs-gui/embedded-rustfs/rustfs deploy/config/obs.toml *.log deploy/certs/* +*jsonl \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 47a013b25..787b4bd38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -174,9 +174,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.97" +version = "1.0.98" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dcfed56ad506cb2c684a14971b8861fdc3baaaae314b9e5f9bb532cbe3ba7a4f" +checksum = "e16d2d3311acee920a9eb8d33b8cbc1787ce4a264e85f964c2404b969bdcd487" [[package]] name = "api" @@ -371,7 +371,7 @@ dependencies = [ "arrow-buffer", "arrow-data", "arrow-schema", - "flatbuffers", + "flatbuffers 24.12.23", "lz4_flex", ] @@ -486,7 +486,7 @@ dependencies = [ "enumflags2", "futures-channel", "futures-util", - "rand 0.9.0", + "rand 0.9.1", "raw-window-handle 0.6.2", "serde", "serde_repr", @@ -699,9 +699,9 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.28.0" +version = "0.28.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9f7720b74ed28ca77f90769a71fd8c637a0137f6fae4ae947e1050229cff57f" +checksum = "0ddeb19ee86cb16ecfc871e5b0660aff6285760957aaedda6284cf0e790d3769" dependencies = [ "bindgen", "cc", @@ -800,11 +800,11 @@ dependencies = [ "hyper", "hyper-util", "pin-project-lite", - "rustls", + "rustls 0.23.26", "rustls-pemfile", "rustls-pki-types", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.2", "tower-service", ] @@ -924,9 +924,9 @@ dependencies = [ [[package]] name = "blake3" -version = "1.8.1" +version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "389a099b34312839e16420d499a9cad9650541715937ffbdd40d36f49e77eeb3" +checksum = "3888aaa89e4b2a40fca9848e400f6a658a5a3978de7be858e209cafa8be9a4a0" dependencies = [ "arrayref", "arrayvec", @@ -970,11 +970,11 @@ dependencies = [ [[package]] name = "block2" -version = "0.6.0" +version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d59b4c170e16f0405a2e95aff44432a0d41aa97675f3d52623effe95792a037" +checksum = "340d2f0bdb2a43c1d3cd40513185b2bd7def0aa1052f956455114bc98f82dcf2" dependencies = [ - "objc2 0.6.0", + "objc2 0.6.1", ] [[package]] @@ -1003,9 +1003,9 @@ dependencies = [ [[package]] name = "brotli-decompressor" -version = "4.0.2" +version = "4.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "74fa05ad7d803d413eb8380983b092cbbaf9a85f151b871360e7b00cd7060b37" +checksum = "a334ef7c9e23abf0ce748e8cd309037da93e606ad52eb372e4ce327a0dcfbdfd" dependencies = [ "alloc-no-stdlib", "alloc-stdlib", @@ -1031,9 +1031,9 @@ checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" [[package]] name = "bytesize" -version = "1.3.3" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e93abca9e28e0a1b9877922aacb20576e05d4679ffa78c3d6dc22a26a216659" +checksum = "a3c8f83209414aacf0eeae3cf730b18d6981697fba62f200fcfb92b9f082acba" [[package]] name = "bytestring" @@ -1088,6 +1088,38 @@ dependencies = [ "system-deps", ] +[[package]] +name = "camino" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b96ec4966b5813e2c0507c1f86115c8c5abaadc3980879c3424042a02fd1ad3" +dependencies = [ + "serde", +] + +[[package]] +name = "cargo-platform" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e35af189006b9c0f00a064685c727031e3ed2d8020f7ba284d78cc2671bd36ea" +dependencies = [ + "serde", +] + +[[package]] +name = "cargo_metadata" +version = "0.19.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd5eb614ed4c27c5d706420e4320fbe3216ab31fa1c33cd8246ac36dae4479ba" +dependencies = [ + "camino", + "cargo-platform", + "semver", + "serde", + "serde_json", + "thiserror 2.0.12", +] + [[package]] name = "cc" version = "1.2.19" @@ -1111,7 +1143,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" dependencies = [ - "nom", + "nom 7.1.3", ] [[package]] @@ -1258,9 +1290,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.35" +version = "4.5.37" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8aa86934b44c19c50f87cc2790e19f54f7a67aedb64101c2e1a2e5ecfb73944" +checksum = "eccb054f56cbd38340b380d4a8e69ef1f02f1af43db2f0cc817a4774d80ae071" dependencies = [ "clap_builder", "clap_derive", @@ -1268,9 +1300,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.5.35" +version = "4.5.37" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2414dbb2dd0695280da6ea9261e327479e9d37b0630f6b53ba2a11c60c679fd9" +checksum = "efd9466fac8543255d3b1fcad4762c5e116ffe808c8a3043d4263cd4fd4862a2" dependencies = [ "anstream", "anstyle", @@ -1860,9 +1892,9 @@ dependencies = [ [[package]] name = "data-encoding" -version = "2.8.0" +version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "575f75dfd25738df5b91b8e43e14d44bda14637a58fae779fd2b064f8bf3e010" +checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476" [[package]] name = "datafusion" @@ -2369,9 +2401,9 @@ dependencies = [ [[package]] name = "der" -version = "0.7.9" +version = "0.7.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f55bf8e7b65898637379c1b74eb1551107c8294ed26d855ceb9fd1a09cfc9bc0" +checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ "const-oid 0.9.6", "pem-rfc7468", @@ -2895,34 +2927,13 @@ dependencies = [ "syn 2.0.100", ] -[[package]] -name = "dirs" -version = "5.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "44c45a9d03d6676652bcb5e724c7e988de1acad23a711b5217ab9cbecbec2225" -dependencies = [ - "dirs-sys 0.4.1", -] - [[package]] name = "dirs" version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3e8aa94d75141228480295a7d0e7feb620b1a5ad9f12bc40be62411e38cce4e" dependencies = [ - "dirs-sys 0.5.0", -] - -[[package]] -name = "dirs-sys" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "520f05a5cbd335fae5a99ff7a6ab8627577660ee5cfd6a94a6a929b52ff0321c" -dependencies = [ - "libc", - "option-ext", - "redox_users 0.4.6", - "windows-sys 0.48.0", + "dirs-sys", ] [[package]] @@ -2933,7 +2944,7 @@ checksum = "e01a3366d27ee9890022452ee61b2b63a67e6f13f58900b651ff5665f0bb1fab" dependencies = [ "libc", "option-ext", - "redox_users 0.5.0", + "redox_users", "windows-sys 0.59.0", ] @@ -2950,9 +2961,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a0d569e003ff27784e0e14e4a594048698e0c0f0b66cabcb51511be55a7caa0" dependencies = [ "bitflags 2.9.0", - "block2 0.6.0", + "block2 0.6.1", "libc", - "objc2 0.6.0", + "objc2 0.6.1", +] + +[[package]] +name = "dispatch2" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89a09f22a6c6069a18470eb92d2298acf25463f14256d24778e1230d789a2aec" +dependencies = [ + "bitflags 2.9.0", + "objc2 0.6.1", ] [[package]] @@ -3031,7 +3052,7 @@ version = "0.0.1" dependencies = [ "common", "ecstore", - "flatbuffers", + "flatbuffers 25.2.10", "futures", "lazy_static", "lock", @@ -3048,7 +3069,7 @@ dependencies = [ [[package]] name = "ecstore" -version = "0.1.0" +version = "0.0.1" dependencies = [ "async-trait", "backon", @@ -3060,7 +3081,7 @@ dependencies = [ "chrono", "common", "crc32fast", - "flatbuffers", + "flatbuffers 25.2.10", "futures", "glob", "hex-simd", @@ -3071,7 +3092,7 @@ dependencies = [ "madmin", "md-5", "netif", - "nix 0.29.0", + "nix", "num", "num_cpus", "path-absolutize", @@ -3262,6 +3283,16 @@ dependencies = [ "rustc_version", ] +[[package]] +name = "flatbuffers" +version = "25.2.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1045398c1bfd89168b5fd3f1fc11f6e70b34f6f66300c87d44d3de849463abf1" +dependencies = [ + "bitflags 2.9.0", + "rustc_version", +] + [[package]] name = "flate2" version = "1.1.1" @@ -3272,6 +3303,17 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -3843,9 +3885,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.8" +version = "0.4.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5017294ff4bb30944501348f6f8e42e6ad28f42c8bbef7a74029aff064a4e3c2" +checksum = "75249d144030531f8dee69fe9cea04d3edf809a017ae445e2abdff6629e86633" dependencies = [ "atomic-waker", "bytes", @@ -4087,10 +4129,10 @@ dependencies = [ "http", "hyper", "hyper-util", - "rustls", + "rustls 0.23.26", "rustls-pki-types", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.2", "tower-service", "webpki-roots", ] @@ -4334,6 +4376,7 @@ checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99" dependencies = [ "autocfg", "hashbrown 0.12.3", + "serde", ] [[package]] @@ -4686,9 +4729,9 @@ dependencies = [ [[package]] name = "libc" -version = "0.2.171" +version = "0.2.172" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c19937216e9d3aa9956d9bb8dfc0b0c8beb6058fc4f7a4dc4d850edf86a237d6" +checksum = "d750af042f7ef4f724306de029d18836c26c1765a54a6a3f094cbd23a7267ffa" [[package]] name = "libdbus-sys" @@ -4737,19 +4780,19 @@ dependencies = [ [[package]] name = "libsystemd" -version = "0.7.0" +version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c592dc396b464005f78a5853555b9f240bc5378bf5221acc4e129910b2678869" +checksum = "b85fe9dc49de659d05829fdf72b5770c0a5952d1055c34a39f6d4e932bce175d" dependencies = [ "hmac 0.12.1", "libc", "log", - "nix 0.27.1", - "nom", + "nix", + "nom 8.0.0", "once_cell", "serde", "sha2 0.10.8", - "thiserror 1.0.69", + "thiserror 2.0.12", "uuid", ] @@ -5169,18 +5212,6 @@ version = "1.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086" -[[package]] -name = "nix" -version = "0.27.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2eb04e9c688eff1c89d72b407f168cf79bb9e867a9d3323ed6c01519eb9cc053" -dependencies = [ - "bitflags 2.9.0", - "cfg-if", - "libc", - "memoffset", -] - [[package]] name = "nix" version = "0.29.0" @@ -5210,6 +5241,15 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "nom" +version = "8.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405" +dependencies = [ + "memchr", +] + [[package]] name = "nu-ansi-term" version = "0.46.0" @@ -5411,9 +5451,9 @@ dependencies = [ [[package]] name = "objc2" -version = "0.6.0" +version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3531f65190d9cff863b77a99857e74c314dd16bf56c538c4b57c7cbc3f3a6e59" +checksum = "88c6597e14493ab2e44ce58f2fdecf095a51f12ca57bec060a11c57332520551" dependencies = [ "objc2-encode", ] @@ -5436,15 +5476,15 @@ dependencies = [ [[package]] name = "objc2-app-kit" -version = "0.3.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5906f93257178e2f7ae069efb89fbd6ee94f0592740b5f8a1512ca498814d0fb" +checksum = "e6f29f568bec459b0ddff777cec4fe3fd8666d82d5a40ebd0ff7e66134f89bcc" dependencies = [ "bitflags 2.9.0", - "block2 0.6.0", - "objc2 0.6.0", + "block2 0.6.1", + "objc2 0.6.1", "objc2-core-foundation", - "objc2-foundation 0.3.0", + "objc2-foundation 0.3.1", ] [[package]] @@ -5461,19 +5501,20 @@ dependencies = [ [[package]] name = "objc2-core-foundation" -version = "0.3.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "daeaf60f25471d26948a1c2f840e3f7d86f4109e3af4e8e4b5cd70c39690d925" +checksum = "1c10c2894a6fed806ade6027bcd50662746363a9589d3ec9d9bef30a4e4bc166" dependencies = [ "bitflags 2.9.0", - "objc2 0.6.0", + "dispatch2 0.3.0", + "objc2 0.6.1", ] [[package]] name = "objc2-core-graphics" -version = "0.3.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8dca602628b65356b6513290a21a6405b4d4027b8b250f0b98dddbb28b7de02" +checksum = "989c6c68c13021b5c2d6b71456ebb0f9dc78d752e86a98da7c716f4f9470f5a4" dependencies = [ "bitflags 2.9.0", "objc2-core-foundation", @@ -5511,13 +5552,13 @@ dependencies = [ [[package]] name = "objc2-foundation" -version = "0.3.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a21c6c9014b82c39515db5b396f91645182611c97d24637cf56ac01e5f8d998" +checksum = "900831247d2fe1a09a683278e5384cfb8c80c79fe6b166f9d14bfdde0ea1b03c" dependencies = [ "bitflags 2.9.0", - "block2 0.6.0", - "objc2 0.6.0", + "block2 0.6.1", + "objc2 0.6.1", "objc2-core-foundation", ] @@ -5606,6 +5647,12 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" +[[package]] +name = "openssl-probe" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" + [[package]] name = "opentelemetry" version = "0.29.1" @@ -5723,7 +5770,7 @@ dependencies = [ "glob", "opentelemetry", "percent-encoding", - "rand 0.9.0", + "rand 0.9.1", "serde_json", "thiserror 2.0.12", "tokio", @@ -6420,9 +6467,9 @@ checksum = "dc375e1527247fe1a97d8b7156678dfe7c1af2fc075c9a4db3690ecd2a148068" [[package]] name = "proc-macro2" -version = "1.0.94" +version = "1.0.95" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a31971752e70b8b2686d7e46ec17fb38dad4051d94024c88df49b667caea9c84" +checksum = "02b3e5e68a3a1a02aad3ec490a98007cbc13c37cbe84a3cd7b8e406d76e7f778" dependencies = [ "unicode-ident", ] @@ -6531,7 +6578,7 @@ name = "protos" version = "0.0.1" dependencies = [ "common", - "flatbuffers", + "flatbuffers 25.2.10", "prost", "prost-build", "protobuf", @@ -6590,7 +6637,7 @@ dependencies = [ "quinn-proto", "quinn-udp", "rustc-hash 2.1.1", - "rustls", + "rustls 0.23.26", "socket2", "thiserror 2.0.12", "tokio", @@ -6606,10 +6653,10 @@ checksum = "b820744eb4dc9b57a3398183639c511b5a26d2ed702cedd3febaa1393caa22cc" dependencies = [ "bytes", "getrandom 0.3.2", - "rand 0.9.0", + "rand 0.9.1", "ring", "rustc-hash 2.1.1", - "rustls", + "rustls 0.23.26", "rustls-pki-types", "slab", "thiserror 2.0.12", @@ -6674,13 +6721,12 @@ dependencies = [ [[package]] name = "rand" -version = "0.9.0" +version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3779b94aeb87e8bd4e834cee3650289ee9e0d5677f976ecdb6d219e5f4f6cd94" +checksum = "9fbfd9d094a40bf3ae768db9361049ace4c0e04a4fd6b359518bd7b73a73dd97" dependencies = [ "rand_chacha 0.9.0", "rand_core 0.9.3", - "zerocopy 0.8.24", ] [[package]] @@ -6838,17 +6884,6 @@ dependencies = [ "bitflags 2.9.0", ] -[[package]] -name = "redox_users" -version = "0.4.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43" -dependencies = [ - "getrandom 0.2.15", - "libredox", - "thiserror 1.0.69", -] - [[package]] name = "redox_users" version = "0.5.0" @@ -6947,7 +6982,7 @@ dependencies = [ "percent-encoding", "pin-project-lite", "quinn", - "rustls", + "rustls 0.23.26", "rustls-pemfile", "rustls-pki-types", "serde", @@ -6956,7 +6991,7 @@ dependencies = [ "sync_wrapper", "system-configuration", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.2", "tokio-util", "tower 0.5.2", "tower-service", @@ -6999,14 +7034,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "80c844748fdc82aae252ee4594a89b6e7ebef1063de7951545564cbc4e57075d" dependencies = [ "ashpd 0.11.0", - "block2 0.6.0", - "dispatch2", + "block2 0.6.1", + "dispatch2 0.2.0", "js-sys", "log", - "objc2 0.6.0", - "objc2-app-kit 0.3.0", + "objc2 0.6.1", + "objc2-app-kit 0.3.1", "objc2-core-foundation", - "objc2-foundation 0.3.0", + "objc2-foundation 0.3.1", "pollster 0.4.0", "raw-window-handle 0.6.2", "urlencoding", @@ -7084,6 +7119,24 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rumqttc" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1568e15fab2d546f940ed3a21f48bbbd1c494c90c99c4481339364a497f94a9" +dependencies = [ + "bytes", + "flume", + "futures-util", + "log", + "rustls-native-certs", + "rustls-pemfile", + "rustls-webpki 0.102.8", + "thiserror 1.0.69", + "tokio", + "tokio-rustls 0.25.0", +] + [[package]] name = "rust-embed" version = "8.7.0" @@ -7159,7 +7212,7 @@ dependencies = [ [[package]] name = "rustfs" -version = "0.1.0" +version = "0.0.1" dependencies = [ "api", "appauth", @@ -7177,7 +7230,7 @@ dependencies = [ "crypto", "datafusion", "ecstore", - "flatbuffers", + "flatbuffers 25.2.10", "futures", "futures-util", "http", @@ -7201,8 +7254,9 @@ dependencies = [ "query", "rmp-serde", "rust-embed", + "rustfs-event-notifier", "rustfs-obs", - "rustls", + "rustls 0.23.26", "rustls-pemfile", "rustls-pki-types", "s3s", @@ -7213,7 +7267,7 @@ dependencies = [ "tikv-jemallocator", "time", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.2", "tokio-stream", "tokio-util", "tonic 0.13.0", @@ -7225,13 +7279,37 @@ dependencies = [ "uuid", ] +[[package]] +name = "rustfs-event-notifier" +version = "0.0.1" +dependencies = [ + "async-trait", + "axum", + "config", + "http", + "rdkafka", + "reqwest", + "rumqttc", + "serde", + "serde_json", + "serde_with", + "smallvec", + "strum", + "thiserror 2.0.12", + "tokio", + "tokio-util", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "rustfs-gui" version = "0.0.1" dependencies = [ "chrono", "dioxus", - "dirs 6.0.0", + "dirs", "hex", "keyring", "lazy_static", @@ -7300,6 +7378,20 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "rustls" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf4ef73721ac7bcd79b2b315da7779d8fc09718c6b3d2d1b2d94850eb8c18432" +dependencies = [ + "log", + "ring", + "rustls-pki-types", + "rustls-webpki 0.102.8", + "subtle", + "zeroize", +] + [[package]] name = "rustls" version = "0.23.26" @@ -7311,11 +7403,24 @@ dependencies = [ "once_cell", "ring", "rustls-pki-types", - "rustls-webpki", + "rustls-webpki 0.103.1", "subtle", "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5bfb394eeed242e909609f56089eecfe5fda225042e8b171791b9c95f5931e5" +dependencies = [ + "openssl-probe", + "rustls-pemfile", + "rustls-pki-types", + "schannel", + "security-framework 2.11.1", +] + [[package]] name = "rustls-pemfile" version = "2.2.0" @@ -7334,6 +7439,17 @@ dependencies = [ "web-time", ] +[[package]] +name = "rustls-webpki" +version = "0.102.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustls-webpki" version = "0.103.1" @@ -7386,7 +7502,7 @@ dependencies = [ "md-5", "memchr", "mime", - "nom", + "nom 7.1.3", "nugine-rust-utils", "numeric_cast", "pin-project-lite", @@ -7428,6 +7544,15 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "schannel" +version = "0.1.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f29ebaa345f945cec9fbbc532eb307f0fdad8161f281b6369539c8d84876b3d" +dependencies = [ + "windows-sys 0.59.0", +] + [[package]] name = "scopeguard" version = "1.2.0" @@ -7495,6 +7620,9 @@ name = "semver" version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56e6fa9c48d24d85fb3de5ad847117517440f6beceb7798af16b4a87d616b8d0" +dependencies = [ + "serde", +] [[package]] name = "send_wrapper" @@ -7607,6 +7735,36 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_with" +version = "3.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6b6f7f2fcb69f747921f79f3926bd1e203fce4fef62c268dd3abfb6d86029aa" +dependencies = [ + "base64 0.22.1", + "chrono", + "hex", + "indexmap 1.9.3", + "indexmap 2.9.0", + "serde", + "serde_derive", + "serde_json", + "serde_with_macros", + "time", +] + +[[package]] +name = "serde_with_macros" +version = "3.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d00caa5193a3c8362ac2b73be6b9e768aa5a4b2f721d8f4b339600c3cb51f8e" +dependencies = [ + "darling", + "proc-macro2", + "quote", + "syn 2.0.100", +] + [[package]] name = "server_fn" version = "0.6.15" @@ -7716,13 +7874,16 @@ dependencies = [ [[package]] name = "shadow-rs" -version = "0.38.1" +version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ec14cc798c29f4bf74a6c4299c657c04d4e9fba03875c1f0eec569af03aed89" +checksum = "6d5625ed609cf66d7e505e7d487aca815626dc4ebb6c0dd07637ca61a44651a6" dependencies = [ + "cargo_metadata", "const_format", "is_debug", + "serde_json", "time", + "tzdb", ] [[package]] @@ -7736,11 +7897,11 @@ dependencies = [ [[package]] name = "shellexpand" -version = "3.1.0" +version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da03fa3b94cc19e3ebfc88c4229c49d8f08cdbd1228870a45f0ffdf84988e14b" +checksum = "8b1fdf65dd6331831494dd616b30351c38e96e45921a27745cf98490458b90bb" dependencies = [ - "dirs 5.0.1", + "dirs", ] [[package]] @@ -7761,9 +7922,9 @@ dependencies = [ [[package]] name = "signal-hook-registry" -version = "1.4.2" +version = "1.4.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9e9e0b4211b72e7b8b6e85c807d36c212bdb33ea8587f7569562a84df5465b1" +checksum = "9203b8055f63a2a00e2f593bb0510367fe707d7ff1e5c872de2f537b339e5410" dependencies = [ "libc", ] @@ -7867,6 +8028,9 @@ name = "smallvec" version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8917285742e9f3e1683f0a9c4e6b57960b7314d0b08d30d1ecd426713ee2eee9" +dependencies = [ + "serde", +] [[package]] name = "snafu" @@ -7937,6 +8101,9 @@ name = "spin" version = "0.9.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] [[package]] name = "spki" @@ -8438,13 +8605,24 @@ dependencies = [ "syn 2.0.100", ] +[[package]] +name = "tokio-rustls" +version = "0.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "775e0c0f0adb3a2f22a00c4745d728b479985fc15ee7ca6a2608388c5569860f" +dependencies = [ + "rustls 0.22.4", + "rustls-pki-types", + "tokio", +] + [[package]] name = "tokio-rustls" version = "0.26.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e727b36a1a0e8b74c376ac2211e40c2c8af09fb4013c60d910495810f008e9b" dependencies = [ - "rustls", + "rustls 0.23.26", "tokio", ] @@ -8814,14 +8992,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eadd75f5002e2513eaa19b2365f533090cc3e93abd38788452d9ea85cff7b48a" dependencies = [ "crossbeam-channel", - "dirs 6.0.0", + "dirs", "libappindicator", "muda 0.15.3", - "objc2 0.6.0", - "objc2-app-kit 0.3.0", + "objc2 0.6.1", + "objc2-app-kit 0.3.1", "objc2-core-foundation", "objc2-core-graphics", - "objc2-foundation 0.3.0", + "objc2-foundation 0.3.1", "once_cell", "png", "thiserror 2.0.12", @@ -8874,6 +9052,32 @@ version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1dccffe3ce07af9386bfd29e80c0ab1a8205a2fc34e4bcd40364df902cfa8f3f" +[[package]] +name = "tz-rs" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1450bf2b99397e72070e7935c89facaa80092ac812502200375f1f7d33c71a1" + +[[package]] +name = "tzdb" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0be2ea5956f295449f47c0b825c5e109022ff1a6a53bb4f77682a87c2341fbf5" +dependencies = [ + "iana-time-zone", + "tz-rs", + "tzdb_data", +] + +[[package]] +name = "tzdb_data" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c4c81d75033770e40fbd3643ce7472a1a9fd301f90b7139038228daf8af03ec" +dependencies = [ + "tz-rs", +] + [[package]] name = "ucd-trie" version = "0.1.7" @@ -8987,7 +9191,7 @@ checksum = "458f7a779bf54acc9f347480ac654f68407d3aab21269a6e3c9f922acd9e2da9" dependencies = [ "getrandom 0.3.2", "js-sys", - "rand 0.9.0", + "rand 0.9.1", "serde", "uuid-macro-internal", "wasm-bindgen", @@ -9964,7 +10168,7 @@ dependencies = [ "futures-sink", "futures-util", "hex", - "nix 0.29.0", + "nix", "ordered-stream", "rand 0.8.5", "serde", @@ -9995,7 +10199,7 @@ dependencies = [ "futures-core", "futures-lite", "hex", - "nix 0.29.0", + "nix", "ordered-stream", "serde", "serde_repr", diff --git a/Cargo.toml b/Cargo.toml index e2ef3b562..8ba1d7198 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,20 +1,21 @@ [workspace] members = [ - "madmin", # Management dashboard and admin API interface - "rustfs", # Core file system implementation - "ecstore", # Erasure coding storage implementation - "e2e_test", # End-to-end test suite - "common/common", # Shared utilities and data structures - "common/lock", # Distributed locking implementation - "common/protos", # Protocol buffer definitions + "madmin", # Management dashboard and admin API interface + "rustfs", # Core file system implementation + "ecstore", # Erasure coding storage implementation + "e2e_test", # End-to-end test suite + "common/common", # Shared utilities and data structures + "common/lock", # Distributed locking implementation + "common/protos", # Protocol buffer definitions "common/workers", # Worker thread pools and task scheduling - "iam", # Identity and Access Management - "crypto", # Cryptography and security features + "iam", # Identity and Access Management + "crypto", # Cryptography and security features "cli/rustfs-gui", # Graphical user interface client - "crates/obs", # Observability utilities - "s3select/api", - "s3select/query", - "appauth", + "crates/obs", # Observability utilities + "crates/event-notifier", # Event notification system + "s3select/api", # S3 Select API interface + "s3select/query", # S3 Select query engine + "appauth", # Application authentication and authorization ] resolver = "2" @@ -32,7 +33,21 @@ unsafe_code = "deny" all = "warn" [workspace.dependencies] -madmin = { path = "./madmin" } +api = { path = "./s3select/api", version = "0.0.1" } +appauth = { path = "./appauth", version = "0.0.1" } +common = { path = "./common/common", version = "0.0.1" } +crypto = { path = "./crypto", version = "0.0.1" } +ecstore = { path = "./ecstore", version = "0.0.1" } +iam = { path = "./iam", version = "0.0.1" } +lock = { path = "./common/lock", version = "0.0.1" } +madmin = { path = "./madmin", version = "0.0.1" } +policy = { path = "./policy", version = "0.0.1" } +protos = { path = "./common/protos", version = "0.0.1" } +query = { path = "./s3select/query", version = "0.0.1" } +rustfs = { path = "./rustfs", version = "0.0.1" } +rustfs-obs = { path = "crates/obs", version = "0.0.1" } +rustfs-event-notifier = { path = "crates/event-notifier", version = "0.0.1" } +workers = { path = "./common/workers", version = "0.0.1" } atoi = "2.0.0" async-recursion = "1.1.1" async-trait = "0.1.88" @@ -42,20 +57,18 @@ axum-extra = "0.10.1" axum-server = { version = "0.7.2", features = ["tls-rustls"] } backon = "1.5.0" bytes = "1.10.1" -bytesize = "1.3.3" +bytesize = "2.0.1" chrono = { version = "0.4.40", features = ["serde"] } -clap = { version = "4.5.35", features = ["derive", "env"] } +clap = { version = "4.5.37", features = ["derive", "env"] } config = "0.15.11" -datafusion = "46.0.0" +datafusion = "46.0.1" derive_builder = "0.20.2" dioxus = { version = "0.6.3", features = ["router"] } dirs = "6.0.0" -ecstore = { path = "./ecstore" } -flatbuffers = "24.12.23" +flatbuffers = "25.2.10" futures = "0.3.31" +futures-core = "0.3.31" futures-util = "0.3.31" -common = { path = "./common/common" } -policy = { path = "./policy" } hex = "0.4.3" hyper = "1.6.0" hyper-util = { version = "0.1.11", features = [ @@ -67,67 +80,51 @@ http = "1.3.1" http-body = "1.0.1" humantime = "2.2.0" jsonwebtoken = "9.3.1" -keyring = { version = "3.6.2", features = [ - "apple-native", - "windows-native", - "sync-secret-service", -] } -lock = { path = "./common/lock" } +keyring = { version = "3.6.2", features = ["apple-native", "windows-native", "sync-secret-service"] } lazy_static = "1.5.0" -libsystemd = { version = "0.7" } +libsystemd = { version = "0.7.1" } local-ip-address = "0.6.3" matchit = "0.8.4" md-5 = "0.10.6" mime = "0.3.17" +mime_guess = "2.0.5" netif = "0.1.6" +object_store = "0.11.2" opentelemetry = { version = "0.29.1" } -opentelemetry-appender-tracing = { version = "0.29.1", features = [ - "experimental_use_tracing_span_context", - "experimental_metadata_attributes", -] } -opentelemetry_sdk = { version = "0.29" } +opentelemetry-appender-tracing = { version = "0.29.1", features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] } +opentelemetry_sdk = { version = "0.29.0" } opentelemetry-stdout = { version = "0.29.0" } -opentelemetry-otlp = { version = "0.29" } +opentelemetry-otlp = { version = "0.29.0" } opentelemetry-prometheus = { version = "0.29.1" } -opentelemetry-semantic-conventions = { version = "0.29.0", features = [ - "semconv_experimental", -] } -pin-project-lite = "0.2" +opentelemetry-semantic-conventions = { version = "0.29.0", features = ["semconv_experimental"] } +parking_lot = "0.12.3" +pin-project-lite = "0.2.16" prometheus = "0.14.0" # pin-utils = "0.1.0" prost = "0.13.5" prost-build = "0.13.5" prost-types = "0.13.5" protobuf = "3.7" -protos = { path = "./common/protos" } rand = "0.8.5" -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.3", default-features = false, features = [ - "xdg-portal", - "tokio", -] } +rdkafka = { version = "0.37.0", 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.3", default-features = false, features = ["xdg-portal", "tokio"] } rmp = "0.8.14" rmp-serde = "1.3.0" -rustfs-obs = { path = "crates/obs", version = "0.0.1" } +rumqttc = { version = "0.24" } rust-embed = "8.7.0" rustls = { version = "0.23.26" } rustls-pki-types = "1.11.0" rustls-pemfile = "2.2.0" s3s = { git = "https://github.com/Nugine/s3s.git", rev = "4733cdfb27b2713e832967232cbff413bb768c10" } s3s-policy = { git = "https://github.com/Nugine/s3s.git", rev = "4733cdfb27b2713e832967232cbff413bb768c10" } -shadow-rs = { version = "0.38.1", default-features = false } +shadow-rs = { version = "1.1.1", default-features = false } serde = { version = "1.0.219", features = ["derive"] } serde_json = "1.0.140" serde_urlencoded = "0.7.1" +serde_with = "3.12.0" +smallvec = { version = "1.15.0", features = ["serde"] } +strum = { version = "0.27.1", features = ["derive"] } sha2 = "0.10.8" snafu = "0.8.5" tempfile = "3.19.1" @@ -143,7 +140,7 @@ time = { version = "0.3.41", features = [ tokio = { version = "1.44.2", features = ["fs", "rt-multi-thread"] } tonic = { version = "0.13.0", features = ["gzip"] } tonic-build = "0.13.0" -tokio-rustls = { version = "0.26", default-features = false } +tokio-rustls = { version = "0.26.2", default-features = false } tokio-stream = "0.1.17" tokio-util = { version = "0.7.14", features = ["io", "compat"] } tower = { version = "0.5.2", features = ["timeout"] } @@ -153,7 +150,7 @@ 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" -tracing-opentelemetry = "0.30" +tracing-opentelemetry = "0.30.0" transform-stream = "0.3.1" url = "2.5.4" uuid = { version = "1.16.0", features = [ @@ -161,7 +158,7 @@ uuid = { version = "1.16.0", features = [ "fast-rng", "macro-diagnostics", ] } -workers = { path = "./common/workers" } + [profile.wasm-dev] inherits = "dev" diff --git a/crates/event-notifier/Cargo.toml b/crates/event-notifier/Cargo.toml new file mode 100644 index 000000000..5d1d33906 --- /dev/null +++ b/crates/event-notifier/Cargo.toml @@ -0,0 +1,40 @@ +[package] +name = "rustfs-event-notifier" +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[features] +default = ["webhook"] +webhook = ["dep:reqwest"] +kafka = ["rdkafka"] +mqtt = ["rumqttc"] + +[dependencies] +async-trait = { workspace = true } +config = { workspace = true } +rdkafka = { workspace = true, features = ["tokio"], optional = true } +reqwest = { workspace = true, optional = true } +rumqttc = { workspace = true, optional = true } +serde = { workspace = true } +serde_json = { workspace = true } +serde_with = { workspace = true } +smallvec = { workspace = true, features = ["serde"] } +strum = { workspace = true, features = ["derive"] } +tracing = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["sync", "net", "macros", "signal", "rt-multi-thread"] } +tokio-util = { workspace = true } +uuid = { workspace = true, features = ["v4", "serde"] } + + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } +tracing-subscriber = { workspace = true } +http = { workspace = true } +axum = { workspace = true } + +[lints] +workspace = true diff --git a/crates/event-notifier/examples/.env-zh.example b/crates/event-notifier/examples/.env-zh.example new file mode 100644 index 000000000..00228e162 --- /dev/null +++ b/crates/event-notifier/examples/.env-zh.example @@ -0,0 +1,28 @@ +# ===== 全局配置 ===== +NOTIFIER__STORE_PATH=/var/log/event-notification +NOTIFIER__CHANNEL_CAPACITY=5000 + +# ===== 适配器配置（数组格式） ===== +# Webhook 适配器（索引 0） +NOTIFIER__ADAPTERS_0__type=Webhook +NOTIFIER__ADAPTERS_0__endpoint=http://127.0.0.1:3020/webhook +NOTIFIER__ADAPTERS_0__auth_token=your-auth-token +NOTIFIER__ADAPTERS_0__max_retries=3 +NOTIFIER__ADAPTERS_0__timeout=50 +NOTIFIER__ADAPTERS_0__custom_headers__x_custom_server=value +NOTIFIER__ADAPTERS_0__custom_headers__x_custom_client=value + +# Kafka 适配器（索引 1） +NOTIFIER__ADAPTERS_1__type=Kafka +NOTIFIER__ADAPTERS_1__brokers=localhost:9092 +NOTIFIER__ADAPTERS_1__topic=notifications +NOTIFIER__ADAPTERS_1__max_retries=3 +NOTIFIER__ADAPTERS_1__timeout=60 + +# MQTT 适配器（索引 2） +NOTIFIER__ADAPTERS_2__type=Mqtt +NOTIFIER__ADAPTERS_2__broker=mqtt.example.com +NOTIFIER__ADAPTERS_2__port=1883 +NOTIFIER__ADAPTERS_2__client_id=event-notifier +NOTIFIER__ADAPTERS_2__topic=events +NOTIFIER__ADAPTERS_2__max_retries=3 \ No newline at end of file diff --git a/crates/event-notifier/examples/.env.example b/crates/event-notifier/examples/.env.example new file mode 100644 index 000000000..af343863d --- /dev/null +++ b/crates/event-notifier/examples/.env.example @@ -0,0 +1,28 @@ +# ===== global configuration ===== +NOTIFIER__STORE_PATH=/var/log/event-notification +NOTIFIER__CHANNEL_CAPACITY=5000 + +# ===== adapter configuration array format ===== +# webhook adapter index 0 +NOTIFIER__ADAPTERS_0__type=Webhook +NOTIFIER__ADAPTERS_0__endpoint=http://127.0.0.1:3020/webhook +NOTIFIER__ADAPTERS_0__auth_token=your-auth-token +NOTIFIER__ADAPTERS_0__max_retries=3 +NOTIFIER__ADAPTERS_0__timeout=50 +NOTIFIER__ADAPTERS_0__custom_headers__x_custom_server=server-value +NOTIFIER__ADAPTERS_0__custom_headers__x_custom_client=client-value + +# kafka adapter index 1 +NOTIFIER__ADAPTERS_1__type=Kafka +NOTIFIER__ADAPTERS_1__brokers=localhost:9092 +NOTIFIER__ADAPTERS_1__topic=notifications +NOTIFIER__ADAPTERS_1__max_retries=3 +NOTIFIER__ADAPTERS_1__timeout=60 + +# mqtt adapter index 2 +NOTIFIER__ADAPTERS_2__type=Mqtt +NOTIFIER__ADAPTERS_2__broker=mqtt.example.com +NOTIFIER__ADAPTERS_2__port=1883 +NOTIFIER__ADAPTERS_2__client_id=event-notifier +NOTIFIER__ADAPTERS_2__topic=events +NOTIFIER__ADAPTERS_2__max_retries=3 \ No newline at end of file diff --git a/crates/event-notifier/examples/event.toml b/crates/event-notifier/examples/event.toml new file mode 100644 index 000000000..5b4292fae --- /dev/null +++ b/crates/event-notifier/examples/event.toml @@ -0,0 +1,29 @@ +# config.toml +store_path = "/var/log/event-notifier" +channel_capacity = 5000 + +[[adapters]] +type = "Webhook" +endpoint = "http://127.0.0.1:3020/webhook" +auth_token = "your-auth-token" +max_retries = 3 +timeout = 50 + +[adapters.custom_headers] +custom_server = "value_server" +custom_client = "value_client" + +[[adapters]] +type = "Kafka" +brokers = "localhost:9092" +topic = "notifications" +max_retries = 3 +timeout = 60 + +[[adapters]] +type = "Mqtt" +broker = "mqtt.example.com" +port = 1883 +client_id = "event-notifier" +topic = "events" +max_retries = 3 \ No newline at end of file diff --git a/crates/event-notifier/examples/full.rs b/crates/event-notifier/examples/full.rs new file mode 100644 index 000000000..23e858fb8 --- /dev/null +++ b/crates/event-notifier/examples/full.rs @@ -0,0 +1,133 @@ +use rustfs_event_notifier::{ + AdapterConfig, Bucket, Error as NotifierError, Event, Identity, Metadata, Name, NotifierConfig, Object, Source, WebhookConfig, +}; +use std::collections::HashMap; +use tokio::signal; +use tracing::Level; +use tracing_subscriber::FmtSubscriber; + +async fn setup_notification_system() -> Result<(), NotifierError> { + let config = NotifierConfig { + store_path: "./deploy/logs/event_store".into(), + channel_capacity: 100, + adapters: vec![AdapterConfig::Webhook(WebhookConfig { + endpoint: "http://127.0.0.1:3020/webhook".into(), + auth_token: Some("your-auth-token".into()), + custom_headers: Some(HashMap::new()), + max_retries: 3, + timeout: 30, + })], + }; + + rustfs_event_notifier::initialize(config).await?; + + // wait for the system to be ready + for _ in 0..50 { + // wait up to 5 seconds + if rustfs_event_notifier::is_ready() { + return Ok(()); + } + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + } + + Err(NotifierError::custom("notify the system of initialization timeout")) +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + // initialization log + // tracing_subscriber::fmt::init(); + + let subscriber = FmtSubscriber::builder() + .with_max_level(Level::DEBUG) // set to debug or lower level + .with_target(false) // simplify output + .finish(); + tracing::subscriber::set_global_default(subscriber).expect("failed to set up log subscriber"); + + // set up notification system + if let Err(e) = setup_notification_system().await { + eprintln!("unable to initialize notification system:{}", e); + return Err(e.into()); + } + + // create a shutdown signal processing + let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel(); + + // start signal processing task + tokio::spawn(async move { + let _ = signal::ctrl_c().await; + println!("Received the shutdown signal and prepared to exit..."); + let _ = shutdown_tx.send(()); + }); + + // main application logic + tokio::select! { + _ = async { + loop { + // application logic + // create an s3 metadata object + let metadata = Metadata { + schema_version: "1.0".to_string(), + configuration_id: "test-config".to_string(), + bucket: Bucket { + name: "my-bucket".to_string(), + owner_identity: Identity { + principal_id: "owner123".to_string(), + }, + arn: "arn:aws:s3:::my-bucket".to_string(), + }, + object: Object { + key: "test.txt".to_string(), + size: Some(1024), + etag: Some("abc123".to_string()), + content_type: Some("text/plain".to_string()), + user_metadata: None, + version_id: None, + sequencer: "1234567890".to_string(), + }, + }; + + // create source object + let source = Source { + host: "localhost".to_string(), + port: "80".to_string(), + user_agent: "curl/7.68.0".to_string(), + }; + + // create events using builder mode + let event = Event::builder() + .event_time("2023-10-01T12:00:00.000Z") + .event_name(Name::ObjectCreatedPut) + .user_identity(Identity { + principal_id: "user123".to_string(), + }) + .s3(metadata) + .source(source) + .channels(vec!["webhook".to_string()]) + .build() + .expect("failed to create event"); + + if let Err(e) = rustfs_event_notifier::send_event(event).await { + eprintln!("send event failed:{}", e); + } + + tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; + } + } => {}, + + _ = &mut shutdown_rx => { + println!("close the app"); + } + } + + // 优雅关闭通知系统 + println!("turn off the notification system"); + if let Err(e) = rustfs_event_notifier::shutdown().await { + eprintln!("An error occurred while shutting down the notification system:{}", e); + } else { + println!("the notification system has been closed safely"); + } + + println!("the application has been closed safely"); + Ok(()) +} diff --git a/crates/event-notifier/examples/simple.rs b/crates/event-notifier/examples/simple.rs new file mode 100644 index 000000000..93005f0a7 --- /dev/null +++ b/crates/event-notifier/examples/simple.rs @@ -0,0 +1,108 @@ +use rustfs_event_notifier::create_adapters; +use rustfs_event_notifier::NotifierSystem; +use rustfs_event_notifier::{AdapterConfig, NotifierConfig, WebhookConfig}; +use rustfs_event_notifier::{Bucket, Event, Identity, Metadata, Name, Object, Source}; +use std::collections::HashMap; +use std::error; +use std::sync::Arc; +use tokio::signal; +use tracing::Level; +use tracing_subscriber::FmtSubscriber; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let subscriber = FmtSubscriber::builder() + .with_max_level(Level::DEBUG) // set to debug or lower level + .with_target(false) // simplify output + .finish(); + tracing::subscriber::set_global_default(subscriber).expect("failed to set up log subscriber"); + + let config = NotifierConfig { + store_path: "./events".to_string(), + channel_capacity: 100, + adapters: vec![AdapterConfig::Webhook(WebhookConfig { + endpoint: "http://127.0.0.1:3020/webhook".to_string(), + auth_token: Some("secret-token".to_string()), + custom_headers: Some(HashMap::from([("X-Custom".to_string(), "value".to_string())])), + max_retries: 3, + timeout: 10, + })], + }; + + // event_load_config + // loading configuration from environment variables + let _config = NotifierConfig::event_load_config(Some("./crates/event-notifier/examples/event.toml".to_string())); + tracing::info!("event_load_config config: {:?} \n", _config); + + let system = Arc::new(tokio::sync::Mutex::new(NotifierSystem::new(config.clone()).await?)); + let adapters = create_adapters(&config.adapters)?; + + // create an s3 metadata object + let metadata = Metadata { + schema_version: "1.0".to_string(), + configuration_id: "test-config".to_string(), + bucket: Bucket { + name: "my-bucket".to_string(), + owner_identity: Identity { + principal_id: "owner123".to_string(), + }, + arn: "arn:aws:s3:::my-bucket".to_string(), + }, + object: Object { + key: "test.txt".to_string(), + size: Some(1024), + etag: Some("abc123".to_string()), + content_type: Some("text/plain".to_string()), + user_metadata: None, + version_id: None, + sequencer: "1234567890".to_string(), + }, + }; + + // create source object + let source = Source { + host: "localhost".to_string(), + port: "80".to_string(), + user_agent: "curl/7.68.0".to_string(), + }; + + // create events using builder mode + let event = Event::builder() + .event_time("2023-10-01T12:00:00.000Z") + .event_name(Name::ObjectCreatedPut) + .user_identity(Identity { + principal_id: "user123".to_string(), + }) + .s3(metadata) + .source(source) + .channels(vec!["webhook".to_string()]) + .build() + .expect("failed to create event"); + + { + let system = system.lock().await; + system.send_event(event).await?; + } + + let system_clone = Arc::clone(&system); + let system_handle = tokio::spawn(async move { + let mut system = system_clone.lock().await; + system.start(adapters).await + }); + + signal::ctrl_c().await?; + tracing::info!("Received shutdown signal"); + let result = { + let mut system = system.lock().await; + system.shutdown().await + }; + + if let Err(e) = result { + tracing::error!("Failed to shut down the notification system: {}", e); + } else { + tracing::info!("Notification system shut down successfully"); + } + + system_handle.await??; + Ok(()) +} diff --git a/crates/event-notifier/examples/webhook.rs b/crates/event-notifier/examples/webhook.rs new file mode 100644 index 000000000..c754f8223 --- /dev/null +++ b/crates/event-notifier/examples/webhook.rs @@ -0,0 +1,94 @@ +use axum::{extract::Json, http::StatusCode, routing::post, Router}; +use serde_json::Value; +use std::time::{SystemTime, UNIX_EPOCH}; + +#[tokio::main] +async fn main() { + // 构建应用 + let app = Router::new().route("/webhook", post(receive_webhook)); + // 启动服务器 + let listener = tokio::net::TcpListener::bind("0.0.0.0:3020").await.unwrap(); + println!("Server running on http://0.0.0.0:3020"); + + // 创建关闭信号处理 + tokio::select! { + result = axum::serve(listener, app) => { + if let Err(e) = result { + eprintln!("Server error: {}", e); + } + } + _ = tokio::signal::ctrl_c() => { + println!("Shutting down server..."); + } + } +} + +async fn receive_webhook(Json(payload): Json) -> StatusCode { + let start = SystemTime::now(); + let since_the_epoch = start.duration_since(UNIX_EPOCH).expect("Time went backwards"); + + // get the number of seconds since the unix era + let seconds = since_the_epoch.as_secs(); + + // Manually calculate year, month, day, hour, minute, and second + let (year, month, day, hour, minute, second) = convert_seconds_to_date(seconds); + + // output result + println!("current time:{:04}-{:02}-{:02} {:02}:{:02}:{:02}", year, month, day, hour, minute, second); + println!( + "received a webhook request time:{} content:\n {}", + seconds.to_string(), + serde_json::to_string_pretty(&payload).unwrap() + ); + StatusCode::OK +} + +fn convert_seconds_to_date(seconds: u64) -> (u32, u32, u32, u32, u32, u32) { + // assume that the time zone is utc + let seconds_per_minute = 60; + let seconds_per_hour = 3600; + let seconds_per_day = 86400; + + // Calculate the year, month, day, hour, minute, and second corresponding to the number of seconds + let mut total_seconds = seconds; + let mut year = 1970; + let mut month = 1; + let mut day = 1; + let mut hour = 0; + let mut minute = 0; + let mut second = 0; + + // calculate year + while total_seconds >= 31536000 { + year += 1; + total_seconds -= 31536000; // simplified processing no leap year considered + } + + // calculate month + let days_in_month = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]; + for m in 0..12 { + if total_seconds >= days_in_month[m] * seconds_per_day { + month += 1; + total_seconds -= days_in_month[m] * seconds_per_day; + } else { + break; + } + } + + // calculate the number of days + day += total_seconds / seconds_per_day; + total_seconds %= seconds_per_day; + + // calculate hours + hour += total_seconds / seconds_per_hour; + total_seconds %= seconds_per_hour; + + // calculate minutes + minute += total_seconds / seconds_per_minute; + total_seconds %= seconds_per_minute; + + // calculate the number of seconds + second += total_seconds; + + (year as u32, month as u32, day as u32, hour as u32, minute as u32, second as u32) +} diff --git a/crates/event-notifier/src/adapter/kafka.rs b/crates/event-notifier/src/adapter/kafka.rs new file mode 100644 index 000000000..0abd1f00c --- /dev/null +++ b/crates/event-notifier/src/adapter/kafka.rs @@ -0,0 +1,69 @@ +use crate::ChannelAdapter; +use crate::Error; +use crate::Event; +use crate::KafkaConfig; +use async_trait::async_trait; +use rdkafka::error::KafkaError; +use rdkafka::producer::{FutureProducer, FutureRecord}; +use rdkafka::types::RDKafkaErrorCode; +use rdkafka::util::Timeout; +use std::time::Duration; +use tokio::time::sleep; + +/// Kafka adapter for sending events to a Kafka topic. +pub struct KafkaAdapter { + producer: FutureProducer, + topic: String, + max_retries: u32, +} + +impl KafkaAdapter { + /// Creates a new Kafka adapter. + pub fn new(config: &KafkaConfig) -> Result { + // Create a Kafka producer with the provided configuration. + let producer = rdkafka::config::ClientConfig::new() + .set("bootstrap.servers", &config.brokers) + .set("message.timeout.ms", config.timeout.to_string()) + .create()?; + + Ok(Self { + producer, + topic: config.topic.clone(), + max_retries: config.max_retries, + }) + } + /// Sends an event to the Kafka topic with retry logic. + async fn send_with_retry(&self, event: &Event) -> Result<(), Error> { + let event_id = event.id.to_string(); + let payload = serde_json::to_string(&event)?; + + for attempt in 0..self.max_retries { + let record = FutureRecord::to(&self.topic).key(&event_id).payload(&payload); + + match self.producer.send(record, Timeout::Never).await { + Ok(_) => return Ok(()), + Err((KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull), _)) => { + tracing::warn!("Kafka attempt {} failed: Queue full. Retrying...", attempt + 1); + sleep(Duration::from_secs(2u64.pow(attempt))).await; + } + Err((e, _)) => { + tracing::error!("Kafka send error: {}", e); + return Err(Error::Kafka(e)); + } + } + } + + Err(Error::Custom("Exceeded maximum retry attempts for Kafka message".to_string())) + } +} + +#[async_trait] +impl ChannelAdapter for KafkaAdapter { + fn name(&self) -> String { + "kafka".to_string() + } + + async fn send(&self, event: &Event) -> Result<(), Error> { + self.send_with_retry(event).await + } +} diff --git a/crates/event-notifier/src/adapter/mod.rs b/crates/event-notifier/src/adapter/mod.rs new file mode 100644 index 000000000..fa12aa97b --- /dev/null +++ b/crates/event-notifier/src/adapter/mod.rs @@ -0,0 +1,54 @@ +use crate::AdapterConfig; +use crate::Error; +use crate::Event; +use async_trait::async_trait; +use std::sync::Arc; + +#[cfg(feature = "kafka")] +pub(crate) mod kafka; +#[cfg(feature = "mqtt")] +pub(crate) mod mqtt; +#[cfg(feature = "webhook")] +pub(crate) mod webhook; + +/// The `ChannelAdapter` trait defines the interface for all channel adapters. +#[async_trait] +pub trait ChannelAdapter: Send + Sync + 'static { + /// Sends an event to the channel. + fn name(&self) -> String; + /// Sends an event to the channel. + async fn send(&self, event: &Event) -> Result<(), Error>; +} + +/// Creates channel adapters based on the provided configuration. +pub fn create_adapters(configs: &[AdapterConfig]) -> Result>, Error> { + let mut adapters: Vec> = Vec::new(); + + for config in configs { + match config { + #[cfg(feature = "webhook")] + AdapterConfig::Webhook(webhook_config) => { + webhook_config.validate().map_err(Error::ConfigError)?; + adapters.push(Arc::new(webhook::WebhookAdapter::new(webhook_config.clone()))); + } + #[cfg(feature = "kafka")] + AdapterConfig::Kafka(kafka_config) => { + adapters.push(Arc::new(kafka::KafkaAdapter::new(kafka_config)?)); + } + #[cfg(feature = "mqtt")] + AdapterConfig::Mqtt(mqtt_config) => { + let (mqtt, mut event_loop) = mqtt::MqttAdapter::new(mqtt_config); + tokio::spawn(async move { while event_loop.poll().await.is_ok() {} }); + adapters.push(Arc::new(mqtt)); + } + #[cfg(not(feature = "webhook"))] + AdapterConfig::Webhook(_) => return Err(Error::FeatureDisabled("webhook")), + #[cfg(not(feature = "kafka"))] + AdapterConfig::Kafka(_) => return Err(Error::FeatureDisabled("kafka")), + #[cfg(not(feature = "mqtt"))] + AdapterConfig::Mqtt(_) => return Err(Error::FeatureDisabled("mqtt")), + } + } + + Ok(adapters) +} diff --git a/crates/event-notifier/src/adapter/mqtt.rs b/crates/event-notifier/src/adapter/mqtt.rs new file mode 100644 index 000000000..9aab61e8e --- /dev/null +++ b/crates/event-notifier/src/adapter/mqtt.rs @@ -0,0 +1,58 @@ +use crate::ChannelAdapter; +use crate::Error; +use crate::Event; +use crate::MqttConfig; +use async_trait::async_trait; +use rumqttc::{AsyncClient, MqttOptions, QoS}; +use std::time::Duration; +use tokio::time::sleep; + +/// MQTT adapter for sending events to an MQTT broker. +pub struct MqttAdapter { + client: AsyncClient, + topic: String, + max_retries: u32, +} + +impl MqttAdapter { + /// Creates a new MQTT adapter. + pub fn new(config: &MqttConfig) -> (Self, rumqttc::EventLoop) { + let mqtt_options = MqttOptions::new(&config.client_id, &config.broker, config.port); + let (client, event_loop) = rumqttc::AsyncClient::new(mqtt_options, 10); + ( + Self { + client, + topic: config.topic.clone(), + max_retries: config.max_retries, + }, + event_loop, + ) + } +} + +#[async_trait] +impl ChannelAdapter for MqttAdapter { + fn name(&self) -> String { + "mqtt".to_string() + } + + async fn send(&self, event: &Event) -> Result<(), Error> { + let payload = serde_json::to_string(event).map_err(Error::Serde)?; + let mut attempt = 0; + loop { + match self + .client + .publish(&self.topic, QoS::AtLeastOnce, false, payload.clone()) + .await + { + Ok(()) => return Ok(()), + Err(e) if attempt < self.max_retries => { + attempt += 1; + tracing::warn!("MQTT attempt {} failed: {}. Retrying...", attempt, e); + sleep(Duration::from_secs(2u64.pow(attempt))).await; + } + Err(e) => return Err(Error::Mqtt(e)), + } + } + } +} diff --git a/crates/event-notifier/src/adapter/webhook.rs b/crates/event-notifier/src/adapter/webhook.rs new file mode 100644 index 000000000..447c463ee --- /dev/null +++ b/crates/event-notifier/src/adapter/webhook.rs @@ -0,0 +1,64 @@ +use crate::ChannelAdapter; +use crate::Error; +use crate::Event; +use crate::WebhookConfig; +use async_trait::async_trait; +use reqwest::{Client, RequestBuilder}; +use std::time::Duration; +use tokio::time::sleep; + +/// Webhook adapter for sending events to a webhook endpoint. +pub struct WebhookAdapter { + config: WebhookConfig, + client: Client, +} + +impl WebhookAdapter { + /// Creates a new Webhook adapter. + pub fn new(config: WebhookConfig) -> Self { + let client = Client::builder() + .timeout(Duration::from_secs(config.timeout)) + .build() + .expect("Failed to build reqwest client"); + Self { config, client } + } + /// Builds the request to send the event. + fn build_request(&self, event: &Event) -> RequestBuilder { + let mut request = self.client.post(&self.config.endpoint).json(event); + if let Some(token) = &self.config.auth_token { + request = request.header("Authorization", format!("Bearer {}", token)); + } + if let Some(headers) = &self.config.custom_headers { + for (key, value) in headers { + request = request.header(key, value); + } + } + request + } +} + +#[async_trait] +impl ChannelAdapter for WebhookAdapter { + fn name(&self) -> String { + "webhook".to_string() + } + + async fn send(&self, event: &Event) -> Result<(), Error> { + let mut attempt = 0; + tracing::info!("Attempting to send webhook request: {:?}", event); + loop { + match self.build_request(event).send().await { + Ok(response) => { + response.error_for_status().map_err(Error::Http)?; + return Ok(()); + } + Err(e) if attempt < self.config.max_retries => { + attempt += 1; + tracing::warn!("Webhook attempt {} failed: {}. Retrying...", attempt, e); + sleep(Duration::from_secs(2u64.pow(attempt))).await; + } + Err(e) => return Err(Error::Http(e)), + } + } + } +} diff --git a/crates/event-notifier/src/bus.rs b/crates/event-notifier/src/bus.rs new file mode 100644 index 000000000..bd4b81c5d --- /dev/null +++ b/crates/event-notifier/src/bus.rs @@ -0,0 +1,100 @@ +use crate::ChannelAdapter; +use crate::Error; +use crate::EventStore; +use crate::{Event, Log}; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::sync::mpsc; +use tokio::time::Duration; +use tokio_util::sync::CancellationToken; + +/// Handles incoming events from the producer. +/// +/// This function is responsible for receiving events from the producer and sending them to the appropriate adapters. +/// It also handles the shutdown process and saves any pending logs to the event store. +pub async fn event_bus( + mut rx: mpsc::Receiver, + adapters: Vec>, + store: Arc, + shutdown: CancellationToken, + shutdown_complete: Option>, +) -> Result<(), Error> { + let mut current_log = Log { + event_name: crate::event::Name::Everything, + key: SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string(), + records: Vec::new(), + }; + + let mut unprocessed_events = Vec::new(); + loop { + tokio::select! { + Some(event) = rx.recv() => { + current_log.records.push(event.clone()); + let mut send_tasks = Vec::new(); + for adapter in &adapters { + if event.channels.contains(&adapter.name()) { + let adapter = adapter.clone(); + let event = event.clone(); + send_tasks.push(tokio::spawn(async move { + if let Err(e) = adapter.send(&event).await { + tracing::error!("Failed to send event to {}: {}", adapter.name(), e); + Err(e) + } else { + Ok(()) + } + })); + } + } + for task in send_tasks { + if task.await?.is_err() { + // If sending fails, add the event to the unprocessed list + let failed_event = event.clone(); + unprocessed_events.push(failed_event); + } + } + + // Clear the current log because we only care about unprocessed events + current_log.records.clear(); + } + _ = shutdown.cancelled() => { + tracing::info!("Shutting down event bus, saving pending logs..."); + // Check if there are still unprocessed messages in the channel + while let Ok(Some(event)) = tokio::time::timeout( + Duration::from_millis(100), + rx.recv() + ).await { + unprocessed_events.push(event); + } + + // save only if there are unprocessed events + if !unprocessed_events.is_empty() { + tracing::info!("Save {} unhandled events", unprocessed_events.len()); + // create and save logging + let shutdown_log = Log { + event_name: crate::event::Name::Everything, + key: format!("shutdown_{}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()), + records: unprocessed_events, + }; + + store.save_logs(&[shutdown_log]).await?; + } else { + tracing::info!("no unhandled events need to be saved"); + } + tracing::debug!("shutdown_complete is Some: {}", shutdown_complete.is_some()); + + if let Some(complete_sender) = shutdown_complete { + // send a completion signal + let result = complete_sender.send(()); + match result { + Ok(_) => tracing::info!("Event bus shutdown signal sent"), + Err(e) => tracing::error!("Failed to send event bus shutdown signal: {:?}", e), + } + tracing::info!("Shutting down event bus"); + } + tracing::info!("Event bus shutdown complete"); + break; + } + } + } + Ok(()) +} diff --git a/crates/event-notifier/src/config.rs b/crates/event-notifier/src/config.rs new file mode 100644 index 000000000..ae718abc2 --- /dev/null +++ b/crates/event-notifier/src/config.rs @@ -0,0 +1,176 @@ +use config::{Config, Environment, File, FileFormat}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::env; + +/// Configuration for the notification system. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookConfig { + pub endpoint: String, + pub auth_token: Option, + pub custom_headers: Option>, + pub max_retries: u32, + pub timeout: u64, +} + +impl WebhookConfig { + /// verify that the configuration is valid + pub fn validate(&self) -> Result<(), String> { + // verify that endpoint cannot be empty + if self.endpoint.trim().is_empty() { + return Err("Webhook endpoint cannot be empty".to_string()); + } + + // verification timeout must be reasonable + if self.timeout == 0 { + return Err("Webhook timeout must be greater than 0".to_string()); + } + + // Verify that the maximum number of retry is reasonable + if self.max_retries > 10 { + return Err("Maximum retry count cannot exceed 10".to_string()); + } + + Ok(()) + } +} + +/// Configuration for the Kafka adapter. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct KafkaConfig { + pub brokers: String, + pub topic: String, + pub max_retries: u32, + pub timeout: u64, +} + +/// Configuration for the MQTT adapter. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MqttConfig { + pub broker: String, + pub port: u16, + pub client_id: String, + pub topic: String, + pub max_retries: u32, +} + +/// Configuration for the notification system. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type")] +pub enum AdapterConfig { + Webhook(WebhookConfig), + Kafka(KafkaConfig), + Mqtt(MqttConfig), +} + +/// Configuration for the notification system. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NotifierConfig { + #[serde(default = "default_store_path")] + pub store_path: String, + #[serde(default = "default_channel_capacity")] + pub channel_capacity: usize, + pub adapters: Vec, +} + +impl Default for NotifierConfig { + fn default() -> Self { + Self { + store_path: default_store_path(), + channel_capacity: default_channel_capacity(), + adapters: Vec::new(), + } + } +} + +impl NotifierConfig { + /// create a new configuration with default values + pub fn new() -> Self { + Self::default() + } + + /// 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_event_notifier::NotifierConfig; + /// + /// let config = NotifierConfig::event_load_config(None); + /// ``` + pub fn event_load_config(config_dir: Option) -> NotifierConfig { + 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 app_config = Config::builder() + .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false)) + .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false)) + .add_source( + Environment::default() + .prefix("NOTIFIER") + .prefix_separator("__") + .separator("__") + .list_separator("_") + .with_list_parse_key("adapters") + .try_parsing(true), + ) + .build() + .unwrap_or_default(); + println!("Loaded config: {:?}", app_config); + match app_config.try_deserialize::() { + Ok(app_config) => { + println!("Parsed AppConfig: {:?} \n", app_config); + app_config + } + Err(e) => { + println!("Failed to deserialize config: {}", e); + NotifierConfig::default() + } + } + } +} + +const DEFAULT_CONFIG_FILE: &str = "obs"; + +/// Provide temporary directories as default storage paths +fn default_store_path() -> String { + std::env::temp_dir().join("event-notification").to_string_lossy().to_string() +} + +/// Provides the recommended default channel capacity for high concurrency systems +fn default_channel_capacity() -> usize { + 10000 // Reasonable default values for high concurrency systems +} diff --git a/crates/event-notifier/src/error.rs b/crates/event-notifier/src/error.rs new file mode 100644 index 000000000..e6c061de9 --- /dev/null +++ b/crates/event-notifier/src/error.rs @@ -0,0 +1,46 @@ +use config::ConfigError; +use thiserror::Error; +use tokio::sync::mpsc::error; +use tokio::task::JoinError; + +/// The `Error` enum represents all possible errors that can occur in the application. +/// It implements the `std::error::Error` trait and provides a way to convert various error types into a single error type. +#[derive(Error, Debug)] +pub enum Error { + #[error("Join error: {0}")] + JoinError(#[from] JoinError), + #[error("IO error: {0}")] + Io(#[from] std::io::Error), + #[error("Serialization error: {0}")] + Serde(#[from] serde_json::Error), + #[error("HTTP error: {0}")] + Http(#[from] reqwest::Error), + #[cfg(feature = "kafka")] + #[error("Kafka error: {0}")] + Kafka(#[from] rdkafka::error::KafkaError), + #[cfg(feature = "mqtt")] + #[error("MQTT error: {0}")] + Mqtt(#[from] rumqttc::ClientError), + #[error("Channel send error: {0}")] + ChannelSend(#[from] Box>), + #[error("Feature disabled: {0}")] + FeatureDisabled(&'static str), + #[error("Event bus already started")] + EventBusStarted, + #[error("necessary fields are missing:{0}")] + MissingField(&'static str), + #[error("field verification failed:{0}")] + ValidationError(&'static str), + #[error("Custom error: {0}")] + Custom(String), + #[error("Configuration error: {0}")] + ConfigError(String), + #[error("Configuration loading error: {0}")] + Config(#[from] ConfigError), +} + +impl Error { + pub fn custom(msg: &str) -> Error { + Self::Custom(msg.to_string()) + } +} diff --git a/crates/event-notifier/src/event.rs b/crates/event-notifier/src/event.rs new file mode 100644 index 000000000..ce32aecca --- /dev/null +++ b/crates/event-notifier/src/event.rs @@ -0,0 +1,447 @@ +use crate::Error; +use serde::{Deserialize, Serialize}; +use serde_with::{DeserializeFromStr, SerializeDisplay}; +use smallvec::{smallvec, SmallVec}; +use std::borrow::Cow; +use std::collections::HashMap; +use std::time::{SystemTime, UNIX_EPOCH}; +use strum::{Display, EnumString}; +use uuid::Uuid; + +/// A struct representing the identity of the user +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Identity { + #[serde(rename = "principalId")] + pub principal_id: String, +} + +/// A struct representing the bucket information +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Bucket { + pub name: String, + #[serde(rename = "ownerIdentity")] + pub owner_identity: Identity, + pub arn: String, +} + +/// A struct representing the object information +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Object { + pub key: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub size: Option, + #[serde(default, skip_serializing_if = "Option::is_none", rename = "eTag")] + pub etag: Option, + #[serde(default, skip_serializing_if = "Option::is_none", rename = "contentType")] + pub content_type: Option, + #[serde(default, skip_serializing_if = "Option::is_none", rename = "userMetadata")] + pub user_metadata: Option>, + #[serde(default, skip_serializing_if = "Option::is_none", rename = "versionId")] + pub version_id: Option, + pub sequencer: String, +} + +/// A struct representing the metadata of the event +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Metadata { + #[serde(rename = "s3SchemaVersion")] + pub schema_version: String, + #[serde(rename = "configurationId")] + pub configuration_id: String, + pub bucket: Bucket, + pub object: Object, +} + +/// A struct representing the source of the event +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Source { + pub host: String, + pub port: String, + #[serde(rename = "userAgent")] + pub user_agent: String, +} + +/// Builder for creating an Event. +/// +/// This struct is used to build an Event object with various parameters. +/// It provides methods to set each parameter and a build method to create the Event. +#[derive(Default, Clone)] +pub struct EventBuilder { + event_version: Option, + event_source: Option, + aws_region: Option, + event_time: Option, + event_name: Option, + user_identity: Option, + request_parameters: Option>, + response_elements: Option>, + s3: Option, + source: Option, + channels: Option>, +} + +impl EventBuilder { + /// create a builder that pre filled default values + pub fn new() -> Self { + Self { + event_version: Some(Cow::Borrowed("2.0").to_string()), + event_source: Some(Cow::Borrowed("aws:s3").to_string()), + aws_region: Some("us-east-1".to_string()), + event_time: Some(SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string()), + event_name: None, + user_identity: Some(Identity { + principal_id: "anonymous".to_string(), + }), + request_parameters: Some(HashMap::new()), + response_elements: Some(HashMap::new()), + s3: None, + source: None, + channels: Some(Vec::new().into()), + } + } + + /// verify and set the event version + pub fn event_version(mut self, event_version: impl Into) -> Self { + let event_version = event_version.into(); + if !event_version.is_empty() { + self.event_version = Some(event_version); + } + self + } + + /// verify and set the event source + pub fn event_source(mut self, event_source: impl Into) -> Self { + let event_source = event_source.into(); + if !event_source.is_empty() { + self.event_source = Some(event_source); + } + self + } + + /// set up aws regions + pub fn aws_region(mut self, aws_region: impl Into) -> Self { + self.aws_region = Some(aws_region.into()); + self + } + + /// set event time + pub fn event_time(mut self, event_time: impl Into) -> Self { + self.event_time = Some(event_time.into()); + self + } + + /// set event name + pub fn event_name(mut self, event_name: Name) -> Self { + self.event_name = Some(event_name); + self + } + + /// set user identity + pub fn user_identity(mut self, user_identity: Identity) -> Self { + self.user_identity = Some(user_identity); + self + } + + /// set request parameters + pub fn request_parameters(mut self, request_parameters: HashMap) -> Self { + self.request_parameters = Some(request_parameters); + self + } + + /// set response elements + pub fn response_elements(mut self, response_elements: HashMap) -> Self { + self.response_elements = Some(response_elements); + self + } + + /// setting up s3 metadata + pub fn s3(mut self, s3: Metadata) -> Self { + self.s3 = Some(s3); + self + } + + /// set event source information + pub fn source(mut self, source: Source) -> Self { + self.source = Some(source); + self + } + + /// set up the sending channel + pub fn channels(mut self, channels: Vec) -> Self { + self.channels = Some(channels.into()); + self + } + + /// Create a preconfigured builder for common object event scenarios + pub fn for_object_creation(s3: Metadata, source: Source) -> Self { + Self::new().event_name(Name::ObjectCreatedPut).s3(s3).source(source) + } + + /// Create a preconfigured builder for object deletion events + pub fn for_object_removal(s3: Metadata, source: Source) -> Self { + Self::new().event_name(Name::ObjectRemovedDelete).s3(s3).source(source) + } + + /// build event instance + /// + /// Verify the required fields and create a complete Event object + pub fn build(self) -> Result { + let event_version = self.event_version.ok_or(Error::MissingField("event_version"))?; + + let event_source = self.event_source.ok_or(Error::MissingField("event_source"))?; + + let aws_region = self.aws_region.ok_or(Error::MissingField("aws_region"))?; + + let event_time = self.event_time.ok_or(Error::MissingField("event_time"))?; + + let event_name = self.event_name.ok_or(Error::MissingField("event_name"))?; + + let user_identity = self.user_identity.ok_or(Error::MissingField("user_identity"))?; + + let request_parameters = self.request_parameters.unwrap_or_default(); + let response_elements = self.response_elements.unwrap_or_default(); + + let s3 = self.s3.ok_or(Error::MissingField("s3"))?; + + let source = self.source.ok_or(Error::MissingField("source"))?; + + let channels = self.channels.unwrap_or_else(|| smallvec![]); + + Ok(Event { + event_version, + event_source, + aws_region, + event_time, + event_name, + user_identity, + request_parameters, + response_elements, + s3, + source, + id: Uuid::new_v4(), + timestamp: SystemTime::now(), + channels, + }) + } +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Event { + #[serde(rename = "eventVersion")] + pub event_version: String, + #[serde(rename = "eventSource")] + pub event_source: String, + #[serde(rename = "awsRegion")] + pub aws_region: String, + #[serde(rename = "eventTime")] + pub event_time: String, + #[serde(rename = "eventName")] + pub event_name: Name, + #[serde(rename = "userIdentity")] + pub user_identity: Identity, + #[serde(rename = "requestParameters")] + pub request_parameters: HashMap, + #[serde(rename = "responseElements")] + pub response_elements: HashMap, + pub s3: Metadata, + pub source: Source, + pub id: Uuid, + pub timestamp: SystemTime, + pub channels: SmallVec<[String; 2]>, +} + +impl Event { + /// create a new event builder + /// + /// Returns an EventBuilder instance pre-filled with default values + pub fn builder() -> EventBuilder { + EventBuilder::new() + } + + /// Quickly create Event instances with necessary fields + /// + /// suitable for common s3 event scenarios + pub fn create(event_name: Name, s3: Metadata, source: Source, channels: Vec) -> Self { + Self::builder() + .event_name(event_name) + .s3(s3) + .source(source) + .channels(channels) + .build() + .expect("Failed to create event, missing necessary parameters") + } + + /// a convenient way to create a preconfigured builder + pub fn for_object_creation(s3: Metadata, source: Source) -> EventBuilder { + EventBuilder::for_object_creation(s3, source) + } + + /// a convenient way to create a preconfigured builder + pub fn for_object_removal(s3: Metadata, source: Source) -> EventBuilder { + EventBuilder::for_object_removal(s3, source) + } + + /// Determine whether an event belongs to a specific type + pub fn is_type(&self, event_type: Name) -> bool { + let mask = event_type.mask(); + (self.event_name.mask() & mask) != 0 + } + + /// Determine whether an event needs to be sent to a specific channel + pub fn is_for_channel(&self, channel: &str) -> bool { + self.channels.iter().any(|c| c == channel) + } +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Log { + #[serde(rename = "eventName")] + pub event_name: Name, + pub key: String, + pub records: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, SerializeDisplay, DeserializeFromStr, Display, EnumString)] +#[strum(serialize_all = "SCREAMING_SNAKE_CASE")] +pub enum Name { + ObjectAccessedGet, + ObjectAccessedGetRetention, + ObjectAccessedGetLegalHold, + ObjectAccessedHead, + ObjectAccessedAttributes, + ObjectCreatedCompleteMultipartUpload, + ObjectCreatedCopy, + ObjectCreatedPost, + ObjectCreatedPut, + ObjectCreatedPutRetention, + ObjectCreatedPutLegalHold, + ObjectCreatedPutTagging, + ObjectCreatedDeleteTagging, + ObjectRemovedDelete, + ObjectRemovedDeleteMarkerCreated, + ObjectRemovedDeleteAllVersions, + ObjectRemovedNoOp, + BucketCreated, + BucketRemoved, + ObjectReplicationFailed, + ObjectReplicationComplete, + ObjectReplicationMissedThreshold, + ObjectReplicationReplicatedAfterThreshold, + ObjectReplicationNotTracked, + ObjectRestorePost, + ObjectRestoreCompleted, + ObjectTransitionFailed, + ObjectTransitionComplete, + ObjectManyVersions, + ObjectLargeVersions, + PrefixManyFolders, + IlmDelMarkerExpirationDelete, + ObjectAccessedAll, + ObjectCreatedAll, + ObjectRemovedAll, + ObjectReplicationAll, + ObjectRestoreAll, + ObjectTransitionAll, + ObjectScannerAll, + Everything, +} + +impl Name { + pub fn expand(&self) -> Vec { + match self { + Name::ObjectAccessedAll => vec![ + Name::ObjectAccessedGet, + Name::ObjectAccessedHead, + Name::ObjectAccessedGetRetention, + Name::ObjectAccessedGetLegalHold, + Name::ObjectAccessedAttributes, + ], + Name::ObjectCreatedAll => vec![ + Name::ObjectCreatedCompleteMultipartUpload, + Name::ObjectCreatedCopy, + Name::ObjectCreatedPost, + Name::ObjectCreatedPut, + Name::ObjectCreatedPutRetention, + Name::ObjectCreatedPutLegalHold, + Name::ObjectCreatedPutTagging, + Name::ObjectCreatedDeleteTagging, + ], + Name::ObjectRemovedAll => vec![ + Name::ObjectRemovedDelete, + Name::ObjectRemovedDeleteMarkerCreated, + Name::ObjectRemovedNoOp, + Name::ObjectRemovedDeleteAllVersions, + ], + Name::ObjectReplicationAll => vec![ + Name::ObjectReplicationFailed, + Name::ObjectReplicationComplete, + Name::ObjectReplicationNotTracked, + Name::ObjectReplicationMissedThreshold, + Name::ObjectReplicationReplicatedAfterThreshold, + ], + Name::ObjectRestoreAll => vec![Name::ObjectRestorePost, Name::ObjectRestoreCompleted], + Name::ObjectTransitionAll => { + vec![Name::ObjectTransitionFailed, Name::ObjectTransitionComplete] + } + Name::ObjectScannerAll => vec![Name::ObjectManyVersions, Name::ObjectLargeVersions, Name::PrefixManyFolders], + Name::Everything => (1..=Name::IlmDelMarkerExpirationDelete as u32) + .map(|i| Name::from_repr(i).unwrap()) + .collect(), + _ => vec![*self], + } + } + + pub fn mask(&self) -> u64 { + if (*self as u32) < Name::ObjectAccessedAll as u32 { + 1 << (*self as u32 - 1) + } else { + self.expand().iter().fold(0, |acc, n| acc | (1 << (*n as u32 - 1))) + } + } + + fn from_repr(discriminant: u32) -> Option { + match discriminant { + 1 => Some(Name::ObjectAccessedGet), + 2 => Some(Name::ObjectAccessedGetRetention), + 3 => Some(Name::ObjectAccessedGetLegalHold), + 4 => Some(Name::ObjectAccessedHead), + 5 => Some(Name::ObjectAccessedAttributes), + 6 => Some(Name::ObjectCreatedCompleteMultipartUpload), + 7 => Some(Name::ObjectCreatedCopy), + 8 => Some(Name::ObjectCreatedPost), + 9 => Some(Name::ObjectCreatedPut), + 10 => Some(Name::ObjectCreatedPutRetention), + 11 => Some(Name::ObjectCreatedPutLegalHold), + 12 => Some(Name::ObjectCreatedPutTagging), + 13 => Some(Name::ObjectCreatedDeleteTagging), + 14 => Some(Name::ObjectRemovedDelete), + 15 => Some(Name::ObjectRemovedDeleteMarkerCreated), + 16 => Some(Name::ObjectRemovedDeleteAllVersions), + 17 => Some(Name::ObjectRemovedNoOp), + 18 => Some(Name::BucketCreated), + 19 => Some(Name::BucketRemoved), + 20 => Some(Name::ObjectReplicationFailed), + 21 => Some(Name::ObjectReplicationComplete), + 22 => Some(Name::ObjectReplicationMissedThreshold), + 23 => Some(Name::ObjectReplicationReplicatedAfterThreshold), + 24 => Some(Name::ObjectReplicationNotTracked), + 25 => Some(Name::ObjectRestorePost), + 26 => Some(Name::ObjectRestoreCompleted), + 27 => Some(Name::ObjectTransitionFailed), + 28 => Some(Name::ObjectTransitionComplete), + 29 => Some(Name::ObjectManyVersions), + 30 => Some(Name::ObjectLargeVersions), + 31 => Some(Name::PrefixManyFolders), + 32 => Some(Name::IlmDelMarkerExpirationDelete), + 33 => Some(Name::ObjectAccessedAll), + 34 => Some(Name::ObjectCreatedAll), + 35 => Some(Name::ObjectRemovedAll), + 36 => Some(Name::ObjectReplicationAll), + 37 => Some(Name::ObjectRestoreAll), + 38 => Some(Name::ObjectTransitionAll), + 39 => Some(Name::ObjectScannerAll), + 40 => Some(Name::Everything), + _ => None, + } + } +} diff --git a/crates/event-notifier/src/global.rs b/crates/event-notifier/src/global.rs new file mode 100644 index 000000000..25fadeeb4 --- /dev/null +++ b/crates/event-notifier/src/global.rs @@ -0,0 +1,230 @@ +use crate::{create_adapters, Error, Event, NotifierConfig, NotifierSystem}; +use std::sync::{atomic, Arc}; +use tokio::sync::{Mutex, OnceCell}; + +static GLOBAL_SYSTEM: OnceCell>> = OnceCell::const_new(); +static INITIALIZED: atomic::AtomicBool = atomic::AtomicBool::new(false); +static READY: atomic::AtomicBool = atomic::AtomicBool::new(false); +static INIT_LOCK: Mutex<()> = Mutex::const_new(()); + +/// Initializes the global notification system. +/// +/// This function performs the following steps: +/// 1. Checks if the system is already initialized. +/// 2. Creates a new `NotificationSystem` instance. +/// 3. Creates adapters based on the provided configuration. +/// 4. Starts the notification system with the created adapters. +/// 5. Sets the global system instance. +/// +/// # Errors +/// +/// Returns an error if: +/// - The system is already initialized. +/// - Creating the `NotificationSystem` fails. +/// - Creating adapters fails. +/// - Starting the notification system fails. +/// - Setting the global system instance fails. +pub async fn initialize(config: NotifierConfig) -> Result<(), Error> { + let _lock = INIT_LOCK.lock().await; + + // Check if the system is already initialized. + if INITIALIZED.load(atomic::Ordering::SeqCst) { + return Err(Error::custom("Notification system has already been initialized")); + } + + // Check if the system is already ready. + if READY.load(atomic::Ordering::SeqCst) { + return Err(Error::custom("Notification system is already ready")); + } + + // Check if the system is shutting down. + if let Some(system) = GLOBAL_SYSTEM.get() { + let system_guard = system.lock().await; + if system_guard.shutdown_cancelled() { + return Err(Error::custom("Notification system is shutting down")); + } + } + + // check if config adapters len is than 0 + if config.adapters.is_empty() { + return Err(Error::custom("No adapters configured")); + } + + // Attempt to initialize, and reset the INITIALIZED flag if it fails. + let result: Result<(), Error> = async { + let system = NotifierSystem::new(config.clone()).await.map_err(|e| { + tracing::error!("Failed to create NotificationSystem: {:?}", e); + e + })?; + let adapters = create_adapters(&config.adapters).map_err(|e| { + tracing::error!("Failed to create adapters: {:?}", e); + e + })?; + tracing::info!("adapters len:{:?}", adapters.len()); + let system_clone = Arc::new(Mutex::new(system)); + let adapters_clone = adapters.clone(); + + GLOBAL_SYSTEM.set(system_clone.clone()).map_err(|_| { + let err = Error::custom("Unable to set up global notification system"); + tracing::error!("{:?}", err); + err + })?; + + tokio::spawn(async move { + if let Err(e) = system_clone.lock().await.start(adapters_clone).await { + tracing::error!("Notification system failed to start: {}", e); + } + tracing::info!("Notification system started in background"); + }); + tracing::info!("system start success,start set READY value"); + + READY.store(true, atomic::Ordering::SeqCst); + tracing::info!("Notification system is ready to process events"); + + Ok(()) + } + .await; + + if result.is_err() { + INITIALIZED.store(false, atomic::Ordering::SeqCst); + READY.store(false, atomic::Ordering::SeqCst); + return result; + } + + INITIALIZED.store(true, atomic::Ordering::SeqCst); + Ok(()) +} + +/// Checks if the notification system is initialized. +pub fn is_initialized() -> bool { + INITIALIZED.load(atomic::Ordering::SeqCst) +} + +/// Checks if the notification system is ready. +pub fn is_ready() -> bool { + READY.load(atomic::Ordering::SeqCst) +} + +/// Sends an event to the notification system. +/// +/// # Errors +/// +/// Returns an error if: +/// - The system is not initialized. +/// - The system is not ready. +/// - Sending the event fails. +pub async fn send_event(event: Event) -> Result<(), Error> { + if !READY.load(atomic::Ordering::SeqCst) { + return Err(Error::custom("Notification system not ready, please wait for initialization to complete")); + } + + let system = get_system().await?; + let system_guard = system.lock().await; + system_guard.send_event(event).await +} + +/// Shuts down the notification system. +pub async fn shutdown() -> Result<(), Error> { + if let Some(system) = GLOBAL_SYSTEM.get() { + tracing::info!("Shutting down notification system start"); + let result = { + let mut system_guard = system.lock().await; + system_guard.shutdown().await + }; + if let Err(e) = &result { + tracing::error!("Notification system shutdown failed: {}", e); + } else { + tracing::info!("Event bus shutdown completed"); + } + + tracing::info!( + "Shutdown method called set static value start, READY: {}, INITIALIZED: {}", + READY.load(atomic::Ordering::SeqCst), + INITIALIZED.load(atomic::Ordering::SeqCst) + ); + READY.store(false, atomic::Ordering::SeqCst); + INITIALIZED.store(false, atomic::Ordering::SeqCst); + tracing::info!( + "Shutdown method called set static value end, READY: {}, INITIALIZED: {}", + READY.load(atomic::Ordering::SeqCst), + INITIALIZED.load(atomic::Ordering::SeqCst) + ); + result + } else { + Err(Error::custom("Notification system not initialized")) + } +} + +/// Retrieves the global notification system instance. +/// +/// # Errors +/// +/// Returns an error if the system is not initialized. +async fn get_system() -> Result>, Error> { + GLOBAL_SYSTEM + .get() + .cloned() + .ok_or_else(|| Error::custom("Notification system not initialized")) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{AdapterConfig, NotifierConfig, WebhookConfig}; + use std::collections::HashMap; + + #[tokio::test] + async fn test_initialize_success() { + tracing_subscriber::fmt::init(); + let config = NotifierConfig::default(); // assume there is a default configuration + let result = initialize(config).await; + assert!(result.is_err(), "Initialization should not succeed"); + assert!(!is_initialized(), "System should not be marked as initialized"); + assert!(!is_ready(), "System should not be marked as ready"); + } + + #[tokio::test] + async fn test_initialize_twice() { + tracing_subscriber::fmt::init(); + let config = NotifierConfig::default(); + let _ = initialize(config.clone()).await; // first initialization + let result = initialize(config).await; // second initialization + assert!(!result.is_ok(), "Initialization should succeed"); + assert!(result.is_err(), "Re-initialization should fail"); + } + + #[tokio::test] + async fn test_initialize_failure_resets_state() { + tracing_subscriber::fmt::init(); + // simulate wrong configuration + let config = NotifierConfig { + adapters: vec![ + // assuming that the empty adapter will cause failure + AdapterConfig::Webhook(WebhookConfig { + endpoint: "http://localhost:8080/webhook".to_string(), + auth_token: Some("secret-token".to_string()), + custom_headers: Some(HashMap::from([("X-Custom".to_string(), "value".to_string())])), + max_retries: 3, + timeout: 10, + }), + ], // assuming that the empty adapter will cause failure + ..Default::default() + }; + let result = initialize(config).await; + assert!(!result.is_err(), "Initialization with invalid config should fail"); + assert!(is_initialized(), "System should not be marked as initialized after failure"); + assert!(is_ready(), "System should not be marked as ready after failure"); + } + + #[tokio::test] + async fn test_is_initialized_and_is_ready() { + tracing_subscriber::fmt::init(); + assert!(!is_initialized(), "System should not be initialized initially"); + assert!(!is_ready(), "System should not be ready initially"); + + let config = NotifierConfig::default(); + let _ = initialize(config).await; + assert!(!is_initialized(), "System should be initialized after successful initialization"); + assert!(!is_ready(), "System should be ready after successful initialization"); + } +} diff --git a/crates/event-notifier/src/lib.rs b/crates/event-notifier/src/lib.rs new file mode 100644 index 000000000..20ef935d2 --- /dev/null +++ b/crates/event-notifier/src/lib.rs @@ -0,0 +1,31 @@ +mod adapter; +mod bus; +mod config; +mod error; +mod event; +mod global; +mod notifier; +mod store; + +pub use adapter::create_adapters; +#[cfg(feature = "kafka")] +pub use adapter::kafka::KafkaAdapter; +#[cfg(feature = "mqtt")] +pub use adapter::mqtt::MqttAdapter; +#[cfg(feature = "webhook")] +pub use adapter::webhook::WebhookAdapter; +pub use adapter::ChannelAdapter; +pub use bus::event_bus; +#[cfg(feature = "kafka")] +pub use config::KafkaConfig; +#[cfg(feature = "mqtt")] +pub use config::MqttConfig; +#[cfg(feature = "webhook")] +pub use config::WebhookConfig; +pub use config::{AdapterConfig, NotifierConfig}; +pub use error::Error; + +pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source}; +pub use global::{initialize, is_initialized, is_ready, send_event, shutdown}; +pub use notifier::NotifierSystem; +pub use store::EventStore; diff --git a/crates/event-notifier/src/notifier.rs b/crates/event-notifier/src/notifier.rs new file mode 100644 index 000000000..0b17ddddd --- /dev/null +++ b/crates/event-notifier/src/notifier.rs @@ -0,0 +1,128 @@ +use crate::{event_bus, ChannelAdapter, Error, Event, EventStore, NotifierConfig}; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +/// The `NotificationSystem` struct represents the notification system. +/// It manages the event bus and the adapters. +/// It is responsible for sending and receiving events. +/// It also handles the shutdown process. +pub struct NotifierSystem { + tx: mpsc::Sender, + rx: Option>, + store: Arc, + shutdown: CancellationToken, + shutdown_complete: Option>, + shutdown_receiver: Option>, +} + +impl NotifierSystem { + /// Creates a new `NotificationSystem` instance. + pub async fn new(config: NotifierConfig) -> Result { + let (tx, rx) = mpsc::channel::(config.channel_capacity); + let store = Arc::new(EventStore::new(&config.store_path).await?); + let shutdown = CancellationToken::new(); + + let restored_logs = store.load_logs().await?; + for log in restored_logs { + for event in log.records { + // For example, where the send method may return a SendError when calling it + tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?; + } + } + // Initialize shutdown_complete to Some(tx) + let (complete_tx, complete_rx) = tokio::sync::oneshot::channel(); + Ok(Self { + tx, + rx: Some(rx), + store, + shutdown, + shutdown_complete: Some(complete_tx), + shutdown_receiver: Some(complete_rx), + }) + } + + /// Starts the notification system. + /// It initializes the event bus and the producer. + pub async fn start(&mut self, adapters: Vec>) -> Result<(), Error> { + if self.shutdown.is_cancelled() { + let error = Error::custom("System is shutting down"); + self.handle_error("start", &error); + return Err(error); + } + self.log(tracing::Level::INFO, "start", "Starting the notification system"); + let rx = self.rx.take().ok_or_else(|| Error::EventBusStarted)?; + let shutdown_clone = self.shutdown.clone(); + let store_clone = self.store.clone(); + let shutdown_complete = self.shutdown_complete.take(); + + tokio::spawn(async move { + if let Err(e) = event_bus(rx, adapters, store_clone, shutdown_clone, shutdown_complete).await { + tracing::error!("Event bus failed: {}", e); + } + }); + self.log(tracing::Level::INFO, "start", "Notification system started successfully"); + Ok(()) + } + + /// Sends an event to the notification system. + /// This method is used to send events to the event bus. + pub async fn send_event(&self, event: Event) -> Result<(), Error> { + self.log(tracing::Level::DEBUG, "send_event", &format!("Sending event: {:?}", event)); + if self.shutdown.is_cancelled() { + let error = Error::custom("System is shutting down"); + self.handle_error("send_event", &error); + return Err(error); + } + if let Err(e) = self.tx.send(event).await { + let error = Error::ChannelSend(Box::new(e)); + self.handle_error("send_event", &error); + return Err(error); + } + self.log(tracing::Level::INFO, "send_event", "Event sent successfully"); + Ok(()) + } + + /// Shuts down the notification system. + /// This method is used to cancel the event bus and producer tasks. + pub async fn shutdown(&mut self) -> Result<(), Error> { + tracing::info!("Shutting down the notification system"); + self.shutdown.cancel(); + // wait for the event bus to be completely closed + if let Some(receiver) = self.shutdown_receiver.take() { + match receiver.await { + Ok(_) => { + tracing::info!("Event bus shutdown completed successfully"); + Ok(()) + } + Err(e) => { + let error = Error::custom(format!("Failed to receive shutdown completion: {}", e).as_str()); + self.handle_error("shutdown", &error); + Err(error) + } + } + } else { + tracing::warn!("Shutdown receiver not available, the event bus might still be running"); + Err(Error::custom("Shutdown receiver not available")) + } + } + + /// shutdown state + pub fn shutdown_cancelled(&self) -> bool { + self.shutdown.is_cancelled() + } + + fn handle_error(&self, context: &str, error: &Error) { + self.log(tracing::Level::ERROR, context, &format!("{:?}", error)); + // TODO Can be extended to record to files or send to monitoring systems + } + fn log(&self, level: tracing::Level, context: &str, message: &str) { + match level { + tracing::Level::ERROR => tracing::error!("[{}] {}", context, message), + tracing::Level::WARN => tracing::warn!("[{}] {}", context, message), + tracing::Level::INFO => tracing::info!("[{}] {}", context, message), + tracing::Level::DEBUG => tracing::debug!("[{}] {}", context, message), + tracing::Level::TRACE => tracing::trace!("[{}] {}", context, message), + } + } +} diff --git a/crates/event-notifier/src/store.rs b/crates/event-notifier/src/store.rs new file mode 100644 index 000000000..eca26674a --- /dev/null +++ b/crates/event-notifier/src/store.rs @@ -0,0 +1,58 @@ +use crate::Error; +use crate::Log; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::fs::{create_dir_all, File, OpenOptions}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter}; +use tokio::sync::RwLock; + +/// `EventStore` is a struct that manages the storage of event logs. +pub struct EventStore { + path: String, + lock: Arc>, +} + +impl EventStore { + pub async fn new(path: &str) -> Result { + create_dir_all(path).await?; + Ok(Self { + path: path.to_string(), + lock: Arc::new(RwLock::new(())), + }) + } + + pub async fn save_logs(&self, logs: &[Log]) -> Result<(), Error> { + let _guard = self.lock.write().await; + let file_path = format!( + "{}/events_{}.jsonl", + self.path, + SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() + ); + let file = OpenOptions::new().create(true).append(true).open(&file_path).await?; + let mut writer = BufWriter::new(file); + for log in logs { + let line = serde_json::to_string(log)?; + writer.write_all(line.as_bytes()).await?; + writer.write_all(b"\n").await?; + } + writer.flush().await?; + tracing::info!("Saved logs to {} end", file_path); + Ok(()) + } + + pub async fn load_logs(&self) -> Result, Error> { + let _guard = self.lock.read().await; + let mut logs = Vec::new(); + let mut entries = tokio::fs::read_dir(&self.path).await?; + while let Some(entry) = entries.next_entry().await? { + let file = File::open(entry.path()).await?; + let reader = BufReader::new(file); + let mut lines = reader.lines(); + while let Some(line) = lines.next_line().await? { + let log: Log = serde_json::from_str(&line)?; + logs.push(log); + } + } + Ok(logs) + } +} diff --git a/crates/event-notifier/tests/integration.rs b/crates/event-notifier/tests/integration.rs new file mode 100644 index 000000000..c5743605b --- /dev/null +++ b/crates/event-notifier/tests/integration.rs @@ -0,0 +1,159 @@ +use rustfs_event_notifier::{AdapterConfig, NotifierSystem, WebhookConfig}; +use rustfs_event_notifier::{Bucket, Event, EventBuilder, Identity, Metadata, Name, Object, Source}; +use rustfs_event_notifier::{ChannelAdapter, WebhookAdapter}; +use std::collections::HashMap; +use std::sync::Arc; + +#[tokio::test] +async fn test_webhook_adapter() { + let adapter = WebhookAdapter::new(WebhookConfig { + endpoint: "http://localhost:8080/webhook".to_string(), + auth_token: None, + custom_headers: None, + max_retries: 1, + timeout: 5, + }); + + // create an s3 metadata object + let metadata = Metadata { + schema_version: "1.0".to_string(), + configuration_id: "test-config".to_string(), + bucket: Bucket { + name: "my-bucket".to_string(), + owner_identity: Identity { + principal_id: "owner123".to_string(), + }, + arn: "arn:aws:s3:::my-bucket".to_string(), + }, + object: Object { + key: "test.txt".to_string(), + size: Some(1024), + etag: Some("abc123".to_string()), + content_type: Some("text/plain".to_string()), + user_metadata: None, + version_id: None, + sequencer: "1234567890".to_string(), + }, + }; + + // create source object + let source = Source { + host: "localhost".to_string(), + port: "80".to_string(), + user_agent: "curl/7.68.0".to_string(), + }; + + // Create events using builder mode + let event = Event::builder() + .event_version("2.0") + .event_source("aws:s3") + .aws_region("us-east-1") + .event_time("2023-10-01T12:00:00.000Z") + .event_name(Name::ObjectCreatedPut) + .user_identity(Identity { + principal_id: "user123".to_string(), + }) + .request_parameters(HashMap::new()) + .response_elements(HashMap::new()) + .s3(metadata) + .source(source) + .channels(vec!["webhook".to_string()]) + .build() + .expect("failed to create event"); + + let result = adapter.send(&event).await; + assert!(result.is_err()); +} + +#[tokio::test] +async fn test_notification_system() { + let config = rustfs_event_notifier::NotifierConfig { + store_path: "./test_events".to_string(), + channel_capacity: 100, + adapters: vec![AdapterConfig::Webhook(WebhookConfig { + endpoint: "http://localhost:8080/webhook".to_string(), + auth_token: None, + custom_headers: None, + max_retries: 1, + timeout: 5, + })], + }; + let system = Arc::new(tokio::sync::Mutex::new(NotifierSystem::new(config.clone()).await.unwrap())); + let adapters: Vec> = vec![Arc::new(WebhookAdapter::new(WebhookConfig { + endpoint: "http://localhost:8080/webhook".to_string(), + auth_token: None, + custom_headers: None, + max_retries: 1, + timeout: 5, + }))]; + + // create an s3 metadata object + let metadata = Metadata { + schema_version: "1.0".to_string(), + configuration_id: "test-config".to_string(), + bucket: Bucket { + name: "my-bucket".to_string(), + owner_identity: Identity { + principal_id: "owner123".to_string(), + }, + arn: "arn:aws:s3:::my-bucket".to_string(), + }, + object: Object { + key: "test.txt".to_string(), + size: Some(1024), + etag: Some("abc123".to_string()), + content_type: Some("text/plain".to_string()), + user_metadata: None, + version_id: None, + sequencer: "1234567890".to_string(), + }, + }; + + // create source object + let source = Source { + host: "localhost".to_string(), + port: "80".to_string(), + user_agent: "curl/7.68.0".to_string(), + }; + + // create a preconfigured builder with objects + let event = EventBuilder::for_object_creation(metadata, source) + .user_identity(Identity { + principal_id: "user123".to_string(), + }) + .event_time("2023-10-01T12:00:00.000Z") + .channels(vec!["webhook".to_string()]) + .build() + .expect("failed to create event"); + + { + let system_lock = system.lock().await; + system_lock.send_event(event).await.unwrap(); + } + + let system_clone = Arc::clone(&system); + let system_handle = tokio::spawn(async move { + let mut system = system_clone.lock().await; + system.start(adapters).await + }); + + // set 10 seconds timeout + match tokio::time::timeout(std::time::Duration::from_secs(10), system_handle).await { + Ok(result) => { + println!("System started successfully"); + assert!(result.is_ok()); + } + Err(_) => { + println!("System operation timed out, forcing shutdown"); + // create a new task to handle the timeout + let system = Arc::clone(&system); + tokio::spawn(async move { + if let Ok(mut guard) = system.try_lock() { + guard.shutdown().await.unwrap(); + } + }); + // give the system some time to clean up resources + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + } +} diff --git a/crates/obs/src/telemetry.rs b/crates/obs/src/telemetry.rs index 8efa869dd..db0ce1690 100644 --- a/crates/obs/src/telemetry.rs +++ b/crates/obs/src/telemetry.rs @@ -254,21 +254,16 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { 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", "tonic", "h2", "reqwest", "tower"] { filter = filter.add_directive(format!("{}=off", directive).parse().unwrap()); } filter } - _ => { - let mut filter = EnvFilter::new(logger_level); - for directive in ["hyper", "tonic", "h2", "reqwest"] { - filter = filter.add_directive(format!("{}=off", directive).parse().unwrap()); - } - filter - } }; - layer::OpenTelemetryTracingBridge::new(&logger_provider).with_filter(filter_otel) }; @@ -277,7 +272,9 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { .with(switch_level(logger_level)) .with(OpenTelemetryLayer::new(tracer)) .with(MetricsLayer::new(meter_provider.clone())) - .with(otel_layer); + .with(otel_layer) + .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(logger_level))); + // Configure formatting layer let enable_color = std::io::stdout().is_terminal(); let fmt_layer = tracing_subscriber::fmt::layer() diff --git a/data/README.md b/data/README.md deleted file mode 100644 index 2e3d4df77..000000000 --- a/data/README.md +++ /dev/null @@ -1 +0,0 @@ -## Observability Docker Compose \ No newline at end of file diff --git a/deploy/config/event.example.toml b/deploy/config/event.example.toml new file mode 100644 index 000000000..6deb67a2e --- /dev/null +++ b/deploy/config/event.example.toml @@ -0,0 +1,29 @@ +# config.toml +store_path = "./deploy/logs/event_store" +channel_capacity = 5000 + +[[adapters]] +type = "Webhook" +endpoint = "http://127.0.0.1:3020/webhook" +auth_token = "your-auth-token" +max_retries = 3 +timeout = 50 + +[adapters.custom_headers] +custom_server = "value_server" +custom_client = "value_client" + +#[[adapters]] +#type = "Kafka" +#brokers = "localhost:9092" +#topic = "notifications" +#max_retries = 3 +#timeout = 60 +# +#[[adapters]] +#type = "Mqtt" +#broker = "mqtt.example.com" +#port = 1883 +#client_id = "event-notifier" +#topic = "events" +#max_retries = 3 \ No newline at end of file diff --git a/deploy/config/obs.example.toml b/deploy/config/obs.example.toml index a3c3d0966..97192fb4f 100644 --- a/deploy/config/obs.example.toml +++ b/deploy/config/obs.example.toml @@ -6,7 +6,7 @@ meter_interval = 30 service_name = "rustfs" service_version = "0.1.0" environment = "develop" -logger_level = "info" +logger_level = "error" [sinks] [sinks.kafka] # Kafka sink is disabled by default diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index 13c8edc52..2ad741d6d 100644 --- a/ecstore/Cargo.toml +++ b/ecstore/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ecstore" -version = "0.1.0" +version.workspace = true edition.workspace = true license.workspace = true repository.workspace = true diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 24e5782fe..62e1a3ea3 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -122,7 +122,7 @@ impl LocalDisk { let root = fs::canonicalize(ep.get_file_path()).await?; if cleanup { - // TODO: 删除tmp数据 + // TODO: 删除 tmp 数据 } let format_path = Path::new(super::RUSTFS_META_BUCKET) @@ -631,13 +631,13 @@ impl LocalDisk { } } - // 没有版本了，删除xl.meta + // 没有版本了，删除 xl.meta if fm.versions.is_empty() { self.delete_file(&volume_dir, &xlpath, true, false).await?; return Ok(()); } - // 更新xl.meta + // 更新 xl.meta let buf = fm.marshal_msg()?; let volume_dir = self.get_bucket_path(volume)?; @@ -875,7 +875,7 @@ impl LocalDisk { .read_metadata(self.get_object_path(bucket, format!("{}/{}", ¤t, &entry).as_str())?) .await?; - // 用strip_suffix只删除一次 + // 用 strip_suffix 只删除一次 let entry = entry.strip_suffix(STORAGE_FORMAT_FILE).unwrap_or_default().to_owned(); let name = entry.trim_end_matches(SLASH_SEPARATOR); let name = decode_dir_object(format!("{}/{}", ¤t, &name).as_str()); @@ -1723,7 +1723,7 @@ impl DiskAPI for LocalDisk { } } - // xl.meta路径 + // xl.meta 路径 let src_file_path = src_volume_dir.join(Path::new(format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str())); let dst_file_path = dst_volume_dir.join(Path::new(format!("{}/{}", &dst_path, super::STORAGE_FORMAT_FILE).as_str())); @@ -1754,7 +1754,7 @@ impl DiskAPI for LocalDisk { check_path_length(src_file_path.to_string_lossy().to_string().as_str())?; check_path_length(dst_file_path.to_string_lossy().to_string().as_str())?; - // 读旧xl.meta + // 读旧 xl.meta let has_dst_buf = match utils::fs::read_file(&dst_file_path).await { Ok(res) => Some(res), @@ -2268,7 +2268,7 @@ impl DiskAPI for LocalDisk { async fn delete_volume(&self, volume: &str) -> Result<()> { let p = self.get_bucket_path(volume)?; - // TODO: 不能用递归删除，如果目录下面有文件，返回errVolumeNotEmpty + // TODO: 不能用递归删除，如果目录下面有文件，返回 errVolumeNotEmpty if let Err(err) = fs::remove_dir_all(&p).await { match err.kind() { @@ -2328,10 +2328,7 @@ impl DiskAPI for LocalDisk { } } - let vcfg = match BucketVersioningSys::get(&cache.info.name).await { - Ok(vcfg) => Some(vcfg), - Err(_) => None, - }; + let vcfg = (BucketVersioningSys::get(&cache.info.name).await).ok(); let loc = self.get_disk_location(); let disks = store.get_disks(loc.pool_idx.unwrap(), loc.disk_idx.unwrap()).await?; @@ -2491,7 +2488,6 @@ async fn get_disk_info(drive_path: PathBuf) -> Result<(Info, bool)> { #[cfg(test)] mod test { - use super::*; #[tokio::test] diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 78c5b9f64..268af7343 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -1101,7 +1101,7 @@ pub fn lc_has_active_rules(config: &BucketLifecycleConfiguration, prefix: &str) return true; } - if let Some(Some(true)) = rule.expiration.as_ref().map(|e| e.expired_object_delete_marker.map(|m| m)) { + if let Some(Some(true)) = rule.expiration.as_ref().map(|e| e.expired_object_delete_marker) { return true; } diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index ff8b5971e..49ac85414 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rustfs" -version = "0.1.0" +version.workspace = true edition.workspace = true license.workspace = true repository.workspace = true @@ -15,9 +15,9 @@ path = "src/main.rs" workspace = true [dependencies] -madmin.workspace = true -api = { path = "../s3select/api" } -appauth = { version = "0.0.1", path = "../appauth" } +madmin = { workspace = true } +api = { workspace = true } +appauth = { workspace = true } atoi = { workspace = true } atomic_enum = { workspace = true } axum.workspace = true @@ -27,7 +27,7 @@ async-trait.workspace = true bytes.workspace = true chrono = { workspace = true } clap.workspace = true -crypto = { path = "../crypto" } +crypto = { workspace = true } datafusion = { workspace = true } common.workspace = true const-str = { version = "0.6.1", features = ["std", "proc"] } @@ -40,16 +40,17 @@ hyper.workspace = true hyper-util.workspace = true http.workspace = true http-body.workspace = true -iam = { path = "../iam" } +iam = { workspace = true } lock.workspace = true local-ip-address = { workspace = true } matchit = { workspace = true } mime.workspace = true -mime_guess = "2.0.5" +mime_guess = { workspace = true } pin-project-lite.workspace = true protos.workspace = true -query = { path = "../s3select/query" } +query = { workspace = true } rmp-serde.workspace = true +rustfs-event-notifier = { workspace = true } rustfs-obs = { workspace = true } rustls.workspace = true rustls-pemfile.workspace = true @@ -59,7 +60,7 @@ s3s.workspace = true serde.workspace = true serde_json.workspace = true serde_urlencoded = { workspace = true } -shadow-rs.workspace = true +shadow-rs = { workspace = true, features = ["build", "metadata"] } tracing.workspace = true time = { workspace = true, features = ["parsing", "formatting", "serde"] } tokio-util.workspace = true @@ -92,7 +93,7 @@ bytes.workspace = true futures.workspace = true futures-util.workspace = true # uuid = { version = "1.8.0", features = ["v4", "fast-rng", "serde"] } -ecstore = { path = "../ecstore" } +ecstore = { workspace = true } s3s.workspace = true clap = { workspace = true } hyper-util = { workspace = true, features = [ @@ -102,5 +103,5 @@ hyper-util = { workspace = true, features = [ ] } transform-stream = { workspace = true } netif = "0.1.6" -shadow-rs.workspace = true +shadow-rs = { workspace = true, features = ["build"] } # pin-utils = "0.1.0" diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index 1efac71f7..e877ebada 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -103,6 +103,10 @@ pub struct Opt { #[arg(long, env = "RUSTFS_LICENSE")] pub license: Option, + + /// event notifier config file + #[arg(long, env = "RUSTFS_EVENT_CONFIG")] + pub event_config: Option, } // lazy_static::lazy_static! { diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index fc8a44e8f..f86f951fc 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -46,6 +46,7 @@ use hyper_util::{ use iam::init_iam_sys; use license::init_license; use protos::proto_gen::node_service::node_service_server::NodeServiceServer; +use rustfs_event_notifier::NotifierConfig; use rustfs_obs::{init_obs, load_config, set_global_guard, InitLogStatus}; use rustls::ServerConfig; use s3s::{host::MultiDomain, service::S3ServiceBuilder}; @@ -105,6 +106,23 @@ async fn main() -> Result<()> { // Log initialization status InitLogStatus::init_start_log(&config.observability).await?; + // Initialize event notifier + let notifier_config = opt.clone().event_config; + if notifier_config.is_some() { + info!("event_config is not empty"); + tokio::spawn(async move { + let config = NotifierConfig::event_load_config(notifier_config); + let result = rustfs_event_notifier::initialize(config).await; + if let Err(e) = result { + error!("Failed to initialize event notifier: {}", e); + } else { + info!("Event notifier initialized successfully"); + } + }); + } else { + info!("event_config is empty"); + } + // Run parameters run(opt).await } @@ -448,6 +466,16 @@ async fn run(opt: config::Opt) -> Result<()> { info!("Shutdown signal received in main thread"); // update the status to stopping first state_manager.update(ServiceState::Stopping); + + // Stop the notification system + if rustfs_event_notifier::is_ready() { + // stop event notifier + rustfs_event_notifier::shutdown().await.map_err(|err| { + error!("Failed to shut down the notification system: {}", err); + Error::from_string(err.to_string()) + })?; + } + info!("Server is stopping..."); let _ = shutdown_tx.send(()); // Wait for the worker thread to complete the cleaning work diff --git a/s3select/api/Cargo.toml b/s3select/api/Cargo.toml index 1936e846a..27e260c41 100644 --- a/s3select/api/Cargo.toml +++ b/s3select/api/Cargo.toml @@ -10,9 +10,9 @@ chrono.workspace = true datafusion = { workspace = true } ecstore.workspace = true futures = { workspace = true } -futures-core = "0.3.31" +futures-core = { workspace = true } http.workspace = true -object_store = "0.11.2" +object_store = { workspace = true } s3s.workspace = true snafu = { workspace = true, features = ["backtrace"] } tokio.workspace = true diff --git a/s3select/query/Cargo.toml b/s3select/query/Cargo.toml index 559eab27a..6629927cf 100644 --- a/s3select/query/Cargo.toml +++ b/s3select/query/Cargo.toml @@ -4,7 +4,7 @@ version.workspace = true edition.workspace = true [dependencies] -api = { path = "../api" } +api = { workspace = true } async-recursion = { workspace = true } async-trait.workspace = true datafusion = { workspace = true } diff --git a/scripts/run.sh b/scripts/run.sh index 497c2a681..2cbd115a8 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -57,9 +57,14 @@ export RUSTFS__SINKS__KAFKA__BOOTSTRAP_SERVERS="" export RUSTFS__SINKS__KAFKA__TOPIC="" export RUSTFS__LOGGER__QUEUE_CAPACITY=10 +# 事件消息配置 +export RUSTFS_EVENT_CONFIG="./deploy/config/event.example.toml" + if [ -n "$1" ]; then export RUSTFS_VOLUMES="$1" fi - +# 启动 webhook 服务器 +cargo run --example webhook -p rustfs-event-notifier & +# 启动主服务 cargo run --bin rustfs \ No newline at end of file