mirror of
https://github.com/alibaba/higress.git
synced 2026-05-08 04:17:27 +08:00
feat: Rust WASM supports Redis database configuration option (#2704)
This commit is contained in:
@@ -11,5 +11,5 @@ RUN cargo build --target wasm32-wasip1 $BUILD_OPTS \
|
|||||||
&& cp target/wasm32-wasip1/release/*.wasm /main.wasm \
|
&& cp target/wasm32-wasip1/release/*.wasm /main.wasm \
|
||||||
&& cargo clean
|
&& cargo clean
|
||||||
|
|
||||||
FROM scratch
|
FROM scratch AS output
|
||||||
COPY --from=builder /main.wasm plugin.wasm
|
COPY --from=builder /main.wasm plugin.wasm
|
||||||
|
|||||||
@@ -20,6 +20,15 @@ build:
|
|||||||
@echo ""
|
@echo ""
|
||||||
@echo "output wasm file: extensions/${PLUGIN_NAME}/plugin.wasm"
|
@echo "output wasm file: extensions/${PLUGIN_NAME}/plugin.wasm"
|
||||||
|
|
||||||
|
build-image:
|
||||||
|
DOCKER_BUILDKIT=1 docker build \
|
||||||
|
--build-arg PLUGIN_NAME=${PLUGIN_NAME} \
|
||||||
|
--build-arg BUILDER=${BUILDER} \
|
||||||
|
-t ${IMG} \
|
||||||
|
.
|
||||||
|
@echo ""
|
||||||
|
@echo "image: ${IMG}"
|
||||||
|
|
||||||
lint-base:
|
lint-base:
|
||||||
cargo fmt --all --check
|
cargo fmt --all --check
|
||||||
cargo clippy --workspace --all-features --all-targets
|
cargo clippy --workspace --all-features --all-targets
|
||||||
|
|||||||
@@ -101,6 +101,7 @@ make lint PLUGIN_NAME=say-hello
|
|||||||
当前 Makefile 包含以下可用目标:
|
当前 Makefile 包含以下可用目标:
|
||||||
|
|
||||||
- `build` - 构建插件(默认插件为 say-hello)
|
- `build` - 构建插件(默认插件为 say-hello)
|
||||||
|
- `build-image` - 构建插件对应镜像(默认插件为 say-hello)
|
||||||
- `lint-base` - 对所有代码进行 lint 检查
|
- `lint-base` - 对所有代码进行 lint 检查
|
||||||
- `lint` - 对指定插件进行 lint 检查
|
- `lint` - 对指定插件进行 lint 检查
|
||||||
- `test-base` - 运行所有测试
|
- `test-base` - 运行所有测试
|
||||||
@@ -384,7 +385,7 @@ docker build -t my-custom-plugin:latest --build-arg PLUGIN_NAME=my-custom-plugin
|
|||||||
# 进入项目目录
|
# 进入项目目录
|
||||||
cd plugins/wasm-rust/
|
cd plugins/wasm-rust/
|
||||||
|
|
||||||
# 构建插件(会自动推送到官方仓库)
|
# 构建插件
|
||||||
make build PLUGIN_NAME=my-plugin PLUGIN_VERSION=1.0.0
|
make build PLUGIN_NAME=my-plugin PLUGIN_VERSION=1.0.0
|
||||||
|
|
||||||
# 构建构建器镜像
|
# 构建构建器镜像
|
||||||
@@ -414,7 +415,7 @@ spec:
|
|||||||
|
|
||||||
不同命令需要在不同的目录下执行:
|
不同命令需要在不同的目录下执行:
|
||||||
|
|
||||||
- **Makefile 命令**(如 `make build`、`make test`、`make lint`):在 `plugins/wasm-rust/` 目录下执行
|
- **Makefile 命令**(如 `make build`、`make build-image`、`make test`、`make lint`):在 `plugins/wasm-rust/` 目录下执行
|
||||||
- **Cargo 命令**(如 `cargo build`、`cargo test`):在具体的插件目录下执行(如 `plugins/wasm-rust/extensions/my-plugin/`)
|
- **Cargo 命令**(如 `cargo build`、`cargo test`):在具体的插件目录下执行(如 `plugins/wasm-rust/extensions/my-plugin/`)
|
||||||
- **Docker 命令**:在 `plugins/wasm-rust/` 目录下执行,需要指定 `PLUGIN_NAME` 参数
|
- **Docker 命令**:在 `plugins/wasm-rust/` 目录下执行,需要指定 `PLUGIN_NAME` 参数
|
||||||
|
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
use higress_wasm_rust::cluster_wrapper::{DnsCluster, StaticIpCluster};
|
use higress_wasm_rust::cluster_wrapper::{DnsCluster, FQDNCluster};
|
||||||
use higress_wasm_rust::log::Log;
|
use higress_wasm_rust::log::Log;
|
||||||
use higress_wasm_rust::plugin_wrapper::{HttpContextWrapper, RootContextWrapper};
|
use higress_wasm_rust::plugin_wrapper::{HttpContextWrapper, RootContextWrapper};
|
||||||
use higress_wasm_rust::redis_wrapper::{RedisClient, RedisClientBuilder, RedisClientConfig};
|
use higress_wasm_rust::redis_wrapper::{RedisClient, RedisClientConfig};
|
||||||
use higress_wasm_rust::rule_matcher::{on_configure, RuleMatcher, SharedRuleMatcher};
|
use higress_wasm_rust::rule_matcher::{on_configure, RuleMatcher, SharedRuleMatcher};
|
||||||
use http::Method;
|
use http::Method;
|
||||||
use multimap::MultiMap;
|
use multimap::MultiMap;
|
||||||
@@ -23,12 +23,45 @@ proxy_wasm::main! {{
|
|||||||
|
|
||||||
const PLUGIN_NAME: &str = "demo-wasm";
|
const PLUGIN_NAME: &str = "demo-wasm";
|
||||||
|
|
||||||
|
#[derive(Debug, Deserialize, Clone)]
|
||||||
|
struct RedisConfig {
|
||||||
|
service_name: String,
|
||||||
|
#[serde(default)]
|
||||||
|
service_port: Option<u16>,
|
||||||
|
#[serde(default)]
|
||||||
|
username: Option<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
password: Option<String>,
|
||||||
|
#[serde(default = "default_redis_timeout")]
|
||||||
|
timeout: u64, // 超时(默认1000ms)
|
||||||
|
#[serde(default)]
|
||||||
|
database: Option<u32>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for RedisConfig {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
service_name: String::new(),
|
||||||
|
service_port: None,
|
||||||
|
username: None,
|
||||||
|
password: None,
|
||||||
|
timeout: default_redis_timeout(),
|
||||||
|
database: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 超时时间默认值
|
||||||
|
fn default_redis_timeout() -> u64 {
|
||||||
|
1000
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Default, Debug, Deserialize, Clone)]
|
#[derive(Default, Debug, Deserialize, Clone)]
|
||||||
struct DemoWasmConfig {
|
struct DemoWasmConfig {
|
||||||
// 配置文件结构体
|
// 配置文件结构体
|
||||||
test: String,
|
test: String,
|
||||||
#[serde(default)]
|
#[serde(rename = "redis")]
|
||||||
password: Option<String>,
|
redis_config: RedisConfig,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn format_body(body: Option<Vec<u8>>) -> String {
|
fn format_body(body: Option<Vec<u8>>) -> String {
|
||||||
@@ -74,50 +107,49 @@ impl HttpContextWrapper<DemoWasmConfig> for DemoWasm {
|
|||||||
// 请求header获取完成回调
|
// 请求header获取完成回调
|
||||||
self.log
|
self.log
|
||||||
.info(&format!("on_http_request_complete_headers {:?}", headers));
|
.info(&format!("on_http_request_complete_headers {:?}", headers));
|
||||||
if let Some(config) = &self.config {
|
// 验证配置是否存在
|
||||||
let _redis_client = RedisClientBuilder::new(
|
let config = match &self.config {
|
||||||
&StaticIpCluster::new("redis", 80, ""),
|
Some(cfg) => cfg,
|
||||||
Duration::from_secs(5),
|
None => {
|
||||||
)
|
self.log.error("Plugin configuration is missing");
|
||||||
.password(config.password.as_ref())
|
return HeaderAction::Continue;
|
||||||
.build();
|
}
|
||||||
|
};
|
||||||
let redis_client = RedisClient::new(
|
// 初始化redis client
|
||||||
RedisClientConfig::new(
|
let redis_client = match self.initialize_redis_client(config) {
|
||||||
&StaticIpCluster::new("redis", 80, ""),
|
Ok(client) => client,
|
||||||
Duration::from_secs(5),
|
Err(err) => {
|
||||||
)
|
self.log
|
||||||
.password(config.password.as_ref()),
|
.error(&format!("Failed to initialize Redis client: {}", err));
|
||||||
);
|
return HeaderAction::Continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
self.redis_client = Some(redis_client);
|
||||||
|
// 调用redis
|
||||||
|
if let Some(redis_client) = &self.redis_client {
|
||||||
if let Some(self_rc) = self.weak.upgrade() {
|
if let Some(self_rc) = self.weak.upgrade() {
|
||||||
let init_res = redis_client.init();
|
let incr_res = redis_client.incr(
|
||||||
self.log.info(&format!("redis init {:?}", init_res));
|
"connect",
|
||||||
if init_res.is_ok() {
|
Box::new(move |res, status, token_id| {
|
||||||
let incr_res = redis_client.incr(
|
self_rc.borrow().log().info(&format!(
|
||||||
"connect",
|
"redis incr finish value_res:{:?}, status: {}, token_id: {}",
|
||||||
Box::new(move |res, status, token_id| {
|
res, status, token_id
|
||||||
self_rc.borrow().log().info(&format!(
|
));
|
||||||
"redis incr finish value_res:{:?}, status: {}, token_id: {}",
|
if let Some(this) = self_rc.borrow_mut().downcast_mut::<DemoWasm>() {
|
||||||
res, status, token_id
|
if let Ok(Value::Int(value)) = res {
|
||||||
));
|
this.cid = *value;
|
||||||
if let Some(this) = self_rc.borrow_mut().downcast_mut::<DemoWasm>() {
|
|
||||||
if let Ok(Value::Int(value)) = res {
|
|
||||||
this.cid = *value;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
self_rc.borrow().resume_http_request();
|
|
||||||
}),
|
|
||||||
);
|
|
||||||
match incr_res {
|
|
||||||
Ok(s) => {
|
|
||||||
self.log.info(&format!("redis incr ok {}", s));
|
|
||||||
return HeaderAction::StopAllIterationAndBuffer;
|
|
||||||
}
|
}
|
||||||
Err(e) => self.log.info(&format!("redis incr error {:?}", e)),
|
self_rc.borrow().resume_http_request();
|
||||||
};
|
}),
|
||||||
}
|
);
|
||||||
self.redis_client = Some(redis_client);
|
match incr_res {
|
||||||
|
Ok(s) => {
|
||||||
|
self.log.info(&format!("redis incr ok {}", s));
|
||||||
|
return HeaderAction::StopAllIterationAndBuffer;
|
||||||
|
}
|
||||||
|
Err(e) => self.log.info(&format!("redis incr error {:?}", e)),
|
||||||
|
};
|
||||||
} else {
|
} else {
|
||||||
self.log.error("self_weak upgrade error");
|
self.log.error("self_weak upgrade error");
|
||||||
}
|
}
|
||||||
@@ -233,6 +265,48 @@ impl HttpContextWrapper<DemoWasmConfig> for DemoWasm {
|
|||||||
DataAction::Continue
|
DataAction::Continue
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl DemoWasm {
|
||||||
|
fn initialize_redis_client(&self, config: &DemoWasmConfig) -> Result<RedisClient, String> {
|
||||||
|
let redis_cfg = &config.redis_config;
|
||||||
|
|
||||||
|
if redis_cfg.service_name.is_empty() {
|
||||||
|
return Err("Redis service_name is required but missing".to_string());
|
||||||
|
}
|
||||||
|
|
||||||
|
let service_port = redis_cfg.service_port.unwrap_or_else(|| {
|
||||||
|
if redis_cfg.service_name.ends_with(".static") {
|
||||||
|
80
|
||||||
|
} else {
|
||||||
|
6379
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
let cluster = FQDNCluster::new(&redis_cfg.service_name, "", service_port);
|
||||||
|
let timeout = Duration::from_millis(redis_cfg.timeout);
|
||||||
|
let mut client_config = RedisClientConfig::new(&cluster, timeout);
|
||||||
|
if let Some(username) = &redis_cfg.username {
|
||||||
|
client_config.username(Some(username));
|
||||||
|
}
|
||||||
|
if let Some(password) = &redis_cfg.password {
|
||||||
|
client_config.password(Some(password));
|
||||||
|
}
|
||||||
|
if let Some(database) = redis_cfg.database {
|
||||||
|
client_config.database(Some(database));
|
||||||
|
}
|
||||||
|
let redis_client = RedisClient::new(&client_config);
|
||||||
|
|
||||||
|
// 初始化redis client
|
||||||
|
match redis_client.init() {
|
||||||
|
Ok(_) => {
|
||||||
|
self.log.info("Redis client initialized successfully");
|
||||||
|
Ok(redis_client)
|
||||||
|
}
|
||||||
|
Err(e) => Err(format!("Failed to initialize Redis client: {:?}", e)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
struct DemoWasmRoot {
|
struct DemoWasmRoot {
|
||||||
log: Log,
|
log: Log,
|
||||||
rule_matcher: SharedRuleMatcher<DemoWasmConfig>,
|
rule_matcher: SharedRuleMatcher<DemoWasmConfig>,
|
||||||
|
|||||||
@@ -1,10 +1,9 @@
|
|||||||
use std::{collections::HashMap, time::Duration};
|
use std::{collections::HashMap, time::Duration};
|
||||||
|
|
||||||
|
use crate::{cluster_wrapper::Cluster, internal};
|
||||||
use proxy_wasm::{hostcalls::RedisCallbackFn, types::Status};
|
use proxy_wasm::{hostcalls::RedisCallbackFn, types::Status};
|
||||||
use redis::{Cmd, ToRedisArgs, Value};
|
use redis::{Cmd, ToRedisArgs, Value};
|
||||||
|
|
||||||
use crate::{cluster_wrapper::Cluster, internal};
|
|
||||||
|
|
||||||
pub type RedisValueCallbackFn = dyn FnOnce(&Result<Value, String>, usize, u32);
|
pub type RedisValueCallbackFn = dyn FnOnce(&Result<Value, String>, usize, u32);
|
||||||
|
|
||||||
fn gen_callback(call_fn: Box<RedisValueCallbackFn>) -> Box<RedisCallbackFn> {
|
fn gen_callback(call_fn: Box<RedisValueCallbackFn>) -> Box<RedisCallbackFn> {
|
||||||
@@ -25,6 +24,7 @@ pub struct RedisClientBuilder {
|
|||||||
username: Option<String>,
|
username: Option<String>,
|
||||||
password: Option<String>,
|
password: Option<String>,
|
||||||
timeout: Duration,
|
timeout: Duration,
|
||||||
|
database: Option<u32>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RedisClientBuilder {
|
impl RedisClientBuilder {
|
||||||
@@ -34,6 +34,7 @@ impl RedisClientBuilder {
|
|||||||
username: None,
|
username: None,
|
||||||
password: None,
|
password: None,
|
||||||
timeout,
|
timeout,
|
||||||
|
database: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -47,9 +48,24 @@ impl RedisClientBuilder {
|
|||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn database(mut self, database: Option<u32>) -> Self {
|
||||||
|
self.database = database;
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
pub fn build(self) -> RedisClient {
|
pub fn build(self) -> RedisClient {
|
||||||
|
let upstream = if let Some(db) = self.database {
|
||||||
|
if db != 0 {
|
||||||
|
format!("{}?db={}", self.upstream, db)
|
||||||
|
} else {
|
||||||
|
self.upstream
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
self.upstream
|
||||||
|
};
|
||||||
|
|
||||||
RedisClient {
|
RedisClient {
|
||||||
upstream: self.upstream,
|
upstream,
|
||||||
username: self.username,
|
username: self.username,
|
||||||
password: self.password,
|
password: self.password,
|
||||||
timeout: self.timeout,
|
timeout: self.timeout,
|
||||||
@@ -62,6 +78,7 @@ pub struct RedisClientConfig {
|
|||||||
username: Option<String>,
|
username: Option<String>,
|
||||||
password: Option<String>,
|
password: Option<String>,
|
||||||
timeout: Duration,
|
timeout: Duration,
|
||||||
|
database: Option<u32>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RedisClientConfig {
|
impl RedisClientConfig {
|
||||||
@@ -71,6 +88,7 @@ impl RedisClientConfig {
|
|||||||
username: None,
|
username: None,
|
||||||
password: None,
|
password: None,
|
||||||
timeout,
|
timeout,
|
||||||
|
database: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -83,6 +101,11 @@ impl RedisClientConfig {
|
|||||||
self.password = password.map(|p| p.as_ref().to_string());
|
self.password = password.map(|p| p.as_ref().to_string());
|
||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn database(&mut self, database: Option<u32>) -> &Self {
|
||||||
|
self.database = database;
|
||||||
|
self
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
@@ -95,8 +118,18 @@ pub struct RedisClient {
|
|||||||
|
|
||||||
impl RedisClient {
|
impl RedisClient {
|
||||||
pub fn new(config: &RedisClientConfig) -> Self {
|
pub fn new(config: &RedisClientConfig) -> Self {
|
||||||
|
let upstream = if let Some(db) = config.database {
|
||||||
|
if db != 0 {
|
||||||
|
format!("{}?db={}", config.upstream, db)
|
||||||
|
} else {
|
||||||
|
config.upstream.clone()
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
config.upstream.clone()
|
||||||
|
};
|
||||||
|
|
||||||
RedisClient {
|
RedisClient {
|
||||||
upstream: config.upstream.clone(),
|
upstream,
|
||||||
username: config.username.clone(),
|
username: config.username.clone(),
|
||||||
password: config.password.clone(),
|
password: config.password.clone(),
|
||||||
timeout: config.timeout,
|
timeout: config.timeout,
|
||||||
|
|||||||
Reference in New Issue
Block a user