ToolJet 集成 Elasticsearch 数据源完全指南：连接配置、13 种查询操作与源码实现解析-编程实验室

ToolJet 集成 Elasticsearch 数据源完全指南：连接配置、13 种查询操作与源码实现解析

【免费下载链接】ToolJetOpen-source foundation of ToolJet AI - the enterprise app generation platform for internal tools, dashboards, business applications, workflows and AI agents. Build visually, from a prompt, or from Claude Code, Codex and Cursor over MCP 🚀项目地址: https://gitcode.com/GitHub_Trending/to/ToolJet

ToolJet 内置了对 Elasticsearch 集群的完整支持，允许用户在可视化画布中直接对索引执行数据读写与各类查询操作。本文以 ToolJet 3.0.0-LTS 版本文档为骨架，结合仓库中 Elasticsearch 插件源码与操作定义文件，系统讲解数据源连接方式、SSL 证书配置、全部 13 种受支持操作的参数与示例，并深入底层剖析查询执行的真实调用链。读完本文，你将能够独立完成 ToolJet 与 Elasticsearch 的对接，并在查询面板中熟练执行搜索、文档增删改查、批量操作、滚动分页等常见任务。

连接 Elasticsearch 数据源

在 ToolJet 中连接 Elasticsearch 数据源有两种入口：

在查询面板（Query Panel）中点击+ Add new data source按钮；
在 ToolJet 仪表盘中导航到 Data Sources 页面添加。

添加时需要填写以下连接信息：

参数	说明
Host	Elasticsearch 集群所在主机地址，默认值为`localhost`
Port	集群对外服务端口，默认值为`9200`
Username	访问集群的用户名（可留空）
Password	访问集群的密码（可留空）

网络可达性提醒：如果你自托管 ToolJet，请确保 Elasticsearch 的Host/IP能从你的 VPC 内访问；如果使用 ToolJet Cloud，则需要将 ToolJet 的 IP 加入 Elasticsearch 的白名单。

SSL 证书连接支持

ToolJet 同时支持基于 SSL 证书的加密连接。在 SSL 配置中，你可以选择以下两种证书方式之一：

CA certificate：上传 CA 证书内容，用于校验服务端身份；
Client certificate：同时提供客户端证书（Client cert）、客户端私钥（Client key）与根证书（Root cert），实现双向 TLS 认证。

从插件的数据源描述文件 manifest.json 可以看到，ssl_certificate是一个下拉组件，可选值为ca_certificate、client_certificate、none（默认），并且password、ca_cert、client_key、client_cert、root_cert均被标记为encrypted: true，意味着这些敏感字段在存储时会经过加密处理。

连接配置的源码级解读

插件的数据源连接逻辑位于 index.ts 的getConnection方法中，其构建客户端的过程清晰揭示了表单字段的底层映射：

const host = sourceOptions.host; const port = sourceOptions.port; const username = encodeURIComponent(sourceOptions.username); const password = encodeURIComponent(sourceOptions.password); const sslEnabled = sourceOptions.ssl_enabled; const protocol = this.determineProtocol(sourceOptions);

URL 组装：用户名与密码会先经过encodeURIComponent编码，再以${protocol}://${username}:${password}@${host}:${port}的形式拼入节点地址；若用户名密码均为空，则直接使用${protocol}://${host}:${port}；
协议判定：determineProtocol方法（index.ts）优先读取历史遗留的scheme字段——早期版本将协议硬编码为https，若存在scheme且未设置ssl_enabled，则向后兼容地返回https；否则按ssl_enabled开关返回https或http；
SSL 装配：当启用 SSL 且选择ca_certificate时，仅注入ca；选择client_certificate时则同时注入ca、cert、key三个字段，分别对应表单中的根证书、客户端证书与客户端私钥；
请求超时：客户端统一设置了requestTimeout: 10000（10 秒），因此对慢查询请结合业务在操作参数层面控制数据规模。

需要说明的是，插件底层使用的是 OpenSearch 官方 Node.js 客户端（@opensearch-project/opensearch，见 package.json），它兼容 Elasticsearch 的 REST API 语义。这一点在响应示例的meta元数据中也有体现："name": "opensearch-js"以及"user-agent": "opensearch-js/1.2.0"。

在查询面板中使用 Elasticsearch

连接建立后，按以下步骤发起查询：

