mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-31 09:18:28 +00:00
chore: improve event and docker-compose ,Improve the permissions of the endpoint health interface, upgrade otel from 0.30.0 to 0.31.0 (#620)
* feat: improve code for notify * upgrade starshard version * upgrade version * Fix ETag format to comply with HTTP standards by wrapping with quotes (#592) * Initial plan * Fix ETag format to comply with HTTP standards by wrapping with quotes Co-authored-by: overtrue <1472352+overtrue@users.noreply.github.com> * bufigx --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: overtrue <1472352+overtrue@users.noreply.github.com> Co-authored-by: overtrue <anzhengchao@gmail.com> * Improve lock (#596) * improve lock Signed-off-by: Mu junxiang <1948535941@qq.com> * feat(tests): add wait_for_object_absence helper and improve lifecycle test reliability Signed-off-by: Mu junxiang <1948535941@qq.com> * chore: remove dirty docs Signed-off-by: Mu junxiang <1948535941@qq.com> --------- Signed-off-by: Mu junxiang <1948535941@qq.com> * feat(append): implement object append operations with state tracking (#599) * feat(append): implement object append operations with state tracking Signed-off-by: junxiang Mu <1948535941@qq.com> * chore: rebase Signed-off-by: junxiang Mu <1948535941@qq.com> --------- Signed-off-by: junxiang Mu <1948535941@qq.com> * build(deps): upgrade s3s (#595) Co-authored-by: loverustfs <155562731+loverustfs@users.noreply.github.com> * fix: validate mqtt broker * improve code for `import` * upgrade otel relation crates version * fix：dep("jsonwebtoken") feature = 'rust_crypto' * fix * fix * fix * upgrade version * improve code for ecfs * chore: improve event and docker-compose ,Improve the permissions of the `endpoint` health interface * fix * fix * fix * fix * improve code * fix --------- Signed-off-by: Mu junxiang <1948535941@qq.com> Signed-off-by: junxiang Mu <1948535941@qq.com> Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com> Co-authored-by: overtrue <1472352+overtrue@users.noreply.github.com> Co-authored-by: overtrue <anzhengchao@gmail.com> Co-authored-by: guojidan <63799833+guojidan@users.noreply.github.com> Co-authored-by: Nugine <nugine@foxmail.com> Co-authored-by: loverustfs <155562731+loverustfs@users.noreply.github.com>
This commit is contained in:
houseme
@@ -0,0 +1,70 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
/// Check if MQTT Broker is available
|
||||
/// # Arguments
|
||||
/// * `broker_url` - URL of MQTT Broker, for example `mqtt://localhost:1883`
|
||||
/// * `topic` - Topic for testing connections
|
||||
/// # Returns
|
||||
/// * `Ok(())` - If the connection is successful
|
||||
/// * `Err(String)` - If the connection fails, contains an error message
|
||||
///
|
||||
/// # Example
|
||||
/// ```rust,no_run
|
||||
/// #[tokio::main]
|
||||
/// async fn main() {
|
||||
/// let result = rustfs_targets::check_mqtt_broker_available("mqtt://localhost:1883", "test/topic").await;
|
||||
/// if result.is_ok() {
|
||||
/// println!("MQTT Broker is available");
|
||||
/// } else {
|
||||
/// println!("MQTT Broker is not available: {}", result.err().unwrap());
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
/// # Note
|
||||
/// Need to add `rumqttc` and `url` dependencies in `Cargo.toml`
|
||||
/// ```toml
|
||||
/// [dependencies]
|
||||
/// rumqttc = "0.25.0"
|
||||
/// url = "2.5.7"
|
||||
/// tokio = { version = "1", features = ["full"] }
|
||||
/// ```
|
||||
pub async fn check_mqtt_broker_available(broker_url: &str, topic: &str) -> Result<(), String> {
|
||||
use rumqttc::{AsyncClient, MqttOptions, QoS};
|
||||
let url = rustfs_utils::parse_url(broker_url).map_err(|e| format!("Broker URL parsing failed:{e}"))?;
|
||||
let url = url.url();
|
||||
|
||||
match url.scheme() {
|
||||
"tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" | "tls" | "tcps" => {}
|
||||
_ => return Err("unsupported broker url scheme".to_string()),
|
||||
}
|
||||
|
||||
let host = url.host_str().ok_or("Broker is missing host")?;
|
||||
let port = url.port().unwrap_or(1883);
|
||||
let mut mqtt_options = MqttOptions::new("rustfs_check", host, port);
|
||||
mqtt_options.set_keep_alive(std::time::Duration::from_secs(5));
|
||||
let (client, mut eventloop) = AsyncClient::new(mqtt_options, 1);
|
||||
|
||||
// Try to connect and subscribe
|
||||
client
|
||||
.subscribe(topic, QoS::AtLeastOnce)
|
||||
.await
|
||||
.map_err(|e| format!("MQTT subscription failed:{e}"))?;
|
||||
// Wait for eventloop to receive at least one event
|
||||
match tokio::time::timeout(std::time::Duration::from_secs(3), eventloop.poll()).await {
|
||||
Ok(Ok(_)) => Ok(()),
|
||||
Ok(Err(e)) => Err(format!("MQTT connection failed:{e}")),
|
||||
Err(_) => Err("MQTT connection timeout".to_string()),
|
||||
}
|
||||
}
|
||||
@@ -13,11 +13,13 @@
|
||||
// limitations under the License.
|
||||
|
||||
pub mod arn;
|
||||
mod check;
|
||||
pub mod error;
|
||||
mod event_name;
|
||||
pub mod store;
|
||||
pub mod target;
|
||||
|
||||
pub use check::check_mqtt_broker_available;
|
||||
pub use error::{StoreError, TargetError};
|
||||
pub use event_name::EventName;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
@@ -324,7 +324,7 @@ async fn run_mqtt_event_loop(
|
||||
Ok(Err(e)) => Err(e),
|
||||
Err(_) => {
|
||||
debug!(target_id = %target_id, "MQTT poll timed out (EVENT_LOOP_POLL_TIMEOUT) while not connected or status pending.");
|
||||
Err(rumqttc::ConnectionError::NetworkTimeout)
|
||||
Err(ConnectionError::NetworkTimeout)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
@@ -376,7 +376,7 @@ async fn run_mqtt_event_loop(
|
||||
connected_status.store(false, Ordering::SeqCst);
|
||||
error!(target_id = %target_id, error = %e, "Error from MQTT event loop poll");
|
||||
|
||||
if matches!(e, rumqttc::ConnectionError::NetworkTimeout) && (!initial_connection_established || !connected_status.load(Ordering::SeqCst)) {
|
||||
if matches!(e, ConnectionError::NetworkTimeout) && (!initial_connection_established || !connected_status.load(Ordering::SeqCst)) {
|
||||
warn!(target_id = %target_id, "Timeout during initial poll or pending state, will retry.");
|
||||
continue;
|
||||
}
|
||||
@@ -395,8 +395,8 @@ async fn run_mqtt_event_loop(
|
||||
error!(target_id = %target_id, error = %e, "Fatal MQTT error, terminating event loop.");
|
||||
break;
|
||||
}
|
||||
// rumqttc's eventloop.poll() may return Err and terminate after some errors,
|
||||
// Or it will handle reconnection internally. The continue here will make select! wait again.
|
||||
// rumqttc's eventloop.poll() may return Err and terminate after some errors,
|
||||
// Or it will handle reconnection internally. To continue here will make select! wait again.
|
||||
// If the error is temporary and rumqttc is handling reconnection, poll() should eventually succeed or return a different error again.
|
||||
// Sleep briefly to avoid busy cycles in case of rapid failure.
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
|
||||
Reference in New Issue
Block a user