gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-13 16:46:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(tls): centralize runtime foundation (#3065)

Browse Source 
* refactor(targets): move notify net helpers from utils

* refactor(tls): centralize runtime foundation

* refactor(targets): move notify net helpers from utils

* refactor(tls): centralize runtime foundation

* feat(tls-runtime): add TLS debug state and admin handler

* refactor(tls-runtime): unify TLS debug consumer status view

* fix(tls): address PR3065 review feedback

* refactor(tls): align debug status payload types

* refactor(targets): harden TLS hot reload paths

* fix(targets): resolve review-4348251652 findings

* fix(targets): finalize tls runtime review follow-ups

* fix(targets): harden tls reload and review follow-ups

* fix(targets): align tls reload handling across targets

* fix(targets): finalize tls reload state and metrics updates

* chore(deps): trim unused TLS deps

* style(targets): normalize TLS reload formatting

* refactor(targets): introduce tls runtime adapter path

* chore: update workspace manifests for tls refactor

* fix(tls): stabilize material reload and audit workflow

* fix(targets): refresh tls fingerprint flow across sinks

* fix(tls): align runtime coordinator and http reader updates

* fix(sftp): simplify protocol error mapping

* fix(tls): harmonize material loading behavior

* fix(server): finalize tls material wiring in startup flow

* fix(protos): tighten tls generation cache and deps
This commit is contained in:
houseme
2026-05-24 14:41:15 +08:00
committed by GitHub
parent 8be787387c
commit d74e6eb042
67 changed files with 4978 additions and 1725 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/targets/src/target/amqp.rs
+104 -8
View File
@@ -22,11 +22,15 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsGeneration, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode,
validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
TargetType, build_queued_payload_with_records, is_connectivity_error, open_target_queue_store,
persist_queued_payload_to_store,
TargetTlsState, TargetType, build_queued_payload_with_records, build_target_tls_fingerprint, is_connectivity_error,
open_target_queue_store, persist_queued_payload_to_store,
},
};
use async_trait::async_trait;
@@ -37,6 +41,7 @@ use lapin::{
};
use parking_lot::Mutex;
use rustfs_config::{AMQP_TLS_CA, AMQP_TLS_CLIENT_CERT, AMQP_TLS_CLIENT_KEY};
use rustfs_tls_runtime::load_cert_bundle_der_bytes;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::fmt;
@@ -196,16 +201,24 @@ async fn build_tls_config(args: &AMQPArgs) -> Result<OwnedTLSConfig, TargetError
let cert_chain = if args.tls_ca.is_empty() {
None
} else {
Some(
tokio::fs::read_to_string(&args.tls_ca)
.await
.map_err(|e| TargetError::Configuration(format!("Failed to read {AMQP_TLS_CA}: {e}")))?,
)
let certs_der = load_cert_bundle_der_bytes(&args.tls_ca)
.map_err(|e| TargetError::Configuration(format!("Failed to parse {AMQP_TLS_CA}: {e}")))?;
if certs_der.is_empty() {
return Err(TargetError::Configuration(format!(
"{AMQP_TLS_CA} did not contain any parsable certificates"
)));
}
let pem = tokio::fs::read_to_string(&args.tls_ca)
.await
.map_err(|e| TargetError::Configuration(format!("Failed to read {AMQP_TLS_CA}: {e}")))?;
Some(pem)
};
let identity = if args.tls_client_cert.is_empty() {
None
} else {
let _ = load_cert_bundle_der_bytes(&args.tls_client_cert)
.map_err(|e| TargetError::Configuration(format!("Failed to parse {AMQP_TLS_CLIENT_CERT}: {e}")))?;
let pem = tokio::fs::read(&args.tls_client_cert)
.await
.map_err(|e| TargetError::Configuration(format!("Failed to read {AMQP_TLS_CLIENT_CERT}: {e}")))?;
@@ -290,6 +303,9 @@ where
id: TargetID,
args: AMQPArgs,
connection: Arc<Mutex<Option<Arc<AMQPConnection>>>>,
tls_state: Arc<Mutex<TargetTlsState>>,
/// When set, the coordinator drives TLS reload; inline fingerprint check is skipped.
tls_adapter: Option<TlsReloadAdapter<AMQPConnection>>,
connect_lock: Arc<AsyncMutex<()>>,
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
delivery_counters: Arc<TargetDeliveryCounters>,
@@ -305,6 +321,8 @@ where
id: self.id.clone(),
args: self.args.clone(),
connection: Arc::clone(&self.connection),
tls_state: Arc::clone(&self.tls_state),
tls_adapter: self.tls_adapter.clone(),
connect_lock: Arc::clone(&self.connect_lock),
store: self.store.as_ref().map(|s| s.boxed_clone()),
delivery_counters: Arc::clone(&self.delivery_counters),
@@ -329,6 +347,8 @@ where
id: target_id,
args,
connection: Arc::new(Mutex::new(None)),
tls_state: Arc::new(Mutex::new(TargetTlsState::default())),
tls_adapter: None,
connect_lock: Arc::new(AsyncMutex::new(())),
store: queue_store,
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
@@ -341,6 +361,27 @@ where
}
async fn get_or_connect(&self) -> Result<Arc<AMQPConnection>, TargetError> {
// When a TLS reload adapter is attached, it drives connection rebuilds
// in the background. The inline per-send fingerprint check is skipped.
if let Some(adapter) = &self.tls_adapter {
let material = adapter.current_material();
if material.connection.status().connected() && material.channel.status().connected() {
return Ok(material);
}
self.clear_connection_handle();
} else {
let next_fingerprint =
build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key).await?;
let tls_changed = {
let tls_state_guard = self.tls_state.lock();
tls_state_guard.needs_update(&next_fingerprint)
};
if tls_changed {
self.clear_connection_handle();
self.tls_state.lock().refresh(next_fingerprint);
}
}
if let Some(connection) = self.connection.lock().clone()
&& connection.connection.status().connected()
&& connection.channel.status().connected()
@@ -362,10 +403,19 @@ where
Ok(connection)
}
fn clear_connection(&self) {
fn clear_connection_handle(&self) {
*self.connection.lock() = None;
}
fn clear_connection_cache(&self) {
self.clear_connection_handle();
self.tls_state.lock().reset();
}
fn clear_connection(&self) {
self.clear_connection_cache();
}
async fn send_body(&self, body: &[u8]) -> Result<(), TargetError> {
let connection = self.get_or_connect().await?;
let publish = connection
@@ -407,6 +457,46 @@ where
}
}
/// Coordinated TLS hot-reload implementation for AMQP targets.
///
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the connection without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for AMQPTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = AMQPConnection;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.tls_ca.clone(),
client_cert_path: self.args.tls_client_cert.clone(),
client_key_path: self.args.tls_client_key.clone(),
target_label: format!("amqp:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
connect_amqp(&self.args).await
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
let mut guard = self.connection.lock();
*guard = Some(material);
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
validate_tls_material(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
}
}
#[async_trait]
impl<E> Target<E> for AMQPTarget<E>
where
@@ -458,6 +548,12 @@ where
.await
.map_err(|e| map_lapin_error(e, "Failed to close AMQP connection"))?;
}
self.tls_state.lock().reset();
// If a TLS reload adapter is attached, reset its error tracking
// so that a future re-init does not inherit stale failure state.
if let Some(adapter) = &self.tls_adapter {
*adapter.runtime_state().last_error.write() = None;
}
info!(target_id = %self.id, "AMQP target closed");
Ok(())
}
crates/targets/src/target/kafka.rs
+108 -2
View File
@@ -16,16 +16,21 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode, fingerprint::TargetTlsGeneration,
validate::validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
TargetType, build_queued_payload, invalidate_cache_on_connectivity_error, open_target_queue_store,
persist_queued_payload_to_store,
TargetTlsState, TargetType, build_queued_payload, build_target_tls_fingerprint, invalidate_cache_on_connectivity_error,
open_target_queue_store, persist_queued_payload_to_store,
},
};
use async_trait::async_trait;
use rustfs_kafka_async::error::{ConnectionError, Error as KafkaError};
use rustfs_kafka_async::{AsyncProducer, AsyncProducerConfig, Record, RequiredAcks, SecurityConfig};
use rustfs_tls_runtime::{load_cert_bundle_der_bytes, load_private_key};
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::{marker::PhantomData, sync::Arc, time::Duration};
@@ -104,6 +109,11 @@ where
args: KafkaArgs,
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
producer: Arc<Mutex<Option<Arc<AsyncProducer>>>>,
tls_state: Arc<Mutex<TargetTlsState>>,
/// Adapter that bridges this target to the TLS reload coordinator.
/// When `Some`, the target uses coordinator-managed material; when `None`,
/// it falls back to inline fingerprint-based change detection.
tls_adapter: Option<TlsReloadAdapter<Arc<AsyncProducer>>>,
delivery_counters: Arc<TargetDeliveryCounters>,
_phantom: PhantomData<E>,
}
@@ -144,6 +154,8 @@ where
args,
store: queue_store,
producer: Arc::new(Mutex::new(None)),
tls_state: Arc::new(Mutex::new(TargetTlsState::default())),
tls_adapter: None,
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
_phantom: PhantomData,
})
@@ -164,9 +176,27 @@ where
if self.args.tls_enable {
let mut security = SecurityConfig::new();
if !self.args.tls_ca.is_empty() {
let certs = load_cert_bundle_der_bytes(&self.args.tls_ca)
.map_err(|e| Self::map_kafka_error(KafkaError::Config(e.to_string()), "Failed to parse Kafka tls_ca"))?;
if certs.is_empty() {
return Err(TargetError::Configuration(
"Kafka tls_ca did not contain any parsable certificates".to_string(),
));
}
security = security.with_ca_cert(self.args.tls_ca.clone());
}
if !self.args.tls_client_cert.is_empty() && !self.args.tls_client_key.is_empty() {
let certs = load_cert_bundle_der_bytes(&self.args.tls_client_cert).map_err(|e| {
Self::map_kafka_error(KafkaError::Config(e.to_string()), "Failed to parse Kafka tls_client_cert")
})?;
if certs.is_empty() {
return Err(TargetError::Configuration(
"Kafka tls_client_cert did not contain any parsable certificates".to_string(),
));
}
let _ = load_private_key(&self.args.tls_client_key).map_err(|e| {
Self::map_kafka_error(KafkaError::Config(e.to_string()), "Failed to parse Kafka tls_client_key")
})?;
security = security.with_client_cert(self.args.tls_client_cert.clone(), self.args.tls_client_key.clone());
}
config = config.with_security(security);
@@ -178,6 +208,31 @@ where
}
async fn get_or_build_producer(&self) -> Result<Arc<AsyncProducer>, TargetError> {
// Adapter-managed path: use the material directly from the TLS reload adapter.
if let Some(adapter) = &self.tls_adapter {
let producer: Arc<AsyncProducer> = (*adapter.current_material()).clone();
// Ensure the producer is also stored locally so that close() can drain it.
{
let mut guard = self.producer.lock().await;
*guard = Some(Arc::clone(&producer));
}
return Ok(producer);
}
// Inline fingerprint fallback path (no coordinator).
let next_fingerprint =
build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key).await?;
let tls_changed = {
let tls_state_guard = self.tls_state.lock().await;
tls_state_guard.needs_update(&next_fingerprint)
};
if tls_changed {
let mut cached = self.producer.lock().await;
*cached = None;
self.tls_state.lock().await.refresh(next_fingerprint);
}
let mut cached = self.producer.lock().await;
if let Some(producer) = cached.as_ref() {
return Ok(Arc::clone(producer));
@@ -191,6 +246,7 @@ where
async fn invalidate_cached_producer(&self) {
let mut cached = self.producer.lock().await;
*cached = None;
self.tls_state.lock().await.reset();
}
/// Serializes the event and builds a QueuedPayload
@@ -230,6 +286,8 @@ where
args: self.args.clone(),
store: self.store.as_ref().map(|s| s.boxed_clone()),
producer: Arc::clone(&self.producer),
tls_state: Arc::clone(&self.tls_state),
tls_adapter: self.tls_adapter.clone(),
delivery_counters: Arc::clone(&self.delivery_counters),
_phantom: PhantomData,
})
@@ -292,6 +350,13 @@ where
}
async fn close(&self) -> Result<(), TargetError> {
{
let mut guard = self.producer.lock().await;
*guard = None;
}
self.tls_state.lock().await.reset();
info!("Kafka target closed: {}", self.id);
Ok(())
}
@@ -318,6 +383,47 @@ where
}
}
/// Coordinated TLS hot-reload implementation for Kafka targets.
///
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the producer without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for KafkaTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = Arc<AsyncProducer>;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.tls_ca.clone(),
client_cert_path: self.args.tls_client_cert.clone(),
client_key_path: self.args.tls_client_key.clone(),
target_label: format!("kafka:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
let producer = self.build_producer().await?;
Ok(Arc::new(producer))
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
let mut guard = self.producer.lock().await;
*guard = Some((*material).clone());
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
validate_tls_material(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
}
}
#[cfg(test)]
mod tests {
use super::*;
crates/targets/src/target/mod.rs
+49
View File
@@ -37,6 +37,13 @@ pub mod pulsar;
pub mod redis;
pub mod webhook;
#[cfg(test)]
pub(crate) use crate::runtime::tls::fingerprint::TargetTlsFingerprint as TargetTlsFingerprintState;
#[cfg(test)]
pub(crate) use crate::runtime::tls::fingerprint::TargetTlsGeneration;
pub(crate) use crate::runtime::tls::fingerprint::TargetTlsState;
pub(crate) use crate::runtime::tls::fingerprint::build_target_tls_fingerprint;
/// A read-only snapshot of delivery counters for a target.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct TargetDeliverySnapshot {
@@ -531,6 +538,48 @@ pub(crate) fn ensure_rustls_provider_installed() {
}
}
#[cfg(test)]
mod tls_state_tests {
use super::{TargetTlsFingerprintState, TargetTlsGeneration, TargetTlsState};
#[test]
fn refresh_increments_generation_only_when_fingerprint_changes() {
let mut state = TargetTlsState::default();
let first = TargetTlsFingerprintState {
ca_sha256: Some([1; 32]),
client_cert_sha256: None,
client_key_sha256: None,
};
let second = TargetTlsFingerprintState {
ca_sha256: Some([2; 32]),
client_cert_sha256: None,
client_key_sha256: None,
};
assert!(state.refresh(first.clone()));
assert_eq!(state.generation, TargetTlsGeneration(1));
assert!(!state.refresh(first));
assert_eq!(state.generation, TargetTlsGeneration(1));
assert!(state.refresh(second));
assert_eq!(state.generation, TargetTlsGeneration(2));
}
#[test]
fn reset_clears_generation_and_fingerprint() {
let mut state = TargetTlsState {
generation: TargetTlsGeneration(5),
fingerprint: Some(TargetTlsFingerprintState {
ca_sha256: Some([9; 32]),
client_cert_sha256: None,
client_key_sha256: None,
}),
};
state.reset();
assert_eq!(state, TargetTlsState::default());
}
}
#[cfg(test)]
mod tests {
use super::*;
crates/targets/src/target/mqtt.rs
+96 -15
View File
@@ -16,6 +16,10 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsGeneration, TargetTlsInputSet, TargetTlsState, TlsReloadAdapter, config::ReloadApplyMode,
validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
@@ -23,6 +27,7 @@ use crate::{
persist_queued_payload_to_store,
},
};
use arc_swap::ArcSwap;
use async_trait::async_trait;
use hyper_rustls::ConfigBuilderExt;
use rumqttc::{
@@ -32,6 +37,7 @@ use rumqttc::{
use rustfs_config::{
EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST,
};
use rustfs_tls_runtime::{load_certs, load_private_key};
use rustls::ClientConfig;
use serde::Serialize;
use serde::de::DeserializeOwned;
@@ -185,8 +191,7 @@ fn validate_path_is_absolute(path: &str, field: &str) -> Result<(), TargetError>
}
fn build_root_store(ca_path: &str, trust_leaf_as_ca: bool) -> Result<rustls::RootCertStore, TargetError> {
let certs =
rustfs_utils::load_certs(ca_path).map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_ca: {e}")))?;
let certs = load_certs(ca_path).map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_ca: {e}")))?;
let mut store = rustls::RootCertStore::empty();
if trust_leaf_as_ca {
@@ -222,9 +227,9 @@ fn build_mqtt_tls_transport(broker: &Url, tls: &MQTTTlsConfig) -> Result<Transpo
if tls.client_cert_path.is_empty() {
builder.with_no_client_auth()
} else {
let certs = rustfs_utils::load_certs(&tls.client_cert_path)
let certs = load_certs(&tls.client_cert_path)
.map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_cert: {e}")))?;
let key = rustfs_utils::load_private_key(&tls.client_key_path)
let key = load_private_key(&tls.client_key_path)
.map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_key: {e}")))?;
builder
.with_client_auth_cert(certs, key)
@@ -237,9 +242,9 @@ fn build_mqtt_tls_transport(broker: &Url, tls: &MQTTTlsConfig) -> Result<Transpo
if tls.client_cert_path.is_empty() {
builder.with_no_client_auth()
} else {
let certs = rustfs_utils::load_certs(&tls.client_cert_path)
let certs = load_certs(&tls.client_cert_path)
.map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_cert: {e}")))?;
let key = rustfs_utils::load_private_key(&tls.client_key_path)
let key = load_private_key(&tls.client_key_path)
.map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_key: {e}")))?;
builder
.with_client_auth_cert(certs, key)
@@ -490,6 +495,12 @@ where
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
connected: Arc<AtomicBool>,
bg_task_manager: Arc<BgTaskManager>,
/// TLS fingerprint tracking for inline fallback path.
tls_state: Arc<parking_lot::Mutex<TargetTlsState>>,
/// When set, the coordinator drives TLS reload; inline fingerprint check is skipped.
tls_adapter: Option<TlsReloadAdapter<MqttOptions>>,
/// Updated MqttOptions from coordinator for use on next reconnection.
pending_mqtt_options: Arc<ArcSwap<MqttOptions>>,
delivery_counters: Arc<TargetDeliveryCounters>,
_phantom: PhantomData<E>,
}
@@ -519,6 +530,17 @@ where
initial_cancel_rx: Mutex::new(Some(cancel_rx)),
});
// Build the initial MqttOptions for TLS reload support.
let initial_mqtt_options = build_mqtt_options(
format!("rustfs_notify_{}", uuid::Uuid::new_v4()),
&args.broker,
Some(args.username.as_str()),
Some(args.password.as_str()),
&args.tls,
args.keep_alive,
Some(MAX_MQTT_PACKET_SIZE_BYTES),
)?;
info!(target_id = %target_id, "MQTT target created");
Ok(MQTTTarget::<E> {
id: target_id,
@@ -527,6 +549,9 @@ where
store: queue_store,
connected: Arc::new(AtomicBool::new(false)),
bg_task_manager,
tls_state: Arc::new(parking_lot::Mutex::new(TargetTlsState::default())),
tls_adapter: None,
pending_mqtt_options: Arc::new(ArcSwap::from(Arc::new(initial_mqtt_options))),
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
_phantom: PhantomData,
})
@@ -544,20 +569,15 @@ where
let connected_arc = Arc::clone(&self.connected);
let target_id_clone = self.id.clone();
let args_clone = self.args.clone();
let pending_mqtt_options = Arc::clone(&self.pending_mqtt_options);
let _ = bg_task_manager
.init_cell
.get_or_try_init(|| async {
debug!(target_id = %target_id_clone, "Initializing MQTT background task.");
let mqtt_options = build_mqtt_options(
format!("rustfs_notify_{}", uuid::Uuid::new_v4()),
&args_clone.broker,
Some(args_clone.username.as_str()),
Some(args_clone.password.as_str()),
&args_clone.tls,
args_clone.keep_alive,
Some(MAX_MQTT_PACKET_SIZE_BYTES),
)?;
// Use the latest MqttOptions (may have been updated by TLS reload coordinator).
let mqtt_options: MqttOptions = (**pending_mqtt_options.load()).clone();
let (new_client, eventloop) = AsyncClient::builder(mqtt_options).capacity(10).build();
@@ -662,12 +682,66 @@ where
store: self.store.as_ref().map(|s| s.boxed_clone()),
connected: self.connected.clone(),
bg_task_manager: self.bg_task_manager.clone(),
tls_state: Arc::clone(&self.tls_state),
tls_adapter: self.tls_adapter.clone(),
pending_mqtt_options: Arc::clone(&self.pending_mqtt_options),
delivery_counters: self.delivery_counters.clone(),
_phantom: PhantomData,
})
}
}
/// Coordinated TLS hot-reload implementation for MQTT targets.
///
/// MQTT uses `MqttOptions` as the material type. The coordinator rebuilds
/// `MqttOptions` on TLS file changes, and `apply_tls_material` stores it in
/// an `ArcSwap` for use on the next reconnection. The running event loop is
/// not interrupted; rumqttc handles reconnection internally.
#[async_trait]
impl<E> ReloadableTargetTls for MQTTTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = MqttOptions;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.tls.ca_path.clone(),
client_cert_path: self.args.tls.client_cert_path.clone(),
client_key_path: self.args.tls.client_key_path.clone(),
target_label: format!("mqtt:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
build_mqtt_options(
format!("rustfs_notify_{}", uuid::Uuid::new_v4()),
&self.args.broker,
Some(self.args.username.as_str()),
Some(self.args.password.as_str()),
&self.args.tls,
self.args.keep_alive,
Some(MAX_MQTT_PACKET_SIZE_BYTES),
)
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
// Store the new MqttOptions for use on next reconnection.
// The running event loop is not interrupted; rumqttc handles reconnection.
self.pending_mqtt_options.store(material);
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
validate_tls_material(&self.args.tls.ca_path, &self.args.tls.client_cert_path, &self.args.tls.client_key_path)
}
}
async fn run_mqtt_event_loop(
mut eventloop: EventLoop,
connected_status: Arc<AtomicBool>,
@@ -968,6 +1042,13 @@ where
}
}
self.tls_state.lock().reset();
// If a TLS reload adapter is attached, reset its error tracking
// so that a future re-init does not inherit stale failure state.
if let Some(adapter) = &self.tls_adapter {
*adapter.runtime_state().last_error.write() = None;
}
self.connected.store(false, Ordering::SeqCst);
info!(target_id = %self.id, "MQTT target close method finished.");
Ok(())
crates/targets/src/target/mysql.rs
+162 -62
View File
@@ -16,6 +16,10 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode, fingerprint::TargetTlsGeneration,
validate::validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
@@ -26,6 +30,7 @@ use crate::{
use async_trait::async_trait;
use mysql_async::{Conn, Opts, OptsBuilder, Pool, PoolConstraints, PoolOpts, SslOpts, prelude::Queryable};
use rustfs_config::{MYSQL_TLS_CA, MYSQL_TLS_CLIENT_CERT, MYSQL_TLS_CLIENT_KEY};
use rustfs_tls_runtime::{load_certs, load_private_key};
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::marker::PhantomData;
@@ -478,6 +483,11 @@ where
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
/// Lazily-initialized MySQL connection pool
pool: Arc<Mutex<Option<Pool>>>,
/// TLS fingerprint tracking for hot reload (inline fallback path)
tls_state: Arc<parking_lot::Mutex<super::TargetTlsState>>,
/// When present, the adapter provides coordinator-managed TLS material;
/// otherwise the inline fingerprint path is used as a fallback.
tls_adapter: Option<TlsReloadAdapter<Pool>>,
/// Success/failure counters exposed via `delivery_snapshot`
delivery_counters: Arc<TargetDeliveryCounters>,
/// Zero-sized marker for the event type `E`
@@ -489,6 +499,9 @@ where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
/// Creates a new MySqlTarget.
///
/// The target starts without a TLS reload coordinator. Use
/// `TlsReloadAdapter::try_register` to opt into coordinated TLS hot-reload.
pub fn new(id: String, args: MySqlArgs) -> Result<Self, TargetError> {
args.validate()?;
@@ -511,6 +524,8 @@ where
store: queue_store,
// Pool is lazily initialized on first use to avoid unnecessary connections at startup and allow for better error handling
pool: Arc::new(Mutex::new(None)),
tls_state: Arc::new(parking_lot::Mutex::new(super::TargetTlsState::default())),
tls_adapter: None,
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
_phantom: PhantomData,
})
@@ -518,6 +533,10 @@ where
/// Returns or lazily initializes the MySQL connection pool.
///
/// When `tls_adapter` is present (coordinator-managed), the pool
/// is sourced from the coordinator's published material.
/// Otherwise, the inline fingerprint-based path is used as a fallback.
///
/// # Errors
///
/// | Scenario | Error variant |
@@ -528,6 +547,31 @@ where
/// | Existing table has incompatible schema | `Initialization` |
/// | DSN parse failure / invalid config | `Configuration` |
async fn get_or_init_pool(&self) -> Result<Pool, TargetError> {
// Adapter-managed path: use the material directly from the coordinator.
if let Some(adapter) = &self.tls_adapter {
let pool: Pool = (*adapter.current_material()).clone();
// Ensure the pool is also stored locally so that close() can drain it.
{
let mut guard = self.pool.lock().await;
*guard = Some(pool.clone());
}
return Ok(pool);
}
// Inline fingerprint fallback path (no coordinator).
let next_fingerprint =
super::build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key).await?;
let tls_changed = {
let tls_state_guard = self.tls_state.lock();
tls_state_guard.needs_update(&next_fingerprint)
};
if tls_changed {
let mut guard = self.pool.lock().await;
*guard = None;
self.tls_state.lock().refresh(next_fingerprint);
}
{
let guard = self.pool.lock().await;
if let Some(pool) = guard.as_ref() {
@@ -535,68 +579,7 @@ where
}
}
let dsn = MySqlDsn::parse(&self.args.dsn_string)?;
let mut builder = OptsBuilder::default()
.user(Some(dsn.user.clone()))
.pass(Some(dsn.password.clone()))
.ip_or_hostname(dsn.host.clone())
.tcp_port(dsn.port)
.db_name(Some(dsn.database.clone()));
if dsn.tls {
super::ensure_rustls_provider_installed();
let mut ssl_opts = SslOpts::default();
if !self.args.tls_ca.is_empty() {
ssl_opts = ssl_opts.with_root_certs(vec![PathBuf::from(self.args.tls_ca.clone()).into()]);
}
if !self.args.tls_client_cert.is_empty() && !self.args.tls_client_key.is_empty() {
let identity = mysql_async::ClientIdentity::new(
PathBuf::from(self.args.tls_client_cert.clone()).into(),
PathBuf::from(self.args.tls_client_key.clone()).into(),
);
ssl_opts = ssl_opts.with_client_identity(Some(identity));
}
builder = builder.ssl_opts(Some(ssl_opts));
} else {
warn!(
"MySQL target '{}' is configured without TLS. This is insecure and should not be used in production.",
self.id
);
}
// When max_open_connections is 0, no explicit upper bound is set —
// mysql_async uses its default pool constraints (10–100).
if self.args.max_open_connections > 0 {
let constraints = PoolConstraints::new(1, self.args.max_open_connections).ok_or_else(|| {
TargetError::Configuration(format!(
"MySQL max_open_connections must be >= 1, got {}",
self.args.max_open_connections
))
})?;
builder = builder.pool_opts(PoolOpts::default().with_constraints(constraints));
}
let opts = Opts::from(builder);
let pool = Pool::new(opts);
// Uses a double-check pattern: the mutex guard is only held for
// short reads/writes to the pool cache. All I/O (connecting,
// DDL, schema validation) happens outside the lock so that
// concurrent callers are not blocked by a slow MySQL server.
let mut conn = pool.get_conn().await.map_err(|_| TargetError::NotConnected)?;
conn.query_drop("SELECT 1").await.map_err(|_| TargetError::NotConnected)?;
let ddl = format!(
"CREATE TABLE IF NOT EXISTS {} (event_time DATETIME(6) NOT NULL, event_data JSON NOT NULL)",
quote_table_name(&self.args.table)?
);
conn.query_drop(ddl)
.await
.map_err(|e| TargetError::Initialization(format!("Failed to create MySQL table: {e}")))?;
validate_existing_schema(&mut conn, &self.args.table).await?;
let pool = build_mysql_pool_from_args(&self.args).await?;
// Double-check: another caller may have initialized the pool
// while we were doing I/O.
@@ -653,12 +636,87 @@ where
args: self.args.clone(),
store: self.store.as_ref().map(|s| s.boxed_clone()),
pool: Arc::clone(&self.pool),
tls_state: Arc::clone(&self.tls_state),
tls_adapter: self.tls_adapter.clone(),
delivery_counters: Arc::clone(&self.delivery_counters),
_phantom: PhantomData,
})
}
}
/// Builds a MySQL connection pool from the given args, including TLS setup,
/// DDL table creation, and schema validation.
///
/// This is a standalone function so it can be called both from
/// `get_or_init_pool` (inline fallback) and from `build_tls_material`
/// (coordinator path).
async fn build_mysql_pool_from_args(args: &MySqlArgs) -> Result<Pool, TargetError> {
let dsn = MySqlDsn::parse(&args.dsn_string)?;
let mut builder = OptsBuilder::default()
.user(Some(dsn.user.clone()))
.pass(Some(dsn.password.clone()))
.ip_or_hostname(dsn.host.clone())
.tcp_port(dsn.port)
.db_name(Some(dsn.database.clone()));
if dsn.tls {
super::ensure_rustls_provider_installed();
let mut ssl_opts = SslOpts::default();
if !args.tls_ca.is_empty() {
let _ =
load_certs(&args.tls_ca).map_err(|e| TargetError::Configuration(format!("Failed to load MySQL tls_ca: {e}")))?;
ssl_opts = ssl_opts.with_root_certs(vec![PathBuf::from(args.tls_ca.clone()).into()]);
}
if !args.tls_client_cert.is_empty() && !args.tls_client_key.is_empty() {
let _ = load_certs(&args.tls_client_cert)
.map_err(|e| TargetError::Configuration(format!("Failed to load MySQL tls_client_cert: {e}")))?;
let _ = load_private_key(&args.tls_client_key)
.map_err(|e| TargetError::Configuration(format!("Failed to load MySQL tls_client_key: {e}")))?;
let identity = mysql_async::ClientIdentity::new(
PathBuf::from(args.tls_client_cert.clone()).into(),
PathBuf::from(args.tls_client_key.clone()).into(),
);
ssl_opts = ssl_opts.with_client_identity(Some(identity));
}
builder = builder.ssl_opts(Some(ssl_opts));
} else {
warn!("MySQL target is configured without TLS. This is insecure and should not be used in production.");
}
// When max_open_connections is 0, no explicit upper bound is set —
// mysql_async uses its default pool constraints (10–100).
if args.max_open_connections > 0 {
let constraints = PoolConstraints::new(1, args.max_open_connections).ok_or_else(|| {
TargetError::Configuration(format!("MySQL max_open_connections must be >= 1, got {}", args.max_open_connections))
})?;
builder = builder.pool_opts(PoolOpts::default().with_constraints(constraints));
}
let opts = Opts::from(builder);
let pool = Pool::new(opts);
// Uses a double-check pattern: the mutex guard is only held for
// short reads/writes to the pool cache. All I/O (connecting,
// DDL, schema validation) happens outside the lock so that
// concurrent callers are not blocked by a slow MySQL server.
let mut conn = pool.get_conn().await.map_err(|_| TargetError::NotConnected)?;
conn.query_drop("SELECT 1").await.map_err(|_| TargetError::NotConnected)?;
let ddl = format!(
"CREATE TABLE IF NOT EXISTS {} (event_time DATETIME(6) NOT NULL, event_data JSON NOT NULL)",
quote_table_name(&args.table)?
);
conn.query_drop(ddl)
.await
.map_err(|e| TargetError::Initialization(format!("Failed to create MySQL table: {e}")))?;
validate_existing_schema(&mut conn, &args.table).await?;
Ok(pool)
}
/// Maps a mysql_async error to `TargetError`:
/// - `Io`/`Driver` → `NotConnected` (connection lost, fixed-delay retry)
/// - `Server(1213|1205|1040)` → `Timeout` (deadlock/lock timeout/too
@@ -793,6 +851,8 @@ where
.map_err(|err| TargetError::Network(format!("Failed to disconnect MySQL pool: {err}")))?;
}
// Adapter cleanup is done by the coordinator; no local state to reset.
info!("MySQL target closed: {}", self.id);
Ok(())
}
@@ -828,6 +888,46 @@ where
}
}
/// Coordinated TLS hot-reload implementation for MySQL targets.
///
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the connection pool without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for MySqlTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = Pool;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.tls_ca.clone(),
client_cert_path: self.args.tls_client_cert.clone(),
client_key_path: self.args.tls_client_key.clone(),
target_label: format!("mysql:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
build_mysql_pool_from_args(&self.args).await
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
let mut guard = self.pool.lock().await;
*guard = Some((*material).clone());
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
validate_tls_material(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
}
}
#[cfg(test)]
mod tests {
use super::*;
crates/targets/src/target/nats.rs
+143 -9
View File
@@ -16,10 +16,15 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsGeneration, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode,
validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
TargetType, build_queued_payload_with_records, open_target_queue_store, persist_queued_payload_to_store,
TargetTlsState, TargetType, build_queued_payload_with_records, build_target_tls_fingerprint, open_target_queue_store,
persist_queued_payload_to_store,
},
};
use async_trait::async_trait;
@@ -28,9 +33,10 @@ use serde::Serialize;
use serde::de::DeserializeOwned;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use tracing::{info, instrument};
use tokio::sync::Mutex;
use tracing::{info, instrument, warn};
#[derive(Debug, Clone)]
pub struct NATSArgs {
@@ -168,7 +174,12 @@ where
{
id: TargetID,
args: NATSArgs,
client: Mutex<Option<async_nats::Client>>,
client: Arc<Mutex<Option<async_nats::Client>>>,
tls_state: Arc<parking_lot::Mutex<TargetTlsState>>,
/// Adapter that bridges this target to the TLS reload coordinator.
/// When `Some`, the target uses coordinator-managed material; when `None`,
/// it falls back to inline fingerprint-based change detection.
tls_adapter: Option<TlsReloadAdapter<async_nats::Client>>,
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
connected: AtomicBool,
delivery_counters: Arc<TargetDeliveryCounters>,
@@ -183,7 +194,9 @@ where
Box::new(NATSTarget::<E> {
id: self.id.clone(),
args: self.args.clone(),
client: Mutex::new(self.client.lock().unwrap().clone()),
client: Arc::clone(&self.client),
tls_state: Arc::clone(&self.tls_state),
tls_adapter: self.tls_adapter.clone(),
store: self.store.as_ref().map(|s| s.boxed_clone()),
connected: AtomicBool::new(self.connected.load(Ordering::SeqCst)),
delivery_counters: Arc::clone(&self.delivery_counters),
@@ -207,7 +220,9 @@ where
Ok(Self {
id: target_id,
args,
client: Mutex::new(None),
client: Arc::new(Mutex::new(None)),
tls_state: Arc::new(parking_lot::Mutex::new(TargetTlsState::default())),
tls_adapter: None,
store: queue_store,
connected: AtomicBool::new(false),
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
@@ -215,11 +230,42 @@ where
})
}
async fn invalidate_cached_client_connection(&self) {
*self.client.lock().await = None;
}
async fn get_or_connect(&self) -> Result<async_nats::Client, TargetError> {
if let Some(client) = self.client.lock().unwrap().clone() {
// Adapter-managed path: use the material directly from the TLS reload adapter.
if let Some(adapter) = &self.tls_adapter {
let client: async_nats::Client = (*adapter.current_material()).clone();
// Ensure the client is also stored locally so that close() can drain it.
{
let mut guard = self.client.lock().await;
*guard = Some(client.clone());
}
return Ok(client);
}
// Inline fingerprint fallback path (no coordinator).
let next_fingerprint =
build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key).await?;
let tls_changed = {
let tls_state_guard = self.tls_state.lock();
tls_state_guard.needs_update(&next_fingerprint)
};
if tls_changed {
self.invalidate_cached_client_connection().await;
self.tls_state.lock().refresh(next_fingerprint);
}
{
let guard = self.client.lock().await;
if let Some(client) = guard.as_ref() {
return Ok(client.clone());
}
}
let client = connect_nats(&self.args).await?;
client
.flush()
@@ -227,7 +273,7 @@ where
.map_err(|e| TargetError::Network(format!("Failed to flush NATS connection: {e}")))?;
self.connected.store(true, Ordering::SeqCst);
let mut guard = self.client.lock().unwrap();
let mut guard = self.client.lock().await;
let shared = guard.get_or_insert_with(|| client.clone()).clone();
Ok(shared)
}
@@ -294,7 +340,11 @@ where
}
async fn close(&self) -> Result<(), TargetError> {
let client = self.client.lock().unwrap().take();
let client = {
let mut guard = self.client.lock().await;
guard.take()
};
self.tls_state.lock().reset();
self.connected.store(false, Ordering::SeqCst);
if let Some(client) = client {
client
@@ -335,3 +385,87 @@ where
self.delivery_counters.record_final_failure();
}
}
/// Coordinated TLS hot-reload implementation for NATS targets.
///
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the NATS client without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for NATSTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = async_nats::Client;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.tls_ca.clone(),
client_cert_path: self.args.tls_client_cert.clone(),
client_key_path: self.args.tls_client_key.clone(),
target_label: format!("nats:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
connect_nats(&self.args).await
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
let mut guard = self.client.lock().await;
*guard = Some((*material).clone());
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
validate_tls_material(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn base_args() -> NATSArgs {
NATSArgs {
enable: true,
address: "nats://127.0.0.1:4222".to_string(),
subject: "rustfs.events".to_string(),
username: String::new(),
password: String::new(),
token: String::new(),
credentials_file: String::new(),
tls_ca: String::new(),
tls_client_cert: String::new(),
tls_client_key: String::new(),
tls_required: false,
queue_dir: String::new(),
queue_limit: 0,
target_type: TargetType::NotifyEvent,
}
}
#[test]
fn validate_nats_rejects_multiple_auth_methods() {
let args = NATSArgs {
token: "abc".to_string(),
username: "user".to_string(),
password: "pass".to_string(),
..base_args()
};
assert!(args.validate().is_err());
}
#[test]
fn validate_nats_rejects_relative_queue_dir() {
let args = NATSArgs {
queue_dir: "relative/path".to_string(),
..base_args()
};
assert!(args.validate().is_err());
}
}
crates/targets/src/target/postgres.rs
+86 -27
View File
@@ -29,6 +29,10 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsGeneration, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode,
validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
@@ -38,12 +42,10 @@ use crate::{
use async_trait::async_trait;
use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
use rustfs_config::{POSTGRES_DSN_STRING, POSTGRES_TLS_CA, POSTGRES_TLS_CLIENT_CERT, POSTGRES_TLS_CLIENT_KEY};
use rustls_pki_types::pem::PemObject;
use rustls_pki_types::{CertificateDer, PrivateKeyDer};
use rustfs_tls_runtime::{load_certs, load_private_key};
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::fmt;
use std::io::BufReader;
use std::path::Path;
use std::sync::Arc;
use tokio_postgres::Config;
@@ -425,11 +427,9 @@ pub fn build_tls_config(args: &PostgresArgs) -> Result<rustls::ClientConfig, Tar
let _ = root_store.add(cert);
}
} else {
let pem = std::fs::read(&args.tls_ca)
.map_err(|e| TargetError::Configuration(format!("failed to read {POSTGRES_TLS_CA}: {e}")))?;
let mut reader = BufReader::new(pem.as_slice());
for cert in CertificateDer::pem_reader_iter(&mut reader) {
let cert = cert.map_err(|e| TargetError::Configuration(format!("invalid {POSTGRES_TLS_CA}: {e}")))?;
let certs =
load_certs(&args.tls_ca).map_err(|e| TargetError::Configuration(format!("invalid {POSTGRES_TLS_CA}: {e}")))?;
for cert in certs {
root_store
.add(cert)
.map_err(|e| TargetError::Configuration(format!("failed to add CA cert: {e}")))?;
@@ -439,16 +439,9 @@ pub fn build_tls_config(args: &PostgresArgs) -> Result<rustls::ClientConfig, Tar
let builder = rustls::ClientConfig::builder().with_root_certificates(root_store);
let client_config = if !args.tls_client_cert.is_empty() && !args.tls_client_key.is_empty() {
let cert_pem = std::fs::read(&args.tls_client_cert)
.map_err(|e| TargetError::Configuration(format!("failed to read {POSTGRES_TLS_CLIENT_CERT}: {e}")))?;
let key_pem = std::fs::read(&args.tls_client_key)
.map_err(|e| TargetError::Configuration(format!("failed to read {POSTGRES_TLS_CLIENT_KEY}: {e}")))?;
let certs: Vec<_> = CertificateDer::pem_reader_iter(&mut BufReader::new(cert_pem.as_slice()))
.collect::<Result<_, _>>()
let certs = load_certs(&args.tls_client_cert)
.map_err(|e| TargetError::Configuration(format!("invalid {POSTGRES_TLS_CLIENT_CERT}: {e}")))?;
let key = PrivateKeyDer::from_pem_reader(&mut BufReader::new(key_pem.as_slice()))
let key = load_private_key(&args.tls_client_key)
.map_err(|e| TargetError::Configuration(format!("invalid {POSTGRES_TLS_CLIENT_KEY}: {e}")))?;
builder
@@ -548,13 +541,22 @@ fn resolve_payload_key(payload: &serde_json::Value, meta: &QueuedPayloadMeta) ->
/// so that `clone_box` does not duplicate connection state. The optional
/// `QueueStore` provides at-least-once delivery semantics consistent with the
/// other built-in targets.
///
/// When `tls_adapter` is `Some`, the target participates in the
/// coordinated TLS hot-reload system driven by `TlsReloadAdapter`,
/// and the inline fingerprint check in `send_body` is skipped. When `None`,
/// the legacy inline fingerprint check is used as a fallback.
pub struct PostgresTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
id: TargetID,
args: PostgresArgs,
pool: Pool,
pool: Arc<parking_lot::Mutex<Pool>>,
tls_state: Arc<parking_lot::Mutex<super::TargetTlsState>>,
/// When present, the adapter provides coordinator-managed TLS material;
/// otherwise the inline fingerprint path is used as a fallback.
tls_adapter: Option<TlsReloadAdapter<Pool>>,
namespace_sql: String,
access_sql: String,
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
@@ -570,7 +572,9 @@ where
Box::new(PostgresTarget::<E> {
id: self.id.clone(),
args: self.args.clone(),
pool: self.pool.clone(),
pool: Arc::clone(&self.pool),
tls_state: Arc::clone(&self.tls_state),
tls_adapter: self.tls_adapter.clone(),
namespace_sql: self.namespace_sql.clone(),
access_sql: self.access_sql.clone(),
store: self.store.as_ref().map(|s| s.boxed_clone()),
@@ -599,7 +603,9 @@ where
namespace_sql: namespace_upsert_sql(&args.schema, &args.table),
access_sql: access_insert_sql(&args.schema, &args.table),
args,
pool,
pool: Arc::new(parking_lot::Mutex::new(pool)),
tls_state: Arc::new(parking_lot::Mutex::new(super::TargetTlsState::default())),
tls_adapter: None,
store: queue_store,
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
_phantom: std::marker::PhantomData,
@@ -611,8 +617,25 @@ where
/// Identifier validation has already happened in `PostgresArgs::validate()`,
/// so `qualified_table` cannot produce a malformed SQL string here.
async fn send_body(&self, body: &[u8], event_id: &str, meta: &QueuedPayloadMeta) -> Result<(), TargetError> {
let client = self
.pool
// When a TLS reload adapter is attached, it drives pool rebuilds in
// the background. The inline per-send fingerprint check is skipped.
if self.tls_adapter.is_none() {
let next_fingerprint =
super::build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
.await?;
let tls_changed = {
let tls_state_guard = self.tls_state.lock();
tls_state_guard.fingerprint.as_ref() != Some(&next_fingerprint)
};
if tls_changed {
let new_pool = build_pool(&self.args)?;
*self.pool.lock() = new_pool;
self.tls_state.lock().refresh(next_fingerprint);
}
}
let pool = self.pool.lock().clone();
let client = pool
.get()
.await
.map_err(|e| map_pool_error(e, "PostgreSQL pool checkout failed"))?;
@@ -645,8 +668,8 @@ where
/// Probes the table from `init()`. Failure is non-fatal when a queue is
/// configured: events buffer in the store until the schema is fixed.
async fn probe_table(&self) -> Result<(), TargetError> {
let client = self
.pool
let pool = self.pool.lock().clone();
let client = pool
.get()
.await
.map_err(|e| map_pool_error(e, "PostgreSQL pool checkout failed during init probe"))?;
@@ -659,6 +682,41 @@ where
}
}
#[async_trait]
impl<E> ReloadableTargetTls for PostgresTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = Pool;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.tls_ca.clone(),
client_cert_path: self.args.tls_client_cert.clone(),
client_key_path: self.args.tls_client_key.clone(),
target_label: format!("postgres:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
build_pool(&self.args)
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
*self.pool.lock() = (*material).clone();
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
validate_tls_material(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
}
}
#[async_trait]
impl<E> Target<E> for PostgresTarget<E>
where
@@ -674,8 +732,8 @@ where
}
match tokio::time::timeout(std::time::Duration::from_secs(10), async {
let client = self
.pool
let pool = self.pool.lock().clone();
let client = pool
.get()
.await
.map_err(|e| map_pool_error(e, "PostgreSQL pool checkout failed"))?;
@@ -728,7 +786,8 @@ where
}
async fn close(&self) -> Result<(), TargetError> {
self.pool.close();
self.pool.lock().close();
// Adapter cleanup is done by the coordinator; no local state to reset.
info!(target_id = %self.id, "PostgreSQL target closed");
Ok(())
}
crates/targets/src/target/pulsar.rs
+148 -2
View File
@@ -16,14 +16,20 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsGeneration, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode,
validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
TargetType, build_queued_payload_with_records, open_target_queue_store, persist_queued_payload_to_store,
TargetTlsState, TargetType, build_queued_payload_with_records, build_target_tls_fingerprint, open_target_queue_store,
persist_queued_payload_to_store,
},
};
use async_trait::async_trait;
use pulsar::{Authentication, Producer, Pulsar, TokioExecutor};
use rustfs_tls_runtime::load_cert_bundle_der_bytes;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::path::Path;
@@ -136,6 +142,13 @@ pub async fn connect_pulsar(args: &PulsarArgs) -> Result<Pulsar<TokioExecutor>,
}
if !args.tls_ca.is_empty() {
let certs = load_cert_bundle_der_bytes(&args.tls_ca)
.map_err(|e| TargetError::Configuration(format!("Failed to parse Pulsar tls_ca: {e}")))?;
if certs.is_empty() {
return Err(TargetError::Configuration(
"Pulsar tls_ca did not contain any parsable certificates".to_string(),
));
}
builder = builder
.with_certificate_chain_file(&args.tls_ca)
.map_err(|e| TargetError::Configuration(format!("Failed to load Pulsar tls_ca: {e}")))?;
@@ -158,6 +171,9 @@ where
id: TargetID,
args: PulsarArgs,
client: Mutex<Option<Pulsar<TokioExecutor>>>,
tls_state: Mutex<TargetTlsState>,
/// When set, the coordinator drives TLS reload; inline fingerprint check is skipped.
tls_adapter: Option<TlsReloadAdapter<Pulsar<TokioExecutor>>>,
producer: AsyncMutex<Option<Producer<TokioExecutor>>>,
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
connected: AtomicBool,
@@ -174,6 +190,8 @@ where
id: self.id.clone(),
args: self.args.clone(),
client: Mutex::new(self.client.lock().unwrap().clone()),
tls_state: Mutex::new(self.tls_state.lock().unwrap().clone()),
tls_adapter: self.tls_adapter.clone(),
producer: AsyncMutex::new(None),
store: self.store.as_ref().map(|s| s.boxed_clone()),
connected: AtomicBool::new(self.connected.load(Ordering::SeqCst)),
@@ -199,6 +217,8 @@ where
id: target_id,
args,
client: Mutex::new(None),
tls_state: Mutex::new(TargetTlsState::default()),
tls_adapter: None,
producer: AsyncMutex::new(None),
store: queue_store,
connected: AtomicBool::new(false),
@@ -207,7 +227,36 @@ where
})
}
fn clear_cached_client_connection(&self) {
self.client.lock().unwrap().take();
}
fn clear_cached_client(&self) {
self.clear_cached_client_connection();
self.tls_state.lock().unwrap().reset();
}
async fn get_or_connect_client(&self) -> Result<Pulsar<TokioExecutor>, TargetError> {
// When a TLS reload adapter is attached, it drives client rebuilds
// in the background. The inline per-send fingerprint check is skipped.
if let Some(adapter) = &self.tls_adapter {
let material = adapter.current_material();
{
let mut guard = self.client.lock().unwrap();
*guard = Some((*material).clone());
}
} else {
let next_fingerprint = build_target_tls_fingerprint(&self.args.tls_ca, "", "").await?;
let tls_changed = {
let tls_state_guard = self.tls_state.lock().unwrap();
tls_state_guard.needs_update(&next_fingerprint)
};
if tls_changed {
self.clear_cached_client_connection();
self.tls_state.lock().unwrap().refresh(next_fingerprint);
}
}
if let Some(client) = self.client.lock().unwrap().clone() {
return Ok(client);
}
@@ -262,6 +311,56 @@ where
}
}
/// Coordinated TLS hot-reload implementation for Pulsar targets.
///
/// Pulsar only uses a CA certificate (no client cert/key).
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the client without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for PulsarTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = Pulsar<TokioExecutor>;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.tls_ca.clone(),
client_cert_path: String::new(),
client_key_path: String::new(),
target_label: format!("pulsar:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
connect_pulsar(&self.args).await
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
// Pulsar client is Clone, so we clone from the Arc and store it.
{
let mut guard = self.client.lock().unwrap();
*guard = Some((*material).clone());
}
// Producer is bound to the old client; clear it so next send rebuilds.
{
let mut producer = self.producer.lock().await;
*producer = None;
}
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
// Pulsar only uses CA, no client cert/key.
validate_tls_material(&self.args.tls_ca, "", "")
}
}
#[async_trait]
impl<E> Target<E> for PulsarTarget<E>
where
@@ -321,8 +420,13 @@ where
.map_err(|e| TargetError::Network(format!("Failed to close Pulsar producer: {e}")))?;
}
*producer = None;
self.client.lock().unwrap().take();
self.clear_cached_client();
self.connected.store(false, Ordering::SeqCst);
// If a TLS reload adapter is attached, reset its error tracking
// so that a future re-init does not inherit stale failure state.
if let Some(adapter) = &self.tls_adapter {
*adapter.runtime_state().last_error.write() = None;
}
info!(target_id = %self.id, "Pulsar target closed");
Ok(())
}
@@ -355,3 +459,45 @@ where
self.delivery_counters.record_final_failure();
}
}
#[cfg(test)]
mod tests {
use super::*;
fn base_args() -> PulsarArgs {
PulsarArgs {
enable: true,
broker: "pulsar://127.0.0.1:6650".to_string(),
topic: "persistent://public/default/rustfs-events".to_string(),
auth_token: String::new(),
username: String::new(),
password: String::new(),
tls_ca: String::new(),
tls_allow_insecure: false,
tls_hostname_verification: true,
queue_dir: String::new(),
queue_limit: 0,
target_type: TargetType::NotifyEvent,
}
}
#[test]
fn validate_pulsar_rejects_mixed_auth_methods() {
let args = PulsarArgs {
auth_token: "token".to_string(),
username: "user".to_string(),
password: "pass".to_string(),
..base_args()
};
assert!(args.validate().is_err());
}
#[test]
fn validate_pulsar_rejects_relative_queue_dir() {
let args = PulsarArgs {
queue_dir: "relative/path".to_string(),
..base_args()
};
assert!(args.validate().is_err());
}
}
crates/targets/src/target/redis.rs
+116 -9
View File
@@ -16,6 +16,10 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsGeneration, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode,
validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
@@ -31,9 +35,12 @@ use redis::{
io::tcp::{TcpSettings, socket2},
};
use rustfs_config::{REDIS_TLS_CA, REDIS_TLS_CLIENT_CERT, REDIS_TLS_CLIENT_KEY, REDIS_TLS_POLICY};
use rustls::pki_types::CertificateDer;
use rustls::pki_types::pem::PemObject;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::fmt;
use std::io::BufReader;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
@@ -297,7 +304,8 @@ where
{
id: TargetID,
args: RedisArgs,
publisher_client: Client,
/// Redis client, wrapped in a lock so TLS hot-reload can atomically replace it.
publisher_client: Arc<parking_lot::Mutex<Client>>,
publisher: Arc<Mutex<Option<ConnectionManager>>>,
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
/// Business-level liveness flag.
@@ -306,6 +314,12 @@ where
/// publish exhausted retries, or the target was explicitly closed). Temporary reconnectable
/// errors only invalidate the cached publisher so that a later request can lazily rebuild it.
connected: Arc<AtomicBool>,
/// TLS fingerprint tracking for hot reload (inline fallback path).
tls_state: Arc<parking_lot::Mutex<super::TargetTlsState>>,
/// Adapter that bridges this target to the TLS reload coordinator.
/// When `Some`, the target uses coordinator-managed material; when `None`,
/// it falls back to inline fingerprint-based change detection.
tls_adapter: Option<TlsReloadAdapter<Client>>,
delivery_counters: Arc<TargetDeliveryCounters>,
_phantom: std::marker::PhantomData<E>,
}
@@ -334,10 +348,12 @@ where
Ok(Self {
id: target_id,
args,
publisher_client,
publisher_client: Arc::new(parking_lot::Mutex::new(publisher_client)),
publisher: Arc::new(Mutex::new(None)),
store: queue_store,
connected: Arc::new(AtomicBool::new(false)),
tls_state: Arc::new(parking_lot::Mutex::new(super::TargetTlsState::default())),
tls_adapter: None,
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
_phantom: std::marker::PhantomData,
})
@@ -347,23 +363,61 @@ where
Box::new(Self {
id: self.id.clone(),
args: self.args.clone(),
publisher_client: self.publisher_client.clone(),
publisher_client: Arc::clone(&self.publisher_client),
publisher: Arc::clone(&self.publisher),
store: self.store.as_ref().map(|s| s.boxed_clone()),
connected: Arc::clone(&self.connected),
tls_state: Arc::clone(&self.tls_state),
tls_adapter: self.tls_adapter.clone(),
delivery_counters: Arc::clone(&self.delivery_counters),
_phantom: std::marker::PhantomData,
})
}
async fn get_or_create_publisher(&self) -> Result<ConnectionManager, TargetError> {
// Adapter-managed path: use the material directly from the TLS reload adapter.
if let Some(adapter) = &self.tls_adapter {
let client: Client = (*adapter.current_material()).clone();
// Ensure the client is also stored locally so close() can drain it.
*self.publisher_client.lock() = client.clone();
let manager = client
.get_connection_manager_lazy(build_redis_connection_manager_config(&self.args))
.map_err(map_redis_error)?;
*self.publisher.lock().await = Some(manager.clone());
return Ok(manager);
}
// Inline fingerprint fallback path (no coordinator).
let secure_scheme = matches!(self.args.url.scheme(), "rediss" | "valkeys");
if secure_scheme {
let next_fingerprint = super::build_target_tls_fingerprint(
&self.args.tls.ca_path,
&self.args.tls.client_cert_path,
&self.args.tls.client_key_path,
)
.await?;
let tls_changed = {
let tls_state_guard = self.tls_state.lock();
tls_state_guard.needs_update(&next_fingerprint)
};
if tls_changed {
let new_client = build_redis_client(&self.args)?;
*self.publisher_client.lock() = new_client;
self.invalidate_cached_publisher().await;
self.tls_state.lock().refresh(next_fingerprint);
}
}
let mut guard = self.publisher.lock().await;
if let Some(manager) = guard.clone() {
return Ok(manager);
}
let manager = self
.publisher_client
let client = self.publisher_client.lock().clone();
let manager = client
.get_connection_manager_lazy(build_redis_connection_manager_config(&self.args))
.map_err(map_redis_error)?;
@@ -475,7 +529,8 @@ where
return Ok(false);
}
match tokio::time::timeout(Duration::from_secs(5), ping_redis_server(&self.publisher_client, &self.args)).await {
let client = self.publisher_client.lock().clone();
match tokio::time::timeout(Duration::from_secs(5), ping_redis_server(&client, &self.args)).await {
Ok(Ok(())) => {
self.connected.store(true, Ordering::SeqCst);
Ok(true)
@@ -557,6 +612,7 @@ where
async fn close(&self) -> Result<(), TargetError> {
self.invalidate_cached_publisher().await;
self.tls_state.lock().reset();
self.connected.store(false, Ordering::SeqCst);
info!(target_id = %self.id, "Redis target closed");
Ok(())
@@ -696,9 +752,20 @@ fn read_root_cert(tls: &RedisTlsConfig) -> Result<Option<Vec<u8>>, TargetError>
return Ok(None);
}
std::fs::read(&tls.ca_path)
.map(Some)
.map_err(|e| TargetError::Configuration(format!("Failed to read Redis root CA cert: {e}")))
let pem =
std::fs::read(&tls.ca_path).map_err(|e| TargetError::Configuration(format!("Failed to read Redis root CA cert: {e}")))?;
let mut reader = BufReader::new(pem.as_slice());
let certs_der = CertificateDer::pem_reader_iter(&mut reader)
.collect::<Result<Vec<_>, _>>()
.map_err(|e| TargetError::Configuration(format!("Failed to parse Redis root CA cert: {e}")))?;
if certs_der.is_empty() {
return Err(TargetError::Configuration(
"Redis root CA cert did not contain any parsable certificates".to_string(),
));
}
Ok(Some(pem))
}
fn map_redis_error(err: RedisError) -> TargetError {
@@ -722,6 +789,46 @@ fn compute_retry_delay(attempt: usize, min_delay: Duration, max_delay: Duration)
min_delay.saturating_mul(factor).min(max_delay)
}
/// Coordinated TLS hot-reload implementation for Redis targets.
///
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the Redis client without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for RedisTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = Client;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.tls.ca_path.clone(),
client_cert_path: self.args.tls.client_cert_path.clone(),
client_key_path: self.args.tls.client_key_path.clone(),
target_label: format!("redis:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
build_redis_client(&self.args)
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
*self.publisher_client.lock() = (*material).clone();
self.invalidate_cached_publisher().await;
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
validate_tls_material(&self.args.tls.ca_path, &self.args.tls.client_cert_path, &self.args.tls.client_key_path)
}
}
#[cfg(test)]
mod tests {
use super::*;
crates/targets/src/target/webhook.rs
+105 -10
View File
@@ -16,14 +16,21 @@ use crate::{
StoreError, Target,
arn::TargetID,
error::TargetError,
runtime::tls::{
ReloadableTargetTls, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode, fingerprint::TargetTlsGeneration,
validate::validate_tls_material,
},
store::{Key, Store},
target::{
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
TargetType, build_queued_payload, open_target_queue_store, persist_queued_payload_to_store,
TargetTlsState, TargetType, build_queued_payload, build_target_tls_fingerprint, open_target_queue_store,
persist_queued_payload_to_store,
},
};
use async_trait::async_trait;
use parking_lot::Mutex;
use reqwest::{Client, StatusCode, Url};
use rustfs_tls_runtime::load_cert_bundle_der_bytes;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::{
@@ -104,7 +111,11 @@ where
id: TargetID,
args: WebhookArgs,
health_check_url: Option<Url>,
http_client: Arc<Client>,
http_client: Arc<Mutex<Client>>,
tls_state: Arc<Mutex<TargetTlsState>>,
/// When present, the adapter provides coordinator-managed TLS material;
/// otherwise the inline fingerprint path is used as a fallback.
tls_adapter: Option<TlsReloadAdapter<Client>>,
// Add Send + Sync constraints to ensure thread safety
store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
initialized: AtomicBool,
@@ -124,6 +135,8 @@ where
args: self.args.clone(),
health_check_url: self.health_check_url.clone(),
http_client: Arc::clone(&self.http_client),
tls_state: Arc::clone(&self.tls_state),
tls_adapter: self.tls_adapter.clone(),
store: self.store.as_ref().map(|s| s.boxed_clone()),
initialized: AtomicBool::new(self.initialized.load(Ordering::SeqCst)),
cancel_sender: self.cancel_sender.clone(),
@@ -146,7 +159,7 @@ where
};
// Build HTTP client using the helper function
let http_client = Arc::new(Self::build_http_client(&args)?);
let http_client = Arc::new(Mutex::new(Self::build_http_client(&args)?));
let queue_store = open_target_queue_store(
&args.queue_dir,
@@ -165,6 +178,8 @@ where
args,
health_check_url,
http_client,
tls_state: Arc::new(Mutex::new(TargetTlsState::default())),
tls_adapter: None,
store: queue_store,
initialized: AtomicBool::new(false),
cancel_sender,
@@ -188,11 +203,18 @@ where
);
} else if !args.client_ca.is_empty() {
// Use user-provided custom CA certificate
let ca_cert_pem = std::fs::read(&args.client_ca)
.map_err(|e| TargetError::Configuration(format!("Failed to read root CA cert: {e}")))?;
let ca_cert = reqwest::Certificate::from_pem(&ca_cert_pem)
let certs_der = load_cert_bundle_der_bytes(&args.client_ca)
.map_err(|e| TargetError::Configuration(format!("Failed to parse root CA cert: {e}")))?;
client_builder = client_builder.add_root_certificate(ca_cert);
if certs_der.is_empty() {
return Err(TargetError::Configuration(
"Webhook client_ca did not contain any parsable certificates".to_string(),
));
}
for cert_der in certs_der {
let ca_cert = reqwest::Certificate::from_der(&cert_der)
.map_err(|e| TargetError::Configuration(format!("Failed to load root CA cert: {e}")))?;
client_builder = client_builder.add_root_certificate(ca_cert);
}
}
// If neither is set, use the system's default trust store
@@ -213,6 +235,29 @@ where
.map_err(|e| TargetError::Configuration(format!("Failed to build HTTP client: {e}")))
}
async fn refresh_tls(&self) -> Result<(), TargetError> {
let next_fingerprint =
build_target_tls_fingerprint(&self.args.client_ca, &self.args.client_cert, &self.args.client_key).await?;
let tls_changed = {
let tls_state_guard = self.tls_state.lock();
tls_state_guard.fingerprint.as_ref() != Some(&next_fingerprint)
};
if !tls_changed {
return Ok(());
}
let new_client = Self::build_http_client(&self.args)?;
{
let mut tls_state_guard = self.tls_state.lock();
if tls_state_guard.fingerprint.as_ref() == Some(&next_fingerprint) {
return Ok(());
}
*self.http_client.lock() = new_client;
tls_state_guard.refresh(next_fingerprint);
}
Ok(())
}
fn health_check_url(endpoint: &Url) -> Result<Url, TargetError> {
endpoint
.host()
@@ -230,7 +275,8 @@ where
return Ok(false);
};
match tokio::time::timeout(Duration::from_secs(5), self.http_client.head(health_check_url.as_str()).send()).await {
let client = self.http_client.lock().clone();
match tokio::time::timeout(Duration::from_secs(5), client.head(health_check_url.as_str()).send()).await {
Ok(Ok(resp)) => {
debug!(
target = %self.id,
@@ -299,8 +345,14 @@ where
"Sending webhook payload"
);
let mut req_builder = self
.http_client
// When a TLS reload adapter is attached, it drives client rebuilds in
// the background. The inline per-send fingerprint check is skipped.
if self.tls_adapter.is_none() {
self.refresh_tls().await?;
}
let client = self.http_client.lock().clone();
let mut req_builder = client
.post(self.args.endpoint.as_str())
.header("Content-Type", meta.content_type.as_str());
@@ -425,6 +477,7 @@ where
async fn close(&self) -> Result<(), TargetError> {
// Send cancel signal to background tasks
let _ = self.cancel_sender.try_send(());
// Adapter cleanup is done by the coordinator; no local state to reset.
info!("Webhook target closed: {}", self.id);
Ok(())
}
@@ -460,6 +513,48 @@ where
}
}
/// Coordinated TLS hot-reload implementation for Webhook targets.
///
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the HTTP client without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for WebhookTarget<E>
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
type Material = Client;
fn tls_input_set(&self) -> TargetTlsInputSet {
TargetTlsInputSet {
ca_path: self.args.client_ca.clone(),
client_cert_path: self.args.client_cert.clone(),
client_key_path: self.args.client_key.clone(),
target_label: format!("webhook:{}", self.id.id),
}
}
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
// build_http_client is synchronous (reads files + configures reqwest).
// The coordinator already runs this in a background task, so the
// synchronous file I/O does not block the send path.
Self::build_http_client(&self.args)
}
async fn apply_tls_material(
&self,
_generation: TargetTlsGeneration,
material: Arc<Self::Material>,
_mode: ReloadApplyMode,
) -> Result<(), TargetError> {
*self.http_client.lock() = (*material).clone();
Ok(())
}
async fn validate_tls_files(&self) -> Result<(), TargetError> {
validate_tls_material(&self.args.client_ca, &self.args.client_cert, &self.args.client_key)
}
}
#[cfg(test)]
mod tests {
use super::{WebhookArgs, WebhookTarget};