feat(nacos): add Nacos management console (#1456)

This commit is contained in:
二丫讲梵 authored and GitHub committed 2026-06-24 01:19:18 +08:00
1 parent ff1559f018
commit 3ebfae05a2
62 files changed
+7314 -55

No files matched your search

+71 -2
View File
@@ -80,6 +80,8 @@ pub enum PoolKind {
/// Message queue admin connection (not a data query pool; serves as a
/// marker that this connection_id is a valid MQ admin connection).
MessageQueue,
/// Nacos admin connection marker.
Nacos,
}
macro_rules! agent_connection_pool_database_type {
@@ -131,6 +133,7 @@ pub struct AppState {
pub storage: Storage,
pub plugins: PluginRegistry,
pub agent_manager: crate::agent_manager::AgentManager,
pub nacos_registry: crate::nacos::NacosAdminRegistry,
#[cfg(feature = "mq-admin")]
pub mq_registry: crate::mq::MqAdminRegistry,
}
@@ -358,6 +361,7 @@ impl AppState {
agent_dir,
app_version,
),
nacos_registry: crate::nacos::NacosAdminRegistry::new(),
#[cfg(feature = "mq-admin")]
mq_registry: crate::mq::MqAdminRegistry::new(),
}
@@ -781,6 +785,12 @@ impl AppState {
db::influxdb_driver::test_connection(&client, connect_timeout).await?;
PoolKind::InfluxDb(client)
}
DatabaseType::Nacos => {
let admin_config = self.nacos_admin_config_for_connection(connection_id, &config).await?;
let adapter = self.nacos_registry.build_transient_config(admin_config).await?;
adapter.test_connection().await?;
PoolKind::Nacos
}
agent_connection_pool_database_type!() => {
let connect_params =
agent_connect_params(&db_config, &host, port, db_config.effective_database().unwrap_or(""));
@@ -1014,6 +1024,20 @@ impl AppState {
Ok(mqc.with_connect_override(&host, port))
}
pub async fn nacos_admin_config_for_connection(
&self,
connection_id: &str,
config: &ConnectionConfig,
) -> Result<crate::nacos::config::NacosAdminConfig, String> {
let nacos_config = crate::nacos::config::NacosAdminConfig::from_connection(config)?;
if !config.has_effective_transport_layers() {
return Ok(nacos_config);
}
let (host, port) = self.connection_host_port(connection_id, config).await?;
Ok(nacos_config.with_connect_override(&host, port))
}
async fn remove_stale_connection_pool(&self, pool_key: &str) -> bool {
if self.running_queries.is_pool_active(pool_key) {
return false;
@@ -1191,7 +1215,8 @@ impl AppState {
| PoolKind::DuckDb(_)
| PoolKind::ExternalTabular(_)
| PoolKind::ExternalDriver { .. }
| PoolKind::MessageQueue => false,
| PoolKind::MessageQueue
| PoolKind::Nacos => false,
}
};
@@ -1570,7 +1595,8 @@ impl AppState {
| PoolKind::DuckDb(_)
| PoolKind::ExternalTabular(_)
| PoolKind::ExternalDriver { .. }
| PoolKind::MessageQueue => true,
| PoolKind::MessageQueue
| PoolKind::Nacos => true,
PoolKind::Redis(_) => unreachable!("Redis handled separately"),
};
if !healthy {
@@ -1795,6 +1821,8 @@ fn connection_remote_endpoint(config: &ConnectionConfig) -> (String, u16) {
.unwrap_or_else(|| (config.host.clone(), config.port))
} else if config.db_type == DatabaseType::MessageQueue {
parse_mq_admin_host_port(config).unwrap_or_else(|| (config.host.clone(), config.port))
} else if config.db_type == DatabaseType::Nacos {
parse_nacos_server_host_port(config).unwrap_or_else(|| (config.host.clone(), config.port))
} else {
(config.host.clone(), config.port)
}
@@ -1817,6 +1845,23 @@ fn parse_mq_admin_host_port(config: &ConnectionConfig) -> Option<(String, u16)>
Some((host, port))
}
fn parse_nacos_server_host_port(config: &ConnectionConfig) -> Option<(String, u16)> {
let value = config
.external_config
.as_ref()?
.get("serverAddr")
.or_else(|| config.external_config.as_ref()?.get("server_addr"))?
.as_str()?
.trim();
if value.is_empty() {
return None;
}
let url = reqwest::Url::parse(value).ok()?;
let host = url.host_str()?.to_string();
let port = url.port_or_known_default()?;
Some((host, port))
}
fn normalize_client_session_id(client_session_id: Option<&str>) -> Option<String> {
client_session_id.map(str::trim).filter(|session| !session.is_empty()).map(|session| session.replace(':', "_"))
}
@@ -1891,6 +1936,7 @@ fn clone_pool_kind(pool: &PoolKind) -> PoolKind {
PoolKind::ExternalDriver { driver_id: driver_id.clone(), config: config.clone(), session: session.clone() }
}
PoolKind::MessageQueue => PoolKind::MessageQueue,
PoolKind::Nacos => PoolKind::Nacos,
PoolKind::Redis(_) => panic!("clone_pool_kind not supported for Redis — handled separately"),
}
}
@@ -1940,6 +1986,7 @@ pub async fn close_pool_kind(pool: PoolKind) {
session.shutdown().await;
}
PoolKind::MessageQueue => {}
PoolKind::Nacos => {}
}
}
@@ -3242,6 +3289,28 @@ mod tests {
let _ = std::fs::remove_dir_all(dir);
}
#[tokio::test]
async fn nacos_admin_config_allows_domain_server_addr_without_transport_override() {
let (state, dir) = test_app_state().await;
let mut config = mysql_config(None);
config.id = "aliyun-nacos".to_string();
config.db_type = DatabaseType::Nacos;
config.host = "example.com".to_string();
config.port = 8848;
config.external_config = Some(serde_json::json!({
"serverAddr": "https://nacos.aliyuncs.com:8848",
"namespace": "public",
"contextPath": "/nacos",
"auth": { "kind": "none" }
}));
let nacos_config = state.nacos_admin_config_for_connection("aliyun-nacos", &config).await.unwrap();
assert_eq!(nacos_config.server_addr, "https://nacos.aliyuncs.com:8848");
assert!(nacos_config.connect_override.is_none());
let _ = std::fs::remove_dir_all(dir);
}
#[tokio::test]
#[ignore = "requires a reachable GaussDB instance via environment variables"]
async fn live_gaussdb_native_connection_succeeds() {
+1
View File
@@ -29,6 +29,7 @@ pub mod models;
pub mod mongo_ops;
#[cfg(feature = "mq-admin")]
pub mod mq;
pub mod nacos;
pub mod object_source_sql;
pub mod path_utils;
pub mod plugins;
+14
View File
@@ -307,6 +307,7 @@ pub enum DatabaseType {
Xugu,
Iotdb,
Etcd,
Nacos,
#[serde(rename = "iris")]
Iris,
#[serde(rename = "turso")]
@@ -793,6 +794,7 @@ impl ConnectionConfig {
}
DatabaseType::Jdbc => "jdbc:<redacted>".to_string(),
DatabaseType::MessageQueue => self.message_queue_admin_url(),
DatabaseType::Nacos => self.nacos_admin_url(),
}
}
@@ -998,6 +1000,7 @@ impl ConnectionConfig {
self.connection_string.as_deref().filter(|value| !value.is_empty()).unwrap_or("jdbc:").to_string()
}
DatabaseType::MessageQueue => self.message_queue_admin_url(),
DatabaseType::Nacos => self.nacos_admin_url(),
}
}
@@ -1012,6 +1015,17 @@ impl ConnectionConfig {
.to_string()
}
fn nacos_admin_url(&self) -> String {
self.external_config
.as_ref()
.and_then(|value| value.get("serverAddr").or_else(|| value.get("server_addr")))
.and_then(|value| value.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("nacos://")
.to_string()
}
fn normalized_url_params(&self) -> String {
let value = self.url_params.as_deref().unwrap_or("").trim();
if self.needs_bare_mysql() {
+193
View File
@@ -0,0 +1,193 @@
use serde::{Deserialize, Serialize};
use crate::models::connection::ConnectionConfig;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "camelCase")]
pub enum NacosAuthConfig {
None,
UsernamePassword { username: String, password: String },
}
impl Default for NacosAuthConfig {
fn default() -> Self {
Self::None
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosAdminConfig {
pub server_addr: String,
#[serde(default)]
pub namespace: String,
#[serde(default)]
pub context_path: String,
#[serde(default)]
pub auth: NacosAuthConfig,
#[serde(default)]
pub tls_skip_verify: bool,
#[serde(default = "default_page_size")]
pub page_size: u32,
#[serde(skip)]
pub connect_override: Option<NacosConnectOverride>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct NacosConnectOverride {
pub host: String,
pub port: u16,
}
pub fn default_page_size() -> u32 {
20
}
impl NacosAdminConfig {
pub fn from_connection(cfg: &ConnectionConfig) -> Result<Self, String> {
let parsed = if let Some(raw) = cfg.external_config.as_ref() {
serde_json::from_value::<NacosAdminConfig>(raw.clone())
.map_err(|e| format!("Failed to parse Nacos admin config: {e}"))?
} else {
let scheme = if cfg.ssl { "https" } else { "http" };
NacosAdminConfig {
server_addr: format!("{scheme}://{}:{}", cfg.host.trim(), cfg.port),
namespace: cfg.database.clone().unwrap_or_default(),
context_path: String::new(),
auth: if cfg.username.trim().is_empty() {
NacosAuthConfig::None
} else {
NacosAuthConfig::UsernamePassword { username: cfg.username.clone(), password: cfg.password.clone() }
},
tls_skip_verify: false,
page_size: default_page_size(),
connect_override: None,
}
};
parsed.validate()
}
pub fn validate(mut self) -> Result<Self, String> {
self.server_addr = self.server_addr.trim().trim_end_matches('/').to_string();
if self.server_addr.is_empty() {
return Err("Nacos server address is empty".to_string());
}
self.context_path = normalize_context_path(&self.context_path);
if self.page_size == 0 {
self.page_size = default_page_size();
}
self.page_size = self.page_size.clamp(1, 500);
Ok(self)
}
pub fn with_connect_override(mut self, host: &str, port: u16) -> Self {
self.connect_override = Some(NacosConnectOverride { host: host.to_string(), port });
self
}
}
pub fn normalize_context_path(path: &str) -> String {
let trimmed = path.trim().trim_end_matches('/');
if trimmed.is_empty() {
return String::new();
}
if trimmed.starts_with('/') {
trimmed.to_string()
} else {
format!("/{trimmed}")
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::models::connection::{default_keepalive_interval_secs, DatabaseType};
fn connection_with_external(value: serde_json::Value) -> ConnectionConfig {
ConnectionConfig {
id: "nacos-1".to_string(),
name: "Nacos".to_string(),
db_type: DatabaseType::Nacos,
driver_profile: None,
driver_label: None,
url_params: None,
host: "127.0.0.1".to_string(),
port: 8848,
username: String::new(),
password: String::new(),
database: None,
visible_databases: None,
attached_databases: Vec::new(),
color: None,
transport_layers: Vec::new(),
connect_timeout_secs: 5,
query_timeout_secs: 30,
idle_timeout_secs: 60,
keepalive_interval_secs: default_keepalive_interval_secs(),
ssl: false,
ca_cert_path: String::new(),
client_cert_path: String::new(),
client_key_path: String::new(),
sysdba: false,
oracle_connection_type: None,
connection_string: None,
redis_connection_mode: None,
redis_sentinel_master: String::new(),
redis_sentinel_nodes: String::new(),
redis_sentinel_username: String::new(),
redis_sentinel_password: String::new(),
redis_sentinel_tls: false,
redis_cluster_nodes: String::new(),
redis_key_separator: ":".to_string(),
etcd_endpoints: String::new(),
gbase_server: String::new(),
informix_server: String::new(),
external_config: Some(value),
jdbc_driver_class: None,
jdbc_driver_paths: Vec::new(),
one_time: false,
read_only: false,
}
}
#[test]
fn parses_external_config() {
let cfg = connection_with_external(serde_json::json!({
"serverAddr": " http://127.0.0.1:8848/ ",
"namespace": "public",
"contextPath": "nacos",
"pageSize": 100,
"auth": { "kind": "usernamePassword", "username": "nacos", "password": "pw" }
}));
let parsed = NacosAdminConfig::from_connection(&cfg).unwrap();
assert_eq!(parsed.server_addr, "http://127.0.0.1:8848");
assert_eq!(parsed.context_path, "/nacos");
assert_eq!(parsed.page_size, 100);
assert_eq!(parsed.namespace, "public");
}
#[test]
fn missing_external_context_path_defaults_to_root() {
let cfg = connection_with_external(serde_json::json!({
"serverAddr": "http://127.0.0.1:8848",
"auth": { "kind": "none" }
}));
let parsed = NacosAdminConfig::from_connection(&cfg).unwrap();
assert_eq!(parsed.context_path, "");
}
#[test]
fn falls_back_to_connection_fields() {
let mut cfg = connection_with_external(serde_json::Value::Null);
cfg.external_config = None;
cfg.username = "nacos".to_string();
cfg.password = "pw".to_string();
let parsed = NacosAdminConfig::from_connection(&cfg).unwrap();
assert_eq!(parsed.server_addr, "http://127.0.0.1:8848");
assert_eq!(parsed.context_path, "");
assert!(matches!(parsed.auth, NacosAuthConfig::UsernamePassword { .. }));
}
}
File diff suppressed because it is too large. Load diff
+85
View File
@@ -0,0 +1,85 @@
//! Nacos admin console support.
//!
//! The public API is intentionally owned by dbx (`NacosAdmin`) instead of
//! exposing SDK/OpenAPI types. The current adapter uses Nacos OpenAPI because
//! the current nacos-sdk-rust releases require Rust edition 2024, while dbx
//! still supports an older Rust toolchain. A future SDK adapter can implement
//! the same port without changing commands, routes, or frontend contracts.
pub mod config;
pub mod http;
pub mod port;
pub mod service;
pub mod types;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};
use crate::models::connection::ConnectionConfig;
use crate::nacos::config::NacosAdminConfig;
use crate::nacos::http::NacosOpenApiAdmin;
use crate::nacos::port::NacosAdmin;
pub use crate::nacos::config::{NacosAdminConfig as NacosConfig, NacosAuthConfig};
pub use crate::nacos::types::*;
#[derive(Default)]
pub struct NacosAdminRegistry {
instances: RwLock<HashMap<String, Arc<dyn NacosAdmin>>>,
build_locks: RwLock<HashMap<String, Arc<Mutex<()>>>>,
}
impl NacosAdminRegistry {
pub fn new() -> Self {
Self { instances: RwLock::new(HashMap::new()), build_locks: RwLock::new(HashMap::new()) }
}
pub async fn get_or_build(&self, cfg: &ConnectionConfig) -> Result<Arc<dyn NacosAdmin>, String> {
let admin_config = NacosAdminConfig::from_connection(cfg)?;
self.get_or_build_config(&cfg.id, admin_config).await
}
pub async fn get_or_build_config(
&self,
connection_id: &str,
cfg: NacosAdminConfig,
) -> Result<Arc<dyn NacosAdmin>, String> {
if let Some(admin) = self.instances.read().await.get(connection_id) {
return Ok(admin.clone());
}
let lock = {
let mut locks = self.build_locks.write().await;
locks.entry(connection_id.to_string()).or_insert_with(|| Arc::new(Mutex::new(()))).clone()
};
let _guard = lock.lock().await;
if let Some(admin) = self.instances.read().await.get(connection_id) {
return Ok(admin.clone());
}
let admin = build_admin(cfg)?;
self.instances.write().await.insert(connection_id.to_string(), admin.clone());
Ok(admin)
}
pub async fn build_transient(&self, cfg: &ConnectionConfig) -> Result<Arc<dyn NacosAdmin>, String> {
let admin_config = NacosAdminConfig::from_connection(cfg)?;
self.build_transient_config(admin_config).await
}
pub async fn build_transient_config(&self, cfg: NacosAdminConfig) -> Result<Arc<dyn NacosAdmin>, String> {
build_admin(cfg)
}
pub async fn drop_connection(&self, connection_id: &str) {
self.instances.write().await.remove(connection_id);
self.build_locks.write().await.remove(connection_id);
}
}
fn build_admin(cfg: NacosAdminConfig) -> Result<Arc<dyn NacosAdmin>, String> {
Ok(Arc::new(NacosOpenApiAdmin::new(cfg)?))
}
+22
View File
@@ -0,0 +1,22 @@
use async_trait::async_trait;
use crate::nacos::types::*;
#[async_trait]
pub trait NacosAdmin: Send + Sync {
async fn test_connection(&self) -> Result<NacosConnectionInfo, String>;
async fn list_namespaces(&self) -> Result<Vec<NacosNamespaceInfo>, String>;
async fn create_namespace(&self, req: NacosNamespaceCreate) -> Result<(), String>;
async fn update_namespace(&self, req: NacosNamespaceUpdate) -> Result<(), String>;
async fn list_configs(&self, query: NacosConfigQuery) -> Result<NacosConfigList, String>;
async fn get_config(&self, key: NacosConfigKey) -> Result<NacosConfigItem, String>;
async fn publish_config(&self, req: NacosConfigUpsert) -> Result<(), String>;
async fn delete_config(&self, key: NacosConfigKey) -> Result<(), String>;
async fn list_config_history(&self, query: NacosConfigHistoryQuery) -> Result<NacosConfigHistoryList, String>;
async fn get_config_history(&self, key: NacosConfigHistoryKey) -> Result<NacosConfigItem, String>;
async fn rollback_config(&self, req: NacosConfigRollbackRequest) -> Result<(), String>;
async fn list_services(&self, query: NacosServiceQuery) -> Result<NacosServiceList, String>;
async fn list_instances(&self, query: NacosInstanceQuery) -> Result<Vec<NacosInstanceInfo>, String>;
async fn update_instance(&self, req: NacosInstanceUpdate) -> Result<(), String>;
async fn raw_request(&self, req: NacosRawRequest) -> Result<NacosRawResponse, String>;
}
+287
View File
@@ -0,0 +1,287 @@
use crate::connection::AppState;
use crate::models::connection::DatabaseType;
use crate::nacos::types::*;
pub async fn nacos_test_connection_core(state: &AppState, conn_id: &str) -> Result<NacosConnectionInfo, String> {
let cfg = state.configs.read().await.get(conn_id).cloned().ok_or("Connection not found")?;
let admin = state.nacos_registry.build_transient(&cfg).await?;
admin.test_connection().await
}
pub async fn nacos_list_namespaces_core(state: &AppState, conn_id: &str) -> Result<Vec<NacosNamespaceInfo>, String> {
let admin = get_admin(state, conn_id).await?;
admin.list_namespaces().await
}
pub async fn nacos_create_namespace_core(
state: &AppState,
conn_id: &str,
req: NacosNamespaceCreate,
) -> Result<(), String> {
ensure_connection_writable(state, conn_id, "Create Nacos namespace").await?;
let admin = get_admin(state, conn_id).await?;
admin.create_namespace(req).await
}
pub async fn nacos_update_namespace_core(
state: &AppState,
conn_id: &str,
req: NacosNamespaceUpdate,
) -> Result<(), String> {
ensure_connection_writable(state, conn_id, "Update Nacos namespace").await?;
let admin = get_admin(state, conn_id).await?;
admin.update_namespace(req).await
}
pub async fn nacos_list_configs_core(
state: &AppState,
conn_id: &str,
query: NacosConfigQuery,
) -> Result<NacosConfigList, String> {
let admin = get_admin(state, conn_id).await?;
admin.list_configs(query).await
}
pub async fn nacos_get_config_core(
state: &AppState,
conn_id: &str,
key: NacosConfigKey,
) -> Result<NacosConfigItem, String> {
let admin = get_admin(state, conn_id).await?;
admin.get_config(key).await
}
pub async fn nacos_publish_config_core(state: &AppState, conn_id: &str, req: NacosConfigUpsert) -> Result<(), String> {
ensure_connection_writable(state, conn_id, "Publish Nacos config").await?;
let admin = get_admin(state, conn_id).await?;
admin.publish_config(req).await
}
pub async fn nacos_delete_config_core(state: &AppState, conn_id: &str, key: NacosConfigKey) -> Result<(), String> {
ensure_connection_writable(state, conn_id, "Delete Nacos config").await?;
let admin = get_admin(state, conn_id).await?;
admin.delete_config(key).await
}
pub async fn nacos_list_config_history_core(
state: &AppState,
conn_id: &str,
query: NacosConfigHistoryQuery,
) -> Result<NacosConfigHistoryList, String> {
let admin = get_admin(state, conn_id).await?;
admin.list_config_history(query).await
}
pub async fn nacos_get_config_history_core(
state: &AppState,
conn_id: &str,
key: NacosConfigHistoryKey,
) -> Result<NacosConfigItem, String> {
let admin = get_admin(state, conn_id).await?;
admin.get_config_history(key).await
}
pub async fn nacos_rollback_config_core(
state: &AppState,
conn_id: &str,
req: NacosConfigRollbackRequest,
) -> Result<(), String> {
ensure_connection_writable(state, conn_id, "Rollback Nacos config").await?;
let admin = get_admin(state, conn_id).await?;
admin.rollback_config(req).await
}
pub async fn nacos_list_services_core(
state: &AppState,
conn_id: &str,
query: NacosServiceQuery,
) -> Result<NacosServiceList, String> {
let admin = get_admin(state, conn_id).await?;
admin.list_services(query).await
}
pub async fn nacos_list_instances_core(
state: &AppState,
conn_id: &str,
query: NacosInstanceQuery,
) -> Result<Vec<NacosInstanceInfo>, String> {
let admin = get_admin(state, conn_id).await?;
admin.list_instances(query).await
}
pub async fn nacos_update_instance_core(
state: &AppState,
conn_id: &str,
req: NacosInstanceUpdate,
) -> Result<(), String> {
ensure_connection_writable(state, conn_id, "Update Nacos instance").await?;
let admin = get_admin(state, conn_id).await?;
admin.update_instance(req).await
}
pub async fn nacos_raw_request_core(
state: &AppState,
conn_id: &str,
req: NacosRawRequest,
) -> Result<NacosRawResponse, String> {
crate::nacos::http::validate_raw_api_path(&req.path)?;
if req.method.to_ascii_uppercase() != "GET" {
ensure_connection_writable(state, conn_id, "Run mutating Nacos raw request").await?;
}
let admin = get_admin(state, conn_id).await?;
admin.raw_request(req).await
}
async fn get_admin(
state: &AppState,
conn_id: &str,
) -> Result<std::sync::Arc<dyn crate::nacos::port::NacosAdmin>, String> {
let cfg = state.configs.read().await.get(conn_id).cloned().ok_or("Connection not found")?;
if cfg.db_type != DatabaseType::Nacos {
return Err("Connection is not a Nacos admin connection".to_string());
}
state.nacos_registry.get_or_build(&cfg).await
}
async fn ensure_connection_writable(state: &AppState, conn_id: &str, action: &str) -> Result<(), String> {
let cfg = state.configs.read().await.get(conn_id).cloned().ok_or("Connection not found")?;
if cfg.read_only {
Err(format!("{action} is blocked because this connection is read-only"))
} else {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn raw_mutation_requires_writable_connection_before_adapter_build() {
let dir = std::env::temp_dir().join(format!("dbx-nacos-service-test-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&dir).unwrap();
let storage = crate::storage::Storage::open(&dir.join("storage.db")).await.unwrap();
let state = AppState::new(storage);
let mut cfg = crate::models::connection::ConnectionConfig {
id: "nacos-1".to_string(),
name: "Nacos".to_string(),
db_type: DatabaseType::Nacos,
driver_profile: None,
driver_label: None,
url_params: None,
host: "127.0.0.1".to_string(),
port: 8848,
username: String::new(),
password: String::new(),
database: None,
visible_databases: None,
attached_databases: Vec::new(),
color: None,
transport_layers: Vec::new(),
connect_timeout_secs: 5,
query_timeout_secs: 30,
idle_timeout_secs: 60,
keepalive_interval_secs: crate::models::connection::default_keepalive_interval_secs(),
ssl: false,
ca_cert_path: String::new(),
client_cert_path: String::new(),
client_key_path: String::new(),
sysdba: false,
oracle_connection_type: None,
connection_string: None,
redis_connection_mode: None,
redis_sentinel_master: String::new(),
redis_sentinel_nodes: String::new(),
redis_sentinel_username: String::new(),
redis_sentinel_password: String::new(),
redis_sentinel_tls: false,
redis_cluster_nodes: String::new(),
redis_key_separator: ":".to_string(),
etcd_endpoints: String::new(),
gbase_server: String::new(),
informix_server: String::new(),
external_config: Some(serde_json::json!({ "serverAddr": "http://127.0.0.1:9" })),
jdbc_driver_class: None,
jdbc_driver_paths: Vec::new(),
one_time: false,
read_only: true,
};
cfg.read_only = true;
state.configs.write().await.insert(cfg.id.clone(), cfg);
let err = nacos_raw_request_core(
&state,
"nacos-1",
NacosRawRequest { method: "POST".to_string(), path: "/v1/cs/configs".to_string(), query: None, body: None },
)
.await
.unwrap_err();
assert!(err.contains("read-only"));
}
#[tokio::test]
async fn config_rollback_requires_writable_connection_before_adapter_build() {
let dir = std::env::temp_dir().join(format!("dbx-nacos-service-test-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&dir).unwrap();
let storage = crate::storage::Storage::open(&dir.join("storage.db")).await.unwrap();
let state = AppState::new(storage);
let cfg = crate::models::connection::ConnectionConfig {
id: "nacos-rollback".to_string(),
name: "Nacos".to_string(),
db_type: DatabaseType::Nacos,
driver_profile: None,
driver_label: None,
url_params: None,
host: "127.0.0.1".to_string(),
port: 8848,
username: String::new(),
password: String::new(),
database: None,
visible_databases: None,
attached_databases: Vec::new(),
color: None,
transport_layers: Vec::new(),
connect_timeout_secs: 5,
query_timeout_secs: 30,
idle_timeout_secs: 60,
keepalive_interval_secs: crate::models::connection::default_keepalive_interval_secs(),
ssl: false,
ca_cert_path: String::new(),
client_cert_path: String::new(),
client_key_path: String::new(),
sysdba: false,
oracle_connection_type: None,
connection_string: None,
redis_connection_mode: None,
redis_sentinel_master: String::new(),
redis_sentinel_nodes: String::new(),
redis_sentinel_username: String::new(),
redis_sentinel_password: String::new(),
redis_sentinel_tls: false,
redis_cluster_nodes: String::new(),
redis_key_separator: ":".to_string(),
etcd_endpoints: String::new(),
gbase_server: String::new(),
informix_server: String::new(),
external_config: Some(serde_json::json!({ "serverAddr": "http://127.0.0.1:9" })),
jdbc_driver_class: None,
jdbc_driver_paths: Vec::new(),
one_time: false,
read_only: true,
};
state.configs.write().await.insert(cfg.id.clone(), cfg);
let err = nacos_rollback_config_core(
&state,
"nacos-rollback",
NacosConfigRollbackRequest {
namespace: None,
data_id: "app.yaml".to_string(),
group: "DEFAULT_GROUP".to_string(),
history_id: "1".to_string(),
nid: None,
},
)
.await
.unwrap_err();
assert!(err.contains("read-only"));
}
}
+334
View File
@@ -0,0 +1,334 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosCapabilities {
pub supports_config_management: bool,
pub supports_config_history: bool,
pub supports_service_management: bool,
pub supports_instance_update: bool,
pub supports_raw_api: bool,
}
impl Default for NacosCapabilities {
fn default() -> Self {
Self {
supports_config_management: true,
supports_config_history: true,
supports_service_management: true,
supports_instance_update: true,
supports_raw_api: true,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConnectionInfo {
pub server_addr: String,
pub namespace: String,
pub server_version: Option<String>,
pub auth: String,
pub capabilities: NacosCapabilities,
#[serde(skip_serializing_if = "Option::is_none")]
pub raw: Option<serde_json::Value>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosNamespaceInfo {
pub namespace: String,
pub namespace_show_name: String,
#[serde(default)]
pub namespace_desc: Option<String>,
#[serde(default)]
pub config_count: Option<u64>,
#[serde(default)]
pub quota: Option<u64>,
#[serde(default)]
pub namespace_type: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosNamespaceCreate {
#[serde(default)]
pub namespace_id: Option<String>,
pub namespace_name: String,
#[serde(default)]
pub namespace_desc: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosNamespaceUpdate {
pub namespace_id: String,
pub namespace_name: String,
#[serde(default)]
pub namespace_desc: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigQuery {
#[serde(default)]
pub namespace: Option<String>,
#[serde(default)]
pub group: Option<String>,
#[serde(default)]
pub data_id: Option<String>,
#[serde(default)]
pub search: Option<String>,
#[serde(default)]
pub page_no: Option<u32>,
#[serde(default)]
pub page_size: Option<u32>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigItem {
pub data_id: String,
pub group: String,
pub namespace: String,
#[serde(default)]
pub app_name: Option<String>,
#[serde(default)]
pub desc: Option<String>,
#[serde(default)]
pub tags: Option<String>,
#[serde(default)]
pub config_type: Option<String>,
#[serde(default)]
pub md5: Option<String>,
#[serde(default)]
pub encrypted_data_key: Option<String>,
#[serde(default)]
pub content: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigList {
pub page_no: u32,
pub page_size: u32,
pub total_count: u64,
pub items: Vec<NacosConfigItem>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigUpsert {
#[serde(default)]
pub namespace: Option<String>,
pub data_id: String,
pub group: String,
pub content: String,
#[serde(default)]
pub config_type: Option<String>,
#[serde(default)]
pub app_name: Option<String>,
#[serde(default)]
pub desc: Option<String>,
#[serde(default)]
pub tags: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigKey {
#[serde(default)]
pub namespace: Option<String>,
pub data_id: String,
pub group: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigHistoryQuery {
#[serde(default)]
pub namespace: Option<String>,
pub data_id: String,
pub group: String,
#[serde(default)]
pub page_no: Option<u32>,
#[serde(default)]
pub page_size: Option<u32>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigHistoryItem {
pub history_id: String,
#[serde(default)]
pub nid: Option<i64>,
pub data_id: String,
pub group: String,
pub namespace: String,
#[serde(default)]
pub app_name: Option<String>,
#[serde(default)]
pub operation: Option<String>,
#[serde(default)]
pub operator: Option<String>,
#[serde(default)]
pub last_modified_time: Option<String>,
#[serde(default)]
pub config_type: Option<String>,
#[serde(default)]
pub tags: Option<String>,
#[serde(default)]
pub md5: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigHistoryList {
pub page_no: u32,
pub page_size: u32,
pub total_count: u64,
pub items: Vec<NacosConfigHistoryItem>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigHistoryKey {
#[serde(default)]
pub namespace: Option<String>,
pub data_id: String,
pub group: String,
pub history_id: String,
#[serde(default)]
pub nid: Option<i64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosConfigRollbackRequest {
#[serde(default)]
pub namespace: Option<String>,
pub data_id: String,
pub group: String,
pub history_id: String,
#[serde(default)]
pub nid: Option<i64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NacosServiceQuery {
#[serde(default)]
pub namespace: Option<String>,
#[serde(default)]
pub group_name: Option<String>,
#[serde(default)]
pub service_name: Option<String>,
#[serde(default)]
pub page_no: Option<u32>,
#[serde(default)]
pub page_size: Option<u32>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosServiceInfo {
pub service_name: String,
#[serde(default)]
pub group_name: Option<String>,
#[serde(default)]
pub cluster_count: Option<u64>,
#[serde(default)]
pub ip_count: Option<u64>,
#[serde(default)]
pub healthy_instance_count: Option<u64>,
#[serde(default)]
pub trigger_flag: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosServiceList {
pub page_no: u32,
pub page_size: u32,
pub total_count: u64,
pub items: Vec<NacosServiceInfo>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosInstanceInfo {
pub ip: String,
pub port: u16,
#[serde(default)]
pub service_name: Option<String>,
#[serde(default)]
pub cluster_name: Option<String>,
#[serde(default)]
pub group_name: Option<String>,
#[serde(default)]
pub healthy: Option<bool>,
#[serde(default)]
pub enabled: Option<bool>,
#[serde(default)]
pub ephemeral: Option<bool>,
#[serde(default)]
pub weight: Option<f64>,
#[serde(default)]
pub metadata: serde_json::Value,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosInstanceQuery {
#[serde(default)]
pub namespace: Option<String>,
pub service_name: String,
#[serde(default)]
pub group_name: Option<String>,
#[serde(default)]
pub clusters: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosInstanceUpdate {
#[serde(default)]
pub namespace: Option<String>,
pub service_name: String,
pub ip: String,
pub port: u16,
#[serde(default)]
pub group_name: Option<String>,
#[serde(default)]
pub cluster_name: Option<String>,
#[serde(default)]
pub healthy: Option<bool>,
#[serde(default)]
pub enabled: Option<bool>,
#[serde(default)]
pub ephemeral: Option<bool>,
#[serde(default)]
pub weight: Option<f64>,
#[serde(default)]
pub metadata: Option<serde_json::Value>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosRawRequest {
pub method: String,
pub path: String,
#[serde(default)]
pub query: Option<std::collections::HashMap<String, String>>,
#[serde(default)]
pub body: Option<serde_json::Value>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NacosRawResponse {
pub status: u16,
pub body: serde_json::Value,
#[serde(default)]
pub text: Option<String>,
}
+2 -1
View File
@@ -1005,6 +1005,7 @@ pub async fn do_execute(
PoolKind::Redis(_) => Err("Use Redis-specific commands".to_string()),
PoolKind::MongoDb(_) => Err("Use MongoDB-specific commands".to_string()),
PoolKind::MessageQueue => Err("Use Message Queue-specific commands".to_string()),
PoolKind::Nacos => Err("Use Nacos-specific commands".to_string()),
PoolKind::InfluxDb(client) => {
let client = client.clone();
let database = pool_key.split(':').nth(1).unwrap_or("default").to_string();
@@ -1757,7 +1758,7 @@ pub async fn execute_statements_in_transaction(
| PoolKind::Turso(_)
| PoolKind::SqlServer(_)
| PoolKind::Agent(_) => TxPath::Explicit,
PoolKind::MessageQueue => TxPath::None,
PoolKind::MessageQueue | PoolKind::Nacos => TxPath::None,
#[cfg(feature = "duckdb-bundled")]
PoolKind::DuckDb(_)
| PoolKind::Redis(_)