mirror of
https://github.com/Syngnat/GoNavi.git
synced 2026-08-21 16:34:21 +08:00
✨ feat(clickhouse): 支持自定义连接兼容 JDBC DSN
- 解析 ClickHouse 与 JDBC 风格 DSN 并映射协议、TLS、认证及数据库参数 - 统一连接缓存、DDL、数据同步与导出 driver-agent 生命周期 - 保持 DSN 加密存储并补充前后端回归测试和多语言提示 Fixes #383
This commit is contained in:
Syngnat
@@ -561,8 +561,8 @@ func shouldRefreshCachedConnection(err error) bool {
|
||||
}
|
||||
|
||||
func (a *App) invalidateCachedDatabase(config connection.ConnectionConfig, reason error) bool {
|
||||
if resolvedConfig, err := a.resolveConnectionSecrets(config); err == nil {
|
||||
config = resolvedConfig
|
||||
if effectiveConfig, err := a.resolveEffectiveConnectionConfig(config); err == nil {
|
||||
config = effectiveConfig
|
||||
}
|
||||
effectiveConfig := config
|
||||
key := getCacheKey(effectiveConfig)
|
||||
@@ -876,15 +876,18 @@ func (a *App) resolveEffectiveConnectionConfig(config connection.ConnectionConfi
|
||||
if err != nil {
|
||||
return config, wrapConnectError(config, err)
|
||||
}
|
||||
return resolvedConfig, nil
|
||||
runtimeConfig, err := a.resolveCustomClickHouseRuntimeConfig(resolvedConfig)
|
||||
if err != nil {
|
||||
return config, wrapConnectError(resolvedConfig, err)
|
||||
}
|
||||
return runtimeConfig, nil
|
||||
}
|
||||
|
||||
func (a *App) getDatabaseWithPing(config connection.ConnectionConfig, forcePing bool) (db.Database, error) {
|
||||
resolvedConfig, err := a.resolveConnectionSecrets(config)
|
||||
effectiveConfig, err := a.resolveEffectiveConnectionConfig(config)
|
||||
if err != nil {
|
||||
return nil, wrapConnectError(config, err)
|
||||
return nil, err
|
||||
}
|
||||
effectiveConfig := resolvedConfig
|
||||
isFileDB := isFileDatabaseType(effectiveConfig.Type)
|
||||
|
||||
key := getCacheKey(effectiveConfig)
|
||||
@@ -992,9 +995,9 @@ func (a *App) getDatabaseWithPing(config connection.ConnectionConfig, forcePing
|
||||
}
|
||||
|
||||
initialKey := key
|
||||
dbInst, connectedConfig, err := a.connectDatabaseWithStartupRetry(resolvedConfig)
|
||||
dbInst, connectedConfig, err := a.connectEffectiveDatabaseWithStartupRetry(effectiveConfig)
|
||||
if err != nil {
|
||||
retryInst, retryConfig, retryErr := a.retryConnectAfterMySQLMaxUserConnections(resolvedConfig, connectedConfig, err)
|
||||
retryInst, retryConfig, retryErr := a.retryConnectAfterMySQLMaxUserConnections(effectiveConfig, connectedConfig, err)
|
||||
if retryErr != nil {
|
||||
failedKey := getCacheKey(retryConfig)
|
||||
a.recordConnectFailureByKey(failedKey, retryErr)
|
||||
@@ -1046,7 +1049,7 @@ func (a *App) retryConnectAfterMySQLMaxUserConnections(rawConfig connection.Conn
|
||||
return nil, failedConfig, withMySQLMaxUserConnectionsHint(err, released)
|
||||
}
|
||||
|
||||
dbInst, connectedConfig, retryErr := a.connectDatabaseWithStartupRetry(rawConfig)
|
||||
dbInst, connectedConfig, retryErr := a.connectEffectiveDatabaseWithStartupRetry(rawConfig)
|
||||
if retryErr != nil {
|
||||
if isMySQLMaxUserConnectionsError(retryErr) {
|
||||
return nil, connectedConfig, withMySQLMaxUserConnectionsHint(retryErr, released)
|
||||
@@ -1154,12 +1157,14 @@ func shortenCacheKey(key string) string {
|
||||
}
|
||||
|
||||
func (a *App) connectDatabaseWithStartupRetry(rawConfig connection.ConnectionConfig) (db.Database, connection.ConnectionConfig, error) {
|
||||
resolvedConfig, err := a.resolveConnectionSecrets(rawConfig)
|
||||
effectiveConfig, err := a.resolveEffectiveConnectionConfig(rawConfig)
|
||||
if err != nil {
|
||||
return nil, rawConfig, wrapConnectError(rawConfig, err)
|
||||
return nil, rawConfig, err
|
||||
}
|
||||
rawConfig = resolvedConfig
|
||||
return a.connectEffectiveDatabaseWithStartupRetry(effectiveConfig)
|
||||
}
|
||||
|
||||
func (a *App) connectEffectiveDatabaseWithStartupRetry(rawConfig connection.ConnectionConfig) (db.Database, connection.ConnectionConfig, error) {
|
||||
var lastErr error
|
||||
var lastEffectiveConfig connection.ConnectionConfig
|
||||
|
||||
|
||||
399
internal/app/custom_clickhouse_runtime.go
Normal file
399
internal/app/custom_clickhouse_runtime.go
Normal file
@@ -0,0 +1,399 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"unicode"
|
||||
|
||||
"GoNavi-Wails/internal/connection"
|
||||
)
|
||||
|
||||
const customClickHouseDSNMaxLength = 4096
|
||||
|
||||
type customClickHouseEndpoint struct {
|
||||
host string
|
||||
port int
|
||||
user string
|
||||
password string
|
||||
database string
|
||||
connectionParams string
|
||||
protocol string
|
||||
useSSL bool
|
||||
sslMode string
|
||||
sslCAPath string
|
||||
sslCertPath string
|
||||
sslKeyPath string
|
||||
}
|
||||
|
||||
func (a *App) resolveCustomClickHouseRuntimeConfig(config connection.ConnectionConfig) (connection.ConnectionConfig, error) {
|
||||
if !strings.EqualFold(strings.TrimSpace(config.Type), "custom") ||
|
||||
!strings.EqualFold(strings.TrimSpace(config.Driver), "clickhouse") {
|
||||
return config, nil
|
||||
}
|
||||
|
||||
dsn := strings.TrimSpace(config.DSN)
|
||||
if dsn == "" {
|
||||
return config, fmt.Errorf("%s", a.appText("db.backend.error.custom_clickhouse_dsn_required", nil))
|
||||
}
|
||||
endpoint, ok := parseCustomClickHouseEndpoint(dsn)
|
||||
if !ok {
|
||||
return config, fmt.Errorf("%s", a.appText("db.backend.error.custom_clickhouse_dsn_invalid", nil))
|
||||
}
|
||||
if config.HasRuntimeDatabaseOverride() {
|
||||
endpoint.database = strings.TrimSpace(config.RuntimeDatabaseOverride())
|
||||
}
|
||||
|
||||
runtimeConfig := config
|
||||
runtimeConfig.Type = "clickhouse"
|
||||
runtimeConfig.Driver = ""
|
||||
runtimeConfig.DSN = ""
|
||||
runtimeConfig.URI = ""
|
||||
runtimeConfig.Host = endpoint.host
|
||||
runtimeConfig.Port = endpoint.port
|
||||
runtimeConfig.User = endpoint.user
|
||||
runtimeConfig.Password = endpoint.password
|
||||
runtimeConfig.Database = endpoint.database
|
||||
runtimeConfig.ConnectionParams = endpoint.connectionParams
|
||||
runtimeConfig.ClickHouseProtocol = endpoint.protocol
|
||||
runtimeConfig.UseSSL = endpoint.useSSL
|
||||
runtimeConfig.SSLMode = endpoint.sslMode
|
||||
runtimeConfig.SSLCAPath = endpoint.sslCAPath
|
||||
runtimeConfig.SSLCertPath = endpoint.sslCertPath
|
||||
runtimeConfig.SSLKeyPath = endpoint.sslKeyPath
|
||||
runtimeConfig.Hosts = nil
|
||||
runtimeConfig.Topology = ""
|
||||
runtimeConfig.OceanBaseProtocol = ""
|
||||
runtimeConfig.RedisDB = 0
|
||||
runtimeConfig.RedisSentinelMaster = ""
|
||||
runtimeConfig.RedisSentinelUser = ""
|
||||
runtimeConfig.RedisSentinelPassword = ""
|
||||
runtimeConfig.MySQLReplicaUser = ""
|
||||
runtimeConfig.MySQLReplicaPassword = ""
|
||||
runtimeConfig.ReplicaSet = ""
|
||||
runtimeConfig.AuthSource = ""
|
||||
runtimeConfig.ReadPreference = ""
|
||||
runtimeConfig.MongoSRV = false
|
||||
runtimeConfig.MongoAuthMechanism = ""
|
||||
runtimeConfig.MongoReplicaUser = ""
|
||||
runtimeConfig.MongoReplicaPassword = ""
|
||||
runtimeConfig.JVM = connection.JVMConfig{}
|
||||
runtimeConfig = runtimeConfig.WithoutRuntimeDatabaseOverride()
|
||||
return runtimeConfig, nil
|
||||
}
|
||||
|
||||
func parseCustomClickHouseEndpoint(rawDSN string) (customClickHouseEndpoint, bool) {
|
||||
dsn := strings.TrimSpace(rawDSN)
|
||||
if dsn == "" || len(dsn) > customClickHouseDSNMaxLength || containsControlCharacter(dsn) {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
endpointText, jdbc, ok := normalizeCustomClickHouseEndpointText(dsn)
|
||||
if !ok {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
parsed, err := url.Parse(endpointText)
|
||||
if err != nil || parsed == nil || parsed.Opaque != "" || parsed.Fragment != "" {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
scheme := strings.ToLower(strings.TrimSpace(parsed.Scheme))
|
||||
if scheme != "clickhouse" && scheme != "http" && scheme != "https" {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
if jdbc && scheme != "http" && scheme != "https" {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
if strings.TrimSpace(parsed.Host) == "" || strings.ContainsAny(parsed.Host, ",;") || strings.HasSuffix(parsed.Host, ":") {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
host := strings.TrimSpace(parsed.Hostname())
|
||||
if host == "" || containsControlCharacter(host) || strings.Contains(host, ":") && !strings.HasPrefix(parsed.Host, "[") {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
explicitPort := false
|
||||
port := 0
|
||||
if portText := strings.TrimSpace(parsed.Port()); portText != "" {
|
||||
parsedPort, convErr := strconv.Atoi(portText)
|
||||
if convErr != nil || parsedPort <= 0 || parsedPort > 65535 {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
port = parsedPort
|
||||
explicitPort = true
|
||||
}
|
||||
|
||||
query, err := url.ParseQuery(parsed.RawQuery)
|
||||
if err != nil || connectionValuesContainControlCharacter(query) {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
user := ""
|
||||
password := ""
|
||||
if parsed.User != nil {
|
||||
user = parsed.User.Username()
|
||||
if parsedPassword, hasPassword := parsed.User.Password(); hasPassword {
|
||||
password = parsedPassword
|
||||
}
|
||||
}
|
||||
if containsControlCharacter(user) || containsControlCharacter(password) {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
queryUser, hasQueryUser := popConnectionValue(query, "user")
|
||||
queryUsername, hasQueryUsername := popConnectionValue(query, "username")
|
||||
if hasQueryUser {
|
||||
user = queryUser
|
||||
} else if hasQueryUsername {
|
||||
user = queryUsername
|
||||
}
|
||||
if queryPassword, exists := popConnectionValue(query, "password"); exists {
|
||||
password = queryPassword
|
||||
}
|
||||
|
||||
database := strings.Trim(strings.TrimSpace(parsed.Path), "/")
|
||||
if database != "" && strings.Contains(database, "/") {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
if queryDatabase, exists := popConnectionValue(query, "database"); exists {
|
||||
database = strings.TrimSpace(queryDatabase)
|
||||
}
|
||||
if containsControlCharacter(user) || containsControlCharacter(password) || containsControlCharacter(database) {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
protocolValue, hasProtocol := popConnectionValue(query, "protocol")
|
||||
protocol, protocolTLS, protocolOK := resolveCustomClickHouseProtocol(scheme, jdbc, protocolValue, hasProtocol, port, explicitPort)
|
||||
if !protocolOK {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
secureProtocol := scheme == "https" || protocolTLS
|
||||
useSSL := secureProtocol
|
||||
sslMode := ""
|
||||
if useSSL {
|
||||
sslMode = "required"
|
||||
}
|
||||
for _, key := range []string{"ssl", "secure"} {
|
||||
if rawValue, exists := popConnectionValue(query, key); exists {
|
||||
enabled, known := parseCustomClickHouseBool(rawValue)
|
||||
if !known {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
useSSL = enabled
|
||||
if enabled {
|
||||
sslMode = "required"
|
||||
} else {
|
||||
sslMode = ""
|
||||
}
|
||||
}
|
||||
}
|
||||
if rawMode, exists := popConnectionValue(query, "sslmode"); exists {
|
||||
var modeOK bool
|
||||
useSSL, sslMode, modeOK = normalizeCustomClickHouseSSLMode(rawMode)
|
||||
if !modeOK {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
}
|
||||
if rawSkipVerify, exists := popConnectionValue(query, "skip_verify"); exists {
|
||||
skipVerify, known := parseCustomClickHouseBool(rawSkipVerify)
|
||||
if !known {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
if skipVerify {
|
||||
useSSL = true
|
||||
sslMode = "skip-verify"
|
||||
}
|
||||
}
|
||||
if secureProtocol {
|
||||
useSSL = true
|
||||
if sslMode == "" {
|
||||
sslMode = "required"
|
||||
}
|
||||
}
|
||||
if !useSSL {
|
||||
sslMode = ""
|
||||
}
|
||||
|
||||
sslCAPath := popFirstConnectionValue(query, "sslrootcert", "ssl_ca", "ca_cert")
|
||||
sslCertPath := popFirstConnectionValue(query, "sslcert", "ssl_cert", "client_cert")
|
||||
sslKeyPath := popFirstConnectionValue(query, "sslkey", "ssl_key", "client_key")
|
||||
if containsControlCharacter(sslCAPath) || containsControlCharacter(sslCertPath) || containsControlCharacter(sslKeyPath) {
|
||||
return customClickHouseEndpoint{}, false
|
||||
}
|
||||
|
||||
if !explicitPort {
|
||||
switch {
|
||||
case protocol == "http" && useSSL:
|
||||
port = 8443
|
||||
case protocol == "http":
|
||||
port = 8123
|
||||
default:
|
||||
port = 9000
|
||||
}
|
||||
}
|
||||
|
||||
return customClickHouseEndpoint{
|
||||
host: host,
|
||||
port: port,
|
||||
user: user,
|
||||
password: password,
|
||||
database: database,
|
||||
connectionParams: query.Encode(),
|
||||
protocol: protocol,
|
||||
useSSL: useSSL,
|
||||
sslMode: sslMode,
|
||||
sslCAPath: strings.TrimSpace(sslCAPath),
|
||||
sslCertPath: strings.TrimSpace(sslCertPath),
|
||||
sslKeyPath: strings.TrimSpace(sslKeyPath),
|
||||
}, true
|
||||
}
|
||||
|
||||
func normalizeCustomClickHouseEndpointText(dsn string) (string, bool, bool) {
|
||||
for _, prefix := range []string{"jdbc:clickhouse:", "jdbc:ch:"} {
|
||||
if len(dsn) >= len(prefix) && strings.EqualFold(dsn[:len(prefix)], prefix) {
|
||||
remainder := strings.TrimSpace(dsn[len(prefix):])
|
||||
switch {
|
||||
case strings.HasPrefix(remainder, "//"):
|
||||
return "http:" + remainder, true, true
|
||||
case hasEndpointScheme(remainder, "http"), hasEndpointScheme(remainder, "https"):
|
||||
return remainder, true, true
|
||||
default:
|
||||
return "", true, false
|
||||
}
|
||||
}
|
||||
}
|
||||
if strings.HasPrefix(strings.ToLower(dsn), "jdbc:") {
|
||||
return "", false, false
|
||||
}
|
||||
for _, scheme := range []string{"clickhouse", "http", "https"} {
|
||||
if hasEndpointScheme(dsn, scheme) {
|
||||
return dsn, false, true
|
||||
}
|
||||
}
|
||||
return "", false, false
|
||||
}
|
||||
|
||||
func hasEndpointScheme(value string, scheme string) bool {
|
||||
prefix := scheme + "://"
|
||||
return len(value) >= len(prefix) && strings.EqualFold(value[:len(prefix)], prefix)
|
||||
}
|
||||
|
||||
func resolveCustomClickHouseProtocol(scheme string, jdbc bool, rawProtocol string, hasProtocol bool, port int, explicitPort bool) (string, bool, bool) {
|
||||
normalizedProtocol := strings.ToLower(strings.TrimSpace(rawProtocol))
|
||||
if jdbc || scheme == "http" || scheme == "https" {
|
||||
if hasProtocol && normalizedProtocol != "" && normalizedProtocol != "http" && normalizedProtocol != "https" {
|
||||
return "", false, false
|
||||
}
|
||||
return "http", scheme == "https" || normalizedProtocol == "https", true
|
||||
}
|
||||
if hasProtocol {
|
||||
switch normalizedProtocol {
|
||||
case "", "auto":
|
||||
return "", false, true
|
||||
case "http":
|
||||
return "http", false, true
|
||||
case "https":
|
||||
return "http", true, true
|
||||
case "native", "tcp":
|
||||
return "native", false, true
|
||||
default:
|
||||
return "", false, false
|
||||
}
|
||||
}
|
||||
if explicitPort && isCustomClickHouseHTTPPort(port) {
|
||||
return "http", false, true
|
||||
}
|
||||
return "", false, true
|
||||
}
|
||||
|
||||
func isCustomClickHouseHTTPPort(port int) bool {
|
||||
switch port {
|
||||
case 8123, 8125, 8132, 8443:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func popConnectionValue(values url.Values, target string) (string, bool) {
|
||||
keys := make([]string, 0, len(values))
|
||||
for key := range values {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
value := ""
|
||||
found := false
|
||||
for _, key := range keys {
|
||||
if !strings.EqualFold(strings.TrimSpace(key), target) {
|
||||
continue
|
||||
}
|
||||
items := values[key]
|
||||
if len(items) > 0 {
|
||||
value = items[len(items)-1]
|
||||
}
|
||||
delete(values, key)
|
||||
found = true
|
||||
}
|
||||
return value, found
|
||||
}
|
||||
|
||||
func popFirstConnectionValue(values url.Values, targets ...string) string {
|
||||
selected := ""
|
||||
hasSelected := false
|
||||
for _, target := range targets {
|
||||
value, exists := popConnectionValue(values, target)
|
||||
if exists && !hasSelected {
|
||||
selected = value
|
||||
hasSelected = true
|
||||
}
|
||||
}
|
||||
return selected
|
||||
}
|
||||
|
||||
func parseCustomClickHouseBool(raw string) (bool, bool) {
|
||||
switch strings.ToLower(strings.TrimSpace(raw)) {
|
||||
case "1", "true", "yes", "on", "enabled":
|
||||
return true, true
|
||||
case "0", "false", "no", "off", "disabled":
|
||||
return false, true
|
||||
default:
|
||||
return false, false
|
||||
}
|
||||
}
|
||||
|
||||
func normalizeCustomClickHouseSSLMode(raw string) (bool, string, bool) {
|
||||
switch strings.ToLower(strings.TrimSpace(raw)) {
|
||||
case "strict", "required", "require", "verify-ca", "verify-full", "on", "true":
|
||||
return true, "required", true
|
||||
case "skip-verify", "skip_verify", "insecure", "insecure-skip-verify":
|
||||
return true, "skip-verify", true
|
||||
case "preferred", "prefer", "allow":
|
||||
return true, "preferred", true
|
||||
case "disable", "disabled", "none", "off", "false":
|
||||
return false, "", true
|
||||
default:
|
||||
return false, "", false
|
||||
}
|
||||
}
|
||||
|
||||
func connectionValuesContainControlCharacter(values url.Values) bool {
|
||||
for key, items := range values {
|
||||
if containsControlCharacter(key) {
|
||||
return true
|
||||
}
|
||||
for _, item := range items {
|
||||
if containsControlCharacter(item) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func containsControlCharacter(value string) bool {
|
||||
return strings.IndexFunc(value, unicode.IsControl) >= 0
|
||||
}
|
||||
753
internal/app/custom_clickhouse_runtime_test.go
Normal file
753
internal/app/custom_clickhouse_runtime_test.go
Normal file
@@ -0,0 +1,753 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"GoNavi-Wails/internal/connection"
|
||||
"GoNavi-Wails/internal/db"
|
||||
)
|
||||
|
||||
type customClickHouseRecordingDB struct {
|
||||
fakeStartupRetryDB
|
||||
closeCalls int
|
||||
}
|
||||
|
||||
func (d *customClickHouseRecordingDB) Close() error {
|
||||
d.closeCalls++
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestResolveEffectiveConnectionConfigCanonicalizesCustomClickHouseJDBCDSN(t *testing.T) {
|
||||
a := NewApp()
|
||||
proxy := connection.ProxyConfig{
|
||||
Type: "socks5",
|
||||
Host: "proxy.internal",
|
||||
Port: 1080,
|
||||
}
|
||||
raw := connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: " ClickHouse ",
|
||||
DSN: "jdbc:clickhouse://alice:p%40ss@[2001:db8::1]:8443/analytics?compress=lz4&ssl=true",
|
||||
Host: "stale.example.com",
|
||||
Port: 3306,
|
||||
User: "stale-user",
|
||||
Password: "stale-password",
|
||||
Database: "stale-database",
|
||||
URI: "mysql://stale.example.com:3306/stale-database",
|
||||
ConnectionParams: "stale=true",
|
||||
ClickHouseProtocol: "native",
|
||||
Hosts: []string{"stale-replica.example.com:3306"},
|
||||
Topology: "replica",
|
||||
SSLCAPath: "stale-ca.pem",
|
||||
SSLCertPath: "stale-cert.pem",
|
||||
SSLKeyPath: "stale-key.pem",
|
||||
Timeout: 42,
|
||||
UseProxy: true,
|
||||
Proxy: proxy,
|
||||
}
|
||||
|
||||
got, err := a.resolveEffectiveConnectionConfig(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("resolveEffectiveConnectionConfig returned error: %v", err)
|
||||
}
|
||||
if got.Type != "clickhouse" {
|
||||
t.Fatalf("expected runtime type clickhouse, got %q", got.Type)
|
||||
}
|
||||
if got.Driver != "" || got.DSN != "" {
|
||||
t.Fatalf("expected custom driver fields to be removed from runtime config, got driver=%q dsn=%q", got.Driver, got.DSN)
|
||||
}
|
||||
if got.URI != "" {
|
||||
t.Fatalf("expected runtime URI to be cleared after extracting fields, got %q", got.URI)
|
||||
}
|
||||
if got.Host != "2001:db8::1" || got.Port != 8443 {
|
||||
t.Fatalf("unexpected endpoint: host=%q port=%d", got.Host, got.Port)
|
||||
}
|
||||
if got.User != "alice" || got.Password != "p@ss" || got.Database != "analytics" {
|
||||
t.Fatalf("unexpected credentials/database mapping: user=%q password=%q database=%q", got.User, got.Password, got.Database)
|
||||
}
|
||||
if got.ClickHouseProtocol != "http" {
|
||||
t.Fatalf("expected JDBC DSN to force HTTP even on a non-standard port, got %q", got.ClickHouseProtocol)
|
||||
}
|
||||
if !got.UseSSL || got.SSLMode != "required" {
|
||||
t.Fatalf("expected ssl=true to enable required TLS, got useSSL=%v sslMode=%q", got.UseSSL, got.SSLMode)
|
||||
}
|
||||
if got.SSLCAPath != "" || got.SSLCertPath != "" || got.SSLKeyPath != "" {
|
||||
t.Fatalf("expected hidden stale TLS paths to be cleared, got ca=%q cert=%q key=%q", got.SSLCAPath, got.SSLCertPath, got.SSLKeyPath)
|
||||
}
|
||||
if got.ConnectionParams != "compress=lz4" {
|
||||
t.Fatalf("expected non-connection query params to be preserved, got %q", got.ConnectionParams)
|
||||
}
|
||||
if len(got.Hosts) != 0 || got.Topology != "" {
|
||||
t.Fatalf("expected stale topology fields to be cleared, got hosts=%v topology=%q", got.Hosts, got.Topology)
|
||||
}
|
||||
if got.Timeout != 42 || !got.UseProxy || !reflect.DeepEqual(got.Proxy, proxy) {
|
||||
t.Fatalf("expected runtime-neutral settings to be preserved, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveEffectiveConnectionConfigAcceptsOfficialClickHouseJDBCHTTPSAlias(t *testing.T) {
|
||||
a := NewApp()
|
||||
raw := connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:ch:https://reporter:secret@clickhouse.example.com:8443/default?skip_verify=true&max_open_conns=8",
|
||||
}
|
||||
|
||||
got, err := a.resolveEffectiveConnectionConfig(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("resolveEffectiveConnectionConfig returned error: %v", err)
|
||||
}
|
||||
if got.URI != "" {
|
||||
t.Fatalf("expected runtime URI to be cleared, got %q", got.URI)
|
||||
}
|
||||
if got.ClickHouseProtocol != "http" {
|
||||
t.Fatalf("expected HTTPS JDBC alias to select HTTP protocol, got %q", got.ClickHouseProtocol)
|
||||
}
|
||||
if !got.UseSSL || got.SSLMode != "skip-verify" {
|
||||
t.Fatalf("expected HTTPS skip_verify mapping, got useSSL=%v sslMode=%q", got.UseSSL, got.SSLMode)
|
||||
}
|
||||
if got.ConnectionParams != "max_open_conns=8" {
|
||||
t.Fatalf("unexpected connection params: %q", got.ConnectionParams)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveEffectiveConnectionConfigUsesJDBCHTTPDefaultsAndQueryOverrides(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
dsn string
|
||||
port int
|
||||
useSSL bool
|
||||
sslMode string
|
||||
}{
|
||||
{
|
||||
name: "plain JDBC defaults to HTTP 8123",
|
||||
dsn: "jdbc:clickhouse://url-user:url-pass@clickhouse.example.com/path_db?user=query-user&password=query%40pass&database=query_db",
|
||||
port: 8123,
|
||||
},
|
||||
{
|
||||
name: "explicit non-standard port remains HTTP",
|
||||
dsn: "jdbc:ch://clickhouse.example.com:9000/default",
|
||||
port: 9000,
|
||||
},
|
||||
{
|
||||
name: "HTTPS alias defaults to 8443",
|
||||
dsn: "jdbc:clickhouse:https://clickhouse.example.com/default",
|
||||
port: 8443,
|
||||
useSSL: true,
|
||||
sslMode: "required",
|
||||
},
|
||||
{
|
||||
name: "HTTPS protocol parameter cannot downgrade to plaintext",
|
||||
dsn: "jdbc:clickhouse://clickhouse.example.com/default?protocol=https",
|
||||
port: 8443,
|
||||
useSSL: true,
|
||||
sslMode: "required",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
a := NewApp()
|
||||
got, err := a.resolveEffectiveConnectionConfig(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: tt.dsn,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("resolveEffectiveConnectionConfig returned error: %v", err)
|
||||
}
|
||||
if got.Host != "clickhouse.example.com" || got.Port != tt.port {
|
||||
t.Fatalf("unexpected endpoint: host=%q port=%d", got.Host, got.Port)
|
||||
}
|
||||
if got.ClickHouseProtocol != "http" {
|
||||
t.Fatalf("expected JDBC protocol HTTP, got %q", got.ClickHouseProtocol)
|
||||
}
|
||||
if got.UseSSL != tt.useSSL || got.SSLMode != tt.sslMode {
|
||||
t.Fatalf("unexpected TLS mapping: useSSL=%v sslMode=%q", got.UseSSL, got.SSLMode)
|
||||
}
|
||||
if tt.name == "plain JDBC defaults to HTTP 8123" {
|
||||
if got.User != "query-user" || got.Password != "query@pass" || got.Database != "query_db" {
|
||||
t.Fatalf("expected query properties to override URL credentials/database, got user=%q password=%q database=%q", got.User, got.Password, got.Database)
|
||||
}
|
||||
if got.ConnectionParams != "" {
|
||||
t.Fatalf("expected connection-only query properties to be removed, got %q", got.ConnectionParams)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveEffectiveConnectionConfigSupportsNativeAndHTTPClickHouseDSN(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
dsn string
|
||||
port int
|
||||
protocol string
|
||||
useSSL bool
|
||||
sslMode string
|
||||
}{
|
||||
{name: "native", dsn: "clickhouse://clickhouse.example.com/analytics", port: 9000},
|
||||
{name: "native HTTP port inference", dsn: "clickhouse://clickhouse.example.com:8123/analytics", port: 8123, protocol: "http"},
|
||||
{name: "HTTP", dsn: "http://clickhouse.example.com/analytics", port: 8123, protocol: "http"},
|
||||
{name: "HTTPS", dsn: "https://clickhouse.example.com/analytics", port: 8443, protocol: "http", useSSL: true, sslMode: "required"},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
a := NewApp()
|
||||
got, err := a.resolveEffectiveConnectionConfig(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: tt.dsn,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("resolveEffectiveConnectionConfig returned error: %v", err)
|
||||
}
|
||||
if got.Host != "clickhouse.example.com" || got.Port != tt.port || got.Database != "analytics" {
|
||||
t.Fatalf("unexpected endpoint mapping: host=%q port=%d database=%q", got.Host, got.Port, got.Database)
|
||||
}
|
||||
if got.ClickHouseProtocol != tt.protocol || got.UseSSL != tt.useSSL || got.SSLMode != tt.sslMode {
|
||||
t.Fatalf("unexpected protocol/TLS mapping: protocol=%q useSSL=%v sslMode=%q", got.ClickHouseProtocol, got.UseSSL, got.SSLMode)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveEffectiveConnectionConfigRejectsInvalidCustomClickHouseDSN(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
dsn string
|
||||
}{
|
||||
{name: "empty", dsn: ""},
|
||||
{name: "wrong scheme", dsn: "jdbc:mysql://db.example.com:3306/app"},
|
||||
{name: "missing host", dsn: "jdbc:clickhouse:///analytics"},
|
||||
{name: "invalid port", dsn: "jdbc:clickhouse://db.example.com:not-a-port/analytics"},
|
||||
{name: "port out of range", dsn: "jdbc:clickhouse://db.example.com:65536/analytics"},
|
||||
{name: "unbracketed IPv6", dsn: "jdbc:clickhouse://2001:db8::1:8123/analytics"},
|
||||
{name: "unsupported grpc", dsn: "jdbc:ch:grpc://db.example.com/analytics"},
|
||||
{name: "unsupported multiple hosts", dsn: "jdbc:clickhouse://db-1.example.com:8123,db-2.example.com:8123/analytics"},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
a := NewApp()
|
||||
_, err := a.resolveEffectiveConnectionConfig(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: tt.dsn,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatal("expected invalid custom ClickHouse DSN to be rejected")
|
||||
}
|
||||
if !strings.Contains(strings.ToLower(err.Error()), "clickhouse") {
|
||||
t.Fatalf("expected ClickHouse-specific error, got %q", err.Error())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveEffectiveConnectionConfigDoesNotExposeInvalidClickHouseDSN(t *testing.T) {
|
||||
a := NewApp()
|
||||
secretDSN := "jdbc:clickhouse://admin:super-secret@db.example.com:not-a-port/analytics"
|
||||
_, err := a.resolveEffectiveConnectionConfig(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: secretDSN,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatal("expected invalid custom ClickHouse DSN to be rejected")
|
||||
}
|
||||
message := err.Error()
|
||||
for _, secret := range []string{secretDSN, "super-secret", "admin"} {
|
||||
if strings.Contains(message, secret) {
|
||||
t.Fatalf("invalid DSN error leaked %q: %q", secret, message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveEffectiveConnectionConfigLeavesOtherCustomDriversUntouched(t *testing.T) {
|
||||
a := NewApp()
|
||||
raw := connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "mysql",
|
||||
DSN: "root:secret@tcp(db.example.com:3306)/app",
|
||||
}
|
||||
|
||||
got, err := a.resolveEffectiveConnectionConfig(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("resolveEffectiveConnectionConfig returned error: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(got, raw) {
|
||||
t.Fatalf("non-ClickHouse custom config changed:\nwant=%+v\n got=%+v", raw, got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeRunConfigCarriesCustomClickHouseDatabaseOverride(t *testing.T) {
|
||||
a := NewApp()
|
||||
raw := connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:clickhouse://clickhouse.example.com:8123/default",
|
||||
Database: "stale-hidden-database",
|
||||
}
|
||||
|
||||
direct, err := a.resolveEffectiveConnectionConfig(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("direct resolve returned error: %v", err)
|
||||
}
|
||||
if direct.Database != "default" {
|
||||
t.Fatalf("expected DSN database to override stale hidden field, got %q", direct.Database)
|
||||
}
|
||||
|
||||
runConfig := normalizeRunConfig(raw, "analytics")
|
||||
effective, err := a.resolveEffectiveConnectionConfig(runConfig)
|
||||
if err != nil {
|
||||
t.Fatalf("run config resolve returned error: %v", err)
|
||||
}
|
||||
if effective.Database != "analytics" {
|
||||
t.Fatalf("expected selected database override analytics, got %q", effective.Database)
|
||||
}
|
||||
if effective.RuntimeDatabaseOverride() != "" {
|
||||
t.Fatalf("expected runtime database marker to be consumed, got %q", effective.RuntimeDatabaseOverride())
|
||||
}
|
||||
if effective.HasRuntimeDatabaseOverride() {
|
||||
t.Fatal("expected runtime database marker state to be consumed")
|
||||
}
|
||||
|
||||
serverLevel, err := a.resolveEffectiveConnectionConfig(raw.WithRuntimeDatabaseOverride(""))
|
||||
if err != nil {
|
||||
t.Fatalf("server-level config resolve returned error: %v", err)
|
||||
}
|
||||
if serverLevel.Database != "" {
|
||||
t.Fatalf("expected explicit empty override to clear DSN database, got %q", serverLevel.Database)
|
||||
}
|
||||
|
||||
ddlConfig := buildRunConfigForDDL(raw, "clickhouse", "reporting")
|
||||
ddlEffective, err := a.resolveEffectiveConnectionConfig(ddlConfig)
|
||||
if err != nil {
|
||||
t.Fatalf("DDL config resolve returned error: %v", err)
|
||||
}
|
||||
if ddlEffective.Database != "reporting" {
|
||||
t.Fatalf("expected DDL database override reporting, got %q", ddlEffective.Database)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCustomClickHouseDatabaseDDLConnectsAtServerLevel(t *testing.T) {
|
||||
originalNewDatabaseFunc := newDatabaseFunc
|
||||
originalDriverRuntimeSupportStatusFunc := driverRuntimeSupportStatusFunc
|
||||
originalVerifyDriverAgentRevisionFunc := verifyDriverAgentRevisionFunc
|
||||
originalResolveDialConfigWithProxyFunc := resolveDialConfigWithProxyFunc
|
||||
t.Cleanup(func() {
|
||||
newDatabaseFunc = originalNewDatabaseFunc
|
||||
driverRuntimeSupportStatusFunc = originalDriverRuntimeSupportStatusFunc
|
||||
verifyDriverAgentRevisionFunc = originalVerifyDriverAgentRevisionFunc
|
||||
resolveDialConfigWithProxyFunc = originalResolveDialConfigWithProxyFunc
|
||||
})
|
||||
|
||||
driverRuntimeSupportStatusFunc = func(dbType string) (bool, string) { return true, "" }
|
||||
verifyDriverAgentRevisionFunc = func(config connection.ConnectionConfig) error { return nil }
|
||||
resolveDialConfigWithProxyFunc = func(config connection.ConnectionConfig) (connection.ConnectionConfig, error) {
|
||||
return config, nil
|
||||
}
|
||||
|
||||
raw := connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:clickhouse://clickhouse.example.com:8123/analytics",
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
run func(*App) connection.QueryResult
|
||||
wantQuery string
|
||||
}{
|
||||
{
|
||||
name: "create database",
|
||||
run: func(a *App) connection.QueryResult {
|
||||
return a.CreateDatabase(raw, "reporting")
|
||||
},
|
||||
wantQuery: "CREATE DATABASE IF NOT EXISTS `reporting`",
|
||||
},
|
||||
{
|
||||
name: "drop current DSN database",
|
||||
run: func(a *App) connection.QueryResult {
|
||||
return a.DropDatabase(raw, "analytics")
|
||||
},
|
||||
wantQuery: "DROP DATABASE `analytics`",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
fakeDB := &fakeCreateDatabaseDB{}
|
||||
newDatabaseFunc = func(dbType string) (db.Database, error) {
|
||||
if dbType != "clickhouse" {
|
||||
t.Fatalf("expected ClickHouse factory, got %q", dbType)
|
||||
}
|
||||
return fakeDB, nil
|
||||
}
|
||||
|
||||
result := tt.run(NewApp())
|
||||
if !result.Success {
|
||||
t.Fatalf("database DDL failed: %s", result.Message)
|
||||
}
|
||||
if fakeDB.connectConfig.Database != "" {
|
||||
t.Fatalf("expected server-level connection, got database %q", fakeDB.connectConfig.Database)
|
||||
}
|
||||
if len(fakeDB.execQueries) != 1 || fakeDB.execQueries[0] != tt.wantQuery {
|
||||
t.Fatalf("unexpected DDL query: %#v", fakeDB.execQueries)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveEffectiveConnectionConfigLoadsSavedOpaqueClickHouseDSN(t *testing.T) {
|
||||
store := newFakeAppSecretStore()
|
||||
a := NewAppWithSecretStore(store)
|
||||
a.configDir = t.TempDir()
|
||||
|
||||
view, err := a.SaveConnection(connection.SavedConnectionInput{
|
||||
ID: "custom-clickhouse-secret",
|
||||
Name: "Custom ClickHouse",
|
||||
Config: connection.ConnectionConfig{
|
||||
ID: "custom-clickhouse-secret",
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:clickhouse://secret-user:secret-password@clickhouse.example.com:8123/analytics",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("SaveConnection returned error: %v", err)
|
||||
}
|
||||
if view.Config.DSN != "" || !view.HasOpaqueDSN {
|
||||
t.Fatalf("expected saved view to keep the ClickHouse DSN opaque, got dsn=%q hasOpaque=%v", view.Config.DSN, view.HasOpaqueDSN)
|
||||
}
|
||||
|
||||
effective, err := a.resolveEffectiveConnectionConfig(view.Config)
|
||||
if err != nil {
|
||||
t.Fatalf("resolveEffectiveConnectionConfig returned error: %v", err)
|
||||
}
|
||||
if effective.Type != "clickhouse" || effective.Host != "clickhouse.example.com" || effective.Port != 8123 {
|
||||
t.Fatalf("unexpected effective ClickHouse endpoint: %+v", effective)
|
||||
}
|
||||
if effective.User != "secret-user" || effective.Password != "secret-password" || effective.Database != "analytics" {
|
||||
t.Fatalf("saved opaque DSN was not restored before conversion: user=%q password=%q database=%q", effective.User, effective.Password, effective.Database)
|
||||
}
|
||||
if effective.DSN != "" || effective.URI != "" {
|
||||
t.Fatalf("expected restored DSN to be removed from runtime config, got dsn=%q uri=%q", effective.DSN, effective.URI)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenDatabaseIsolatedRoutesCustomClickHouseThroughOptionalDriverPipeline(t *testing.T) {
|
||||
originalNewDatabaseFunc := newDatabaseFunc
|
||||
originalDriverRuntimeSupportStatusFunc := driverRuntimeSupportStatusFunc
|
||||
originalVerifyDriverAgentRevisionFunc := verifyDriverAgentRevisionFunc
|
||||
originalResolveDialConfigWithProxyFunc := resolveDialConfigWithProxyFunc
|
||||
t.Cleanup(func() {
|
||||
newDatabaseFunc = originalNewDatabaseFunc
|
||||
driverRuntimeSupportStatusFunc = originalDriverRuntimeSupportStatusFunc
|
||||
verifyDriverAgentRevisionFunc = originalVerifyDriverAgentRevisionFunc
|
||||
resolveDialConfigWithProxyFunc = originalResolveDialConfigWithProxyFunc
|
||||
})
|
||||
|
||||
var supportType string
|
||||
var revisionConfig connection.ConnectionConfig
|
||||
var factoryType string
|
||||
var dialConfig connection.ConnectionConfig
|
||||
var connectConfig connection.ConnectionConfig
|
||||
driverRuntimeSupportStatusFunc = func(dbType string) (bool, string) {
|
||||
supportType = dbType
|
||||
return true, ""
|
||||
}
|
||||
verifyDriverAgentRevisionFunc = func(config connection.ConnectionConfig) error {
|
||||
revisionConfig = config
|
||||
return nil
|
||||
}
|
||||
newDatabaseFunc = func(dbType string) (db.Database, error) {
|
||||
factoryType = dbType
|
||||
return &fakeStartupRetryDB{connect: func(config connection.ConnectionConfig) error {
|
||||
connectConfig = config
|
||||
return nil
|
||||
}}, nil
|
||||
}
|
||||
resolveDialConfigWithProxyFunc = func(config connection.ConnectionConfig) (connection.ConnectionConfig, error) {
|
||||
dialConfig = config
|
||||
return config, nil
|
||||
}
|
||||
|
||||
a := NewApp()
|
||||
inst, err := a.openDatabaseIsolated(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:clickhouse://db.example.com:9000/analytics",
|
||||
UseProxy: true,
|
||||
Proxy: connection.ProxyConfig{
|
||||
Type: "socks5",
|
||||
Host: "proxy.example.com",
|
||||
Port: 1080,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("openDatabaseIsolated returned error: %v", err)
|
||||
}
|
||||
if inst == nil {
|
||||
t.Fatal("expected database instance")
|
||||
}
|
||||
for name, got := range map[string]string{
|
||||
"support": supportType,
|
||||
"revision": revisionConfig.Type,
|
||||
"factory": factoryType,
|
||||
"dial": dialConfig.Type,
|
||||
"connect": connectConfig.Type,
|
||||
} {
|
||||
if got != "clickhouse" {
|
||||
t.Fatalf("expected %s stage to use clickhouse, got %q", name, got)
|
||||
}
|
||||
}
|
||||
if dialConfig.Host != "db.example.com" || dialConfig.Port != 9000 {
|
||||
t.Fatalf("proxy preparation received unresolved endpoint: %+v", dialConfig)
|
||||
}
|
||||
if dialConfig.ClickHouseProtocol != "http" || connectConfig.ClickHouseProtocol != "http" {
|
||||
t.Fatalf("expected JDBC HTTP protocol to stay pinned through proxy/connect stages: dial=%q connect=%q", dialConfig.ClickHouseProtocol, connectConfig.ClickHouseProtocol)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetDatabaseReusesCanonicalCacheForEquivalentClickHouseJDBCDSN(t *testing.T) {
|
||||
originalNewDatabaseFunc := newDatabaseFunc
|
||||
originalDriverRuntimeSupportStatusFunc := driverRuntimeSupportStatusFunc
|
||||
originalVerifyDriverAgentRevisionFunc := verifyDriverAgentRevisionFunc
|
||||
originalResolveDialConfigWithProxyFunc := resolveDialConfigWithProxyFunc
|
||||
t.Cleanup(func() {
|
||||
newDatabaseFunc = originalNewDatabaseFunc
|
||||
driverRuntimeSupportStatusFunc = originalDriverRuntimeSupportStatusFunc
|
||||
verifyDriverAgentRevisionFunc = originalVerifyDriverAgentRevisionFunc
|
||||
resolveDialConfigWithProxyFunc = originalResolveDialConfigWithProxyFunc
|
||||
})
|
||||
|
||||
factoryCalls := 0
|
||||
connectCalls := 0
|
||||
instance := &fakeStartupRetryDB{connect: func(config connection.ConnectionConfig) error {
|
||||
connectCalls++
|
||||
if config.Type != "clickhouse" || config.ClickHouseProtocol != "http" {
|
||||
t.Fatalf("unexpected connect config: %+v", config)
|
||||
}
|
||||
return nil
|
||||
}}
|
||||
driverRuntimeSupportStatusFunc = func(dbType string) (bool, string) {
|
||||
if dbType != "clickhouse" {
|
||||
t.Fatalf("support check used unexpected type %q", dbType)
|
||||
}
|
||||
return true, ""
|
||||
}
|
||||
verifyDriverAgentRevisionFunc = func(config connection.ConnectionConfig) error {
|
||||
if config.Type != "clickhouse" {
|
||||
t.Fatalf("revision check used unexpected type %q", config.Type)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
newDatabaseFunc = func(dbType string) (db.Database, error) {
|
||||
factoryCalls++
|
||||
if dbType != "clickhouse" {
|
||||
t.Fatalf("factory used unexpected type %q", dbType)
|
||||
}
|
||||
return instance, nil
|
||||
}
|
||||
resolveDialConfigWithProxyFunc = func(config connection.ConnectionConfig) (connection.ConnectionConfig, error) {
|
||||
return config, nil
|
||||
}
|
||||
|
||||
a := NewApp()
|
||||
first, err := a.getDatabase(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:clickhouse://alice:secret@clickhouse.example.com:8123/analytics?compress=lz4",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("first getDatabase returned error: %v", err)
|
||||
}
|
||||
second, err := a.getDatabase(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "CLICKHOUSE",
|
||||
DSN: "jdbc:ch:http://clickhouse.example.com:8123/analytics?password=secret&user=alice&compress=lz4",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("second getDatabase returned error: %v", err)
|
||||
}
|
||||
if first != second {
|
||||
t.Fatal("expected equivalent JDBC DSNs to reuse the same cached instance")
|
||||
}
|
||||
if factoryCalls != 1 || connectCalls != 1 || len(a.dbCache) != 1 {
|
||||
t.Fatalf("expected one canonical cached connection, got factory=%d connect=%d cache=%d", factoryCalls, connectCalls, len(a.dbCache))
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetDatabaseSavedOpaqueClickHouseDSNReusesAndReleasesCanonicalCache(t *testing.T) {
|
||||
originalNewDatabaseFunc := newDatabaseFunc
|
||||
originalDriverRuntimeSupportStatusFunc := driverRuntimeSupportStatusFunc
|
||||
originalVerifyDriverAgentRevisionFunc := verifyDriverAgentRevisionFunc
|
||||
originalResolveDialConfigWithProxyFunc := resolveDialConfigWithProxyFunc
|
||||
t.Cleanup(func() {
|
||||
newDatabaseFunc = originalNewDatabaseFunc
|
||||
driverRuntimeSupportStatusFunc = originalDriverRuntimeSupportStatusFunc
|
||||
verifyDriverAgentRevisionFunc = originalVerifyDriverAgentRevisionFunc
|
||||
resolveDialConfigWithProxyFunc = originalResolveDialConfigWithProxyFunc
|
||||
})
|
||||
|
||||
store := newFakeAppSecretStore()
|
||||
a := NewAppWithSecretStore(store)
|
||||
a.configDir = t.TempDir()
|
||||
view, err := a.SaveConnection(connection.SavedConnectionInput{
|
||||
ID: "saved-custom-clickhouse",
|
||||
Name: "Saved Custom ClickHouse",
|
||||
Config: connection.ConnectionConfig{
|
||||
ID: "saved-custom-clickhouse",
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:clickhouse://alice:secret@clickhouse.example.com:8123/analytics",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("SaveConnection returned error: %v", err)
|
||||
}
|
||||
|
||||
factoryCalls := 0
|
||||
connectCalls := 0
|
||||
recording := &customClickHouseRecordingDB{fakeStartupRetryDB: fakeStartupRetryDB{
|
||||
connect: func(config connection.ConnectionConfig) error {
|
||||
connectCalls++
|
||||
if config.Type != "clickhouse" || config.DSN != "" || config.URI != "" {
|
||||
t.Fatalf("saved opaque DSN leaked back into connect config: %+v", config)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}}
|
||||
driverRuntimeSupportStatusFunc = func(dbType string) (bool, string) { return true, "" }
|
||||
verifyDriverAgentRevisionFunc = func(config connection.ConnectionConfig) error { return nil }
|
||||
newDatabaseFunc = func(dbType string) (db.Database, error) {
|
||||
factoryCalls++
|
||||
return recording, nil
|
||||
}
|
||||
resolveDialConfigWithProxyFunc = func(config connection.ConnectionConfig) (connection.ConnectionConfig, error) {
|
||||
return config, nil
|
||||
}
|
||||
|
||||
first, err := a.getDatabase(view.Config)
|
||||
if err != nil {
|
||||
t.Fatalf("first getDatabase returned error: %v", err)
|
||||
}
|
||||
second, err := a.getDatabase(view.Config)
|
||||
if err != nil {
|
||||
t.Fatalf("second getDatabase returned error: %v", err)
|
||||
}
|
||||
if first != second || factoryCalls != 1 || connectCalls != 1 || len(a.dbCache) != 1 {
|
||||
t.Fatalf("expected one reusable canonical connection, same=%v factory=%d connect=%d cache=%d", first == second, factoryCalls, connectCalls, len(a.dbCache))
|
||||
}
|
||||
for _, entry := range a.dbCache {
|
||||
if entry.config.DSN != "" || entry.config.URI != "" || entry.config.Type != "clickhouse" {
|
||||
t.Fatalf("cache retained non-canonical opaque DSN config: %+v", entry.config)
|
||||
}
|
||||
}
|
||||
|
||||
result := a.DBReleaseConnection(view.Config)
|
||||
if !result.Success {
|
||||
t.Fatalf("DBReleaseConnection failed: %s", result.Message)
|
||||
}
|
||||
if recording.closeCalls != 1 || len(a.dbCache) != 0 {
|
||||
t.Fatalf("expected saved canonical connection to be released once, close=%d cache=%d", recording.closeCalls, len(a.dbCache))
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveDataSyncEndpointConfigCanonicalizesCustomClickHouse(t *testing.T) {
|
||||
a := NewApp()
|
||||
effective, selectedDatabase, err := a.resolveDataSyncEndpointConfig(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:clickhouse://clickhouse.example.com:8123/default",
|
||||
}, "analytics")
|
||||
if err != nil {
|
||||
t.Fatalf("resolveDataSyncEndpointConfig returned error: %v", err)
|
||||
}
|
||||
if effective.Type != "clickhouse" || effective.Host != "clickhouse.example.com" || effective.Port != 8123 {
|
||||
t.Fatalf("unexpected data sync ClickHouse endpoint: %+v", effective)
|
||||
}
|
||||
if effective.DSN != "" || effective.ClickHouseProtocol != "http" {
|
||||
t.Fatalf("data sync endpoint was not canonicalized: %+v", effective)
|
||||
}
|
||||
if selectedDatabase != "analytics" {
|
||||
t.Fatalf("expected selected sync database analytics, got %q", selectedDatabase)
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifyOptionalDriverAgentReadyForExportRecognizesCustomClickHouse(t *testing.T) {
|
||||
originalProbe := optionalDriverAgentMetadataProbe
|
||||
originalResolvePath := resolveOptionalDriverAgentExecutablePathFunc
|
||||
t.Cleanup(func() {
|
||||
optionalDriverAgentMetadataProbe = originalProbe
|
||||
resolveOptionalDriverAgentExecutablePathFunc = originalResolvePath
|
||||
})
|
||||
|
||||
resolveCalls := 0
|
||||
resolveOptionalDriverAgentExecutablePathFunc = func(downloadDir string, driverType string) (string, error) {
|
||||
resolveCalls++
|
||||
if driverType != "clickhouse" {
|
||||
t.Fatalf("expected ClickHouse export preflight, got %q", driverType)
|
||||
}
|
||||
return "clickhouse-driver-agent", nil
|
||||
}
|
||||
optionalDriverAgentMetadataProbe = func(driverType string, executablePath string) (db.OptionalDriverAgentMetadata, error) {
|
||||
return db.OptionalDriverAgentMetadata{
|
||||
DriverType: driverType,
|
||||
AgentRevision: db.OptionalDriverAgentRevision(driverType),
|
||||
}, nil
|
||||
}
|
||||
|
||||
if err := verifyOptionalDriverAgentReadyForExport(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
}); err != nil {
|
||||
t.Fatalf("custom ClickHouse export preflight failed: %v", err)
|
||||
}
|
||||
if resolveCalls != 1 {
|
||||
t.Fatalf("expected one ClickHouse agent preflight, got %d", resolveCalls)
|
||||
}
|
||||
|
||||
if err := verifyOptionalDriverAgentReadyForExport(connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "kingbase",
|
||||
}); err != nil {
|
||||
t.Fatalf("unrelated custom driver export preflight changed: %v", err)
|
||||
}
|
||||
if resolveCalls != 1 {
|
||||
t.Fatalf("expected unrelated custom driver to skip optional-agent preflight, got %d calls", resolveCalls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDBReleaseConnectionCanonicalizesCustomClickHouseCacheKey(t *testing.T) {
|
||||
a := NewApp()
|
||||
raw := connection.ConnectionConfig{
|
||||
Type: "custom",
|
||||
Driver: "clickhouse",
|
||||
DSN: "jdbc:clickhouse://db.example.com:8123/analytics",
|
||||
}
|
||||
effective, err := a.resolveEffectiveConnectionConfig(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("resolve effective config failed: %v", err)
|
||||
}
|
||||
recording := &customClickHouseRecordingDB{}
|
||||
a.dbCache[getCacheKey(effective)] = cachedDatabase{
|
||||
inst: recording,
|
||||
config: normalizeCacheKeyConfig(effective),
|
||||
}
|
||||
|
||||
result := a.DBReleaseConnection(raw)
|
||||
if !result.Success {
|
||||
t.Fatalf("DBReleaseConnection failed: %s", result.Message)
|
||||
}
|
||||
if recording.closeCalls != 1 {
|
||||
t.Fatalf("expected cached ClickHouse agent connection to close once, got %d", recording.closeCalls)
|
||||
}
|
||||
if len(a.dbCache) != 0 {
|
||||
t.Fatalf("expected cache to be empty, got %d entries", len(a.dbCache))
|
||||
}
|
||||
}
|
||||
@@ -40,10 +40,14 @@ func normalizeRunConfig(config connection.ConnectionConfig, dbName string) conne
|
||||
if idx, err := strconv.Atoi(name); err == nil && idx >= 0 {
|
||||
runConfig.RedisDB = idx
|
||||
}
|
||||
case "custom":
|
||||
if resolveDDLDBType(config) == "clickhouse" {
|
||||
runConfig = runConfig.WithRuntimeDatabaseOverride(name)
|
||||
}
|
||||
default:
|
||||
// oracle: dbName 表示 schema/owner，不能覆盖 config.Database（服务名）
|
||||
// sqlite: 无需设置 Database
|
||||
// custom: 语义不明确，避免污染缓存 key
|
||||
// 其他 custom: 语义不明确，避免污染缓存 key
|
||||
}
|
||||
|
||||
return runConfig
|
||||
|
||||
@@ -93,15 +93,14 @@ func (a *App) DBReleaseConnection(config connection.ConnectionConfig) connection
|
||||
return connection.QueryResult{Success: true, Message: a.appText("db.backend.message.release_success", nil), Data: map[string]int{"closed": closed}}
|
||||
}
|
||||
|
||||
resolvedConfig, err := a.resolveConnectionSecrets(config)
|
||||
effectiveConfig, err := a.resolveEffectiveConnectionConfig(config)
|
||||
if err != nil {
|
||||
wrapped := wrapConnectError(config, err)
|
||||
logger.Error(wrapped, "DBReleaseConnection 解析连接密文失败：%s", formatConnSummary(config))
|
||||
return connection.QueryResult{Success: false, Message: wrapped.Error()}
|
||||
logger.Error(err, "DBReleaseConnection 解析运行时连接配置失败：%s", formatConnSummary(config))
|
||||
return connection.QueryResult{Success: false, Message: err.Error()}
|
||||
}
|
||||
closed := a.releaseCachedDatabaseConnectionsForConfig(resolvedConfig)
|
||||
closed := a.releaseCachedDatabaseConnectionsForConfig(effectiveConfig)
|
||||
|
||||
logger.Infof("DBReleaseConnection 已释放数据库连接：%s 数量=%d", formatConnSummary(resolvedConfig), closed)
|
||||
logger.Infof("DBReleaseConnection 已释放数据库连接：%s 数量=%d", formatConnSummary(effectiveConfig), closed)
|
||||
return connection.QueryResult{Success: true, Message: a.appText("db.backend.message.release_success", nil), Data: map[string]int{"closed": closed}}
|
||||
}
|
||||
|
||||
@@ -205,6 +204,9 @@ func (a *App) CreateDatabase(config connection.ConnectionConfig, dbName string)
|
||||
|
||||
runConfig := config
|
||||
runConfig.Database = ""
|
||||
if resolveDDLDBType(config) == "clickhouse" && strings.EqualFold(strings.TrimSpace(config.Type), "custom") {
|
||||
runConfig = runConfig.WithRuntimeDatabaseOverride("")
|
||||
}
|
||||
|
||||
dbInst, err := a.getDatabase(runConfig)
|
||||
if err != nil {
|
||||
@@ -758,6 +760,9 @@ func (a *App) DropDatabase(config connection.ConnectionConfig, dbName string) (r
|
||||
case "mysql", "mariadb", "oceanbase", "diros", "starrocks", "tdengine", "clickhouse":
|
||||
runConfig = config
|
||||
runConfig.Database = ""
|
||||
if dbType == "clickhouse" && strings.EqualFold(strings.TrimSpace(config.Type), "custom") {
|
||||
runConfig = runConfig.WithRuntimeDatabaseOverride("")
|
||||
}
|
||||
sql = fmt.Sprintf("DROP DATABASE %s", quoteIdentByType(dbType, dbName))
|
||||
case "postgres", "kingbase", "highgo", "vastbase", "opengauss", "gaussdb":
|
||||
runConfig = resolvePGLikeDatabaseDDLRunConfig(config, dbType, dbName)
|
||||
|
||||
@@ -365,6 +365,10 @@ func tryResolveExportTableTotalRows(dbInst db.Database, config connection.Connec
|
||||
|
||||
func verifyOptionalDriverAgentReadyForExport(config connection.ConnectionConfig) error {
|
||||
driverType := normalizeDriverType(config.Type)
|
||||
if strings.EqualFold(strings.TrimSpace(config.Type), "custom") &&
|
||||
strings.EqualFold(strings.TrimSpace(config.Driver), "clickhouse") {
|
||||
driverType = "clickhouse"
|
||||
}
|
||||
if !db.IsOptionalGoDriver(driverType) {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -56,28 +56,26 @@ func (a *App) resolveDataSyncEndpointConfig(raw connection.ConnectionConfig, sel
|
||||
return resolved, selectedDatabase, err
|
||||
}
|
||||
|
||||
if !strings.EqualFold(strings.TrimSpace(raw.Type), "oracle") || strings.TrimSpace(raw.ID) == "" {
|
||||
return resolved, strings.TrimSpace(selectedDatabase), nil
|
||||
}
|
||||
|
||||
repo := newSavedConnectionRepository(a.configDir, a.secretStore)
|
||||
view, findErr := repo.Find(raw.ID)
|
||||
if findErr != nil {
|
||||
return resolved, strings.TrimSpace(selectedDatabase), nil
|
||||
}
|
||||
|
||||
savedServiceName := strings.TrimSpace(view.Config.Database)
|
||||
if savedServiceName == "" {
|
||||
return resolved, strings.TrimSpace(selectedDatabase), nil
|
||||
}
|
||||
|
||||
selected := strings.TrimSpace(selectedDatabase)
|
||||
incomingDatabase := strings.TrimSpace(raw.Database)
|
||||
if selected == "" && incomingDatabase != "" && !strings.EqualFold(incomingDatabase, savedServiceName) {
|
||||
selected = incomingDatabase
|
||||
if strings.EqualFold(strings.TrimSpace(raw.Type), "oracle") && strings.TrimSpace(raw.ID) != "" {
|
||||
repo := newSavedConnectionRepository(a.configDir, a.secretStore)
|
||||
if view, findErr := repo.Find(raw.ID); findErr == nil {
|
||||
savedServiceName := strings.TrimSpace(view.Config.Database)
|
||||
if savedServiceName != "" {
|
||||
incomingDatabase := strings.TrimSpace(raw.Database)
|
||||
if selected == "" && incomingDatabase != "" && !strings.EqualFold(incomingDatabase, savedServiceName) {
|
||||
selected = incomingDatabase
|
||||
}
|
||||
resolved.Database = savedServiceName
|
||||
}
|
||||
}
|
||||
}
|
||||
resolved.Database = savedServiceName
|
||||
return resolved, selected, nil
|
||||
|
||||
effectiveConfig, err := a.resolveCustomClickHouseRuntimeConfig(resolved)
|
||||
if err != nil {
|
||||
return resolved, selected, err
|
||||
}
|
||||
return effectiveConfig, selected, nil
|
||||
}
|
||||
|
||||
// DataSync executes a data synchronization task
|
||||
|
||||
@@ -133,6 +133,34 @@ type ConnectionConfig struct {
|
||||
MongoReplicaUser string `json:"mongoReplicaUser,omitempty"` // MongoDB replica auth user
|
||||
MongoReplicaPassword string `json:"mongoReplicaPassword,omitempty"` // MongoDB replica auth password
|
||||
JVM JVMConfig `json:"jvm,omitempty"` // JVM connector config
|
||||
runtimeDBOverride string // App-only selected database; never persisted or sent over RPC.
|
||||
runtimeDBOverrideSet bool // Distinguishes an explicit server-level override from no override.
|
||||
}
|
||||
|
||||
// WithRuntimeDatabaseOverride carries a caller-selected database through runtime
|
||||
// connection normalization without letting stale persisted fields override a DSN.
|
||||
func (c ConnectionConfig) WithRuntimeDatabaseOverride(database string) ConnectionConfig {
|
||||
c.runtimeDBOverride = database
|
||||
c.runtimeDBOverrideSet = true
|
||||
return c
|
||||
}
|
||||
|
||||
// RuntimeDatabaseOverride returns the app-only selected database override.
|
||||
func (c ConnectionConfig) RuntimeDatabaseOverride() string {
|
||||
return c.runtimeDBOverride
|
||||
}
|
||||
|
||||
// HasRuntimeDatabaseOverride reports whether the app explicitly selected a
|
||||
// database, including an empty server-level selection.
|
||||
func (c ConnectionConfig) HasRuntimeDatabaseOverride() bool {
|
||||
return c.runtimeDBOverrideSet
|
||||
}
|
||||
|
||||
// WithoutRuntimeDatabaseOverride removes the app-only selected database marker.
|
||||
func (c ConnectionConfig) WithoutRuntimeDatabaseOverride() ConnectionConfig {
|
||||
c.runtimeDBOverride = ""
|
||||
c.runtimeDBOverrideSet = false
|
||||
return c
|
||||
}
|
||||
|
||||
// ResultSetData 表示一个查询结果集（行 + 列名），用于多结果集场景。
|
||||
|
||||
Reference in New Issue
Block a user