add-D4Metric-7

src/main/java/com/demo/seecanalysisbackend/service/D4MetricsService.java

@@ -2,12 +2,8 @@ package com.demo.seecanalysisbackend.service;
 
 import com.demo.seecanalysisbackend.model.D4Metrics;
 import com.demo.seecanalysisbackend.model.LogEntry;
-import com.demo.seecanalysisbackend.repository.LogEntryRepository;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.data.domain.Page;
-import org.springframework.data.domain.PageRequest;
-import org.springframework.data.domain.Pageable;
 import org.springframework.stereotype.Service;
 
 import java.time.Duration;
@@ -22,8 +18,8 @@ import java.util.stream.Collectors;
 @RequiredArgsConstructor
 @Slf4j
 public class D4MetricsService {
-    
-    private final LogEntryRepository logEntryRepository;
+
+    private final ElasticsearchRawQueryService elasticsearchRawQueryService;
     
     private static final Set<String> D4_EVENTS = Set.of(
         "D4_TASK_ASSIGNED", "D4_TASK_COMPLETED", "D4_COLLAB_EFFECTIVE",
@@ -139,22 +135,12 @@ public class D4MetricsService {
     }
 
     private List<LogEntry> loadAllD4Logs(LocalDateTime startTime, LocalDateTime endTime) {
-        Instant startInstant = toUtcInstant(startTime);
-        Instant endInstant = toUtcInstant(endTime);
-        List<LogEntry> allLogs = new ArrayList<>();
-        List<String> d4Events = new ArrayList<>(D4_EVENTS);
-        int page = 0;
-
-        while (true) {
-            Pageable pageable = PageRequest.of(page, D4_QUERY_PAGE_SIZE);
-            Page<LogEntry> pageResult = logEntryRepository.findByTimestampBetweenAndBizNameIn(startInstant, endInstant, d4Events, pageable);
-            allLogs.addAll(pageResult.getContent());
-
-            if (!pageResult.hasNext()) {
-                break;
-            }
-            page++;
-        }
+        List<LogEntry> allLogs = elasticsearchRawQueryService.fetchByBizNames(
+            startTime,
+            endTime,
+            new ArrayList<>(D4_EVENTS),
+            D4_QUERY_PAGE_SIZE
+        );
 
         allLogs.sort(Comparator.comparing(LogEntry::getTimestamp, Comparator.nullsLast(Comparator.reverseOrder())));
 
@@ -456,10 +442,6 @@ public class D4MetricsService {
         return logEntry.getContext();
     }
 
-    private Instant toUtcInstant(LocalDateTime dateTime) {
-        return dateTime.atZone(ZoneOffset.UTC).toInstant();
-    }
-
     private LocalDateTime toUtcLocalDateTime(Instant instant) {
         return instant == null ? null : LocalDateTime.ofInstant(instant, ZoneOffset.UTC);
     }

src/main/java/com/demo/seecanalysisbackend/service/ElasticsearchRawQueryService.java

@@ -0,0 +1,303 @@
+package com.demo.seecanalysisbackend.service;
+
+import com.demo.seecanalysisbackend.dto.LogSearchRequest;
+import com.demo.seecanalysisbackend.dto.LogSearchResponse;
+import com.demo.seecanalysisbackend.model.LogEntry;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.*;
+import org.springframework.stereotype.Service;
+import org.springframework.web.client.RestTemplate;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.*;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class ElasticsearchRawQueryService {
+
+    private final ObjectMapper objectMapper;
+
+    @Value("${spring.elasticsearch.uris}")
+    private String uris;
+
+    @Value("${spring.elasticsearch.username:}")
+    private String username;
+
+    @Value("${spring.elasticsearch.password:}")
+    private String password;
+
+    private static final String INDEX_PATTERN = "k8s-logs-*";
+
+    public LogSearchResponse searchLogs(LogSearchRequest request) {
+        int page = request.getPage() == null ? 0 : request.getPage();
+        int size = request.getSize() == null ? 20 : request.getSize();
+        int from = page * size;
+
+        Map<String, Object> queryBody = buildLogSearchBody(request, from, size);
+        JsonNode root = executeSearch(queryBody);
+
+        JsonNode hitsNode = root.path("hits");
+        long total = extractTotal(hitsNode.path("total"));
+
+        List<LogSearchResponse.LogEntryDto> logs = new ArrayList<>();
+        for (JsonNode hit : hitsNode.path("hits")) {
+            logs.add(toDto(hit));
+        }
+
+        int totalPages = size > 0 ? (int) Math.ceil((double) total / size) : 0;
+
+        return LogSearchResponse.builder()
+            .logs(logs)
+            .totalCount(total)
+            .currentPage(page)
+            .pageSize(size)
+            .totalPages(totalPages)
+            .hasNext(page + 1 < totalPages)
+            .hasPrevious(page > 0)
+            .build();
+    }
+
+    public List<LogEntry> fetchByBizNames(LocalDateTime startTime, LocalDateTime endTime, List<String> bizNames, int pageSize) {
+        List<LogEntry> result = new ArrayList<>();
+        int from = 0;
+
+        while (true) {
+            Map<String, Object> body = new HashMap<>();
+            body.put("from", from);
+            body.put("size", pageSize);
+            body.put("track_total_hits", true);
+            body.put("sort", List.of(Map.of("@timestamp", Map.of("order", "desc"))));
+
+            List<Map<String, Object>> filters = new ArrayList<>();
+            filters.add(Map.of("range", Map.of("@timestamp", Map.of(
+                "gte", toUtcInstant(startTime).toString(),
+                "lte", toUtcInstant(endTime).toString()
+            ))));
+
+            if (bizNames != null && !bizNames.isEmpty()) {
+                filters.add(Map.of("terms", Map.of("bizName.keyword", bizNames)));
+            }
+
+            body.put("query", Map.of("bool", Map.of("filter", filters)));
+
+            JsonNode root = executeSearch(body);
+            JsonNode hitArray = root.path("hits").path("hits");
+            if (!hitArray.isArray() || hitArray.isEmpty()) {
+                break;
+            }
+
+            for (JsonNode hit : hitArray) {
+                result.add(toLogEntry(hit));
+            }
+
+            if (hitArray.size() < pageSize) {
+                break;
+            }
+            from += pageSize;
+        }
+
+        return result;
+    }
+
+    private Map<String, Object> buildLogSearchBody(LogSearchRequest request, int from, int size) {
+        Map<String, Object> body = new HashMap<>();
+        body.put("from", from);
+        body.put("size", size);
+        body.put("track_total_hits", true);
+        body.put("sort", List.of(Map.of(toEsSortField(request.getSortBy()), Map.of(
+            "order", "asc".equalsIgnoreCase(request.getSortDirection()) ? "asc" : "desc"
+        ))));
+
+        List<Map<String, Object>> filters = new ArrayList<>();
+        filters.add(Map.of("range", Map.of("@timestamp", Map.of(
+            "gte", toUtcInstant(request.getStartTime()).toString(),
+            "lte", toUtcInstant(request.getEndTime()).toString()
+        ))));
+
+        if (request.getLevels() != null && !request.getLevels().isEmpty()) {
+            filters.add(Map.of("terms", Map.of("log_level.keyword", request.getLevels())));
+        }
+        if (request.getApplications() != null && !request.getApplications().isEmpty()) {
+            filters.add(Map.of("terms", Map.of("application.keyword", request.getApplications())));
+        }
+        if (request.getServices() != null && !request.getServices().isEmpty()) {
+            filters.add(Map.of("terms", Map.of("service_name.keyword", request.getServices())));
+        }
+        if (request.getBizNames() != null && !request.getBizNames().isEmpty()) {
+            filters.add(Map.of("terms", Map.of("bizName.keyword", request.getBizNames())));
+        }
+        if (request.getTraceId() != null && !request.getTraceId().isBlank()) {
+            filters.add(Map.of("term", Map.of("trace_id.keyword", request.getTraceId())));
+        }
+
+        Map<String, Object> bool = new HashMap<>();
+        bool.put("filter", filters);
+
+        if (request.getSearchText() != null && !request.getSearchText().isBlank()) {
+            bool.put("must", List.of(Map.of("multi_match", Map.of(
+                "query", request.getSearchText(),
+                "fields", List.of("raw_message", "business_info", "method_location", "service_name", "trace_id", "bizName")
+            ))));
+        }
+
+        body.put("query", Map.of("bool", bool));
+        return body;
+    }
+
+    private JsonNode executeSearch(Map<String, Object> body) {
+        String endpoint = firstUri() + "/" + INDEX_PATTERN + "/_search";
+        String payload;
+        try {
+            payload = objectMapper.writeValueAsString(body);
+        } catch (JsonProcessingException e) {
+            throw new IllegalStateException("Failed to serialize ES query body", e);
+        }
+
+        RestTemplate restTemplate = new RestTemplate();
+        HttpHeaders headers = new HttpHeaders();
+        headers.setContentType(MediaType.APPLICATION_JSON);
+        headers.setAccept(List.of(MediaType.APPLICATION_JSON));
+
+        if (username != null && !username.isBlank()) {
+            headers.setBasicAuth(username, password == null ? "" : password);
+        }
+
+        HttpEntity<String> entity = new HttpEntity<>(payload, headers);
+        ResponseEntity<String> response = restTemplate.exchange(endpoint, HttpMethod.POST, entity, String.class);
+
+        try {
+            return objectMapper.readTree(response.getBody());
+        } catch (Exception e) {
+            throw new IllegalStateException("Failed to parse ES response", e);
+        }
+    }
+
+    private LogSearchResponse.LogEntryDto toDto(JsonNode hit) {
+        JsonNode source = hit.path("_source");
+        Map<String, Object> context = parseContext(source);
+        String timestamp = null;
+        Instant ts = parseInstant(source.path("@timestamp").asText(null));
+        if (ts != null) {
+            timestamp = DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(ts.atOffset(ZoneOffset.UTC).toLocalDateTime());
+        }
+
+        return LogSearchResponse.LogEntryDto.builder()
+            .id(hit.path("_id").asText(null))
+            .timestamp(timestamp)
+            .rawMessage(source.path("raw_message").asText(null))
+            .serviceName(source.path("service_name").asText(null))
+            .thread(source.path("process_name").asText(null))
+            .level(source.path("log_level").asText(null))
+            .className(source.path("method_location").asText(null))
+            .methodName(source.path("method_name").asText(null))
+            .traceId(source.path("trace_id").asText(null))
+            .spanId(source.path("span_id").asText(null))
+            .bizName(source.path("bizName").asText(null))
+            .context(context)
+            .build();
+    }
+
+    private LogEntry toLogEntry(JsonNode hit) {
+        JsonNode source = hit.path("_source");
+        return LogEntry.builder()
+            .id(hit.path("_id").asText(null))
+            .timestamp(parseInstant(source.path("@timestamp").asText(null)))
+            .rawMessage(source.path("raw_message").asText(null))
+            .businessInfo(source.path("business_info").asText(null))
+            .serviceName(source.path("service_name").asText(null))
+            .thread(source.path("process_name").asText(null))
+            .level(source.path("log_level").asText(null))
+            .application(source.path("application").asText(null))
+            .className(source.path("method_location").asText(null))
+            .methodName(source.path("method_name").asText(null))
+            .traceId(source.path("trace_id").asText(null))
+            .spanId(source.path("span_id").asText(null))
+            .bizName(source.path("bizName").asText(null))
+            .context(parseContext(source))
+            .build();
+    }
+
+    private Map<String, Object> parseContext(JsonNode source) {
+        if (source.has("context") && source.get("context").isObject()) {
+            return objectMapper.convertValue(source.get("context"), new TypeReference<Map<String, Object>>() {});
+        }
+
+        String businessInfo = source.path("business_info").asText(null);
+        if (businessInfo == null || businessInfo.isBlank()) {
+            return new HashMap<>();
+        }
+
+        Map<String, Object> context = new HashMap<>();
+        String[] pairs = businessInfo.split("\\|");
+        for (String pair : pairs) {
+            int index = pair.indexOf('=');
+            if (index <= 0 || index == pair.length() - 1) {
+                continue;
+            }
+            String key = pair.substring(0, index).trim();
+            String value = pair.substring(index + 1).trim();
+            if (!key.isEmpty() && !value.isEmpty() && !"null".equalsIgnoreCase(value)) {
+                context.put(key, value);
+            }
+        }
+        return context;
+    }
+
+    private long extractTotal(JsonNode totalNode) {
+        if (totalNode == null || totalNode.isMissingNode()) {
+            return 0L;
+        }
+        if (totalNode.isNumber()) {
+            return totalNode.asLong();
+        }
+        return totalNode.path("value").asLong(0L);
+    }
+
+    private String toEsSortField(String sortBy) {
+        if (sortBy == null || sortBy.isBlank()) {
+            return "@timestamp";
+        }
+        return switch (sortBy) {
+            case "timestamp", "@timestamp" -> "@timestamp";
+            case "level", "log_level" -> "log_level.keyword";
+            case "serviceName", "service_name" -> "service_name.keyword";
+            case "traceId", "trace_id" -> "trace_id.keyword";
+            case "spanId", "span_id" -> "span_id.keyword";
+            case "bizName", "biz_name" -> "bizName.keyword";
+            default -> "@timestamp";
+        };
+    }
+
+    private String firstUri() {
+        return Arrays.stream(uris.split(","))
+            .map(String::trim)
+            .filter(s -> !s.isEmpty())
+            .findFirst()
+            .orElseThrow(() -> new IllegalStateException("spring.elasticsearch.uris is empty"));
+    }
+
+    private Instant toUtcInstant(LocalDateTime time) {
+        return time.atZone(ZoneOffset.UTC).toInstant();
+    }
+
+    private Instant parseInstant(String value) {
+        if (value == null || value.isBlank()) {
+            return null;
+        }
+        try {
+            return Instant.parse(value);
+        } catch (Exception ignored) {
+            return null;
+        }
+    }
+}

src/main/java/com/demo/seecanalysisbackend/service/LogSearchService.java

@@ -2,155 +2,19 @@ package com.demo.seecanalysisbackend.service;
 
 import com.demo.seecanalysisbackend.dto.LogSearchRequest;
 import com.demo.seecanalysisbackend.dto.LogSearchResponse;
-import com.demo.seecanalysisbackend.model.LogEntry;
-import com.demo.seecanalysisbackend.repository.LogEntryRepository;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.data.domain.Page;
-import org.springframework.data.domain.PageRequest;
-import org.springframework.data.domain.Pageable;
-import org.springframework.data.domain.Sort;
 import org.springframework.stereotype.Service;
 
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
-import java.time.format.DateTimeFormatter;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
-
 @Service
 @RequiredArgsConstructor
 @Slf4j
 public class LogSearchService {
-    
-    private final LogEntryRepository logEntryRepository;
-    private static final Map<String, String> SORT_FIELD_MAPPING = createSortFieldMapping();
-    private static final Set<String> ALLOWED_SORT_PROPERTIES = Set.of(
-        "timestamp", "level", "serviceName", "traceId", "spanId", "bizName",
-        "className", "methodName", "thread", "application", "rawMessage"
-    );
+
+    private final ElasticsearchRawQueryService elasticsearchRawQueryService;
     
     public LogSearchResponse searchLogs(LogSearchRequest request) {
         log.info("Searching logs with criteria: {}", request);
-        
-        Pageable pageable = createPageable(request);
-        Page<LogEntry> logPage = performSearch(request, pageable);
-        
-        List<LogSearchResponse.LogEntryDto> logDtos = logPage.getContent().stream()
-            .map(this::convertToDto)
-            .collect(Collectors.toList());
-        
-        return LogSearchResponse.builder()
-            .logs(logDtos)
-            .totalCount(logPage.getTotalElements())
-            .currentPage(logPage.getNumber())
-            .pageSize(logPage.getSize())
-            .totalPages(logPage.getTotalPages())
-            .hasNext(logPage.hasNext())
-            .hasPrevious(logPage.hasPrevious())
-            .build();
-    }
-    
-    private Page<LogEntry> performSearch(LogSearchRequest request, Pageable pageable) {
-        Instant startTime = toUtcInstant(request.getStartTime());
-        Instant endTime = toUtcInstant(request.getEndTime());
-
-        if (request.getSearchText() != null && !request.getSearchText().trim().isEmpty()) {
-            return logEntryRepository.findByTimestampBetweenAndMessage(
-                startTime, endTime,
-                request.getSearchText(), pageable);
-        }
-        
-        if (request.getBizNames() != null && !request.getBizNames().isEmpty()) {
-            return logEntryRepository.findByTimestampBetweenAndBizNameIn(
-                startTime, endTime,
-                request.getBizNames(), pageable);
-        }
-        
-        if (request.getServices() != null && !request.getServices().isEmpty()) {
-            String serviceName = request.getServices().getFirst();
-            return logEntryRepository.findByTimestampBetweenAndServiceName(
-                startTime, endTime,
-                serviceName, pageable);
-        }
-        
-        if (request.getTraceId() != null && !request.getTraceId().trim().isEmpty()) {
-            return logEntryRepository.findByTimestampBetweenAndTraceId(
-                startTime, endTime,
-                request.getTraceId(), pageable);
-        }
-        
-        if (request.getLevels() != null && !request.getLevels().isEmpty()) {
-            return logEntryRepository.findByTimestampBetweenAndLevelIn(
-                startTime, endTime,
-                request.getLevels(), pageable);
-        }
-        
-        if (request.getApplications() != null && !request.getApplications().isEmpty()) {
-            return logEntryRepository.findByTimestampBetweenAndApplicationIn(
-                startTime, endTime,
-                request.getApplications(), pageable);
-        }
-        
-        return logEntryRepository.findByTimestampBetween(
-            startTime, endTime, pageable);
-    }
-    
-    private Pageable createPageable(LogSearchRequest request) {
-        Sort.Direction direction = "asc".equalsIgnoreCase(request.getSortDirection()) ? 
-            Sort.Direction.ASC : Sort.Direction.DESC;
-
-        String requestedSortBy = request.getSortBy() == null ? "timestamp" : request.getSortBy();
-        String normalizedSortProperty = SORT_FIELD_MAPPING.getOrDefault(requestedSortBy, requestedSortBy);
-        String actualSortField = ALLOWED_SORT_PROPERTIES.contains(normalizedSortProperty)
-            ? normalizedSortProperty
-            : "timestamp";
-        
-        return PageRequest.of(request.getPage(), request.getSize(), 
-            Sort.by(direction, actualSortField));
-    }
-
-    private static Map<String, String> createSortFieldMapping() {
-        Map<String, String> mapping = new HashMap<>();
-        // Accept external ES field names and normalize them to entity property names.
-        mapping.put("@timestamp", "timestamp");
-        mapping.put("log_level", "level");
-        mapping.put("service_name", "serviceName");
-        mapping.put("trace_id", "traceId");
-        mapping.put("span_id", "spanId");
-        mapping.put("biz_name", "bizName");
-        mapping.put("bizName", "bizName");
-        mapping.put("timestamp", "timestamp");
-        mapping.put("level", "level");
-        mapping.put("serviceName", "serviceName");
-        mapping.put("traceId", "traceId");
-        mapping.put("spanId", "spanId");
-        return mapping;
-    }
-    
-    private LogSearchResponse.LogEntryDto convertToDto(LogEntry logEntry) {
-        return LogSearchResponse.LogEntryDto.builder()
-            .id(logEntry.getId())
-            .timestamp(logEntry.getTimestamp() != null ? 
-                DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(logEntry.getTimestamp().atOffset(ZoneOffset.UTC).toLocalDateTime()) : null)
-            .rawMessage(logEntry.getRawMessage())
-            .serviceName(logEntry.getServiceName())
-            .thread(logEntry.getThread())
-            .level(logEntry.getLevel())
-            .className(logEntry.getClassName())
-            .methodName(logEntry.getMethodName())
-            .traceId(logEntry.getTraceId())
-            .spanId(logEntry.getSpanId())
-            .bizName(logEntry.getBizName())
-            .context(logEntry.getContext())
-            .build();
-    }
-
-    private Instant toUtcInstant(LocalDateTime dateTime) {
-        return dateTime.atZone(ZoneOffset.UTC).toInstant();
+        return elasticsearchRawQueryService.searchLogs(request);
     }
 }