rustfs_targets/
plugin.rs

1// Copyright 2024 RustFS Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use crate::{
16    PluginRuntimeAdapter, RuntimeActivation, Target, TargetError,
17    config::{collect_target_config_results, redact_error_detail_with_config},
18    manifest::{TargetPluginManifest, builtin_target_manifest},
19    target::with_deferred_queue_store_open,
20};
21use hashbrown::HashMap;
22use rustfs_config::server_config::{Config, KVS};
23use serde::Serialize;
24use serde::de::DeserializeOwned;
25use std::collections::HashSet;
26use std::sync::Arc;
27use tracing::{error, info, warn};
28
29type BoxedTarget<E> = Box<dyn Target<E> + Send + Sync>;
30type TargetCreateFn<E> = Arc<dyn Fn(String, &KVS) -> Result<BoxedTarget<E>, TargetError> + Send + Sync>;
31type TargetValidateFn = Arc<dyn Fn(&KVS) -> Result<(), TargetError> + Send + Sync>;
32
33/// Event payload contract shared by all target plugin machinery.
34///
35/// Blanket-implemented for every type meeting the bounds; it exists solely to
36/// keep this composite bound spelled in one place instead of on every generic.
37pub trait PluginEvent: Send + Sync + Clone + Serialize + DeserializeOwned + 'static {}
38
39impl<T> PluginEvent for T where T: Send + Sync + Clone + Serialize + DeserializeOwned + 'static {}
40
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum TargetRequestValidator {
43    Webhook,
44    Mqtt,
45    Amqp(crate::target::TargetType),
46    Kafka(crate::target::TargetType),
47    MySql(crate::target::TargetType),
48    Nats(crate::target::TargetType),
49    Postgres(crate::target::TargetType),
50    Pulsar(crate::target::TargetType),
51    Redis {
52        default_channel: &'static str,
53        target_type: crate::target::TargetType,
54    },
55}
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq)]
58pub struct TargetAdminMetadata {
59    subsystem: &'static str,
60    request_validator: TargetRequestValidator,
61}
62
63impl TargetAdminMetadata {
64    pub fn new(subsystem: &'static str, request_validator: TargetRequestValidator) -> Self {
65        Self {
66            subsystem,
67            request_validator,
68        }
69    }
70
71    #[inline]
72    pub fn subsystem(&self) -> &'static str {
73        self.subsystem
74    }
75
76    #[inline]
77    pub fn request_validator(&self) -> TargetRequestValidator {
78        self.request_validator
79    }
80}
81
82#[derive(Debug, Clone, Copy, PartialEq, Eq)]
83pub struct BuiltinTargetAdminDescriptor {
84    manifest: TargetPluginManifest,
85    valid_fields: &'static [&'static str],
86    admin: TargetAdminMetadata,
87}
88
89impl BuiltinTargetAdminDescriptor {
90    pub fn new(manifest: TargetPluginManifest, valid_fields: &'static [&'static str], admin: TargetAdminMetadata) -> Self {
91        Self {
92            manifest,
93            valid_fields,
94            admin,
95        }
96    }
97
98    #[inline]
99    pub fn manifest(&self) -> &TargetPluginManifest {
100        &self.manifest
101    }
102
103    #[inline]
104    pub fn valid_fields(&self) -> &'static [&'static str] {
105        self.valid_fields
106    }
107
108    #[inline]
109    pub fn admin_metadata(&self) -> TargetAdminMetadata {
110        self.admin
111    }
112}
113
114#[derive(Clone)]
115pub struct TargetPluginDescriptor<E>
116where
117    E: PluginEvent,
118{
119    create_target: TargetCreateFn<E>,
120    manifest: TargetPluginManifest,
121    target_type: &'static str,
122    valid_fields: &'static [&'static str],
123    valid_fields_set: Arc<HashSet<String>>,
124    validate_config: TargetValidateFn,
125}
126
127impl<E> TargetPluginDescriptor<E>
128where
129    E: PluginEvent,
130{
131    pub fn new<Create, Validate>(
132        target_type: &'static str,
133        valid_fields: &'static [&'static str],
134        validate_config: Validate,
135        create_target: Create,
136    ) -> Self
137    where
138        Create: Fn(String, &KVS) -> Result<BoxedTarget<E>, TargetError> + Send + Sync + 'static,
139        Validate: Fn(&KVS) -> Result<(), TargetError> + Send + Sync + 'static,
140    {
141        Self::with_manifest(builtin_target_manifest(target_type), valid_fields, validate_config, create_target)
142    }
143
144    pub fn with_manifest<Create, Validate>(
145        manifest: TargetPluginManifest,
146        valid_fields: &'static [&'static str],
147        validate_config: Validate,
148        create_target: Create,
149    ) -> Self
150    where
151        Create: Fn(String, &KVS) -> Result<BoxedTarget<E>, TargetError> + Send + Sync + 'static,
152        Validate: Fn(&KVS) -> Result<(), TargetError> + Send + Sync + 'static,
153    {
154        Self {
155            create_target: Arc::new(create_target),
156            manifest,
157            target_type: manifest.target_type,
158            valid_fields,
159            valid_fields_set: Arc::new(valid_fields.iter().map(|field| (*field).to_string()).collect()),
160            validate_config: Arc::new(validate_config),
161        }
162    }
163
164    #[inline]
165    pub fn target_type(&self) -> &'static str {
166        self.target_type
167    }
168
169    #[inline]
170    pub fn manifest(&self) -> &TargetPluginManifest {
171        &self.manifest
172    }
173
174    #[inline]
175    pub fn valid_fields(&self) -> &'static [&'static str] {
176        self.valid_fields
177    }
178
179    #[inline]
180    pub fn valid_fields_set(&self) -> &HashSet<String> {
181        self.valid_fields_set.as_ref()
182    }
183
184    #[inline]
185    pub fn validate_config(&self, config: &KVS) -> Result<(), TargetError> {
186        (self.validate_config)(config)
187    }
188
189    #[inline]
190    pub fn create_target(&self, id: String, config: &KVS) -> Result<BoxedTarget<E>, TargetError> {
191        (self.create_target)(id, config)
192    }
193}
194
195#[derive(Clone)]
196pub struct BuiltinTargetDescriptor<E>
197where
198    E: PluginEvent,
199{
200    plugin: TargetPluginDescriptor<E>,
201    admin: TargetAdminMetadata,
202}
203
204impl<E> BuiltinTargetDescriptor<E>
205where
206    E: PluginEvent,
207{
208    pub fn new(subsystem: &'static str, request_validator: TargetRequestValidator, plugin: TargetPluginDescriptor<E>) -> Self {
209        Self {
210            plugin,
211            admin: TargetAdminMetadata::new(subsystem, request_validator),
212        }
213    }
214
215    #[inline]
216    pub fn plugin(&self) -> &TargetPluginDescriptor<E> {
217        &self.plugin
218    }
219
220    #[inline]
221    pub fn admin_metadata(&self) -> TargetAdminMetadata {
222        self.admin
223    }
224
225    #[inline]
226    pub fn request_validator(&self) -> TargetRequestValidator {
227        self.admin.request_validator()
228    }
229
230    #[inline]
231    pub fn subsystem(&self) -> &'static str {
232        self.admin.subsystem()
233    }
234}
235
236impl<E> From<BuiltinTargetDescriptor<E>> for BuiltinTargetAdminDescriptor
237where
238    E: PluginEvent,
239{
240    fn from(descriptor: BuiltinTargetDescriptor<E>) -> Self {
241        Self::new(
242            *descriptor.plugin().manifest(),
243            descriptor.plugin().valid_fields(),
244            descriptor.admin_metadata(),
245        )
246    }
247}
248
249pub struct TargetPluginRegistry<E>
250where
251    E: PluginEvent,
252{
253    plugins: HashMap<String, TargetPluginDescriptor<E>>,
254}
255
256impl<E> Default for TargetPluginRegistry<E>
257where
258    E: PluginEvent,
259{
260    fn default() -> Self {
261        Self::new()
262    }
263}
264
265impl<E> TargetPluginRegistry<E>
266where
267    E: PluginEvent,
268{
269    pub fn new() -> Self {
270        Self { plugins: HashMap::new() }
271    }
272
273    pub fn register(&mut self, plugin: TargetPluginDescriptor<E>) -> Option<TargetPluginDescriptor<E>> {
274        let replaced = self.plugins.insert(plugin.target_type().to_string(), plugin);
275        if let Some(previous) = &replaced {
276            warn!(
277                target_type = %previous.target_type(),
278                plugin_id = %previous.manifest().plugin_id,
279                "replacing previously registered target plugin descriptor"
280            );
281        }
282        replaced
283    }
284
285    pub fn register_all<I>(&mut self, plugins: I)
286    where
287        I: IntoIterator<Item = TargetPluginDescriptor<E>>,
288    {
289        for plugin in plugins {
290            self.register(plugin);
291        }
292    }
293
294    pub fn supports_target_type(&self, target_type: &str) -> bool {
295        self.plugins.contains_key(target_type)
296    }
297
298    pub fn registered_target_types(&self) -> Vec<String> {
299        self.plugins.keys().cloned().collect()
300    }
301
302    pub fn create_target(&self, target_type: &str, id: String, config: &KVS) -> Result<BoxedTarget<E>, TargetError> {
303        let plugin = self
304            .plugins
305            .get(target_type)
306            .ok_or_else(|| TargetError::Configuration(format!("Unknown target type: {target_type}")))?;
307        plugin.validate_config(config)?;
308        plugin.create_target(id, config)
309    }
310
311    /// Creates every enabled target instance found in `config`.
312    ///
313    /// Creation is fault-isolated per instance: one broken target must not
314    /// prevent the remaining targets from activating, so failures are logged
315    /// and summarized instead of aborting the whole activation.
316    pub async fn create_targets_from_config(
317        &self,
318        config: &Config,
319        route_prefix: &str,
320    ) -> Result<Vec<BoxedTarget<E>>, TargetError> {
321        self.create_targets_from_config_with_store_mode(config, route_prefix, false)
322            .await
323            .map(|(targets, _)| targets)
324    }
325
326    /// Creates targets while deferring queue-store open until runtime handoff.
327    /// Unlike the compatibility activation API, lifecycle preparation reports
328    /// any invalid or unconstructable configured instance so the originating
329    /// Admin request cannot report a false success.
330    pub async fn create_dormant_targets_from_config(
331        &self,
332        config: &Config,
333        route_prefix: &str,
334    ) -> Result<(Vec<BoxedTarget<E>>, Vec<String>), TargetError> {
335        self.create_targets_from_config_with_store_mode(config, route_prefix, true)
336            .await
337    }
338
339    async fn create_targets_from_config_with_store_mode(
340        &self,
341        config: &Config,
342        route_prefix: &str,
343        defer_store_open: bool,
344    ) -> Result<(Vec<BoxedTarget<E>>, Vec<String>), TargetError> {
345        let mut successful_targets = Vec::new();
346        let mut failures = Vec::new();
347
348        for (target_type, plugin) in &self.plugins {
349            info!(target_type = %target_type, "Start working on target type");
350            // Per-instance fault isolation: an invalid instance (e.g. an
351            // unparseable `enable` value) is recorded as a failure and skipped,
352            // never aborting the remaining instances or other target types.
353            let (collected, invalid_instances) =
354                collect_target_config_results(config, route_prefix, target_type, plugin.valid_fields_set());
355            for detail in invalid_instances {
356                error!(target_type = %target_type, reason = "invalid_config", detail = %detail, "Skipping target instance with invalid configuration");
357                failures.push(detail);
358            }
359            for (id, merged_config) in collected {
360                info!(target_type = %target_type, instance_id = %id, "Target is enabled, ready to create");
361                let created = if defer_store_open {
362                    with_deferred_queue_store_open(|| self.create_target(target_type, id.clone(), &merged_config))
363                } else {
364                    self.create_target(target_type, id.clone(), &merged_config)
365                };
366                match created {
367                    Ok(target) => {
368                        info!(target_type = %target.id().name, instance_id = %id, "Create target successfully");
369                        successful_targets.push(target);
370                    }
371                    Err(err) => {
372                        // The underlying error names the root cause (egress policy
373                        // rejection, queue-store open failure, ...); scrub it against
374                        // the instance config so credential-bearing values never
375                        // reach the log or the Admin-visible failure summary.
376                        let detail = redact_error_detail_with_config(&err.to_string(), &merged_config);
377                        failures.push(format!("{target_type}/{id}: target construction failed: {detail}"));
378                        error!(target_type = %target_type, instance_id = %id, reason = "construction_failed", detail = %detail, "Failed to create target");
379                    }
380                }
381            }
382        }
383
384        if !failures.is_empty() {
385            warn!(
386                created = successful_targets.len(),
387                failed = failures.len(),
388                "Some configured targets failed to create and were skipped"
389            );
390        }
391        info!(
392            count = successful_targets.len(),
393            failed = failures.len(),
394            "All target processing completed"
395        );
396        Ok((successful_targets, failures))
397    }
398
399    pub async fn create_activation_from_config<A>(
400        &self,
401        config: &Config,
402        route_prefix: &str,
403        adapter: &A,
404    ) -> Result<RuntimeActivation<E>, TargetError>
405    where
406        A: PluginRuntimeAdapter<E> + ?Sized,
407    {
408        let targets = self.create_targets_from_config(config, route_prefix).await?;
409        Ok(adapter.activate_with_replay(targets).await)
410    }
411}
412
413pub fn boxed_target<E, T>(target: T) -> BoxedTarget<E>
414where
415    E: PluginEvent,
416    T: Target<E> + Send + Sync + 'static,
417{
418    Box::new(target)
419}
420
421#[cfg(test)]
422mod tests {
423    use super::{TargetPluginDescriptor, TargetPluginRegistry};
424    use crate::TargetError;
425    use crate::runtime::adapter::BuiltinPluginRuntimeAdapter;
426    use crate::testkit::MockTarget;
427    use rustfs_config::ENABLE_KEY;
428    use rustfs_config::server_config::{Config, KVS};
429    use std::collections::HashMap;
430    use std::sync::Arc;
431    use std::time::Duration;
432
433    fn builtin_adapter() -> BuiltinPluginRuntimeAdapter<String> {
434        BuiltinPluginRuntimeAdapter::new(
435            Arc::new(|_event| Box::pin(async {})),
436            Arc::new(|_target_id, _has_replay| {}),
437            None,
438            Duration::from_millis(10),
439            Duration::from_millis(10),
440            "stopping plugin registry test replay worker",
441        )
442    }
443
444    #[tokio::test]
445    async fn registry_creates_activation_from_config_via_runtime_adapter() {
446        let mut registry = TargetPluginRegistry::new();
447        registry.register(TargetPluginDescriptor::new(
448            "test",
449            &[ENABLE_KEY, "endpoint"],
450            |_config| Ok(()),
451            |id, _config| Ok(Box::new(MockTarget::new(&id, "test"))),
452        ));
453
454        let mut cfg = Config(HashMap::new());
455        let mut section = HashMap::new();
456        let mut primary = KVS::new();
457        primary.insert(ENABLE_KEY.to_string(), "on".to_string());
458        primary.insert("endpoint".to_string(), "https://example.com/hook".to_string());
459        section.insert("primary".to_string(), primary);
460        cfg.0.insert("notify_test".to_string(), section);
461
462        let adapter = builtin_adapter();
463        let activation = registry
464            .create_activation_from_config(&cfg, "notify_", &adapter)
465            .await
466            .expect("activation should be created through runtime adapter");
467
468        assert_eq!(activation.targets.len(), 1);
469        assert_eq!(activation.targets[0].id().to_string(), "primary:test");
470        assert!(activation.replay_workers.is_empty());
471    }
472
473    // Regression: a single instance with a malformed `enable` value must not
474    // abort the remaining instances or unrelated target types. Before this fix
475    // the collector short-circuited the whole create path, so one typo took
476    // down every notify/audit target.
477    #[tokio::test]
478    async fn create_dormant_isolates_invalid_enable_and_still_loads_other_targets() {
479        let mut registry = TargetPluginRegistry::<String>::new();
480        for target_type in ["alpha", "beta"] {
481            registry.register(TargetPluginDescriptor::new(
482                target_type,
483                &[ENABLE_KEY, "endpoint"],
484                |_config| Ok(()),
485                move |id, _config| Ok(Box::new(MockTarget::new(&id, target_type))),
486            ));
487        }
488
489        let mut cfg = Config(HashMap::new());
490
491        // alpha: one healthy instance plus one with a malformed `enable` value
492        // ("enable" is a typo -- EnableState accepts "enabled"/"on", not "enable").
493        let mut alpha = HashMap::new();
494        let mut alpha_good = KVS::new();
495        alpha_good.insert(ENABLE_KEY.to_string(), "on".to_string());
496        alpha_good.insert("endpoint".to_string(), "https://example.com/alpha".to_string());
497        alpha.insert("good".to_string(), alpha_good);
498        let mut alpha_bad = KVS::new();
499        alpha_bad.insert(ENABLE_KEY.to_string(), "enable".to_string());
500        alpha.insert("bad".to_string(), alpha_bad);
501        cfg.0.insert("notify_alpha".to_string(), alpha);
502
503        // beta: a healthy instance in a different target type must survive.
504        let mut beta = HashMap::new();
505        let mut beta_primary = KVS::new();
506        beta_primary.insert(ENABLE_KEY.to_string(), "on".to_string());
507        beta_primary.insert("endpoint".to_string(), "https://example.com/beta".to_string());
508        beta.insert("primary".to_string(), beta_primary);
509        cfg.0.insert("notify_beta".to_string(), beta);
510
511        let (targets, failures) = registry
512            .create_dormant_targets_from_config(&cfg, "notify_")
513            .await
514            .expect("a malformed instance must not abort target creation");
515
516        let mut created: Vec<String> = targets.iter().map(|target| target.id().to_string()).collect();
517        created.sort();
518        assert_eq!(created, vec!["good:alpha".to_string(), "primary:beta".to_string()]);
519
520        // The malformed instance is surfaced (so an Admin write can't report a
521        // false success) rather than silently dropped or fatally aborting.
522        assert_eq!(failures.len(), 1);
523        assert!(failures[0].contains("alpha/bad"), "unexpected failure summary: {}", failures[0]);
524    }
525
526    // Regression (#5115 debugging): a construction failure must carry the
527    // underlying error detail (e.g. an egress-policy rejection) in the failure
528    // summary instead of an opaque "target construction failed", while
529    // credential-bearing config values stay redacted.
530    #[tokio::test]
531    async fn construction_failure_surfaces_redacted_error_detail() {
532        let mut registry = TargetPluginRegistry::<String>::new();
533        registry.register(TargetPluginDescriptor::new(
534            "gamma",
535            &[ENABLE_KEY, "endpoint", "auth_token"],
536            |_config| Ok(()),
537            |_id, config| {
538                let endpoint = config.lookup("endpoint").unwrap_or_default();
539                let token = config.lookup("auth_token").unwrap_or_default();
540                Err(TargetError::Configuration(format!(
541                    "webhook endpoint is not allowed: {endpoint} (auth_token {token})"
542                )))
543            },
544        ));
545
546        let mut cfg = Config(HashMap::new());
547        let mut section = HashMap::new();
548        let mut primary = KVS::new();
549        primary.insert(ENABLE_KEY.to_string(), "on".to_string());
550        primary.insert("endpoint".to_string(), "https://example.com/private/hook?sig=hunter2".to_string());
551        primary.insert("auth_token".to_string(), "hook-secret-token".to_string());
552        section.insert("primary".to_string(), primary);
553        cfg.0.insert("notify_gamma".to_string(), section);
554
555        let (targets, failures) = registry
556            .create_dormant_targets_from_config(&cfg, "notify_")
557            .await
558            .expect("a failing instance must not abort target creation");
559
560        assert!(targets.is_empty());
561        assert_eq!(failures.len(), 1);
562        let failure = &failures[0];
563        assert!(failure.contains("gamma/primary"), "unexpected failure summary: {failure}");
564        // The root cause is surfaced instead of an opaque generic message.
565        assert!(failure.contains("webhook endpoint is not allowed"), "missing error detail: {failure}");
566        // The endpoint is reduced to its origin; path, query, and token are gone.
567        assert!(failure.contains("https://example.com"), "endpoint origin should stay visible: {failure}");
568        assert!(!failure.contains("/private/hook"), "endpoint path must be redacted: {failure}");
569        assert!(!failure.contains("hunter2"), "endpoint query must be redacted: {failure}");
570        assert!(!failure.contains("hook-secret-token"), "auth token must be redacted: {failure}");
571    }
572}
rustfs_targets/plugin.rs

rustfs_targets/
plugin.rs
