This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11984-c44f5b3388a8664a997b915f2d4b55c580ab3c90 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 9ec9c4fe8aa62c628c638ba5a625d2624b8a4849 Author: SEZ <[email protected]> AuthorDate: Sat Aug 29 02:09:54 2026 +0000 [Feature][Zeta] Read and change log levels on the v2 REST API (#11984) --- docs/en/engines/zeta/logging.md | 20 ++ docs/en/engines/zeta/rest-api-v2.md | 166 ++++++++++++ .../introduction/concepts/incompatible-changes.md | 6 + docs/zh/engines/zeta/logging.md | 18 ++ docs/zh/engines/zeta/rest-api-v2.md | 160 +++++++++++ .../introduction/concepts/incompatible-changes.md | 6 + .../org/apache/seatunnel/engine/e2e/RestApiIT.java | 102 +++++++ .../seatunnel/engine/server/JettyService.java | 4 + .../server/log/Log4j2HttpPostCommandProcessor.java | 25 +- .../seatunnel/engine/server/log/LogLevels.java | 241 +++++++++++++++++ .../seatunnel/engine/server/rest/RestConstant.java | 2 + .../server/rest/service/LoggerLevelService.java | 301 +++++++++++++++++++++ .../engine/server/rest/servlet/LoggersServlet.java | 186 +++++++++++++ .../seatunnel/engine/server/log/LogLevelsTest.java | 209 ++++++++++++++ 14 files changed, 1439 insertions(+), 7 deletions(-) diff --git a/docs/en/engines/zeta/logging.md b/docs/en/engines/zeta/logging.md index 0fb82ae44b..3abb488629 100644 --- a/docs/en/engines/zeta/logging.md +++ b/docs/en/engines/zeta/logging.md @@ -95,6 +95,26 @@ For more details, please refer to the [REST-API](rest-api-v2.md). For a workflow to collect, redact, and analyze runtime logs, see [Diagnose Runtime Logs with AI Tools](log-analysis-with-ai.md). +### Change Log Levels at Runtime + +There are two ways to change a log level, and they behave differently: + +- **Edit `log4j2.properties`**: Log4j 2 picks the change up on its next scan of the file (every 60 + seconds by default, see `monitorInterval`), the change survives a restart, and it has to be rolled + out to every node. +- **Call the `/loggers` REST API**: the change takes effect immediately on the node that serves the + request, or on every node with `?scope=cluster`, and it is lost when the node restarts. + +**Usage examples:** +- List the loggers of a node and where their level comes from: `http://localhost:8080/loggers` +- Raise one connector to `DEBUG` on the whole cluster: + `curl -X POST 'http://localhost:8080/loggers/org.apache.seatunnel.connectors.seatunnel.jdbc?level=DEBUG&scope=cluster'` +- Drop the override again: `curl -X DELETE 'http://localhost:8080/loggers/org.apache.seatunnel.connectors.seatunnel.jdbc?scope=cluster'` + +A logger that was changed through the API reports `"origin": "runtime-override"`, so an override is +never mistaken for the configured level. For the full request and response format, please refer to +the [REST-API](rest-api-v2.md). + ## SeaTunnel Log Configuration ### Scheduled deletion of old logs diff --git a/docs/en/engines/zeta/rest-api-v2.md b/docs/en/engines/zeta/rest-api-v2.md index 5bd9b821c2..4cbf697ca6 100644 --- a/docs/en/engines/zeta/rest-api-v2.md +++ b/docs/en/engines/zeta/rest-api-v2.md @@ -1378,6 +1378,172 @@ To get the content of a log file: `http://localhost:5801/log/job-898380162133917 </details> +------------------------------------------------------------------------------------------ + +### Read And Change Log Levels + +A log level changed through these endpoints is a runtime override: it takes effect immediately, is +node local, and is lost when the node restarts. Levels that should survive a restart belong in +`config/log4j2.properties`, see [Logging](logging.md). + +The root logger is addressed as `root`. + +<details> + <summary><code>GET</code> <code><b>/loggers</b></code> <code>(Returns the loggers of the running configuration.)</code></summary> + +#### Query Parameters + +> | Parameter Name | Required | Type | Description | +> |------------------|--------------|---------|--------------------------------------------------------------------------------| +> | scope | optional | string | `node` (default) answers for the node that serves the request, `cluster` asks every member | + +#### Response + +```json +{ + "node": "localhost:8080", + "loggers": [ + { + "name": "root", + "level": "INFO", + "origin": "file" + }, + { + "name": "org.apache.seatunnel.connectors.seatunnel.jdbc", + "level": "DEBUG", + "origin": "runtime-override", + "fileLevel": "INFO" + } + ] +} +``` + +`origin` tells where the current level comes from: `file` for the level of the log4j2 configuration +file, `runtime-override` for a level that was set through one of the log level endpoints. `fileLevel` +is only present when an overridden logger is configured in the file as well, and reports the level a +`DELETE` puts back. + +With `?scope=cluster` the answer is one entry per member: + +```json +{ + "scope": "cluster", + "status": "SUCCESS", + "nodes": [ + { + "node": "localhost:8080", + "loggers": [ + { + "name": "root", + "level": "INFO", + "origin": "file" + } + ] + } + ] +} +``` + +`status` is `SUCCESS` when every member answered, `PARTIAL_FAILURE` when some did not, and `FAILURE` +when none did; the member that failed carries its own `status` and `error`. A cluster request reaches +every member on the REST port of its configuration, so it does not reach members that took a +different port through `enable-dynamic-port`. + +</details> + +<details> + <summary><code>GET</code> <code><b>/loggers/:name</b></code> <code>(Returns the effective level of one logger.)</code></summary> + +#### Response + +The level is resolved through the closest configured ancestor, so a logger that is not configured +itself can be asked about as well. + +```json +{ + "name": "org.apache.seatunnel.connectors.seatunnel.jdbc", + "level": "INFO", + "origin": "file", + "node": "localhost:8080" +} +``` + +</details> + +<details> + <summary><code>POST</code> <code><b>/loggers/:name</b></code> <code>(Overrides the level of one logger.)</code></summary> + +#### Query Parameters + +> | Parameter Name | Required | Type | Description | +> |------------------|--------------|---------|--------------------------------------------------------------------------------| +> | level | optional | string | `OFF`, `FATAL`, `ERROR`, `WARN`, `INFO`, `DEBUG`, `TRACE` or `ALL`, any letter case; may also be sent in the body | +> | scope | optional | string | `node` (default) changes the node that serves the request, `cluster` changes every member | + +#### Body + +```json +{ + "level": "DEBUG" +} +``` + +#### Response + +```json +{ + "name": "org.apache.seatunnel.connectors.seatunnel.jdbc", + "level": "DEBUG", + "origin": "runtime-override", + "node": "localhost:8080", + "previousLevel": "INFO", + "status": "SUCCESS" +} +``` + +An unknown level is rejected with `400` and the list of valid levels instead of being reported as +applied. Every change is written to the node log as a single `INFO` line with the logger, the old and +the new level, the scope and the address of the caller. + +#### Examples + +Raise the JDBC connector to `DEBUG` on one node: +`curl -X POST 'http://localhost:8080/loggers/org.apache.seatunnel.connectors.seatunnel.jdbc?level=DEBUG'` + +Raise it on every member of the cluster: +`curl -X POST 'http://localhost:8080/loggers/org.apache.seatunnel.connectors.seatunnel.jdbc?level=DEBUG&scope=cluster'` + +</details> + +<details> + <summary><code>DELETE</code> <code><b>/loggers/:name</b></code> <code>(Reverts a runtime override.)</code></summary> + +#### Query Parameters + +> | Parameter Name | Required | Type | Description | +> |------------------|--------------|---------|--------------------------------------------------------------------------------| +> | scope | optional | string | `node` (default) reverts the node that serves the request, `cluster` reverts every member | + +#### Response + +The logger goes back to the level it had before its first override, which is the level of the +configuration file, or the level inherited from its parent when the file does not configure it. + +```json +{ + "name": "org.apache.seatunnel.connectors.seatunnel.jdbc", + "level": "INFO", + "origin": "file", + "node": "localhost:8080", + "previousLevel": "DEBUG", + "status": "SUCCESS" +} +``` + +`status` is `NO_OVERRIDE` when the logger was never overridden through an endpoint; nothing is +changed in that case. + +</details> ### Get Node Metrics diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 41437be0be..b64f17b2b6 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -60,6 +60,12 @@ You need to check this document before you upgrade to related version. } ``` +- **Breaking Change: An unknown log level is rejected by the runtime log level endpoint** + - **Affected component**: SeaTunnel Engine REST API — `POST /hazelcast/rest/maps/log-level` + - **Description**: The endpoint answered `200` with `{"status":"SUCCESS"}` for every request, including a level name it could not resolve (`DEBUGG`, `verbose`, a lowercase name of a level that does not exist, an empty value). Nothing was applied in that case, and the unresolved level was handed to log4j2 as `null`, which removes the explicit level of the logger instead of leaving it alone — so the logger silently fell back to its parent, or to `ERROR` for the root logger. An unknown le [...] + - **Impact**: Scripts and automation that only check the HTTP status now see `400` where they used to see `200`, for requests that never took effect in the first place. Requests with a resolvable level are unchanged. + - **Migration Guide**: Send a level log4j2 knows (`OFF`, `FATAL`, `ERROR`, `WARN`, `INFO`, `DEBUG`, `TRACE`, `ALL`, or a level registered by the configuration). The response body of a rejected request names the levels the node accepts. + - **Breaking Change: `Condition.of(option, null)` no longer allowed** - **Affected component**: `seatunnel-api` — `org.apache.seatunnel.api.configuration.util.Condition` - **Description**: The `Condition` constructor now validates that binary literal operators (such as `EQUAL`, `NOT_EQUAL`, `GREATER_THAN`, etc.) must have a non-null `expectValue`. Previously, `Condition.of(option, null)` was silently accepted; it now throws `IllegalArgumentException` at construction time. diff --git a/docs/zh/engines/zeta/logging.md b/docs/zh/engines/zeta/logging.md index d439920dd8..d1cfcdce22 100644 --- a/docs/zh/engines/zeta/logging.md +++ b/docs/zh/engines/zeta/logging.md @@ -94,6 +94,24 @@ SeaTunnel 提供了一个 API，用于查询日志。 有关收集、脱敏和分析运行日志的流程，请参阅[使用 AI 辅助诊断运行日志](log-analysis-with-ai.md)。 +### 运行期修改日志级别 + +修改日志级别有两种方式，二者行为不同： + +- **修改 `log4j2.properties`**：Log4j 2 在下一次扫描配置文件时生效（默认 60 秒一次，由 + `monitorInterval` 控制），重启后依然有效，但需要在每个节点上分发。 +- **调用 `/loggers` REST 接口**：立即在接收请求的节点上生效，使用 `?scope=cluster` 时在所有节点上生效， + 但节点重启后丢失。 + +**使用样例：** +- 查看某个节点的 logger 列表以及各自级别的来源：`http://localhost:8080/loggers` +- 在整个集群上把某个连接器调整为 `DEBUG`： + `curl -X POST 'http://localhost:8080/loggers/org.apache.seatunnel.connectors.seatunnel.jdbc?level=DEBUG&scope=cluster'` +- 撤销该覆盖：`curl -X DELETE 'http://localhost:8080/loggers/org.apache.seatunnel.connectors.seatunnel.jdbc?scope=cluster'` + +通过接口修改过的 logger 会返回 `"origin": "runtime-override"`，因此运行时覆盖不会被误认为是配置文件中的 +级别。完整的请求与响应格式请参阅 [REST-API](rest-api-v2.md)。 + ## SeaTunnel 日志配置 ### 定时删除旧日志 diff --git a/docs/zh/engines/zeta/rest-api-v2.md b/docs/zh/engines/zeta/rest-api-v2.md index 81ba2da7be..fb744e73ae 100644 --- a/docs/zh/engines/zeta/rest-api-v2.md +++ b/docs/zh/engines/zeta/rest-api-v2.md @@ -1362,6 +1362,166 @@ curl --location 'http://127.0.0.1:8080/submit-job/upload?restoreMode=CHECKPOINT&; </details> +------------------------------------------------------------------------------------------ + +### 查看与修改日志级别 + +通过这些接口修改的日志级别属于运行时覆盖：立即生效、仅作用于单个节点、节点重启后丢失。需要在重启后依然 +生效的级别请写入 `config/log4j2.properties`，参见 [日志](logging.md)。 + +根 logger 的名字是 `root`。 + +<details> + <summary><code>GET</code> <code><b>/loggers</b></code> <code>(返回当前生效配置中的 logger 列表。)</code></summary> + +#### 请求参数 + +> | 参数名 | 是否必填 | 类型 | 描述 | +> |----------------|--------------|---------|---------------------------------------------------------------------------| +> | scope | optional | string | `node`（默认）只处理接收请求的节点，`cluster` 会请求集群中的每个节点 | + +#### 响应 + +```json +{ + "node": "localhost:8080", + "loggers": [ + { + "name": "root", + "level": "INFO", + "origin": "file" + }, + { + "name": "org.apache.seatunnel.connectors.seatunnel.jdbc", + "level": "DEBUG", + "origin": "runtime-override", + "fileLevel": "INFO" + } + ] +} +``` + +`origin` 表示当前级别的来源：`file` 表示来自 log4j2 配置文件，`runtime-override` 表示通过日志级别接口 +设置。只有当被覆盖的 logger 同时也在配置文件中配置过时才会返回 `fileLevel`，它就是 `DELETE` 会恢复的级别。 + +使用 `?scope=cluster` 时每个节点返回一条记录： + +```json +{ + "scope": "cluster", + "status": "SUCCESS", + "nodes": [ + { + "node": "localhost:8080", + "loggers": [ + { + "name": "root", + "level": "INFO", + "origin": "file" + } + ] + } + ] +} +``` + +所有节点都返回结果时 `status` 为 `SUCCESS`，部分节点失败时为 `PARTIAL_FAILURE`，全部失败时为 +`FAILURE`；失败的节点会带上自己的 `status` 与 `error`。集群请求按各节点配置中的 REST 端口访问，因此无法 +访问通过 `enable-dynamic-port` 使用了其它端口的节点。 + +</details> + +<details> + <summary><code>GET</code> <code><b>/loggers/:name</b></code> <code>(返回单个 logger 的生效级别。)</code></summary> + +#### 响应 + +级别通过最近的已配置父级 logger 解析，因此也可以查询本身没有被配置的 logger。 + +```json +{ + "name": "org.apache.seatunnel.connectors.seatunnel.jdbc", + "level": "INFO", + "origin": "file", + "node": "localhost:8080" +} +``` + +</details> + +<details> + <summary><code>POST</code> <code><b>/loggers/:name</b></code> <code>(修改单个 logger 的级别。)</code></summary> + +#### 请求参数 + +> | 参数名 | 是否必填 | 类型 | 描述 | +> |----------------|--------------|---------|---------------------------------------------------------------------------| +> | level | optional | string | `OFF`、`FATAL`、`ERROR`、`WARN`、`INFO`、`DEBUG`、`TRACE` 或 `ALL`，不区分大小写；也可以放在请求体中 | +> | scope | optional | string | `node`（默认）只修改接收请求的节点，`cluster` 会修改集群中的每个节点 | + +#### 请求体 + +```json +{ + "level": "DEBUG" +} +``` + +#### 响应 + +```json +{ + "name": "org.apache.seatunnel.connectors.seatunnel.jdbc", + "level": "DEBUG", + "origin": "runtime-override", + "node": "localhost:8080", + "previousLevel": "INFO", + "status": "SUCCESS" +} +``` + +未知级别不会被当作已生效处理，而是返回 `400` 并列出所有合法级别。每次修改都会在节点日志中输出一行 `INFO` +记录，包含 logger 名、修改前后的级别、作用范围以及调用方地址。 + +#### 例子 + +把单个节点上的 JDBC 连接器调整为 `DEBUG`： +`curl -X POST 'http://localhost:8080/loggers/org.apache.seatunnel.connectors.seatunnel.jdbc?level=DEBUG'` + +在集群的每个节点上调整： +`curl -X POST 'http://localhost:8080/loggers/org.apache.seatunnel.connectors.seatunnel.jdbc?level=DEBUG&scope=cluster'` + +</details> + +<details> + <summary><code>DELETE</code> <code><b>/loggers/:name</b></code> <code>(撤销运行时覆盖。)</code></summary> + +#### 请求参数 + +> | 参数名 | 是否必填 | 类型 | 描述 | +> |----------------|--------------|---------|---------------------------------------------------------------------------| +> | scope | optional | string | `node`（默认）只撤销接收请求的节点，`cluster` 会撤销集群中的每个节点 | + +#### 响应 + +logger 会恢复到首次被覆盖之前的状态：配置文件中配置的级别，或者在配置文件没有配置它时恢复为从父级 logger +继承的级别。 + +```json +{ + "name": "org.apache.seatunnel.connectors.seatunnel.jdbc", + "level": "INFO", + "origin": "file", + "node": "localhost:8080", + "previousLevel": "DEBUG", + "status": "SUCCESS" +} +``` + +如果该 logger 从未通过接口被覆盖过，`status` 为 `NO_OVERRIDE`，此时不做任何修改。 + +</details> + ### 获取节点指标信息 <details> diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index 7616859aef..8def490447 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -54,6 +54,12 @@ } ``` +- **破坏性变更：运行期日志级别接口拒绝无法识别的级别** + - **影响范围**：SeaTunnel Engine REST API — `POST /hazelcast/rest/maps/log-level` + - **变更说明**：该接口此前对任何请求都返回 `200` 和 `{"status":"SUCCESS"}`，包括无法识别的级别名（`DEBUGG`、`verbose`、不存在的级别、空值）。这类请求实际上什么都没有生效，并且无法识别的级别会以 `null` 传给 log4j2，而 `null` 并不是"保持不变"：它会清除该 logger 上显式设置的级别，于是 logger 静默回退到父级别，root logger 则回退到 `ERROR`。现在无法识别的级别、空级别以及缺少 `level` 参数都会返回 `400`，并在响应中列出有效级别；级别名仍然不区分大小写。 + - **影响**：只检查 HTTP 状态码的脚本和自动化流程，对于原本就没有生效的请求，会从 `200` 变为 `400`。能够正确识别级别的请求行为不变。 + - **升级指南**：请传入 log4j2 能识别的级别（`OFF`、`FATAL`、`ERROR`、`WARN`、`INFO`、`DEBUG`、`TRACE`、`ALL`，或配置中注册的自定义级别）。被拒绝请求的响应体会列出该节点接受的级别。 + - **破坏性变更：`Condition.of(option, null)` 不再允许** - **影响范围**：`seatunnel-api` — `org.apache.seatunnel.api.configuration.util.Condition` - **变更说明**：`Condition` 构造器新增校验：二元字面量操作符（如 `EQUAL`、`NOT_EQUAL`、`GREATER_THAN` 等）的 `expectValue` 不能为 null。此前 `Condition.of(option, null)` 会被静默接受，现在会在构造时抛出 `IllegalArgumentException`。 diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java index ab0accdfa3..a25362a965 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java @@ -67,6 +67,7 @@ import static org.apache.seatunnel.e2e.common.util.ContainerUtil.PROJECT_ROOT_PA import static org.apache.seatunnel.engine.server.rest.RestConstant.CONTEXT_PATH; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasItem; import static org.hamcrest.Matchers.hasKey; import static org.hamcrest.Matchers.hasSize; import static org.hamcrest.Matchers.lessThan; @@ -246,6 +247,107 @@ public class RestApiIT { })); } + @Test + public void testLoggers() { + String loggersUrl = + HOST + + node1Config.getEngineConfig().getHttpConfig().getPort() + + node1Config.getEngineConfig().getHttpConfig().getContextPath() + + RestConstant.REST_URL_LOGGERS; + // a logger of this test only, so that changing its level cannot hide job logs + String logger = "org.apache.seatunnel.engine.e2e.RestApiIT.loggers"; + + // the list reports every configured logger, the root logger included + given().get(loggersUrl) + .then() + .statusCode(200) + .body("node", notNullValue()) + .body("loggers.name", hasItem("root")); + + given().get(loggersUrl + "/root") + .then() + .statusCode(200) + .body("name", equalTo("root")) + .body("level", notNullValue()) + .body("origin", equalTo("file")); + + // an unknown level is rejected instead of being reported as applied + given().post(loggersUrl + "/" + logger + "?level=DEBUGG") + .then() + .statusCode(400) + .body("message", containsString("valid levels are")); + + // a request without a level does not silently do nothing either + given().post(loggersUrl + "/" + logger) + .then() + .statusCode(400) + .body("message", containsString("level is required")); + + given().post(loggersUrl + "?level=DEBUG") + .then() + .statusCode(400) + .body("message", containsString("logger name")); + + given().get(loggersUrl + "?scope=galaxy") + .then() + .statusCode(400) + .body("message", containsString("valid scopes are")); + + // the level is applied and reported as a runtime override, not as the file level + given().body("{\"level\":\"debug\"}") + .post(loggersUrl + "/" + logger) + .then() + .statusCode(200) + .body("status", equalTo("SUCCESS")) + .body("name", equalTo(logger)) + .body("level", equalTo("DEBUG")) + .body("origin", equalTo("runtime-override")); + + given().get(loggersUrl + "/" + logger) + .then() + .statusCode(200) + .body("level", equalTo("DEBUG")) + .body("origin", equalTo("runtime-override")); + + // and the override can be reverted, which the legacy log-level endpoint cannot do + given().delete(loggersUrl + "/" + logger) + .then() + .statusCode(200) + .body("status", equalTo("SUCCESS")) + .body("previousLevel", equalTo("DEBUG")) + .body("origin", equalTo("file")); + + given().delete(loggersUrl + "/" + logger) + .then() + .statusCode(200) + .body("status", equalTo("NO_OVERRIDE")); + + // the cluster scope answers for every member + given().get(loggersUrl + "?scope=cluster") + .then() + .statusCode(200) + .body("scope", equalTo("cluster")) + .body("status", equalTo("SUCCESS")) + .body("nodes", hasSize(ports.size())); + + given().post(loggersUrl + "/" + logger + "?scope=cluster&level=TRACE") + .then() + .statusCode(200) + .body("scope", equalTo("cluster")) + .body("status", equalTo("SUCCESS")) + .body("level", equalTo("TRACE")) + .body("nodes", hasSize(ports.size())) + .body("nodes[0].level", equalTo("TRACE")) + .body("nodes[0].origin", equalTo("runtime-override")); + + given().delete(loggersUrl + "/" + logger + "?scope=cluster") + .then() + .statusCode(200) + .body("scope", equalTo("cluster")) + .body("status", equalTo("SUCCESS")) + .body("nodes[0].origin", equalTo("file")); + } + private CheckpointMonitorService resolveCheckpointMonitorService( HazelcastInstanceImpl instance) { try { diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java index 9686960fb5..0c45ada83f 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java @@ -37,6 +37,7 @@ import org.apache.seatunnel.engine.server.rest.servlet.CurrentNodeLogServlet; import org.apache.seatunnel.engine.server.rest.servlet.EncryptConfigServlet; import org.apache.seatunnel.engine.server.rest.servlet.FinishedJobsServlet; import org.apache.seatunnel.engine.server.rest.servlet.JobInfoServlet; +import org.apache.seatunnel.engine.server.rest.servlet.LoggersServlet; import org.apache.seatunnel.engine.server.rest.servlet.MetricsServlet; import org.apache.seatunnel.engine.server.rest.servlet.OptionRulesServlet; import org.apache.seatunnel.engine.server.rest.servlet.OverviewServlet; @@ -73,6 +74,7 @@ import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_FINI import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_GET_ALL_LOG_NAME; import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_JOB_INFO; import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_LOG; +import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_LOGGERS; import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_LOGS; import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_METRICS; import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_OPEN_METRICS; @@ -210,6 +212,7 @@ public class JettyService { ServletHolder currentNodeLogServlet = new ServletHolder(new CurrentNodeLogServlet(nodeEngine)); ServletHolder allLogNameServlet = new ServletHolder(new AllLogNameServlet(nodeEngine)); + ServletHolder loggersServlet = new ServletHolder(new LoggersServlet(nodeEngine)); ServletHolder metricsServlet = new ServletHolder(new MetricsServlet(nodeEngine)); ServletHolder realtimeMetricsServlet = @@ -246,6 +249,7 @@ public class JettyService { context.addServlet(allNodeLogServletHolder, convertUrlToPath(REST_URL_LOGS)); context.addServlet(currentNodeLogServlet, convertUrlToPath(REST_URL_LOG)); context.addServlet(allLogNameServlet, convertUrlToPath(REST_URL_GET_ALL_LOG_NAME)); + context.addServlet(loggersServlet, convertUrlToPath(REST_URL_LOGGERS)); context.addServlet(metricsServlet, convertUrlToPath(REST_URL_METRICS)); context.addServlet(metricsServlet, convertUrlToPath(REST_URL_OPEN_METRICS)); context.addServlet(realtimeMetricsServlet, convertUrlToPath(REST_URL_REALTIME_METRICS)); diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/log/Log4j2HttpPostCommandProcessor.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/log/Log4j2HttpPostCommandProcessor.java index 33b300b76f..c19d470ff8 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/log/Log4j2HttpPostCommandProcessor.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/log/Log4j2HttpPostCommandProcessor.java @@ -18,8 +18,6 @@ package org.apache.seatunnel.engine.server.log; import org.apache.logging.log4j.Level; -import org.apache.logging.log4j.core.config.Configurator; -import org.apache.logging.log4j.core.config.LoggerConfig; import com.hazelcast.internal.ascii.TextCommandService; import com.hazelcast.internal.ascii.rest.HttpCommandProcessor; @@ -27,6 +25,7 @@ import com.hazelcast.internal.ascii.rest.HttpPostCommand; import com.hazelcast.internal.ascii.rest.HttpPostCommandProcessor; import com.hazelcast.internal.json.JsonObject; +import static com.hazelcast.internal.ascii.rest.HttpStatusCode.SC_400; import static com.hazelcast.internal.ascii.rest.HttpStatusCode.SC_500; public class Log4j2HttpPostCommandProcessor extends HttpCommandProcessor<HttpPostCommand> { @@ -72,19 +71,31 @@ public class Log4j2HttpPostCommandProcessor extends HttpCommandProcessor<HttpPos * <p>Request Body(application/text): * * <p>your_username&your_password&com.example.logger1&ERROR + * + * <p>An unknown level name is rejected with {@code 400} instead of being applied as a {@code + * null} level, which log4j2 silently ignores. */ @SuppressWarnings("MagicNumber") private void setLoggerLevel(HttpPostCommand request) { try { String[] params = decodeParamsAndAuthenticate(request, 4); String logger = params[2]; - String level = params[3]; - if (LoggerConfig.ROOT.equals(logger)) { - Configurator.setRootLevel(Level.getLevel(level)); + String levelName = params[3]; + Level level = LogLevels.parse(levelName); + if (logger == null || logger.trim().isEmpty()) { + prepareResponse(SC_400, request, "Logger name is required!"); + } else if (level == null) { + prepareResponse( + SC_400, + request, + "Unknown logger level '" + + levelName + + "', valid levels are: " + + LogLevels.validNames()); } else { - Configurator.setLevel(logger, Level.getLevel(level)); + LogLevels.apply(logger, level); + prepareResponse(request, new JsonObject().add("status", "SUCCESS")); } - prepareResponse(request, new JsonObject().add("status", "SUCCESS")); } catch (Throwable e) { prepareResponse(SC_500, request, exceptionResponse(e)); } diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/log/LogLevels.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/log/LogLevels.java new file mode 100644 index 0000000000..260cf4dc77 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/log/LogLevels.java @@ -0,0 +1,241 @@ +/* + * 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.seatunnel.engine.server.log; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LoggerContext; +import org.apache.logging.log4j.core.config.Configuration; +import org.apache.logging.log4j.core.config.Configurator; +import org.apache.logging.log4j.core.config.LoggerConfig; + +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** Parsing, applying and reverting of runtime log levels, shared by the log level endpoints. */ +public final class LogLevels { + + /** The level of the logger comes from the log4j2 configuration file. */ + public static final String ORIGIN_FILE = "file"; + + /** The level was set through a log level endpoint and is lost when the node restarts. */ + public static final String ORIGIN_RUNTIME_OVERRIDE = "runtime-override"; + + /** + * State every overridden logger had before its first runtime override, so that {@link + * #reset(String)} can put it back and so that the endpoints can tell where the current level + * comes from. + */ + private static final Map<String, OriginalState> OVERRIDDEN_LEVELS = new ConcurrentHashMap<>(); + + private LogLevels() {} + + /** + * Resolves a level name to a log4j2 {@link Level}, accepting any letter case, or returns {@code + * null} when no such level is registered. Callers must reject {@code null} instead of handing + * it to {@link Configurator}: a {@code null} level does not leave the logger alone, it removes + * the explicit level so the logger falls back to its parent, and the root logger falls back to + * {@code ERROR}. Silently lowering a level is worse than answering with an error. + */ + public static Level parse(String name) { + if (name == null) { + return null; + } + String trimmed = name.trim(); + if (trimmed.isEmpty()) { + return null; + } + Level level = Level.getLevel(trimmed); + if (level == null) { + level = Level.getLevel(trimmed.toUpperCase(Locale.ROOT)); + } + return level; + } + + /** + * Names of all levels currently registered in log4j2, most severe first. {@link Level#values()} + * reads a hash map, so the natural order of {@link Level} is applied to keep the message of a + * rejected request stable. + */ + public static String validNames() { + return Stream.of(Level.values()) + .sorted() + .map(Level::name) + .collect(Collectors.joining(", ")); + } + + /** + * Applies a level to one logger, or to the root logger for {@link LoggerConfig#ROOT}, and + * returns the effective level it replaced. The level is read and replaced while holding the + * monitor of this class, so two requests that change the same logger at the same time can not + * both report the level that was in place before either of them ran. + */ + public static synchronized Level apply(String logger, Level level) { + String name = trim(logger); + Level previousLevel = effectiveLevel(name); + OVERRIDDEN_LEVELS.computeIfAbsent(name, OriginalState::of); + if (LoggerConfig.ROOT.equals(name)) { + Configurator.setRootLevel(level); + } else { + Configurator.setLevel(name, level); + } + return previousLevel; + } + + /** + * Reverts a logger to the state it had before its first runtime override. A logger that was not + * configured at all loses the configuration the override added and inherits from its parent + * again. Like {@link #apply(String, Level)} this runs under the monitor of this class, so the + * returned levels always belong to the same revert. + */ + public static synchronized Reverted reset(String logger) { + String name = trim(logger); + Level previousLevel = effectiveLevel(name); + OriginalState original = OVERRIDDEN_LEVELS.remove(name); + if (original == null) { + return new Reverted(false, previousLevel, previousLevel); + } + if (LoggerConfig.ROOT.equals(name)) { + // the root logger is always configured, the fallback is only defensive + Configurator.setRootLevel(original.level == null ? Level.INFO : original.level); + } else if (original.configured) { + Configurator.setLevel(name, original.level); + } else { + LoggerContext context = LoggerContext.getContext(false); + context.getConfiguration().removeLogger(name); + context.updateLoggers(); + } + return new Reverted(true, previousLevel, effectiveLevel(name)); + } + + /** Whether the current level of a logger comes from a runtime override. */ + public static boolean isOverridden(String logger) { + return OVERRIDDEN_LEVELS.containsKey(trim(logger)); + } + + /** Either {@link #ORIGIN_FILE} or {@link #ORIGIN_RUNTIME_OVERRIDE} for one logger. */ + public static String origin(String logger) { + return isOverridden(logger) ? ORIGIN_RUNTIME_OVERRIDE : ORIGIN_FILE; + } + + /** + * The level a logger was configured with before its first runtime override, {@code null} when + * it was never overridden. + */ + public static Level levelBeforeOverride(String logger) { + OriginalState original = OVERRIDDEN_LEVELS.get(trim(logger)); + return original == null ? null : original.level; + } + + /** + * Effective level of a logger name, resolved through the closest configured ancestor, so that + * names which are not configured themselves can be asked about as well. + */ + public static Level effectiveLevel(String logger) { + return configuration().getLoggerConfig(configurationName(trim(logger))).getLevel(); + } + + /** + * Effective level of every logger of the running configuration, keyed by the name the endpoints + * use ({@link LoggerConfig#ROOT} for the root logger), in configuration order. + */ + public static Map<String, Level> loggers() { + Map<String, Level> loggers = new LinkedHashMap<>(); + configuration() + .getLoggers() + .forEach((name, config) -> loggers.put(endpointName(name), config.getLevel())); + return loggers; + } + + private static Configuration configuration() { + return LoggerContext.getContext(false).getConfiguration(); + } + + /** The root logger is named {@code root} on the endpoints and {@code ""} inside log4j2. */ + private static String configurationName(String logger) { + return LoggerConfig.ROOT.equals(logger) ? LogManager.ROOT_LOGGER_NAME : logger; + } + + private static String endpointName(String configurationName) { + return LogManager.ROOT_LOGGER_NAME.equals(configurationName) + ? LoggerConfig.ROOT + : configurationName; + } + + private static String trim(String logger) { + return logger == null ? "" : logger.trim(); + } + + /** + * Outcome of {@link #reset(String)}: whether anything was reverted, and the levels around the + * revert. + */ + public static final class Reverted { + + private final boolean reverted; + private final Level previousLevel; + private final Level level; + + private Reverted(boolean reverted, Level previousLevel, Level level) { + this.reverted = reverted; + this.previousLevel = previousLevel; + this.level = level; + } + + /** Whether the logger was overridden through an endpoint and has now been put back. */ + public boolean isReverted() { + return reverted; + } + + /** Effective level immediately before the revert. */ + public Level getPreviousLevel() { + return previousLevel; + } + + /** Effective level after the revert, the previous level when nothing was reverted. */ + public Level getLevel() { + return level; + } + } + + /** What a logger looked like in the configuration before it was overridden. */ + private static final class OriginalState { + + /** Whether the logger had a configuration of its own, and not only an inherited one. */ + private final boolean configured; + + /** Level the logger was configured with. */ + private final Level level; + + private OriginalState(boolean configured, Level level) { + this.configured = configured; + this.level = level; + } + + private static OriginalState of(String logger) { + LoggerConfig config = configuration().getLoggers().get(configurationName(logger)); + return config == null + ? new OriginalState(false, null) + : new OriginalState(true, config.getLevel()); + } + } +} diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java index 11964f1686..883a16c5e3 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java @@ -111,6 +111,8 @@ public class RestConstant { public static final String REST_URL_LOG = "/log"; // Code internal Use , Get Node Log Name public static final String REST_URL_GET_ALL_LOG_NAME = "/get-all-log-name"; + // Read and change runtime log levels + public static final String REST_URL_LOGGERS = "/loggers"; public static final String REST_URL_METRICS = "/metrics"; public static final String REST_URL_OPEN_METRICS = "/openmetrics"; public static final String REST_URL_TRACE_TASK_MAPPING = "/trace/task-mapping"; diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/LoggerLevelService.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/LoggerLevelService.java new file mode 100644 index 0000000000..628bc79336 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/LoggerLevelService.java @@ -0,0 +1,301 @@ +/* + * 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.seatunnel.engine.server.rest.service; + +import org.apache.seatunnel.engine.common.config.server.HttpConfig; +import org.apache.seatunnel.engine.server.SeaTunnelServer; +import org.apache.seatunnel.engine.server.log.LogLevels; +import org.apache.seatunnel.engine.server.operation.GetNodeHttpPortOperation; +import org.apache.seatunnel.engine.server.rest.RestConstant; +import org.apache.seatunnel.engine.server.utils.NodeEngineUtil; + +import org.apache.logging.log4j.Level; + +import com.hazelcast.cluster.Member; +import com.hazelcast.internal.json.Json; +import com.hazelcast.internal.json.JsonArray; +import com.hazelcast.internal.json.JsonObject; +import com.hazelcast.spi.impl.NodeEngineImpl; +import lombok.extern.slf4j.Slf4j; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.Map; + +/** + * Reads and changes log levels of the node that serves the request, and optionally of every member + * of the cluster. Runtime overrides are node local and are lost when a node restarts, so the + * cluster scope is a convenience for the operator and not a replicated setting. + */ +@Slf4j +public class LoggerLevelService extends BaseService { + + public static final String SCOPE = "scope"; + public static final String SCOPE_CLUSTER = "cluster"; + public static final String SCOPE_NODE = "node"; + public static final String LEVEL = "level"; + + private static final String NODE = "node"; + private static final String NODES = "nodes"; + private static final String LOGGERS = "loggers"; + private static final String NAME = "name"; + private static final String ORIGIN = "origin"; + private static final String FILE_LEVEL = "fileLevel"; + private static final String PREVIOUS_LEVEL = "previousLevel"; + private static final String STATUS = "status"; + private static final String ERROR = "error"; + private static final String SUCCESS = "SUCCESS"; + private static final String FAILURE = "FAILURE"; + private static final String PARTIAL_FAILURE = "PARTIAL_FAILURE"; + private static final String NO_OVERRIDE = "NO_OVERRIDE"; + private static final int REQUEST_TIMEOUT_MS = 5000; + + public LoggerLevelService(NodeEngineImpl nodeEngine) { + super(nodeEngine); + } + + /** Every logger of the running configuration of this node. */ + public JsonObject loggers() { + JsonArray loggers = new JsonArray(); + for (Map.Entry<String, Level> logger : LogLevels.loggers().entrySet()) { + loggers.add(loggerJson(logger.getKey(), logger.getValue())); + } + return new JsonObject().add(NODE, nodeId()).add(LOGGERS, loggers); + } + + /** One logger of this node, resolved through its closest configured ancestor. */ + public JsonObject logger(String name) { + return loggerJson(name, LogLevels.effectiveLevel(name)).add(NODE, nodeId()); + } + + /** + * Overrides the level of one logger on this node. The level that was replaced comes from the + * change itself; the rest of the answer is read afterwards, so a change another request makes + * in between is already reflected in it. + */ + public JsonObject setLevel(String name, Level level, String client) { + Level previousLevel = LogLevels.apply(name, level); + log.info( + "Logger level changed: logger={}, {} -> {}, scope={}, client={}", + name, + previousLevel, + level, + SCOPE_NODE, + client); + return loggerJson(name, LogLevels.effectiveLevel(name)) + .add(NODE, nodeId()) + .add(PREVIOUS_LEVEL, previousLevel == null ? null : previousLevel.name()) + .add(STATUS, SUCCESS); + } + + /** Reverts one logger of this node to the level it had before the first override. */ + public JsonObject resetLevel(String name, String client) { + LogLevels.Reverted reverted = LogLevels.reset(name); + Level previousLevel = reverted.getPreviousLevel(); + if (reverted.isReverted()) { + log.info( + "Logger level reverted: logger={}, {} -> {}, scope={}, client={}", + name, + previousLevel, + reverted.getLevel(), + SCOPE_NODE, + client); + } + return loggerJson(name, reverted.getLevel()) + .add(NODE, nodeId()) + .add(PREVIOUS_LEVEL, previousLevel == null ? null : previousLevel.name()) + .add(STATUS, reverted.isReverted() ? SUCCESS : NO_OVERRIDE); + } + + /** Every logger of every member of the cluster. */ + public JsonObject clusterLoggers(String name) { + return fanOut("GET", name, null); + } + + /** Overrides the level of one logger on every member of the cluster. */ + public JsonObject clusterSetLevel(String name, Level level, String client) { + log.info( + "Logger level change requested for the whole cluster: logger={}, level={}, " + + "client={}", + name, + level, + client); + return fanOut("POST", name, level); + } + + /** Reverts one logger to its pre-override level on every member of the cluster. */ + public JsonObject clusterResetLevel(String name, String client) { + log.info( + "Logger level revert requested for the whole cluster: logger={}, client={}", + name, + client); + return fanOut("DELETE", name, null); + } + + /** + * Runs the node local endpoint of every member and collects the answers. Members are + * independent of each other, so a member that cannot be reached does not undo the change on the + * members that were already changed; the per member status makes that visible instead. + */ + private JsonObject fanOut(String method, String name, Level level) { + HttpConfig httpConfig = httpConfig(); + JsonArray nodes = new JsonArray(); + int reached = 0; + int failed = 0; + for (Member member : nodeEngine.getClusterService().getMembers()) { + String memberId = member.getAddress().getHost(); + try { + int httpPort = + (int) + NodeEngineUtil.sendOperationToMemberNode( + nodeEngine, + new GetNodeHttpPortOperation(), + member.getAddress()) + .get(); + memberId = member.getAddress().getHost() + ":" + httpPort; + String url = + "http://"; + + member.getAddress().getHost() + + ":" + + httpPort + + httpConfig.getContextPath() + + RestConstant.REST_URL_LOGGERS + + (name == null ? "" : "/" + encode(name)) + + (level == null ? "" : "?" + LEVEL + "=" + encode(level.name())); + nodes.add(request(url, method, httpConfig)); + reached++; + } catch (Throwable t) { + log.warn("Logger level request to member {} failed", memberId, t); + nodes.add( + new JsonObject() + .add(NODE, memberId) + .add(STATUS, FAILURE) + .add(ERROR, errorMessage(t))); + failed++; + } + } + String status = failed == 0 ? SUCCESS : (reached == 0 ? FAILURE : PARTIAL_FAILURE); + JsonObject response = new JsonObject().add(SCOPE, SCOPE_CLUSTER).add(STATUS, status); + if (name != null) { + response.add(NAME, name); + } + if (level != null) { + response.add(LEVEL, level.name()); + } + return response.add(NODES, nodes); + } + + private JsonObject request(String url, String method, HttpConfig httpConfig) + throws IOException { + HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection(); + try { + connection.setRequestMethod(method); + connection.setConnectTimeout(REQUEST_TIMEOUT_MS); + connection.setReadTimeout(REQUEST_TIMEOUT_MS); + if (httpConfig.isEnableBasicAuth()) { + String credentials = + httpConfig.getBasicAuthUsername() + ":" + httpConfig.getBasicAuthPassword(); + connection.setRequestProperty( + "Authorization", + "Basic " + + Base64.getEncoder() + .encodeToString( + credentials.getBytes(StandardCharsets.UTF_8))); + } + connection.connect(); + int code = connection.getResponseCode(); + if (code != HttpURLConnection.HTTP_OK) { + throw new IOException( + "HTTP " + code + " from " + url + ": " + read(connection.getErrorStream())); + } + return Json.parse(read(connection.getInputStream())).asObject(); + } finally { + connection.disconnect(); + } + } + + private JsonObject loggerJson(String name, Level level) { + JsonObject logger = + new JsonObject() + .add(NAME, name) + .add(LEVEL, level == null ? null : level.name()) + .add(ORIGIN, LogLevels.origin(name)); + Level fileLevel = LogLevels.levelBeforeOverride(name); + if (fileLevel != null) { + logger.add(FILE_LEVEL, fileLevel.name()); + } + return logger; + } + + private HttpConfig httpConfig() { + SeaTunnelServer seaTunnelServer = getSeaTunnelServer(false); + if (seaTunnelServer == null) { + throw new IllegalStateException("SeaTunnel server is not available on this node."); + } + return seaTunnelServer.getSeaTunnelConfig().getEngineConfig().getHttpConfig(); + } + + private String nodeId() { + return nodeEngine.getThisAddress().getHost() + ":" + httpConfig().getPort(); + } + + /** + * Short reason of a failed member request, the stack trace goes to the node log instead. The + * root cause is reported, because the wrappers a failed operation collects on the way out + * ({@code CompletionException} around an {@code ExecutionException} around the real failure) + * say nothing about what went wrong. + */ + private static String errorMessage(Throwable t) { + Throwable cause = t; + while (cause.getCause() != null && cause.getCause() != cause) { + cause = cause.getCause(); + } + return cause.getMessage() == null + ? cause.getClass().getName() + : cause.getClass().getSimpleName() + ": " + cause.getMessage(); + } + + private static String encode(String value) { + try { + return URLEncoder.encode(value, StandardCharsets.UTF_8.name()); + } catch (IOException e) { + throw new IllegalArgumentException("Can not encode '" + value + "'", e); + } + } + + private static String read(InputStream stream) throws IOException { + if (stream == null) { + return ""; + } + try (InputStream input = stream; + ByteArrayOutputStream output = new ByteArrayOutputStream()) { + byte[] buffer = new byte[4096]; + int length; + while ((length = input.read(buffer)) != -1) { + output.write(buffer, 0, length); + } + return output.toString(StandardCharsets.UTF_8.name()); + } + } +} diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/LoggersServlet.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/LoggersServlet.java new file mode 100644 index 0000000000..bc695b47ec --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/LoggersServlet.java @@ -0,0 +1,186 @@ +/* + * 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.seatunnel.engine.server.rest.servlet; + +import org.apache.seatunnel.engine.server.log.LogLevels; +import org.apache.seatunnel.engine.server.rest.service.LoggerLevelService; + +import org.apache.logging.log4j.Level; + +import com.hazelcast.internal.json.Json; +import com.hazelcast.internal.json.JsonObject; +import com.hazelcast.internal.json.JsonValue; +import com.hazelcast.spi.impl.NodeEngineImpl; + +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; + +/** + * Reads and changes runtime log levels. + * + * <p>{@code GET /loggers} lists every configured logger, {@code GET /loggers/{name}} reads one, + * {@code POST /loggers/{name}} overrides its level and {@code DELETE /loggers/{name}} reverts the + * override. {@code ?scope=cluster} runs the same request on every member of the cluster. + */ +public class LoggersServlet extends BaseServlet { + + private final LoggerLevelService loggerLevelService; + + public LoggersServlet(NodeEngineImpl nodeEngine) { + super(nodeEngine); + this.loggerLevelService = new LoggerLevelService(nodeEngine); + } + + @Override + protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException { + String name = loggerName(req); + if (!isScopeValid(req, resp)) { + return; + } + if (isClusterRequested(req)) { + writeJson(resp, loggerLevelService.clusterLoggers(name)); + } else if (name == null) { + writeJson(resp, loggerLevelService.loggers()); + } else { + writeJson(resp, loggerLevelService.logger(name)); + } + } + + @Override + protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException { + String name = loggerName(req); + if (name == null) { + badRequest(resp, "The logger name must not be empty, use /loggers/{name}."); + return; + } + if (!isScopeValid(req, resp)) { + return; + } + String levelName = requestedLevel(req); + if (levelName == null) { + badRequest( + resp, + "The level is required, send it as ?level=DEBUG or as {\"level\":\"DEBUG\"}."); + return; + } + Level level = LogLevels.parse(levelName); + if (level == null) { + badRequest( + resp, + "Unknown logger level '" + + levelName + + "', valid levels are: " + + LogLevels.validNames()); + return; + } + String client = req.getRemoteAddr(); + if (isClusterRequested(req)) { + writeJson(resp, loggerLevelService.clusterSetLevel(name, level, client)); + } else { + writeJson(resp, loggerLevelService.setLevel(name, level, client)); + } + } + + @Override + protected void doDelete(HttpServletRequest req, HttpServletResponse resp) throws IOException { + String name = loggerName(req); + if (name == null) { + badRequest(resp, "The logger name must not be empty, use /loggers/{name}."); + return; + } + if (!isScopeValid(req, resp)) { + return; + } + String client = req.getRemoteAddr(); + if (isClusterRequested(req)) { + writeJson(resp, loggerLevelService.clusterResetLevel(name, client)); + } else { + writeJson(resp, loggerLevelService.resetLevel(name, client)); + } + } + + /** The logger name of the request path, {@code null} when the whole list is addressed. */ + private String loggerName(HttpServletRequest req) { + String pathInfo = req.getPathInfo(); + if (pathInfo == null) { + return null; + } + String name = pathInfo.startsWith("/") ? pathInfo.substring(1) : pathInfo; + name = name.trim(); + return name.isEmpty() ? null : name; + } + + /** + * The level of the {@code level} query parameter, falling back to the {@code level} field of + * the request body, or {@code null} when neither carries one. + */ + private String requestedLevel(HttpServletRequest req) throws IOException { + String level = req.getParameter(LoggerLevelService.LEVEL); + if (level != null) { + return level; + } + byte[] body = requestBody(req); + if (body.length == 0) { + return null; + } + try { + JsonValue json = Json.parse(new String(body, StandardCharsets.UTF_8)); + return json.isObject() + ? json.asObject().getString(LoggerLevelService.LEVEL, null) + : null; + } catch (RuntimeException e) { + // an unparsable body is reported as a missing level instead of a server error + return null; + } + } + + private boolean isClusterRequested(HttpServletRequest req) { + return LoggerLevelService.SCOPE_CLUSTER.equalsIgnoreCase( + req.getParameter(LoggerLevelService.SCOPE)); + } + + /** Rejects an unknown scope and returns whether the request may go on. */ + private boolean isScopeValid(HttpServletRequest req, HttpServletResponse resp) + throws IOException { + String scope = req.getParameter(LoggerLevelService.SCOPE); + if (scope == null + || LoggerLevelService.SCOPE_CLUSTER.equalsIgnoreCase(scope) + || LoggerLevelService.SCOPE_NODE.equalsIgnoreCase(scope)) { + return true; + } + badRequest( + resp, + "Unknown scope '" + + scope + + "', valid scopes are: " + + LoggerLevelService.SCOPE_NODE + + ", " + + LoggerLevelService.SCOPE_CLUSTER); + return false; + } + + private void badRequest(HttpServletResponse resp, String message) throws IOException { + writeJson( + resp, + new JsonObject().add("status", "fail").add("message", message), + HttpServletResponse.SC_BAD_REQUEST); + } +} diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/log/LogLevelsTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/log/LogLevelsTest.java new file mode 100644 index 0000000000..2b0e90fe45 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/log/LogLevelsTest.java @@ -0,0 +1,209 @@ +/* + * 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.seatunnel.engine.server.log; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.config.Configurator; +import org.apache.logging.log4j.core.config.LoggerConfig; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +public class LogLevelsTest { + + /** A logger name owned by this test only, so mutating its level cannot affect other tests. */ + private static final String TEST_LOGGER = + "org.apache.seatunnel.engine.server.log.LogLevelsTest"; + + @AfterEach + void revertOverrides() { + LogLevels.reset(TEST_LOGGER); + } + + @ParameterizedTest + @ValueSource(strings = {"DEBUG", "debug", "Debug", " DEBUG ", "\tdebug\n"}) + void testLevelNameIsParsedCaseInsensitively(String name) { + Assertions.assertEquals(Level.DEBUG, LogLevels.parse(name)); + } + + @ParameterizedTest + @ValueSource(strings = {"DEBUGG", "verbose", "1", "INFO,DEBUG"}) + void testUnknownLevelNameIsRejected(String name) { + // Level.getLevel returns null instead of throwing, so an unchecked value would reach + // Configurator as a null level, which clears the level of the logger instead of leaving it + // alone while the caller is told the request succeeded + Assertions.assertNull(LogLevels.parse(name)); + } + + @ParameterizedTest + @ValueSource(strings = {"", " ", "\t"}) + void testBlankLevelNameIsRejected(String name) { + Assertions.assertNull(LogLevels.parse(name)); + } + + @Test + void testNullLevelNameIsRejected() { + Assertions.assertNull(LogLevels.parse(null)); + } + + @Test + void testValidNamesListsEveryStandardLevel() { + String validNames = LogLevels.validNames(); + for (Level level : Level.values()) { + Assertions.assertTrue( + validNames.contains(level.name()), + () -> "Level " + level.name() + " missing from: " + validNames); + } + } + + @Test + void testValidNamesAreOrderedBySeverity() { + // Level.values() reads a hash map, so without an explicit order the same rejected request + // would list the valid levels differently between runs + String validNames = LogLevels.validNames(); + List<Level> bySeverity = + Arrays.asList( + Level.OFF, + Level.FATAL, + Level.ERROR, + Level.WARN, + Level.INFO, + Level.DEBUG, + Level.TRACE, + Level.ALL); + int previousIndex = -1; + for (Level level : bySeverity) { + int index = validNames.indexOf(level.name()); + Assertions.assertTrue( + index > previousIndex, + () -> level.name() + " is out of severity order in: " + validNames); + previousIndex = index; + } + } + + @Test + void testApplyChangesTheEffectiveLevel() { + Level originalLevel = LogManager.getLogger(TEST_LOGGER).getLevel(); + Level target = originalLevel == Level.TRACE ? Level.ERROR : Level.TRACE; + + Level replaced = LogLevels.apply(TEST_LOGGER, target); + + Assertions.assertEquals(target, LogManager.getLogger(TEST_LOGGER).getLevel()); + // the level that was replaced is reported by the change itself, so an audit line and a + // previousLevel field can not be built from a level a concurrent change already replaced + Assertions.assertEquals(originalLevel, replaced); + } + + @Test + void testOverrideOfAnUnconfiguredLoggerIsTrackedAndReverted() { + String logger = TEST_LOGGER + ".unconfigured"; + Level inherited = LogLevels.effectiveLevel(logger); + Assertions.assertEquals(LogLevels.ORIGIN_FILE, LogLevels.origin(logger)); + + LogLevels.apply(logger, Level.TRACE); + + Assertions.assertEquals(Level.TRACE, LogLevels.effectiveLevel(logger)); + Assertions.assertEquals(LogLevels.ORIGIN_RUNTIME_OVERRIDE, LogLevels.origin(logger)); + // the logger had no configuration of its own, so there is no file level to report + Assertions.assertNull(LogLevels.levelBeforeOverride(logger)); + + LogLevels.Reverted reverted = LogLevels.reset(logger); + + Assertions.assertTrue(reverted.isReverted()); + Assertions.assertEquals(Level.TRACE, reverted.getPreviousLevel()); + Assertions.assertEquals(inherited, reverted.getLevel()); + Assertions.assertEquals(inherited, LogLevels.effectiveLevel(logger)); + Assertions.assertEquals(LogLevels.ORIGIN_FILE, LogLevels.origin(logger)); + Assertions.assertFalse(LogLevels.isOverridden(logger)); + } + + @Test + void testConfiguredLevelIsRememberedAndRestored() { + String logger = TEST_LOGGER + ".configured"; + // stands in for a logger that log4j2.properties configures with a level of its own + Configurator.setLevel(logger, Level.WARN); + + LogLevels.apply(logger, Level.DEBUG); + + Assertions.assertEquals(Level.DEBUG, LogLevels.effectiveLevel(logger)); + Assertions.assertEquals(Level.WARN, LogLevels.levelBeforeOverride(logger)); + + // a second override must not forget the level the file configured + LogLevels.apply(logger, Level.TRACE); + Assertions.assertEquals(Level.WARN, LogLevels.levelBeforeOverride(logger)); + + LogLevels.Reverted reverted = LogLevels.reset(logger); + + Assertions.assertTrue(reverted.isReverted()); + Assertions.assertEquals(Level.TRACE, reverted.getPreviousLevel()); + Assertions.assertEquals(Level.WARN, reverted.getLevel()); + Assertions.assertEquals(Level.WARN, LogLevels.effectiveLevel(logger)); + Assertions.assertEquals(LogLevels.ORIGIN_FILE, LogLevels.origin(logger)); + Assertions.assertNull(LogLevels.levelBeforeOverride(logger)); + } + + @Test + void testResetOfALoggerThatWasNeverOverriddenReportsNoChange() { + String logger = TEST_LOGGER + ".neverTouched"; + Level inherited = LogLevels.effectiveLevel(logger); + + LogLevels.Reverted reverted = LogLevels.reset(logger); + + Assertions.assertFalse(reverted.isReverted()); + // nothing changed, so both levels are the one the logger already had + Assertions.assertEquals(inherited, reverted.getPreviousLevel()); + Assertions.assertEquals(inherited, reverted.getLevel()); + } + + @Test + void testRootLoggerCanBeOverriddenAndReverted() { + Level configuredRootLevel = LogLevels.effectiveLevel(LoggerConfig.ROOT); + Level target = configuredRootLevel == Level.TRACE ? Level.ERROR : Level.TRACE; + try { + LogLevels.apply(LoggerConfig.ROOT, target); + + Assertions.assertEquals(target, LogLevels.effectiveLevel(LoggerConfig.ROOT)); + Assertions.assertEquals( + LogLevels.ORIGIN_RUNTIME_OVERRIDE, LogLevels.origin(LoggerConfig.ROOT)); + Assertions.assertEquals( + configuredRootLevel, LogLevels.levelBeforeOverride(LoggerConfig.ROOT)); + } finally { + Assertions.assertTrue(LogLevels.reset(LoggerConfig.ROOT).isReverted()); + } + Assertions.assertEquals(configuredRootLevel, LogLevels.effectiveLevel(LoggerConfig.ROOT)); + } + + @Test + void testLoggersListReportsTheRootLoggerUnderItsEndpointName() { + Map<String, Level> loggers = LogLevels.loggers(); + + Assertions.assertTrue( + loggers.containsKey(LoggerConfig.ROOT), + () -> "root logger missing from: " + loggers.keySet()); + Assertions.assertFalse(loggers.containsKey(LogManager.ROOT_LOGGER_NAME)); + Assertions.assertNotNull(loggers.get(LoggerConfig.ROOT)); + } +}