点击编辑器底部查询管理器（Query Manager）中的+ Add按钮，选择前面添加的 Elasticsearch 数据源；
选择要执行的操作（Operation）。

提示：查询结果可以通过 Transformations（数据变换）进一步加工处理，具体用法参见 Transformations 教程。

插件支持的 13 种操作由 operations.json 中的operation下拉列表定义，运行时则由 index.ts 中的switch语句分派到对应的执行函数。下面逐一说明每种操作的参数与示例。

支持的 13 种操作详解

Search（搜索）

执行搜索查询并返回匹配的命中结果（hits）。

必填参数

Index：要搜索的索引名称；
Query：JSON 格式的搜索查询体。

可选参数

Scroll：滚动（scroll）保留时间，用于分批拉取大量结果。

示例

Index: books Query: { "query": { "match": { "title": "The Great Gatsby" } }, "size": 20 } Scroll: 1m # 时间格式支持 1m、1h、1d 等

源码实现（operations.ts）将query通过JSON.parse解析后传入client.search，并把可选的scroll透传给底层客户端——这正是 Search 操作中Scroll参数能生效的原理。

Index a Document（索引文档）

向指定索引或数据流新增一条 JSON 文档。

必填参数

Index：要写入文档的索引名称；
Body：JSON 格式的文档内容。

示例

Index: books Body: { "title": "1984", "author": "George Orwell", "year": 1949, "genre": "Dystopian Fiction" }

Get a Document（获取文档）

按文档 ID 从索引中取回指定 JSON 文档。

必填参数

Index：文档所在的索引名称；
Id：要检索的文档 ID。

示例

Index: books Id: FJXTSZEBsuzUn2y4wZ-W

Update a Document（更新文档）

使用脚本或部分文档对指定文档执行更新。

必填参数

Index：文档所在索引名称；
Id：要更新的文档 ID；
Body：JSON 格式的更新脚本或部分文档（如{ "doc": { ... } }）。

示例

Index: books Id: FJXTSZEBsuzUn2y4wZ-W Body: { "doc": { "title": "1984", "author": "George Orwell", "year": 1949, "genre": "Fiction" } }

Delete a Document（删除文档）

从指定索引中删除一条 JSON 文档。

必填参数

Index：文档所在索引名称；
Id：要删除的文档 ID。

示例

Index: books Id: FJXTSZEBsuzUn2y4wZ-W

Bulk Operation（批量操作）

在一次 API 调用中执行多条 index/update/delete 操作，适合大批量数据写入场景。

必填参数

Operations：JSON 格式的批量操作列表。

示例

