Rust wrappers (#1367)

plugins/wasm-rust/src/cluster_wrapper.rs

@@ -0,0 +1,259 @@
+use crate::{internal::get_property, request_wrapper::get_request_host};
+
+pub trait Cluster {
+    fn cluster_name(&self) -> String;
+    fn host_name(&self) -> String;
+}
+#[derive(Debug, Clone)]
+pub struct RouteCluster {
+    host: String,
+}
+impl RouteCluster {
+    pub fn new(host: &str) -> Self {
+        RouteCluster {
+            host: host.to_string(),
+        }
+    }
+}
+impl Cluster for RouteCluster {
+    fn cluster_name(&self) -> String {
+        if let Some(res) = get_property(vec!["cluster_name"]) {
+            if let Ok(r) = String::from_utf8(res) {
+                return r;
+            }
+        }
+        String::new()
+    }
+
+    fn host_name(&self) -> String {
+        if !self.host.is_empty() {
+            return self.host.clone();
+        }
+
+        get_request_host()
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct K8sCluster {
+    service_name: String,
+    namespace: String,
+    port: String,
+    version: String,
+    host: String,
+}
+
+impl K8sCluster {
+    pub fn new(service_name: &str, namespace: &str, port: &str, version: &str, host: &str) -> Self {
+        K8sCluster {
+            service_name: service_name.to_string(),
+            namespace: namespace.to_string(),
+            port: port.to_string(),
+            version: version.to_string(),
+            host: host.to_string(),
+        }
+    }
+}
+
+impl Cluster for K8sCluster {
+    fn cluster_name(&self) -> String {
+        format!(
+            "outbound|{}|{}|{}.{}.svc.cluster.local",
+            self.port,
+            self.version,
+            self.service_name,
+            if self.namespace.is_empty() {
+                "default"
+            } else {
+                &self.namespace
+            }
+        )
+    }
+
+    fn host_name(&self) -> String {
+        if self.host.is_empty() {
+            format!("{}.{}.svc.cluster.local", self.service_name, self.namespace)
+        } else {
+            self.host.clone()
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct NacosCluster {
+    service_name: String,
+    group: String,
+    namespace_id: String,
+    port: u16,
+    is_ext_registry: bool,
+    version: String,
+    host: String,
+}
+
+impl NacosCluster {
+    pub fn new(
+        service_name: &str,
+        group: &str,
+        namespace_id: &str,
+        port: u16,
+        is_ext_registry: bool,
+        version: &str,
+        host: &str,
+    ) -> Self {
+        NacosCluster {
+            service_name: service_name.to_string(),
+            group: group.to_string(),
+            namespace_id: namespace_id.to_string(),
+            port,
+            is_ext_registry,
+            version: version.to_string(),
+            host: host.to_string(),
+        }
+    }
+}
+impl Cluster for NacosCluster {
+    fn cluster_name(&self) -> String {
+        let group = if self.group.is_empty() {
+            "DEFAULT-GROUP".to_string()
+        } else {
+            self.group.replace('_', "-")
+        };
+        let tail = if self.is_ext_registry {
+            "nacos-ext"
+        } else {
+            "nacos"
+        };
+        format!(
+            "outbound|{}|{}|{}.{}.{}.{}",
+            self.port, self.version, self.service_name, group, self.namespace_id, tail
+        )
+    }
+
+    fn host_name(&self) -> String {
+        if self.host.is_empty() {
+            self.service_name.clone()
+        } else {
+            self.host.clone()
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct StaticIpCluster {
+    service_name: String,
+    port: u16,
+    host: String,
+}
+
+impl StaticIpCluster {
+    pub fn new(service_name: &str, port: u16, host: &str) -> Self {
+        StaticIpCluster {
+            service_name: service_name.to_string(),
+            port,
+            host: host.to_string(),
+        }
+    }
+}
+impl Cluster for StaticIpCluster {
+    fn cluster_name(&self) -> String {
+        format!("outbound|{}||{}.static", self.port, self.service_name)
+    }
+
+    fn host_name(&self) -> String {
+        if self.host.is_empty() {
+            self.service_name.clone()
+        } else {
+            self.host.clone()
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct DnsCluster {
+    service_name: String,
+    domain: String,
+    port: u16,
+}
+
+impl DnsCluster {
+    pub fn new(service_name: &str, domain: &str, port: u16) -> Self {
+        DnsCluster {
+            service_name: service_name.to_string(),
+            domain: domain.to_string(),
+            port,
+        }
+    }
+}
+impl Cluster for DnsCluster {
+    fn cluster_name(&self) -> String {
+        format!("outbound|{}||{}.dns", self.port, self.service_name)
+    }
+
+    fn host_name(&self) -> String {
+        self.domain.clone()
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct ConsulCluster {
+    service_name: String,
+    datacenter: String,
+    port: u16,
+    host: String,
+}
+
+impl ConsulCluster {
+    pub fn new(service_name: &str, datacenter: &str, port: u16, host: &str) -> Self {
+        ConsulCluster {
+            service_name: service_name.to_string(),
+            datacenter: datacenter.to_string(),
+            port,
+            host: host.to_string(),
+        }
+    }
+}
+impl Cluster for ConsulCluster {
+    fn cluster_name(&self) -> String {
+        format!(
+            "outbound|{}||{}.{}.consul",
+            self.port, self.service_name, self.datacenter
+        )
+    }
+
+    fn host_name(&self) -> String {
+        if self.host.is_empty() {
+            self.service_name.clone()
+        } else {
+            self.host.clone()
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct FQDNCluster {
+    fqdn: String,
+    host: String,
+    port: u16,
+}
+
+impl FQDNCluster {
+    pub fn new(fqdn: &str, host: &str, port: u16) -> Self {
+        FQDNCluster {
+            fqdn: fqdn.to_string(),
+            host: host.to_string(),
+            port,
+        }
+    }
+}
+impl Cluster for FQDNCluster {
+    fn cluster_name(&self) -> String {
+        format!("outbound|{}||{}", self.port, self.fqdn)
+    }
+    fn host_name(&self) -> String {
+        if self.host.is_empty() {
+            self.fqdn.clone()
+        } else {
+            self.host.clone()
+        }
+    }
+}

plugins/wasm-rust/src/lib.rs

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod cluster_wrapper;
 pub mod error;
 mod internal;
 pub mod log;
 pub mod plugin_wrapper;
+pub mod request_wrapper;
 pub mod rule_matcher;

plugins/wasm-rust/src/plugin_wrapper.rs

@@ -12,15 +12,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+use std::time::Duration;
+
+use crate::cluster_wrapper::Cluster;
+use crate::log::Log;
 use crate::rule_matcher::SharedRuleMatcher;
+use http::{method::Method, Uri};
+use lazy_static::lazy_static;
 use multimap::MultiMap;
-use proxy_wasm::hostcalls::log;
 use proxy_wasm::traits::{Context, HttpContext, RootContext};
-use proxy_wasm::types::LogLevel;
-use proxy_wasm::types::{Action, Bytes, DataAction, HeaderAction};
+use proxy_wasm::types::{Action, Bytes, DataAction, HeaderAction, Status};
 use serde::de::DeserializeOwned;
 
-pub trait RootContextWrapper<PluginConfig>: RootContext
+lazy_static! {
+    static ref LOG: Log = Log::new("plugin_wrapper".to_string());
+}
+
+pub trait RootContextWrapper<PluginConfig, HttpCallArg: 'static = ()>: RootContext
 where
     PluginConfig: Default + DeserializeOwned + 'static + Clone,
 {
@@ -39,11 +48,37 @@ where
     fn create_http_context_wrapper(
         &self,
         _context_id: u32,
-    ) -> Option<Box<dyn HttpContextWrapper<PluginConfig>>> {
+    ) -> Option<Box<dyn HttpContextWrapper<PluginConfig, HttpCallArg>>> {
         None
     }
 }
-pub trait HttpContextWrapper<PluginConfig>: HttpContext {
+pub type HttpCallbackFn<T> = dyn FnOnce(&mut T, u16, &MultiMap<String, String>, Option<Vec<u8>>);
+
+pub struct HttpCallArgStorage<HttpCallArg> {
+    args: HashMap<u32, HttpCallArg>,
+}
+impl<HttpCallArg> Default for HttpCallArgStorage<HttpCallArg> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+impl<HttpCallArg> HttpCallArgStorage<HttpCallArg> {
+    pub fn new() -> Self {
+        HttpCallArgStorage {
+            args: HashMap::new(),
+        }
+    }
+    pub fn set(&mut self, token_id: u32, arg: HttpCallArg) {
+        self.args.insert(token_id, arg);
+    }
+    pub fn pop(&mut self, token_id: u32) -> Option<HttpCallArg> {
+        self.args.remove(&token_id)
+    }
+}
+pub trait HttpContextWrapper<PluginConfig, HttpCallArg = ()>: HttpContext {
+    fn log(&self) -> &Log {
+        &LOG
+    }
     fn on_config(&mut self, _config: &PluginConfig) {}
     fn on_http_request_complete_headers(
         &mut self,
@@ -69,26 +104,96 @@ pub trait HttpContextWrapper<PluginConfig>: HttpContext {
     fn on_http_response_complete_body(&mut self, _res_body: &Bytes) -> DataAction {
         DataAction::Continue
     }
+
+    #[allow(clippy::too_many_arguments)]
+    fn on_http_call_response_detail(
+        &mut self,
+        _token_id: u32,
+        _arg: HttpCallArg,
+        _status_code: u16,
+        _headers: &MultiMap<String, String>,
+        _body: Option<Vec<u8>>,
+    ) {
+    }
     fn replace_http_request_body(&mut self, body: &[u8]) {
         self.set_http_request_body(0, i32::MAX as usize, body)
     }
     fn replace_http_response_body(&mut self, body: &[u8]) {
         self.set_http_response_body(0, i32::MAX as usize, body)
     }
+
+    fn get_http_call_storage(&mut self) -> Option<&mut HttpCallArgStorage<HttpCallArg>> {
+        None
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn http_call(
+        &mut self,
+        cluster: &dyn Cluster,
+        method: &Method,
+        raw_url: &str,
+        headers: MultiMap<String, String>,
+        body: Option<&[u8]>,
+        arg: HttpCallArg,
+        timeout: Duration,
+    ) -> Result<u32, Status> {
+        if let Ok(uri) = raw_url.parse::<Uri>() {
+            let mut authority = cluster.host_name();
+            if let Some(host) = uri.host() {
+                authority = host.to_string();
+            }
+            let mut path = uri.path().to_string();
+            if let Some(query) = uri.query() {
+                path = format!("{}?{}", path, query);
+            }
+            let mut headers_vec = Vec::new();
+            for (k, v) in headers.iter() {
+                headers_vec.push((k.as_str(), v.as_str()));
+            }
+            headers_vec.push((":method", method.as_str()));
+            headers_vec.push((":path", &path));
+            headers_vec.push((":authority", &authority));
+            let ret = self.dispatch_http_call(
+                &cluster.cluster_name(),
+                headers_vec,
+                body,
+                Vec::new(),
+                timeout,
+            );
+
+            if let Ok(token_id) = ret {
+                if let Some(storage) = self.get_http_call_storage() {
+                    storage.set(token_id, arg);
+                    self.log().debug(
+                        &format!(
+                            "http call start, id: {}, cluster: {}, method: {}, url: {}, body: {:?}, timeout: {:?}",
+                            token_id, cluster.cluster_name(), method.as_str(), raw_url, body, timeout
+                        )
+                    );
+                } else {
+                    return Err(Status::InternalFailure);
+                }
+            }
+            ret
+        } else {
+            self.log().critical(&format!("invalid raw_url:{}", raw_url));
+            Err(Status::ParseFailure)
+        }
+    }
 }
-pub struct PluginHttpWrapper<PluginConfig> {
+pub struct PluginHttpWrapper<PluginConfig, HttpCallArg = ()> {
     req_headers: MultiMap<String, String>,
     res_headers: MultiMap<String, String>,
     req_body_len: usize,
     res_body_len: usize,
     config: Option<PluginConfig>,
     rule_matcher: SharedRuleMatcher<PluginConfig>,
-    http_content: Box<dyn HttpContextWrapper<PluginConfig>>,
+    http_content: Box<dyn HttpContextWrapper<PluginConfig, HttpCallArg>>,
 }
-impl<PluginConfig> PluginHttpWrapper<PluginConfig> {
+impl<PluginConfig, HttpCallArg> PluginHttpWrapper<PluginConfig, HttpCallArg> {
     pub fn new(
         rule_matcher: &SharedRuleMatcher<PluginConfig>,
-        http_content: Box<dyn HttpContextWrapper<PluginConfig>>,
+        http_content: Box<dyn HttpContextWrapper<PluginConfig, HttpCallArg>>,
     ) -> Self {
         PluginHttpWrapper {
             req_headers: MultiMap::new(),
@@ -100,8 +205,15 @@ impl<PluginConfig> PluginHttpWrapper<PluginConfig> {
             http_content,
         }
     }
+    fn get_http_call_arg(&mut self, token_id: u32) -> Option<HttpCallArg> {
+        if let Some(storage) = self.http_content.get_http_call_storage() {
+            storage.pop(token_id)
+        } else {
+            None
+        }
+    }
 }
-impl<PluginConfig> Context for PluginHttpWrapper<PluginConfig> {
+impl<PluginConfig, HttpCallArg> Context for PluginHttpWrapper<PluginConfig, HttpCallArg> {
     fn on_http_call_response(
         &mut self,
         token_id: u32,
@@ -109,8 +221,50 @@ impl<PluginConfig> Context for PluginHttpWrapper<PluginConfig> {
         body_size: usize,
         num_trailers: usize,
     ) {
-        self.http_content
-            .on_http_call_response(token_id, num_headers, body_size, num_trailers)
+        if let Some(arg) = self.get_http_call_arg(token_id) {
+            let body = self.get_http_call_response_body(0, body_size);
+            let mut headers = MultiMap::new();
+            let mut status_code = 502;
+            let mut normal_response = false;
+            for (k, v) in self.get_http_call_response_headers_bytes() {
+                match String::from_utf8(v) {
+                    Ok(header_value) => {
+                        if k == ":status" {
+                            if let Ok(code) = header_value.parse::<u16>() {
+                                status_code = code;
+                                normal_response = true;
+                            } else {
+                                self.http_content
+                                    .log()
+                                    .error(&format!("failed to parse status: {}", header_value));
+                                status_code = 500;
+                            }
+                        }
+                        headers.insert(k, header_value);
+                    }
+                    Err(_) => {
+                        self.http_content.log().warn(&format!(
+                            "http call response header contains non-ASCII characters header: {}",
+                            k
+                        ));
+                    }
+                }
+            }
+            self.http_content.log().warn(&format!(
+                "http call end, id: {}, code: {}, normal: {}, body: {:?}",
+                token_id, status_code, normal_response, body
+            ));
+            self.http_content.on_http_call_response_detail(
+                token_id,
+                arg,
+                status_code,
+                &headers,
+                body,
+            )
+        } else {
+            self.http_content
+                .on_http_call_response(token_id, num_headers, body_size, num_trailers)
+        }
     }
 
     fn on_grpc_call_response(&mut self, token_id: u32, status_code: u32, response_size: usize) {
@@ -138,7 +292,7 @@ impl<PluginConfig> Context for PluginHttpWrapper<PluginConfig> {
         self.http_content.on_done()
     }
 }
-impl<PluginConfig> HttpContext for PluginHttpWrapper<PluginConfig>
+impl<PluginConfig, HttpCallArg> HttpContext for PluginHttpWrapper<PluginConfig, HttpCallArg>
 where
     PluginConfig: Default + DeserializeOwned + Clone,
 {
@@ -152,15 +306,10 @@ where
                     self.req_headers.insert(k, header_value);
                 }
                 Err(_) => {
-                    log(
-                        LogLevel::Warn,
-                        format!(
-                            "request http header contains non-ASCII characters header: {}",
-                            k
-                        )
-                        .as_str(),
-                    )
-                    .unwrap();
+                    self.http_content.log().warn(&format!(
+                        "request http header contains non-ASCII characters header: {}",
+                        k
+                    ));
                 }
             }
         }
@@ -212,15 +361,10 @@ where
                     self.res_headers.insert(k, header_value);
                 }
                 Err(_) => {
-                    log(
-                        LogLevel::Warn,
-                        format!(
-                            "response http header contains non-ASCII characters header: {}",
-                            k
-                        )
-                        .as_str(),
-                    )
-                    .unwrap();
+                    self.http_content.log().warn(&format!(
+                        "response http header contains non-ASCII characters header: {}",
+                        k
+                    ));
                 }
             }
         }

plugins/wasm-rust/src/request_wrapper.rs

@@ -0,0 +1,82 @@
+use proxy_wasm::hostcalls;
+
+use crate::internal;
+
+fn get_request_head(head: &str, log_flag: &str) -> String {
+    if let Some(value) = internal::get_http_request_header(head) {
+        value
+    } else {
+        hostcalls::log(
+            proxy_wasm::types::LogLevel::Error,
+            &format!("get request {} failed", log_flag),
+        )
+        .unwrap();
+        String::new()
+    }
+}
+pub fn get_request_scheme() -> String {
+    get_request_head(":scheme", "head")
+}
+
+pub fn get_request_host() -> String {
+    get_request_head(":authority", "host")
+}
+
+pub fn get_request_path() -> String {
+    get_request_head(":path", "path")
+}
+
+pub fn get_request_method() -> String {
+    get_request_head(":method", "method")
+}
+
+pub fn is_binary_request_body() -> bool {
+    if let Some(content_type) = internal::get_http_request_header("content-type") {
+        if content_type.contains("octet-stream") || content_type.contains("grpc") {
+            return true;
+        }
+    }
+    if let Some(encoding) = internal::get_http_request_header("content-encoding") {
+        if !encoding.is_empty() {
+            return true;
+        }
+    }
+    false
+}
+
+pub fn is_binary_response_body() -> bool {
+    if let Some(content_type) = internal::get_http_response_header("content-type") {
+        if content_type.contains("octet-stream") || content_type.contains("grpc") {
+            return true;
+        }
+    }
+    if let Some(encoding) = internal::get_http_response_header("content-encoding") {
+        if !encoding.is_empty() {
+            return true;
+        }
+    }
+    false
+}
+pub fn has_request_body() -> bool {
+    let content_type = internal::get_http_request_header("content-type");
+    let content_length_str = internal::get_http_request_header("content-length");
+    let transfer_encoding = internal::get_http_request_header("transfer-encoding");
+    hostcalls::log(
+        proxy_wasm::types::LogLevel::Debug,
+        &format!(
+            "check has request body: content_type:{:?}, content_length_str:{:?}, transfer_encoding:{:?}",
+            content_type, content_length_str, transfer_encoding
+        )
+    ).unwrap();
+    if !content_type.is_some_and(|x| !x.is_empty()) {
+        return true;
+    }
+    if let Some(cl) = content_length_str {
+        if let Ok(content_length) = cl.parse::<i32>() {
+            if content_length > 0 {
+                return true;
+            }
+        }
+    }
+    transfer_encoding.is_some_and(|x| x == "chunked")
+}