feat(call): handle resume errors as a new 'error' output port

docs/datakit.md

@@ -58,6 +58,8 @@ An HTTP dispatch call.
 
 * `body`: body returned as the dispatch response.
 * `headers`: headers returned as the dispatch response.
+* `error`: triggered if a dispatch error occurs, such as a DNS resolver timeout, etc.
+  The port returns the error message.
 
 #### Supported attributes:
 

src/config.rs

@@ -1110,7 +1110,7 @@ mod test {
             &[&[], &[]],
             &[&[], &[]],
             &[&[(5, 0)]],
-            &[&[(6, 0)], &[]],
+            &[&[(6, 0)], &[], &[]],
             &[],
         ];
         for (i, &output_list) in output_lists.iter().enumerate() {

src/nodes/call.rs

@@ -102,11 +102,16 @@ impl Node for Call {
     }
 
     fn resume(&self, ctx: &dyn HttpContext, _inputs: &Input) -> State {
-        log::debug!("call: resume");
+        let headers = payload::from_pwm_headers(ctx.get_http_call_response_headers());
 
-        let headers = Some(payload::from_pwm_headers(
-            ctx.get_http_call_response_headers(),
-        ));
+        if let Some(dispatch_status) = headers.get_str(":dispatch_status") {
+            if dispatch_status != "ok" {
+                #[cfg(debug_assertions)]
+                log::debug!("call: resume failure status: {dispatch_status}");
+
+                return Done(vec![None, None, Some(Payload::Raw(dispatch_status.into()))]);
+            }
+        }
 
         let body = if let Some(body) = ctx.get_http_call_response_body(0, usize::MAX) {
             let content_type = ctx.get_http_call_response_header("Content-Type");
@@ -117,9 +122,8 @@ impl Node for Call {
         };
 
         // TODO only produce an output if it is connected
-        // TODO produce a Fail() status on HTTP >= 400
 
-        Done(vec![body, headers])
+        Done(vec![body, Some(headers), None])
     }
 }
 
@@ -134,7 +138,7 @@ impl NodeFactory for CallFactory {
     }
     fn default_output_ports(&self) -> PortConfig {
         PortConfig {
-            defaults: Some(PortConfig::names(&["body", "headers"])),
+            defaults: Some(PortConfig::names(&["body", "headers", "error"])),
             user_defined_ports: false,
         }
     }

src/nodes/exit.rs

@@ -34,18 +34,7 @@ impl NodeConfig for ExitConfig {
     }
 
     fn default_outputs(&self) -> Option<Vec<NodeDefaultLink>> {
-        Some(vec![
-            NodeDefaultLink {
-                this_port: "body".into(),
-                other_node: "response".into(),
-                other_port: "body".into(),
-            },
-            NodeDefaultLink {
-                this_port: "headers".into(),
-                other_node: "response".into(),
-                other_port: "headers".into(),
-            },
-        ])
+        None
     }
 }
 

src/payload.rs

@@ -157,6 +157,17 @@ impl Payload {
         }
     }
 
+    pub fn get(&self, key: &str) -> Option<&serde_json::Value> {
+        match self {
+            Payload::Json(serde_json::Value::Object(map)) => map.get(key),
+            _ => None,
+        }
+    }
+
+    pub fn get_str(&self, key: &str) -> Option<&str> {
+        self.get(key).and_then(serde_json::Value::as_str)
+    }
+
     pub fn json_null() -> Self {
         Self::Json(Json::Null)
     }