From 8e8ab80aa0804877c1ca9074993400f59efbd2ec Mon Sep 17 00:00:00 2001 From: Yang Chen <1597081640@qq.com> Date: Wed, 3 Dec 2025 12:32:45 +0800 Subject: [PATCH] Revert "[refractor] Using Spring jdbc to query Greptime log" (#3885) --- .../log/alert/LogPeriodicAlertE2eTest.java | 15 +- .../storage/GreptimeLogStorageE2eTest.java | 23 +- .../log/controller/LogQueryController.java | 161 +++++----- .../controller/LogQueryControllerTest.java | 111 +++---- .../src/main/resources/application-test.yml | 1 - .../src/main/resources/application.yml | 1 - hertzbeat-warehouse/pom.xml | 25 +- .../db/GreptimeSqlQueryExecutor.java | 234 +++++--------- .../store/history/tsdb/HistoryDataReader.java | 34 +- .../tsdb/greptime/GreptimeDbDataStorage.java | 300 +++++++++++++----- .../tsdb/greptime/GreptimeProperties.java | 21 +- .../db/GreptimeSqlQueryExecutorTest.java | 107 ++++--- .../greptime/GreptimeDbDataStorageTest.java | 96 +++--- script/application.yml | 1 - .../conf/application.yml | 1 - .../log/log-manage/log-manage.component.html | 14 +- .../log/log-manage/log-manage.component.ts | 10 +- web-app/src/app/service/log.service.ts | 14 +- 18 files changed, 592 insertions(+), 577 deletions(-) diff --git a/hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/alert/LogPeriodicAlertE2eTest.java b/hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/alert/LogPeriodicAlertE2eTest.java index 8628d5b72f..e3c387dbe6 100644 --- a/hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/alert/LogPeriodicAlertE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/alert/LogPeriodicAlertE2eTest.java @@ -6,7 +6,7 @@ * (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 + * 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, @@ -70,7 +70,6 @@ public class LogPeriodicAlertE2eTest { private static final String GREPTIME_IMAGE = "greptime/greptimedb:latest"; private static final int GREPTIME_HTTP_PORT = 4000; private static final int GREPTIME_GRPC_PORT = 4001; - private static final int GREPTIME_PG_PORT = 4003; private static final Duration CONTAINER_STARTUP_TIMEOUT = Duration.ofSeconds(120); @LocalServerPort @@ -86,17 +85,16 @@ public class LogPeriodicAlertE2eTest { private AlarmCommonReduce alarmCommonReduce; static GenericContainer vector; - + static GenericContainer greptimedb; static { greptimedb = new GenericContainer<>(DockerImageName.parse(GREPTIME_IMAGE)) - .withExposedPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT, GREPTIME_PG_PORT) + .withExposedPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT) .withCommand("standalone", "start", "--http-addr", "0.0.0.0:" + GREPTIME_HTTP_PORT, - "--rpc-bind-addr", "0.0.0.0:" + GREPTIME_GRPC_PORT, - "--postgres-addr", "0.0.0.0:" + GREPTIME_PG_PORT) - .waitingFor(Wait.forListeningPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT, GREPTIME_PG_PORT)) + "--rpc-bind-addr", "0.0.0.0:" + GREPTIME_GRPC_PORT) + .waitingFor(Wait.forListeningPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT)) .withStartupTimeout(CONTAINER_STARTUP_TIMEOUT); greptimedb.start(); } @@ -108,7 +106,6 @@ public class LogPeriodicAlertE2eTest { r.add("warehouse.store.greptime.enabled", () -> "true"); r.add("warehouse.store.greptime.http-endpoint", () -> "http://localhost:" + greptimedb.getMappedPort(GREPTIME_HTTP_PORT)); r.add("warehouse.store.greptime.grpc-endpoints", () -> "localhost:" + greptimedb.getMappedPort(GREPTIME_GRPC_PORT)); - r.add("warehouse.store.greptime.postgres-endpoint", () -> "localhost:" + greptimedb.getMappedPort(GREPTIME_PG_PORT)); r.add("warehouse.store.greptime.username", () -> ""); r.add("warehouse.store.greptime.password", () -> ""); } @@ -230,4 +227,4 @@ public class LogPeriodicAlertE2eTest { errorLabelsByIndividual.put("type", "error_count"); errorCountAlertByIndividual.setLabels(errorLabelsByIndividual); } -} \ No newline at end of file +} diff --git a/hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/storage/GreptimeLogStorageE2eTest.java b/hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/storage/GreptimeLogStorageE2eTest.java index f491d8562f..4aba2f1eb2 100644 --- a/hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/storage/GreptimeLogStorageE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/storage/GreptimeLogStorageE2eTest.java @@ -6,7 +6,7 @@ * (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 + * 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, @@ -59,7 +59,6 @@ public class GreptimeLogStorageE2eTest { private static final String GREPTIME_IMAGE = "greptime/greptimedb:latest"; private static final int GREPTIME_HTTP_PORT = 4000; private static final int GREPTIME_GRPC_PORT = 4001; - private static final int GREPTIME_PG_PORT = 4003; private static final Duration CONTAINER_STARTUP_TIMEOUT = Duration.ofSeconds(120); @LocalServerPort @@ -76,12 +75,11 @@ public class GreptimeLogStorageE2eTest { static { greptimedb = new GenericContainer<>(DockerImageName.parse(GREPTIME_IMAGE)) - .withExposedPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT, GREPTIME_PG_PORT) + .withExposedPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT) .withCommand("standalone", "start", "--http-addr", "0.0.0.0:" + GREPTIME_HTTP_PORT, - "--rpc-bind-addr", "0.0.0.0:" + GREPTIME_GRPC_PORT, - "--postgres-addr", "0.0.0.0:" + GREPTIME_PG_PORT) - .waitingFor(Wait.forListeningPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT, GREPTIME_PG_PORT)) + "--rpc-bind-addr", "0.0.0.0:" + GREPTIME_GRPC_PORT) + .waitingFor(Wait.forListeningPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT)) .withStartupTimeout(CONTAINER_STARTUP_TIMEOUT); greptimedb.start(); } @@ -92,7 +90,6 @@ public class GreptimeLogStorageE2eTest { r.add("warehouse.store.greptime.enabled", () -> "true"); r.add("warehouse.store.greptime.http-endpoint", () -> "http://localhost:" + greptimedb.getMappedPort(GREPTIME_HTTP_PORT)); r.add("warehouse.store.greptime.grpc-endpoints", () -> "localhost:" + greptimedb.getMappedPort(GREPTIME_GRPC_PORT)); - r.add("warehouse.store.greptime.postgres-endpoint", () -> "localhost:" + greptimedb.getMappedPort(GREPTIME_PG_PORT)); r.add("warehouse.store.greptime.username", () -> ""); r.add("warehouse.store.greptime.password", () -> ""); } @@ -119,7 +116,7 @@ public class GreptimeLogStorageE2eTest { void testLogStorageToGreptimeDb() { List capturedLogs = new ArrayList<>(); - + // Wait for Vector to generate and send logs to HertzBeat await().atMost(Duration.ofSeconds(30)) .pollInterval(Duration.ofSeconds(3)) @@ -134,7 +131,7 @@ public class GreptimeLogStorageE2eTest { Thread.currentThread().interrupt(); throw new RuntimeException("Test interrupted", e); } - + // Assert that we have captured at least some logs assertFalse(capturedLogs.isEmpty(), "Should have captured at least one log entry"); }); @@ -145,7 +142,7 @@ public class GreptimeLogStorageE2eTest { assertNotNull(firstLog, "First log should not be null"); assertNotNull(firstLog.getBody(), "Log body should not be null"); assertNotNull(firstLog.getSeverityText(), "Severity text should not be null"); - + // Additional wait to ensure logs are persisted to GreptimeDB await().atMost(Duration.ofSeconds(30)) .pollInterval(Duration.ofSeconds(2)) @@ -162,8 +159,8 @@ public class GreptimeLogStorageE2eTest { private List queryStoredLogs() { long endTime = System.currentTimeMillis(); long startTime = endTime - Duration.ofMinutes(5).toMillis(); // Look back 5 minutes - + return greptimeDbDataStorage.queryLogsByMultipleConditions( - startTime, endTime, null, null, null, null, null); + startTime, endTime, null, null, null, null); } -} \ No newline at end of file +} diff --git a/hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/LogQueryController.java b/hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/LogQueryController.java index 5d0bed1f45..2de4de5ab2 100644 --- a/hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/LogQueryController.java +++ b/hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/LogQueryController.java @@ -61,54 +61,50 @@ public class LogQueryController { } @GetMapping("/list") - @Operation(summary = "Query logs by time range with optional filters", - description = "Query logs by [start,end] in ms and optional filters with pagination. Returns paginated log entries sorted by timestamp in descending order.") + @Operation(summary = "Query logs by time range with optional filters", + description = "Query logs by [start,end] in ms and optional filters with pagination. Returns paginated log entries sorted by timestamp in descending order.") public ResponseEntity>> list( - @Parameter(description = "Start timestamp in milliseconds (Unix timestamp)", example = "1640995200000") + @Parameter(description = "Start timestamp in milliseconds (Unix timestamp)", example = "1640995200000") @RequestParam(value = "start", required = false) Long start, - @Parameter(description = "End timestamp in milliseconds (Unix timestamp)", example = "1641081600000") + @Parameter(description = "End timestamp in milliseconds (Unix timestamp)", example = "1641081600000") @RequestParam(value = "end", required = false) Long end, - @Parameter(description = "Trace ID for distributed tracing", example = "1234567890abcdef") + @Parameter(description = "Trace ID for distributed tracing", example = "1234567890abcdef") @RequestParam(value = "traceId", required = false) String traceId, - @Parameter(description = "Span ID for distributed tracing", example = "abcdef1234567890") + @Parameter(description = "Span ID for distributed tracing", example = "abcdef1234567890") @RequestParam(value = "spanId", required = false) String spanId, - @Parameter(description = "Log severity number (1-24 according to OpenTelemetry standard)", example = "9") + @Parameter(description = "Log severity number (1-24 according to OpenTelemetry standard)", example = "9") @RequestParam(value = "severityNumber", required = false) Integer severityNumber, - @Parameter(description = "Log severity text (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)", example = "INFO") + @Parameter(description = "Log severity text (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)", example = "INFO") @RequestParam(value = "severityText", required = false) String severityText, - @Parameter(description = "Log content search keyword", example = "error") - @RequestParam(value = "search", required = false) String search, - @Parameter(description = "Page index starting from 0", example = "0") + @Parameter(description = "Page index starting from 0", example = "0") @RequestParam(value = "pageIndex", required = false, defaultValue = "0") Integer pageIndex, - @Parameter(description = "Number of items per page", example = "20") + @Parameter(description = "Number of items per page", example = "20") @RequestParam(value = "pageSize", required = false, defaultValue = "20") Integer pageSize) { - Page result = getPagedLogs(start, end, traceId, spanId, severityNumber, severityText, search, pageIndex, pageSize); + Page result = getPagedLogs(start, end, traceId, spanId, severityNumber, severityText, pageIndex, pageSize); return ResponseEntity.ok(Message.success(result)); } @GetMapping("/stats/overview") - @Operation(summary = "Log overview statistics", - description = "Overall counts and basic statistics with filters. Provides counts by severity levels according to OpenTelemetry standard.") + @Operation(summary = "Log overview statistics", + description = "Overall counts and basic statistics with filters. Provides counts by severity levels according to OpenTelemetry standard.") public ResponseEntity>> overviewStats( - @Parameter(description = "Start timestamp in milliseconds (Unix timestamp)", example = "1640995200000") + @Parameter(description = "Start timestamp in milliseconds (Unix timestamp)", example = "1640995200000") @RequestParam(value = "start", required = false) Long start, - @Parameter(description = "End timestamp in milliseconds (Unix timestamp)", example = "1641081600000") + @Parameter(description = "End timestamp in milliseconds (Unix timestamp)", example = "1641081600000") @RequestParam(value = "end", required = false) Long end, - @Parameter(description = "Trace ID for distributed tracing", example = "1234567890abcdef") + @Parameter(description = "Trace ID for distributed tracing", example = "1234567890abcdef") @RequestParam(value = "traceId", required = false) String traceId, - @Parameter(description = "Span ID for distributed tracing", example = "abcdef1234567890") + @Parameter(description = "Span ID for distributed tracing", example = "abcdef1234567890") @RequestParam(value = "spanId", required = false) String spanId, - @Parameter(description = "Log severity number (1-24 according to OpenTelemetry standard)", example = "9") + @Parameter(description = "Log severity number (1-24 according to OpenTelemetry standard)", example = "9") @RequestParam(value = "severityNumber", required = false) Integer severityNumber, - @Parameter(description = "Log severity text (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)", example = "INFO") - @RequestParam(value = "severityText", required = false) String severityText, - @Parameter(description = "Log content search keyword", example = "error") - @RequestParam(value = "search", required = false) String search) { - List logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText, search); - + @Parameter(description = "Log severity text (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)", example = "INFO") + @RequestParam(value = "severityText", required = false) String severityText) { + List logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText); + Map overview = new HashMap<>(); overview.put("totalCount", logs.size()); - + // Count by severity levels according to OpenTelemetry standard // TRACE: 1-4, DEBUG: 5-8, INFO: 9-12, WARN: 13-16, ERROR: 17-20, FATAL: 21-24 long fatalCount = logs.stream().filter(log -> log.getSeverityNumber() != null && log.getSeverityNumber() >= 21 && log.getSeverityNumber() <= 24).count(); @@ -117,117 +113,114 @@ public class LogQueryController { long infoCount = logs.stream().filter(log -> log.getSeverityNumber() != null && log.getSeverityNumber() >= 9 && log.getSeverityNumber() <= 12).count(); long debugCount = logs.stream().filter(log -> log.getSeverityNumber() != null && log.getSeverityNumber() >= 5 && log.getSeverityNumber() <= 8).count(); long traceCount = logs.stream().filter(log -> log.getSeverityNumber() != null && log.getSeverityNumber() >= 1 && log.getSeverityNumber() <= 4).count(); - + overview.put("fatalCount", fatalCount); overview.put("errorCount", errorCount); overview.put("warnCount", warnCount); overview.put("infoCount", infoCount); overview.put("debugCount", debugCount); overview.put("traceCount", traceCount); - + return ResponseEntity.ok(Message.success(overview)); } @GetMapping("/stats/trace-coverage") - @Operation(summary = "Trace coverage statistics", - description = "Statistics about trace information availability. Shows how many logs have trace IDs, span IDs, or both for distributed tracing analysis.") + @Operation(summary = "Trace coverage statistics", + description = "Statistics about trace information availability. Shows how many logs have trace IDs, span IDs, or both for distributed tracing analysis.") public ResponseEntity>> traceCoverageStats( - @Parameter(description = "Start timestamp in milliseconds (Unix timestamp)", example = "1640995200000") + @Parameter(description = "Start timestamp in milliseconds (Unix timestamp)", example = "1640995200000") @RequestParam(value = "start", required = false) Long start, - @Parameter(description = "End timestamp in milliseconds (Unix timestamp)", example = "1641081600000") + @Parameter(description = "End timestamp in milliseconds (Unix timestamp)", example = "1641081600000") @RequestParam(value = "end", required = false) Long end, - @Parameter(description = "Trace ID for distributed tracing", example = "1234567890abcdef") + @Parameter(description = "Trace ID for distributed tracing", example = "1234567890abcdef") @RequestParam(value = "traceId", required = false) String traceId, - @Parameter(description = "Span ID for distributed tracing", example = "abcdef1234567890") + @Parameter(description = "Span ID for distributed tracing", example = "abcdef1234567890") @RequestParam(value = "spanId", required = false) String spanId, - @Parameter(description = "Log severity number (1-24 according to OpenTelemetry standard)", example = "9") + @Parameter(description = "Log severity number (1-24 according to OpenTelemetry standard)", example = "9") @RequestParam(value = "severityNumber", required = false) Integer severityNumber, - @Parameter(description = "Log severity text (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)", example = "INFO") - @RequestParam(value = "severityText", required = false) String severityText, - @Parameter(description = "Log content search keyword", example = "error") - @RequestParam(value = "search", required = false) String search) { - List logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText, search); - + @Parameter(description = "Log severity text (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)", example = "INFO") + @RequestParam(value = "severityText", required = false) String severityText) { + List logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText); + Map result = new HashMap<>(); - + // Trace coverage statistics long withTraceId = logs.stream().filter(log -> log.getTraceId() != null && !log.getTraceId().isEmpty()).count(); long withSpanId = logs.stream().filter(log -> log.getSpanId() != null && !log.getSpanId().isEmpty()).count(); - long withBothTraceAndSpan = logs.stream().filter(log -> - log.getTraceId() != null && !log.getTraceId().isEmpty() - && log.getSpanId() != null && !log.getSpanId().isEmpty()).count(); + long withBothTraceAndSpan = logs.stream().filter(log -> + log.getTraceId() != null && !log.getTraceId().isEmpty() + && log.getSpanId() != null && !log.getSpanId().isEmpty()).count(); long withoutTrace = logs.size() - withTraceId; - + Map traceCoverage = new HashMap<>(); traceCoverage.put("withTrace", withTraceId); traceCoverage.put("withoutTrace", withoutTrace); traceCoverage.put("withSpan", withSpanId); traceCoverage.put("withBothTraceAndSpan", withBothTraceAndSpan); - + result.put("traceCoverage", traceCoverage); return ResponseEntity.ok(Message.success(result)); } @GetMapping("/stats/trend") - @Operation(summary = "Log trend over time", - description = "Count logs by hour intervals with filters. Groups logs by hour and provides time-series data for trend analysis.") + @Operation(summary = "Log trend over time", + description = "Count logs by hour intervals with filters. Groups logs by hour and provides time-series data for trend analysis.") public ResponseEntity>> trendStats( - @Parameter(description = "Start timestamp in milliseconds (Unix timestamp)", example = "1640995200000") + @Parameter(description = "Start timestamp in milliseconds (Unix timestamp)", example = "1640995200000") @RequestParam(value = "start", required = false) Long start, - @Parameter(description = "End timestamp in milliseconds (Unix timestamp)", example = "1641081600000") + @Parameter(description = "End timestamp in milliseconds (Unix timestamp)", example = "1641081600000") @RequestParam(value = "end", required = false) Long end, - @Parameter(description = "Trace ID for distributed tracing", example = "1234567890abcdef") + @Parameter(description = "Trace ID for distributed tracing", example = "1234567890abcdef") @RequestParam(value = "traceId", required = false) String traceId, - @Parameter(description = "Span ID for distributed tracing", example = "abcdef1234567890") + @Parameter(description = "Span ID for distributed tracing", example = "abcdef1234567890") @RequestParam(value = "spanId", required = false) String spanId, - @Parameter(description = "Log severity number (1-24 according to OpenTelemetry standard)", example = "9") + @Parameter(description = "Log severity number (1-24 according to OpenTelemetry standard)", example = "9") @RequestParam(value = "severityNumber", required = false) Integer severityNumber, - @Parameter(description = "Log severity text (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)", example = "INFO") - @RequestParam(value = "severityText", required = false) String severityText, - @Parameter(description = "Log content search keyword", example = "error") - @RequestParam(value = "search", required = false) String search) { - List logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText, search); - + @Parameter(description = "Log severity text (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)", example = "INFO") + @RequestParam(value = "severityText", required = false) String severityText) { + List logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText); + // Group by hour Map hourlyStats = logs.stream() - .filter(log -> log.getTimeUnixNano() != null) - .collect(Collectors.groupingBy( - log -> { - long timestampMs = log.getTimeUnixNano() / 1_000_000L; - LocalDateTime dateTime = LocalDateTime.ofInstant( - Instant.ofEpochMilli(timestampMs), - ZoneId.systemDefault()); - return dateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:00")); - }, - Collectors.counting())); - + .filter(log -> log.getTimeUnixNano() != null) + .collect(Collectors.groupingBy( + log -> { + long timestampMs = log.getTimeUnixNano() / 1_000_000L; + LocalDateTime dateTime = LocalDateTime.ofInstant( + Instant.ofEpochMilli(timestampMs), + ZoneId.systemDefault()); + return dateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:00")); + }, + Collectors.counting())); + Map result = new HashMap<>(); result.put("hourlyStats", hourlyStats); return ResponseEntity.ok(Message.success(result)); } - private List getFilteredLogs(Long start, Long end, String traceId, String spanId, - Integer severityNumber, String severityText, String search) { + private List getFilteredLogs(Long start, Long end, String traceId, String spanId, + Integer severityNumber, String severityText) { // Use the new multi-condition query method - return historyDataReader.queryLogsByMultipleConditions(start, end, traceId, spanId, severityNumber, severityText, search); + return historyDataReader.queryLogsByMultipleConditions(start, end, traceId, spanId, severityNumber, severityText); } - private Page getPagedLogs(Long start, Long end, String traceId, String spanId, - Integer severityNumber, String severityText, String search, - Integer pageIndex, Integer pageSize) { + private Page getPagedLogs(Long start, Long end, String traceId, String spanId, + Integer severityNumber, String severityText, Integer pageIndex, Integer pageSize) { // Calculate pagination parameters int offset = pageIndex * pageSize; - + // Get total count and paginated data - long totalElements = historyDataReader.countLogsByMultipleConditions(start, end, traceId, spanId, severityNumber, severityText, search); + long totalElements = historyDataReader.countLogsByMultipleConditions(start, end, traceId, spanId, severityNumber, severityText); List pagedLogs = historyDataReader.queryLogsByMultipleConditionsWithPagination( - start, end, traceId, spanId, severityNumber, severityText, search, offset, pageSize); - + start, end, traceId, spanId, severityNumber, severityText, offset, pageSize); + // Create PageRequest (sorted by timestamp descending) Sort sort = Sort.by(Sort.Direction.DESC, "timeUnixNano"); PageRequest pageRequest = PageRequest.of(pageIndex, pageSize, sort); - + // Return Spring Data Page object return new PageImpl<>(pagedLogs, pageRequest, totalElements); } -} \ No newline at end of file +} + + diff --git a/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogQueryControllerTest.java b/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogQueryControllerTest.java index 3c1afb932a..96173c2a24 100644 --- a/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogQueryControllerTest.java +++ b/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogQueryControllerTest.java @@ -6,7 +6,7 @@ * (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 + * 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, @@ -87,26 +87,24 @@ class LogQueryControllerTest { List mockLogs = Arrays.asList(logEntry1, logEntry2); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.countLogsByMultipleConditions(anyLong(), anyLong(), anyString(), - anyString(), anyInt(), anyString(), isNull())).thenReturn(2L); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.queryLogsByMultipleConditionsWithPagination(anyLong(), anyLong(), - anyString(), anyString(), anyInt(), anyString(), isNull(), anyInt(), anyInt())) + when(historyDataReader.countLogsByMultipleConditions(anyLong(), anyLong(), anyString(), + anyString(), anyInt(), anyString())).thenReturn(2L); + when(historyDataReader.queryLogsByMultipleConditionsWithPagination(anyLong(), anyLong(), + anyString(), anyString(), anyInt(), anyString(), anyInt(), anyInt())) .thenReturn(mockLogs); mockMvc.perform( - MockMvcRequestBuilders - .get("/api/logs/list") - .param("start", "1734005477000") - .param("end", "1734005478000") - .param("traceId", "trace123") - .param("spanId", "span456") - .param("severityNumber", "9") - .param("severityText", "INFO") - .param("pageIndex", "0") - .param("pageSize", "20") - ) + MockMvcRequestBuilders + .get("/api/logs/list") + .param("start", "1734005477000") + .param("end", "1734005478000") + .param("traceId", "trace123") + .param("spanId", "span456") + .param("severityNumber", "9") + .param("severityText", "INFO") + .param("pageIndex", "0") + .param("pageSize", "20") + ) .andDo(print()) .andExpect(status().isOk()) .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE)) @@ -129,18 +127,16 @@ class LogQueryControllerTest { .build() ); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.countLogsByMultipleConditions(isNull(), isNull(), isNull(), - isNull(), isNull(), isNull(), isNull())).thenReturn(1L); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.queryLogsByMultipleConditionsWithPagination(isNull(), isNull(), - isNull(), isNull(), isNull(), isNull(), isNull(), eq(0), eq(20))) + when(historyDataReader.countLogsByMultipleConditions(isNull(), isNull(), isNull(), + isNull(), isNull(), isNull())).thenReturn(1L); + when(historyDataReader.queryLogsByMultipleConditionsWithPagination(isNull(), isNull(), + isNull(), isNull(), isNull(), isNull(), eq(0), eq(20))) .thenReturn(mockLogs); mockMvc.perform( - MockMvcRequestBuilders - .get("/api/logs/list") - ) + MockMvcRequestBuilders + .get("/api/logs/list") + ) .andExpect(status().isOk()) .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE)) .andExpect(jsonPath("$.data.content").isArray()) @@ -167,14 +163,13 @@ class LogQueryControllerTest { LogEntry.builder().severityNumber(21).build() ); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.queryLogsByMultipleConditions(isNull(), isNull(), isNull(), - isNull(), isNull(), isNull(), isNull())).thenReturn(mockLogs); + when(historyDataReader.queryLogsByMultipleConditions(isNull(), isNull(), isNull(), + isNull(), isNull(), isNull())).thenReturn(mockLogs); mockMvc.perform( - MockMvcRequestBuilders - .get("/api/logs/stats/overview") - ) + MockMvcRequestBuilders + .get("/api/logs/stats/overview") + ) .andExpect(status().isOk()) .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE)) .andExpect(jsonPath("$.data.totalCount").value(8)) @@ -193,16 +188,15 @@ class LogQueryControllerTest { LogEntry.builder().severityNumber(17).build() ); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.queryLogsByMultipleConditions(eq(1734005477000L), eq(1734005478000L), - isNull(), isNull(), isNull(), isNull(), isNull())).thenReturn(mockLogs); + when(historyDataReader.queryLogsByMultipleConditions(eq(1734005477000L), eq(1734005478000L), + isNull(), isNull(), isNull(), isNull())).thenReturn(mockLogs); mockMvc.perform( - MockMvcRequestBuilders - .get("/api/logs/stats/overview") - .param("start", "1734005477000") - .param("end", "1734005478000") - ) + MockMvcRequestBuilders + .get("/api/logs/stats/overview") + .param("start", "1734005477000") + .param("end", "1734005478000") + ) .andExpect(status().isOk()) .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE)) .andExpect(jsonPath("$.data.totalCount").value(2)); @@ -223,14 +217,13 @@ class LogQueryControllerTest { LogEntry.builder().build() // null values ); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.queryLogsByMultipleConditions(isNull(), isNull(), isNull(), - isNull(), isNull(), isNull(), isNull())).thenReturn(mockLogs); + when(historyDataReader.queryLogsByMultipleConditions(isNull(), isNull(), isNull(), + isNull(), isNull(), isNull())).thenReturn(mockLogs); mockMvc.perform( - MockMvcRequestBuilders - .get("/api/logs/stats/trace-coverage") - ) + MockMvcRequestBuilders + .get("/api/logs/stats/trace-coverage") + ) .andExpect(status().isOk()) .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE)) .andExpect(jsonPath("$.data.traceCoverage.withTrace").value(3)) @@ -251,14 +244,13 @@ class LogQueryControllerTest { LogEntry.builder().timeUnixNano(1734009077630000000L).build() ); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.queryLogsByMultipleConditions(isNull(), isNull(), isNull(), - isNull(), isNull(), isNull(), isNull())).thenReturn(mockLogs); + when(historyDataReader.queryLogsByMultipleConditions(isNull(), isNull(), isNull(), + isNull(), isNull(), isNull())).thenReturn(mockLogs); mockMvc.perform( - MockMvcRequestBuilders - .get("/api/logs/stats/trend") - ) + MockMvcRequestBuilders + .get("/api/logs/stats/trend") + ) .andExpect(status().isOk()) .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE)) .andExpect(jsonPath("$.data.hourlyStats").isMap()); @@ -271,17 +263,16 @@ class LogQueryControllerTest { LogEntry.builder().timeUnixNano(null).build() // This should be filtered out ); - // Fixed: Added searchContent param matcher (7th arg) - when(historyDataReader.queryLogsByMultipleConditions(isNull(), isNull(), isNull(), - isNull(), isNull(), isNull(), isNull())).thenReturn(mockLogs); + when(historyDataReader.queryLogsByMultipleConditions(isNull(), isNull(), isNull(), + isNull(), isNull(), isNull())).thenReturn(mockLogs); mockMvc.perform( - MockMvcRequestBuilders - .get("/api/logs/stats/trend") - ) + MockMvcRequestBuilders + .get("/api/logs/stats/trend") + ) .andExpect(status().isOk()) .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE)) .andExpect(jsonPath("$.data.hourlyStats").isMap()); } -} \ No newline at end of file +} diff --git a/hertzbeat-startup/src/main/resources/application-test.yml b/hertzbeat-startup/src/main/resources/application-test.yml index 53bcfa3b00..4aa13715ff 100644 --- a/hertzbeat-startup/src/main/resources/application-test.yml +++ b/hertzbeat-startup/src/main/resources/application-test.yml @@ -91,7 +91,6 @@ warehouse: enabled: false grpc-endpoints: localhost:4001 http-endpoint: http://localhost:4000 - postgres-endpoint: localhost:4003 database: public username: greptime password: greptime diff --git a/hertzbeat-startup/src/main/resources/application.yml b/hertzbeat-startup/src/main/resources/application.yml index 8d36940e2c..e141b4606c 100644 --- a/hertzbeat-startup/src/main/resources/application.yml +++ b/hertzbeat-startup/src/main/resources/application.yml @@ -209,7 +209,6 @@ warehouse: enabled: false grpc-endpoints: localhost:4001 http-endpoint: http://localhost:4000 - postgres-endpoint: localhost:4003 # if you config other database name, you should create them first database: public username: greptime diff --git a/hertzbeat-warehouse/pom.xml b/hertzbeat-warehouse/pom.xml index 0d4ba07e18..95a32a5012 100644 --- a/hertzbeat-warehouse/pom.xml +++ b/hertzbeat-warehouse/pom.xml @@ -29,20 +29,24 @@ ${project.artifactId} + org.apache.hertzbeat hertzbeat-common provided + org.apache.hertzbeat hertzbeat-plugin provided + commons-net commons-net + org.springframework.boot spring-boot-starter-web @@ -57,16 +61,13 @@ spring-boot-configuration-processor true - - org.springframework.boot - spring-boot-starter-jdbc - provided - + com.taosdata.jdbc taos-jdbcdriver ${taos-jdbcdriver.version} + org.apache.iotdb iotdb-session @@ -82,16 +83,19 @@ + org.questdb questdb ${questdb.version} + org.influxdb influxdb-java ${influxdb.version} + io.greptime ingester-all @@ -115,6 +119,7 @@ + org.apache.kafka kafka-clients @@ -125,11 +130,13 @@ + io.lettuce lettuce-core provided + org.springdoc springdoc-openapi-starter-webmvc-ui @@ -143,15 +150,11 @@ org.apache.arrow arrow-memory-netty - + org.xerial.snappy snappy-java ${snappy-java.version} - - org.postgresql - postgresql - - \ No newline at end of file + diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimeSqlQueryExecutor.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimeSqlQueryExecutor.java index aab903b745..7e4c64341b 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimeSqlQueryExecutor.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimeSqlQueryExecutor.java @@ -7,7 +7,7 @@ * "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 + * 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 @@ -19,199 +19,111 @@ package org.apache.hertzbeat.warehouse.db; -import com.fasterxml.jackson.core.type.TypeReference; -import com.zaxxer.hikari.HikariDataSource; import lombok.extern.slf4j.Slf4j; -import org.apache.hertzbeat.common.entity.log.LogEntry; -import org.apache.hertzbeat.common.util.JsonUtil; +import org.apache.hertzbeat.common.constants.NetworkConstants; +import org.apache.hertzbeat.common.constants.SignConstants; +import org.apache.hertzbeat.common.util.Base64Util; import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeProperties; -import org.springframework.beans.factory.annotation.Autowired; +import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeSqlQueryContent; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.jdbc.core.BeanPropertyRowMapper; -import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpEntity; +import org.springframework.http.MediaType; +import org.springframework.http.HttpMethod; +import org.springframework.http.ResponseEntity; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; import org.springframework.web.client.RestTemplate; -import java.beans.PropertyDescriptor; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Timestamp; +import java.util.HashMap; +import java.util.LinkedList; import java.util.List; import java.util.Map; /** - * query executor for GreptimeDB SQL via JDBC + * query executor for GreptimeDB SQL */ @Slf4j @Component("greptimeSqlQueryExecutor") @ConditionalOnProperty(prefix = "warehouse.store.greptime", name = "enabled", havingValue = "true") public class GreptimeSqlQueryExecutor extends SqlQueryExecutor { + private static final String QUERY_PATH = "/v1/sql"; private static final String DATASOURCE = "Greptime-sql"; - private static final String DRIVER_CLASS_NAME = "org.postgresql.Driver"; - private static final String JDBC_URL_PREFIX = "jdbc:postgresql://"; - private final JdbcTemplate jdbcTemplate; - private final HikariDataSource dataSource; + private final GreptimeProperties greptimeProperties; - @Autowired - public GreptimeSqlQueryExecutor(GreptimeProperties greptimeProperties) { - super(null, null); // No longer using RestTemplate or HttpSqlProperties - // Initialize JDBC DataSource - this.dataSource = new HikariDataSource(); - - // Construct JDBC URL: jdbc:postgresql://endpoint/database - String jdbcUrl = JDBC_URL_PREFIX + greptimeProperties.postgresEndpoint() + "/" + greptimeProperties.database(); - this.dataSource.setJdbcUrl(jdbcUrl); - - // Fixed driver class name for PostgreSQL protocol - this.dataSource.setDriverClassName(DRIVER_CLASS_NAME); - - if (greptimeProperties.username() != null) { - this.dataSource.setUsername(greptimeProperties.username()); - } - if (greptimeProperties.password() != null) { - this.dataSource.setPassword(greptimeProperties.password()); - } - this.dataSource.setMaximumPoolSize(10); - this.dataSource.setMinimumIdle(2); - this.dataSource.setConnectionTimeout(30000); - - this.jdbcTemplate = new JdbcTemplate(this.dataSource); - log.info("Initialized GreptimeDB JDBC connection to {}", jdbcUrl); - } - - /** - * Constructor for compatibility with existing tests. - * delegating to the main constructor. - * @param greptimeProperties greptime properties - * @param restTemplate (unused) rest template - */ public GreptimeSqlQueryExecutor(GreptimeProperties greptimeProperties, RestTemplate restTemplate) { - this(greptimeProperties); - } - - /** - * Constructor for testing purposes only. - * @param jdbcTemplate Mocked JdbcTemplate - */ - public GreptimeSqlQueryExecutor(JdbcTemplate jdbcTemplate) { - super(null, null); - this.dataSource = null; - this.jdbcTemplate = jdbcTemplate; + super(restTemplate, new SqlQueryExecutor.HttpSqlProperties(greptimeProperties.httpEndpoint() + QUERY_PATH, + greptimeProperties.username(), greptimeProperties.password())); + this.greptimeProperties = greptimeProperties; } @Override - public List> execute(String sql) { - log.debug("Executing GreptimeDB SQL: {}", sql); + public List> execute(String queryString) { + List> results = new LinkedList<>(); try { - return jdbcTemplate.queryForList(sql); - } catch (Exception e) { - log.error("Failed to execute GreptimeDB SQL: {}", sql, e); - throw e; - } - } + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED); + headers.setAccept(List.of(MediaType.APPLICATION_JSON)); + if (StringUtils.hasText(greptimeProperties.username()) + && StringUtils.hasText(greptimeProperties.password())) { + String authStr = greptimeProperties.username() + ":" + greptimeProperties.password(); + String encodedAuth = Base64Util.encode(authStr); + headers.add(HttpHeaders.AUTHORIZATION, NetworkConstants.BASIC + SignConstants.BLANK + encodedAuth); + } - /** - * Execute SQL query with arguments (Prepared Statement) - * @param sql SQL query with ? placeholders - * @param args Arguments for placeholders - * @return List of rows - */ - public List query(String sql, Object... args) { - log.debug("Executing GreptimeDB SQL: {} with args: {}", sql, args); - try { - // Use custom RowMapper that extends BeanPropertyRowMapper - return jdbcTemplate.query(sql, new GreptimeLogEntryRowMapper(), args); - } catch (Exception e) { - log.error("Failed to execute GreptimeDB SQL: {}", sql, e); - throw e; - } - } + String requestBody = "sql=" + queryString; + HttpEntity httpEntity = new HttpEntity<>(requestBody, headers); - /** - * Execute count SQ - * @param sql SQL - * @return count - */ - public Long count(String sql, Object... args) { - try { - return jdbcTemplate.queryForObject(sql, Long.class, args); + String url = greptimeProperties.httpEndpoint() + QUERY_PATH; + if (StringUtils.hasText(greptimeProperties.database())) { + url += "?db=" + greptimeProperties.database(); + } + + ResponseEntity responseEntity = restTemplate.exchange(url, + HttpMethod.POST, httpEntity, GreptimeSqlQueryContent.class); + + if (responseEntity.getStatusCode().is2xxSuccessful()) { + GreptimeSqlQueryContent responseBody = responseEntity.getBody(); + if (responseBody != null && responseBody.getCode() == 0 + && responseBody.getOutput() != null && !responseBody.getOutput().isEmpty()) { + + for (GreptimeSqlQueryContent.Output output : responseBody.getOutput()) { + if (output.getRecords() != null && output.getRecords().getRows() != null) { + GreptimeSqlQueryContent.Output.Records.Schema schema = output.getRecords().getSchema(); + List> rows = output.getRecords().getRows(); + + for (List row : rows) { + Map rowMap = new HashMap<>(); + if (schema != null && schema.getColumnSchemas() != null) { + for (int i = 0; i < Math.min(schema.getColumnSchemas().size(), row.size()); i++) { + String columnName = schema.getColumnSchemas().get(i).getName(); + Object value = row.get(i); + rowMap.put(columnName, value); + } + } else { + for (int i = 0; i < row.size(); i++) { + rowMap.put("col_" + i, row.get(i)); + } + } + results.add(rowMap); + } + } + } + } + } else { + log.error("query metrics data from greptime failed. {}", responseEntity); + } } catch (Exception e) { - log.error("Failed to execute GreptimeDB SQL: {}", sql, e); - throw e; + log.error("query metrics data from greptime error. {}", e.getMessage(), e); } + return results; } @Override public String getDatasource() { return DATASOURCE; } - - // Ensure to close the datasource when the bean is destroyed - public void close() { - if (this.dataSource != null && !this.dataSource.isClosed()) { - this.dataSource.close(); - } - } - - /** - * Custom RowMapper that extends BeanPropertyRowMapper to leverage default mapping - * while overriding specific fields that need type conversion (Timestamp -> Long, JSON String -> Object). - */ - private static class GreptimeLogEntryRowMapper extends BeanPropertyRowMapper { - - public GreptimeLogEntryRowMapper() { - super(LogEntry.class); - } - - @Override - protected Object getColumnValue(ResultSet rs, int index, PropertyDescriptor pd) throws SQLException { - String propertyName = pd.getName(); - - // 1. Handle Timestamp to Long (nanoseconds) conversion - if ("timeUnixNano".equals(propertyName) || "observedTimeUnixNano".equals(propertyName)) { - Timestamp timestamp = rs.getTimestamp(index); - if (timestamp == null) { - return null; - } - long seconds = timestamp.getTime() / 1000; - long nanos = timestamp.getNanos(); - return seconds * 1_000_000_000L + nanos; - } - - // 2. Handle JSON String to Map conversion (attributes, resource) - if ("attributes".equals(propertyName) || "resource".equals(propertyName)) { - String json = rs.getString(index); - if (!StringUtils.hasText(json)) { - return null; - } - try { - return JsonUtil.fromJson(json, new TypeReference>() {}); - } catch (Exception e) { - log.warn("Failed to parse JSON map for {}: {}", propertyName, json); - return null; - } - } - - // 3. Handle JSON String to InstrumentationScope conversion - if ("instrumentationScope".equals(propertyName)) { - String json = rs.getString(index); - if (!StringUtils.hasText(json)) { - return null; - } - try { - return JsonUtil.fromJson(json, LogEntry.InstrumentationScope.class); - } catch (Exception e) { - log.warn("Failed to parse instrumentationScope: {}", json); - return null; - } - } - - // 4. Default handling for other fields (traceId, spanId, severityText, body, etc.) - return super.getColumnValue(rs, index, pd); - } - } -} \ No newline at end of file +} diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/HistoryDataReader.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/HistoryDataReader.java index 567e27177e..89fe0f3adb 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/HistoryDataReader.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/HistoryDataReader.java @@ -65,67 +65,45 @@ public interface HistoryDataReader { * @param spanId span ID filter * @param severityNumber severity number filter * @param severityText severity text filter - * @param searchContent search content in log body * @return filtered log entries */ default List queryLogsByMultipleConditions(Long startTime, Long endTime, String traceId, String spanId, Integer severityNumber, - String severityText, String searchContent) { + String severityText) { throw new UnsupportedOperationException("query logs by multiple conditions is not supported"); } /** - * Query logs with multiple filter conditions and pagination (Legacy) - */ - default List queryLogsByMultipleConditionsWithPagination(Long startTime, Long endTime, String traceId, - String spanId, Integer severityNumber, - String severityText, Integer offset, Integer limit) { - return queryLogsByMultipleConditionsWithPagination(startTime, endTime, traceId, spanId, severityNumber, severityText, null, offset, limit); - } - - /** - * Query logs with multiple filter conditions and pagination including search content + * Query logs with multiple filter conditions and pagination * @param startTime start time in milliseconds * @param endTime end time in milliseconds * @param traceId trace ID filter * @param spanId span ID filter * @param severityNumber severity number filter * @param severityText severity text filter - * @param searchContent search content in log body * @param offset pagination offset * @param limit pagination limit * @return filtered log entries with pagination */ default List queryLogsByMultipleConditionsWithPagination(Long startTime, Long endTime, String traceId, String spanId, Integer severityNumber, - String severityText, String searchContent, - Integer offset, Integer limit) { + String severityText, Integer offset, Integer limit) { throw new UnsupportedOperationException("query logs by multiple conditions with pagination is not supported"); } /** - * Count logs with multiple filter conditions (Legacy) - */ - default long countLogsByMultipleConditions(Long startTime, Long endTime, String traceId, - String spanId, Integer severityNumber, - String severityText) { - return countLogsByMultipleConditions(startTime, endTime, traceId, spanId, severityNumber, severityText, null); - } - - /** - * Count logs with multiple filter conditions including search content + * Count logs with multiple filter conditions * @param startTime start time in milliseconds * @param endTime end time in milliseconds * @param traceId trace ID filter * @param spanId span ID filter * @param severityNumber severity number filter * @param severityText severity text filter - * @param searchContent search content in log body * @return count of matching log entries */ default long countLogsByMultipleConditions(Long startTime, Long endTime, String traceId, String spanId, Integer severityNumber, - String severityText, String searchContent) { + String severityText) { throw new UnsupportedOperationException("count logs by multiple conditions is not supported"); } -} \ No newline at end of file +} diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java index c62e66a035..c2834f5ab9 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java @@ -26,32 +26,6 @@ import io.greptime.models.Table; import io.greptime.models.TableSchema; import io.greptime.models.WriteOk; import io.greptime.options.GreptimeOptions; -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.math.NumberUtils; -import org.apache.hertzbeat.common.constants.CommonConstants; -import org.apache.hertzbeat.common.constants.MetricDataConstants; -import org.apache.hertzbeat.common.entity.arrow.RowWrapper; -import org.apache.hertzbeat.common.entity.dto.Value; -import org.apache.hertzbeat.common.entity.log.LogEntry; -import org.apache.hertzbeat.common.entity.message.CollectRep; -import org.apache.hertzbeat.common.util.Base64Util; -import org.apache.hertzbeat.common.util.JsonUtil; -import org.apache.hertzbeat.common.util.TimePeriodUtil; -import org.apache.hertzbeat.warehouse.db.GreptimeSqlQueryExecutor; -import org.apache.hertzbeat.warehouse.store.history.tsdb.AbstractHistoryDataStorage; -import org.apache.hertzbeat.warehouse.store.history.tsdb.vm.PromQlQueryContent; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.http.HttpEntity; -import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpMethod; -import org.springframework.http.MediaType; -import org.springframework.http.ResponseEntity; -import org.springframework.stereotype.Component; -import org.springframework.util.MultiValueMap; -import org.springframework.util.StringUtils; -import org.springframework.web.client.RestTemplate; -import org.springframework.web.util.UriComponents; -import org.springframework.web.util.UriComponentsBuilder; import java.math.BigDecimal; import java.math.RoundingMode; @@ -75,8 +49,35 @@ import java.util.function.BiConsumer; import java.util.function.Function; import java.util.stream.Collectors; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.math.NumberUtils; +import org.apache.hertzbeat.common.constants.CommonConstants; +import org.apache.hertzbeat.common.constants.MetricDataConstants; +import org.apache.hertzbeat.common.entity.arrow.RowWrapper; +import org.apache.hertzbeat.common.entity.dto.Value; +import org.apache.hertzbeat.common.entity.log.LogEntry; +import org.apache.hertzbeat.common.entity.message.CollectRep; +import org.apache.hertzbeat.common.util.Base64Util; +import org.apache.hertzbeat.common.util.JsonUtil; +import org.apache.hertzbeat.common.util.TimePeriodUtil; +import org.apache.hertzbeat.warehouse.store.history.tsdb.AbstractHistoryDataStorage; +import org.apache.hertzbeat.warehouse.store.history.tsdb.vm.PromQlQueryContent; +import org.apache.hertzbeat.warehouse.db.GreptimeSqlQueryExecutor; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.stereotype.Component; +import org.springframework.util.MultiValueMap; +import org.springframework.util.StringUtils; +import org.springframework.web.client.RestTemplate; +import org.springframework.web.util.UriComponents; +import org.springframework.web.util.UriComponentsBuilder; + /** - * GreptimeDB data storage + * GreptimeDB data storage, only supports GreptimeDB version >= v0.5 */ @Component @ConditionalOnProperty(prefix = "warehouse.store.greptime", name = "enabled", havingValue = "true") @@ -94,8 +95,11 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { private static final int LOG_BATCH_SIZE = 500; private GreptimeDB greptimeDb; + private final GreptimeProperties greptimeProperties; + private final RestTemplate restTemplate; + private final GreptimeSqlQueryExecutor greptimeSqlQueryExecutor; public GreptimeDbDataStorage(GreptimeProperties greptimeProperties, RestTemplate restTemplate, @@ -123,6 +127,7 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { log.error("[warehouse greptime] Fail to start GreptimeDB client"); return false; } + return true; } @@ -161,15 +166,19 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { RowWrapper rowWrapper = metricsData.readRow(); while (rowWrapper.hasNextRow()) { rowWrapper = rowWrapper.nextRow(); + AtomicInteger index = new AtomicInteger(-1); rowWrapper.cellStream().forEach(cell -> { index.getAndIncrement(); + if (CommonConstants.NULL_VALUE.equals(cell.getValue())) { values[2 + index.get()] = null; return; } + Boolean label = cell.getMetadataAsBoolean(MetricDataConstants.LABEL); Byte type = cell.getMetadataAsByte(MetricDataConstants.TYPE); + if (label) { values[2 + index.get()] = cell.getValue(); } else { @@ -180,8 +189,10 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { } } }); + table.addRow(values); } + CompletableFuture> writeFuture = greptimeDb.write(table); try { Result result = writeFuture.get(10, TimeUnit.SECONDS); @@ -201,6 +212,7 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { Map timeRange = getTimeRange(history); Long start = timeRange.get(LABEL_KEY_START_TIME); Long end = timeRange.get(LABEL_KEY_END_TIME); + String step = getTimeStep(start, end); return getHistoryData(start, end, step, instance, app, metrics, metric); @@ -216,6 +228,7 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { Map timeRange = getTimeRange(history); Long start = timeRange.get(LABEL_KEY_START_TIME); Long end = timeRange.get(LABEL_KEY_END_TIME); + String step = getTimeStep(start, end); Map> instanceValuesMap = getHistoryData(start, end, step, instance, app, metrics, metric); @@ -224,25 +237,32 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { // Therefore, we restrict the valid range by obtaining the post-query timeframe. // Since `gretime`'s `end` excludes the specified time, we add 4 hours. List values = instanceValuesMap.get(instanceValuesMap.keySet().stream().toList().get(0)); + // effective time long effectiveStart = values.get(0).getTime() / 1000; long effectiveEnd = values.get(values.size() - 1).getTime() / 1000 + Duration.ofHours(4).getSeconds(); + String name = getTableName(metrics); String timeSeriesSelector = name + "{" + LABEL_KEY_INSTANCE + "=\"" + instance + "\""; if (!CommonConstants.PROMETHEUS.equals(app)) { timeSeriesSelector = timeSeriesSelector + "," + LABEL_KEY_FIELD + "=\"" + metric + "\""; } timeSeriesSelector = timeSeriesSelector + "}"; + try { + // max String finalTimeSeriesSelector = timeSeriesSelector; URI uri = getUri(effectiveStart, effectiveEnd, step, uriComponents -> "max_over_time(" + finalTimeSeriesSelector + "[" + step + "])"); requestIntervalMetricAndPutValue(uri, instanceValuesMap, Value::setMax); + // min uri = getUri(effectiveStart, effectiveEnd, step, uriComponents -> "min_over_time(" + finalTimeSeriesSelector + "[" + step + "])"); requestIntervalMetricAndPutValue(uri, instanceValuesMap, Value::setMin); + // avg uri = getUri(effectiveStart, effectiveEnd, step, uriComponents -> "avg_over_time(" + finalTimeSeriesSelector + "[" + step + "])"); requestIntervalMetricAndPutValue(uri, instanceValuesMap, Value::setMean); } catch (Exception e) { log.error("query interval metrics data from greptime error. {}", e.getMessage(), e); } + return instanceValuesMap; } @@ -282,6 +302,7 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { * @return step */ private String getTimeStep(long start, long end) { + // get step String step = "60s"; if (end - start < Duration.ofDays(7).getSeconds() && end - start > Duration.ofDays(1).getSeconds()) { step = "1h"; @@ -310,9 +331,11 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { if (!CommonConstants.PROMETHEUS.equals(app)) { timeSeriesSelector = timeSeriesSelector + "," + LABEL_KEY_FIELD + "=\"" + metric + "\""; } + Map> instanceValuesMap = new HashMap<>(8); try { HttpEntity httpEntity = getHttpEntity(); + String finalTimeSeriesSelector = timeSeriesSelector; URI uri = getUri(start, end, step, uriComponents -> { MultiValueMap queryParams = uriComponents.getQueryParams(); @@ -321,13 +344,16 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { } return null; }); + ResponseEntity responseEntity = null; if (uri != null) { - responseEntity = restTemplate.exchange(uri, HttpMethod.GET, httpEntity, PromQlQueryContent.class); + responseEntity = restTemplate.exchange(uri, + HttpMethod.GET, httpEntity, PromQlQueryContent.class); } if (responseEntity != null && responseEntity.getStatusCode().is2xxSuccessful()) { log.debug("query metrics data from greptime success. {}", uri); - if (responseEntity.getBody() != null && responseEntity.getBody().getData() != null && responseEntity.getBody().getData().getResult() != null) { + if (responseEntity.getBody() != null && responseEntity.getBody().getData() != null + && responseEntity.getBody().getData().getResult() != null) { List contents = responseEntity.getBody().getData().getResult(); for (PromQlQueryContent.ContentData.Content content : contents) { Map labels = content.getMetric(); @@ -362,7 +388,8 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); headers.setAccept(List.of(MediaType.APPLICATION_JSON)); - if (StringUtils.hasText(greptimeProperties.username()) && StringUtils.hasText(greptimeProperties.password())) { + if (StringUtils.hasText(greptimeProperties.username()) + && StringUtils.hasText(greptimeProperties.password())) { String authStr = greptimeProperties.username() + ":" + greptimeProperties.password(); String encodedAuth = Base64Util.encode(authStr); headers.add(HttpHeaders.AUTHORIZATION, BASIC + " " + encodedAuth); @@ -410,7 +437,8 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { return; } HttpEntity httpEntity = getHttpEntity(); - ResponseEntity responseEntity = restTemplate.exchange(uri, HttpMethod.GET, httpEntity, PromQlQueryContent.class); + ResponseEntity responseEntity = restTemplate.exchange(uri, + HttpMethod.GET, httpEntity, PromQlQueryContent.class); if (!responseEntity.getStatusCode().is2xxSuccessful()) { log.error("query interval metrics data from greptime failed. {}", responseEntity); return; @@ -447,9 +475,6 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { this.greptimeDb.shutdownGracefully(); this.greptimeDb = null; } - if (this.greptimeSqlQueryExecutor != null) { - this.greptimeSqlQueryExecutor.close(); - } } @Override @@ -457,13 +482,15 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { if (!isServerAvailable()) { return; } + try { + // Create table schema TableSchema.Builder tableSchemaBuilder = TableSchema.newBuilder(LOG_TABLE_NAME); tableSchemaBuilder.addTimestamp("time_unix_nano", DataType.TimestampNanosecond) .addField("observed_time_unix_nano", DataType.TimestampNanosecond) .addField("severity_number", DataType.Int32) .addField("severity_text", DataType.String) - .addField("body", DataType.String) + .addField("body", DataType.Json) .addField("trace_id", DataType.String) .addField("span_id", DataType.String) .addField("trace_flags", DataType.Int32) @@ -471,13 +498,16 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { .addField("resource", DataType.Json) .addField("instrumentation_scope", DataType.Json) .addField("dropped_attributes_count", DataType.Int32); + Table table = Table.from(tableSchemaBuilder.build()); - Object[] values = new Object[]{ + + // Convert LogEntry to table row + Object[] values = new Object[] { logEntry.getTimeUnixNano() != null ? logEntry.getTimeUnixNano() : System.nanoTime(), logEntry.getObservedTimeUnixNano() != null ? logEntry.getObservedTimeUnixNano() : System.nanoTime(), logEntry.getSeverityNumber(), logEntry.getSeverityText(), - logEntry.getBody(), + JsonUtil.toJson(logEntry.getBody()), logEntry.getTraceId(), logEntry.getSpanId(), logEntry.getTraceFlags(), @@ -486,9 +516,13 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { JsonUtil.toJson(logEntry.getInstrumentationScope()), logEntry.getDroppedAttributesCount() }; + table.addRow(values); + + // Write to GreptimeDB CompletableFuture> writeFuture = greptimeDb.write(table); Result result = writeFuture.get(10, TimeUnit.SECONDS); + if (result.isOk()) { log.debug("[warehouse greptime-log] Write successful"); } else { @@ -502,16 +536,14 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { @Override public List queryLogsByMultipleConditions(Long startTime, Long endTime, String traceId, String spanId, Integer severityNumber, - String severityText, String searchContent) { + String severityText) { try { StringBuilder sql = new StringBuilder("SELECT * FROM ").append(LOG_TABLE_NAME); - List args = new ArrayList<>(); - buildWhereConditions(sql, args, startTime, endTime, traceId, spanId, severityNumber, severityText, searchContent); + buildWhereConditions(sql, startTime, endTime, traceId, spanId, severityNumber, severityText); sql.append(" ORDER BY time_unix_nano DESC"); - // Execute via JDBC executor using parameters - List rows = greptimeSqlQueryExecutor.query(sql.toString(), args.toArray()); - return rows; + List> rows = greptimeSqlQueryExecutor.execute(sql.toString()); + return mapRowsToLogEntries(rows); } catch (Exception e) { log.error("[warehouse greptime-log] queryLogsByMultipleConditions error: {}", e.getMessage(), e); return List.of(); @@ -521,24 +553,22 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { @Override public List queryLogsByMultipleConditionsWithPagination(Long startTime, Long endTime, String traceId, String spanId, Integer severityNumber, - String severityText, String searchContent, - Integer offset, Integer limit) { + String severityText, Integer offset, Integer limit) { try { StringBuilder sql = new StringBuilder("SELECT * FROM ").append(LOG_TABLE_NAME); - List args = new ArrayList<>(); - buildWhereConditions(sql, args, startTime, endTime, traceId, spanId, severityNumber, severityText, searchContent); + buildWhereConditions(sql, startTime, endTime, traceId, spanId, severityNumber, severityText); sql.append(" ORDER BY time_unix_nano DESC"); + // Add pagination if (limit != null && limit > 0) { - sql.append(" LIMIT ?"); - args.add(limit); + sql.append(" LIMIT ").append(limit); if (offset != null && offset > 0) { - sql.append(" OFFSET ?"); - args.add(offset); + sql.append(" OFFSET ").append(offset); } } - return greptimeSqlQueryExecutor.query(sql.toString(), args.toArray()); + List> rows = greptimeSqlQueryExecutor.execute(sql.toString()); + return mapRowsToLogEntries(rows); } catch (Exception e) { log.error("[warehouse greptime-log] queryLogsByMultipleConditionsWithPagination error: {}", e.getMessage(), e); return List.of(); @@ -547,14 +577,20 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { @Override public long countLogsByMultipleConditions(Long startTime, Long endTime, String traceId, - String spanId, Integer severityNumber, - String severityText, String searchContent) { + String spanId, Integer severityNumber, + String severityText) { try { StringBuilder sql = new StringBuilder("SELECT COUNT(*) as count FROM ").append(LOG_TABLE_NAME); - List args = new ArrayList<>(); - buildWhereConditions(sql, args, startTime, endTime, traceId, spanId, severityNumber, severityText, searchContent); + buildWhereConditions(sql, startTime, endTime, traceId, spanId, severityNumber, severityText); - return greptimeSqlQueryExecutor.count(sql.toString(), args.toArray()); + List> rows = greptimeSqlQueryExecutor.execute(sql.toString()); + if (rows != null && !rows.isEmpty()) { + Object countObj = rows.get(0).get("count"); + if (countObj instanceof Number) { + return ((Number) countObj).longValue(); + } + } + return 0; } catch (Exception e) { log.error("[warehouse greptime-log] countLogsByMultipleConditions error: {}", e.getMessage(), e); return 0; @@ -565,6 +601,12 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { return ms * 1_000_000L; } + private static String safeString(String input) { + if (input == null) { + return ""; + } + return input.replace("'", "''"); + } /** * build WHERE conditions @@ -575,57 +617,151 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { * @param spanId span id * @param severityNumber severity number */ - private void buildWhereConditions(StringBuilder sql, List args, Long startTime, Long endTime, - String traceId, String spanId, Integer severityNumber, - String severityText, String searchContent) { + private void buildWhereConditions(StringBuilder sql, Long startTime, Long endTime, String traceId, + String spanId, Integer severityNumber, String severityText) { List conditions = new ArrayList<>(); + + // Time range condition if (startTime != null && endTime != null) { - conditions.add("time_unix_nano >= ? AND time_unix_nano <= ?"); - args.add(msToNs(startTime)); - args.add(msToNs(endTime)); + conditions.add("time_unix_nano >= " + msToNs(startTime) + " AND time_unix_nano <= " + msToNs(endTime)); } + + // TraceId condition if (StringUtils.hasText(traceId)) { - conditions.add("trace_id = ?"); - args.add(traceId); + conditions.add("trace_id = '" + safeString(traceId) + "'"); } + + // SpanId condition if (StringUtils.hasText(spanId)) { - conditions.add("span_id = ?"); - args.add(spanId); + conditions.add("span_id = '" + safeString(spanId) + "'"); } + + // Severity condition if (severityNumber != null) { - conditions.add("severity_number = ?"); - args.add(severityNumber); + conditions.add("severity_number = " + severityNumber); } + + // SeverityText condition if (StringUtils.hasText(severityText)) { - conditions.add("severity_text = ?"); - args.add(severityText); - } - if (StringUtils.hasText(searchContent)) { - // Using CAST(body AS String) to search within JSON/String content. - // GreptimeDB supports PostgreSQL protocol, this syntax is generally safe. - conditions.add("CAST(body AS String) LIKE ?"); - args.add("%" + searchContent + "%"); + conditions.add("severity_text = '" + safeString(severityText) + "'"); } + + // Add WHERE clause if there are conditions if (!conditions.isEmpty()) { sql.append(" WHERE ").append(String.join(" AND ", conditions)); } } + private List mapRowsToLogEntries(List> rows) { + List list = new LinkedList<>(); + if (rows == null || rows.isEmpty()) { + return list; + } + for (Map row : rows) { + try { + LogEntry.InstrumentationScope scope = null; + Object scopeObj = row.get("instrumentation_scope"); + if (scopeObj instanceof String scopeStr && StringUtils.hasText(scopeStr)) { + try { + scope = JsonUtil.fromJson(scopeStr, LogEntry.InstrumentationScope.class); + } catch (Exception ignore) { + scope = null; + } + } + + Object bodyObj = parseJsonMaybe(row.get("body")); + Map attributes = castToMap(parseJsonMaybe(row.get("attributes"))); + Map resource = castToMap(parseJsonMaybe(row.get("resource"))); + + LogEntry entry = LogEntry.builder() + .timeUnixNano(castToLong(row.get("time_unix_nano"))) + .observedTimeUnixNano(castToLong(row.get("observed_time_unix_nano"))) + .severityNumber(castToInteger(row.get("severity_number"))) + .severityText(castToString(row.get("severity_text"))) + .body(bodyObj) + .traceId(castToString(row.get("trace_id"))) + .spanId(castToString(row.get("span_id"))) + .traceFlags(castToInteger(row.get("trace_flags"))) + .attributes(attributes) + .resource(resource) + .instrumentationScope(scope) + .droppedAttributesCount(castToInteger(row.get("dropped_attributes_count"))) + .build(); + list.add(entry); + } catch (Exception e) { + log.warn("[warehouse greptime-log] map row to LogEntry error: {}", e.getMessage()); + } + } + return list; + } + + private static Object parseJsonMaybe(Object value) { + if (value == null) return null; + if (value instanceof Map) return value; + if (value instanceof String str) { + String s = str.trim(); + if ((s.startsWith("{") && s.endsWith("}")) || (s.startsWith("[") && s.endsWith("]"))) { + try { + return JsonUtil.fromJson(s, Object.class); + } catch (Exception e) { + return s; + } + } + return s; + } + return value; + } + + @SuppressWarnings("unchecked") + private static Map castToMap(Object obj) { + if (obj instanceof Map) { + return (Map) obj; + } + return null; + } + + private static Long castToLong(Object obj) { + if (obj == null) return null; + if (obj instanceof Number n) return n.longValue(); + try { + return Long.parseLong(String.valueOf(obj)); + } catch (Exception e) { + return null; + } + } + + private static Integer castToInteger(Object obj) { + if (obj == null) return null; + if (obj instanceof Number n) return n.intValue(); + try { + return Integer.parseInt(String.valueOf(obj)); + } catch (Exception e) { + return null; + } + } + + private static String castToString(Object obj) { + return obj == null ? null : String.valueOf(obj); + } + @Override public boolean batchDeleteLogs(List timeUnixNanos) { if (!isServerAvailable() || timeUnixNanos == null || timeUnixNanos.isEmpty()) { return false; } + try { StringBuilder sql = new StringBuilder("DELETE FROM ").append(LOG_TABLE_NAME).append(" WHERE time_unix_nano IN ("); - // Construct placeholders (?,?,?) - String placeholders = timeUnixNanos.stream().map(t -> "?").collect(Collectors.joining(", ")); - sql.append(placeholders).append(")"); + sql.append(timeUnixNanos.stream() + .filter(time -> time != null) + .map(String::valueOf) + .collect(Collectors.joining(", "))); + sql.append(")"); - // Convert list to array for varargs - greptimeSqlQueryExecutor.query(sql.toString(), timeUnixNanos.toArray()); + greptimeSqlQueryExecutor.execute(sql.toString()); log.info("[warehouse greptime-log] Batch delete executed successfully for {} logs", timeUnixNanos.size()); return true; + } catch (Exception e) { log.error("[warehouse greptime-log] batchDeleteLogs error: {}", e.getMessage(), e); return false; diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeProperties.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeProperties.java index f356eea5c6..7913eec5fd 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeProperties.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeProperties.java @@ -27,16 +27,11 @@ import org.springframework.boot.context.properties.bind.DefaultValue; * GrepTimeDB configuration information */ @ConfigurationProperties(prefix = ConfigConstants.FunctionModuleConstants.WAREHOUSE - + SignConstants.DOT - + WarehouseConstants.STORE - + SignConstants.DOT - + WarehouseConstants.HistoryName.GREPTIME) -public record GreptimeProperties( - @DefaultValue("false") boolean enabled, - @DefaultValue("127.0.0.1:4001") String grpcEndpoints, - @DefaultValue("http://127.0.0.1:4000") String httpEndpoint, - @DefaultValue("127.0.0.1:4003") String postgresEndpoint, - @DefaultValue("public") String database, - String username, - String password) { -} \ No newline at end of file + + SignConstants.DOT + + WarehouseConstants.STORE + + SignConstants.DOT + + WarehouseConstants.HistoryName.GREPTIME) +public record GreptimeProperties(@DefaultValue("false") boolean enabled, + @DefaultValue("127.0.0.1:4001") String grpcEndpoints, @DefaultValue("http://127.0.0.1:4000") String httpEndpoint, + @DefaultValue("public") String database, String username, String password) { +} diff --git a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/db/GreptimeSqlQueryExecutorTest.java b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/db/GreptimeSqlQueryExecutorTest.java index 9a518c8a54..99f83855ae 100644 --- a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/db/GreptimeSqlQueryExecutorTest.java +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/db/GreptimeSqlQueryExecutorTest.java @@ -7,7 +7,7 @@ * "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 + * 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 @@ -21,23 +21,26 @@ package org.apache.hertzbeat.warehouse.db; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.util.ArrayList; import java.util.List; import java.util.Map; -import org.apache.hertzbeat.common.entity.log.LogEntry; +import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeProperties; +import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeSqlQueryContent; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.jdbc.core.RowMapper; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.web.client.RestTemplate; /** * Test case for {@link GreptimeSqlQueryExecutor} @@ -46,23 +49,36 @@ import org.springframework.jdbc.core.RowMapper; class GreptimeSqlQueryExecutorTest { @Mock - private JdbcTemplate jdbcTemplate; + private GreptimeProperties greptimeProperties; + + @Mock + private RestTemplate restTemplate; private GreptimeSqlQueryExecutor greptimeSqlQueryExecutor; @BeforeEach void setUp() { - // Use the constructor capable of dependency injection for mocking - greptimeSqlQueryExecutor = new GreptimeSqlQueryExecutor(jdbcTemplate); + when(greptimeProperties.httpEndpoint()).thenReturn("http://127.0.0.1:4000"); + when(greptimeProperties.database()).thenReturn("hertzbeat"); + when(greptimeProperties.username()).thenReturn("username"); + when(greptimeProperties.password()).thenReturn("password"); + + greptimeSqlQueryExecutor = new GreptimeSqlQueryExecutor(greptimeProperties, restTemplate); } @Test void testExecuteSuccess() { - // Mock successful response for queryForList - List> mockResult = new ArrayList<>(); - mockResult.add(Map.of("metric_name", "cpu", "value", 85.5)); + // Mock successful response + GreptimeSqlQueryContent mockResponse = createMockResponse(); + ResponseEntity responseEntity = + new ResponseEntity<>(mockResponse, HttpStatus.OK); - when(jdbcTemplate.queryForList(any(String.class))).thenReturn(mockResult); + when(restTemplate.exchange( + any(String.class), + eq(HttpMethod.POST), + any(HttpEntity.class), + eq(GreptimeSqlQueryContent.class) + )).thenReturn(responseEntity); // Execute List> result = greptimeSqlQueryExecutor.execute("SELECT * FROM metrics"); @@ -77,41 +93,48 @@ class GreptimeSqlQueryExecutorTest { @Test void testExecuteError() { // Mock error response - when(jdbcTemplate.queryForList(any(String.class))).thenThrow(new RuntimeException("Connection error")); - - // Execute and verify exception - assertThrows(RuntimeException.class, () -> greptimeSqlQueryExecutor.execute("SELECT * FROM metrics")); - } - - @Test - void testQuerySuccess() { - // Mock success response for query (using RowMapper) - List mockLogs = List.of(LogEntry.builder().traceId("123").build()); - when(jdbcTemplate.query(any(String.class), any(RowMapper.class), any(Object[].class))) - .thenReturn(mockLogs); + when(restTemplate.exchange( + any(String.class), + eq(HttpMethod.POST), + any(HttpEntity.class), + eq(GreptimeSqlQueryContent.class) + )).thenThrow(new RuntimeException("Connection error")); // Execute - List result = greptimeSqlQueryExecutor.query("SELECT * FROM logs WHERE id = ?", 1); + List> result = greptimeSqlQueryExecutor.execute("SELECT * FROM metrics"); - // Verify + // Verify returns empty list on error assertNotNull(result); - assertEquals(1, result.size()); - assertEquals("123", result.get(0).getTraceId()); - - // Verify args passed - verify(jdbcTemplate).query(eq("SELECT * FROM logs WHERE id = ?"), any(RowMapper.class), eq(1)); + assertTrue(result.isEmpty()); } - @Test - void testCountSuccess() { - // Mock success response for queryForObject (count) - when(jdbcTemplate.queryForObject(any(String.class), eq(Long.class), any(Object[].class))) - .thenReturn(10L); + private GreptimeSqlQueryContent createMockResponse() { + GreptimeSqlQueryContent response = new GreptimeSqlQueryContent(); + response.setCode(0); - // Execute - Long count = greptimeSqlQueryExecutor.count("SELECT COUNT(*) FROM logs"); + // Create simple schema + List columnSchemas = new ArrayList<>(); + columnSchemas.add(new GreptimeSqlQueryContent.Output.Records.Schema.ColumnSchema("metric_name", "String")); + columnSchemas.add(new GreptimeSqlQueryContent.Output.Records.Schema.ColumnSchema("value", "Float64")); - // Verify - assertEquals(10L, count); + GreptimeSqlQueryContent.Output.Records.Schema schema = + new GreptimeSqlQueryContent.Output.Records.Schema(); + schema.setColumnSchemas(columnSchemas); + + // Create simple row + List> rows = new ArrayList<>(); + rows.add(List.of("cpu", 85.5)); + + // Build response structure + GreptimeSqlQueryContent.Output.Records records = + new GreptimeSqlQueryContent.Output.Records(); + records.setSchema(schema); + records.setRows(rows); + + GreptimeSqlQueryContent.Output output = new GreptimeSqlQueryContent.Output(); + output.setRecords(records); + + response.setOutput(List.of(output)); + return response; } -} \ No newline at end of file +} diff --git a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java index 4b4a825608..45a1234c11 100644 --- a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java @@ -82,7 +82,7 @@ class GreptimeDbDataStorageTest { @Mock private GreptimeDB greptimeDb; - + private GreptimeDbDataStorage greptimeDbDataStorage; @BeforeEach @@ -120,14 +120,14 @@ class GreptimeDbDataStorageTest { void testSaveData() { try (MockedStatic mockedStatic = mockStatic(GreptimeDB.class)) { mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb); - + // Mock the write result @SuppressWarnings("unchecked") Result mockResult = mock(Result.class); when(mockResult.isOk()).thenReturn(true); CompletableFuture> mockFuture = CompletableFuture.completedFuture(mockResult); when(greptimeDb.write(any(Table.class))).thenReturn(mockFuture); - + greptimeDbDataStorage = new GreptimeDbDataStorage(greptimeProperties, restTemplate, greptimeSqlQueryExecutor); // Test with valid metrics data @@ -153,7 +153,7 @@ class GreptimeDbDataStorageTest { @Test void testGetHistoryMetricData() { greptimeDbDataStorage = new GreptimeDbDataStorage(greptimeProperties, restTemplate, greptimeSqlQueryExecutor); - + PromQlQueryContent content = createMockPromQlQueryContent(); ResponseEntity responseEntity = new ResponseEntity<>(content, HttpStatus.OK); @@ -172,14 +172,14 @@ class GreptimeDbDataStorageTest { void testSaveLogData() { try (MockedStatic mockedStatic = mockStatic(GreptimeDB.class)) { mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb); - + // Mock the write result @SuppressWarnings("unchecked") Result mockResult = mock(Result.class); when(mockResult.isOk()).thenReturn(true); CompletableFuture> mockFuture = CompletableFuture.completedFuture(mockResult); when(greptimeDb.write(any(Table.class))).thenReturn(mockFuture); - + greptimeDbDataStorage = new GreptimeDbDataStorage(greptimeProperties, restTemplate, greptimeSqlQueryExecutor); LogEntry logEntry = createMockLogEntry(); @@ -194,61 +194,51 @@ class GreptimeDbDataStorageTest { try (MockedStatic mockedStatic = mockStatic(GreptimeDB.class)) { mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb); greptimeDbDataStorage = new GreptimeDbDataStorage(greptimeProperties, restTemplate, greptimeSqlQueryExecutor); - - // Mock list return for query - List mockLogs = List.of(createMockLogEntry()); - - // The query passes 7 arguments: start, end, traceId, spanId, severityNumber, severityText, content - when(greptimeSqlQueryExecutor.query(anyString(), any(), any(), any(), any(), any(), any(), any())) - .thenReturn(mockLogs); - + List> mockLogRows = createMockLogRows(); + when(greptimeSqlQueryExecutor.execute(anyString())).thenReturn(mockLogRows); + // Test basic query List result = greptimeDbDataStorage.queryLogsByMultipleConditions( - System.currentTimeMillis() - 3600000, System.currentTimeMillis(), "trace123", "span456", 1, "INFO", "content" + System.currentTimeMillis() - 3600000, System.currentTimeMillis(), "trace123", "span456", 1, "INFO" ); assertNotNull(result); assertEquals(1, result.size()); assertEquals("trace123", result.get(0).getTraceId()); - // Mock count return - // The count query passes 2 arguments: start, end (other fields are null) - when(greptimeSqlQueryExecutor.count(anyString(), any(), any())).thenReturn(5L); + // Test count query + List> mockCountResult = List.of(Map.of("count", 5L)); + when(greptimeSqlQueryExecutor.execute(anyString())).thenReturn(mockCountResult); long count = greptimeDbDataStorage.countLogsByMultipleConditions( - System.currentTimeMillis() - 3600000, System.currentTimeMillis(), null, null, null, null, null + System.currentTimeMillis() - 3600000, System.currentTimeMillis(), null, null, null, null ); assertEquals(5L, count); - + // Test count query with executor error - when(greptimeSqlQueryExecutor.count(anyString(), any(), any())).thenThrow(new RuntimeException("Database error")); - long errorCount = greptimeDbDataStorage.countLogsByMultipleConditions(System.currentTimeMillis() - 3600000, System.currentTimeMillis(), null, null, null, null, null); + when(greptimeSqlQueryExecutor.execute(anyString())).thenThrow(new RuntimeException("Database error")); + long errorCount = greptimeDbDataStorage.countLogsByMultipleConditions(System.currentTimeMillis() - 3600000, System.currentTimeMillis(), null, null, null, null); assertEquals(0L, errorCount); } } - + @Test void testQueryLogsWithPagination() { try (MockedStatic mockedStatic = mockStatic(GreptimeDB.class)) { mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb); greptimeDbDataStorage = new GreptimeDbDataStorage(greptimeProperties, restTemplate, greptimeSqlQueryExecutor); - - List mockLogs = List.of(createMockLogEntry()); - // The query passes 4 arguments: start, end, limit, offset - when(greptimeSqlQueryExecutor.query(anyString(), any(), any(), any(), any())).thenReturn(mockLogs); - + when(greptimeSqlQueryExecutor.execute(anyString())).thenReturn(createMockLogRows()); + ArgumentCaptor sqlCaptor = ArgumentCaptor.forClass(String.class); greptimeDbDataStorage.queryLogsByMultipleConditionsWithPagination( System.currentTimeMillis() - 3600000, System.currentTimeMillis(), - null, null, null, null, null, 1, 10 + null, null, null, null, 1, 10 ); - // Verify that the query method was called with the SQL containing LIMIT and OFFSET - // And 4 parameters - verify(greptimeSqlQueryExecutor).query(sqlCaptor.capture(), any(), any(), any(), any()); + verify(greptimeSqlQueryExecutor).execute(sqlCaptor.capture()); String capturedSql = sqlCaptor.getValue(); - - assertTrue(capturedSql.toLowerCase().contains("limit ?")); - assertTrue(capturedSql.toLowerCase().contains("offset ?")); + + assertTrue(capturedSql.toLowerCase().contains("limit 10")); + assertTrue(capturedSql.toLowerCase().contains("offset 1")); } } @@ -262,8 +252,7 @@ class GreptimeDbDataStorageTest { // Test with valid list boolean result = greptimeDbDataStorage.batchDeleteLogs(List.of(1L, 2L)); assertTrue(result); - // Verify query method is called with 2 arguments (for 2 IDs) - verify(greptimeSqlQueryExecutor, times(1)).query(anyString(), eq(1L), eq(2L)); + verify(greptimeSqlQueryExecutor, times(1)).execute(anyString()); } } @@ -276,8 +265,7 @@ class GreptimeDbDataStorageTest { // Test with empty list boolean emptyResult = greptimeDbDataStorage.batchDeleteLogs(Collections.emptyList()); assertFalse(emptyResult); - // Verify query was never called - verify(greptimeSqlQueryExecutor, never()).query(anyString(), any()); + verify(greptimeSqlQueryExecutor, never()).execute(anyString()); } } @@ -290,8 +278,7 @@ class GreptimeDbDataStorageTest { // Test with null list boolean nullResult = greptimeDbDataStorage.batchDeleteLogs(null); assertFalse(nullResult); - // Verify query was never called - verify(greptimeSqlQueryExecutor, never()).query(anyString(), any()); + verify(greptimeSqlQueryExecutor, never()).execute(anyString()); } } @@ -312,7 +299,7 @@ class GreptimeDbDataStorageTest { lenient().when(mockMetricsData.getCode()).thenReturn(CollectRep.Code.SUCCESS); lenient().when(mockMetricsData.getMetrics()).thenReturn("cpu"); lenient().when(mockMetricsData.getId()).thenReturn(1L); - + if (!hasValues) { lenient().when(mockMetricsData.getValues()).thenReturn(Collections.emptyList()); return mockMetricsData; @@ -324,36 +311,36 @@ class GreptimeDbDataStorageTest { lenient().when(mockField1.getName()).thenReturn("usage"); lenient().when(mockField1.getLabel()).thenReturn(false); lenient().when(mockField1.getType()).thenReturn((int) CommonConstants.TYPE_NUMBER); - + CollectRep.Field mockField2 = mock(CollectRep.Field.class); lenient().when(mockField2.getName()).thenReturn("instance"); lenient().when(mockField2.getLabel()).thenReturn(true); lenient().when(mockField2.getType()).thenReturn((int) CommonConstants.TYPE_STRING); - + lenient().when(mockMetricsData.getFields()).thenReturn(List.of(mockField1, mockField2)); - + // Create ValueRow mock CollectRep.ValueRow mockValueRow = mock(CollectRep.ValueRow.class); lenient().when(mockValueRow.getColumnsList()).thenReturn(List.of("server1", "85.5")); - + lenient().when(mockMetricsData.getValues()).thenReturn(List.of(mockValueRow)); // Mock RowWrapper for readRow() RowWrapper mockRowWrapper = mock(RowWrapper.class); lenient().when(mockRowWrapper.hasNextRow()).thenReturn(true, false); lenient().when(mockRowWrapper.nextRow()).thenReturn(mockRowWrapper); - + // Mock cell stream ArrowCell mockCell1 = mock(ArrowCell.class); lenient().when(mockCell1.getValue()).thenReturn("85.5"); lenient().when(mockCell1.getMetadataAsBoolean(any())).thenReturn(false); lenient().when(mockCell1.getMetadataAsByte(any())).thenReturn(CommonConstants.TYPE_NUMBER); - + ArrowCell mockCell2 = mock(ArrowCell.class); lenient().when(mockCell2.getValue()).thenReturn("server1"); lenient().when(mockCell2.getMetadataAsBoolean(any())).thenReturn(true); lenient().when(mockCell2.getMetadataAsByte(any())).thenReturn(CommonConstants.TYPE_STRING); - + lenient().when(mockRowWrapper.cellStream()).thenReturn(java.util.stream.Stream.of(mockCell1, mockCell2)); lenient().when(mockMetricsData.readRow()).thenReturn(mockRowWrapper); @@ -389,4 +376,15 @@ class GreptimeDbDataStorageTest { .spanId("span456") .build(); } + + private List> createMockLogRows() { + Map row = new HashMap<>(); + row.put("time_unix_nano", System.nanoTime()); + row.put("severity_text", "INFO"); + row.put("body", "\"Test log message\""); + row.put("trace_id", "trace123"); + row.put("span_id", "span456"); + + return List.of(row); + } } \ No newline at end of file diff --git a/script/application.yml b/script/application.yml index 8d36940e2c..e141b4606c 100644 --- a/script/application.yml +++ b/script/application.yml @@ -209,7 +209,6 @@ warehouse: enabled: false grpc-endpoints: localhost:4001 http-endpoint: http://localhost:4000 - postgres-endpoint: localhost:4003 # if you config other database name, you should create them first database: public username: greptime diff --git a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/application.yml b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/application.yml index ea8347f19b..1bfe96d105 100644 --- a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/application.yml +++ b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/application.yml @@ -149,7 +149,6 @@ warehouse: enabled: true grpc-endpoints: greptime:4001 http-endpoint: http://greptime:4000 - postgres-endpoint: localhost:4003 # if you config other database name, you should create them first database: public username: greptime diff --git a/web-app/src/app/routes/log/log-manage/log-manage.component.html b/web-app/src/app/routes/log/log-manage/log-manage.component.html index 3ee454cd9d..cb95696ac9 100644 --- a/web-app/src/app/routes/log/log-manage/log-manage.component.html +++ b/web-app/src/app/routes/log/log-manage/log-manage.component.html @@ -5,9 +5,9 @@ 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. @@ -23,7 +23,9 @@ + +
@@ -54,7 +56,6 @@ ERROR FATAL - @@ -100,7 +101,9 @@
+
+
@@ -176,6 +179,7 @@
+
@@ -196,6 +200,7 @@
+ +
+
@@ -346,6 +353,7 @@
+