featherbit/plugins/native/
openwhisk.rs

1//! Apache OpenWhisk serverless-upstream plugin (`openwhisk`).
2//!
3//! Port of APISIX's `openwhisk` plugin (3.17). Invokes an OpenWhisk action
4//! (blocking, with the result inlined) and returns the action's reply as the
5//! gateway response — it **replaces the upstream**, so the node's `success`
6//! port should be wired straight to `client.in`.
7//!
8//! The request body is POSTed as the action parameters to
9//! `<api_host>/api/v1/namespaces/<namespace>/actions/<package/><action>?blocking=true&result=<result>&timeout=<ms>`
10//! with an `Authorization: Basic <base64(service_token)>` header.
11//!
12//! # Response mapping
13//!
14//! OpenWhisk returns a JSON envelope. An action may return just a body, or set
15//! `statusCode` and `headers` explicitly. [`map_response`] mirrors APISIX:
16//! `statusCode` (when present) becomes the response status, `headers` are
17//! applied, and `body` (or the raw envelope when absent) becomes the body. A
18//! non-JSON envelope fails the node with `503` through the error port.
19
20use async_trait::async_trait;
21use base64::engine::general_purpose::STANDARD as BASE64;
22use base64::Engine;
23use bytes::Bytes;
24use std::collections::HashMap;
25use std::sync::Arc;
26use std::time::Duration;
27
28use crate::context::{Context, GatewayError};
29use crate::outbound::{OutboundClient, OutboundRequest, OutboundResponse};
30use crate::plugins::resources::PluginResources;
31use crate::plugins::{Plugin, PluginExecutionError, PluginOutput, PluginResult};
32
33use super::faas;
34
35/// Invokes an OpenWhisk action and maps its reply into `Context.response`.
36pub struct OpenWhiskPlugin {
37    api_host: String,
38    /// Pre-computed `Basic <base64(service_token)>` header value.
39    authorization: String,
40    namespace: String,
41    package: Option<String>,
42    action: String,
43    result: bool,
44    ssl_verify: bool,
45    timeout: Duration,
46    /// `timeout` in milliseconds, passed through as the action query param.
47    timeout_ms: u64,
48    client: Arc<OutboundClient>,
49}
50
51impl OpenWhiskPlugin {
52    /// Builds the plugin from node config.
53    ///
54    /// Accepted keys:
55    /// - `api_host` (string, **required**): the OpenWhisk API host
56    ///   (e.g. `https://ow.example.com`). Missing/empty is a config error.
57    /// - `service_token` (string, **required**): `user:pass` action token, sent
58    ///   base64-encoded as HTTP Basic auth. Missing/empty is a config error.
59    /// - `action` (string, **required**): the action name to invoke.
60    /// - `namespace` (string, default `_`): the OpenWhisk namespace.
61    /// - `package` (string, optional): the package the action belongs to.
62    /// - `result` (bool, default `true`): request `result=true` (inline the
63    ///   action result rather than the full activation record).
64    /// - `ssl_verify` (bool, default `true`): verify TLS certificates.
65    /// - `timeout` (integer ms, default `3000`): whole-call deadline, also
66    ///   passed as the action `timeout` query parameter.
67    ///
68    /// ```yaml
69    /// type: openwhisk
70    /// config:
71    ///   api_host: https://ow.example.com
72    ///   service_token: ${OPENWHISK_TOKEN}
73    ///   namespace: guest
74    ///   action: hello
75    ///   result: true
76    ///   ssl_verify: true
77    ///   timeout: 3000
78    /// ```
79    pub fn from_config(
80        config: &HashMap<String, serde_json::Value>,
81        resources: &Arc<PluginResources>,
82    ) -> Result<Self, String> {
83        let api_host = config
84            .get("api_host")
85            .and_then(|v| v.as_str())
86            .filter(|s| !s.is_empty())
87            .ok_or_else(|| "openwhisk plugin requires 'api_host'".to_string())?
88            .trim_end_matches('/')
89            .to_string();
90
91        let service_token = config
92            .get("service_token")
93            .and_then(|v| v.as_str())
94            .filter(|s| !s.is_empty())
95            .ok_or_else(|| "openwhisk plugin requires 'service_token'".to_string())?;
96        let authorization = format!("Basic {}", BASE64.encode(service_token));
97
98        let action = config
99            .get("action")
100            .and_then(|v| v.as_str())
101            .filter(|s| !s.is_empty())
102            .ok_or_else(|| "openwhisk plugin requires 'action'".to_string())?
103            .to_string();
104
105        let namespace = config
106            .get("namespace")
107            .and_then(|v| v.as_str())
108            .filter(|s| !s.is_empty())
109            .unwrap_or("_")
110            .to_string();
111
112        let package = config
113            .get("package")
114            .and_then(|v| v.as_str())
115            .filter(|s| !s.is_empty())
116            .map(String::from);
117
118        let result = config
119            .get("result")
120            .and_then(|v| v.as_bool())
121            .unwrap_or(true);
122
123        let ssl_verify = config
124            .get("ssl_verify")
125            .and_then(|v| v.as_bool())
126            .unwrap_or(true);
127
128        let timeout_ms = config
129            .get("timeout")
130            .and_then(|v| v.as_u64())
131            .unwrap_or(3000);
132
133        Ok(Self {
134            api_host,
135            authorization,
136            namespace,
137            package,
138            action,
139            result,
140            ssl_verify,
141            timeout: Duration::from_millis(timeout_ms),
142            timeout_ms,
143            client: resources.outbound.clone(),
144        })
145    }
146
147    /// Builds the OpenWhisk action-invocation URL.
148    fn endpoint(&self) -> String {
149        let package = self
150            .package
151            .as_ref()
152            .map(|p| format!("{}/", p))
153            .unwrap_or_default();
154        format!(
155            "{}/api/v1/namespaces/{}/actions/{}{}?blocking=true&result={}&timeout={}",
156            self.api_host, self.namespace, package, self.action, self.result, self.timeout_ms
157        )
158    }
159
160    /// Builds the outbound POST request carrying the client body as the action
161    /// parameters.
162    fn build_request(&self, ctx: &Context) -> OutboundRequest {
163        let headers = vec![
164            ("authorization".to_string(), self.authorization.clone()),
165            ("content-type".to_string(), "application/json".to_string()),
166        ];
167        OutboundRequest {
168            method: http::Method::POST,
169            url: self.endpoint(),
170            headers,
171            body: ctx.request.body.clone(),
172            timeout: self.timeout,
173            ssl_verify: self.ssl_verify,
174            tls: None,
175        }
176    }
177}
178
179/// The status, headers, and body mapped out of an OpenWhisk envelope.
180struct MappedResponse {
181    status: u16,
182    headers: HashMap<String, Vec<String>>,
183    body: Bytes,
184}
185
186/// Maps an OpenWhisk activation reply into a `(status, headers, body)` triple.
187///
188/// An empty body passes the transport status/body through untouched. A
189/// well-formed JSON envelope may override the status via `statusCode`, set
190/// response `headers`, and provide `body` (string or nested JSON); when
191/// `statusCode`/`body` are absent the transport status / raw envelope are used.
192/// A non-JSON, non-empty body is an error (mapped to `503` by the caller).
193fn map_response(status: u16, raw: &Bytes) -> Result<MappedResponse, String> {
194    if raw.is_empty() {
195        return Ok(MappedResponse {
196            status,
197            headers: HashMap::new(),
198            body: raw.clone(),
199        });
200    }
201
202    let envelope: serde_json::Value = serde_json::from_slice(raw)
203        .map_err(|e| format!("failed to parse openwhisk response: {}", e))?;
204
205    let mut headers: HashMap<String, Vec<String>> = HashMap::new();
206    if let Some(hdrs) = envelope.get("headers").and_then(|v| v.as_object()) {
207        for (name, value) in hdrs {
208            let value = match value {
209                serde_json::Value::String(s) => s.clone(),
210                other => other.to_string(),
211            };
212            headers
213                .entry(name.to_ascii_lowercase())
214                .or_default()
215                .push(value);
216        }
217    }
218
219    let code = envelope
220        .get("statusCode")
221        .and_then(|v| v.as_u64())
222        .map(|c| c as u16)
223        .unwrap_or(status);
224
225    let body = match envelope.get("body") {
226        Some(serde_json::Value::String(s)) => Bytes::from(s.clone().into_bytes()),
227        Some(other) => Bytes::from(other.to_string().into_bytes()),
228        None => raw.clone(),
229    };
230
231    Ok(MappedResponse {
232        status: code,
233        headers,
234        body,
235    })
236}
237
238#[async_trait]
239impl Plugin for OpenWhiskPlugin {
240    fn plugin_type(&self) -> &str {
241        "openwhisk"
242    }
243
244    async fn execute(
245        &self,
246        mut ctx: Context,
247        _named_inputs: &HashMap<String, serde_json::Value>,
248    ) -> PluginResult {
249        let request = self.build_request(&ctx);
250
251        let response: OutboundResponse = match self.client.request(request).await {
252            Ok(resp) => resp,
253            Err(e) => {
254                let (status, message) = faas::classify_error("openwhisk", &e);
255                return Err(reject(ctx, status, message));
256            }
257        };
258
259        match map_response(response.status, &response.body) {
260            Ok(mapped) => {
261                ctx.response.status_code = mapped.status;
262                ctx.response.headers = mapped.headers;
263                ctx.response.body = mapped.body;
264                Ok(PluginOutput {
265                    context: ctx,
266                    named_outputs: HashMap::new(),
267                })
268            }
269            Err(message) => Err(reject(ctx, 503, message)),
270        }
271    }
272}
273
274/// Builds the `OPENWHISK_CALLOUT_ERROR` rejection carrying the context.
275fn reject(mut ctx: Context, status: u16, message: String) -> PluginExecutionError {
276    ctx.response.status_code = status;
277    PluginExecutionError {
278        context: ctx,
279        error: GatewayError {
280            node_id: String::new(),
281            code: "OPENWHISK_CALLOUT_ERROR".to_string(),
282            message,
283            metadata: HashMap::new(),
284        },
285    }
286}
287
288#[cfg(test)]
289mod tests {
290    use super::*;
291    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
292
293    fn ctx() -> Context {
294        Context {
295            request: GatewayRequest {
296                method: "POST".to_string(),
297                path: "/orig".to_string(),
298                host: "gw".to_string(),
299                scheme: "http".to_string(),
300                headers: HashMap::new(),
301                query_params: HashMap::new(),
302                body: Bytes::from_static(b"{\"name\":\"x\"}"),
303                remote_addr: "1.2.3.4:5".to_string(),
304                protocol: Protocol::Http1,
305            },
306            response: GatewayResponse {
307                status_code: 0,
308                headers: HashMap::new(),
309                body: Bytes::new(),
310            },
311            message: HashMap::new(),
312            errors: Vec::new(),
313        }
314    }
315
316    fn plugin(config: serde_json::Value) -> OpenWhiskPlugin {
317        let map: HashMap<String, serde_json::Value> = serde_json::from_value(config).unwrap();
318        OpenWhiskPlugin::from_config(&map, &PluginResources::empty()).unwrap()
319    }
320
321    fn base_config() -> serde_json::Value {
322        serde_json::json!({
323            "api_host": "https://ow.example.com",
324            "service_token": "user:pass",
325            "namespace": "guest",
326            "action": "hello"
327        })
328    }
329
330    #[test]
331    fn test_requires_api_host_and_token_and_action() {
332        assert!(OpenWhiskPlugin::from_config(&HashMap::new(), &PluginResources::empty()).is_err());
333        let mut cfg: HashMap<String, serde_json::Value> =
334            serde_json::from_value(serde_json::json!({ "api_host": "https://x" })).unwrap();
335        assert!(OpenWhiskPlugin::from_config(&cfg, &PluginResources::empty()).is_err());
336        cfg.insert("service_token".to_string(), serde_json::json!("t"));
337        // still missing action
338        assert!(OpenWhiskPlugin::from_config(&cfg, &PluginResources::empty()).is_err());
339    }
340
341    #[test]
342    fn test_endpoint_and_auth() {
343        let p = plugin(base_config());
344        let req = p.build_request(&ctx());
345        assert_eq!(
346            req.url,
347            "https://ow.example.com/api/v1/namespaces/guest/actions/hello?blocking=true&result=true&timeout=3000"
348        );
349        let authz = req
350            .headers
351            .iter()
352            .find(|(k, _)| k == "authorization")
353            .map(|(_, v)| v.clone())
354            .unwrap();
355        // "user:pass" base64 = dXNlcjpwYXNz
356        assert_eq!(authz, "Basic dXNlcjpwYXNz");
357        // body forwarded as action params
358        assert_eq!(req.body, Bytes::from_static(b"{\"name\":\"x\"}"));
359    }
360
361    #[test]
362    fn test_endpoint_with_package_and_result_false() {
363        let mut cfg = base_config();
364        cfg["package"] = serde_json::json!("mypkg");
365        cfg["result"] = serde_json::json!(false);
366        let p = plugin(cfg);
367        assert_eq!(
368            p.endpoint(),
369            "https://ow.example.com/api/v1/namespaces/guest/actions/mypkg/hello?blocking=true&result=false&timeout=3000"
370        );
371    }
372
373    #[test]
374    fn test_map_response_status_code_and_headers_and_body() {
375        let raw = Bytes::from_static(
376            br#"{"statusCode":201,"headers":{"Content-Type":"application/json"},"body":"hi"}"#,
377        );
378        let mapped = map_response(200, &raw).unwrap();
379        assert_eq!(mapped.status, 201);
380        assert_eq!(mapped.body, Bytes::from_static(b"hi"));
381        assert_eq!(
382            mapped.headers.get("content-type"),
383            Some(&vec!["application/json".to_string()])
384        );
385    }
386
387    #[test]
388    fn test_map_response_falls_back_to_transport_status() {
389        let raw = Bytes::from_static(br#"{"greeting":"hello"}"#);
390        let mapped = map_response(200, &raw).unwrap();
391        assert_eq!(mapped.status, 200);
392        // no `body` field -> raw envelope passed through
393        assert_eq!(mapped.body, raw);
394    }
395
396    #[test]
397    fn test_map_response_empty_body_passthrough() {
398        let mapped = map_response(204, &Bytes::new()).unwrap();
399        assert_eq!(mapped.status, 204);
400        assert!(mapped.body.is_empty());
401    }
402
403    #[test]
404    fn test_map_response_invalid_json_errors() {
405        assert!(map_response(200, &Bytes::from_static(b"not json")).is_err());
406    }
407
408    #[tokio::test]
409    async fn test_callout_failure_routes_error() {
410        let mut cfg = base_config();
411        cfg["api_host"] = serde_json::json!("http://127.0.0.1:1");
412        cfg["timeout"] = serde_json::json!(200);
413        let p = plugin(cfg);
414        let err = p.execute(ctx(), &HashMap::new()).await.unwrap_err();
415        assert_eq!(err.error.code, "OPENWHISK_CALLOUT_ERROR");
416        assert!(err.context.response.status_code >= 502);
417    }
418}
featherbit/plugins/native/openwhisk.rs

featherbit/plugins/native/
openwhisk.rs
