opssignal_core/providers/
external.rs

1//! Providers that a cloud add-on delivers for a build without that
2//! cloud compiled in, the way [`crate::source::ExternalFetcher`] serves
3//! cloud config sources.
4//!
5//! The base Python wheel has no cloud code. When a config names a
6//! cloud-only transport (email over Amazon SES, say), the provider hands
7//! the provider config and the signal, as JSON, to the sender the
8//! `opssignal[aws]` add-on registered. The add-on runs the same core
9//! code with its cloud feature on, through [`send_blocking`].
10
11use crate::signal::{Signal, SignalError};
12use std::collections::HashMap;
13use std::sync::{Arc, OnceLock, RwLock};
14use std::time::Duration;
15
16/// Guards against a sender that never returns; the sender applies its
17/// own, shorter request timeouts.
18const SEND_TIMEOUT: Duration = Duration::from_secs(60);
19
20/// Delivers one signal through a cloud-only transport. `provider` is
21/// the provider's name (`email`), `config_json` its config and
22/// `signal_json` the signal, both as JSON.
23pub trait ExternalSender: Send + Sync {
24    fn send(&self, provider: &str, config_json: &str, signal_json: &str) -> Result<(), String>;
25}
26
27type SenderRegistry = RwLock<HashMap<String, Arc<dyn ExternalSender>>>;
28
29fn senders() -> &'static SenderRegistry {
30    static SENDERS: OnceLock<SenderRegistry> = OnceLock::new();
31    SENDERS.get_or_init(Default::default)
32}
33
34/// Registers the sender for every cloud-only transport of `cloud`
35/// (`aws`, `gcp` or `azure`). This is what an add-on package does.
36pub fn register_cloud_sender(cloud: &str, sender: Arc<dyn ExternalSender>) {
37    senders()
38        .write()
39        .unwrap_or_else(|e| e.into_inner())
40        .insert(cloud.to_string(), sender);
41}
42
43fn registered_sender(cloud: &str) -> Option<Arc<dyn ExternalSender>> {
44    senders()
45        .read()
46        .unwrap_or_else(|e| e.into_inner())
47        .get(cloud)
48        .cloned()
49}
50
51/// Whether this build can deliver through `cloud`'s transports, natively
52/// or through a registered add-on.
53pub fn cloud_supported(cloud: &str) -> bool {
54    let built_in = (cloud == "aws" && cfg!(feature = "aws"))
55        || (cloud == "gcp" && cfg!(feature = "gcp"))
56        || (cloud == "azure" && cfg!(feature = "azure"));
57    built_in || registered_sender(cloud).is_some()
58}
59
60/// The error for a config that needs a cloud this build lacks.
61pub(crate) fn missing_support(what: &str, cloud: &str) -> String {
62    format!(
63        "{what} needs the {cloud} add-on: pip install 'opssignal[{cloud}]' for Python, \
64         or a CLI or C library built with `--features {cloud}`"
65    )
66}
67
68// Only a build without the cloud compiled in hands off to an add-on.
69#[cfg_attr(feature = "aws", allow(dead_code))]
70/// Hands a delivery to `cloud`'s registered sender.
71pub(crate) async fn send_external(
72    cloud: &str,
73    provider: &str,
74    config_json: String,
75    signal: &Signal,
76) -> Result<(), SignalError> {
77    let sender = registered_sender(cloud)
78        .ok_or_else(|| SignalError::Delivery(missing_support(provider, cloud)))?;
79    let signal_json =
80        serde_json::to_string(signal).map_err(|e| SignalError::Delivery(e.to_string()))?;
81    let provider = provider.to_string();
82    let (tx, rx) = tokio::sync::oneshot::channel();
83    // A plain thread rather than spawn_blocking: the sender may start
84    // its own runtime, which tokio forbids on its blocking pool.
85    std::thread::Builder::new()
86        .name("opssignal-send".into())
87        .spawn(move || {
88            let _ = tx.send(sender.send(&provider, &config_json, &signal_json));
89        })
90        .map_err(|e| SignalError::Delivery(format!("cannot start send thread: {e}")))?;
91    match tokio::time::timeout(SEND_TIMEOUT, rx).await {
92        Ok(Ok(result)) => result.map_err(SignalError::Delivery),
93        Ok(Err(_)) => Err(SignalError::Delivery(
94            "the add-on sender failed without an answer".into(),
95        )),
96        Err(_) => Err(SignalError::Delivery(format!(
97            "timed out after {SEND_TIMEOUT:?}"
98        ))),
99    }
100}
101
102/// Delivers a signal for an add-on, on a throwaway runtime, with errors
103/// in the form an [`ExternalSender`] returns them. This is what add-on
104/// packages call to serve a send.
105pub fn send_blocking(provider: &str, config_json: &str, signal_json: &str) -> Result<(), String> {
106    use crate::delivery::Provider;
107
108    let signal: Signal =
109        serde_json::from_str(signal_json).map_err(|e| format!("invalid signal: {e}"))?;
110    let provider: Box<dyn Provider> = match provider {
111        "email" => {
112            let config = serde_json::from_str(config_json)
113                .map_err(|e| format!("invalid email config: {e}"))?;
114            Box::new(super::email::EmailProvider::new(config))
115        }
116        other => return Err(format!("unknown provider {other:?}")),
117    };
118    let runtime = tokio::runtime::Builder::new_current_thread()
119        .enable_all()
120        .build()
121        .map_err(|e| format!("failed to start runtime: {e}"))?;
122    runtime
123        .block_on(provider.send(&signal))
124        .map_err(|e| match e {
125            SignalError::Delivery(msg) => msg,
126            other => other.to_string(),
127        })
128}
opssignal_core/providers/external.rs

opssignal_core/providers/
external.rs