[ { "index": { "_index": "books", "_id": "book1" } }, { "title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "year": 1925 }, { "delete": { "_index": "books", "_id": "book2" } }, { "index": { "_index": "books", "_id": "book3" } }, { "title": "Moby-Dick", "author": "Herman Melville", "year": 1851 }, { "delete": { "_index": "books", "_id": "book4" } } ]

批量操作在源码中直接映射为client.bulk({ body: JSON.parse(operations) })（operations.ts），操作列表整体作为 bulk body 提交。

Count Documents（统计文档数）

返回满足搜索条件的文档数量。

必填参数

Index：要统计文档数的索引。

可选参数

Query：JSON 格式的过滤查询（可选）。

示例

{ "query": { "range": { "timestamp": { "gte": 1901 } } } }

源码实现（operations.ts）会在query为空时直接不传 body，相当于统计索引内全部文档。

Check Document Existence（检查文档是否存在）

判断指定索引中是否存在某条文档。

必填参数

Index：待检查的索引名称；
Id：待检查的文档 ID。

示例

Index: books Id: FJXTSZEBsuzUn2y4wZ-W

该操作底层调用client.exists，返回布尔值，适合在画布逻辑中做条件判断。

Multi Get（批量获取）

通过一次请求批量取回多条文档。

必填参数

Operations：JSON 格式的批量获取操作。

示例

{ "docs": [ { "_index": "books", "_id": "book124" }, { "_index": "books", "_id": "book125" } ] }

Scroll Search（滚动搜索）

基于上一次 Search 返回的 Scroll ID，持续分批拉取单次搜索请求的大量结果。

必填参数

Scroll ID：搜索返回的滚动 ID；
Scroll：滚动保留时间。

示例

Scroll ID: DXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAOWQWYm9vbDItY1NCOUExal9TcTBjeUEyZw Scroll: 60m

使用滚动搜索的典型流程是：先用带Scroll参数的 Search 操作取得首批结果与_scroll_id，再反复使用 Scroll Search 拉取后续批次，最后用 Clear Scroll 释放搜索上下文。

Clear Scroll（清理滚动）

释放指定滚动搜索占用的服务端上下文。

必填参数

Scroll ID：要清理的滚动 ID。

示例

Scroll ID: DXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAOWQWYm9vbDItY1NCOUExal9TcTBjeUEyZw

Get Cat Indices（获取索引列表）

以紧凑、列对齐的形式查看集群中的索引概况。该操作无必填参数，点击即可执行。底层调用client.cat.indices({ format: 'json' })（operations.ts），因此返回体中的body是一个 JSON 数组。

响应示例

{ "body": [ { "health": "yellow", "status": "open", "index": "1", "uuid": "JQOzqxK7Rdar7ROOlqXwkA", "pri": "1", "rep": "1", "docs.count": "2", "docs.deleted": "0", "store.size": "9.2kb", "pri.store.size": "9.2kb" }, { "health": "yellow", "status": "open", "index": "recipes", "uuid": "eNGdAsG4TMWvs9f0eLERlQ", "pri": "1", "rep": "1", "docs.count": "20", "docs.deleted": "0", "store.size": "30kb", "pri.store.size": "30kb" } ], "statusCode": 200, "headers": { "x-elastic-product": "Elasticsearch", "content-type": "application/json", "content-length": "558" }, "meta": { "name": "opensearch-js", "connection": { "url": "http://xx.2xx.183.199:9200/", "status": "alive" } } }

Get Cluster Health（获取集群健康状态）

返回集群健康状态信息（green/yellow/red），无必填参数。底层调用client.cluster.health()（operations.ts）。

响应示例

{ "body": { "cluster_name": "docker-cluster", "status": "yellow", "timed_out": false, "number_of_nodes": 1, "number_of_data_nodes": 1, "active_primary_shards": 10, "active_shards": 10, "relocating_shards": 0, "initializing_shards": 0, "unassigned_shards": 3, "delayed_unassigned_shards": 0, "number_of_pending_tasks": 0, "number_of_in_flight_fetch": 0, "task_max_waiting_in_queue_millis": 0, "active_shards_percent_as_number": 76.92307692307693 }, "statusCode": 200 }

上述两个响应示例中的meta字段来自底层客户端（opensearch-js），其中connection.url即你配置的集群节点地址。实际使用时可将{{queries.<查询名>.data.body}}之类的表达式用于前端组件渲染，实现索引列表与集群状态的实时可视化。

查询执行链路与错误处理

从源码结构看，所有操作共用同一条执行链路（index.ts）：

run方法首先通过getConnection根据数据源配置构建并复用 OpenSearch 客户端；
根据queryOptions.operation进入switch分支，将index、query、body、id、operations、scroll_id、scroll等参数分发到 operations.ts 中对应的函数（其中query、body、operations均先经过JSON.parse解析为对象）；
任一环节抛出异常都会被捕获并包装为QueryError('Query could not be completed', err.message, {})返回，避免将底层堆栈直接暴露给前端；
成功时统一返回{ status: 'ok', data: result }结构。

此外，数据源还实现了testConnection方法（index.ts），通过调用client.info()探测集群连通性，这也是你在连接配置界面点击“测试连接”按钮时的底层行为。

实战建议

参数中的表达式能力：所有操作的 Index、Id、Query、Body 等输入框均为代码提示（codehinter）组件（见 operations.json），支持直接引用画布组件状态、全局变量或前一个查询的输出（如{{ components.table1.selectedRow.id }}），因此完全可以把 Search 的Id或Query做成动态值，实现“选中表格行 → 查询对应文档”的典型联动场景；
大批量导出优先滚动：数据量超过单次返回上限（如size: 20）时，使用 Scroll 参数 + Scroll Search 分批拉取，结束后务必 Clear Scroll 释放上下文；
敏感信息加密存储：密码与各类证书在插件定义中均标记为encrypted，可以放心将连接配置保存在组织内共享；
测试文件位置：插件的自动化测试骨架位于 plugins/packages/elasticsearch/tests/elasticsearch.test.js，后续如需为自定义操作补充回归用例，可在此扩展。