diff --git a/.github/workflows/collector-native-build.yml b/.github/workflows/collector-native-build.yml
new file mode 100644
index 00000000000..d44877dba8f
--- /dev/null
+++ b/.github/workflows/collector-native-build.yml
@@ -0,0 +1,89 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: Collector Native Release
+
+run-name: Native collector release build (${{ github.ref_name }})
+
+on:
+ workflow_dispatch:
+
+jobs:
+ build-native-collector:
+ name: Native collector (${{ matrix.platform }})
+ permissions:
+ contents: read
+ timeout-minutes: 120
+ runs-on: ${{ matrix.runner }}
+ strategy:
+ fail-fast: false
+ matrix:
+ include:
+ - platform: linux-amd64
+ runner: ubuntu-24.04
+ archive_ext: tar.gz
+ - platform: linux-arm64
+ runner: ubuntu-24.04-arm
+ archive_ext: tar.gz
+ - platform: macos-amd64
+ runner: macos-13
+ archive_ext: tar.gz
+ - platform: macos-arm64
+ runner: macos-14
+ archive_ext: tar.gz
+ - platform: windows-amd64
+ runner: windows-latest
+ archive_ext: zip
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up GraalVM JDK 25
+ uses: graalvm/setup-graalvm@v1
+ with:
+ distribution: graalvm-community
+ java-version: '25'
+ github-token: ${{ secrets.GITHUB_TOKEN }}
+ cache: maven
+ native-image-job-reports: 'true'
+
+ - name: Verify toolchain
+ shell: pwsh
+ run: |
+ java -version
+ native-image --version
+ mvn -version
+
+ - name: Build native collector package
+ run: mvn -B -pl hertzbeat-collector/hertzbeat-collector-collector -am -Pnative -DskipTests package
+
+ - name: Locate native collector package
+ id: package
+ shell: pwsh
+ run: |
+ $package = Get-ChildItem -Path "dist/apache-hertzbeat-collector-native-*-${{ matrix.platform }}-bin.${{ matrix.archive_ext }}" | Select-Object -First 1
+ if (-not $package) {
+ throw "Native collector package not found for ${{ matrix.platform }}"
+ }
+ "archive=$($package.FullName)" >> $env:GITHUB_OUTPUT
+
+ - name: Upload native collector package
+ uses: actions/upload-artifact@v4
+ with:
+ name: apache-hertzbeat-collector-native-${{ matrix.platform }}
+ path: ${{ steps.package.outputs.archive }}
+ retention-days: 14
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 1a1738267c9..9f329883578 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -32,7 +32,7 @@ Even small corrections to typos are very welcome :)
#### Backend start
-1. Requires `maven3+`, `java21` and `lombok` environments
+1. Requires `maven3+`, `java25` and `lombok` environments
2. (Optional) Modify the configuration file: `hertzbeat-startup/src/main/resources/application.yml`
@@ -172,7 +172,7 @@ Add WeChat account `ahertzbeat` to pull you into the WeChat group.
#### 后端启动
-1. 需要 `maven3+`, `java21` 和 `lombok` 环境
+1. 需要 `maven3+`, `java25` 和 `lombok` 环境
2. (可选)修改配置文件配置信息-`hertzbeat-startup/src/main/resources/application.yml`
diff --git a/README.md b/README.md
index 0d8cff96a67..eb8cac645f8 100644
--- a/README.md
+++ b/README.md
@@ -130,12 +130,12 @@ Detailed config refer to [Install HertzBeat via Docker](https://hertzbeat.apache
##### 2:Install via package
-1. Download the release package `hertzbeat-xx.tar.gz` [Download](https://hertzbeat.apache.org/docs/download)
+1. Download the release package `apache-hertzbeat-xx-bin.tar.gz` [Download](https://hertzbeat.apache.org/docs/download)
2. Configure the HertzBeat configuration yml file `hertzbeat/config/application.yml` (optional)
3. Run command `$ ./bin/startup.sh ` or `bin/startup.bat`
4. Access `http://localhost:1157` to start, default account: `admin/hertzbeat`
5. Deploy collector clusters (Optional)
- - Download the release package `hertzbeat-collector-xx.tar.gz` to new machine [Download](https://hertzbeat.apache.org/docs/download)
+ - Download the release package `apache-hertzbeat-collector-xx-bin.tar.gz` (JVM collector) or the native collector package for your platform, such as `apache-hertzbeat-collector-native-xx-linux-amd64-bin.tar.gz` or `apache-hertzbeat-collector-native-xx-windows-amd64-bin.zip`, to the new machine [Download](https://hertzbeat.apache.org/docs/download)
- Configure the collector configuration yml file `hertzbeat-collector/config/application.yml`: unique `identity` name, running `mode` (public or private), hertzbeat `manager-host`, hertzbeat `manager-port`
```yaml
collector:
@@ -148,7 +148,9 @@ Detailed config refer to [Install HertzBeat via Docker](https://hertzbeat.apache
manager-host: ${MANAGER_HOST:127.0.0.1}
manager-port: ${MANAGER_PORT:1158}
```
- - Run command `$ ./bin/startup.sh ` or `bin/startup.bat`
+ - If you do not provide JDBC drivers in `ext-lib`, MySQL, MariaDB, and OceanBase can use the built-in query engine and run on the native collector package as well. TiDB follows the same rule for its SQL query metric set.
+ - If `mysql-connector-j` is present in `ext-lib`, the built-in server collector or JVM collector automatically prefers JDBC after restart for MySQL, MariaDB, and OceanBase. TiDB follows the same rule for its SQL query metric set, while its HTTP metrics are unchanged. Oracle and DB2 still require the JVM collector package because they depend on external JDBC drivers.
+ - Run `$ ./bin/startup.sh ` or `bin/startup.bat` for the JVM collector package. Run `$ ./bin/startup.sh ` for Linux or macOS native collector packages, and `bin\\startup.bat` for the Windows native collector package.
- Access `http://localhost:1157` and you will see the registered new collector in dashboard
Detailed config refer to [Install HertzBeat via Package](https://hertzbeat.apache.org/docs/start/package-deploy)
@@ -156,7 +158,7 @@ Detailed config refer to [Install HertzBeat via Package](https://hertzbeat.apach
##### 3:Start via source code
1. Local source code debugging needs to start the back-end project `hertzbeat-startup` and the front-end project `web-app`.
-2. Backend:need `maven3+`, `java21`, `lombok`, add VM options in IDE: ` --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED `, then start the `hertzbeat-startup` service.
+2. Backend:need `maven3+`, `java25`, `lombok`, add VM options in IDE: ` --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED `, then start the `hertzbeat-startup` service.
3. Web:need `nodejs npm angular-cli` environment, Run `ng serve --open` in `web-app` directory after backend startup.
4. Access `http://localhost:4200` to start, default account: `admin/hertzbeat`
diff --git a/README_CN.md b/README_CN.md
index 7393ace7010..80058e8f7b4 100644
--- a/README_CN.md
+++ b/README_CN.md
@@ -127,12 +127,12 @@
##### 方式二:通过安装包安装
-1. 下载您系统环境对应的安装包 `hertzbeat-xx.tar.gz` [Download](https://hertzbeat.apache.org/docs/download)
+1. 下载您系统环境对应的安装包 `apache-hertzbeat-xx-bin.tar.gz` [Download](https://hertzbeat.apache.org/docs/download)
2. 配置 HertzBeat 的配置文件 `hertzbeat/config/application.yml` (可选)
3. 部署启动 `$ ./bin/startup.sh ` 或 `bin/startup.bat`
4. 浏览器访问 `http://localhost:1157` 即可开始,默认账号密码 `admin/hertzbeat`
5. 部署采集器集群(可选)
- - 下载采集器安装包 `hertzbeat-collector-xx.tar.gz` 到规划的另一台部署主机上 [Download](https://hertzbeat.apache.org/docs/download)
+ - 下载采集器安装包 `apache-hertzbeat-collector-xx-bin.tar.gz`(JVM 采集器)或与你目标平台匹配的 Native 采集器安装包,例如 `apache-hertzbeat-collector-native-xx-linux-amd64-bin.tar.gz`、`apache-hertzbeat-collector-native-xx-windows-amd64-bin.zip`,到规划的另一台部署主机上 [Download](https://hertzbeat.apache.org/docs/download)
- 配置采集器的配置文件 `hertzbeat-collector/config/application.yml` 里面的连接主 HertzBeat 服务的对外 IP,端口,当前采集器名称(需保证唯一性)等参数 `identity` `mode` (public or private) `manager-host` `manager-port`
```yaml
collector:
@@ -145,7 +145,9 @@
manager-host: ${MANAGER_HOST:127.0.0.1}
manager-port: ${MANAGER_PORT:1158}
```
- - 启动 `$ ./bin/startup.sh ` 或 `bin/startup.bat`
+ - 如果没有在 `ext-lib` 中提供 JDBC 驱动,MySQL、MariaDB、OceanBase 可以直接使用内置查询引擎,也可以使用 Native 采集器安装包;TiDB 的 SQL 查询指标也遵循同样规则。
+ - 如果在 `ext-lib` 中放入了 `mysql-connector-j`,主程序内置采集器或 JVM 采集器会在重启后自动优先走 JDBC;这一点现在适用于 MySQL、MariaDB、OceanBase,TiDB 的 SQL 查询指标也遵循同样规则,而它的 HTTP 指标不受影响。Oracle、DB2 仍然必须使用 JVM 采集器安装包,因为它们依赖外置 JDBC 驱动。
+ - JVM 采集器安装包使用 `$ ./bin/startup.sh ` 或 `bin/startup.bat` 启动。Linux 或 macOS 的 Native 采集器安装包使用 `$ ./bin/startup.sh ` 启动,Windows 的 Native 采集器安装包使用 `bin\\startup.bat` 启动
- 浏览器访问主 HertzBeat 服务 `http://localhost:1157` 查看概览页面即可看到注册上来的新采集器
更多配置详细步骤参考 [通过安装包安装HertzBeat](https://hertzbeat.apache.org/docs/start/package-deploy)
@@ -153,7 +155,7 @@
##### 方式三:本地代码启动
1. 此为前后端分离项目,本地代码调试需要分别启动后端工程 `hertzbeat-startup` 和前端工程 `web-app`
-2. 后端:需要 `maven3+`, `java21` 和 `lombok` 环境,修改 `YML` 配置信息,添加JVM参数`--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED`后启动 `hertzbeat-startup` 服务即可。
+2. 后端:需要 `maven3+`, `java25` 和 `lombok` 环境,修改 `YML` 配置信息,添加JVM参数`--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED`后启动 `hertzbeat-startup` 服务即可。
3. 前端:需要 `nodejs npm angular-cli`环境,待本地后端启动后,在 `web-app` 目录下启动 `ng serve --open`
4. 浏览器访问 `http://localhost:4200` 即可开始,默认账号密码 `admin/hertzbeat`
diff --git a/README_JP.md b/README_JP.md
index 13c49de6a4f..4487344ce90 100644
--- a/README_JP.md
+++ b/README_JP.md
@@ -108,7 +108,7 @@
docker run -d -p 1157:1157 -p 1158:1158 --name hertzbeat apache/hertzbeat
```
-2. スタート:`http://localhost:4200`にアクセスします。デフォルトのアカウントとパスワード:`admin/hertzbeat`。
+2. スタート:`http://localhost:1157`にアクセスします。デフォルトのアカウントとパスワード:`admin/hertzbeat`。
3. コレクタークラスタのデプロイメント(オプション)
@@ -126,13 +126,13 @@
##### 方式2:インストールパッケージ
-1. リリースパッケージをダウンロード `hertzbeat-xx.tar.gz` [Download](https://hertzbeat.apache.org/docs/download)
-2. HertzBeatのymlファイルを設定 `hertzbeat/config/application.yml` (オプション)
-3. コマンド`$ ./bin/startup.sh`または`bin/startup.bat`を実行
-4. スタート:`http://localhost:4200`にアクセスします。デフォルトのアカウントとパスワード:`admin/hertzbeat`
+1. リリースパッケージ `apache-hertzbeat-xx-bin.tar.gz` をダウンロードします [Download](https://hertzbeat.apache.org/docs/download)
+2. HertzBeat の設定ファイル `hertzbeat/config/application.yml` を編集します(任意)
+3. コマンド `$ ./bin/startup.sh` または `bin/startup.bat` を実行します
+4. ブラウザで `http://localhost:1157` にアクセスします。デフォルトのアカウントとパスワードは `admin/hertzbeat` です
5. コレクタークラスタのデプロイメント(オプション)
- - コレクターパッケージを別のホストにダウンロード `hertzbeat-collector-xx.tar.gz` [Download](https://hertzbeat.apache.org/docs/download)
- - コレクターのymlファイルを設定 `hertzbeat-collector/config/application.yml`
+ - 別ホストにコレクターのインストールパッケージ `apache-hertzbeat-collector-xx-bin.tar.gz`(JVM コレクター)または対象プラットフォーム向けの Native コレクターパッケージ(例: `apache-hertzbeat-collector-native-xx-linux-amd64-bin.tar.gz`、`apache-hertzbeat-collector-native-xx-windows-amd64-bin.zip`)をダウンロードします [Download](https://hertzbeat.apache.org/docs/download)
+ - コレクターの設定ファイル `hertzbeat-collector/config/application.yml` を編集します
```yaml
collector:
dispatch:
@@ -148,15 +148,17 @@
- `mode: ${MODE:public}`:実行モード(パブリッククラスタまたはプライベートクラウドエッジ)。
- `manager-host: ${MANAGER_HOST:127.0.0.1}`:メインhertzbeatサーバーのIP。
- `manager-port: ${MANAGER_PORT:1158}`:メインhertzbeatサーバポート。
- - コマンド`$ ./bin/startup.sh`または`bin/startup.bat`を実行。
- - `http://localhost:1157`にアクセスし、登録された新しいコレクターを見ることがでます。
+ - `ext-lib` に JDBC ドライバーを置かない場合、MySQL、MariaDB、OceanBase は組み込みのクエリエンジンを使って Native コレクターパッケージでも監視できます。TiDB も SQL クエリのメトリクスセットについては同じルールです。
+ - `ext-lib` に `mysql-connector-j` を置いた場合は、再起動後に組み込みサーバーコレクターまたは JVM コレクターが MySQL、MariaDB、OceanBase で自動的に JDBC を優先します。TiDB も SQL クエリのメトリクスセットについては同じルールで、HTTP メトリクスは影響を受けません。Oracle と DB2 は引き続き外部 JDBC ドライバーに依存するため、JVM コレクターパッケージを使用してください。
+ - JVM コレクターのインストールパッケージは `$ ./bin/startup.sh` または `bin/startup.bat`、Linux/macOS の Native コレクターパッケージは `$ ./bin/startup.sh`、Windows の Native コレクターパッケージは `bin\\startup.bat` で起動します。
+ - メインの HertzBeat サービス `http://localhost:1157` にアクセスすると、登録された新しいコレクターを確認できます。
詳細ステップ [通过安装包安装HertzBeat](https://hertzbeat.apache.org/docs/start/package-deploy)
##### 方式3:ローカルの実行
1. ローカルの実行には、バックエンドのプロジェクト`hertzbeat-startup`とフロントエンドのプロジェクト`web-app`を起動する必要があります。
-2. バックエンド:`maven3+`、`Java21`と`lombok`の環境は必要です。`YML` 設定を修正し、Java仮想マシンパラメータに`--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED` を追加し、`hertzbeat-startup` を起動します。
+2. バックエンド:`maven3+`、`Java25`、`lombok` の環境が必要です。`YML` 設定を修正し、Java 仮想マシンパラメータに `--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED` を追加して `hertzbeat-startup` を起動します。
3. フロントエンド:`nodejs npm angular-cli`の環境は必要です。ローカルのバックエンドが立ち上がったら、`web-app` ディレクトリで `ng serve --open` というコマンドを実行します。
4. スタート:`http://localhost:4200`にアクセスします。デフォルトのアカウントとパスワード:`admin/hertzbeat`。
diff --git a/hertzbeat-ai/pom.xml b/hertzbeat-ai/pom.xml
index 2b9298b6b5e..235b2481dce 100644
--- a/hertzbeat-ai/pom.xml
+++ b/hertzbeat-ai/pom.xml
@@ -27,7 +27,7 @@
${hertzbeat.version}
1.1.1
- 21
+ 25
diff --git a/hertzbeat-collector/hertzbeat-collector-basic/pom.xml b/hertzbeat-collector/hertzbeat-collector-basic/pom.xml
index c43972cda0a..63c88e262df 100644
--- a/hertzbeat-collector/hertzbeat-collector-basic/pom.xml
+++ b/hertzbeat-collector/hertzbeat-collector-basic/pom.xml
@@ -30,8 +30,8 @@
${project.artifactId}
- 17
- 17
+ ${java.version}
+ ${java.version}
UTF-8
1.2.5
@@ -67,12 +67,6 @@
commons-net
commons-net
-
-
- com.mysql
- mysql-connector-j
- provided
-
com.clickhouse
@@ -189,4 +183,4 @@
${zookeeper.version}
-
\ No newline at end of file
+
diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/JdbcCommonCollect.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/JdbcCommonCollect.java
index 42529c72842..76e9421a6a2 100644
--- a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/JdbcCommonCollect.java
+++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/JdbcCommonCollect.java
@@ -20,7 +20,6 @@
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
-import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
@@ -34,6 +33,9 @@
import org.apache.hertzbeat.collector.collect.common.cache.CacheIdentifier;
import org.apache.hertzbeat.collector.collect.common.cache.GlobalConnectionCache;
import org.apache.hertzbeat.collector.collect.common.cache.JdbcConnect;
+import org.apache.hertzbeat.collector.collect.database.query.JdbcQueryExecutor;
+import org.apache.hertzbeat.collector.collect.database.query.JdbcQueryExecutorRegistry;
+import org.apache.hertzbeat.collector.collect.database.query.JdbcQueryRowSet;
import org.apache.hertzbeat.collector.collect.common.ssh.SshTunnelHelper;
import org.apache.hertzbeat.collector.constants.CollectorConstants;
import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
@@ -205,31 +207,26 @@ public void preCheck(Metrics metrics) throws IllegalArgumentException {
public void collect(CollectRep.MetricsData.Builder builder, Metrics metrics) {
long startTime = System.currentTimeMillis();
JdbcProtocol jdbcProtocol = metrics.getJdbc();
- SshTunnel sshTunnel = jdbcProtocol.getSshTunnel();
-
int timeout = CollectUtil.getTimeout(jdbcProtocol.getTimeout());
boolean reuseConnection = Boolean.parseBoolean(jdbcProtocol.getReuseConnection());
- Statement statement = null;
- String databaseUrl;
try {
- if (sshTunnel != null && Boolean.parseBoolean(sshTunnel.getEnable())) {
- int localPort = SshTunnelHelper.localPortForward(sshTunnel, jdbcProtocol.getHost(), jdbcProtocol.getPort());
- databaseUrl = constructDatabaseUrl(jdbcProtocol, "localhost", String.valueOf(localPort));
- } else {
- databaseUrl = constructDatabaseUrl(jdbcProtocol, jdbcProtocol.getHost(), jdbcProtocol.getPort());
- }
-
- statement = getConnection(jdbcProtocol.getUsername(),
- jdbcProtocol.getPassword(), databaseUrl, timeout, reuseConnection);
switch (jdbcProtocol.getQueryType()) {
- case QUERY_TYPE_ONE_ROW -> queryOneRow(statement, jdbcProtocol.getSql(), metrics.getAliasFields(), builder, startTime);
- case QUERY_TYPE_MULTI_ROW -> queryMultiRow(statement, jdbcProtocol.getSql(), metrics.getAliasFields(), builder, startTime);
- case QUERY_TYPE_COLUMNS -> queryOneRowByMatchTwoColumns(statement, jdbcProtocol.getSql(), metrics.getAliasFields(), builder, startTime);
- case RUN_SCRIPT -> {
- Connection connection = statement.getConnection();
- FileSystemResource rc = new FileSystemResource(jdbcProtocol.getSql());
- ScriptUtils.executeSqlScript(connection, rc);
+ case QUERY_TYPE_ONE_ROW -> {
+ try (JdbcQueryRowSet rowSet = executeQuery(metrics, timeout, reuseConnection, 1)) {
+ queryOneRow(rowSet, metrics.getAliasFields(), builder, startTime);
+ }
+ }
+ case QUERY_TYPE_MULTI_ROW -> {
+ try (JdbcQueryRowSet rowSet = executeQuery(metrics, timeout, reuseConnection, 1000)) {
+ queryMultiRow(rowSet, metrics.getAliasFields(), builder, startTime);
+ }
}
+ case QUERY_TYPE_COLUMNS -> {
+ try (JdbcQueryRowSet rowSet = executeQuery(metrics, timeout, reuseConnection, 1000)) {
+ queryOneRowByMatchTwoColumns(rowSet, metrics.getAliasFields(), builder, startTime);
+ }
+ }
+ case RUN_SCRIPT -> runScript(metrics, timeout, reuseConnection);
default -> {
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg("Not support database query type: " + jdbcProtocol.getQueryType());
@@ -261,23 +258,6 @@ public void collect(CollectRep.MetricsData.Builder builder, Metrics metrics) {
log.error("Jdbc error: {}.", errorMessage, e);
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg("Query Error: " + errorMessage);
- } finally {
- if (statement != null) {
- Connection connection = null;
- try {
- connection = statement.getConnection();
- statement.close();
- } catch (Exception e) {
- log.error("Jdbc close statement error: {}", e.getMessage());
- }
- try {
- if (!reuseConnection && connection != null) {
- connection.close();
- }
- } catch (Exception e) {
- log.error("Jdbc close connection error: {}", e.getMessage());
- }
- }
}
}
@@ -286,6 +266,52 @@ public String supportProtocol() {
return DispatchConstants.PROTOCOL_JDBC;
}
+ private JdbcQueryRowSet executeQuery(Metrics metrics, int timeout, boolean reuseConnection, int maxRows) throws Exception {
+ Optional executor = JdbcQueryExecutorRegistry.resolve(metrics);
+ if (executor.isPresent()) {
+ return executor.get().executeQuery(metrics, timeout, maxRows);
+ }
+ return executeJdbcQuery(metrics.getJdbc(), timeout, reuseConnection, maxRows);
+ }
+
+ private JdbcQueryRowSet executeJdbcQuery(JdbcProtocol jdbcProtocol, int timeout, boolean reuseConnection,
+ int maxRows) throws Exception {
+ Statement statement = null;
+ try {
+ String databaseUrl = resolveDatabaseUrl(jdbcProtocol);
+ statement = getConnection(jdbcProtocol.getUsername(),
+ jdbcProtocol.getPassword(), databaseUrl, timeout, reuseConnection);
+ statement.setMaxRows(maxRows);
+ return new ResultSetJdbcQueryRowSet(statement, statement.executeQuery(jdbcProtocol.getSql()), reuseConnection);
+ } catch (Exception exception) {
+ closeStatementAndConnection(statement, reuseConnection);
+ throw exception;
+ }
+ }
+
+ private void runScript(Metrics metrics, int timeout, boolean reuseConnection) throws Exception {
+ JdbcProtocol jdbcProtocol = metrics.getJdbc();
+ Statement statement = null;
+ try {
+ String databaseUrl = resolveDatabaseUrl(jdbcProtocol);
+ statement = getConnection(jdbcProtocol.getUsername(),
+ jdbcProtocol.getPassword(), databaseUrl, timeout, reuseConnection);
+ Connection connection = statement.getConnection();
+ FileSystemResource rc = new FileSystemResource(jdbcProtocol.getSql());
+ ScriptUtils.executeSqlScript(connection, rc);
+ } finally {
+ closeStatementAndConnection(statement, reuseConnection);
+ }
+ }
+
+ private String resolveDatabaseUrl(JdbcProtocol jdbcProtocol) throws Exception {
+ SshTunnel sshTunnel = jdbcProtocol.getSshTunnel();
+ if (sshTunnel != null && Boolean.parseBoolean(sshTunnel.getEnable())) {
+ int localPort = SshTunnelHelper.localPortForward(sshTunnel, jdbcProtocol.getHost(), jdbcProtocol.getPort());
+ return constructDatabaseUrl(jdbcProtocol, "localhost", String.valueOf(localPort));
+ }
+ return constructDatabaseUrl(jdbcProtocol, jdbcProtocol.getHost(), jdbcProtocol.getPort());
+ }
private Statement getConnection(String username, String password, String url, Integer timeout, boolean reuseConnection) throws Exception {
CacheIdentifier identifier = CacheIdentifier.builder()
@@ -343,29 +369,25 @@ private Statement getConnection(String username, String password, String url, In
* query metrics:one tow three four
* query sql:select one, tow, three, four from book limit 1;
*
- * @param statement statement
- * @param sql sql
+ * @param rowSet row set
* @param columns query metrics field list
* @throws Exception when error happen
*/
- private void queryOneRow(Statement statement, String sql, List columns,
+ private void queryOneRow(JdbcQueryRowSet rowSet, List columns,
CollectRep.MetricsData.Builder builder, long startTime) throws Exception {
- statement.setMaxRows(1);
- try (ResultSet resultSet = statement.executeQuery(sql)) {
- if (resultSet.next()) {
- CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
- for (String column : columns) {
- if (CollectorConstants.RESPONSE_TIME.equals(column)) {
- long time = System.currentTimeMillis() - startTime;
- valueRowBuilder.addColumn(String.valueOf(time));
- } else {
- String value = resultSet.getString(column);
- value = value == null ? CommonConstants.NULL_VALUE : value;
- valueRowBuilder.addColumn(value);
- }
+ if (rowSet.next()) {
+ CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
+ for (String column : columns) {
+ if (CollectorConstants.RESPONSE_TIME.equals(column)) {
+ long time = System.currentTimeMillis() - startTime;
+ valueRowBuilder.addColumn(String.valueOf(time));
+ } else {
+ String value = rowSet.getString(column);
+ value = value == null ? CommonConstants.NULL_VALUE : value;
+ valueRowBuilder.addColumn(value);
}
- builder.addValueRow(valueRowBuilder.build());
}
+ builder.addValueRow(valueRowBuilder.build());
}
}
@@ -380,33 +402,30 @@ private void queryOneRow(Statement statement, String sql, List columns,
* three - value3
* four - value4
*
- * @param statement statement
- * @param sql sql
+ * @param rowSet row set
* @param columns query metrics field list
* @throws Exception when error happen
*/
- private void queryOneRowByMatchTwoColumns(Statement statement, String sql, List columns,
+ private void queryOneRowByMatchTwoColumns(JdbcQueryRowSet rowSet, List columns,
CollectRep.MetricsData.Builder builder, long startTime) throws Exception {
- try (ResultSet resultSet = statement.executeQuery(sql)) {
- HashMap values = new HashMap<>(columns.size());
- while (resultSet.next()) {
- if (resultSet.getString(1) != null) {
- values.put(resultSet.getString(1).toLowerCase().trim(), resultSet.getString(2));
- }
+ HashMap values = new HashMap<>(columns.size());
+ while (rowSet.next()) {
+ if (rowSet.getString(1) != null) {
+ values.put(rowSet.getString(1).toLowerCase().trim(), rowSet.getString(2));
}
- CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
- for (String column : columns) {
- if (CollectorConstants.RESPONSE_TIME.equals(column)) {
- long time = System.currentTimeMillis() - startTime;
- valueRowBuilder.addColumn(String.valueOf(time));
- } else {
- String value = values.get(column.toLowerCase());
- value = value == null ? CommonConstants.NULL_VALUE : value;
- valueRowBuilder.addColumn(value);
- }
+ }
+ CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
+ for (String column : columns) {
+ if (CollectorConstants.RESPONSE_TIME.equals(column)) {
+ long time = System.currentTimeMillis() - startTime;
+ valueRowBuilder.addColumn(String.valueOf(time));
+ } else {
+ String value = values.get(column.toLowerCase());
+ value = value == null ? CommonConstants.NULL_VALUE : value;
+ valueRowBuilder.addColumn(value);
}
- builder.addValueRow(valueRowBuilder.build());
}
+ builder.addValueRow(valueRowBuilder.build());
}
/**
@@ -416,28 +435,45 @@ private void queryOneRowByMatchTwoColumns(Statement statement, String sql, List<
* query sql:select one, tow, three, four from book;
* and return multi row record mapping with the metrics
*
- * @param statement statement
- * @param sql sql
+ * @param rowSet row set
* @param columns query metrics field list
* @throws Exception when error happen
*/
- private void queryMultiRow(Statement statement, String sql, List columns,
+ private void queryMultiRow(JdbcQueryRowSet rowSet, List columns,
CollectRep.MetricsData.Builder builder, long startTime) throws Exception {
- try (ResultSet resultSet = statement.executeQuery(sql)) {
- while (resultSet.next()) {
- CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
- for (String column : columns) {
- if (CollectorConstants.RESPONSE_TIME.equals(column)) {
- long time = System.currentTimeMillis() - startTime;
- valueRowBuilder.addColumn(String.valueOf(time));
- } else {
- String value = resultSet.getString(column);
- value = value == null ? CommonConstants.NULL_VALUE : value;
- valueRowBuilder.addColumn(value);
- }
+ while (rowSet.next()) {
+ CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
+ for (String column : columns) {
+ if (CollectorConstants.RESPONSE_TIME.equals(column)) {
+ long time = System.currentTimeMillis() - startTime;
+ valueRowBuilder.addColumn(String.valueOf(time));
+ } else {
+ String value = rowSet.getString(column);
+ value = value == null ? CommonConstants.NULL_VALUE : value;
+ valueRowBuilder.addColumn(value);
}
- builder.addValueRow(valueRowBuilder.build());
}
+ builder.addValueRow(valueRowBuilder.build());
+ }
+ }
+
+ private void closeStatementAndConnection(Statement statement, boolean reuseConnection) {
+ if (statement == null) {
+ return;
+ }
+ Connection connection = null;
+ try {
+ connection = statement.getConnection();
+ statement.close();
+ } catch (Exception exception) {
+ log.error("Jdbc close statement error: {}", exception.getMessage());
+ }
+ try {
+ if (!reuseConnection && connection != null) {
+ connection.close();
+ }
+ } catch (Exception exception) {
+ log.error("Jdbc close connection error: {}", exception.getMessage());
}
}
@@ -548,4 +584,53 @@ private String constructDatabaseUrl(JdbcProtocol jdbcProtocol, String host, Stri
default -> throw new IllegalArgumentException("Not support database platform: " + jdbcProtocol.getPlatform());
};
}
+
+ private static final class ResultSetJdbcQueryRowSet implements JdbcQueryRowSet {
+
+ private final Statement statement;
+ private final java.sql.ResultSet resultSet;
+ private final boolean reuseConnection;
+
+ private ResultSetJdbcQueryRowSet(Statement statement, java.sql.ResultSet resultSet, boolean reuseConnection) {
+ this.statement = statement;
+ this.resultSet = resultSet;
+ this.reuseConnection = reuseConnection;
+ }
+
+ @Override
+ public boolean next() throws Exception {
+ return resultSet.next();
+ }
+
+ @Override
+ public String getString(String column) throws Exception {
+ return resultSet.getString(column);
+ }
+
+ @Override
+ public String getString(int index) throws Exception {
+ return resultSet.getString(index);
+ }
+
+ @Override
+ public void close() throws Exception {
+ Connection connection = null;
+ try {
+ connection = statement.getConnection();
+ } catch (Exception ignored) {
+ // ignore
+ }
+ try {
+ resultSet.close();
+ } finally {
+ try {
+ statement.close();
+ } finally {
+ if (!reuseConnection && connection != null) {
+ connection.close();
+ }
+ }
+ }
+ }
+ }
}
diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryExecutor.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryExecutor.java
new file mode 100644
index 00000000000..3d3c9076da3
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryExecutor.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.query;
+
+import org.apache.hertzbeat.common.entity.job.Metrics;
+
+/**
+ * Adapter point for replacing only the SQL query execution part of JdbcCommonCollect.
+ */
+public interface JdbcQueryExecutor {
+
+ boolean supports(Metrics metrics);
+
+ JdbcQueryRowSet executeQuery(Metrics metrics, int timeout, int maxRows) throws Exception;
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryExecutorRegistry.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryExecutorRegistry.java
new file mode 100644
index 00000000000..bb9f1c0baa6
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryExecutorRegistry.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.query;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+
+/**
+ * Static registry used by JdbcCommonCollect to discover optional query executors.
+ */
+public final class JdbcQueryExecutorRegistry {
+
+ private static final List EXECUTORS = new CopyOnWriteArrayList<>();
+
+ private JdbcQueryExecutorRegistry() {
+ }
+
+ public static void register(JdbcQueryExecutor executor) {
+ if (executor == null || EXECUTORS.contains(executor)) {
+ return;
+ }
+ EXECUTORS.add(executor);
+ }
+
+ public static void unregister(JdbcQueryExecutor executor) {
+ if (executor == null) {
+ return;
+ }
+ EXECUTORS.remove(executor);
+ }
+
+ public static Optional resolve(Metrics metrics) {
+ return EXECUTORS.stream()
+ .filter(executor -> executor.supports(metrics))
+ .findFirst();
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryRowSet.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryRowSet.java
new file mode 100644
index 00000000000..a93cef11d77
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/query/JdbcQueryRowSet.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.query;
+
+/**
+ * A minimal row cursor abstraction shared by JDBC and R2DBC-backed database queries.
+ */
+public interface JdbcQueryRowSet extends AutoCloseable {
+
+ boolean next() throws Exception;
+
+ String getString(String column) throws Exception;
+
+ String getString(int index) throws Exception;
+
+ @Override
+ void close() throws Exception;
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/pom.xml b/hertzbeat-collector/hertzbeat-collector-collector/pom.xml
index 8a108c99de0..5fcca5107d9 100644
--- a/hertzbeat-collector/hertzbeat-collector-collector/pom.xml
+++ b/hertzbeat-collector/hertzbeat-collector-collector/pom.xml
@@ -29,12 +29,18 @@
${project.artifactId}
- 17
- 17
+ ${java.version}
+ ${java.version}
UTF-8
+
+ org.apache.hertzbeat
+ hertzbeat-common-spring
+ ${hertzbeat.version}
+
+
org.apache.hertzbeat
@@ -42,6 +48,12 @@
${hertzbeat.version}
+
+ org.apache.hertzbeat
+ hertzbeat-collector-mysql-r2dbc
+ ${hertzbeat.version}
+
+
org.apache.hertzbeat
@@ -93,6 +105,23 @@
io.micrometer
micrometer-registry-prometheus
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+ com.mysql
+ mysql-connector-j
+ test
+
+
+ org.testcontainers
+ testcontainers
+ ${testcontainers.version}
+ test
+
@@ -142,8 +171,8 @@
org.apache.maven.plugins
maven-compiler-plugin
- 17
- 17
+ ${java.version}
+ ${java.version}
@@ -216,8 +245,8 @@
org.apache.maven.plugins
maven-compiler-plugin
- 17
- 17
+ ${java.version}
+ ${java.version}
@@ -344,5 +373,193 @@
+
+ native
+
+ hertzbeat-collector-collector
+
+ ../../script/assembly/collector/bin-native
+ tar.gz
+ unsupported
+ ${native.target.platform}-bin
+ apache-hertzbeat-collector-native-${hzb.version}-${native.target.platform}-bin
+ target/${native.image.name}${native.binary.extension}
+ ${project.build.finalName}${native.binary.extension}
+
+
+ apache-hertzbeat-collector-native-${hzb.version}
+
+
+ org.apache.maven.plugins
+ maven-compiler-plugin
+
+ full
+
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+
+ org.apache.maven.plugins
+ maven-enforcer-plugin
+ 3.5.0
+
+
+ require-supported-native-platform
+ validate
+
+ enforce
+
+
+
+
+ native.target.platform
+ ^(linux-amd64|linux-arm64|macos-amd64|macos-arm64|windows-amd64)$
+ Unsupported native collector platform. Build the native collector on a supported GitHub runner or matching local host: linux-amd64, linux-arm64, macos-amd64, macos-arm64, windows-amd64.
+
+
+
+
+
+
+
+ org.graalvm.buildtools
+ native-maven-plugin
+ 0.11.4
+
+
+ build-native-image
+ package
+
+ compile-no-fork
+
+
+
+
+ ${native.image.name}
+
+ -H:+UnlockExperimentalVMOptions
+ -H:-AddAllFileSystemProviders
+ -H:ServiceLoaderFeatureExcludeServiceProviders=org.apache.sshd.common.file.root.RootedFileSystemProvider,org.apache.sshd.sftp.client.fs.SftpFileSystemProvider
+ --initialize-at-build-time=net.i2p.crypto.eddsa.EdDSASecurityProvider,org.apache.arrow.memory.util.MemoryUtil
+ -J--add-opens=java.base/java.nio=ALL-UNNAMED
+ -J-Dorg.apache.sshd.security.registrars=org.apache.sshd.common.util.security.eddsa.EdDSASecurityProviderRegistrar
+
+
+
+
+ org.apache.maven.plugins
+ maven-assembly-plugin
+ ${maven-assembly-plugin.version}
+
+
+ native-bin
+ package
+
+ single
+
+
+
+ ../../script/assembly/collector/assembly-native.xml
+
+ ../../dist
+
+
+
+
+
+
+
+
+ native-platform-linux-amd64
+
+
+ Linux
+ amd64
+
+
+
+ linux-amd64
+
+
+
+ native-platform-linux-x86_64
+
+
+ Linux
+ x86_64
+
+
+
+ linux-amd64
+
+
+
+ native-platform-linux-arm64
+
+
+ Linux
+ aarch64
+
+
+
+ linux-arm64
+
+
+
+ native-platform-macos-amd64
+
+
+ mac
+ x86_64
+
+
+
+ macos-amd64
+
+
+
+ native-platform-macos-arm64
+
+
+ mac
+ aarch64
+
+
+
+ macos-arm64
+
+
+
+ native-platform-windows-amd64
+
+
+ windows
+ amd64
+
+
+
+ .exe
+ ../../script/assembly/collector/bin-native-win
+ zip
+ windows-amd64
+
+
+
+ native-platform-windows-x86_64
+
+
+ windows
+ x86_64
+
+
+
+ .exe
+ ../../script/assembly/collector/bin-native-win
+ zip
+ windows-amd64
+
+
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/Collector.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/Collector.java
index 3e41f228010..0eca3ce9e35 100644
--- a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/Collector.java
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/Collector.java
@@ -18,10 +18,13 @@
package org.apache.hertzbeat.collector;
import jakarta.annotation.PostConstruct;
+import org.apache.hertzbeat.collector.nativex.CollectorRuntimeHintsRegistrar;
+import org.apache.hertzbeat.collector.nativex.NativeCollectorDefaults;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.ImportRuntimeHints;
/**
* collector startup
@@ -29,13 +32,20 @@
@ComponentScan(basePackages = {"org.apache.hertzbeat"})
@ConfigurationPropertiesScan(basePackages = {"org.apache.hertzbeat"})
@SpringBootApplication
+@ImportRuntimeHints(CollectorRuntimeHintsRegistrar.class)
public class Collector {
public static void main(String[] args) {
- SpringApplication.run(Collector.class, args);
+ SpringApplication application = new SpringApplication(Collector.class);
+ NativeCollectorDefaults.applyTo(application);
+ application.run(args);
}
@PostConstruct
public void init() {
System.setProperty("jdk.jndi.object.factoriesFilter", "!com.zaxxer.hikari.HikariJNDIFactory");
+ if (System.getProperty("arrow.allocation.manager.type") == null
+ && System.getenv("ARROW_ALLOCATION_MANAGER_TYPE") == null) {
+ System.setProperty("arrow.allocation.manager.type", "Netty");
+ }
}
}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlCollectorProperties.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlCollectorProperties.java
new file mode 100644
index 00000000000..ac34a37ab84
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlCollectorProperties.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.mysql;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Collector-side MySQL query engine routing.
+ */
+@Getter
+@Setter
+@ConfigurationProperties(prefix = "hertzbeat.collector.mysql")
+public class MysqlCollectorProperties {
+
+ private QueryEngine queryEngine = QueryEngine.AUTO;
+
+ public QueryEngine resolveQueryEngine(boolean mysqlJdbcDriverAvailable) {
+ if (queryEngine == QueryEngine.AUTO) {
+ return mysqlJdbcDriverAvailable ? QueryEngine.JDBC : QueryEngine.R2DBC;
+ }
+ return queryEngine;
+ }
+
+ public boolean useR2dbc(boolean mysqlJdbcDriverAvailable) {
+ return resolveQueryEngine(mysqlJdbcDriverAvailable) == QueryEngine.R2DBC;
+ }
+
+ /**
+ * Supported collector-side query engines.
+ */
+ public enum QueryEngine {
+ AUTO,
+ JDBC,
+ R2DBC
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcDriverAvailability.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcDriverAvailability.java
new file mode 100644
index 00000000000..8c1466aedeb
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcDriverAvailability.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.mysql;
+
+import java.net.URL;
+import java.security.CodeSource;
+import java.util.Locale;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * Detects whether a MySQL JDBC driver is available from the external ext-lib path.
+ */
+@Component
+public class MysqlJdbcDriverAvailability {
+
+ private static final String[] MYSQL_DRIVER_CLASSES = {
+ "com.mysql.cj.jdbc.Driver",
+ "com.mysql.jdbc.Driver"
+ };
+
+ public boolean hasMysqlJdbcDriver() {
+ ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
+ for (String driverClass : MYSQL_DRIVER_CLASSES) {
+ if (!ClassUtils.isPresent(driverClass, classLoader)) {
+ continue;
+ }
+ try {
+ if (isExternalExtLibDriver(ClassUtils.forName(driverClass, classLoader))) {
+ return true;
+ }
+ } catch (ClassNotFoundException ignored) {
+ // Race-free enough for runtime detection: keep probing other known driver class names.
+ }
+ }
+ return false;
+ }
+
+ boolean isExternalExtLibDriver(Class> driverClass) {
+ String location = resolveLocation(driverClass);
+ return isExtLibLocation(location);
+ }
+
+ static boolean isExtLibLocation(String location) {
+ if (!StringUtils.hasText(location)) {
+ return false;
+ }
+ String normalized = location
+ .replace('\\', '/')
+ .toLowerCase(Locale.ROOT);
+ return normalized.contains("/ext-lib/");
+ }
+
+ private String resolveLocation(Class> driverClass) {
+ CodeSource codeSource = driverClass.getProtectionDomain().getCodeSource();
+ if (codeSource != null && codeSource.getLocation() != null) {
+ return codeSource.getLocation().toExternalForm();
+ }
+ String resourceName = ClassUtils.convertClassNameToResourcePath(driverClass.getName()) + ".class";
+ ClassLoader classLoader = driverClass.getClassLoader();
+ URL resource = classLoader != null ? classLoader.getResource(resourceName) : ClassLoader.getSystemResource(resourceName);
+ return resource != null ? resource.toExternalForm() : null;
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlR2dbcJdbcQueryExecutor.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlR2dbcJdbcQueryExecutor.java
new file mode 100644
index 00000000000..55837095c60
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlR2dbcJdbcQueryExecutor.java
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.mysql;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import org.apache.hertzbeat.collector.collect.common.ssh.SshTunnelHelper;
+import org.apache.hertzbeat.collector.collect.database.query.JdbcQueryExecutor;
+import org.apache.hertzbeat.collector.collect.database.query.JdbcQueryExecutorRegistry;
+import org.apache.hertzbeat.collector.collect.database.query.JdbcQueryRowSet;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlQueryExecutor;
+import org.apache.hertzbeat.collector.mysql.r2dbc.QueryOptions;
+import org.apache.hertzbeat.collector.mysql.r2dbc.QueryResult;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.job.SshTunnel;
+import org.apache.hertzbeat.common.entity.job.protocol.JdbcProtocol;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+
+/**
+ * MySQL-compatible query-only adapter that lets JdbcCommonCollect execute read-only queries through the built-in
+ * R2DBC path when no MySQL JDBC driver is present.
+ */
+@Component
+public class MysqlR2dbcJdbcQueryExecutor implements JdbcQueryExecutor, InitializingBean, DisposableBean {
+
+ private static final String QUERY_TYPE_ONE_ROW = "oneRow";
+ private static final String QUERY_TYPE_MULTI_ROW = "multiRow";
+ private static final String QUERY_TYPE_COLUMNS = "columns";
+
+ private final MysqlCollectorProperties properties;
+ private final MysqlQueryExecutor mysqlQueryExecutor;
+ private final MysqlJdbcDriverAvailability mysqlJdbcDriverAvailability;
+
+ public MysqlR2dbcJdbcQueryExecutor(MysqlCollectorProperties properties,
+ MysqlQueryExecutor mysqlQueryExecutor,
+ MysqlJdbcDriverAvailability mysqlJdbcDriverAvailability) {
+ this.properties = properties;
+ this.mysqlQueryExecutor = mysqlQueryExecutor;
+ this.mysqlJdbcDriverAvailability = mysqlJdbcDriverAvailability;
+ }
+
+ @Override
+ public boolean supports(Metrics metrics) {
+ if (metrics == null || metrics.getJdbc() == null) {
+ return false;
+ }
+ JdbcProtocol jdbcProtocol = metrics.getJdbc();
+ if (!isMysqlCompatiblePlatform(jdbcProtocol.getPlatform())) {
+ return false;
+ }
+ String queryType = jdbcProtocol.getQueryType();
+ return properties.useR2dbc(mysqlJdbcDriverAvailability.hasMysqlJdbcDriver())
+ && (QUERY_TYPE_ONE_ROW.equals(queryType)
+ || QUERY_TYPE_MULTI_ROW.equals(queryType)
+ || QUERY_TYPE_COLUMNS.equals(queryType));
+ }
+
+ private boolean isMysqlCompatiblePlatform(String platform) {
+ return "mysql".equalsIgnoreCase(platform) || "mariadb".equalsIgnoreCase(platform);
+ }
+
+ @Override
+ public JdbcQueryRowSet executeQuery(Metrics metrics, int timeout, int maxRows) {
+ JdbcProtocol jdbcProtocol = metrics.getJdbc();
+ QueryOptions options = buildQueryOptions(jdbcProtocol, timeout, maxRows);
+ QueryResult queryResult = mysqlQueryExecutor.execute(jdbcProtocol.getSql(), options);
+ if (queryResult.hasError()) {
+ throw new IllegalStateException("R2DBC MySQL query failed: " + queryResult.getError());
+ }
+ return new QueryResultRowSet(queryResult);
+ }
+
+ @Override
+ public void afterPropertiesSet() {
+ JdbcQueryExecutorRegistry.register(this);
+ }
+
+ @Override
+ public void destroy() {
+ JdbcQueryExecutorRegistry.unregister(this);
+ }
+
+ private QueryOptions buildQueryOptions(JdbcProtocol jdbcProtocol, int timeout, int maxRows) {
+ MysqlTarget target = resolveTarget(jdbcProtocol);
+ SshTunnel sshTunnel = jdbcProtocol.getSshTunnel();
+ String host = target.host();
+ int port = target.port();
+ if (sshTunnel != null && Boolean.parseBoolean(sshTunnel.getEnable())) {
+ try {
+ int localPort = SshTunnelHelper.localPortForward(sshTunnel, host, String.valueOf(port));
+ host = "127.0.0.1";
+ port = localPort;
+ } catch (Exception exception) {
+ throw new IllegalStateException("R2DBC MySQL query adapter failed to establish SSH tunnel", exception);
+ }
+ }
+ return QueryOptions.builder()
+ .host(host)
+ .port(port)
+ .username(jdbcProtocol.getUsername())
+ .password(jdbcProtocol.getPassword())
+ .database(target.database())
+ .schema(target.database())
+ .timeout(Duration.ofMillis(timeout))
+ .maxRows(maxRows)
+ .fetchSize(256)
+ .readOnly(true)
+ .build();
+ }
+
+ private MysqlTarget resolveTarget(JdbcProtocol jdbcProtocol) {
+ if (StringUtils.hasText(jdbcProtocol.getUrl())) {
+ return parseJdbcUrl(jdbcProtocol.getUrl(), jdbcProtocol.getDatabase());
+ }
+ if (!StringUtils.hasText(jdbcProtocol.getHost()) || !StringUtils.hasText(jdbcProtocol.getPort())) {
+ throw new IllegalArgumentException("R2DBC MySQL query adapter requires host/port or a jdbc:mysql URL");
+ }
+ return new MysqlTarget(jdbcProtocol.getHost(), Integer.parseInt(jdbcProtocol.getPort()), jdbcProtocol.getDatabase());
+ }
+
+ private MysqlTarget parseJdbcUrl(String url, String fallbackDatabase) {
+ String trimmed = url.trim();
+ if (!(trimmed.startsWith("jdbc:mysql://") || trimmed.startsWith("jdbc:mariadb://"))) {
+ throw new IllegalArgumentException("R2DBC MySQL query adapter only supports jdbc:mysql:// or jdbc:mariadb:// URLs");
+ }
+ URI uri = URI.create(trimmed.substring("jdbc:".length()));
+ String host = uri.getHost();
+ int port = uri.getPort() > 0 ? uri.getPort() : 3306;
+ if (!StringUtils.hasText(host)) {
+ throw new IllegalArgumentException("R2DBC MySQL query adapter URL must include a host");
+ }
+ String path = uri.getPath();
+ String database = StringUtils.hasText(path) && path.length() > 1 ? path.substring(1) : fallbackDatabase;
+ return new MysqlTarget(host, port, database);
+ }
+
+ private record MysqlTarget(String host, int port, String database) {
+ }
+
+ private static final class QueryResultRowSet implements JdbcQueryRowSet {
+
+ private final List> rows;
+ private final Map columnIndexMap;
+ private int currentIndex = -1;
+
+ private QueryResultRowSet(QueryResult queryResult) {
+ this.rows = queryResult.getRows();
+ this.columnIndexMap = buildColumnIndexMap(queryResult.getColumns());
+ }
+
+ @Override
+ public boolean next() {
+ currentIndex++;
+ return currentIndex < rows.size();
+ }
+
+ @Override
+ public String getString(String column) {
+ Integer index = columnIndexMap.get(column.toLowerCase(Locale.ROOT));
+ if (index == null) {
+ throw new IllegalArgumentException("Column not found in R2DBC MySQL result: " + column);
+ }
+ return getString(index + 1);
+ }
+
+ @Override
+ public String getString(int index) {
+ if (currentIndex < 0 || currentIndex >= rows.size()) {
+ throw new IllegalStateException("R2DBC MySQL result cursor is not positioned on a row");
+ }
+ int zeroBased = index - 1;
+ List row = rows.get(currentIndex);
+ if (zeroBased < 0 || zeroBased >= row.size()) {
+ throw new IllegalArgumentException("Column index out of bounds in R2DBC MySQL result: " + index);
+ }
+ return row.get(zeroBased);
+ }
+
+ @Override
+ public void close() {
+ // QueryResult is fully materialized, so there is nothing left to close here.
+ }
+
+ private static Map buildColumnIndexMap(List columns) {
+ Map indexMap = new HashMap<>(columns.size());
+ for (int index = 0; index < columns.size(); index++) {
+ indexMap.put(columns.get(index).toLowerCase(Locale.ROOT), index);
+ }
+ return indexMap;
+ }
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/strategy/CollectStrategyFactory.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/strategy/CollectStrategyFactory.java
index 4c6517ab905..28b9d2db88e 100644
--- a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/strategy/CollectStrategyFactory.java
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/collect/strategy/CollectStrategyFactory.java
@@ -20,6 +20,7 @@
import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;
+import lombok.extern.slf4j.Slf4j;
import org.apache.hertzbeat.collector.collect.AbstractCollect;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;
@@ -29,6 +30,7 @@
/**
* Specific metrics collection factory
*/
+@Slf4j
@Configuration
@Order(value = Ordered.HIGHEST_PRECEDENCE + 1)
public class CollectStrategyFactory implements CommandLineRunner {
@@ -49,10 +51,18 @@ public static AbstractCollect invoke(String protocol) {
@Override
public void run(String... args) throws Exception {
+ COLLECT_STRATEGY.clear();
// spi load and registry protocol and collect instance
ServiceLoader loader = ServiceLoader.load(AbstractCollect.class, AbstractCollect.class.getClassLoader());
for (AbstractCollect collect : loader) {
COLLECT_STRATEGY.put(collect.supportProtocol(), collect);
}
+ if (COLLECT_STRATEGY.isEmpty()) {
+ throw new IllegalStateException(
+ "No collect strategies were registered. "
+ + "Verify META-INF/services/org.apache.hertzbeat.collector.collect.AbstractCollect "
+ + "is present on the runtime classpath.");
+ }
+ log.info("Registered {} collect strategies: {}", COLLECT_STRATEGY.size(), COLLECT_STRATEGY.keySet());
}
}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/CollectorRuntimeHintsRegistrar.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/CollectorRuntimeHintsRegistrar.java
new file mode 100644
index 00000000000..3464caa1b1b
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/CollectorRuntimeHintsRegistrar.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.nativex;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.arrow.memory.netty.NettyAllocationManager;
+import org.apache.arrow.vector.types.DateUnit;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.IntervalUnit;
+import org.apache.arrow.vector.types.MetadataVersion;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.UnionMode;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.hertzbeat.common.entity.dto.ServerInfo;
+import org.springframework.aot.hint.BindingReflectionHintsRegistrar;
+import org.springframework.aot.hint.MemberCategory;
+import org.springframework.aot.hint.RuntimeHints;
+import org.springframework.aot.hint.RuntimeHintsRegistrar;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
+import org.springframework.core.type.filter.TypeFilter;
+import org.springframework.lang.NonNull;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Registers native binding hints for collector-side message payloads.
+ */
+@Slf4j
+public class CollectorRuntimeHintsRegistrar implements RuntimeHintsRegistrar {
+
+ private static final String JOB_PACKAGE = "org.apache.hertzbeat.common.entity.job";
+ private static final String JOB_PROTOCOL_PACKAGE = "org.apache.hertzbeat.common.entity.job.protocol";
+
+ @Override
+ public void registerHints(@NonNull RuntimeHints hints, ClassLoader classLoader) {
+ BindingReflectionHintsRegistrar bindingRegistrar = new BindingReflectionHintsRegistrar();
+ registerType(bindingRegistrar, hints, ServerInfo.class);
+ scanBindingPackage(classLoader, bindingRegistrar, hints, JOB_PACKAGE);
+ scanBindingPackage(classLoader, bindingRegistrar, hints, JOB_PROTOCOL_PACKAGE);
+ hints.reflection().registerType(NettyAllocationManager.class, MemberCategory.DECLARED_FIELDS);
+ registerType(bindingRegistrar, hints, Schema.class);
+ registerType(bindingRegistrar, hints, Field.class);
+ registerType(bindingRegistrar, hints, FieldType.class);
+ registerType(bindingRegistrar, hints, DictionaryEncoding.class);
+ registerType(bindingRegistrar, hints, DateUnit.class);
+ registerType(bindingRegistrar, hints, FloatingPointPrecision.class);
+ registerType(bindingRegistrar, hints, IntervalUnit.class);
+ registerType(bindingRegistrar, hints, MetadataVersion.class);
+ registerType(bindingRegistrar, hints, TimeUnit.class);
+ registerType(bindingRegistrar, hints, UnionMode.class);
+ for (Class> nestedClass : ArrowType.class.getDeclaredClasses()) {
+ if (!nestedClass.isAnnotation() && !nestedClass.isInterface()) {
+ registerType(bindingRegistrar, hints, nestedClass);
+ }
+ }
+ }
+
+ private void scanBindingPackage(ClassLoader classLoader, BindingReflectionHintsRegistrar bindingRegistrar,
+ RuntimeHints hints, String basePackage) {
+ for (Class> clazz : findBindingTypes(basePackage, classLoader)) {
+ registerType(bindingRegistrar, hints, clazz);
+ }
+ }
+
+ private void registerType(BindingReflectionHintsRegistrar bindingRegistrar, RuntimeHints hints, Class> clazz) {
+ bindingRegistrar.registerReflectionHints(hints.reflection(), clazz);
+ }
+
+ private Set> findBindingTypes(String basePackage, ClassLoader classLoader) {
+ Set> bindingTypes = new LinkedHashSet<>();
+ ClassPathScanningCandidateComponentProvider scanner = new ClassPathScanningCandidateComponentProvider(false);
+ TypeFilter includeAll = (metadataReader, metadataReaderFactory) -> true;
+ scanner.addIncludeFilter(includeAll);
+ for (BeanDefinition candidate : scanner.findCandidateComponents(basePackage)) {
+ String className = candidate.getBeanClassName();
+ if (className == null) {
+ continue;
+ }
+ try {
+ Class> clazz = ClassUtils.forName(className, classLoader);
+ if (!clazz.isAnnotation() && !clazz.isInterface()) {
+ bindingTypes.add(clazz);
+ }
+ } catch (Throwable ex) {
+ log.debug("Skip native binding hint registration for {}", className, ex);
+ }
+ }
+ return bindingTypes;
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/NativeCollectorDefaults.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/NativeCollectorDefaults.java
new file mode 100644
index 00000000000..1926c48cfb8
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/NativeCollectorDefaults.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.nativex;
+
+import java.util.Map;
+import org.springframework.boot.SpringApplication;
+import org.springframework.core.NativeDetector;
+
+/**
+ * Applies collector defaults without forking {@code application.yml}.
+ */
+public final class NativeCollectorDefaults {
+
+ static final String AUTOCONFIGURE_EXCLUDE_PROPERTY = "spring.autoconfigure.exclude";
+ static final String JVM_AUTOCONFIGURE_EXCLUDES = String.join(",",
+ "org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration",
+ "org.springframework.boot.autoconfigure.data.mongo.MongoDataAutoConfiguration",
+ "org.springframework.boot.jdbc.autoconfigure.DataSourceAutoConfiguration",
+ "org.springframework.boot.hibernate.autoconfigure.HibernateJpaAutoConfiguration");
+ static final String NATIVE_AUTOCONFIGURE_EXCLUDES = String.join(",",
+ JVM_AUTOCONFIGURE_EXCLUDES,
+ "org.springframework.boot.data.jpa.autoconfigure.DataJpaRepositoriesAutoConfiguration",
+ "org.springframework.boot.jdbc.autoconfigure.DataSourceInitializationAutoConfiguration",
+ "org.springframework.boot.jdbc.autoconfigure.DataSourceTransactionManagerAutoConfiguration",
+ "org.springframework.boot.jdbc.autoconfigure.health.DataSourceHealthContributorAutoConfiguration",
+ "org.springframework.boot.jdbc.autoconfigure.metrics.DataSourcePoolMetricsAutoConfiguration",
+ "org.springframework.boot.tomcat.autoconfigure.metrics.TomcatMetricsAutoConfiguration");
+
+ private NativeCollectorDefaults() {
+ }
+
+ public static void applyTo(SpringApplication application) {
+ application.setDefaultProperties(defaultProperties(NativeDetector.inNativeImage()));
+ }
+
+ static Map defaultProperties(boolean nativeImage) {
+ return Map.of(
+ AUTOCONFIGURE_EXCLUDE_PROPERTY,
+ nativeImage ? NATIVE_AUTOCONFIGURE_EXCLUDES : JVM_AUTOCONFIGURE_EXCLUDES);
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/META-INF/spring.factories b/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/META-INF/spring.factories
new file mode 100644
index 00000000000..731da64e615
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,17 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Intentionally left blank.
+# Native collector defaults are applied from Collector.main.
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml b/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml
index 9615e547b94..03702ef5627 100644
--- a/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml
@@ -27,9 +27,6 @@ spring:
timeout-per-shutdown-phase: 10s
jackson:
default-property-inclusion: ALWAYS
- # need to disable spring boot mongodb auto config, or default mongodb connection tried and failed...
- autoconfigure:
- exclude: org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration, org.springframework.boot.autoconfigure.data.mongo.MongoDataAutoConfiguration, org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration, org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration
management:
endpoints:
web:
@@ -77,6 +74,13 @@ common:
type: netty
hertzbeat:
+ collector:
+ mysql:
+ # MySQL-compatible query engine routing for MySQL, MariaDB, OceanBase, and TiDB SQL metrics.
+ # auto : prefer JDBC only when mysql-connector-j is available from ext-lib, otherwise use the built-in query engine
+ # jdbc : always use JDBC
+ # r2dbc : always use the built-in query engine
+ query-engine: ${HERTZBEAT_COLLECTOR_MYSQL_QUERY_ENGINE:auto}
# Optional virtual-thread overrides. Remove this whole block to use built-in defaults.
vthreads:
enabled: true
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/native/resources/META-INF/native-image/org.apache.hertzbeat/hertzbeat-collector-collector/reflect-config.json b/hertzbeat-collector/hertzbeat-collector-collector/src/native/resources/META-INF/native-image/org.apache.hertzbeat/hertzbeat-collector-collector/reflect-config.json
new file mode 100644
index 00000000000..227a978fe57
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/native/resources/META-INF/native-image/org.apache.hertzbeat/hertzbeat-collector-collector/reflect-config.json
@@ -0,0 +1,54 @@
+[
+ {
+ "name": "org.apache.hertzbeat.collector.Collector__ApplicationContextInitializer",
+ "allDeclaredConstructors": true,
+ "allDeclaredMethods": true
+ },
+ {
+ "name": "org.apache.hertzbeat.collector.Collector__BeanFactoryRegistrations",
+ "allDeclaredConstructors": true,
+ "allDeclaredMethods": true
+ },
+ {
+ "name": "org.apache.hertzbeat.common.entity.dto.ServerInfo",
+ "allDeclaredConstructors": true,
+ "allDeclaredFields": true,
+ "allDeclaredMethods": true
+ },
+ {
+ "name": "io.netty.channel.kqueue.KQueueDatagramChannel",
+ "allPublicConstructors": true
+ },
+ {
+ "name": "io.netty.channel.kqueue.KQueueSocketChannel",
+ "allPublicConstructors": true
+ },
+ {
+ "name": "io.netty.channel.kqueue.KQueueEventLoopGroup",
+ "allPublicConstructors": true
+ },
+ {
+ "name": "io.netty.channel.epoll.EpollDatagramChannel",
+ "allPublicConstructors": true
+ },
+ {
+ "name": "io.netty.channel.epoll.EpollSocketChannel",
+ "allPublicConstructors": true
+ },
+ {
+ "name": "io.netty.channel.epoll.EpollEventLoopGroup",
+ "allPublicConstructors": true
+ },
+ {
+ "name": "io.netty.channel.socket.nio.NioDatagramChannel",
+ "allPublicConstructors": true
+ },
+ {
+ "name": "io.netty.channel.socket.nio.NioSocketChannel",
+ "allPublicConstructors": true
+ },
+ {
+ "name": "io.netty.channel.nio.NioEventLoopGroup",
+ "allPublicConstructors": true
+ }
+]
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/native/resources/META-INF/services/org.apache.hertzbeat.collector.collect.AbstractCollect b/hertzbeat-collector/hertzbeat-collector-collector/src/native/resources/META-INF/services/org.apache.hertzbeat.collector.collect.AbstractCollect
new file mode 100644
index 00000000000..81ed2990010
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/native/resources/META-INF/services/org.apache.hertzbeat.collector.collect.AbstractCollect
@@ -0,0 +1,37 @@
+org.apache.hertzbeat.collector.collect.http.HttpCollectImpl
+org.apache.hertzbeat.collector.collect.http.SslCertificateCollectImpl
+org.apache.hertzbeat.collector.collect.database.JdbcCommonCollect
+org.apache.hertzbeat.collector.collect.icmp.IcmpCollectImpl
+org.apache.hertzbeat.collector.collect.jmx.JmxCollectImpl
+org.apache.hertzbeat.collector.collect.redis.RedisCommonCollectImpl
+org.apache.hertzbeat.collector.collect.mongodb.MongodbSingleCollectImpl
+org.apache.hertzbeat.collector.collect.rocketmq.RocketmqSingleCollectImpl
+org.apache.hertzbeat.collector.collect.snmp.SnmpCollectImpl
+org.apache.hertzbeat.collector.collect.ssh.SshCollectImpl
+org.apache.hertzbeat.collector.collect.telnet.TelnetCollectImpl
+org.apache.hertzbeat.collector.collect.smtp.SmtpCollectImpl
+org.apache.hertzbeat.collector.collect.ntp.NtpCollectImpl
+org.apache.hertzbeat.collector.collect.websocket.WebsocketCollectImpl
+org.apache.hertzbeat.collector.collect.ftp.FtpCollectImpl
+org.apache.hertzbeat.collector.collect.udp.UdpCollectImpl
+org.apache.hertzbeat.collector.collect.push.PushCollectImpl
+org.apache.hertzbeat.collector.collect.dns.DnsCollectImpl
+org.apache.hertzbeat.collector.collect.nginx.NginxCollectImpl
+org.apache.hertzbeat.collector.collect.memcached.MemcachedCollectImpl
+org.apache.hertzbeat.collector.collect.nebulagraph.NebulaGraphCollectImpl
+org.apache.hertzbeat.collector.collect.pop3.Pop3CollectImpl
+org.apache.hertzbeat.collector.collect.registry.RegistryImpl
+org.apache.hertzbeat.collector.collect.redfish.RedfishCollectImpl
+org.apache.hertzbeat.collector.collect.nebulagraph.NgqlCollectImpl
+org.apache.hertzbeat.collector.collect.imap.ImapCollectImpl
+org.apache.hertzbeat.collector.collect.script.ScriptCollectImpl
+org.apache.hertzbeat.collector.collect.mqtt.MqttCollectImpl
+org.apache.hertzbeat.collector.collect.ipmi2.IpmiCollectImpl
+org.apache.hertzbeat.collector.collect.kafka.KafkaCollectImpl
+org.apache.hertzbeat.collector.collect.sd.HttpSdCollectImpl
+org.apache.hertzbeat.collector.collect.sd.NacosSdCollectImpl
+org.apache.hertzbeat.collector.collect.sd.DnsSdCollectImpl
+org.apache.hertzbeat.collector.collect.sd.EurekaSdCollectImpl
+org.apache.hertzbeat.collector.collect.sd.ConsulSdCollectImpl
+org.apache.hertzbeat.collector.collect.modbus.ModbusCollectImpl
+org.apache.hertzbeat.collector.collect.s7.S7CollectImpl
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MariadbJdbcQueryAdapterTemplateIntegrationTest.java b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MariadbJdbcQueryAdapterTemplateIntegrationTest.java
new file mode 100644
index 00000000000..1ac654969d0
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MariadbJdbcQueryAdapterTemplateIntegrationTest.java
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.mysql;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hertzbeat.collector.collect.strategy.CollectStrategyFactory;
+import org.apache.hertzbeat.collector.dispatch.CollectDataDispatch;
+import org.apache.hertzbeat.collector.dispatch.MetricsCollect;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlR2dbcConnectionFactoryProvider;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlR2dbcQueryExecutor;
+import org.apache.hertzbeat.collector.mysql.r2dbc.ResultSetMapper;
+import org.apache.hertzbeat.collector.mysql.r2dbc.SqlGuard;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.job.protocol.JdbcProtocol;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.apache.hertzbeat.common.util.JsonUtil;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.DynamicTest;
+import org.junit.jupiter.api.TestFactory;
+import org.junit.jupiter.api.TestInstance;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import org.yaml.snakeyaml.Yaml;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class MariadbJdbcQueryAdapterTemplateIntegrationTest {
+
+ private static final String TEST_DATABASE = "hzb";
+ private static final String TEST_USERNAME = "test";
+ private static final String TEST_PASSWORD = "test123";
+ private static final String ROOT_PASSWORD = "root123";
+
+ private GenericContainer> container;
+ private MysqlR2dbcJdbcQueryExecutor jdbcQueryExecutor;
+ private List mariadbTemplateMetrics;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ Assumptions.assumeTrue(DockerClientFactory.instance().isDockerAvailable(), "Docker is required for integration tests");
+ new CollectStrategyFactory().run();
+ container = new GenericContainer<>(DockerImageName.parse("mariadb:11.4"))
+ .withExposedPorts(3306)
+ .withEnv("MARIADB_DATABASE", TEST_DATABASE)
+ .withEnv("MARIADB_USER", TEST_USERNAME)
+ .withEnv("MARIADB_PASSWORD", TEST_PASSWORD)
+ .withEnv("MARIADB_ROOT_PASSWORD", ROOT_PASSWORD)
+ .waitingFor(Wait.forListeningPort());
+ container.start();
+ awaitTcpLoginReady(container, TEST_USERNAME, TEST_PASSWORD, TEST_DATABASE);
+ initMonitoringData(container);
+
+ MysqlCollectorProperties properties = new MysqlCollectorProperties();
+ properties.setQueryEngine(MysqlCollectorProperties.QueryEngine.R2DBC);
+ jdbcQueryExecutor = new MysqlR2dbcJdbcQueryExecutor(
+ properties,
+ new MysqlR2dbcQueryExecutor(
+ new MysqlR2dbcConnectionFactoryProvider(),
+ new ResultSetMapper(),
+ new SqlGuard()),
+ new MysqlJdbcDriverAvailability());
+ jdbcQueryExecutor.afterPropertiesSet();
+ mariadbTemplateMetrics = loadMariadbTemplate().getMetrics();
+ }
+
+ @AfterAll
+ void tearDown() throws Exception {
+ if (jdbcQueryExecutor != null) {
+ jdbcQueryExecutor.destroy();
+ }
+ if (container != null) {
+ container.stop();
+ }
+ }
+
+ @TestFactory
+ Stream shouldCollectOfficialMariadbTemplateThroughJdbcQueryAdapter() {
+ return mariadbTemplateMetrics.stream()
+ .map(templateMetric -> DynamicTest.dynamicTest(templateMetric.getName(),
+ () -> verifyTemplateMetric(templateMetric)));
+ }
+
+ private void verifyTemplateMetric(Metrics templateMetric) throws Exception {
+ Metrics metric = materializeMetric(templateMetric);
+ if ("process_state".equals(metric.getName())) {
+ startBackgroundSleepQuery(container);
+ }
+ if ("slow_sql".equals(metric.getName())) {
+ generateSlowQuery(container);
+ }
+ CollectRep.MetricsData metricsData = collect(metric);
+ assertEquals(CollectRep.Code.SUCCESS, metricsData.getCode(),
+ () -> metric.getName() + " failed: " + metricsData.getMsg());
+ assertEquals(metric.getFields().size(), metricsData.getFieldsCount(),
+ () -> metric.getName() + " fields should still be produced by the original parser");
+ if ("columns".equals(metric.getJdbc().getQueryType())) {
+ assertEquals(1, metricsData.getValuesCount(), () -> metric.getName() + " should keep the original single-row shape");
+ }
+ if ("basic".equals(metric.getName())) {
+ assertTrue(metricsData.getValuesCount() > 0);
+ assertNotNull(metricsData.getValues().getFirst().getColumns(0));
+ assertTrue(!Objects.equals(CommonConstants.NULL_VALUE, metricsData.getValues().getFirst().getColumns(0)),
+ "basic.version should be collected through the adapted query path");
+ }
+ if ("process_state".equals(metric.getName()) || "slow_sql".equals(metric.getName())) {
+ assertTrue(metricsData.getValuesCount() > 0, () -> metric.getName() + " should return at least one row");
+ }
+ }
+
+ private CollectRep.MetricsData collect(Metrics metric) {
+ Job job = Job.builder()
+ .monitorId(1L)
+ .tenantId(1L)
+ .app("mariadb")
+ .defaultInterval(600L)
+ .metadata(new HashMap<>(0))
+ .labels(new HashMap<>(0))
+ .annotations(new HashMap<>(0))
+ .configmap(new ArrayList<>(0))
+ .metrics(new ArrayList<>(List.of(metric)))
+ .build();
+ WheelTimerTask timerTask = new WheelTimerTask(job, timeout -> {
+ });
+ CapturingCollectDataDispatch collectDataDispatch = new CapturingCollectDataDispatch();
+ MetricsCollect metricsCollect = new MetricsCollect(
+ metric,
+ new StubTimeout(timerTask),
+ collectDataDispatch,
+ "collector-test",
+ List.of());
+ metricsCollect.run();
+ assertNotNull(collectDataDispatch.metricsData, metric.getName() + " should dispatch metrics data");
+ return collectDataDispatch.metricsData;
+ }
+
+ private Metrics materializeMetric(Metrics templateMetric) {
+ Metrics metric = JsonUtil.fromJson(JsonUtil.toJson(templateMetric), Metrics.class);
+ JdbcProtocol jdbcProtocol = metric.getJdbc();
+ jdbcProtocol.setHost(container.getHost());
+ jdbcProtocol.setPort(String.valueOf(container.getMappedPort(3306)));
+ jdbcProtocol.setUsername(TEST_USERNAME);
+ jdbcProtocol.setPassword(TEST_PASSWORD);
+ jdbcProtocol.setTimeout(String.valueOf(Duration.ofSeconds(8).toMillis()));
+ jdbcProtocol.setReuseConnection("false");
+ jdbcProtocol.setUrl(null);
+ jdbcProtocol.setSshTunnel(null);
+ if (jdbcProtocol.getDatabase() == null || jdbcProtocol.getDatabase().contains("^_^")) {
+ jdbcProtocol.setDatabase(TEST_DATABASE);
+ }
+ if (metric.getAliasFields() == null || metric.getAliasFields().isEmpty()) {
+ metric.setAliasFields(metric.getFields().stream().map(Metrics.Field::getField).collect(Collectors.toList()));
+ }
+ return metric;
+ }
+
+ private Job loadMariadbTemplate() throws IOException {
+ Path template = Path.of("..", "..", "hertzbeat-manager", "src", "main", "resources", "define", "app-mariadb.yml")
+ .toAbsolutePath()
+ .normalize();
+ Yaml yaml = new Yaml();
+ try (Reader reader = Files.newBufferedReader(template)) {
+ return yaml.loadAs(reader, Job.class);
+ }
+ }
+
+ private void initMonitoringData(GenericContainer> mariaDb) throws Exception {
+ execRoot(mariaDb,
+ "GRANT SELECT ON mysql.* TO '" + TEST_USERNAME + "'@'%';"
+ + " GRANT PROCESS ON *.* TO '" + TEST_USERNAME + "'@'%';"
+ + " SET GLOBAL log_output='TABLE';"
+ + " SET GLOBAL slow_query_log='ON';"
+ + " SET GLOBAL long_query_time=0;"
+ + " FLUSH PRIVILEGES;");
+ generateSlowQuery(mariaDb);
+ }
+
+ private void generateSlowQuery(GenericContainer> mariaDb) throws Exception {
+ execUser(mariaDb, TEST_DATABASE, "SELECT SLEEP(0.2);");
+ Thread.sleep(300);
+ }
+
+ private void startBackgroundSleepQuery(GenericContainer> mariaDb) throws Exception {
+ String command = String.join(" ",
+ "CLIENT=$(command -v mysql || command -v mariadb)",
+ "&&",
+ "nohup sh -lc",
+ "'$CLIENT --protocol=TCP -h127.0.0.1 -P3306",
+ "-u" + TEST_USERNAME,
+ "-p" + TEST_PASSWORD,
+ TEST_DATABASE,
+ "-e",
+ "\"SELECT SLEEP(15)\" >/tmp/process-state.log 2>&1'",
+ ">/dev/null 2>&1 &");
+ mariaDb.execInContainer("sh", "-lc", command);
+ Thread.sleep(500);
+ }
+
+ private void awaitTcpLoginReady(GenericContainer> mariaDb, String username, String password, String database) throws Exception {
+ long deadline = System.currentTimeMillis() + Duration.ofSeconds(30).toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ var result = mariaDb.execInContainer("sh", "-lc", mysqlCliCommand(username, password, database, "SELECT 1"));
+ if (result.getExitCode() == 0) {
+ return;
+ }
+ } catch (Exception ignored) {
+ // Wait for the MariaDB entrypoint to finish bootstrapping and switch to the final TCP listener.
+ }
+ Thread.sleep(1000);
+ }
+ throw new IllegalStateException("Timed out waiting for MariaDB TCP login to become ready");
+ }
+
+ private void execRoot(GenericContainer> mariaDb, String sql) throws Exception {
+ var result = mariaDb.execInContainer("sh", "-lc", mysqlCliCommand("root", ROOT_PASSWORD, "mysql", sql));
+ if (result.getExitCode() != 0) {
+ throw new IllegalStateException("root mysql command failed: " + result.getStderr());
+ }
+ }
+
+ private void execUser(GenericContainer> mariaDb, String database, String sql) throws Exception {
+ var result = mariaDb.execInContainer("sh", "-lc", mysqlCliCommand(TEST_USERNAME, TEST_PASSWORD, database, sql));
+ if (result.getExitCode() != 0) {
+ throw new IllegalStateException("user mysql command failed: " + result.getStderr());
+ }
+ }
+
+ private String mysqlCliCommand(String username, String password, String database, String sql) {
+ return String.join(" ",
+ "CLIENT=$(command -v mysql || command -v mariadb)",
+ "&&",
+ "$CLIENT --protocol=TCP -h127.0.0.1 -P3306",
+ "-u" + username,
+ "-p" + password,
+ database,
+ "-e",
+ "\"" + sql.replace("\"", "\\\"") + "\"");
+ }
+
+ private static final class CapturingCollectDataDispatch implements CollectDataDispatch {
+
+ private CollectRep.MetricsData metricsData;
+
+ @Override
+ public void dispatchCollectData(Timeout timeout, Metrics metrics, CollectRep.MetricsData metricsData) {
+ this.metricsData = metricsData;
+ }
+
+ @Override
+ public void dispatchCollectData(Timeout timeout, Metrics metrics, List metricsDataList) {
+ if (metricsDataList != null && !metricsDataList.isEmpty()) {
+ this.metricsData = metricsDataList.getFirst();
+ }
+ }
+ }
+
+ private record StubTimeout(WheelTimerTask wheelTimerTask) implements Timeout {
+
+ @Override
+ public org.apache.hertzbeat.common.timer.Timer timer() {
+ return null;
+ }
+
+ @Override
+ public org.apache.hertzbeat.common.timer.TimerTask task() {
+ return wheelTimerTask;
+ }
+
+ @Override
+ public boolean isExpired() {
+ return false;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public boolean cancel() {
+ return false;
+ }
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcDriverAvailabilityTest.java b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcDriverAvailabilityTest.java
new file mode 100644
index 00000000000..9c5cabe8428
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcDriverAvailabilityTest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.mysql;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+class MysqlJdbcDriverAvailabilityTest {
+
+ @Test
+ void shouldTreatOnlyExtLibLocationsAsAutoJdbcSignal() {
+ assertTrue(MysqlJdbcDriverAvailability.isExtLibLocation("/opt/hertzbeat/ext-lib/mysql-connector-j-9.0.0.jar"));
+ assertTrue(MysqlJdbcDriverAvailability.isExtLibLocation("file:/C:/hertzbeat/ext-lib/mysql-connector-j-9.0.0.jar"));
+ assertFalse(MysqlJdbcDriverAvailability.isExtLibLocation("/Users/dev/.m2/repository/com/mysql/mysql-connector-j/9.0.0/mysql-connector-j-9.0.0.jar"));
+ assertFalse(MysqlJdbcDriverAvailability.isExtLibLocation(null));
+ }
+
+ @Test
+ void shouldIgnoreTestClasspathMysqlDriverWhenItIsNotFromExtLib() {
+ MysqlJdbcDriverAvailability availability = new MysqlJdbcDriverAvailability();
+
+ assertFalse(availability.hasMysqlJdbcDriver());
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryAdapterCompatibilityIntegrationTest.java b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryAdapterCompatibilityIntegrationTest.java
new file mode 100644
index 00000000000..38d7265ddb8
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryAdapterCompatibilityIntegrationTest.java
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.mysql;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hertzbeat.collector.collect.strategy.CollectStrategyFactory;
+import org.apache.hertzbeat.collector.dispatch.CollectDataDispatch;
+import org.apache.hertzbeat.collector.dispatch.MetricsCollect;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlR2dbcConnectionFactoryProvider;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlR2dbcQueryExecutor;
+import org.apache.hertzbeat.collector.mysql.r2dbc.ResultSetMapper;
+import org.apache.hertzbeat.collector.mysql.r2dbc.SqlGuard;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.job.protocol.JdbcProtocol;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.apache.hertzbeat.common.util.JsonUtil;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.DynamicTest;
+import org.junit.jupiter.api.TestFactory;
+import org.junit.jupiter.api.TestInstance;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import org.yaml.snakeyaml.Yaml;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class MysqlJdbcQueryAdapterCompatibilityIntegrationTest {
+
+ private static final String TEST_DATABASE = "hzb";
+ private static final String TEST_USERNAME = "test";
+ private static final String TEST_PASSWORD = "test123";
+ private static final String ROOT_PASSWORD = "root123";
+ private static final Set REPRESENTATIVE_TEMPLATE_METRICS = Set.of("basic", "process_state");
+
+ private List representativeTemplateMetrics;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ Assumptions.assumeTrue(DockerClientFactory.instance().isDockerAvailable(), "Docker is required for integration tests");
+ new CollectStrategyFactory().run();
+ representativeTemplateMetrics = loadMysqlTemplate().getMetrics().stream()
+ .filter(metric -> REPRESENTATIVE_TEMPLATE_METRICS.contains(metric.getName()))
+ .collect(Collectors.toList());
+ }
+
+ @TestFactory
+ Stream shouldCollectRepresentativeTemplateMetricsAcrossCompatibilityMatrix() {
+ return Stream.of(
+ new DatabaseTarget("mysql-5.7.44", DockerImageName.parse("mysql:5.7.44"), false),
+ new DatabaseTarget("mysql-8.0.36", DockerImageName.parse("mysql:8.0.36"), false),
+ new DatabaseTarget("mariadb-11.4", DockerImageName.parse("mariadb:11.4"), true))
+ .map(target -> DynamicTest.dynamicTest(target.name(), () -> verifyRepresentativeMetrics(target)));
+ }
+
+ private void verifyRepresentativeMetrics(DatabaseTarget target) throws Exception {
+ MysqlCollectorProperties properties = new MysqlCollectorProperties();
+ properties.setQueryEngine(MysqlCollectorProperties.QueryEngine.R2DBC);
+ MysqlR2dbcJdbcQueryExecutor jdbcQueryExecutor = new MysqlR2dbcJdbcQueryExecutor(
+ properties,
+ new MysqlR2dbcQueryExecutor(
+ new MysqlR2dbcConnectionFactoryProvider(),
+ new ResultSetMapper(),
+ new SqlGuard()),
+ new MysqlJdbcDriverAvailability());
+ try (GenericContainer> container = createContainer(target)) {
+ jdbcQueryExecutor.afterPropertiesSet();
+ container.start();
+ awaitTcpLoginReady(container, TEST_USERNAME, TEST_PASSWORD, TEST_DATABASE);
+ initMonitoringData(container);
+
+ for (Metrics templateMetric : representativeTemplateMetrics) {
+ Metrics metric = materializeMetric(templateMetric, container);
+ if ("process_state".equals(metric.getName())) {
+ startBackgroundSleepQuery(container);
+ }
+ CollectRep.MetricsData metricsData = collect(metric);
+ assertEquals(CollectRep.Code.SUCCESS, metricsData.getCode(),
+ () -> target.name() + " " + metric.getName() + " failed: " + metricsData.getMsg());
+ assertEquals(metric.getFields().size(), metricsData.getFieldsCount(),
+ () -> target.name() + " " + metric.getName() + " should keep the original parser output shape");
+ if ("basic".equals(metric.getName())) {
+ assertTrue(metricsData.getValuesCount() > 0, () -> target.name() + " basic should return data");
+ assertNotNull(metricsData.getValues().getFirst().getColumns(0));
+ assertTrue(!Objects.equals(CommonConstants.NULL_VALUE, metricsData.getValues().getFirst().getColumns(0)),
+ () -> target.name() + " basic.version should be collected");
+ }
+ if ("process_state".equals(metric.getName())) {
+ assertTrue(metricsData.getValuesCount() > 0,
+ () -> target.name() + " process_state should return at least one grouped state row");
+ }
+ }
+ } finally {
+ jdbcQueryExecutor.destroy();
+ }
+ }
+
+ private GenericContainer> createContainer(DatabaseTarget target) {
+ GenericContainer> container = new GenericContainer<>(target.image())
+ .withExposedPorts(3306)
+ .waitingFor(Wait.forListeningPort());
+ if (target.mariaDb()) {
+ return container.withEnv("MARIADB_DATABASE", TEST_DATABASE)
+ .withEnv("MARIADB_USER", TEST_USERNAME)
+ .withEnv("MARIADB_PASSWORD", TEST_PASSWORD)
+ .withEnv("MARIADB_ROOT_PASSWORD", ROOT_PASSWORD);
+ }
+ return container.withEnv("MYSQL_DATABASE", TEST_DATABASE)
+ .withEnv("MYSQL_USER", TEST_USERNAME)
+ .withEnv("MYSQL_PASSWORD", TEST_PASSWORD)
+ .withEnv("MYSQL_ROOT_PASSWORD", ROOT_PASSWORD);
+ }
+
+ private CollectRep.MetricsData collect(Metrics metric) {
+ Job job = Job.builder()
+ .monitorId(1L)
+ .tenantId(1L)
+ .app("mysql")
+ .defaultInterval(600L)
+ .metadata(new HashMap<>(0))
+ .labels(new HashMap<>(0))
+ .annotations(new HashMap<>(0))
+ .configmap(new ArrayList<>(0))
+ .metrics(new ArrayList<>(List.of(metric)))
+ .build();
+ WheelTimerTask timerTask = new WheelTimerTask(job, timeout -> {
+ });
+ CapturingCollectDataDispatch collectDataDispatch = new CapturingCollectDataDispatch();
+ MetricsCollect metricsCollect = new MetricsCollect(
+ metric,
+ new StubTimeout(timerTask),
+ collectDataDispatch,
+ "collector-test",
+ List.of());
+ metricsCollect.run();
+ return collectDataDispatch.metricsData;
+ }
+
+ private Metrics materializeMetric(Metrics templateMetric, GenericContainer> container) {
+ Metrics metric = JsonUtil.fromJson(JsonUtil.toJson(templateMetric), Metrics.class);
+ JdbcProtocol jdbcProtocol = metric.getJdbc();
+ jdbcProtocol.setHost(container.getHost());
+ jdbcProtocol.setPort(String.valueOf(container.getMappedPort(3306)));
+ jdbcProtocol.setUsername(TEST_USERNAME);
+ jdbcProtocol.setPassword(TEST_PASSWORD);
+ jdbcProtocol.setTimeout(String.valueOf(Duration.ofSeconds(8).toMillis()));
+ jdbcProtocol.setReuseConnection("false");
+ jdbcProtocol.setUrl(null);
+ jdbcProtocol.setSshTunnel(null);
+ if (jdbcProtocol.getDatabase() == null || jdbcProtocol.getDatabase().contains("^_^")) {
+ jdbcProtocol.setDatabase(TEST_DATABASE);
+ }
+ if (metric.getAliasFields() == null || metric.getAliasFields().isEmpty()) {
+ metric.setAliasFields(metric.getFields().stream().map(Metrics.Field::getField).collect(Collectors.toList()));
+ }
+ return metric;
+ }
+
+ private Job loadMysqlTemplate() throws IOException {
+ Path template = Path.of("..", "..", "hertzbeat-manager", "src", "main", "resources", "define", "app-mysql.yml")
+ .toAbsolutePath()
+ .normalize();
+ Yaml yaml = new Yaml();
+ try (Reader reader = Files.newBufferedReader(template)) {
+ return yaml.loadAs(reader, Job.class);
+ }
+ }
+
+ private void initMonitoringData(GenericContainer> mysql) throws Exception {
+ execRoot(mysql,
+ "GRANT SELECT ON mysql.* TO '" + TEST_USERNAME + "'@'%';"
+ + " GRANT PROCESS ON *.* TO '" + TEST_USERNAME + "'@'%';"
+ + " SET GLOBAL log_output='TABLE';"
+ + " SET GLOBAL slow_query_log='ON';"
+ + " SET GLOBAL long_query_time=0;"
+ + " FLUSH PRIVILEGES;");
+ }
+
+ private void startBackgroundSleepQuery(GenericContainer> mysql) throws Exception {
+ String command = String.join(" ",
+ "CLIENT=$(command -v mysql || command -v mariadb)",
+ "&&",
+ "nohup sh -lc",
+ "'$CLIENT --protocol=TCP -h127.0.0.1 -P3306",
+ "-u" + TEST_USERNAME,
+ "-p" + TEST_PASSWORD,
+ TEST_DATABASE,
+ "-e",
+ "\"SELECT SLEEP(15)\" >/tmp/process-state.log 2>&1'",
+ ">/dev/null 2>&1 &");
+ mysql.execInContainer("sh", "-lc", command);
+ Thread.sleep(500);
+ }
+
+ private void awaitTcpLoginReady(GenericContainer> mysql, String username, String password, String database) throws Exception {
+ long deadline = System.currentTimeMillis() + Duration.ofSeconds(30).toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ var result = mysql.execInContainer("sh", "-lc", mysqlCliCommand(username, password, database, "SELECT 1"));
+ if (result.getExitCode() == 0) {
+ return;
+ }
+ } catch (Exception ignored) {
+ // Wait for the database entrypoint to finish bootstrapping and switch to the final TCP listener.
+ }
+ Thread.sleep(1000);
+ }
+ throw new IllegalStateException("Timed out waiting for MySQL-compatible TCP login to become ready");
+ }
+
+ private void execRoot(GenericContainer> mysql, String sql) throws Exception {
+ var result = mysql.execInContainer("sh", "-lc", mysqlCliCommand("root", ROOT_PASSWORD, "mysql", sql));
+ if (result.getExitCode() != 0) {
+ throw new IllegalStateException("root mysql command failed: " + result.getStderr());
+ }
+ }
+
+ private String mysqlCliCommand(String username, String password, String database, String sql) {
+ return String.join(" ",
+ "CLIENT=$(command -v mysql || command -v mariadb)",
+ "&&",
+ "$CLIENT --protocol=TCP -h127.0.0.1 -P3306",
+ "-u" + username,
+ "-p" + password,
+ database,
+ "-e",
+ "\"" + sql.replace("\"", "\\\"") + "\"");
+ }
+
+ private record DatabaseTarget(String name, DockerImageName image, boolean mariaDb) {
+ }
+
+ private static final class CapturingCollectDataDispatch implements CollectDataDispatch {
+
+ private CollectRep.MetricsData metricsData;
+
+ @Override
+ public void dispatchCollectData(Timeout timeout, Metrics metrics, CollectRep.MetricsData metricsData) {
+ this.metricsData = metricsData;
+ }
+
+ @Override
+ public void dispatchCollectData(Timeout timeout, Metrics metrics, List metricsDataList) {
+ if (metricsDataList != null && !metricsDataList.isEmpty()) {
+ this.metricsData = metricsDataList.getFirst();
+ }
+ }
+ }
+
+ private record StubTimeout(WheelTimerTask wheelTimerTask) implements Timeout {
+
+ @Override
+ public org.apache.hertzbeat.common.timer.Timer timer() {
+ return null;
+ }
+
+ @Override
+ public org.apache.hertzbeat.common.timer.TimerTask task() {
+ return wheelTimerTask;
+ }
+
+ @Override
+ public boolean isExpired() {
+ return false;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public boolean cancel() {
+ return false;
+ }
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryAdapterTemplateIntegrationTest.java b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryAdapterTemplateIntegrationTest.java
new file mode 100644
index 00000000000..3d91c7107d6
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryAdapterTemplateIntegrationTest.java
@@ -0,0 +1,322 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.mysql;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hertzbeat.collector.collect.strategy.CollectStrategyFactory;
+import org.apache.hertzbeat.collector.dispatch.CollectDataDispatch;
+import org.apache.hertzbeat.collector.dispatch.MetricsCollect;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlR2dbcConnectionFactoryProvider;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlR2dbcQueryExecutor;
+import org.apache.hertzbeat.collector.mysql.r2dbc.ResultSetMapper;
+import org.apache.hertzbeat.collector.mysql.r2dbc.SqlGuard;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.job.protocol.JdbcProtocol;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.apache.hertzbeat.common.util.JsonUtil;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.DynamicTest;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestFactory;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import org.yaml.snakeyaml.Yaml;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class MysqlJdbcQueryAdapterTemplateIntegrationTest {
+
+ private static final String TEST_DATABASE = "hzb";
+ private static final String TEST_USERNAME = "test";
+ private static final String TEST_PASSWORD = "test123";
+ private static final String ROOT_PASSWORD = "root123";
+
+ private GenericContainer> container;
+ private MysqlR2dbcJdbcQueryExecutor jdbcQueryExecutor;
+ private List mysqlTemplateMetrics;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ Assumptions.assumeTrue(DockerClientFactory.instance().isDockerAvailable(), "Docker is required for integration tests");
+ new CollectStrategyFactory().run();
+ container = new GenericContainer<>(DockerImageName.parse("mysql:8.0.36"))
+ .withExposedPorts(3306)
+ .withEnv("MYSQL_DATABASE", TEST_DATABASE)
+ .withEnv("MYSQL_USER", TEST_USERNAME)
+ .withEnv("MYSQL_PASSWORD", TEST_PASSWORD)
+ .withEnv("MYSQL_ROOT_PASSWORD", ROOT_PASSWORD)
+ .waitingFor(Wait.forListeningPort());
+ container.start();
+ awaitTcpLoginReady(container, TEST_USERNAME, TEST_PASSWORD, TEST_DATABASE);
+ initMonitoringData(container);
+
+ MysqlCollectorProperties properties = new MysqlCollectorProperties();
+ properties.setQueryEngine(MysqlCollectorProperties.QueryEngine.R2DBC);
+ jdbcQueryExecutor = new MysqlR2dbcJdbcQueryExecutor(
+ properties,
+ new MysqlR2dbcQueryExecutor(
+ new MysqlR2dbcConnectionFactoryProvider(),
+ new ResultSetMapper(),
+ new SqlGuard()),
+ new MysqlJdbcDriverAvailability());
+ jdbcQueryExecutor.afterPropertiesSet();
+ mysqlTemplateMetrics = loadMysqlTemplate().getMetrics();
+ }
+
+ @AfterAll
+ void tearDown() throws Exception {
+ if (jdbcQueryExecutor != null) {
+ jdbcQueryExecutor.destroy();
+ }
+ if (container != null) {
+ container.stop();
+ }
+ }
+
+ @TestFactory
+ Stream shouldCollectOfficialMysqlTemplateThroughJdbcQueryAdapter() {
+ return mysqlTemplateMetrics.stream()
+ .map(templateMetric -> DynamicTest.dynamicTest(templateMetric.getName(),
+ () -> verifyTemplateMetric(templateMetric)));
+ }
+
+ private void verifyTemplateMetric(Metrics templateMetric) throws Exception {
+ Metrics metric = materializeMetric(templateMetric);
+ if ("process_state".equals(metric.getName())) {
+ startBackgroundSleepQuery(container);
+ }
+ if ("slow_sql".equals(metric.getName())) {
+ generateSlowQuery(container);
+ }
+ CollectRep.MetricsData metricsData = collect(metric);
+ assertEquals(CollectRep.Code.SUCCESS, metricsData.getCode(),
+ () -> metric.getName() + " failed: " + metricsData.getMsg());
+ assertEquals(metric.getFields().size(), metricsData.getFieldsCount(),
+ () -> metric.getName() + " fields should still be produced by the original parser");
+ if ("columns".equals(metric.getJdbc().getQueryType())) {
+ assertEquals(1, metricsData.getValuesCount(), () -> metric.getName() + " should keep the original single-row shape");
+ }
+ if ("basic".equals(metric.getName())) {
+ assertTrue(metricsData.getValuesCount() > 0);
+ assertNotNull(metricsData.getValues().getFirst().getColumns(0));
+ assertTrue(!Objects.equals(CommonConstants.NULL_VALUE, metricsData.getValues().getFirst().getColumns(0)),
+ "basic.version should be collected through the adapted query path");
+ }
+ if ("process_state".equals(metric.getName())
+ || "slow_sql".equals(metric.getName())
+ || "account_expiry".equals(metric.getName())) {
+ assertTrue(metricsData.getValuesCount() > 0, () -> metric.getName() + " should return at least one row");
+ }
+ }
+
+ private CollectRep.MetricsData collect(Metrics metric) {
+ Job job = Job.builder()
+ .monitorId(1L)
+ .tenantId(1L)
+ .app("mysql")
+ .defaultInterval(600L)
+ .metadata(new HashMap<>(0))
+ .labels(new HashMap<>(0))
+ .annotations(new HashMap<>(0))
+ .configmap(new ArrayList<>(0))
+ .metrics(new ArrayList<>(List.of(metric)))
+ .build();
+ WheelTimerTask timerTask = new WheelTimerTask(job, timeout -> {
+ });
+ CapturingCollectDataDispatch collectDataDispatch = new CapturingCollectDataDispatch();
+ MetricsCollect metricsCollect = new MetricsCollect(
+ metric,
+ new StubTimeout(timerTask),
+ collectDataDispatch,
+ "collector-test",
+ List.of());
+ metricsCollect.run();
+ assertNotNull(collectDataDispatch.metricsData, metric.getName() + " should dispatch metrics data");
+ return collectDataDispatch.metricsData;
+ }
+
+ private Metrics materializeMetric(Metrics templateMetric) {
+ Metrics metric = JsonUtil.fromJson(JsonUtil.toJson(templateMetric), Metrics.class);
+ JdbcProtocol jdbcProtocol = metric.getJdbc();
+ jdbcProtocol.setHost(container.getHost());
+ jdbcProtocol.setPort(String.valueOf(container.getMappedPort(3306)));
+ jdbcProtocol.setUsername(TEST_USERNAME);
+ jdbcProtocol.setPassword(TEST_PASSWORD);
+ jdbcProtocol.setTimeout(String.valueOf(Duration.ofSeconds(8).toMillis()));
+ jdbcProtocol.setReuseConnection("false");
+ jdbcProtocol.setUrl(null);
+ jdbcProtocol.setSshTunnel(null);
+ if (jdbcProtocol.getDatabase() == null || jdbcProtocol.getDatabase().contains("^_^")) {
+ jdbcProtocol.setDatabase(TEST_DATABASE);
+ }
+ if (metric.getAliasFields() == null || metric.getAliasFields().isEmpty()) {
+ metric.setAliasFields(metric.getFields().stream().map(Metrics.Field::getField).collect(Collectors.toList()));
+ }
+ return metric;
+ }
+
+ private Job loadMysqlTemplate() throws IOException {
+ Path template = Path.of("..", "..", "hertzbeat-manager", "src", "main", "resources", "define", "app-mysql.yml")
+ .toAbsolutePath()
+ .normalize();
+ Yaml yaml = new Yaml();
+ try (Reader reader = Files.newBufferedReader(template)) {
+ return yaml.loadAs(reader, Job.class);
+ }
+ }
+
+ private void initMonitoringData(GenericContainer> mysql) throws Exception {
+ execRoot(mysql,
+ "GRANT SELECT ON mysql.* TO '" + TEST_USERNAME + "'@'%';"
+ + " GRANT PROCESS ON *.* TO '" + TEST_USERNAME + "'@'%';"
+ + " SET GLOBAL log_output='TABLE';"
+ + " SET GLOBAL slow_query_log='ON';"
+ + " SET GLOBAL long_query_time=0;"
+ + " FLUSH PRIVILEGES;");
+ generateSlowQuery(mysql);
+ }
+
+ private void generateSlowQuery(GenericContainer> mysql) throws Exception {
+ execUser(mysql, TEST_DATABASE, "SELECT SLEEP(0.2);");
+ Thread.sleep(300);
+ }
+
+ private void startBackgroundSleepQuery(GenericContainer> mysql) throws Exception {
+ String command = String.join(" ",
+ "CLIENT=$(command -v mysql || command -v mariadb)",
+ "&&",
+ "nohup sh -lc",
+ "'$CLIENT --protocol=TCP -h127.0.0.1 -P3306",
+ "-u" + TEST_USERNAME,
+ "-p" + TEST_PASSWORD,
+ TEST_DATABASE,
+ "-e",
+ "\"SELECT SLEEP(15)\" >/tmp/process-state.log 2>&1'",
+ ">/dev/null 2>&1 &");
+ mysql.execInContainer("sh", "-lc", command);
+ Thread.sleep(500);
+ }
+
+ private void awaitTcpLoginReady(GenericContainer> mysql, String username, String password, String database) throws Exception {
+ long deadline = System.currentTimeMillis() + Duration.ofSeconds(30).toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ var result = mysql.execInContainer("sh", "-lc", mysqlCliCommand(username, password, database, "SELECT 1"));
+ if (result.getExitCode() == 0) {
+ return;
+ }
+ } catch (Exception ignored) {
+ // Wait for the MySQL entrypoint to finish bootstrapping and switch to the final TCP listener.
+ }
+ Thread.sleep(1000);
+ }
+ throw new IllegalStateException("Timed out waiting for MySQL TCP login to become ready");
+ }
+
+ private void execRoot(GenericContainer> mysql, String sql) throws Exception {
+ var result = mysql.execInContainer("sh", "-lc", mysqlCliCommand("root", ROOT_PASSWORD, "mysql", sql));
+ if (result.getExitCode() != 0) {
+ throw new IllegalStateException("root mysql command failed: " + result.getStderr());
+ }
+ }
+
+ private void execUser(GenericContainer> mysql, String database, String sql) throws Exception {
+ var result = mysql.execInContainer("sh", "-lc", mysqlCliCommand(TEST_USERNAME, TEST_PASSWORD, database, sql));
+ if (result.getExitCode() != 0) {
+ throw new IllegalStateException("user mysql command failed: " + result.getStderr());
+ }
+ }
+
+ private String mysqlCliCommand(String username, String password, String database, String sql) {
+ return String.join(" ",
+ "CLIENT=$(command -v mysql || command -v mariadb)",
+ "&&",
+ "$CLIENT --protocol=TCP -h127.0.0.1 -P3306",
+ "-u" + username,
+ "-p" + password,
+ database,
+ "-e",
+ "\"" + sql.replace("\"", "\\\"") + "\"");
+ }
+
+ private static final class CapturingCollectDataDispatch implements CollectDataDispatch {
+
+ private CollectRep.MetricsData metricsData;
+
+ @Override
+ public void dispatchCollectData(Timeout timeout, Metrics metrics, CollectRep.MetricsData metricsData) {
+ this.metricsData = metricsData;
+ }
+
+ @Override
+ public void dispatchCollectData(Timeout timeout, Metrics metrics, List metricsDataList) {
+ if (metricsDataList != null && !metricsDataList.isEmpty()) {
+ this.metricsData = metricsDataList.getFirst();
+ }
+ }
+ }
+
+ private record StubTimeout(WheelTimerTask wheelTimerTask) implements Timeout {
+
+ @Override
+ public org.apache.hertzbeat.common.timer.Timer timer() {
+ return null;
+ }
+
+ @Override
+ public org.apache.hertzbeat.common.timer.TimerTask task() {
+ return wheelTimerTask;
+ }
+
+ @Override
+ public boolean isExpired() {
+ return false;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public boolean cancel() {
+ return false;
+ }
+ }
+}
diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryParityIntegrationTest.java b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryParityIntegrationTest.java
new file mode 100644
index 00000000000..650e6e92c5c
--- /dev/null
+++ b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/mysql/MysqlJdbcQueryParityIntegrationTest.java
@@ -0,0 +1,393 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.collector.collect.database.mysql;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.lang.reflect.Field;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hertzbeat.collector.collect.strategy.CollectStrategyFactory;
+import org.apache.hertzbeat.collector.collect.database.query.JdbcQueryExecutorRegistry;
+import org.apache.hertzbeat.collector.dispatch.CollectDataDispatch;
+import org.apache.hertzbeat.collector.dispatch.MetricsCollect;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlR2dbcConnectionFactoryProvider;
+import org.apache.hertzbeat.collector.mysql.r2dbc.MysqlR2dbcQueryExecutor;
+import org.apache.hertzbeat.collector.mysql.r2dbc.ResultSetMapper;
+import org.apache.hertzbeat.collector.mysql.r2dbc.SqlGuard;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.job.protocol.JdbcProtocol;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.apache.hertzbeat.common.util.JsonUtil;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.DynamicTest;
+import org.junit.jupiter.api.TestFactory;
+import org.junit.jupiter.api.TestInstance;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import org.yaml.snakeyaml.Yaml;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class MysqlJdbcQueryParityIntegrationTest {
+
+ private static final String TEST_DATABASE = "hzb";
+ private static final String TEST_USERNAME = "test";
+ private static final String TEST_PASSWORD = "test123";
+ private static final String ROOT_PASSWORD = "root123";
+ private static final String PARITY_TABLE = "collector_parity_metrics";
+
+ private Metrics basicTemplateMetric;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ Assumptions.assumeTrue(DockerClientFactory.instance().isDockerAvailable(), "Docker is required for integration tests");
+ new CollectStrategyFactory().run();
+ assertDoesNotThrow(() -> Class.forName("com.mysql.cj.jdbc.Driver"));
+ basicTemplateMetric = loadMysqlTemplate().getMetrics().stream()
+ .filter(metric -> "basic".equals(metric.getName()))
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException("Unable to locate the basic metric in app-mysql.yml"));
+ }
+
+ @AfterEach
+ void clearRegisteredExecutors() throws Exception {
+ Field executorsField = JdbcQueryExecutorRegistry.class.getDeclaredField("EXECUTORS");
+ executorsField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ CopyOnWriteArrayList