github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-15 14:52:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(db): 支持 VictoriaMetrics 数据库

Browse Source 
* feat(victoriametrics): support native MetricsQL queries

* fix(victoriametrics): 完善指标浏览和结果导出

* fix(victoriametrics): 修复指标对象通用操作入口

* fix(victoriametrics): 修正指标注释和截断提示

* fix(i18n): 补齐 VictoriaMetrics 连接配置翻译

---------

Co-authored-by: zipg <4047349+zipg@users.noreply.github.com>
This commit is contained in:
zipgandzipg authored and GitHub committed 2026-08-05 23:58:11 +08:00
1 parent 930aa48c20
commit 2ede64f1cb
64 files changed
+1293 -64
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/connection.rs
+40
View File
@@ -106,6 +106,7 @@ pub enum PoolKind {
HBase(db::hbase_driver::HBaseClient),
VectorDb(db::vector_driver::VectorClient),
InfluxDb(db::influxdb_driver::InfluxdbClient),
VictoriaMetrics(db::victoriametrics_driver::VictoriaMetricsClient),
Agent(Arc<db::agent_driver::PooledAgentClient>),
ExternalDriver {
driver_id: String,
@@ -1950,6 +1951,15 @@ impl AppState {
db::influxdb_driver::test_connection(&client, connect_timeout).await?;
PoolKind::InfluxDb(client)
}
DatabaseType::VictoriaMetrics => {
let client = db::victoriametrics_driver::VictoriaMetricsClient::new_for_config(
&url,
&db_config,
connect_timeout,
)?;
db::victoriametrics_driver::test_connection(&client, connect_timeout).await?;
PoolKind::VictoriaMetrics(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?;
@@ -2892,6 +2902,18 @@ impl AppState {
}
}
}
PoolKind::VictoriaMetrics(client) => {
let client = client.clone();
drop(connections);
let timeout = crate::db::connection_timeout();
match db::victoriametrics_driver::test_connection(&client, timeout).await {
Ok(()) => false,
Err(err) => {
log::warn!("VictoriaMetrics connection pool '{pool_key}' is stale: {err}");
true
}
}
}
PoolKind::Rqlite(client) => {
let client = client.clone();
drop(connections);
@@ -3788,6 +3810,15 @@ impl AppState {
false
}
},
PoolKind::VictoriaMetrics(client) => {
match db::victoriametrics_driver::test_connection(client, timeout).await {
Ok(()) => true,
Err(e) => {
log::warn!("VictoriaMetrics connection pool '{key}' is unhealthy: {e}");
false
}
}
}
PoolKind::Rqlite(client) => match db::rqlite_driver::test_connection(client, timeout).await {
Ok(()) => true,
Err(e) => {
@@ -4060,6 +4091,7 @@ enum KeepaliveTarget {
HBase(db::hbase_driver::HBaseClient),
VectorDb(db::vector_driver::VectorClient),
InfluxDb(db::influxdb_driver::InfluxdbClient),
VictoriaMetrics(db::victoriametrics_driver::VictoriaMetricsClient),
Agent(Arc<db::agent_driver::PooledAgentClient>),
}
@@ -4153,6 +4185,7 @@ fn keepalive_target_from_pool(pool: &PoolKind, config: &ConnectionConfig) -> Opt
PoolKind::HBase(client) => Some(KeepaliveTarget::HBase(client.clone())),
PoolKind::VectorDb(client) => Some(KeepaliveTarget::VectorDb(client.clone())),
PoolKind::InfluxDb(client) => Some(KeepaliveTarget::InfluxDb(client.clone())),
PoolKind::VictoriaMetrics(client) => Some(KeepaliveTarget::VictoriaMetrics(client.clone())),
PoolKind::Agent(client) => Some(KeepaliveTarget::Agent(client.clone())),
_ => None,
}
@@ -4199,6 +4232,9 @@ async fn ping_keepalive_target(target: &mut KeepaliveTarget, timeout: Duration)
KeepaliveTarget::InfluxDb(client) => {
db::influxdb_driver::test_connection(client, timeout).await.map_err(Into::into)
}
KeepaliveTarget::VictoriaMetrics(client) => {
db::victoriametrics_driver::test_connection(client, timeout).await.map_err(Into::into)
}
KeepaliveTarget::Agent(client) => {
let Ok(mut client) = client.try_lock() else {
return Ok(());
@@ -4429,6 +4465,7 @@ fn clone_pool_kind(pool: &PoolKind) -> PoolKind {
PoolKind::HBase(client) => PoolKind::HBase(client.clone()),
PoolKind::VectorDb(client) => PoolKind::VectorDb(client.clone()),
PoolKind::InfluxDb(client) => PoolKind::InfluxDb(client.clone()),
PoolKind::VictoriaMetrics(client) => PoolKind::VictoriaMetrics(client.clone()),
PoolKind::Agent(client) => PoolKind::Agent(client.clone()),
PoolKind::ExternalDriver { driver_id, config, session } => {
PoolKind::ExternalDriver { driver_id: driver_id.clone(), config: config.clone(), session: session.clone() }
@@ -4484,6 +4521,9 @@ async fn close_pool_kind(pool: PoolKind) -> Result<(), String> {
PoolKind::InfluxDb(client) => {
drop(client);
}
PoolKind::VictoriaMetrics(client) => {
drop(client);
}
PoolKind::Agent(client) => {
let mut client = client.lock().await;
client.disconnect().await?;
crates/dbx-core/src/data_grid_sql.rs
+30
View File
@@ -150,6 +150,9 @@ fn supports_data_grid_context_filter_mode(
database_type: Option<DatabaseType>,
mode: DataGridContextFilterMode,
) -> bool {
if database_type == Some(DatabaseType::VictoriaMetrics) {
return false;
}
!matches!(
(database_type, mode),
(
@@ -3767,6 +3770,33 @@ mod tests {
}
}
#[test]
fn does_not_emit_sql_filters_for_victoriametrics() {
for mode in [
DataGridContextFilterMode::Equals,
DataGridContextFilterMode::Like,
DataGridContextFilterMode::GreaterThan,
DataGridContextFilterMode::IsNull,
DataGridContextFilterMode::In,
DataGridContextFilterMode::Between,
] {
assert_eq!(
build_data_grid_context_filter_condition(DataGridContextFilterConditionOptions {
database_type: Some(DatabaseType::VictoriaMetrics),
identifier_quote: None,
column_name: "value".to_string(),
mode,
value: json!(10),
values: vec![json!(10), json!(20)],
end_value: Some(json!(20)),
column_info: Some(column("value", "double", false, None)),
}),
None,
"VictoriaMetrics must not emit SQL filter mode {mode:?}"
);
}
}
#[test]
fn context_membership_and_range_filters_require_complete_values() {
let column_info = Some(column("score", "int", false, None));
crates/dbx-core/src/database_capabilities.rs
+2
View File
@@ -33,6 +33,7 @@ pub fn is_single_connection_pool(db_type: &DatabaseType) -> bool {
| DatabaseType::OceanbaseOracle
| DatabaseType::Access
| DatabaseType::Jdbc
| DatabaseType::VictoriaMetrics
)
}
@@ -57,6 +58,7 @@ pub fn skips_tcp_probe(db_type: &DatabaseType) -> bool {
| DatabaseType::Jdbc
| DatabaseType::MessageQueue
| DatabaseType::Mqtt
| DatabaseType::VictoriaMetrics
) || is_agent_type(db_type)
}
crates/dbx-core/src/db/mod.rs
+1
View File
@@ -36,6 +36,7 @@ pub mod starrocks;
pub mod transport_layer_tunnel;
pub mod turso_driver;
pub mod vector_driver;
pub mod victoriametrics_driver;
pub mod wkb;
use reqwest::ClientBuilder;
crates/dbx-core/src/db/victoriametrics_driver.rs
+569
View File
@@ -0,0 +1,569 @@
use chrono::{DateTime, Utc};
use reqwest::{Certificate, Client as HttpClient, Response};
use serde::Deserialize;
use serde_json::{json, Value};
use std::collections::{BTreeSet, HashMap};
use std::fs;
use std::time::{Duration, Instant};
use super::with_connection_timeout;
use crate::models::connection::ConnectionConfig;
use crate::types::{ColumnInfo, DatabaseInfo, ObjectStatistics, QueryResult, TableInfo};
const DEFAULT_API_PATH: &str = "/prometheus";
const DEFAULT_LOOKBACK: &str = "1h";
#[derive(Clone)]
pub struct VictoriaMetricsClient {
http: HttpClient,
base_url: String,
api_path: String,
username: Option<String>,
password: Option<String>,
database_label: String,
lookback: String,
}
#[derive(Debug, Deserialize)]
#[serde(bound(deserialize = "T: Deserialize<'de>"))]
struct ApiResponse<T> {
status: String,
#[serde(default)]
data: Option<T>,
#[serde(default)]
error_type: Option<String>,
#[serde(default)]
error: Option<String>,
}
#[derive(Debug, Deserialize)]
struct QueryData {
#[serde(rename = "resultType")]
result_type: String,
result: Value,
}
#[derive(Debug, Deserialize)]
struct SeriesResult {
#[serde(default)]
metric: HashMap<String, String>,
#[serde(default)]
value: Option<(f64, String)>,
#[serde(default)]
values: Vec<(f64, String)>,
}
#[derive(Debug, Deserialize)]
struct TsdbStatus {
#[serde(rename = "seriesCountByMetricName", default)]
series_count_by_metric_name: Vec<TsdbMetricStatistic>,
}
#[derive(Debug, Deserialize)]
struct TsdbMetricStatistic {
name: String,
value: Value,
}
impl VictoriaMetricsClient {
pub fn new_for_config(url: &str, config: &ConnectionConfig, timeout: Duration) -> Result<Self, String> {
let api_path = external_string(config, &["apiPath", "api_path"])
.map(|value| normalize_api_path(&value))
.unwrap_or_else(|| DEFAULT_API_PATH.to_string());
let lookback = external_string(config, &["lookback"])
.filter(|value| valid_lookback(value))
.unwrap_or_else(|| DEFAULT_LOOKBACK.to_string());
let database_label = config
.database
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("metrics")
.to_string();
let http = build_http_client(Some(&config.ca_cert_path), timeout)?;
Ok(Self {
http,
base_url: url.trim_end_matches('/').to_string(),
api_path,
username: (!config.username.trim().is_empty()).then(|| config.username.clone()),
password: (!config.password.is_empty()).then(|| config.password.clone()),
database_label,
lookback,
})
}
fn endpoint(&self, suffix: &str) -> String {
format!("{}{}{}", self.base_url, self.api_path, suffix)
}
fn request(&self, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
match (&self.username, &self.password) {
(Some(username), password) => request.basic_auth(username, password.as_deref()),
_ => request,
}
}
}
fn external_string(config: &ConnectionConfig, keys: &[&str]) -> Option<String> {
let value = config.external_config.as_ref()?;
keys.iter()
.find_map(|key| value.get(key))
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string)
}
fn normalize_api_path(value: &str) -> String {
let trimmed = value.trim().trim_end_matches('/');
if trimmed.is_empty() || trimmed == "/" {
String::new()
} else if trimmed.starts_with('/') {
trimmed.to_string()
} else {
format!("/{trimmed}")
}
}
fn valid_lookback(value: &str) -> bool {
let value = value.trim();
let split = value.find(|character: char| !character.is_ascii_digit()).unwrap_or(value.len());
split > 0
&& split < value.len()
&& value[..split].parse::<u64>().is_ok_and(|number| number > 0)
&& matches!(&value[split..], "ms" | "s" | "m" | "h" | "d" | "w" | "y")
}
fn build_http_client(ca_cert_path: Option<&str>, timeout: Duration) -> Result<HttpClient, String> {
let mut builder = HttpClient::builder().connect_timeout(timeout);
if let Some(path) = ca_cert_path.map(str::trim).filter(|path| !path.is_empty()) {
let path = expand_cert_path(path);
let cert_bytes = fs::read(&path)
.map_err(|error| format!("Failed to read VictoriaMetrics CA certificate at {path}: {error}"))?;
let cert = Certificate::from_pem(&cert_bytes)
.or_else(|_| Certificate::from_der(&cert_bytes))
.map_err(|error| format!("Failed to parse VictoriaMetrics CA certificate at {path}: {error}"))?;
builder = builder.add_root_certificate(cert);
}
builder.build().map_err(|error| format!("Failed to configure VictoriaMetrics HTTP client: {error}"))
}
fn expand_cert_path(path: &str) -> String {
let home = || std::env::var(if cfg!(windows) { "USERPROFILE" } else { "HOME" }).ok();
if path == "~" || path.starts_with("~/") || path.starts_with("~\\") {
if let Some(home) = home() {
return format!("{}{}", home, &path[1..]);
}
}
if let Some(rest) = path.strip_prefix("$HOME") {
if let Some(home) = home() {
return format!("{home}{rest}");
}
}
if let Some(rest) = path.strip_prefix("${HOME}") {
if let Some(home) = home() {
return format!("{home}{rest}");
}
}
if let Some(rest) = path.strip_prefix("%USERPROFILE%") {
if let Ok(home) = std::env::var("USERPROFILE") {
return format!("{home}{rest}");
}
}
path.to_string()
}
pub async fn test_connection(client: &VictoriaMetricsClient, timeout: Duration) -> Result<(), String> {
let request = client.request(client.http.get(client.endpoint("/api/v1/status/buildinfo")));
let response = with_connection_timeout("VictoriaMetrics", timeout, async {
request.send().await.map_err(|error| format!("VictoriaMetrics connection failed: {error}"))
})
.await?;
parse_api_response::<Value>(response).await.map(|_| ())
}
pub async fn list_databases(client: &VictoriaMetricsClient) -> Result<Vec<DatabaseInfo>, String> {
Ok(vec![DatabaseInfo { name: client.database_label.clone() }])
}
pub async fn list_tables(client: &VictoriaMetricsClient) -> Result<Vec<TableInfo>, String> {
let request = client.request(
client
.http
.get(client.endpoint("/api/v1/label/__name__/values"))
.query(&[("start", format!("-{}", client.lookback))]),
);
let mut names = parse_api_response::<Vec<String>>(request.send().await.map_err(request_error)?).await?;
names.sort();
names.dedup();
Ok(names
.into_iter()
.filter(|name| !name.trim().is_empty())
.map(|name| TableInfo {
name,
table_type: "TABLE".to_string(),
comment: None,
parent_schema: None,
parent_name: None,
})
.collect())
}
pub async fn get_columns(client: &VictoriaMetricsClient, metric_name: &str) -> Result<Vec<ColumnInfo>, String> {
let selector = metric_selector(metric_name);
let request = client.request(
client
.http
.post(client.endpoint("/api/v1/series"))
.form(&[("match[]", selector.as_str()), ("start", format!("-{}", client.lookback).as_str())]),
);
let series =
parse_api_response::<Vec<HashMap<String, String>>>(request.send().await.map_err(request_error)?).await?;
let label_names = series
.iter()
.flat_map(|labels| labels.keys())
.filter(|name| name.as_str() != "__name__")
.cloned()
.collect::<BTreeSet<_>>();
let label_names = label_names.into_iter().collect::<Vec<_>>();
let (timestamp_name, value_name, metric_name) = result_column_names(&label_names);
let mut columns = vec![
column(&timestamp_name, "timestamp", false, true, Some("Sample timestamp")),
column(&value_name, "double", false, false, Some("Sample value")),
column(&metric_name, "string", false, true, Some("Metric name")),
];
columns.extend(label_names.into_iter().map(|name| column(&name, "string", true, true, Some("Metric label"))));
Ok(columns)
}
pub async fn list_object_statistics(client: &VictoriaMetricsClient) -> Result<Vec<ObjectStatistics>, String> {
let request = client.request(client.http.get(client.endpoint("/api/v1/status/tsdb"))).query(&[("topN", "100000")]);
let status = parse_api_response::<TsdbStatus>(request.send().await.map_err(request_error)?).await?;
Ok(status
.series_count_by_metric_name
.into_iter()
.filter(|item| !item.name.trim().is_empty())
.map(|item| ObjectStatistics {
name: item.name,
schema: None,
estimated_rows: item.value.as_i64().or_else(|| item.value.as_str().and_then(|value| value.parse().ok())),
// VictoriaMetrics exposes series counts here, but no per-metric storage size.
total_bytes: None,
})
.collect())
}
fn column(name: &str, data_type: &str, is_nullable: bool, is_primary_key: bool, comment: Option<&str>) -> ColumnInfo {
ColumnInfo {
name: name.to_string(),
data_type: data_type.to_string(),
is_nullable,
is_primary_key,
comment: comment.map(str::to_string),
..Default::default()
}
}
pub async fn execute_query(client: &VictoriaMetricsClient, query: &str) -> Result<QueryResult, String> {
let start = Instant::now();
let query = query.trim().trim_end_matches(';').trim();
if query.is_empty() {
return Err("VictoriaMetrics query is empty".to_string());
}
let request = client.request(client.http.post(client.endpoint("/api/v1/query")).form(&[("query", query)]));
let data = parse_api_response::<QueryData>(request.send().await.map_err(request_error)?).await?;
query_data_to_result(data, start)
}
fn query_data_to_result(data: QueryData, start: Instant) -> Result<QueryResult, String> {
match data.result_type.as_str() {
"vector" | "matrix" => {
let series = serde_json::from_value::<Vec<SeriesResult>>(data.result)
.map_err(|error| format!("VictoriaMetrics response parse failed: {error}"))?;
Ok(series_result_to_query_result(series, start))
}
"scalar" => {
let (timestamp, value) = serde_json::from_value::<(f64, String)>(data.result)
.map_err(|error| format!("VictoriaMetrics response parse failed: {error}"))?;
Ok(simple_result(vec![vec![timestamp_value(timestamp), sample_value(&value)]], "double", start))
}
"string" => {
let (timestamp, value) = serde_json::from_value::<(f64, String)>(data.result)
.map_err(|error| format!("VictoriaMetrics response parse failed: {error}"))?;
Ok(simple_result(vec![vec![timestamp_value(timestamp), Value::String(value)]], "string", start))
}
other => Err(format!("VictoriaMetrics returned unsupported result type: {other}")),
}
}
fn series_result_to_query_result(series: Vec<SeriesResult>, start: Instant) -> QueryResult {
let labels = series
.iter()
.flat_map(|item| item.metric.keys())
.filter(|name| name.as_str() != "__name__")
.cloned()
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
let (timestamp_name, value_name, metric_name) = result_column_names(&labels);
let mut columns = vec![timestamp_name, value_name, metric_name];
columns.extend(labels.iter().cloned());
let mut rows = Vec::new();
for item in series {
let metric_name = item.metric.get("__name__").cloned().map(Value::String).unwrap_or(Value::Null);
let samples = item.value.into_iter().chain(item.values);
for (timestamp, value) in samples {
let mut row = vec![timestamp_value(timestamp), sample_value(&value), metric_name.clone()];
row.extend(
labels.iter().map(|label| item.metric.get(label).cloned().map(Value::String).unwrap_or(Value::Null)),
);
rows.push(row);
}
}
QueryResult {
column_types: std::iter::once("timestamp".to_string())
.chain(std::iter::once("double".to_string()))
.chain(std::iter::repeat_n("string".to_string(), columns.len().saturating_sub(2)))
.collect(),
column_sortables: vec![true; columns.len()],
columns,
affected_rows: rows.len() as u64,
rows,
execution_time_ms: start.elapsed().as_millis(),
spatial_columns: vec![],
spatial_values: vec![],
truncated: false,
session_id: None,
has_more: false,
elasticsearch_raw_body: None,
}
}
fn result_column_names(labels: &[String]) -> (String, String, String) {
let mut used = labels.iter().cloned().collect::<BTreeSet<_>>();
let timestamp = unique_result_column_name("timestamp", "sample_timestamp", &mut used);
let value = unique_result_column_name("value", "sample_value", &mut used);
let metric = unique_result_column_name("metric", "metric_name", &mut used);
(timestamp, value, metric)
}
fn unique_result_column_name(preferred: &str, fallback: &str, used: &mut BTreeSet<String>) -> String {
let mut candidate = if used.contains(preferred) { fallback.to_string() } else { preferred.to_string() };
let mut suffix = 2;
while used.contains(&candidate) {
candidate = format!("{fallback}_{suffix}");
suffix += 1;
}
used.insert(candidate.clone());
candidate
}
fn simple_result(rows: Vec<Vec<Value>>, value_type: &str, start: Instant) -> QueryResult {
QueryResult {
columns: vec!["timestamp".to_string(), "value".to_string()],
column_types: vec!["timestamp".to_string(), value_type.to_string()],
column_sortables: vec![true, true],
affected_rows: rows.len() as u64,
rows,
execution_time_ms: start.elapsed().as_millis(),
spatial_columns: vec![],
spatial_values: vec![],
truncated: false,
session_id: None,
has_more: false,
elasticsearch_raw_body: None,
}
}
fn timestamp_value(timestamp: f64) -> Value {
DateTime::<Utc>::from_timestamp(timestamp.trunc() as i64, (timestamp.fract() * 1_000_000_000.0) as u32)
.map(|value| Value::String(value.to_rfc3339()))
.unwrap_or_else(|| json!(timestamp))
}
fn sample_value(value: &str) -> Value {
value
.parse::<f64>()
.ok()
.filter(|number| number.is_finite())
.map(|number| json!(number))
.unwrap_or_else(|| json!(value))
}
pub fn metric_selector(metric_name: &str) -> String {
format!(r#"{{__name__="{}"}}"#, escape_label_value(metric_name))
}
pub fn metric_range_query(metric_name: &str, lookback: &str) -> String {
format!("{}[{}]", metric_selector(metric_name), if valid_lookback(lookback) { lookback } else { DEFAULT_LOOKBACK })
}
fn escape_label_value(value: &str) -> String {
value.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n")
}
async fn parse_api_response<T: for<'de> Deserialize<'de>>(response: Response) -> Result<T, String> {
let status = response.status();
let body = response.text().await.unwrap_or_default();
if !status.is_success() {
return Err(http_error(status, &body));
}
let parsed = serde_json::from_str::<ApiResponse<T>>(&body)
.map_err(|error| format!("VictoriaMetrics response parse failed: {error}; response: {body}"))?;
if parsed.status != "success" {
let kind =
parsed.error_type.filter(|value| !value.is_empty()).map(|value| format!("{value}: ")).unwrap_or_default();
return Err(format!(
"VictoriaMetrics error: {kind}{}",
parsed.error.unwrap_or_else(|| "unknown error".to_string())
));
}
parsed.data.ok_or_else(|| "VictoriaMetrics response is missing data".to_string())
}
fn http_error(status: reqwest::StatusCode, body: &str) -> String {
let detail = serde_json::from_str::<Value>(body)
.ok()
.and_then(|value| value.get("error").and_then(Value::as_str).map(str::to_string))
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| body.trim().to_string());
let detail =
if detail.is_empty() { status.canonical_reason().unwrap_or("request failed").to_string() } else { detail };
format!("VictoriaMetrics error: status={}, message={detail}", status.as_u16())
}
fn request_error(error: reqwest::Error) -> String {
format!("VictoriaMetrics request failed: {error}")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn normalizes_api_paths_and_lookbacks() {
assert_eq!(normalize_api_path("prometheus/"), "/prometheus");
assert_eq!(normalize_api_path("/select/42/prometheus/"), "/select/42/prometheus");
assert_eq!(normalize_api_path("/"), "");
assert!(valid_lookback("15m"));
assert!(valid_lookback("250ms"));
assert!(!valid_lookback("0h"));
assert!(!valid_lookback("1 hour"));
}
#[test]
fn metric_queries_escape_label_values() {
assert_eq!(metric_selector("rack\\\"temp"), r#"{__name__="rack\\\"temp"}"#);
assert_eq!(metric_range_query("cpu_usage", "2h"), r#"{__name__="cpu_usage"}[2h]"#);
assert_eq!(metric_range_query("cpu_usage", "bad"), r#"{__name__="cpu_usage"}[1h]"#);
}
#[test]
fn parses_metric_statistics() {
let status: TsdbStatus = serde_json::from_value(json!({
"seriesCountByMetricName": [{"name": "flag", "value": "276"}]
}))
.unwrap();
assert_eq!(status.series_count_by_metric_name[0].name, "flag");
assert_eq!(status.series_count_by_metric_name[0].value.as_str(), Some("276"));
}
#[test]
fn parses_matrix_results_into_rows() {
let data = QueryData {
result_type: "matrix".to_string(),
result: json!([
{
"metric": {"__name__": "temperature", "site": "shanghai"},
"values": [[1720000000.0, "23.5"], [1720000060.0, "24"]]
}
]),
};
let result = query_data_to_result(data, Instant::now()).unwrap();
assert_eq!(result.columns, vec!["timestamp", "value", "metric", "site"]);
assert_eq!(result.rows.len(), 2);
assert_eq!(result.rows[0][1], json!(23.5));
assert_eq!(result.rows[0][2], json!("temperature"));
assert_eq!(result.rows[0][3], json!("shanghai"));
}
#[test]
fn keeps_reserved_label_names_unique() {
let data = QueryData {
result_type: "vector".to_string(),
result: json!([
{
"metric": {
"__name__": "flag",
"metric": "label-metric",
"timestamp": "label-timestamp",
"value": "label-value"
},
"value": [1720000000.0, "1"]
}
]),
};
let result = query_data_to_result(data, Instant::now()).unwrap();
assert_eq!(
result.columns,
vec!["sample_timestamp", "sample_value", "metric_name", "metric", "timestamp", "value"]
);
assert_eq!(result.rows[0].len(), result.columns.len());
assert_eq!(result.rows[0][3], json!("label-metric"));
assert_eq!(result.rows[0][5], json!("label-value"));
}
#[test]
fn parses_scalar_and_preserves_non_finite_values() {
let scalar = QueryData { result_type: "scalar".to_string(), result: json!([1720000000.0, "42"]) };
let result = query_data_to_result(scalar, Instant::now()).unwrap();
assert_eq!(result.rows[0][1], json!(42.0));
assert_eq!(sample_value("NaN"), json!("NaN"));
}
#[test]
fn keeps_string_query_results_as_strings() {
let string = QueryData { result_type: "string".to_string(), result: json!([1720000000.0, "healthy"]) };
let result = query_data_to_result(string, Instant::now()).unwrap();
assert_eq!(result.column_types, vec!["timestamp", "string"]);
assert_eq!(result.rows[0][1], json!("healthy"));
}
#[tokio::test]
#[ignore = "requires DBX_LIVE_VICTORIAMETRICS_* env vars pointing at a VictoriaMetrics endpoint"]
async fn live_connection_metadata_and_query() {
let url = std::env::var("DBX_LIVE_VICTORIAMETRICS_URL").expect("DBX_LIVE_VICTORIAMETRICS_URL");
let username = std::env::var("DBX_LIVE_VICTORIAMETRICS_USER").unwrap_or_default();
let password = std::env::var("DBX_LIVE_VICTORIAMETRICS_PASSWORD").unwrap_or_default();
let api_path = std::env::var("DBX_LIVE_VICTORIAMETRICS_API_PATH").unwrap_or_else(|_| "/prometheus".into());
let metric = std::env::var("DBX_LIVE_VICTORIAMETRICS_METRIC").expect("DBX_LIVE_VICTORIAMETRICS_METRIC");
let config: ConnectionConfig = serde_json::from_value(json!({
"id": "live-victoriametrics",
"name": "Live VictoriaMetrics",
"db_type": "victoriametrics",
"host": "unused",
"port": 8428,
"username": username,
"password": password,
"database": "metrics",
"external_config": { "apiPath": api_path, "lookback": "1h" }
}))
.expect("live VictoriaMetrics config");
let client = VictoriaMetricsClient::new_for_config(&url, &config, Duration::from_secs(10)).unwrap();
test_connection(&client, Duration::from_secs(10)).await.unwrap();
let tables = list_tables(&client).await.unwrap();
assert!(tables.iter().any(|table| table.name == metric));
let columns = get_columns(&client, &metric).await.unwrap();
assert!(columns.iter().any(|column| column.name == "timestamp"));
assert!(columns.iter().any(|column| column.name == "value"));
assert!(columns.iter().any(|column| column.name == "metric"));
let statistics = list_object_statistics(&client).await.unwrap();
assert!(statistics.iter().any(|stat| stat.name == metric && stat.estimated_rows.unwrap_or_default() > 0));
let result = execute_query(&client, &metric_range_query(&metric, "1h")).await.unwrap();
assert!(!result.rows.is_empty());
}
}
crates/dbx-core/src/models/connection.rs
+5 -2
View File
@@ -538,6 +538,8 @@ pub enum DatabaseType {
CloudflareD1,
#[serde(rename = "influxdb")]
InfluxDb,
#[serde(rename = "victoriametrics")]
VictoriaMetrics,
#[serde(rename = "questdb")]
Questdb,
Jdbc,
@@ -890,6 +892,7 @@ impl ConnectionConfig {
DatabaseType::H2 => Some("test"),
DatabaseType::Informix => Some("sysmaster"),
DatabaseType::Neo4j => Some("neo4j"),
DatabaseType::VictoriaMetrics => Some("metrics"),
_ => None,
}
}
@@ -1100,7 +1103,7 @@ impl ConnectionConfig {
format!("zookeeper://{host}:{port}")
}
DatabaseType::Iris => format!("iris://{host}:{port}{db_part}"),
DatabaseType::InfluxDb => {
DatabaseType::InfluxDb | DatabaseType::VictoriaMetrics => {
let scheme = if self.ssl { "https" } else { "http" };
format!("{scheme}://{host}:{port}")
}
@@ -1340,7 +1343,7 @@ impl ConnectionConfig {
DatabaseType::Iris => {
format!("iris://{}:{}@{host}:{port}{db_part}", username, password)
}
DatabaseType::InfluxDb => {
DatabaseType::InfluxDb | DatabaseType::VictoriaMetrics => {
let scheme = if self.ssl { "https" } else { "http" };
format!("{scheme}://{host}:{port}")
}
crates/dbx-core/src/query.rs
+19
View File
@@ -851,6 +851,7 @@ fn should_discard_pool_after_query_timeout(db_type: Option<DatabaseType>) -> boo
| DatabaseType::Weaviate
| DatabaseType::ChromaDb
| DatabaseType::InfluxDb
| DatabaseType::VictoriaMetrics
)
}
@@ -1562,6 +1563,22 @@ async fn do_execute_typed(
}
result
}
PoolKind::VictoriaMetrics(client) => {
let client = client.clone();
let max_rows = options.max_rows;
drop(connections);
let result = wait_for_query_opt(
cancel_token,
query_timeout,
db::victoriametrics_driver::execute_query(&client, sql),
)
.await
.map(|result| truncate_result_with_max_rows(result, max_rows));
if matches!(result.as_ref(), Err(err) if should_discard_pool_after_error(pool_db_type, err)) {
state.remove_pool_by_key(pool_key).await;
}
result
}
PoolKind::Agent(client) => {
let client = client.clone();
let source_client = client.clone();
@@ -2840,6 +2857,7 @@ fn pool_kind_has_transactional_path(pool: &PoolKind) -> bool {
| PoolKind::Easysearch(_)
| PoolKind::VectorDb(_)
| PoolKind::InfluxDb(_)
| PoolKind::VictoriaMetrics(_)
| PoolKind::ExternalDriver { .. } => false,
#[cfg(feature = "mq-admin")]
PoolKind::Mqtt(_) => false,
@@ -3084,6 +3102,7 @@ pub async fn execute_statements_in_transaction_on_pool_typed(
| PoolKind::Easysearch(_)
| PoolKind::VectorDb(_)
| PoolKind::InfluxDb(_)
| PoolKind::VictoriaMetrics(_)
| PoolKind::ExternalDriver { .. } => TxPath::None,
})
};
crates/dbx-core/src/query_execution_sql.rs
+21 -1
View File
@@ -141,7 +141,7 @@ pub fn is_safe_explain_sql(sql: &str) -> bool {
}
/// Returns true for databases that support SQL query execution (execute_query / get_sample_data).
/// Non-SQL databases (Redis, MongoDB, Elasticsearch, InfluxDB, Neo4j, etcd) are excluded.
/// Non-SQL databases (Redis, MongoDB, Elasticsearch, InfluxDB, VictoriaMetrics, Neo4j, etcd) are excluded.
pub fn supports_sql_query(database_type: DatabaseType) -> bool {
!matches!(
database_type,
@@ -154,6 +154,7 @@ pub fn supports_sql_query(database_type: DatabaseType) -> bool {
| DatabaseType::Weaviate
| DatabaseType::ChromaDb
| DatabaseType::InfluxDb
| DatabaseType::VictoriaMetrics
| DatabaseType::Neo4j
| DatabaseType::Etcd
)
@@ -228,6 +229,11 @@ pub fn is_write_sql(sql: &str) -> bool {
/// executable comments and file exports, plus PostgreSQL-family/SQL Server
/// `SELECT ... INTO` table creation.
pub fn is_write_sql_for_database(sql: &str, database_type: DatabaseType) -> bool {
// VictoriaMetrics execution is hard-wired to the read-only query API; MetricsQL
// does not use SQL verbs and must not be rejected by SQL write classification.
if database_type == DatabaseType::VictoriaMetrics {
return false;
}
if let Some(risk) = classify_search_engine_query_risk(sql, database_type) {
return risk != SearchEngineQueryRisk::ReadOnly;
}
@@ -1530,4 +1536,18 @@ mod tests {
DatabaseType::Easysearch
));
}
#[test]
fn excludes_victoriametrics_from_sql_query_paths() {
assert!(!supports_sql_query(DatabaseType::VictoriaMetrics));
assert!(supports_sql_query(DatabaseType::Postgres));
assert_eq!(
check_read_only(
r#"{__name__="dbx_issue_5352_temperature_celsius"}[5m]"#,
"VictoriaMetrics",
DatabaseType::VictoriaMetrics,
),
Ok(())
);
}
}
crates/dbx-core/src/schema.rs
+18
View File
@@ -581,6 +581,10 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result<Ve
drop(connections);
return db::influxdb_driver::list_databases(&client).await;
}
if let Some(client) = extract_pool!(&connections, connection_id, VictoriaMetrics) {
drop(connections);
return db::victoriametrics_driver::list_databases(&client).await;
}
try_sqlserver!(connections, connection_id, list_databases);
if let Some(client) = extract_pool!(&connections, connection_id, Agent) {
let is_mongo =
@@ -1854,6 +1858,12 @@ async fn list_tables_once(
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types, table_name_filter));
}
if let Some(client) = extract_pool!(&connections, &pool_key, VictoriaMetrics) {
drop(connections);
return db::victoriametrics_driver::list_tables(&client)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types, table_name_filter));
}
if let Some(linked) = crate::sql_dialect::parse_sqlserver_linked_schema_ref(schema) {
if let Some(client) = extract_pool!(&connections, &pool_key, SqlServer) {
drop(connections);
@@ -4538,6 +4548,10 @@ async fn list_object_statistics_once(
.await;
}
}
if let Some(client) = extract_pool!(&connections, &pool_key, VictoriaMetrics) {
drop(connections);
return db::victoriametrics_driver::list_object_statistics(&client).await;
}
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
match pool {
PoolKind::Mysql(p, mode) => {
@@ -5217,6 +5231,10 @@ async fn get_columns_core_for_session_inner(
drop(connections);
return db::influxdb_driver::get_columns(&client, database, table).await.map(deduplicate_column_infos);
}
if let Some(client) = extract_pool!(&connections, &pool_key, VictoriaMetrics) {
drop(connections);
return db::victoriametrics_driver::get_columns(&client, table).await.map(deduplicate_column_infos);
}
if let Some(linked) = crate::sql_dialect::parse_sqlserver_linked_schema_ref(schema) {
if let Some(client) = extract_pool!(&connections, &pool_key, SqlServer) {
drop(connections);
crates/dbx-core/src/sql_dialect/ddl_profile.rs
+3 -1
View File
@@ -674,7 +674,9 @@ pub fn profile_for(db_type: DatabaseType) -> DdlDialectProfile {
// Non-tabular / not applicable for relational CREATE TABLE
Redis | MongoDb | Elasticsearch | Easysearch | Qdrant | Milvus | Weaviate | ChromaDb | Neo4j | Cassandra
| Etcd | ZooKeeper | Nacos | InfluxDb | MessageQueue | Mqtt | Hbase => conservative_ansi(db_type),
| Etcd | ZooKeeper | Nacos | InfluxDb | VictoriaMetrics | MessageQueue | Mqtt | Hbase => {
conservative_ansi(db_type)
}
}
}
crates/dbx-core/src/sql_dialect/table_select.rs
+26
View File
@@ -13,6 +13,9 @@ use super::types::{
};
pub fn build_count_table_sql(database_type: Option<DatabaseType>, schema: Option<&str>, table_name: &str) -> String {
if database_type == Some(DatabaseType::VictoriaMetrics) {
return format!("count({})", victoriametrics_metric_selector(table_name));
}
format!("SELECT COUNT(*) AS row_count FROM {}", qualified_table_name(database_type, schema, table_name))
}
@@ -22,6 +25,9 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
if database_type == Some(DatabaseType::Neo4j) {
return build_neo4j_table_select_sql(&options, limit);
}
if database_type == Some(DatabaseType::VictoriaMetrics) {
return format!("{}[1h]", victoriametrics_metric_selector(&options.table_name));
}
// Doris / StarRocks multi-catalog: prefix the catalog for external-catalog tables.
let table = if uses_connection_identifier_quote(database_type, options.identifier_quote.as_deref()) {
@@ -214,6 +220,9 @@ fn is_view_table_type(table_type: Option<&str>) -> bool {
pub fn build_table_select_sql(options: TableSelectSqlOptions<'_>) -> String {
let database_type = options.database_type;
if database_type == Some(DatabaseType::VictoriaMetrics) {
return format!("{}[1h]", victoriametrics_metric_selector(options.table_name));
}
let table = qualified_table_name(database_type, options.schema, options.table_name);
let select_columns = quoted_table_columns_or_star(database_type, options.columns);
let order_by = if options.order_columns.is_empty() {
@@ -257,6 +266,11 @@ pub fn build_table_select_sql(options: TableSelectSqlOptions<'_>) -> String {
}
}
fn victoriametrics_metric_selector(metric_name: &str) -> String {
let escaped = metric_name.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n");
format!(r#"{{__name__="{escaped}"}}"#)
}
fn informix_row_limit_clause(limit: usize, offset: usize) -> String {
if offset > 0 {
format!("SKIP {offset} FIRST {limit}")
@@ -555,4 +569,16 @@ mod tests {
assert!(!sql.contains("iceberg_catalog"), "sql was: {sql}");
assert!(sql.contains("orders"), "sql was: {sql}");
}
#[test]
fn victoriametrics_builds_metric_queries_without_sql_identifiers() {
assert_eq!(
build_table_data_select_sql(opts(DatabaseType::VictoriaMetrics, None, None, "rack_temperature")),
r#"{__name__="rack_temperature"}[1h]"#
);
assert_eq!(
build_count_table_sql(Some(DatabaseType::VictoriaMetrics), None, "rack\\\"temperature"),
r#"count({__name__="rack\\\"temperature"})"#
);
}
}
crates/dbx-core/src/table_export.rs
+24 -1
View File
@@ -430,6 +430,29 @@ async fn fetch_table_export_batch(
});
}
// VictoriaMetrics metrics are read through MetricsQL rather than SQL table
// pagination. Execute the range query once and let the normal writers emit
// the returned matrix/vector rows.
if *db_type == DatabaseType::VictoriaMetrics {
*table_read_attempted = true;
*table_read_completed = true;
let query = crate::db::victoriametrics_driver::metric_range_query(&request.table_name, "1h");
return execute_sql_statement_with_options(
state,
&request.connection_id,
&request.database,
&query,
request.schema.as_deref(),
Some(cancel_token),
QueryExecutionOptions {
max_rows: request.row_limit,
client_session_id: Some(table_export_client_session_id(&request.export_id)),
..Default::default()
},
)
.await;
}
if !*table_read_attempted {
match table_export_cursor_kind(state, pool_key).await {
Some(TableExportCursorKind::Agent) => {
@@ -1205,7 +1228,7 @@ async fn export_table_data_core_inner(
// grid exports skip this by default because COUNT can be the slowest query
// on large HANA/JDBC tables, especially with filters.
let row_limit = request.row_limit;
let total_rows = if request.skip_count {
let total_rows = if request.skip_count || db_type == DatabaseType::VictoriaMetrics {
None
} else {
let count_query = count_sql_with_where_and_identifier_quote(