diff --git a/frontend/src/store.test.ts b/frontend/src/store.test.ts index c861d168..a5c6c931 100644 --- a/frontend/src/store.test.ts +++ b/frontend/src/store.test.ts @@ -789,6 +789,8 @@ describe('store appearance persistence', () => { { id: 'kingbase', name: 'Kingbase', config: { id: 'kingbase', type: 'kingbase8', host: 'kingbase.local', port: 54321, user: 'system' } }, { id: 'dm', name: 'Dameng', config: { id: 'dm', type: 'dm8', host: 'dm.local', port: 5236, user: 'SYSDBA' } }, { id: 'sqlite', name: 'SQLite', config: { id: 'sqlite', type: 'sqlite3', host: 'D:/db/app.sqlite', port: 0, user: '' } }, + { id: 'milvusdb', name: 'Milvus DB', config: { id: 'milvusdb', type: 'milvusdb', host: 'milvus.local', port: 19530, user: '' } }, + { id: 'milvus-db', name: 'Milvus DB Alias', config: { id: 'milvus-db', type: 'milvus-db', host: 'milvus-alias.local', port: 19530, user: '' } }, ]); expect(useStore.getState().connections.map((conn) => conn.config.type)).toEqual([ @@ -797,9 +799,45 @@ describe('store appearance persistence', () => { 'kingbase', 'dameng', 'sqlite', + 'milvus', + 'milvus', ]); }); + it('preserves built-in document, vector, and messaging datasource types', async () => { + const { useStore } = await importStore(); + + const datasourceTypes = [ + ['chroma', 8000], + ['qdrant', 6333], + ['milvus', 19530], + ['rocketmq', 9876], + ['mqtt', 1883], + ['rabbitmq', 15672], + ] as const; + + useStore.getState().replaceConnections( + datasourceTypes.map(([type, port]) => ({ + id: `conn-${type}`, + name: type, + config: { + id: `conn-${type}`, + type, + host: `${type}.local`, + port, + user: '', + }, + })), + ); + + expect( + useStore.getState().connections.map((conn) => [ + conn.config.type, + conn.config.port, + ]), + ).toEqual(datasourceTypes); + }); + it('preserves SSL certificate paths for SSL-capable saved connections', async () => { const { useStore } = await importStore(); diff --git a/frontend/src/store.ts b/frontend/src/store.ts index 3777617e..9dfc1ae8 100644 --- a/frontend/src/store.ts +++ b/frontend/src/store.ts @@ -126,6 +126,12 @@ import { sanitizeSidebarHiddenObjectGroups, type SidebarObjectGroupKey, } from "./utils/sidebarObjectVisibility"; +import { + CONNECTION_TYPE_GROUPS, + getConnectionTypeDefaultPort, +} from "./utils/connectionTypeCatalog"; +import { supportsSSLForType } from "./utils/connectionTypeCapabilities"; +import { normalizeDriverType } from "./utils/connectionDriverType"; export type TableDoubleClickAction = "open-data" | "open-design"; export type ThemeMode = "light" | "dark"; @@ -422,119 +428,12 @@ const resolveOceanBaseProtocol = ( } }; const SUPPORTED_CONNECTION_TYPES = new Set([ - "mysql", - "goldendb", - "mariadb", - "oceanbase", + ...CONNECTION_TYPE_GROUPS.flatMap((group) => + group.items.map((item) => item.key), + ), + // Legacy persisted alias, normalized to diros before support is checked. "doris", - "diros", - "starrocks", - "sphinx", - "clickhouse", - "trino", - "postgres", - "redis", - "tdengine", - "iotdb", - "kafka", - "oracle", - "dameng", - "kingbase", - "sqlserver", - "iris", - "mongodb", - "elasticsearch", - "highgo", - "vastbase", - "opengauss", - "gaussdb", - "jvm", - "sqlite", - "duckdb", - "custom", ]); -const SSL_SUPPORTED_CONNECTION_TYPES = new Set([ - "mysql", - "goldendb", - "mariadb", - "oceanbase", - "diros", - "starrocks", - "sphinx", - "dameng", - "clickhouse", - "trino", - "postgres", - "sqlserver", - "oracle", - "kingbase", - "highgo", - "vastbase", - "opengauss", - "gaussdb", - "mongodb", - "redis", - "elasticsearch", - "tdengine", - "kafka", -]); - -const getDefaultPortByType = (type: string): number => { - switch (type) { - case "jvm": - return DEFAULT_JVM_PORT; - case "mysql": - case "mariadb": - return 3306; - case "goldendb": - return 1523; - case "oceanbase": - return 2881; - case "doris": - case "diros": - case "starrocks": - return 9030; - case "duckdb": - return 0; - case "sphinx": - return 9306; - case "clickhouse": - return 9000; - case "trino": - return 8080; - case "postgres": - case "vastbase": - case "opengauss": - case "gaussdb": - return 5432; - case "redis": - return 6379; - case "tdengine": - return 6041; - case "iotdb": - return 6667; - case "kafka": - return 9092; - case "oracle": - return 1521; - case "dameng": - return 5236; - case "kingbase": - return 54321; - case "sqlserver": - return 1433; - case "iris": - return 1972; - case "mongodb": - return 27017; - case "elasticsearch": - return 9200; - case "highgo": - return 5866; - default: - return 3306; - } -}; const toTrimmedString = (value: unknown, fallback = ""): string => { if (typeof value === "string") { @@ -701,7 +600,7 @@ const sanitizeConnectionIconColor = (value: unknown): string | undefined => { }; const normalizeConnectionType = (value: unknown): string => { - const type = toTrimmedString(value).toLowerCase(); + const type = normalizeDriverType(toTrimmedString(value)); if (type === "doris") { return "diros"; } @@ -890,11 +789,11 @@ const sanitizeConnectionConfig = (value: unknown): ConnectionConfig => { ? (value as Record) : {}; const type = normalizeConnectionType(raw.type); - const defaultPort = getDefaultPortByType(type); + const defaultPort = getConnectionTypeDefaultPort(type); const savePassword = typeof raw.savePassword === "boolean" ? raw.savePassword : true; const mongoSrv = !!raw.mongoSrv; - const sslCapable = SSL_SUPPORTED_CONNECTION_TYPES.has(type); + const sslCapable = supportsSSLForType(type); const sslModeRaw = toTrimmedString(raw.sslMode, "preferred").toLowerCase(); const sslMode: "preferred" | "required" | "skip-verify" | "disable" = sslModeRaw === "required" diff --git a/go.mod b/go.mod index ed149786..04e8fc78 100644 --- a/go.mod +++ b/go.mod @@ -79,7 +79,7 @@ require ( filippo.io/edwards25519 v1.1.0 // indirect github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 // indirect github.com/99designs/keyring v1.2.2 - github.com/ClickHouse/ch-go v0.71.0 // indirect + github.com/ClickHouse/ch-go v0.71.0 github.com/andybalholm/brotli v1.2.0 // indirect github.com/apache/arrow-go/v18 v18.5.1 // indirect github.com/bep/debounce v1.2.1 // indirect diff --git a/internal/db/clickhouse_impl.go b/internal/db/clickhouse_impl.go index 0db7ff15..c50098f6 100644 --- a/internal/db/clickhouse_impl.go +++ b/internal/db/clickhouse_impl.go @@ -6,6 +6,7 @@ import ( "bytes" "context" "database/sql" + "database/sql/driver" "fmt" "io" "net" @@ -41,6 +42,7 @@ const ( type ClickHouseDB struct { conn *sql.DB + legacyHTTP *clickHouseLegacyHTTPClient pingTimeout time.Duration forwarder *ssh.LocalForwarder database string @@ -701,6 +703,10 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error { _ = c.conn.Close() c.conn = nil } + if c.legacyHTTP != nil { + _ = c.legacyHTTP.Close() + c.legacyHTTP = nil + } runConfig := normalizeClickHouseConfig(config) c.pingTimeout = getConnectTimeout(runConfig) @@ -785,6 +791,20 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error { } continue } + if protocol == clickhouse.HTTP && stripHTTPClientProtocolVersion { + legacyClient, legacyErr := c.connectClickHouseLegacyHTTP(opts) + if legacyErr == nil { + c.legacyHTTP = legacyClient + protocolSuccess = true + logger.Warnf("ClickHouse HTTP 兼容握手无法解码旧版 Native block，已切换 legacy JSON HTTP 模式") + break + } + lastProtocolErr = legacyErr + legacyFailure := sanitizeClickHouseErrorMessage(legacyErr) + failures = append(failures, clickHouseAttemptValidationFailedMessage(idx+1, "legacy-http", legacyFailure)) + logger.Warnf("ClickHouse legacy JSON HTTP 连接尝试失败：第%d组/%d 地址=%s:%d SSL=%t 原因=%s", + idx+1, len(attempts), protocolConfig.Host, protocolConfig.Port, protocolConfig.UseSSL, legacyFailure) + } break } protocolSuccess = true @@ -815,6 +835,24 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error { return fmt.Errorf("%s", clickHouseConnectFailureSummary(runConfig, failures)) } +func (c *ClickHouseDB) connectClickHouseLegacyHTTP(opts *clickhouse.Options) (*clickHouseLegacyHTTPClient, error) { + legacyClient, err := newClickHouseLegacyHTTPClient(opts) + if err != nil { + return nil, err + } + timeout := c.pingTimeout + if timeout <= 0 { + timeout = 5 * time.Second + } + ctx, cancel := utils.ContextWithTimeout(timeout) + defer cancel() + if err := legacyClient.Ping(ctx); err != nil { + _ = legacyClient.Close() + return nil, err + } + return legacyClient, nil +} + func (c *ClickHouseDB) Close() error { if c.forwarder != nil { if err := c.forwarder.Close(); err != nil { @@ -823,12 +861,32 @@ func (c *ClickHouseDB) Close() error { c.forwarder = nil } if c.conn != nil { - return c.conn.Close() + err := c.conn.Close() + c.conn = nil + if err != nil { + return err + } + } + if c.legacyHTTP != nil { + err := c.legacyHTTP.Close() + c.legacyHTTP = nil + if err != nil { + return err + } } return nil } func (c *ClickHouseDB) Ping() error { + if c.legacyHTTP != nil { + timeout := c.pingTimeout + if timeout <= 0 { + timeout = 5 * time.Second + } + ctx, cancel := utils.ContextWithTimeout(timeout) + defer cancel() + return c.legacyHTTP.Ping(ctx) + } if c.conn == nil { return fmt.Errorf("连接未打开") } @@ -879,6 +937,9 @@ func (c *ClickHouseDB) validateQueryPath() error { } func (c *ClickHouseDB) QueryContext(ctx context.Context, query string) ([]map[string]interface{}, []string, error) { + if c.legacyHTTP != nil { + return c.legacyHTTP.Query(ctx, query) + } if c.conn == nil { return nil, nil, fmt.Errorf("连接未打开") } @@ -891,6 +952,9 @@ func (c *ClickHouseDB) QueryContext(ctx context.Context, query string) ([]map[st } func (c *ClickHouseDB) Query(query string) ([]map[string]interface{}, []string, error) { + if c.legacyHTTP != nil { + return c.legacyHTTP.Query(context.Background(), query) + } if c.conn == nil { return nil, nil, fmt.Errorf("连接未打开") } @@ -903,6 +967,9 @@ func (c *ClickHouseDB) Query(query string) ([]map[string]interface{}, []string, } func (c *ClickHouseDB) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error { + if c.legacyHTTP != nil { + return c.legacyHTTP.StreamQuery(ctx, query, consumer) + } if c.conn == nil { return fmt.Errorf("连接未打开") } @@ -919,6 +986,9 @@ func (c *ClickHouseDB) StreamQuery(query string, consumer QueryStreamConsumer) e } func (c *ClickHouseDB) ExecContext(ctx context.Context, query string) (int64, error) { + if c.legacyHTTP != nil { + return c.legacyHTTP.Exec(ctx, query) + } if c.conn == nil { return 0, fmt.Errorf("连接未打开") } @@ -930,6 +1000,9 @@ func (c *ClickHouseDB) ExecContext(ctx context.Context, query string) (int64, er } func (c *ClickHouseDB) Exec(query string) (int64, error) { + if c.legacyHTTP != nil { + return c.legacyHTTP.Exec(context.Background(), query) + } if c.conn == nil { return 0, fmt.Errorf("连接未打开") } @@ -1333,7 +1406,7 @@ func isClickHouseTruthy(value interface{}) bool { } func (c *ClickHouseDB) ApplyChanges(tableName string, changes connection.ChangeSet) error { - if c.conn == nil { + if c.conn == nil && c.legacyHTTP == nil { return fmt.Errorf("连接未打开") } @@ -1349,7 +1422,7 @@ func (c *ClickHouseDB) ApplyChanges(tableName string, changes connection.ChangeS continue } query := fmt.Sprintf("ALTER TABLE %s DELETE WHERE %s", qualifiedTable, whereExpr) - if _, err := c.conn.Exec(query); err != nil { + if _, err := c.Exec(query); err != nil { return localizedDatabaseRuntimeError("db.backend.error.clickhouse_delete_failed_with_sql", map[string]any{ "detail": err.Error(), "sql": query, @@ -1364,7 +1437,7 @@ func (c *ClickHouseDB) ApplyChanges(tableName string, changes connection.ChangeS continue } query := fmt.Sprintf("ALTER TABLE %s UPDATE %s WHERE %s", qualifiedTable, setExpr, whereExpr) - if _, err := c.conn.Exec(query); err != nil { + if _, err := c.Exec(query); err != nil { return localizedDatabaseRuntimeError("db.backend.error.clickhouse_update_failed_with_sql", map[string]any{ "detail": err.Error(), "sql": query, @@ -1372,14 +1445,14 @@ func (c *ClickHouseDB) ApplyChanges(tableName string, changes connection.ChangeS } } - if err := execClickHouseInsertBatches(c.conn, qualifiedTable, changes.Inserts); err != nil { + if err := execClickHouseInsertBatches(c.Exec, qualifiedTable, changes.Inserts); err != nil { return err } return nil } -func execClickHouseInsertBatches(conn *sql.DB, qualifiedTable string, rows []map[string]interface{}) error { - if conn == nil { +func execClickHouseInsertBatches(exec func(string) (int64, error), qualifiedTable string, rows []map[string]interface{}) error { + if exec == nil { return fmt.Errorf("连接未打开") } return execLiteralInsertBatches(literalInsertConfig{ @@ -1388,7 +1461,8 @@ func execClickHouseInsertBatches(conn *sql.DB, qualifiedTable string, rows []map QuoteColumn: quoteClickHouseIdentifier, Literal: clickHouseLiteral, Exec: func(query string) (sql.Result, error) { - return conn.Exec(query) + affected, err := exec(query) + return driver.RowsAffected(affected), err }, }) } diff --git a/internal/db/clickhouse_impl_test.go b/internal/db/clickhouse_impl_test.go index 0528830b..f3dadf62 100644 --- a/internal/db/clickhouse_impl_test.go +++ b/internal/db/clickhouse_impl_test.go @@ -8,8 +8,13 @@ import ( "database/sql/driver" "errors" "io" + "net" "net/http" + "net/http/httptest" + "net/url" "os" + "path/filepath" + "strconv" "strings" "sync" "sync/atomic" @@ -19,7 +24,10 @@ import ( "GoNavi-Wails/internal/connection" "GoNavi-Wails/shared/i18n" + chproto "github.com/ClickHouse/ch-go/proto" clickhouse "github.com/ClickHouse/clickhouse-go/v2" + clickhousecolumn "github.com/ClickHouse/clickhouse-go/v2/lib/column" + clickhouseproto "github.com/ClickHouse/clickhouse-go/v2/lib/proto" ) const fakeClickHouseDriverName = "gonavi-fake-clickhouse" @@ -889,6 +897,143 @@ func TestClickHouseHTTPCompatibilityStripperLeavesOtherBodiesUnchanged(t *testin } } +func TestClickHouseConnectFallsBackToLegacyHTTPForRevisionZeroServer(t *testing.T) { + installClickHouseRuntimeMarkerForTest(t) + + var ( + mu sync.Mutex + sawProtocolVersion bool + sawRevisionZeroHandshake bool + sawLegacyJSONQuery bool + ) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read request body: %v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + mu.Lock() + defer mu.Unlock() + if r.URL.Query().Get("client_protocol_version") != "" { + sawProtocolVersion = true + w.WriteHeader(http.StatusNotFound) + _, _ = io.WriteString(w, "Code: 115. DB::Exception: Unknown setting client_protocol_version. (UNKNOWN_SETTING)") + return + } + + switch strings.TrimSpace(string(body)) { + case clickHouseServerHelloCompatQuery: + sawRevisionZeroHandshake = true + writeClickHouseRevisionZeroHelloBlock(t, w) + case "SELECT currentDatabase()": + if r.URL.Query().Get("default_format") != "JSONCompactEachRowWithNamesAndTypes" { + t.Errorf("legacy query format = %q", r.URL.Query().Get("default_format")) + w.WriteHeader(http.StatusBadRequest) + return + } + sawLegacyJSONQuery = true + _, _ = io.WriteString(w, "[\"currentDatabase()\"]\n[\"String\"]\n[\"default\"]\n") + default: + t.Errorf("unexpected ClickHouse query: %q", string(body)) + w.WriteHeader(http.StatusBadRequest) + } + })) + defer server.Close() + + serverURL, err := url.Parse(server.URL) + if err != nil { + t.Fatalf("parse ClickHouse test server URL: %v", err) + } + host, portText, err := net.SplitHostPort(serverURL.Host) + if err != nil { + t.Fatalf("split ClickHouse test server address: %v", err) + } + port, err := strconv.Atoi(portText) + if err != nil { + t.Fatalf("parse ClickHouse test server port: %v", err) + } + client := &ClickHouseDB{} + err = client.Connect(connection.ConnectionConfig{ + Type: "clickhouse", + Host: host, + Port: port, + Database: "default", + User: "default", + ClickHouseProtocol: clickHouseProtocolHTTP, + Timeout: 2, + }) + if err != nil { + t.Fatalf("Connect should fall back to legacy HTTP for ClickHouse 22.8: %v", err) + } + defer client.Close() + + mu.Lock() + defer mu.Unlock() + if !sawProtocolVersion || !sawRevisionZeroHandshake || !sawLegacyJSONQuery { + t.Fatalf( + "expected modern, revision-zero, and legacy requests; modern=%t revisionZero=%t legacy=%t", + sawProtocolVersion, + sawRevisionZeroHandshake, + sawLegacyJSONQuery, + ) + } +} + +func installClickHouseRuntimeMarkerForTest(t *testing.T) { + t.Helper() + previousRoot := currentExternalDriverDownloadDirectory() + t.Cleanup(func() { SetExternalDriverDownloadDirectory(previousRoot) }) + + root := t.TempDir() + SetExternalDriverDownloadDirectory(root) + markerPath, err := ResolveOptionalGoDriverMarkerPath(root, "clickhouse") + if err != nil { + t.Fatalf("resolve ClickHouse marker path: %v", err) + } + if err := os.MkdirAll(filepath.Dir(markerPath), 0o755); err != nil { + t.Fatalf("create ClickHouse marker directory: %v", err) + } + if err := os.WriteFile(markerPath, []byte("{}"), 0o644); err != nil { + t.Fatalf("write ClickHouse marker: %v", err) + } + executablePath, err := ResolveOptionalDriverAgentExecutablePath(root, "clickhouse") + if err != nil { + t.Fatalf("resolve ClickHouse agent path: %v", err) + } + if err := os.WriteFile(executablePath, []byte("test agent placeholder"), 0o755); err != nil { + t.Fatalf("write ClickHouse agent placeholder: %v", err) + } +} + +func writeClickHouseRevisionZeroHelloBlock(t *testing.T, w http.ResponseWriter) { + t.Helper() + block := clickhouseproto.NewBlock() + for _, column := range []struct { + name string + typeName clickhousecolumn.Type + }{ + {name: "hostName()", typeName: "String"}, + {name: "version()", typeName: "String"}, + {name: "revision()", typeName: "UInt64"}, + {name: "timezone()", typeName: "String"}, + } { + if err := block.AddColumn(column.name, column.typeName); err != nil { + t.Fatalf("add ClickHouse block column: %v", err) + } + } + if err := block.Append("legacy-clickhouse", "22.8.20.11", uint64(54460), "UTC"); err != nil { + t.Fatalf("append ClickHouse block row: %v", err) + } + buffer := &chproto.Buffer{} + if err := block.Encode(buffer, 0); err != nil { + t.Fatalf("encode revision-zero ClickHouse block: %v", err) + } + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = w.Write(buffer.Buf) +} + func TestWithClickHouseProtocolForcesProtocolSelection(t *testing.T) { httpConfig := withClickHouseProtocol(connection.ConnectionConfig{ Type: "clickhouse", diff --git a/internal/db/clickhouse_legacy_http.go b/internal/db/clickhouse_legacy_http.go new file mode 100644 index 00000000..02d2c49b --- /dev/null +++ b/internal/db/clickhouse_legacy_http.go @@ -0,0 +1,322 @@ +//go:build gonavi_full_drivers || gonavi_clickhouse_driver + +package db + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strings" + + clickhouse "github.com/ClickHouse/clickhouse-go/v2" +) + +const ( + clickHouseLegacyHTTPFormat = "JSONCompactEachRowWithNamesAndTypes" + clickHouseLegacyHTTPErrorLimit = 64 << 10 + clickHouseLegacyHTTPUserAgent = "GoNavi ClickHouse legacy HTTP client" + clickHouseLegacyHTTPDatabaseName = "database" +) + +// clickHouseLegacyHTTPClient is used only after the current clickhouse-go HTTP +// handshake proves that the server predates client_protocol_version support. +// JSON avoids decoding Native blocks with a wire revision the driver cannot +// negotiate with those servers. +type clickHouseLegacyHTTPClient struct { + endpoint *url.URL + http *http.Client + transport *http.Transport + username string + password string + headers http.Header + params url.Values +} + +func newClickHouseLegacyHTTPClient(opts *clickhouse.Options) (*clickHouseLegacyHTTPClient, error) { + if opts == nil { + return nil, fmt.Errorf("ClickHouse legacy HTTP options are required") + } + if len(opts.Addr) == 0 || strings.TrimSpace(opts.Addr[0]) == "" { + return nil, fmt.Errorf("ClickHouse legacy HTTP address is required") + } + + scheme := "http" + if opts.TLS != nil { + scheme = "https" + } + path := strings.TrimSpace(opts.HttpUrlPath) + if path != "" && !strings.HasPrefix(path, "/") { + path = "/" + path + } + endpoint := &url.URL{ + Scheme: scheme, + Host: strings.TrimSpace(opts.Addr[0]), + Path: path, + } + + proxy := http.ProxyFromEnvironment + if opts.HTTPProxyURL != nil { + proxy = http.ProxyURL(opts.HTTPProxyURL) + } + transport := &http.Transport{ + Proxy: proxy, + DialContext: (&net.Dialer{Timeout: opts.DialTimeout}).DialContext, + MaxIdleConns: 1, + MaxConnsPerHost: opts.HttpMaxConnsPerHost, + IdleConnTimeout: opts.ConnMaxLifetime, + ResponseHeaderTimeout: opts.ReadTimeout, + TLSClientConfig: opts.TLS, + DisableCompression: true, + } + if opts.DialContext != nil { + transport.DialContext = func(ctx context.Context, _, address string) (net.Conn, error) { + return opts.DialContext(ctx, address) + } + } + + params := make(url.Values, len(opts.Settings)+1) + if database := strings.TrimSpace(opts.Auth.Database); database != "" { + params.Set(clickHouseLegacyHTTPDatabaseName, database) + } + for key, value := range opts.Settings { + key = strings.TrimSpace(key) + if key == "" || strings.EqualFold(key, "default_format") || strings.EqualFold(key, "client_protocol_version") { + continue + } + if custom, ok := value.(clickhouse.CustomSetting); ok { + value = custom.Value + } + params.Set(key, fmt.Sprint(value)) + } + + headers := make(http.Header, len(opts.HttpHeaders)+2) + for key, value := range opts.HttpHeaders { + headers.Set(key, value) + } + headers.Set("User-Agent", clickHouseLegacyHTTPUserAgent) + headers.Set("Content-Type", "text/plain; charset=utf-8") + + return &clickHouseLegacyHTTPClient{ + endpoint: endpoint, + http: &http.Client{ + Transport: transport, + }, + transport: transport, + username: opts.Auth.Username, + password: opts.Auth.Password, + headers: headers, + params: params, + }, nil +} + +func (c *clickHouseLegacyHTTPClient) Close() error { + if c != nil && c.transport != nil { + c.transport.CloseIdleConnections() + } + return nil +} + +func (c *clickHouseLegacyHTTPClient) Ping(ctx context.Context) error { + rows, _, err := c.Query(ctx, "SELECT currentDatabase()") + if err != nil { + return err + } + if len(rows) == 0 { + return fmt.Errorf("ClickHouse legacy HTTP validation returned no rows") + } + return nil +} + +func (c *clickHouseLegacyHTTPClient) Query(ctx context.Context, query string) ([]map[string]interface{}, []string, error) { + collector := &clickHouseLegacyHTTPCollector{} + if err := c.StreamQuery(ctx, query, collector); err != nil { + return collector.rows, collector.columns, err + } + return collector.rows, collector.columns, nil +} + +func (c *clickHouseLegacyHTTPClient) StreamQuery(ctx context.Context, query string, consumer QueryStreamConsumer) error { + if consumer == nil { + return fmt.Errorf("query stream consumer required") + } + response, err := c.do(ctx, query, false) + if err != nil { + return err + } + defer response.Body.Close() + + decoder := json.NewDecoder(response.Body) + decoder.UseNumber() + + var columns []string + if err := decoder.Decode(&columns); err != nil { + return c.decodeError(decoder, response.Body, "column names", err) + } + if len(columns) == 0 { + return fmt.Errorf("ClickHouse legacy HTTP response has no columns") + } + columns = ensureUniqueQueryColumnNames(columns) + + var typeNames []string + if err := decoder.Decode(&typeNames); err != nil { + return c.decodeError(decoder, response.Body, "column types", err) + } + if len(typeNames) != len(columns) { + return fmt.Errorf("ClickHouse legacy HTTP column metadata mismatch: names=%d types=%d", len(columns), len(typeNames)) + } + if err := consumer.SetColumns(columns); err != nil { + return err + } + valueConsumer, useValueConsumer := consumer.(QueryStreamValueConsumer) + + for { + var values []interface{} + err := decoder.Decode(&values) + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return c.decodeError(decoder, response.Body, "row", err) + } + if len(values) != len(columns) { + return fmt.Errorf("ClickHouse legacy HTTP row width mismatch: columns=%d values=%d", len(columns), len(values)) + } + for index := range values { + values[index] = normalizeQueryValueWithDBType(values[index], typeNames[index]) + } + if useValueConsumer { + if err := valueConsumer.ConsumeRowValues(values); err != nil { + return err + } + continue + } + row := make(map[string]interface{}, len(columns)) + for index, column := range columns { + row[column] = values[index] + } + if err := consumer.ConsumeRow(row); err != nil { + return err + } + } +} + +func (c *clickHouseLegacyHTTPClient) Exec(ctx context.Context, query string) (int64, error) { + response, err := c.do(ctx, query, true) + if err != nil { + return 0, err + } + defer response.Body.Close() + + body, readErr := io.ReadAll(io.LimitReader(response.Body, clickHouseLegacyHTTPErrorLimit+1)) + if readErr != nil { + return 0, readErr + } + if exception := clickHouseLegacyHTTPException(body); exception != "" { + return 0, fmt.Errorf("%s", exception) + } + // clickhouse-go also reports zero because ClickHouse does not provide a + // database/sql affected-row count for ordinary HTTP executions. + return 0, nil +} + +func (c *clickHouseLegacyHTTPClient) do(ctx context.Context, query string, waitForEnd bool) (*http.Response, error) { + if c == nil || c.endpoint == nil || c.http == nil { + return nil, fmt.Errorf("ClickHouse legacy HTTP connection is not open") + } + requestURL := *c.endpoint + params := cloneURLValues(c.params) + params.Set("default_format", clickHouseLegacyHTTPFormat) + params.Del("client_protocol_version") + if waitForEnd { + params.Set("wait_end_of_query", "1") + } + requestURL.RawQuery = params.Encode() + + request, err := http.NewRequestWithContext(ctx, http.MethodPost, requestURL.String(), strings.NewReader(query)) + if err != nil { + return nil, err + } + request.Header = c.headers.Clone() + if c.username != "" || c.password != "" { + request.SetBasicAuth(c.username, c.password) + } + + response, err := c.http.Do(request) + if err != nil { + return nil, err + } + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + defer response.Body.Close() + body, _ := io.ReadAll(io.LimitReader(response.Body, clickHouseLegacyHTTPErrorLimit)) + detail := clickHouseLegacyHTTPException(body) + if detail == "" { + detail = strings.TrimSpace(string(body)) + } + if detail == "" { + detail = response.Status + } + return nil, fmt.Errorf("ClickHouse legacy HTTP request failed: status=%d detail=%s", response.StatusCode, detail) + } + if code := strings.TrimSpace(response.Header.Get("X-ClickHouse-Exception-Code")); code != "" && code != "0" { + defer response.Body.Close() + body, _ := io.ReadAll(io.LimitReader(response.Body, clickHouseLegacyHTTPErrorLimit)) + detail := clickHouseLegacyHTTPException(body) + if detail == "" { + detail = strings.TrimSpace(string(body)) + } + return nil, fmt.Errorf("ClickHouse legacy HTTP exception code=%s detail=%s", code, detail) + } + return response, nil +} + +func (c *clickHouseLegacyHTTPClient) decodeError(decoder *json.Decoder, body io.Reader, section string, decodeErr error) error { + var tail []byte + if decoder != nil { + tail, _ = io.ReadAll(io.LimitReader(io.MultiReader(decoder.Buffered(), body), clickHouseLegacyHTTPErrorLimit)) + } + if exception := clickHouseLegacyHTTPException(tail); exception != "" { + return fmt.Errorf("%s", exception) + } + return fmt.Errorf("decode ClickHouse legacy HTTP %s: %w", section, decodeErr) +} + +func clickHouseLegacyHTTPException(raw []byte) string { + text := sanitizeClickHouseErrorMessage(errors.New(strings.TrimSpace(string(raw)))) + if text == "" { + return "" + } + lower := strings.ToLower(text) + if strings.Contains(lower, "db::exception") || + (strings.Contains(lower, "code:") && strings.Contains(lower, "exception")) { + return text + } + return "" +} + +func cloneURLValues(source url.Values) url.Values { + result := make(url.Values, len(source)) + for key, values := range source { + result[key] = append([]string(nil), values...) + } + return result +} + +type clickHouseLegacyHTTPCollector struct { + columns []string + rows []map[string]interface{} +} + +func (c *clickHouseLegacyHTTPCollector) SetColumns(columns []string) error { + c.columns = append([]string(nil), columns...) + return nil +} + +func (c *clickHouseLegacyHTTPCollector) ConsumeRow(row map[string]interface{}) error { + c.rows = append(c.rows, row) + return nil +} diff --git a/internal/db/clickhouse_legacy_http_test.go b/internal/db/clickhouse_legacy_http_test.go new file mode 100644 index 00000000..a490fddb --- /dev/null +++ b/internal/db/clickhouse_legacy_http_test.go @@ -0,0 +1,289 @@ +//go:build gonavi_full_drivers || gonavi_clickhouse_driver + +package db + +import ( + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "reflect" + "strings" + "sync" + "testing" + "time" + + "GoNavi-Wails/internal/connection" + + clickhouse "github.com/ClickHouse/clickhouse-go/v2" +) + +func TestClickHouseLegacyHTTPQueryPreservesValuesAndRequestOptions(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/clickhouse" { + t.Errorf("HTTP path = %q", r.URL.Path) + } + if got := r.URL.Query().Get("default_format"); got != clickHouseLegacyHTTPFormat { + t.Errorf("default_format = %q", got) + } + if got := r.URL.Query().Get("database"); got != "analytics" { + t.Errorf("database = %q", got) + } + if got := r.URL.Query().Get("client_protocol_version"); got != "" { + t.Errorf("client_protocol_version should be absent, got %q", got) + } + if got := r.URL.Query().Get("max_execution_time"); got != "60" { + t.Errorf("max_execution_time = %q", got) + } + username, password, ok := r.BasicAuth() + if !ok || username != "reporter" || password != "secret" { + t.Errorf("basic auth = (%q, %q, %t)", username, password, ok) + } + _, _ = io.WriteString(w, strings.Join([]string{ + `["id","id","nullable","created_at","items","attrs"]`, + `["UInt64","UInt8","Nullable(String)","DateTime","Array(UInt64)","Map(String, UInt64)"]`, + `[9007199254740993,2,null,"2022-08-03 12:13:14",[1,9007199254740994],{"small":3,"large":9007199254740995}]`, + "", + }, "\n")) + })) + defer server.Close() + + client := newClickHouseLegacyHTTPTestClientWithOptions(t, server, func(opts *clickhouse.Options) { + opts.Auth = clickhouse.Auth{Database: "analytics", Username: "reporter", Password: "secret"} + opts.HttpUrlPath = "clickhouse" + opts.Settings = clickhouse.Settings{ + "max_execution_time": 60, + "client_protocol_version": 54485, + } + }) + + rows, columns, err := client.Query(context.Background(), "SELECT values") + if err != nil { + t.Fatalf("legacy query failed: %v", err) + } + if !reflect.DeepEqual(columns, []string{"id", "id_2", "nullable", "created_at", "items", "attrs"}) { + t.Fatalf("columns = %#v", columns) + } + if len(rows) != 1 { + t.Fatalf("rows = %#v", rows) + } + row := rows[0] + if row["id"] != "9007199254740993" || row["id_2"] != int64(2) || row["nullable"] != nil { + t.Fatalf("scalar values = %#v", row) + } + if !reflect.DeepEqual(row["items"], []interface{}{int64(1), "9007199254740994"}) { + t.Fatalf("items = %#v", row["items"]) + } + if !reflect.DeepEqual(row["attrs"], map[string]interface{}{"small": int64(3), "large": "9007199254740995"}) { + t.Fatalf("attrs = %#v", row["attrs"]) + } +} + +func TestClickHouseLegacyHTTPStreamDeliversRowsBeforeEOF(t *testing.T) { + firstChunkWritten := make(chan struct{}) + releaseResponse := make(chan struct{}) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseResponse) }) } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + t.Error("test server does not support flushing") + return + } + _, _ = io.WriteString(w, "[\"n\"]\n[\"UInt64\"]\n[1]\n") + flusher.Flush() + close(firstChunkWritten) + <-releaseResponse + _, _ = io.WriteString(w, "[2]\n") + })) + defer func() { + release() + server.Close() + }() + + client := newClickHouseLegacyHTTPTestClient(t, server, nil) + consumer := &legacyHTTPTestStreamConsumer{firstRow: make(chan struct{})} + errCh := make(chan error, 1) + go func() { + errCh <- client.StreamQuery(context.Background(), "SELECT stream", consumer) + }() + + select { + case <-firstChunkWritten: + case <-time.After(2 * time.Second): + t.Fatal("server did not write the first response chunk") + } + select { + case <-consumer.firstRow: + case <-time.After(2 * time.Second): + t.Fatal("first row was not delivered before response EOF") + } + release() + if err := <-errCh; err != nil { + t.Fatalf("stream query failed: %v", err) + } + if !reflect.DeepEqual(consumer.columns, []string{"n"}) { + t.Fatalf("columns = %#v", consumer.columns) + } + if !reflect.DeepEqual(consumer.values, [][]interface{}{{int64(1)}, {int64(2)}}) { + t.Fatalf("values = %#v", consumer.values) + } +} + +func TestClickHouseLegacyHTTPQueryReportsExceptionAppendedAfterRows(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, "[\"n\"]\n[\"UInt64\"]\n[1]\nCode: 241. DB::Exception: memory limit exceeded. (MEMORY_LIMIT_EXCEEDED)\n") + })) + defer server.Close() + + client := newClickHouseLegacyHTTPTestClient(t, server, nil) + rows, columns, err := client.Query(context.Background(), "SELECT fails_late") + if err == nil || !strings.Contains(err.Error(), "memory limit exceeded") { + t.Fatalf("expected trailing ClickHouse exception, got rows=%#v columns=%#v err=%v", rows, columns, err) + } + if len(rows) != 1 || len(columns) != 1 { + t.Fatalf("decoded prefix should remain available, rows=%#v columns=%#v", rows, columns) + } +} + +func TestClickHouseLegacyHTTPExecHandlesSuccessAndErrors(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if got := r.URL.Query().Get("wait_end_of_query"); got != "1" { + t.Errorf("wait_end_of_query = %q for %q", got, body) + } + switch string(body) { + case "ALTER OK": + w.WriteHeader(http.StatusOK) + case "ALTER STATUS ERROR": + w.WriteHeader(http.StatusInternalServerError) + _, _ = io.WriteString(w, "Code: 60. DB::Exception: table does not exist. (UNKNOWN_TABLE)") + case "ALTER TAIL ERROR": + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, "Code: 241. DB::Exception: memory limit exceeded. (MEMORY_LIMIT_EXCEEDED)") + default: + w.WriteHeader(http.StatusBadRequest) + } + })) + defer server.Close() + + client := newClickHouseLegacyHTTPTestClient(t, server, nil) + if affected, err := client.Exec(context.Background(), "ALTER OK"); err != nil || affected != 0 { + t.Fatalf("successful exec = (%d, %v)", affected, err) + } + if _, err := client.Exec(context.Background(), "ALTER STATUS ERROR"); err == nil || !strings.Contains(err.Error(), "table does not exist") { + t.Fatalf("expected HTTP status error, got %v", err) + } + if _, err := client.Exec(context.Background(), "ALTER TAIL ERROR"); err == nil || !strings.Contains(err.Error(), "memory limit exceeded") { + t.Fatalf("expected HTTP 200 body exception, got %v", err) + } +} + +func TestClickHouseApplyChangesUsesLegacyHTTPBackend(t *testing.T) { + var ( + mu sync.Mutex + queries []string + ) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + mu.Lock() + queries = append(queries, string(body)) + mu.Unlock() + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + legacyClient := newClickHouseLegacyHTTPTestClient(t, server, nil) + client := &ClickHouseDB{legacyHTTP: legacyClient, database: "analytics"} + err := client.ApplyChanges("events", connection.ChangeSet{ + Deletes: []map[string]interface{}{{"id": int64(1)}}, + Updates: []connection.UpdateRow{{ + Keys: map[string]interface{}{"id": int64(2)}, + Values: map[string]interface{}{"name": "updated"}, + }}, + Inserts: []map[string]interface{}{{"id": int64(3), "name": "inserted"}}, + }) + if err != nil { + t.Fatalf("ApplyChanges with legacy HTTP failed: %v", err) + } + + mu.Lock() + defer mu.Unlock() + if len(queries) != 3 { + t.Fatalf("queries = %#v", queries) + } + for _, fragment := range []string{"ALTER TABLE `analytics`.`events` DELETE", "ALTER TABLE `analytics`.`events` UPDATE", "INSERT INTO `analytics`.`events`"} { + if !containsStringFragment(queries, fragment) { + t.Fatalf("missing query fragment %q in %#v", fragment, queries) + } + } +} + +func newClickHouseLegacyHTTPTestClient(t *testing.T, server *httptest.Server, settings clickhouse.Settings) *clickHouseLegacyHTTPClient { + t.Helper() + return newClickHouseLegacyHTTPTestClientWithOptions(t, server, func(opts *clickhouse.Options) { + opts.Settings = settings + }) +} + +func newClickHouseLegacyHTTPTestClientWithOptions(t *testing.T, server *httptest.Server, configure func(*clickhouse.Options)) *clickHouseLegacyHTTPClient { + t.Helper() + parsed, err := url.Parse(server.URL) + if err != nil { + t.Fatalf("parse test server URL: %v", err) + } + opts := &clickhouse.Options{ + Protocol: clickhouse.HTTP, + Addr: []string{parsed.Host}, + Auth: clickhouse.Auth{Database: "default"}, + DialTimeout: 2 * time.Second, + ReadTimeout: 2 * time.Second, + } + if configure != nil { + configure(opts) + } + client, err := newClickHouseLegacyHTTPClient(opts) + if err != nil { + t.Fatalf("new legacy HTTP client: %v", err) + } + t.Cleanup(func() { _ = client.Close() }) + return client +} + +type legacyHTTPTestStreamConsumer struct { + mu sync.Mutex + columns []string + values [][]interface{} + firstRow chan struct{} + once sync.Once +} + +func (c *legacyHTTPTestStreamConsumer) SetColumns(columns []string) error { + c.mu.Lock() + defer c.mu.Unlock() + c.columns = append([]string(nil), columns...) + return nil +} + +func (c *legacyHTTPTestStreamConsumer) ConsumeRow(map[string]interface{}) error { + return fmt.Errorf("value consumer fast path was not used") +} + +func (c *legacyHTTPTestStreamConsumer) ConsumeRowValues(values []interface{}) error { + c.mu.Lock() + c.values = append(c.values, append([]interface{}(nil), values...)) + c.mu.Unlock() + c.once.Do(func() { close(c.firstRow) }) + return nil +} + +func containsStringFragment(values []string, fragment string) bool { + for _, value := range values { + if strings.Contains(value, fragment) { + return true + } + } + return false +} diff --git a/internal/db/driver_agent_revisions_gen.go b/internal/db/driver_agent_revisions_gen.go index 12e70499..a5a6b8df 100644 --- a/internal/db/driver_agent_revisions_gen.go +++ b/internal/db/driver_agent_revisions_gen.go @@ -22,8 +22,8 @@ func init() { "mongodb": "src-2610395b35c2e708", "tdengine": "src-779b9b537f08856f", "iotdb": "src-7edea4aba8d4869e", - "clickhouse": "src-0197342ca5afa8b5", - "elasticsearch": "src-08e8e80cb17a409a", + "clickhouse": "src-5872eb0d28e8ab82", + "elasticsearch": "src-85079509f9c31623", "trino": "src-ba947f211ce7b19f", } } diff --git a/internal/db/elasticsearch_impl.go b/internal/db/elasticsearch_impl.go index cd98e90f..c5e1058e 100644 --- a/internal/db/elasticsearch_impl.go +++ b/internal/db/elasticsearch_impl.go @@ -102,6 +102,12 @@ func (e *ElasticsearchDB) Connect(config connection.ConnectionConfig) error { lastErr = err continue } + if _, err := e.GetDatabases(); err != nil { + e.client = nil + logger.Warnf("Elasticsearch 索引枚举验证失败：%d/%d 模式=%s 错误=%v", idx+1, len(attempts), sslLabel, err) + lastErr = err + continue + } logger.Infof("Elasticsearch 连接成功：%d/%d 模式=%s", idx+1, len(attempts), sslLabel) if idx > 0 { @@ -331,9 +337,13 @@ func (e *ElasticsearchDB) GetDatabases() ([]string, error) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - res, err := e.client.Indices.Get( - []string{"*"}, - e.client.Indices.Get.WithContext(ctx), + // Alias API 只返回索引名和别名，避免列表加载下载全部 settings/mappings。 + res, err := e.client.Indices.GetAlias( + e.client.Indices.GetAlias.WithContext(ctx), + e.client.Indices.GetAlias.WithIndex("*"), + e.client.Indices.GetAlias.WithExpandWildcards("all"), + e.client.Indices.GetAlias.WithAllowNoIndices(true), + e.client.Indices.GetAlias.WithIgnoreUnavailable(true), ) if err != nil { return nil, fmt.Errorf("获取索引列表失败：%w", err) diff --git a/internal/db/elasticsearch_impl_test.go b/internal/db/elasticsearch_impl_test.go index 8ee89afd..89adaab3 100644 --- a/internal/db/elasticsearch_impl_test.go +++ b/internal/db/elasticsearch_impl_test.go @@ -9,6 +9,7 @@ import ( "net/http/httptest" "slices" "strings" + "sync/atomic" "testing" "GoNavi-Wails/internal/connection" @@ -104,20 +105,101 @@ func TestElasticsearchPing(t *testing.T) { }) } +func TestElasticsearchConnectValidatesIndexListing(t *testing.T) { + var aliasListingRequested atomic.Bool + server := newMockESServer(t, func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodHead && r.URL.Path == "/": + w.WriteHeader(http.StatusOK) + case r.Method == http.MethodGet && r.URL.Path == "/*/_alias": + aliasListingRequested.Store(true) + w.WriteHeader(http.StatusForbidden) + default: + w.WriteHeader(http.StatusNotFound) + } + }) + + host, port, ok := parseHostPortWithDefault(strings.TrimPrefix(server.URL, "http://"), defaultEsPort) + if !ok { + t.Fatalf("无法解析测试服务器地址：%s", server.URL) + } + + db := &ElasticsearchDB{} + err := db.Connect(connection.ConnectionConfig{ + Type: "elasticsearch", + Host: host, + Port: port, + Timeout: 2, + }) + if err == nil { + t.Fatal("Connect 应在索引枚举被拒绝时失败") + } + if !aliasListingRequested.Load() { + t.Fatal("Connect 应使用轻量 alias 端点验证索引枚举能力") + } + if !strings.Contains(err.Error(), "获取索引列表失败") || !strings.Contains(err.Error(), "403") { + t.Fatalf("Connect 应返回索引枚举错误，实际：%v", err) + } +} + +func TestElasticsearchConnectAllowsEmptyCluster(t *testing.T) { + var aliasListingRequested atomic.Bool + server := newMockESServer(t, func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodHead && r.URL.Path == "/": + w.WriteHeader(http.StatusOK) + case r.Method == http.MethodGet && r.URL.Path == "/*/_alias": + aliasListingRequested.Store(true) + query := r.URL.Query() + if query.Get("allow_no_indices") != "true" || query.Get("ignore_unavailable") != "true" { + w.WriteHeader(http.StatusBadRequest) + return + } + writeJSON(w, map[string]interface{}{}) + default: + w.WriteHeader(http.StatusNotFound) + } + }) + + host, port, ok := parseHostPortWithDefault(strings.TrimPrefix(server.URL, "http://"), defaultEsPort) + if !ok { + t.Fatalf("无法解析测试服务器地址：%s", server.URL) + } + + db := &ElasticsearchDB{} + if err := db.Connect(connection.ConnectionConfig{ + Type: "elasticsearch", + Host: host, + Port: port, + Timeout: 2, + }); err != nil { + t.Fatalf("Connect 应允许没有索引的空集群：%v", err) + } + if !aliasListingRequested.Load() { + t.Fatal("Connect 应验证空集群的索引枚举能力") + } +} + // TestElasticsearchGetDatabases 测试获取索引列表。 func TestElasticsearchGetDatabases(t *testing.T) { - t.Run("正常获取全部索引", func(t *testing.T) { + t.Run("使用轻量别名端点获取全部索引", func(t *testing.T) { + var fullIndexDefinitionsRequested atomic.Bool server := newMockESServer(t, func(w http.ResponseWriter, r *http.Request) { - if r.Method == http.MethodGet && (r.URL.Path == "/" || r.URL.Path == "/*") && !strings.Contains(r.URL.Path, "_") { + if r.Method == http.MethodGet && r.URL.Path == "/*/_alias" { writeJSON(w, map[string]interface{}{ - "logs-2024": map[string]interface{}{}, - "users": map[string]interface{}{}, - ".security": map[string]interface{}{}, - ".kibana_1": map[string]interface{}{}, - "products": map[string]interface{}{}, + "logs-2024": map[string]interface{}{"aliases": map[string]interface{}{}}, + "users": map[string]interface{}{"aliases": map[string]interface{}{}}, + ".security": map[string]interface{}{"aliases": map[string]interface{}{}}, + ".kibana_1": map[string]interface{}{"aliases": map[string]interface{}{}}, + "products": map[string]interface{}{"aliases": map[string]interface{}{}}, }) return } + if r.Method == http.MethodGet && r.URL.Path == "/*" { + fullIndexDefinitionsRequested.Store(true) + w.WriteHeader(http.StatusForbidden) + return + } w.WriteHeader(http.StatusNotFound) }) @@ -137,6 +219,9 @@ func TestElasticsearchGetDatabases(t *testing.T) { t.Fatalf("索引 [%d] 期望 %q，实际 %q", i, name, databases[i]) } } + if fullIndexDefinitionsRequested.Load() { + t.Fatal("GetDatabases 不应请求包含 settings/mappings 的完整索引定义") + } }) t.Run("连接未打开时返回错误", func(t *testing.T) { @@ -998,13 +1083,9 @@ func TestESMockIntegration(t *testing.T) { case r.Method == http.MethodHead && path == "/": w.WriteHeader(http.StatusOK) - // Indices.Get("*") — 返回所有索引 - case r.Method == http.MethodGet && (path == "/" || path == "/*") && !strings.Contains(path, "_"): - writeJSON(w, map[string]interface{}{ - "products": map[string]interface{}{}, - "orders": map[string]interface{}{}, - ".internal": map[string]interface{}{}, - }) + // 完整索引定义响应可能过大，列表加载不应请求该端点。 + case r.Method == http.MethodGet && path == "/*": + w.WriteHeader(http.StatusForbidden) // Indices.GetAlias — 返回别名映射 case strings.Contains(path, "/_alias") && r.Method == http.MethodGet: @@ -1014,6 +1095,8 @@ func TestESMockIntegration(t *testing.T) { "products-alias": map[string]interface{}{}, }, }, + "orders": map[string]interface{}{"aliases": map[string]interface{}{}}, + ".internal": map[string]interface{}{"aliases": map[string]interface{}{}}, }) // Mapping diff --git a/tools/generate-driver-agent-revisions.sh b/tools/generate-driver-agent-revisions.sh index 403535a6..3da5547a 100755 --- a/tools/generate-driver-agent-revisions.sh +++ b/tools/generate-driver-agent-revisions.sh @@ -138,6 +138,7 @@ mongodb:internal/db/mongodb_impl_v1.go|\ tdengine:internal/db/tdengine_impl.go|\ iotdb:internal/db/iotdb_impl.go|\ clickhouse:internal/db/clickhouse_impl.go|\ +clickhouse:internal/db/clickhouse_legacy_http.go|\ elasticsearch:internal/db/elasticsearch_impl.go|\ elasticsearch:internal/db/elasticsearch_helpers.go|\ trino:internal/db/trino_impl.go)