cronoq_server/
scaler.rs

1//! Webhook scaler: forwards desired worker counts to an external orchestrator.
2
3use async_trait::async_trait;
4use cronoq::{ScaleError, Scaler};
5
6/// Posts `{"pool": ..., "desired": n}` to a URL whenever a pool's target capacity changes.
7/// Capacity itself follows from workers connecting, not from this call.
8pub struct WebhookScaler {
9    url: String,
10    token: Option<String>,
11    client: reqwest::Client,
12}
13
14impl std::fmt::Debug for WebhookScaler {
15    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
16        f.debug_struct("WebhookScaler")
17            .field("url", &self.url)
18            .field("token", &self.token.as_ref().map(|_| "<redacted>"))
19            .finish()
20    }
21}
22
23impl WebhookScaler {
24    /// Scaler posting to `url`, with an optional bearer token.
25    pub fn new(url: String, token: Option<String>) -> anyhow::Result<Self> {
26        let parsed = reqwest::Url::parse(&url)
27            .map_err(|e| anyhow::anyhow!("scaler webhook url '{url}': {e}"))?;
28        match parsed.scheme() {
29            "https" => {}
30            "http" => {
31                let local = matches!(parsed.host_str(), Some("localhost" | "127.0.0.1" | "[::1]"));
32                if !local {
33                    tracing::warn!(url = %url, "scaler webhook uses plain http; use https outside a trusted network");
34                    if token.is_some() {
35                        tracing::warn!("the scaler webhook token is sent over plain http");
36                    }
37                }
38            }
39            s => anyhow::bail!("scaler webhook url must be http or https, not {s}"),
40        }
41        // reqwest is built without a bundled crypto provider; use ring, like the listeners. An
42        // error means a provider is already installed, which is fine.
43        let _ = rustls::crypto::ring::default_provider().install_default();
44        let client = reqwest::Client::builder()
45            .timeout(std::time::Duration::from_secs(10))
46            .redirect(reqwest::redirect::Policy::none())
47            .build()?;
48        Ok(WebhookScaler { url, token, client })
49    }
50}
51
52#[async_trait]
53impl Scaler for WebhookScaler {
54    async fn scale_to(&self, pool: &str, desired: u32) -> Result<u32, ScaleError> {
55        let mut req = self
56            .client
57            .post(&self.url)
58            .json(&serde_json::json!({ "pool": pool, "desired": desired }));
59        if let Some(t) = &self.token {
60            req = req.bearer_auth(t);
61        }
62        let resp = req
63            .send()
64            .await
65            .map_err(|e| ScaleError::Other(e.without_url().to_string()))?;
66        if !resp.status().is_success() {
67            return Err(ScaleError::Other(format!(
68                "scaler webhook returned {}",
69                resp.status()
70            )));
71        }
72        // Capacity follows from workers connecting, not from this call.
73        Err(ScaleError::External)
74    }
75}
cronoq_server/scaler.rs

cronoq_server/
scaler.rs
