This is an automated email from the ASF dual-hosted git repository. yuluo-yx pushed a commit to branch 0712-yuluo/fix-sql in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
commit 165c78de09cf7b43fc0cafbae205aa91aa4158b8 Author: yuluo-yx <[email protected]> AuthorDate: Sun Jul 12 22:22:57 2026 +0800 fix: fix mcp log sql error Signed-off-by: yuluo-yx <[email protected]> --- mcp-servers/mcp-log-server/README.md | 63 ++++++- .../mcp/log/server/service/LogService.java | 182 ++++++++++++++----- .../mcp/server/service/LogServiceTest.java | 201 +++++++++++++++++++++ 3 files changed, 397 insertions(+), 49 deletions(-) diff --git a/mcp-servers/mcp-log-server/README.md b/mcp-servers/mcp-log-server/README.md index 89bac97f12..b353142df1 100644 --- a/mcp-servers/mcp-log-server/README.md +++ b/mcp-servers/mcp-log-server/README.md @@ -1,12 +1,50 @@ -# HertzBeat MCP +# HertzBeat 日志 MCP -## HertzBeat Log MCP +该 MCP 服务基于 GreptimeDB 查询 HertzBeat 自身运行日志。使用前需要启用 GreptimeDB 日志写入。 -Log MCP Service Based on GreptimeDB. +## 结构化只读查询 -- GreptimeDB log writing needs to be enabled. +服务只暴露 `query_logs`，不再接收或执行调用方提供的 SQL。服务端生成的查询固定为： -## Claude Desktop Integration (stdio) +```sql +SELECT timestamp, severity_text, body +FROM hzb_logs +WHERE <结构化过滤条件> +ORDER BY timestamp DESC +LIMIT <1-100> +``` + +`query_logs` 支持以下可选参数： + +| 参数 | 类型 | 说明 | +| --- | --- | --- | +| `severity` | 字符串 | `TRACE`、`DEBUG`、`INFO`、`WARN`、`ERROR` 或 `FATAL` | +| `keyword` | 字符串 | 日志正文关键词，最长 256 个字符 | +| `startTime` | 整数 | 开始时间，Unix 毫秒时间戳 | +| `endTime` | 整数 | 结束时间，Unix 毫秒时间戳 | +| `limit` | 整数 | 返回条数，默认为 20，范围为 1～100 | + +调用示例： + +```json +{ + "severity": "ERROR", + "keyword": "connection refused", + "startTime": 1783785600000, + "endTime": 1783872000000, + "limit": 20 +} +``` + +当前 `hzb_logs` 表没有 `monitorId` 字段，因此本接口不提供无效的监控 ID 过滤。如果后续需要该能力，应先在 OpenTelemetry 日志写入链中定义并提取统一的监控 ID 字段。 + +## GreptimeDB 账号 + +服务支持通过 `greptime.username` 和 `greptime.password` 发送 HTTP Basic Authentication。用户名和密码必须同时配置；建议配合 HTTPS 或可信内网使用。 + +项目当前 Docker Compose 使用 GreptimeDB `v0.14.3`，该版本只提供身份认证，不能限制用户为只读权限。因此当前真正生效的安全边界是“删除原始 SQL参数并固定生成单条 `SELECT`”。使用 GreptimeDB 1.0 及以上版本时，应为该 MCP 配置独立的 `ro`/`readonly` 账号。 + +## Claude Desktop 集成（stdio） ```json { @@ -17,11 +55,20 @@ Log MCP Service Based on GreptimeDB. "-Dspring.ai.mcp.server.stdio=true", "-Dspring.main.web-application-type=none", "-Dlogging.pattern.console=", - "-Dgreptime.url=http://${IP}:4000";, "-jar", "${PATH}/hertzbeat-mcp-2.0-SNAPSHOT.jar" - ] + ], + "env": { + "GREPTIME_URL": "http://${IP}:4000";, + "GREPTIME_DATABASE": "public", + "GREPTIME_USERNAME": "${READ_ONLY_USERNAME}", + "GREPTIME_PASSWORD": "${READ_ONLY_PASSWORD}" + } } } } -``` \ No newline at end of file +``` + +不启用 GreptimeDB 认证时，可以省略 `GREPTIME_USERNAME` 和 `GREPTIME_PASSWORD`；不能只配置其中一个。 + +旧的 `getHertzbeatLog({"querySql": "..."})` Tool 已直接移除，不保留兼容入口，以免继续暴露任意 SQL执行能力。 diff --git a/mcp-servers/mcp-log-server/src/main/java/org/apache/hertzbeat/mcp/log/server/service/LogService.java b/mcp-servers/mcp-log-server/src/main/java/org/apache/hertzbeat/mcp/log/server/service/LogService.java index c5023dc106..64ec99889f 100644 --- a/mcp-servers/mcp-log-server/src/main/java/org/apache/hertzbeat/mcp/log/server/service/LogService.java +++ b/mcp-servers/mcp-log-server/src/main/java/org/apache/hertzbeat/mcp/log/server/service/LogService.java @@ -22,7 +22,9 @@ import com.jayway.jsonpath.ReadContext; import lombok.extern.slf4j.Slf4j; import org.springframework.ai.tool.annotation.Tool; import org.springframework.ai.tool.annotation.ToolParam; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.stereotype.Service; import org.springframework.util.LinkedMultiValueMap; @@ -33,79 +35,177 @@ import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneId; import java.time.format.DateTimeFormatter; +import java.util.ArrayList; import java.util.List; +import java.util.Locale; import java.util.Map; +import java.util.Set; /** - * Log query service + * 日志查询服务。 */ @Service @Slf4j public class LogService { + private static final String BASE_QUERY = "SELECT timestamp, severity_text, body FROM hzb_logs"; private static final String TIMESTAMP_COLUMN = "timestamp"; private static final String SEVERITY_TEXT_COLUMN = "severity_text"; private static final String BODY_COLUMN = "body"; + private static final int DEFAULT_LIMIT = 20; + private static final int MAX_LIMIT = 100; + private static final int MAX_KEYWORD_LENGTH = 256; + private static final long NANOS_PER_MILLISECOND = 1_000_000L; + private static final Set<String> SUPPORTED_SEVERITIES = + Set.of("TRACE", "DEBUG", "INFO", "WARN", "ERROR", "FATAL"); private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); - + private final RestClient restClient; + private final String database; - public LogService(@Value("${greptime.url}") String greptimeUrl) { - this.restClient = RestClient.builder() - .baseUrl(greptimeUrl) - .defaultHeader("Accept", "application/json") - .defaultHeader("Content-Type", "application/x-www-form-urlencoded") - .build(); + /** + * 创建日志查询服务。 + * + * @param greptimeUrl GreptimeDB 地址 + * @param database GreptimeDB 数据库名 + * @param username GreptimeDB 用户名 + * @param password GreptimeDB 密码 + */ + @Autowired + public LogService(@Value("${greptime.url}") String greptimeUrl, + @Value("${greptime.database:public}") String database, + @Value("${greptime.username:}") String username, + @Value("${greptime.password:}") String password) { + this(RestClient.builder(), greptimeUrl, database, username, password); } - @Tool(description = "System log query tool that supports filtering by time, log level, and content") - public String getHertzbeatLog( - @ToolParam(description = """ - Query system logs with support for filtering by time, log level, and content. - - Usage: - 1. Table name: hzb_log - 2. Common query examples: - - Get latest 10 logs: SELECT * FROM hzb_log ORDER BY timestamp DESC LIMIT 10 - - Query ERROR level logs: SELECT * FROM hzb_log WHERE severity_number=17 - - Query specific time range: SELECT * FROM hzb_log WHERE timestamp > '2024-01-01 00:00:00' - - Query last ten minutes: SELECT * FROM hzb_log WHERE timestamp >= now() - Interval '10m' AND timestamp < now() - - Query logs at or above the INFO level: SELECT * FROM hzb_log WHERE severity_number >= 9 - - Field descriptions: - 1. severity_number (log level): - - 5: DEBUG - - 9: INFO - - 13: WARN - - 17: ERROR - 2. timestamp: log timestamp - 3. body: log content - """) String querySql) { - - if (!isValidQuery(querySql)) { - return "Invalid query statement"; + LogService(RestClient.Builder restClientBuilder, String greptimeUrl, String database, + String username, String password) { + boolean hasUsername = username != null && !username.isBlank(); + boolean hasPassword = password != null && !password.isBlank(); + if (hasUsername != hasPassword) { + throw new IllegalArgumentException("GreptimeDB 用户名和密码必须同时配置"); + } + + RestClient.Builder builder = restClientBuilder + .baseUrl(greptimeUrl) + .defaultHeader(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE) + .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_FORM_URLENCODED_VALUE); + if (hasUsername) { + builder.defaultHeaders(headers -> headers.setBasicAuth(username, password)); } + this.restClient = builder.build(); + this.database = database == null || database.isBlank() ? "public" : database; + } + /** + * 使用结构化条件查询 HertzBeat 系统日志。 + * + * @param severity 日志级别 + * @param keyword 日志正文关键词 + * @param startTime 开始时间，Unix 毫秒时间戳 + * @param endTime 结束时间，Unix 毫秒时间戳 + * @param limit 最大返回条数 + * @return 格式化后的日志查询结果 + */ + @Tool(name = "query_logs", description = "使用结构化过滤条件只读查询 HertzBeat 系统日志") + public String queryLogs( + @ToolParam(required = false, + description = "日志级别，可选值为 TRACE、DEBUG、INFO、WARN、ERROR、FATAL") + String severity, + @ToolParam(required = false, description = "日志正文关键词，最长 256 个字符") + String keyword, + @ToolParam(required = false, description = "开始时间，Unix 毫秒时间戳") + Long startTime, + @ToolParam(required = false, description = "结束时间，Unix 毫秒时间戳") + Long endTime, + @ToolParam(required = false, description = "最多返回条数，默认为 20，最大为 100") + Integer limit) { try { - String response = executeQuery(querySql); + String response = executeQuery(buildQuery(severity, keyword, startTime, endTime, limit)); return formatQueryResults(response); + } catch (IllegalArgumentException e) { + return "查询参数无效：" + e.getMessage(); } catch (Exception e) { log.error("Failed to query logs", e); - return "Failed to query logs: " + e.getMessage(); + return "日志查询失败"; + } + } + + static String buildQuery(String severity, String keyword, Long startTime, Long endTime, Integer limit) { + if (startTime != null && startTime < 0) { + throw new IllegalArgumentException("开始时间不能为负数"); + } + if (endTime != null && endTime < 0) { + throw new IllegalArgumentException("结束时间不能为负数"); + } + if (startTime != null && endTime != null && startTime > endTime) { + throw new IllegalArgumentException("开始时间不能晚于结束时间"); + } + + int queryLimit = limit == null ? DEFAULT_LIMIT : limit; + if (queryLimit < 1 || queryLimit > MAX_LIMIT) { + throw new IllegalArgumentException("返回条数必须在 1 到 100 之间"); + } + + List<String> conditions = new ArrayList<>(4); + if (startTime != null) { + conditions.add("timestamp >= " + toNanoseconds(startTime, "开始时间")); + } + if (endTime != null) { + conditions.add("timestamp <= " + toNanoseconds(endTime, "结束时间")); + } + + String normalizedSeverity = normalizeSeverity(severity); + if (normalizedSeverity != null) { + conditions.add("severity_text = '" + normalizedSeverity + "'"); + } + + if (keyword != null && !keyword.isBlank()) { + String normalizedKeyword = keyword.strip(); + if (normalizedKeyword.length() > MAX_KEYWORD_LENGTH) { + throw new IllegalArgumentException("日志关键词不能超过 256 个字符"); + } + conditions.add("matches_term(body, '" + escapeSqlLiteral(normalizedKeyword) + "')"); + } + + StringBuilder query = new StringBuilder(BASE_QUERY); + if (!conditions.isEmpty()) { + query.append(" WHERE ").append(String.join(" AND ", conditions)); + } + return query.append(" ORDER BY timestamp DESC LIMIT ").append(queryLimit).toString(); + } + + private static String normalizeSeverity(String severity) { + if (severity == null || severity.isBlank()) { + return null; + } + String normalizedSeverity = severity.strip().toUpperCase(Locale.ROOT); + if (!SUPPORTED_SEVERITIES.contains(normalizedSeverity)) { + throw new IllegalArgumentException("日志级别不受支持"); + } + return normalizedSeverity; + } + + private static long toNanoseconds(long epochMilliseconds, String fieldName) { + try { + return Math.multiplyExact(epochMilliseconds, NANOS_PER_MILLISECOND); + } catch (ArithmeticException e) { + throw new IllegalArgumentException(fieldName + "超出支持范围", e); } } - private boolean isValidQuery(String sql) { - return sql != null && sql.toLowerCase().contains("hzb_log"); + private static String escapeSqlLiteral(String value) { + return value.replace("'", "''"); } private String executeQuery(String sql) { MultiValueMap<String, String> formData = new LinkedMultiValueMap<>(); formData.add("sql", sql); - log.debug("Executing SQL query: {}", sql); - + log.debug("Executing structured log query"); + return restClient.post() - .uri("/v1/sql?db=public") + .uri(uriBuilder -> uriBuilder.path("/v1/sql").queryParam("db", database).build()) .contentType(MediaType.APPLICATION_FORM_URLENCODED) .body(formData) .retrieve() diff --git a/mcp-servers/mcp-log-server/src/test/java/org/apache/hertzbeat/mcp/server/service/LogServiceTest.java b/mcp-servers/mcp-log-server/src/test/java/org/apache/hertzbeat/mcp/server/service/LogServiceTest.java new file mode 100644 index 0000000000..bdfeedec96 --- /dev/null +++ b/mcp-servers/mcp-log-server/src/test/java/org/apache/hertzbeat/mcp/server/service/LogServiceTest.java @@ -0,0 +1,201 @@ +/* + * 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.mcp.server.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.springframework.test.web.client.match.MockRestRequestMatchers.content; +import static org.springframework.test.web.client.match.MockRestRequestMatchers.header; +import static org.springframework.test.web.client.match.MockRestRequestMatchers.method; +import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo; +import static org.springframework.test.web.client.response.MockRestResponseCreators.withServerError; +import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess; + +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.springframework.ai.tool.ToolCallback; +import org.springframework.ai.tool.method.MethodToolCallbackProvider; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.MediaType; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; +import org.springframework.web.client.RestClient; +import org.springframework.test.web.client.MockRestServiceServer; + +/** + * {@link LogService} 安全查询测试。 + */ +class LogServiceTest { + + private static final String BASE_QUERY = "SELECT timestamp, severity_text, body FROM hzb_logs"; + + @Test + void shouldBuildDefaultReadOnlyQuery() { + assertThat(LogService.buildQuery(null, null, null, null, null)) + .isEqualTo(BASE_QUERY + " ORDER BY timestamp DESC LIMIT 20"); + } + + @Test + void shouldExposeOnlyStructuredToolParameters() { + LogService service = new LogService( + RestClient.builder(), "http://localhost:4000";, "public", "", ""); + ToolCallback[] callbacks = MethodToolCallbackProvider.builder().toolObjects(service).build().getToolCallbacks(); + + assertThat(callbacks).singleElement().satisfies(callback -> { + assertThat(callback.getToolDefinition().name()).isEqualTo("query_logs"); + assertThat(callback.getToolDefinition().inputSchema()) + .contains("severity", "keyword", "startTime", "endTime", "limit") + .doesNotContain("querySql"); + }); + } + + @Test + void shouldBuildQueryFromValidatedFilters() { + String query = LogService.buildQuery( + " error ", " x'); DELETE FROM hzb_logs; -- ", 1L, 2L, 10); + + assertThat(query).isEqualTo(BASE_QUERY + + " WHERE timestamp >= 1000000" + + " AND timestamp <= 2000000" + + " AND severity_text = 'ERROR'" + + " AND matches_term(body, 'x''); DELETE FROM hzb_logs; --')" + + " ORDER BY timestamp DESC LIMIT 10"); + } + + @Test + void shouldRejectInvalidFilters() { + assertThatIllegalArgumentException() + .isThrownBy(() -> LogService.buildQuery("NOTICE", null, null, null, null)); + assertThatIllegalArgumentException() + .isThrownBy(() -> LogService.buildQuery(null, null, -1L, null, null)); + assertThatIllegalArgumentException() + .isThrownBy(() -> LogService.buildQuery(null, null, null, -1L, null)); + assertThatIllegalArgumentException() + .isThrownBy(() -> LogService.buildQuery(null, null, 2L, 1L, null)); + assertThatIllegalArgumentException() + .isThrownBy(() -> LogService.buildQuery(null, null, Long.MAX_VALUE, null, null)); + assertThatIllegalArgumentException() + .isThrownBy(() -> LogService.buildQuery(null, "x".repeat(257), null, null, null)); + assertThatIllegalArgumentException() + .isThrownBy(() -> LogService.buildQuery(null, null, null, null, 0)); + assertThatIllegalArgumentException() + .isThrownBy(() -> LogService.buildQuery(null, null, null, null, 101)); + } + + @Test + void shouldSendGeneratedSelectWithBasicAuthentication() { + RestClient.Builder restClientBuilder = RestClient.builder(); + MockRestServiceServer server = MockRestServiceServer.bindTo(restClientBuilder).build(); + LogService service = new LogService( + restClientBuilder, "http://localhost:4000";, "observability", "reader", "secret"); + String query = BASE_QUERY + " ORDER BY timestamp DESC LIMIT 20"; + MultiValueMap<String, String> expectedForm = new LinkedMultiValueMap<>(); + expectedForm.setAll(Map.of("sql", query)); + String credentials = Base64.getEncoder() + .encodeToString("reader:secret".getBytes(StandardCharsets.UTF_8)); + + server.expect(requestTo("http://localhost:4000/v1/sql?db=observability";)) + .andExpect(method(HttpMethod.POST)) + .andExpect(header(HttpHeaders.AUTHORIZATION, "Basic " + credentials)) + .andExpect(content().formData(expectedForm)) + .andRespond(withSuccess(emptyResult(), MediaType.APPLICATION_JSON)); + + assertThat(service.queryLogs(null, null, null, null, null)).contains("No data"); + server.verify(); + } + + @Test + void shouldFormatReturnedLogs() { + RestClient.Builder restClientBuilder = RestClient.builder(); + MockRestServiceServer server = MockRestServiceServer.bindTo(restClientBuilder).build(); + LogService service = new LogService(restClientBuilder, "http://localhost:4000";, "public", "", ""); + server.expect(requestTo("http://localhost:4000/v1/sql?db=public";)) + .andRespond(withSuccess(resultWithOneLog(), MediaType.APPLICATION_JSON)); + + assertThat(service.queryLogs("ERROR", "failure", null, null, 1)) + .contains("ERROR", "failure", "Total 1 records"); + server.verify(); + } + + @Test + void shouldReturnSafeErrorMessages() { + RestClient.Builder restClientBuilder = RestClient.builder(); + MockRestServiceServer server = MockRestServiceServer.bindTo(restClientBuilder).build(); + LogService service = new LogService(restClientBuilder, "http://localhost:4000";, "public", "", ""); + + assertThat(service.queryLogs("NOTICE", null, null, null, null)) + .isEqualTo("查询参数无效：日志级别不受支持"); + + server.expect(requestTo("http://localhost:4000/v1/sql?db=public";)) + .andRespond(withServerError()); + assertThat(service.queryLogs(null, null, null, null, null)).isEqualTo("日志查询失败"); + server.verify(); + } + + @Test + void shouldRequireCompleteCredentials() { + assertThatIllegalArgumentException().isThrownBy(() -> new LogService( + RestClient.builder(), "http://localhost:4000";, "public", "reader", "")); + assertThatIllegalArgumentException().isThrownBy(() -> new LogService( + RestClient.builder(), "http://localhost:4000";, "public", "", "secret")); + } + + @Test + void shouldCreateWithProductionConstructor() { + assertThat(new LogService("http://localhost:4000";, "public", "", "")).isNotNull(); + } + + private static String emptyResult() { + return """ + { + "output": [{ + "records": { + "schema": {"column_schemas": [ + {"name": "timestamp"}, + {"name": "severity_text"}, + {"name": "body"} + ]}, + "rows": [], + "total_rows": 0 + } + }] + } + """; + } + + private static String resultWithOneLog() { + return """ + { + "output": [{ + "records": { + "schema": {"column_schemas": [ + {"name": "timestamp"}, + {"name": "severity_text"}, + {"name": "body"} + ]}, + "rows": [[0, "ERROR", "failure"]], + "total_rows": 1 + } + }] + } + """; + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
