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

improve code for event

Browse Source
This commit is contained in:
houseme
2025-05-25 17:52:22 +08:00
parent 0f22b21c8d
commit dee19fdc35
20 changed files with 1258 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2 -1
View File
@@ -7499,8 +7499,8 @@ dependencies = [
"config",
"dotenvy",
"ecstore",
"http",
"lazy_static",
"once_cell",
"rdkafka",
"reqwest",
"rumqttc",
@@ -7508,6 +7508,7 @@ dependencies = [
"serde_json",
"serde_with",
"smallvec",
"snap",
"strum",
"thiserror 2.0.12",
"tokio",
Cargo.toml
+2
View File
@@ -109,6 +109,7 @@ nix = { version = "0.30.1", features = ["fs"] }
num_cpus = { version = "1.16.0" }
nvml-wrapper = "0.10.0"
object_store = "0.11.2"
once_cell = "1.21.3"
opentelemetry = { version = "0.29.1" }
opentelemetry-appender-tracing = { version = "0.29.1", features = [
"experimental_use_tracing_span_context",
@@ -162,6 +163,7 @@ serde_with = "3.12.0"
sha2 = "0.10.9"
smallvec = { version = "1.15.0", features = ["serde"] }
snafu = "0.8.5"
snap = "1.1.1"
socket2 = "0.5.9"
strum = { version = "0.27.1", features = ["derive"] }
sysinfo = "0.34.2"
crates/event/Cargo.toml
+3 -1
View File
@@ -17,8 +17,8 @@ async-trait = { workspace = true }
config = { workspace = true }
common = { workspace = true }
ecstore = { workspace = true }
http = { workspace = true }
lazy_static = { workspace = true }
once_cell = { workspace = true }
reqwest = { workspace = true, optional = true }
rumqttc = { workspace = true, optional = true }
serde = { workspace = true }
@@ -31,6 +31,8 @@ thiserror = { workspace = true }
tokio = { workspace = true, features = ["sync", "net", "macros", "signal", "rt-multi-thread"] }
tokio-util = { workspace = true }
uuid = { workspace = true, features = ["v4", "serde"] }
snap = { workspace = true }
# Only enable kafka features and related dependencies on Linux
[target.'cfg(target_os = "linux")'.dependencies]
crates/event/examples/full.rs
+11 -1
View File
@@ -16,7 +16,17 @@ async fn setup_notification_system() -> Result<(), NotifierError> {
auth_token: Some("your-auth-token".into()),
custom_headers: Some(HashMap::new()),
max_retries: 3,
timeout: 30,
timeout: Some(30),
retry_interval: Some(5),
client_cert: None,
client_key: None,
common: rustfs_event::AdapterCommon {
identifier: "webhook".into(),
comment: "webhook".into(),
enable: true,
queue_dir: "./deploy/logs/event_queue".into(),
queue_limit: 100,
},
})],
};
crates/event/examples/simple.rs
+11 -1
View File
@@ -25,7 +25,17 @@ async fn main() -> Result<(), Box<dyn error::Error>> {
auth_token: Some("secret-token".to_string()),
custom_headers: Some(HashMap::from([("X-Custom".to_string(), "value".to_string())])),
max_retries: 3,
timeout: 10,
timeout: Some(30),
retry_interval: Some(5),
client_cert: None,
client_key: None,
common: rustfs_event::AdapterCommon {
identifier: "webhook".to_string(),
comment: "webhook".to_string(),
enable: true,
queue_dir: "./deploy/logs/event_queue".to_string(),
queue_limit: 100,
},
})],
};
crates/event/examples/webhook.rs
+4 -1
View File
@@ -1,3 +1,4 @@
use axum::routing::get;
use axum::{extract::Json, http::StatusCode, routing::post, Router};
use serde_json::Value;
use std::time::{SystemTime, UNIX_EPOCH};
@@ -5,7 +6,9 @@ use std::time::{SystemTime, UNIX_EPOCH};
#[tokio::main]
async fn main() {
// 构建应用
let app = Router::new().route("/webhook", post(receive_webhook));
let app = Router::new()
.route("/webhook", post(receive_webhook))
.route("/webhook", get(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");
crates/event/src/adapter/kafka.rs
+20 -8
View File
@@ -1,20 +1,22 @@
use crate::Error;
use crate::Event;
use crate::KafkaConfig;
use crate::{ChannelAdapter, ChannelAdapterType};
use crate::{Error, QueueStore};
use async_trait::async_trait;
use rdkafka::error::KafkaError;
use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::types::RDKafkaErrorCode;
use rdkafka::util::Timeout;
use std::path::PathBuf;
use std::sync::Arc;
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,
store: Option<Arc<QueueStore<Event>>>,
config: KafkaConfig,
}
impl KafkaAdapter {
@@ -26,11 +28,21 @@ impl KafkaAdapter {
.set("message.timeout.ms", config.timeout.to_string())
.create()?;
Ok(Self {
producer,
topic: config.topic.clone(),
max_retries: config.max_retries,
})
// create a queue store if enabled
let store = if !config.queue_dir.is_empty() {
let store_path = PathBuf::from(&config.queue_dir);
let store = QueueStore::new(store_path, config.queue_limit, Some(".kafka".to_string()));
if let Err(e) = store.open() {
tracing::error!("Unable to open queue storage: {}", e);
None
} else {
Some(Arc::new(store))
}
} else {
None
};
Ok(Self { config, producer, store })
}
/// Sends an event to the Kafka topic with retry logic.
async fn send_with_retry(&self, event: &Event) -> Result<(), Error> {
crates/event/src/adapter/webhook.rs
+156 -25
View File
@@ -1,15 +1,20 @@
use crate::Error;
use crate::Event;
use crate::store::queue::Store;
use crate::WebhookConfig;
use crate::{ChannelAdapter, ChannelAdapterType};
use crate::{Error, QueueStore};
use crate::{Event, DEFAULT_RETRY_INTERVAL};
use async_trait::async_trait;
use reqwest::{Client, RequestBuilder};
use reqwest::{self, Client, Identity, RequestBuilder};
use std::fs;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::sleep;
/// Webhook adapter for sending events to a webhook endpoint.
pub struct WebhookAdapter {
config: WebhookConfig,
store: Option<Arc<QueueStore<Event>>>,
client: Client,
}
@@ -17,17 +22,154 @@ impl WebhookAdapter {
/// Creates a new Webhook adapter.
pub fn new(config: WebhookConfig) -> Self {
let mut builder = Client::builder();
if config.timeout > 0 {
builder = builder.timeout(Duration::from_secs(config.timeout));
if config.timeout.is_some() {
// Set the timeout for the client
match config.timeout {
Some(t) => builder = builder.timeout(Duration::from_secs(t)),
None => tracing::warn!("Timeout is not set, using default timeout"),
}
}
let client = builder.build().expect("Failed to build reqwest client");
Self { config, client }
let client = if let (Some(cert_path), Some(key_path)) = (&config.client_cert, &config.client_key) {
let cert_path = PathBuf::from(cert_path);
let key_path = PathBuf::from(key_path);
// Check if the certificate file exists
if !cert_path.exists() || !key_path.exists() {
tracing::warn!("Certificate files not found, falling back to default client");
builder.build()
} else {
// Try to read and load the certificate
match (fs::read(&cert_path), fs::read(&key_path)) {
(Ok(cert_data), Ok(key_data)) => {
// Create an identity
let mut pem_data = cert_data;
pem_data.extend_from_slice(&key_data);
match Identity::from_pem(&pem_data) {
Ok(identity) => {
tracing::info!("Successfully loaded client certificate");
builder.identity(identity).build()
}
Err(e) => {
tracing::warn!("Failed to create identity from PEM: {}, falling back to default client", e);
builder.build()
}
}
}
_ => {
tracing::warn!("Failed to read certificate files, falling back to default client");
builder.build()
}
}
}
} else {
builder.build()
}
.expect("Failed to create HTTP client");
// create a queue store if enabled
let store = if !config.common.queue_dir.len() > 0 {
let store_path = PathBuf::from(&config.common.queue_dir);
let store = QueueStore::new(store_path, config.common.queue_limit, Some(".webhook".to_string()));
if let Err(e) = store.open() {
tracing::error!("Unable to open queue storage: {}", e);
None
} else {
Some(Arc::new(store))
}
} else {
None
};
Self { config, store, client }
}
/// Handle backlog events in storage
pub async fn process_backlog(&self) -> Result<(), Error> {
if let Some(store) = &self.store {
let keys = store.list();
for key in keys {
match store.get_multiple(&key) {
Ok(events) => {
for event in events {
if let Err(e) = self.send_with_retry(&event).await {
tracing::error!("Processing of backlog events failed: {}", e);
continue;
}
}
// Deleted after successful processing
let _ = store.del(&key);
}
Err(e) => {
tracing::error!("Failed to read events from storage: {}", e);
// delete the broken entries
let _ = store.del(&key);
}
}
}
}
Ok(())
}
///Send events to the webhook endpoint with retry logic
async fn send_with_retry(&self, event: &Event) -> Result<(), Error> {
let retry_interval = match self.config.retry_interval {
Some(t) => Duration::from_secs(t),
None => Duration::from_secs(DEFAULT_RETRY_INTERVAL), // Default to 3 seconds if not set
};
let mut attempts = 0;
loop {
attempts += 1;
match self.send_request(event).await {
Ok(_) => return Ok(()),
Err(e) => {
if attempts <= self.config.max_retries {
tracing::warn!("Send to webhook fails and will be retried after 3 seconds:{}", e);
sleep(retry_interval).await;
} else if let Some(store) = &self.store {
// store in a queue for later processing
tracing::warn!("The maximum number of retries is reached, and the event is stored in a queue:{}", e);
if let Err(store_err) = store.put(event.clone()) {
tracing::error!("Events cannot be stored to a queue:{}", store_err);
}
return Err(e);
} else {
return Err(e);
}
}
}
}
}
/// Send a single HTTP request
async fn send_request(&self, event: &Event) -> Result<(), Error> {
// Send a request
let response = self.build_request(event).send().await?;
// Check the response status
if !response.status().is_success() {
return Err(Error::Custom(format!("Webhook request failed, status code:{}", response.status())));
}
Ok(())
}
/// 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);
let mut request = self
.client
.post(&self.config.endpoint)
.json(event)
.header("Content-Type", "application/json");
if let Some(token) = &self.config.auth_token {
request = request.header("Authorization", format!("Bearer {}", token));
let tokens: Vec<&str> = token.split_whitespace().collect();
match tokens.len() {
2 => request = request.header("Authorization", token),
1 => request = request.header("Authorization", format!("Bearer {}", token)),
_ => tracing::warn!("Invalid auth token format, skipping Authorization header"),
}
}
if let Some(headers) = &self.config.custom_headers {
for (key, value) in headers {
@@ -45,21 +187,10 @@ impl ChannelAdapter for WebhookAdapter {
}
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)),
}
}
// Deal with the backlog of events first
let _ = self.process_backlog().await;
// Send the current event
self.send_with_retry(event).await
}
}
crates/event/src/config.rs
+240 -16
View File
@@ -2,17 +2,64 @@ use config::{Config, File, FileFormat};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::env;
use std::path::Path;
use tracing::info;
/// The default configuration file name
const DEFAULT_CONFIG_FILE: &str = "event";
/// Configuration for the webhook adapter.
/// The prefix for the configuration file
pub(crate) const STORE_PREFIX: &str = "rustfs";
/// The default retry interval for the webhook adapter
pub const DEFAULT_RETRY_INTERVAL: u64 = 3;
/// The default maximum retry count for the webhook adapter
pub const DEFAULT_MAX_RETRIES: u32 = 3;
/// Add a common field for the adapter configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AdapterCommon {
/// Adapter identifier for unique identification
pub identifier: String,
/// Adapter description information
pub comment: String,
/// Whether to enable this adapter
#[serde(default)]
pub enable: bool,
#[serde(default = "default_queue_dir")]
pub queue_dir: String,
#[serde(default = "default_queue_limit")]
pub queue_limit: u64,
}
impl Default for AdapterCommon {
fn default() -> Self {
Self {
identifier: String::new(),
comment: String::new(),
enable: false,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
}
}
}
/// Configuration for the webhook adapter.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct WebhookConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub endpoint: String,
pub auth_token: Option<String>,
pub custom_headers: Option<HashMap<String, String>>,
pub max_retries: u32,
pub timeout: u64,
pub retry_interval: Option<u64>,
pub timeout: Option<u64>,
#[serde(default)]
pub client_cert: Option<String>,
#[serde(default)]
pub client_key: Option<String>,
}
impl WebhookConfig {
@@ -26,25 +73,41 @@ impl WebhookConfig {
///
/// ```
/// use rustfs_event::WebhookConfig;
/// use rustfs_event::AdapterCommon;
/// use rustfs_event::DEFAULT_RETRY_INTERVAL;
///
/// let config = WebhookConfig {
/// endpoint: "http://example.com/webhook".to_string(),
/// common: AdapterCommon::default(),
/// endpoint: "https://example.com/webhook".to_string(),
/// auth_token: Some("my_token".to_string()),
/// custom_headers: None,
/// max_retries: 3,
/// timeout: 5000,
/// retry_interval: Some(DEFAULT_RETRY_INTERVAL),
/// timeout: Some(5000),
/// client_cert: None,
/// client_key: None,
/// };
///
/// assert!(config.validate().is_ok());
pub fn validate(&self) -> Result<(), String> {
// If not enabled, the other fields are not validated
if !self.common.enable {
return Ok(());
}
// 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());
if self.timeout.is_some() {
match self.timeout {
Some(timeout) if timeout > 0 => {
info!("Webhook timeout is set to {}", timeout);
}
_ => return Err("Webhook timeout must be greater than 0".to_string()),
}
}
// Verify that the maximum number of retry is reasonable
@@ -52,22 +115,82 @@ impl WebhookConfig {
return Err("Maximum retry count cannot exceed 10".to_string());
}
// Verify the queue directory path
if !self.common.queue_dir.is_empty() && !Path::new(&self.common.queue_dir).is_absolute() {
return Err("Queue directory path should be absolute".to_string());
}
// The authentication certificate and key must appear in pairs
if (self.client_cert.is_some() && self.client_key.is_none()) || (self.client_cert.is_none() && self.client_key.is_some())
{
return Err("Certificate and key must be specified as a pair".to_string());
}
Ok(())
}
/// Create a new webhook configuration
pub fn new(identifier: impl Into<String>, endpoint: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
endpoint: endpoint.into(),
..Default::default()
}
}
}
/// Configuration for the Kafka adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KafkaConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub brokers: String,
pub topic: String,
pub max_retries: u32,
pub timeout: u64,
}
impl Default for KafkaConfig {
fn default() -> Self {
Self {
common: AdapterCommon::default(),
brokers: String::new(),
topic: String::new(),
max_retries: 3,
timeout: 5000,
}
}
}
impl KafkaConfig {
/// Create a new Kafka configuration
pub fn new(identifier: impl Into<String>, brokers: impl Into<String>, topic: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
brokers: brokers.into(),
topic: topic.into(),
..Default::default()
}
}
}
/// Configuration for the MQTT adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MqttConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub broker: String,
pub port: u16,
pub client_id: String,
@@ -75,6 +198,37 @@ pub struct MqttConfig {
pub max_retries: u32,
}
impl Default for MqttConfig {
fn default() -> Self {
Self {
common: AdapterCommon::default(),
broker: String::new(),
port: 1883,
client_id: String::new(),
topic: String::new(),
max_retries: 3,
}
}
}
impl MqttConfig {
/// Create a new MQTT configuration
pub fn new(identifier: impl Into<String>, broker: impl Into<String>, topic: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
broker: broker.into(),
topic: topic.into(),
..Default::default()
}
}
}
/// Configuration for the adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
@@ -104,18 +258,18 @@ pub enum AdapterConfig {
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NotifierConfig {
#[serde(default = "default_store_path")]
#[serde(default = "default_queue_dir")]
pub store_path: String,
#[serde(default = "default_channel_capacity")]
pub channel_capacity: usize,
#[serde(default = "default_queue_limit")]
pub channel_capacity: u64,
pub adapters: Vec<AdapterConfig>,
}
impl Default for NotifierConfig {
fn default() -> Self {
Self {
store_path: default_store_path(),
channel_capacity: default_channel_capacity(),
store_path: default_queue_dir(),
channel_capacity: default_queue_limit(),
adapters: Vec::new(),
}
}
@@ -150,7 +304,7 @@ impl NotifierConfig {
DEFAULT_CONFIG_FILE.to_string()
} else {
// Use the provided path
let path = std::path::Path::new(&path);
let 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()
@@ -209,17 +363,87 @@ impl NotifierConfig {
}
/// Provide temporary directories as default storage paths
fn default_store_path() -> String {
env::var("EVENT_STORE_PATH").unwrap_or_else(|e| {
tracing::info!("Failed to get `EVENT_STORE_PATH` failed err: {}", e.to_string());
fn default_queue_dir() -> String {
env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| {
tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string());
env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string()
})
}
/// Provides the recommended default channel capacity for high concurrency systems
fn default_channel_capacity() -> usize {
fn default_queue_limit() -> u64 {
env::var("EVENT_CHANNEL_CAPACITY")
.unwrap_or_else(|_| "10000".to_string())
.parse()
.unwrap_or(10000) // Default to 10000 if parsing fails
}
/// Event Notifier Configuration
/// This struct contains the configuration for the event notifier system,
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct EventNotifierConfig {
/// A collection of webhook configurations, with the key being a unique identifier
#[serde(default)]
pub webhook: HashMap<String, WebhookConfig>,
/// A collection of Kafka configurations, with the key being a unique identifier
#[serde(default)]
pub kafka: HashMap<String, KafkaConfig>,
///MQTT configuration collection, with the key being a unique identifier
#[serde(default)]
pub mqtt: HashMap<String, MqttConfig>,
}
impl EventNotifierConfig {
/// Create a new default configuration
pub fn new() -> Self {
Self::default()
}
/// Load the configuration from the file
pub fn event_load_config(_config_dir: Option<String>) -> EventNotifierConfig {
// The existing implementation remains the same, but returns EventNotifierConfig
// ...
EventNotifierConfig::default()
}
/// Deserialization configuration
pub fn unmarshal(data: &[u8]) -> common::error::Result<EventNotifierConfig> {
let m: EventNotifierConfig = serde_json::from_slice(data)?;
Ok(m)
}
/// Serialization configuration
pub fn marshal(&self) -> common::error::Result<Vec<u8>> {
let data = serde_json::to_vec(&self)?;
Ok(data)
}
/// Convert this configuration to a list of adapter configurations
pub fn to_adapter_configs(&self) -> Vec<AdapterConfig> {
let mut adapters = Vec::new();
// Add all enabled webhook configurations
for (_, webhook) in &self.webhook {
if webhook.common.enable {
adapters.push(AdapterConfig::Webhook(webhook.clone()));
}
}
// Add all enabled Kafka configurations
for (_, kafka) in &self.kafka {
if kafka.common.enable {
adapters.push(AdapterConfig::Kafka(kafka.clone()));
}
}
// Add all enabled MQTT configurations
for (_, mqtt) in &self.mqtt {
if mqtt.common.enable {
adapters.push(AdapterConfig::Mqtt(mqtt.clone()));
}
}
adapters
}
}
crates/event/src/error.rs
+2
View File
@@ -37,6 +37,8 @@ pub enum Error {
ConfigError(String),
#[error("Configuration loading error: {0}")]
Config(#[from] ConfigError),
#[error("create adapter failed error: {0}")]
AdapterCreationFailed(String),
}
impl Error {
crates/event/src/global.rs
+12 -2
View File
@@ -174,7 +174,7 @@ async fn get_system() -> Result<Arc<Mutex<NotifierSystem>>, Error> {
#[cfg(test)]
mod tests {
use super::*;
use crate::{AdapterConfig, NotifierConfig, WebhookConfig};
use crate::{AdapterCommon, AdapterConfig, NotifierConfig, WebhookConfig};
use std::collections::HashMap;
#[tokio::test]
@@ -205,11 +205,21 @@ mod tests {
adapters: vec![
// assuming that the empty adapter will cause failure
AdapterConfig::Webhook(WebhookConfig {
common: AdapterCommon {
identifier: "empty".to_string(),
comment: "empty".to_string(),
enable: true,
queue_dir: "".to_string(),
queue_limit: 10,
},
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,
timeout: Some(10),
retry_interval: Some(5),
client_cert: None,
client_key: None,
}),
], // assuming that the empty adapter will cause failure
..Default::default()
crates/event/src/lib.rs
+5 -1
View File
@@ -6,6 +6,7 @@ mod event;
mod global;
mod notifier;
mod store;
mod target;
pub use adapter::create_adapters;
#[cfg(all(feature = "kafka", target_os = "linux"))]
@@ -17,13 +18,15 @@ pub use adapter::webhook::WebhookAdapter;
pub use adapter::ChannelAdapter;
pub use adapter::ChannelAdapterType;
pub use bus::event_bus;
pub use config::AdapterCommon;
pub use config::EventNotifierConfig;
#[cfg(all(feature = "kafka", target_os = "linux"))]
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 config::{AdapterConfig, NotifierConfig, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL};
pub use error::Error;
pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source};
@@ -32,6 +35,7 @@ pub use notifier::NotifierSystem;
pub use store::event::EventStore;
pub use store::get_event_notifier_config;
pub use store::queue::QueueStore;
pub use store::EventSys;
pub use store::GLOBAL_EventSys;
crates/event/src/notifier.rs
+1 -1
View File
@@ -21,7 +21,7 @@ impl NotifierSystem {
/// Creates a new `NotificationSystem` instance.
#[instrument(skip(config))]
pub async fn new(config: NotifierConfig) -> Result<Self, Error> {
let (tx, rx) = mpsc::channel::<Event>(config.channel_capacity);
let (tx, rx) = mpsc::channel::<Event>(config.channel_capacity.try_into().unwrap());
let store = Arc::new(EventStore::new(&config.store_path).await?);
let shutdown = CancellationToken::new();
crates/event/src/store/manager.rs
+236
View File
@@ -0,0 +1,236 @@
use crate::store::{CONFIG_FILE, EVENT};
use crate::{adapter, ChannelAdapter, EventNotifierConfig, WebhookAdapter};
use common::error::{Error, Result};
use ecstore::config::com::{read_config, save_config, CONFIG_PREFIX};
use ecstore::disk::RUSTFS_META_BUCKET;
use ecstore::store::ECStore;
use ecstore::store_api::ObjectOptions;
use ecstore::utils::path::SLASH_SEPARATOR;
use ecstore::StorageAPI;
use once_cell::sync::Lazy;
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::instrument;
/// Global storage API access point
pub static GLOBAL_STORE_API: Lazy<Mutex<Option<Arc<ECStore>>>> = Lazy::new(|| Mutex::new(None));
/// Global event system configuration
pub static GLOBAL_EVENT_CONFIG: Lazy<Mutex<Option<EventNotifierConfig>>> = Lazy::new(|| Mutex::new(None));
/// EventManager Responsible for managing all operations of the event system
#[derive(Debug)]
pub struct EventManager {
api: Arc<ECStore>,
}
impl EventManager {
/// Create a new Event Manager
pub async fn new(api: Arc<ECStore>) -> Self {
// Update the global access point at the same time
{
let mut global_api = GLOBAL_STORE_API.lock().await;
*global_api = Some(api.clone());
}
Self { api }
}
/// Initialize the Event Manager
///
/// # Returns
/// If it succeeds, it returns configuration information, and if it fails, it returns an error
#[instrument(skip_all)]
pub async fn init(&self) -> Result<EventNotifierConfig> {
tracing::info!("Event system configuration initialization begins");
let cfg = match read_event_config(self.api.clone()).await {
Ok(cfg) => cfg,
Err(err) => {
tracing::error!("Failed to initialize the event system configuration:{:?}", err);
return Err(err);
}
};
*GLOBAL_EVENT_CONFIG.lock().await = Some(cfg.clone());
tracing::info!("The initialization of the event system configuration is complete");
Ok(cfg)
}
/// Create a new configuration
///
/// # Parameters
/// - `cfg`: The configuration to be created
///
/// # Returns
/// The result of the operation
pub async fn create_config(&self, cfg: &EventNotifierConfig) -> Result<()> {
// Check whether the configuration already exists
if let Ok(_) = read_event_config(self.api.clone()).await {
return Err(Error::msg("The configuration already exists, use the update action"));
}
save_event_config(self.api.clone(), cfg).await?;
*GLOBAL_EVENT_CONFIG.lock().await = Some(cfg.clone());
Ok(())
}
/// Update the configuration
///
/// # Parameters
/// - `cfg`: The configuration to be updated
///
/// # Returns
/// The result of the operation
pub async fn update_config(&self, cfg: &EventNotifierConfig) -> Result<()> {
// Read the existing configuration first to merge
let current_cfg = read_event_config(self.api.clone()).await.unwrap_or_default();
// This is where the merge logic can be implemented
let merged_cfg = self.merge_configs(current_cfg, cfg.clone());
save_event_config(self.api.clone(), &merged_cfg).await?;
*GLOBAL_EVENT_CONFIG.lock().await = Some(merged_cfg);
Ok(())
}
/// Merge the two configurations
fn merge_configs(&self, current: EventNotifierConfig, new: EventNotifierConfig) -> EventNotifierConfig {
let mut merged = current;
// Merge webhook configurations
for (id, config) in new.webhook {
merged.webhook.insert(id, config);
}
// Merge Kafka configurations
for (id, config) in new.kafka {
merged.kafka.insert(id, config);
}
// Merge MQTT configurations
for (id, config) in new.mqtt {
merged.mqtt.insert(id, config);
}
merged
}
/// Delete the configuration
pub async fn delete_config(&self) -> Result<()> {
let config_file = get_event_config_file();
self.api
.delete_object(
RUSTFS_META_BUCKET,
&config_file,
ObjectOptions {
delete_prefix: true,
delete_prefix_object: true,
..Default::default()
},
)
.await?;
// Reset the global configuration to default
// let _ = GLOBAL_EventSysConfig.set(self.read_config().await?);
Ok(())
}
/// Read the configuration
pub async fn read_config(&self) -> Result<EventNotifierConfig> {
read_event_config(self.api.clone()).await
}
/// Create all enabled adapters
pub async fn create_adapters(&self) -> Result<Vec<Arc<dyn ChannelAdapter>>> {
let config = match GLOBAL_EVENT_CONFIG.lock().await.clone() {
Some(cfg) => cfg,
None => return Err(Error::msg("The global configuration is not initialized")),
};
let mut adapters: Vec<Arc<dyn ChannelAdapter>> = Vec::new();
// Create a webhook adapter
for (_, webhook_config) in &config.webhook {
if webhook_config.common.enable {
#[cfg(feature = "webhook")]
{
adapters.push(Arc::new(WebhookAdapter::new(webhook_config.clone())));
}
#[cfg(not(feature = "webhook"))]
{
return Err(Error::msg("The webhook feature is not enabled"));
}
}
}
// Create a Kafka adapter
for (_, kafka_config) in &config.kafka {
if kafka_config.common.enable {
#[cfg(all(feature = "kafka", target_os = "linux"))]
{
match KafkaAdapter::new(kafka_config.clone()) {
Ok(adapter) => adapters.push(Arc::new(adapter)),
Err(e) => tracing::error!("Failed to create a Kafka adapter:{}", e),
}
}
#[cfg(any(not(feature = "kafka"), not(target_os = "linux")))]
{
return Err(Error::msg("Kafka functionality is not enabled or is not a Linux environment"));
}
}
}
// Create an MQTT adapter
for (_, mqtt_config) in &config.mqtt {
if mqtt_config.common.enable {
#[cfg(feature = "mqtt")]
{
// Implement MQTT adapter creation logic
// ...
}
#[cfg(not(feature = "mqtt"))]
{
return Err(Error::msg("MQTT The feature is not enabled"));
}
}
}
Ok(adapters)
}
}
/// Get the Global Storage API
pub async fn get_global_event_config() -> Option<EventNotifierConfig> {
GLOBAL_EVENT_CONFIG.lock().await.clone()
}
/// Read event configuration
async fn read_event_config(api: Arc<ECStore>) -> Result<EventNotifierConfig> {
let config_file = get_event_config_file();
let data = read_config(api, &config_file).await?;
EventNotifierConfig::unmarshal(&data)
}
/// Save the event configuration
async fn save_event_config(api: Arc<ECStore>, config: &EventNotifierConfig) -> Result<()> {
let config_file = get_event_config_file();
let data = config.marshal()?;
save_config(api, &config_file, data).await?;
Ok(())
}
/// Get the event profile path
fn get_event_config_file() -> String {
// "event/config.json".to_string()
format!("{}{}{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, EVENT, SLASH_SEPARATOR, CONFIG_FILE)
}
crates/event/src/store/mod.rs
+2
View File
@@ -1,4 +1,6 @@
pub(crate) mod event;
mod manager;
pub(crate) mod queue;
use crate::NotifierConfig;
use common::error::Result;
crates/event/src/store/queue.rs
+449
View File
@@ -0,0 +1,449 @@
use common::error::{Error, Result};
use serde::{de::DeserializeOwned, Serialize};
use snap::raw::{Decoder, Encoder};
use std::collections::HashMap;
use std::io::Read;
use std::marker::PhantomData;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};
use std::time::{SystemTime, UNIX_EPOCH};
use std::{fs, io};
use uuid::Uuid;
/// Keys in storage
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Key {
/// Key name
pub name: String,
/// Whether or not to compress
pub compress: bool,
/// filename extension
pub extension: String,
/// Number of items
pub item_count: usize,
}
impl Key {
/// Create a new key
pub fn new(name: impl Into<String>, extension: impl Into<String>, compress: bool) -> Self {
Self {
name: name.into(),
compress,
extension: extension.into(),
item_count: 1,
}
}
/// Convert to string form
pub fn to_string(&self) -> String {
let mut key_str = self.name.clone();
if self.item_count > 1 {
key_str = format!("{}:{}", self.item_count, self.name);
}
let compress_ext = if self.compress { ".snappy" } else { "" };
format!("{}{}{}", key_str, self.extension, compress_ext)
}
}
/// Parse key from file name
pub fn parse_key(filename: &str) -> Key {
let compress = filename.ends_with(".snappy");
let filename = if compress {
&filename[..filename.len() - 7] // 移除 ".snappy"
} else {
filename
};
let mut parts = filename.splitn(2, '.');
let name_part = parts.next().unwrap_or("");
let extension = parts
.next()
.map_or_else(|| String::new(), |ext| format!(".{}", ext))
.to_string();
let mut name = name_part.to_string();
let mut item_count = 1;
if let Some(pos) = name_part.find(':') {
if let Ok(count) = name_part[..pos].parse::<usize>() {
item_count = count;
name = name_part[pos + 1..].to_string();
}
}
Key {
name,
compress,
extension,
item_count,
}
}
/// Store the characteristics of the project
pub trait Store<T>: Send + Sync
where
T: Serialize + DeserializeOwned + Send + Sync,
{
/// Store a single item
fn put(&self, item: T) -> Result<Key>;
/// Store multiple projects
fn put_multiple(&self, items: Vec<T>) -> Result<Key>;
/// Get a single item
fn get(&self, key: &Key) -> Result<T>;
/// Get multiple items
fn get_multiple(&self, key: &Key) -> Result<Vec<T>>;
/// Get the raw bytes
fn get_raw(&self, key: &Key) -> Result<Vec<u8>>;
/// Stores raw bytes
fn put_raw(&self, data: &[u8]) -> Result<Key>;
/// Gets the number of items in storage
fn len(&self) -> usize;
/// Whether it is empty or not
fn is_empty(&self) -> bool {
self.len() == 0
}
/// Lists all keys
fn list(&self) -> Vec<Key>;
/// Delete the key
fn del(&self, key: &Key) -> Result<()>;
/// Open Storage
fn open(&self) -> Result<()>;
/// Delete the storage
fn delete(&self) -> Result<()>;
}
const DEFAULT_LIMIT: u64 = 100000;
const DEFAULT_EXT: &str = ".unknown";
const COMPRESS_EXT: &str = ".snappy";
/// Queue storage implementation
pub struct QueueStore<T> {
/// Project Limitations
entry_limit: u64,
/// Storage directory
directory: PathBuf,
/// filename extension
file_ext: String,
/// Item mapping: key -> modified time (Unix nanoseconds)
entries: Arc<RwLock<HashMap<String, i64>>>,
/// Type tags
_phantom: PhantomData<T>,
}
impl<T> QueueStore<T>
where
T: Serialize + DeserializeOwned + Send + Sync,
{
/// Create a new queue store
pub fn new(directory: impl Into<PathBuf>, limit: u64, ext: Option<String>) -> Self {
let limit = if limit == 0 { DEFAULT_LIMIT } else { limit };
let ext = ext.unwrap_or_else(|| DEFAULT_EXT.to_string());
Self {
directory: directory.into(),
entry_limit: limit,
file_ext: ext,
entries: Arc::new(RwLock::new(HashMap::with_capacity(limit as usize))),
_phantom: PhantomData,
}
}
/// Lists all files in the directory, sorted by modification time (oldest takes precedence.)）
fn list_files(&self) -> Result<Vec<fs::DirEntry>> {
let mut files = Vec::new();
for entry in fs::read_dir(&self.directory)? {
let entry = entry?;
let metadata = entry.metadata()?;
if metadata.is_file() {
files.push(entry);
}
}
// Sort by modification time
files.sort_by(|a, b| {
let a_time = a
.metadata()
.map(|m| m.modified())
.unwrap_or(Ok(UNIX_EPOCH))
.unwrap_or(UNIX_EPOCH);
let b_time = b
.metadata()
.map(|m| m.modified())
.unwrap_or(Ok(UNIX_EPOCH))
.unwrap_or(UNIX_EPOCH);
a_time.cmp(&b_time)
});
Ok(files)
}
/// Write the object to a file
fn write_object(&self, key: &Key, item: &T) -> Result<()> {
// 序列化对象
let data = serde_json::to_vec(item)?;
self.write_bytes(key, &data)
}
/// Write multiple objects to a file
fn write_multiple_objects(&self, key: &Key, items: &[T]) -> Result<()> {
let mut data = Vec::new();
for item in items {
let item_data = serde_json::to_vec(item)?;
data.extend_from_slice(&item_data);
data.push(b'\n');
}
self.write_bytes(key, &data)
}
/// Write bytes to a file
fn write_bytes(&self, key: &Key, data: &[u8]) -> Result<()> {
let path = self.directory.join(key.to_string());
let file_data = if key.compress {
// 使用 snap 压缩数据
let mut encoder = Encoder::new();
encoder
.compress_vec(data)
.map_err(|e| Error::msg(format!("Compression failed:{}", e)))?
} else {
data.to_vec()
};
// Make sure the directory exists
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
// Write to the file
fs::write(&path, &file_data)?;
// Update the item mapping
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64;
let mut entries = self.entries.write().map_err(|_| Error::msg("获取写锁失败"))?;
entries.insert(key.to_string(), now);
Ok(())
}
/// Read bytes from a file
fn read_bytes(&self, key: &Key) -> Result<Vec<u8>> {
let path = self.directory.join(key.to_string());
let data = fs::read(&path)?;
if data.is_empty() {
return Err(Error::msg("The file is empty"));
}
if key.compress {
// Use Snap to extract the data
let mut decoder = Decoder::new();
decoder
.decompress_vec(&data)
.map_err(|e| Error::msg(format!("Failed to decompress:{}", e)))
} else {
Ok(data)
}
}
}
impl<T> Store<T> for QueueStore<T>
where
T: Serialize + DeserializeOwned + Send + Sync + Clone,
{
fn open(&self) -> Result<()> {
// Create a directory (if it doesn't exist)
fs::create_dir_all(&self.directory)?;
// Read existing files
let files = self.list_files()?;
let mut entries = self.entries.write().map_err(|_| Error::msg("获取写锁失败"))?;
for file in files {
if let Ok(meta) = file.metadata() {
if let Ok(modified) = meta.modified() {
if let Ok(since_epoch) = modified.duration_since(UNIX_EPOCH) {
entries.insert(file.file_name().to_string_lossy().to_string(), since_epoch.as_nanos() as i64);
}
}
}
}
Ok(())
}
fn delete(&self) -> Result<()> {
fs::remove_dir_all(&self.directory)?;
Ok(())
}
fn put(&self, item: T) -> Result<Key> {
{
let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?;
if entries.len() as u64 >= self.entry_limit {
return Err(Error::msg("The storage limit has been reached"));
}
}
// generate a new uuid
let uuid = Uuid::new_v4();
let key = Key::new(uuid.to_string(), &self.file_ext, true);
self.write_object(&key, &item)?;
Ok(key)
}
fn put_multiple(&self, items: Vec<T>) -> Result<Key> {
if items.is_empty() {
return Err(Error::msg("The list of items is empty"));
}
{
let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?;
if entries.len() as u64 >= self.entry_limit {
return Err(Error::msg("The storage limit has been reached"));
}
}
// Generate a new UUID
let uuid = Uuid::new_v4();
let mut key = Key::new(uuid.to_string(), &self.file_ext, true);
key.item_count = items.len();
self.write_multiple_objects(&key, &items)?;
Ok(key)
}
fn get(&self, key: &Key) -> Result<T> {
let items = self.get_multiple(key)?;
if items.is_empty() {
return Err(Error::msg("Item not found"));
}
Ok(items[0].clone())
}
fn get_multiple(&self, key: &Key) -> Result<Vec<T>> {
let data = self.get_raw(key)?;
let mut items = Vec::with_capacity(key.item_count);
let mut reader = io::Cursor::new(&data);
// Try to read each JSON object
let mut buffer = Vec::new();
// if the read fails try parsing it once
if reader.read_to_end(&mut buffer).is_err() {
// Try to parse the entire data as a single object
match serde_json::from_slice::<T>(&data) {
Ok(item) => {
items.push(item);
return Ok(items);
}
Err(_) => {
// An attempt was made to resolve to an array of objects
match serde_json::from_slice::<Vec<T>>(&data) {
Ok(array_items) => return Ok(array_items),
Err(e) => return Err(Error::msg(format!("Failed to parse the data:{}", e))),
}
}
}
}
// Read JSON objects by row
for line in buffer.split(|&b| b == b'\n') {
if !line.is_empty() {
match serde_json::from_slice::<T>(line) {
Ok(item) => items.push(item),
Err(e) => tracing::warn!("Failed to parse row data:{}", e),
}
}
}
if items.is_empty() {
return Err(Error::msg("Failed to resolve any items"));
}
Ok(items)
}
fn get_raw(&self, key: &Key) -> Result<Vec<u8>> {
let data = self.read_bytes(key)?;
// Delete the wrong file
if data.is_empty() {
let _ = self.del(key);
return Err(Error::msg("the file is empty"));
}
Ok(data)
}
fn put_raw(&self, data: &[u8]) -> Result<Key> {
{
let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?;
if entries.len() as u64 >= self.entry_limit {
return Err(Error::msg("the storage limit has been reached"));
}
}
// Generate a new UUID
let uuid = Uuid::new_v4();
let key = Key::new(uuid.to_string(), &self.file_ext, true);
self.write_bytes(&key, data)?;
Ok(key)
}
fn len(&self) -> usize {
self.entries.read().map(|e| e.len()).unwrap_or(0)
}
fn list(&self) -> Vec<Key> {
let entries = match self.entries.read() {
Ok(guard) => guard,
Err(_) => return Vec::new(),
};
// Convert entries to vectors and sort by timestamp
let mut entries_vec: Vec<_> = entries.iter().collect();
entries_vec.sort_by(|a, b| a.1.cmp(b.1));
// Parsing key
entries_vec.iter().map(|(filename, _)| parse_key(filename)).collect()
}
fn del(&self, key: &Key) -> Result<()> {
let path = self.directory.join(key.to_string());
// Delete the file
if let Err(e) = fs::remove_file(&path) {
if e.kind() != io::ErrorKind::NotFound {
return Err(e.into());
}
}
// Remove the entry from the map
let mut entries = self.entries.write().map_err(|_| Error::msg("获取写锁失败"))?;
entries.remove(&key.to_string());
Ok(())
}
}
crates/event/src/target.rs
+76
View File
@@ -0,0 +1,76 @@
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::fmt;
///TargetID - Holds the identity and name string of the notification target
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TargetID {
/// Destination ID
pub id: String,
/// Target name
pub name: String,
}
impl TargetID {
/// Create a new TargetID
pub fn new(id: impl Into<String>, name: impl Into<String>) -> Self {
Self {
id: id.into(),
name: name.into(),
}
}
/// Convert to ARN
pub fn to_arn(&self, region: &str) -> ARN {
ARN {
target_id: self.clone(),
region: region.to_string(),
}
}
/// The parsed string is TargetID
pub fn parse(s: &str) -> Result<Self, String> {
let tokens: Vec<&str> = s.split(':').collect();
if tokens.len() != 2 {
return Err(format!("Invalid TargetID format '{}'", s));
}
Ok(Self {
id: tokens[0].to_string(),
name: tokens[1].to_string(),
})
}
}
impl fmt::Display for TargetID {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}:{}", self.id, self.name)
}
}
impl Serialize for TargetID {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(&self.to_string())
}
}
impl<'de> Deserialize<'de> for TargetID {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let s = String::deserialize(deserializer)?;
TargetID::parse(&s).map_err(serde::de::Error::custom)
}
}
/// ARN - Amazon Resource Name structure
#[derive(Debug, Clone)]
pub struct ARN {
/// Destination ID
pub target_id: TargetID,
/// region
pub region: String,
}
crates/event/tests/integration.rs
+22 -4
View File
@@ -1,4 +1,4 @@
use rustfs_event::{AdapterConfig, ChannelAdapterType, NotifierSystem, WebhookConfig};
use rustfs_event::{AdapterCommon, AdapterConfig, ChannelAdapterType, NotifierSystem, WebhookConfig};
use rustfs_event::{Bucket, Event, EventBuilder, Identity, Metadata, Name, Object, Source};
use rustfs_event::{ChannelAdapter, WebhookAdapter};
use std::collections::HashMap;
@@ -7,11 +7,21 @@ use std::sync::Arc;
#[tokio::test]
async fn test_webhook_adapter() {
let adapter = WebhookAdapter::new(WebhookConfig {
common: AdapterCommon {
identifier: "webhook".to_string(),
comment: "webhook".to_string(),
enable: true,
queue_dir: "./deploy/logs/event_queue".to_string(),
queue_limit: 100,
},
endpoint: "http://localhost:8080/webhook".to_string(),
auth_token: None,
custom_headers: None,
max_retries: 1,
timeout: 5,
timeout: Some(5),
retry_interval: Some(5),
client_cert: None,
client_key: None,
});
// create an s3 metadata object
@@ -71,20 +81,28 @@ async fn test_notification_system() {
store_path: "./test_events".to_string(),
channel_capacity: 100,
adapters: vec![AdapterConfig::Webhook(WebhookConfig {
common: Default::default(),
endpoint: "http://localhost:8080/webhook".to_string(),
auth_token: None,
custom_headers: None,
max_retries: 1,
timeout: 5,
timeout: Some(5),
retry_interval: Some(5),
client_cert: None,
client_key: None,
})],
};
let system = Arc::new(tokio::sync::Mutex::new(NotifierSystem::new(config.clone()).await.unwrap()));
let adapters: Vec<Arc<dyn ChannelAdapter>> = vec![Arc::new(WebhookAdapter::new(WebhookConfig {
common: Default::default(),
endpoint: "http://localhost:8080/webhook".to_string(),
auth_token: None,
custom_headers: None,
max_retries: 1,
timeout: 5,
timeout: Some(5),
retry_interval: Some(5),
client_cert: None,
client_key: None,
}))];
// create an s3 metadata object
ecstore/src/store.rs
+1 -1
View File
@@ -1671,7 +1671,7 @@ impl StorageAPI for ECStore {
let object = utils::path::encode_dir_object(object);
let object = object.as_str();
// 查询在哪个 pool
// Query which pool it is in
let (mut pinfo, errs) = self
.get_pool_info_existing_with_opts(bucket, object, &opts)
.await
rustfs/src/admin/handlers/sts.rs
+3 -3
View File
@@ -50,7 +50,7 @@ impl Operation for AssumeRoleHandle {
let (cred, _owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &user.access_key).await?;
// // TODO: 判断权限, 不允许sts访问
// // TODO: 判断权限，不允许 sts 访问
if cred.is_temp() || cred.is_service_account() {
return Err(s3_error!(InvalidRequest, "AccessDenied"));
}
@@ -68,11 +68,11 @@ impl Operation for AssumeRoleHandle {
let body: AssumeRoleRequest = from_bytes(&bytes).map_err(|_e| s3_error!(InvalidRequest, "get body failed"))?;
if body.action.as_str() != ASSUME_ROLE_ACTION {
return Err(s3_error!(InvalidArgument, "not suport action"));
return Err(s3_error!(InvalidArgument, "not support action"));
}
if body.version.as_str() != ASSUME_ROLE_VERSION {
return Err(s3_error!(InvalidArgument, "not suport version"));
return Err(s3_error!(InvalidArgument, "not support version"));
}
let mut claims = cred.claims.unwrap_or_default();