chore: import upstream snapshot with attribution

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/LogIngestionController.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.extern.slf4j.Slf4j;
+import java.util.List;
+
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.dto.Message;
+import org.apache.hertzbeat.log.service.LogProtocolAdapter;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Generic Log Ingestion Controller
+ * Provides a fallback endpoint for log protocols that don't have dedicated controllers.
+ * For OTLP protocol, use OtlpLogController instead.
+ */
+@Tag(name = "Log Ingestion Controller")
+@RestController
+@RequestMapping(path = "/api/logs", produces = MediaType.APPLICATION_JSON_VALUE)
+@Slf4j
+public class LogIngestionController {
+
+    private final List<LogProtocolAdapter> protocolAdapters;
+
+    public LogIngestionController(List<LogProtocolAdapter> protocolAdapters) {
+        this.protocolAdapters = protocolAdapters;
+    }
+
+    /**
+     * Receive log payload pushed from external system specifying the log protocol.
+     *
+     * @param protocol log protocol identifier (e.g., "vector", "loki")
+     * @param content  raw request body
+     */
+    @Operation(summary = "Ingest logs by protocol name")
+    @PostMapping(value = "/ingest/{protocol}", consumes = MediaType.APPLICATION_JSON_VALUE)
+    public ResponseEntity<Message<Void>> ingestLog(@PathVariable("protocol") String protocol,
+                                                   @RequestBody String content) {
+        log.debug("Receive log from protocol: {}, content length: {}", protocol, content == null ? 0 : content.length());
+        
+        for (LogProtocolAdapter adapter : protocolAdapters) {
+            if (adapter.supportProtocol().equalsIgnoreCase(protocol)) {
+                try {
+                    adapter.ingest(content);
+                    return ResponseEntity.ok(Message.success("Add extern log success"));
+                } catch (Exception e) {
+                    log.error("Add log failed: {}", e.getMessage(), e);
+                    return ResponseEntity.status(HttpStatus.BAD_REQUEST)
+                            .body(Message.fail(CommonConstants.FAIL_CODE, "Add extern log failed: " + e.getMessage()));
+                }
+            }
+        }
+        log.warn("Not support extern log from protocol: {}", protocol);
+        return ResponseEntity.status(HttpStatus.BAD_REQUEST)
+                .body(Message.fail(CommonConstants.FAIL_CODE, "Not support the " + protocol + " protocol log"));
+    }
+}

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/LogManagerController.java

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.common.entity.dto.Message;
+import org.apache.hertzbeat.warehouse.store.history.tsdb.HistoryDataWriter;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+import static org.apache.hertzbeat.common.constants.CommonConstants.FAIL_CODE;
+
+/**
+ * Controller for managing log entries in HertzBeat.
+ */
+@RestController
+@RequestMapping(path = "/api/logs", produces = "application/json")
+@Tag(name = "Log Management Controller")
+@Slf4j
+public class LogManagerController {
+
+    private final HistoryDataWriter historyDataWriter;
+
+    public LogManagerController(HistoryDataWriter historyDataWriter) {
+        this.historyDataWriter = historyDataWriter;
+    }
+
+    @DeleteMapping
+    @Operation(summary = "Batch delete logs", 
+               description = "Batch delete logs by time timestamps. Deletes multiple log entries based on their Unix nanosecond timestamps.")
+    public ResponseEntity<Message<String>> batchDelete(
+            @Parameter(description = "List of Unix nanosecond timestamps for logs to delete", example = "1640995200000000000") 
+            @RequestParam(required = false) List<Long> timeUnixNanos) {
+        boolean result = historyDataWriter.batchDeleteLogs(timeUnixNanos);
+        if (result) {
+            return ResponseEntity.ok(Message.success("Logs deleted successfully"));
+        } else {
+            return ResponseEntity.ok(Message.fail(FAIL_CODE, "Failed to delete logs"));
+        }
+    }
+}

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/LogQueryController.java

@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.tags.Tag;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.hertzbeat.warehouse.store.history.tsdb.HistoryDataReader;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageImpl;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.common.entity.dto.Message;
+import org.apache.hertzbeat.common.entity.log.LogEntry;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Log query and statistics APIs for UI consumption
+ */
+@RestController
+@RequestMapping(path = "/api/logs", produces = "application/json")
+@Tag(name = "Log Query Controller")
+@Slf4j
+public class LogQueryController {
+
+    private final HistoryDataReader historyDataReader;
+
+    public LogQueryController(HistoryDataReader historyDataReader) {
+        this.historyDataReader = historyDataReader;
+    }
+
+    @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.")
+    public ResponseEntity<Message<Page<LogEntry>>> list(
+            @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")
+            @RequestParam(value = "end", required = false) Long end,
+            @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")
+            @RequestParam(value = "spanId", required = false) String spanId,
+            @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,
+            @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")
+            @RequestParam(value = "pageSize", required = false, defaultValue = "20") Integer pageSize) {
+        Page<LogEntry> result = getPagedLogs(start, end, traceId, spanId, severityNumber, severityText, search, 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.")
+    public ResponseEntity<Message<Map<String, Object>>> overviewStats(
+            @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")
+            @RequestParam(value = "end", required = false) Long end,
+            @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")
+            @RequestParam(value = "spanId", required = false) String spanId,
+            @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<LogEntry> logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText, search);
+
+        Map<String, Object> 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();
+        long errorCount = logs.stream().filter(log -> log.getSeverityNumber() != null && log.getSeverityNumber() >= 17 && log.getSeverityNumber() <= 20).count();
+        long warnCount = logs.stream().filter(log -> log.getSeverityNumber() != null && log.getSeverityNumber() >= 13 && log.getSeverityNumber() <= 16).count();
+        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.")
+    public ResponseEntity<Message<Map<String, Object>>> traceCoverageStats(
+            @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")
+            @RequestParam(value = "end", required = false) Long end,
+            @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")
+            @RequestParam(value = "spanId", required = false) String spanId,
+            @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<LogEntry> logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText, search);
+
+        Map<String, Object> 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 withoutTrace = logs.size() - withTraceId;
+
+        Map<String, Long> 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.")
+    public ResponseEntity<Message<Map<String, Object>>> trendStats(
+            @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")
+            @RequestParam(value = "end", required = false) Long end,
+            @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")
+            @RequestParam(value = "spanId", required = false) String spanId,
+            @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<LogEntry> logs = getFilteredLogs(start, end, traceId, spanId, severityNumber, severityText, search);
+
+        // Group by hour
+        Map<String, Long> 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()));
+
+        Map<String, Object> result = new HashMap<>();
+        result.put("hourlyStats", hourlyStats);
+        return ResponseEntity.ok(Message.success(result));
+    }
+
+    private List<LogEntry> getFilteredLogs(Long start, Long end, String traceId, String spanId,
+                                           Integer severityNumber, String severityText, String search) {
+        // Use the new multi-condition query method
+        return historyDataReader.queryLogsByMultipleConditions(start, end, traceId, spanId, severityNumber, severityText, search);
+    }
+
+    private Page<LogEntry> getPagedLogs(Long start, Long end, String traceId, String spanId,
+                                        Integer severityNumber, String severityText, String search,
+                                        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);
+        List<LogEntry> pagedLogs = historyDataReader.queryLogsByMultipleConditionsWithPagination(
+                start, end, traceId, spanId, severityNumber, severityText, search, 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);
+    }
+}

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/LogSseController.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import static org.springframework.http.MediaType.TEXT_EVENT_STREAM_VALUE;
+import org.apache.hertzbeat.common.util.SnowFlakeIdGenerator;
+import org.apache.hertzbeat.log.notice.LogSseFilterCriteria;
+import org.apache.hertzbeat.log.notice.LogSseManager;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.ModelAttribute;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+import io.swagger.v3.oas.annotations.Operation;
+
+/**
+ * SSE controller for log streaming with filtering support
+ */
+@RestController
+@RequestMapping(path = "/api/logs/sse", produces = {TEXT_EVENT_STREAM_VALUE})
+public class LogSseController {
+
+    private final LogSseManager emitterManager;
+
+    public LogSseController(LogSseManager emitterManager) {
+        this.emitterManager = emitterManager;
+    }
+
+    /**
+     * Subscribe to log events with optional filtering
+     * @param filterCriteria Filter criteria for log events (all parameters are optional)
+     * @return SSE emitter for streaming log events
+     */
+    @GetMapping(path = "/subscribe")
+    @Operation(summary = "Subscribe to log events with optional filtering", description = "Subscribe to log events with optional filtering")
+    public SseEmitter subscribe(@ModelAttribute LogSseFilterCriteria filterCriteria) {
+        Long clientId = SnowFlakeIdGenerator.generateId();
+        return emitterManager.createEmitter(clientId, filterCriteria);
+    }
+} 

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/controller/OtlpLogController.java

@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import com.fasterxml.jackson.core.io.JsonStringEncoder;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.JsonFormat;
+import com.google.rpc.Status;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceResponse;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.log.service.impl.OtlpLogProtocolAdapter;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * OTLP Log Ingestion Controller
+ * Implements OTLP/HTTP specification for log ingestion.
+ * Supports both binary-encoded Protobuf (application/x-protobuf) and JSON-encoded Protobuf (application/json).
+ *
+ * @see <a href="https://opentelemetry.io/docs/specs/otlp/#otlphttp">OTLP/HTTP Specification</a>
+ */
+@Tag(name = "OTLP Log Controller")
+@RestController
+@RequestMapping(path = "/api/logs/otlp")
+@Slf4j
+public class OtlpLogController {
+
+    private static final String CONTENT_TYPE_PROTOBUF = "application/x-protobuf";
+
+    private static final ExportLogsServiceResponse EMPTY_RESPONSE = ExportLogsServiceResponse.newBuilder().build();
+
+    private final OtlpLogProtocolAdapter otlpLogProtocolAdapter;
+
+    public OtlpLogController(OtlpLogProtocolAdapter otlpLogProtocolAdapter) {
+        this.otlpLogProtocolAdapter = otlpLogProtocolAdapter;
+    }
+
+    /**
+     * OTLP/HTTP standard endpoint for logs with JSON-encoded Protobuf payload.
+     * Content-Type: application/json
+     *
+     * Response follows OTLP specification:
+     * - Success: HTTP 200 with ExportLogsServiceResponse (JSON encoded)
+     * - Failure: HTTP 400 with google.rpc.Status (JSON encoded)
+     *
+     * @param content JSON-encoded ExportLogsServiceRequest
+     * @return ExportLogsServiceResponse on success, Status on failure
+     */
+    @Operation(summary = "Ingest OTLP logs (JSON format)")
+    @PostMapping(value = "/v1/logs", consumes = MediaType.APPLICATION_JSON_VALUE, produces = MediaType.APPLICATION_JSON_VALUE)
+    public ResponseEntity<String> ingestJsonLogs(@RequestBody String content) {
+        log.debug("Receive OTLP JSON logs, content length: {}", content == null ? 0 : content.length());
+        try {
+            otlpLogProtocolAdapter.ingest(content);
+            return ResponseEntity.ok(toJsonResponse(EMPTY_RESPONSE));
+        } catch (IllegalArgumentException e) {
+            return ResponseEntity.status(HttpStatus.BAD_REQUEST)
+                    .body(toJsonErrorResponse(e.getMessage()));
+        } catch (Exception e) {
+            // Server-side errors - unexpected failure
+            log.error("Unexpected error ingesting OTLP JSON logs: {}", e.getMessage(), e);
+            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
+                    .body(toJsonErrorResponse(e.getMessage()));
+        }
+    }
+
+    /**
+     * OTLP/HTTP standard endpoint for logs with binary-encoded Protobuf payload.
+     * Content-Type: application/x-protobuf
+     *
+     * Response follows OTLP specification:
+     * - Success: HTTP 200 with ExportLogsServiceResponse (binary encoded)
+     * - Failure: HTTP 400 with google.rpc.Status (binary encoded)
+     *
+     * @param content binary-encoded ExportLogsServiceRequest
+     * @return ExportLogsServiceResponse on success, Status on failure
+     */
+    @Operation(summary = "Ingest OTLP logs (binary Protobuf format)")
+    @PostMapping(value = "/v1/logs", consumes = CONTENT_TYPE_PROTOBUF, produces = CONTENT_TYPE_PROTOBUF)
+    public ResponseEntity<byte[]> ingestBinaryLogs(@RequestBody byte[] content) {
+        log.debug("Receive OTLP binary logs, content length: {}", content == null ? 0 : content.length);
+        try {
+            otlpLogProtocolAdapter.ingestBinary(content);
+            return ResponseEntity.ok(EMPTY_RESPONSE.toByteArray());
+        } catch (IllegalArgumentException e) {
+            // Client-side validation errors - malformed request
+            return ResponseEntity.status(HttpStatus.BAD_REQUEST)
+                    .body(createBinaryErrorResponse(e.getMessage()));
+        } catch (Exception e) {
+            // Server-side errors - unexpected failure
+            log.error("Unexpected error ingesting OTLP binary logs: {}", e.getMessage(), e);
+            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
+                    .body(createBinaryErrorResponse(e.getMessage()));
+        }
+    }
+
+    private String toJsonResponse(ExportLogsServiceResponse response) {
+        try {
+            return JsonFormat.printer().print(response);
+        } catch (InvalidProtocolBufferException e) {
+            log.error("Failed to convert ExportLogsServiceResponse to JSON: {}", e.getMessage(), e);
+            return "{}";
+        }
+    }
+
+    private String toJsonErrorResponse(String message) {
+        Status status = Status.newBuilder()
+                .setMessage(message != null ? message : "Unknown error")
+                .build();
+        try {
+            return JsonFormat.printer().print(status);
+        } catch (InvalidProtocolBufferException e) {
+            return "{\"message\":\"" + escapeJson(message) + "\"}";
+        }
+    }
+
+    /**
+     * Escapes a string value for safe inclusion in JSON.
+     *
+     * @param message the string to escape
+     * @return the escaped string, or empty string if message is null
+     */
+    private String escapeJson(String message) {
+        if (message == null) {
+            return "";
+        }
+
+        char[] escaped = JsonStringEncoder.getInstance().quoteAsString(message);
+        return new String(escaped);
+    }
+
+    private byte[] createBinaryErrorResponse(String message) {
+        return Status.newBuilder()
+                .setMessage(message != null ? message : "Unknown error")
+                .build()
+                .toByteArray();
+    }
+}

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/notice/LogSseFilterCriteria.java

@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hertzbeat.log.notice;
+
+import lombok.Data;
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
+import org.apache.hertzbeat.common.entity.log.LogEntry;
+import org.springframework.util.StringUtils;
+import io.swagger.v3.oas.annotations.media.Schema;
+import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
+
+/**
+ * Log filtering criteria for SSE (Server-Sent Events) log streaming
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Schema(description = "Log filtering criteria for SSE (Server-Sent Events) log streaming")
+public class LogSseFilterCriteria {
+    /**
+     * Numerical value of the severity.
+     * Smaller numerical values correspond to less severe events (such as debug events),
+     * larger numerical values correspond to more severe events (such as errors and critical events).
+     */
+    @Schema(description = "Numerical value of the severity.", example = "1", accessMode = READ_WRITE)
+    private Integer severityNumber;
+
+    /**
+     * The severity text (also known as log level).
+     * This is the original string representation of the severity as it is known at the source.
+     */
+    @Schema(description = "The severity text (also known as log level).", example = "INFO", accessMode = READ_WRITE)
+    private String severityText;
+
+    /**
+     * Log content text filtering (case-insensitive contains match).
+     */
+    @Schema(description = "Log content text filtering", example = "error occurred", accessMode = READ_WRITE)
+    private String logContent;
+
+    /**
+     * A unique identifier for a trace.
+     * All spans from the same trace share the same trace_id.
+     * The ID is a 16-byte array represented as a hex string.
+     */
+    @Schema(description = "A unique identifier for a trace.", example = "1234567890", accessMode = READ_WRITE)
+    private String traceId;
+
+    /**
+     * A unique identifier for a span within a trace.
+     * The ID is an 8-byte array represented as a hex string.
+     */
+    @Schema(description = "A unique identifier for a span.", example = "1234567890", accessMode = READ_WRITE)
+    private String spanId;
+
+
+    /**
+     * Core filtering logic to determine if a log entry matches the criteria
+     * @param log Log entry to be checked
+     * @return boolean Whether the log entry matches the filter criteria
+     */
+    public boolean matches(LogEntry log) {
+        if (log == null) return false;
+        // Check severity text match
+        if (StringUtils.hasText(severityText) && !severityText.equalsIgnoreCase(log.getSeverityText())) {
+            return false;
+        }
+
+        // Check severity number match (if both are present)
+        if (severityNumber != null && log.getSeverityNumber() != null
+                && !severityNumber.equals(log.getSeverityNumber())) {
+            return false;
+        }
+
+        // Check log content match
+        if (StringUtils.hasText(logContent)) {
+            Object body = log.getBody();
+            if (body == null) {
+                return false;
+            }
+            String bodyStr = body.toString();
+            if (!StringUtils.hasText(bodyStr) || !bodyStr.toLowerCase().contains(logContent.toLowerCase())) {
+                return false;
+            }
+        }
+
+        // Check trace ID match
+        if (StringUtils.hasText(traceId) && !traceId.equalsIgnoreCase(log.getTraceId())) {
+            return false;
+        }
+
+        // Check span ID match
+        if (StringUtils.hasText(spanId) && !spanId.equalsIgnoreCase(log.getSpanId())) {
+            return false;
+        }
+        return true;
+    }
+}

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/notice/LogSseManager.java

@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hertzbeat.log.notice;
+
+import jakarta.annotation.PreDestroy;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.common.entity.log.LogEntry;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * SSE manager for log with batch processing support for high TPS scenarios
+ */
+@Component
+@Slf4j
+@Getter
+public class LogSseManager {
+    
+    private static final long BATCH_INTERVAL_MS = 200;
+    private static final int MAX_BATCH_SIZE = 1000;
+    private static final int MAX_QUEUE_SIZE = 10000;
+    
+    private final Map<Long, SseSubscriber> emitters = new ConcurrentHashMap<>();
+    private final Queue<LogEntry> logQueue = new ConcurrentLinkedQueue<>();
+    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+        Thread t = new Thread(r, "sse-batch-scheduler");
+        t.setDaemon(true);
+        return t;
+    });
+    private final ExecutorService senderPool = Executors.newThreadPerTaskExecutor(Thread.ofVirtual()
+            .name("sse-sender-", 0)
+            .uncaughtExceptionHandler((thread, throwable) ->
+                    log.error("SSE sender has uncaughtException.", throwable))
+            .factory());
+    private final AtomicLong queueSize = new AtomicLong(0);
+
+    public LogSseManager() {
+        scheduler.scheduleAtFixedRate(this::flushBatch, BATCH_INTERVAL_MS, BATCH_INTERVAL_MS, TimeUnit.MILLISECONDS);
+    }
+
+    @PreDestroy
+    public void shutdown() {
+        scheduler.shutdown();
+        senderPool.shutdown();
+        try {
+            scheduler.awaitTermination(2, TimeUnit.SECONDS);
+            senderPool.awaitTermination(2, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+        scheduler.shutdownNow();
+        senderPool.shutdownNow();
+    }
+
+    /**
+     * Create a new SSE emitter for a client with specified filters
+     */
+    public SseEmitter createEmitter(Long clientId, LogSseFilterCriteria filters) {
+        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
+        emitter.onCompletion(() -> removeEmitter(clientId));
+        emitter.onTimeout(() -> removeEmitter(clientId));
+        emitter.onError((ex) -> removeEmitter(clientId));
+
+        emitters.put(clientId, new SseSubscriber(emitter, filters));
+        return emitter;
+    }
+
+    /**
+     * Queue log entry for batch processing
+     */
+    public void broadcast(LogEntry logEntry) {
+        if (queueSize.incrementAndGet() > MAX_QUEUE_SIZE) {
+            queueSize.decrementAndGet();
+            return;
+        }
+        boolean offered = logQueue.offer(logEntry);
+        if (!offered) {
+            queueSize.decrementAndGet();
+            log.warn("Failed to enqueue log entry: {}", logEntry);
+        }
+    }
+
+    /**
+     * Flush queued logs to all subscribers in batch
+     */
+    private void flushBatch() {
+        try {
+            if (logQueue.isEmpty() || emitters.isEmpty()) {
+                return;
+            }
+
+            List<LogEntry> batch = new ArrayList<>(MAX_BATCH_SIZE);
+            LogEntry entry;
+            while (batch.size() < MAX_BATCH_SIZE && (entry = logQueue.poll()) != null) {
+                batch.add(entry);
+                queueSize.decrementAndGet();
+            }
+
+            if (batch.isEmpty()) {
+                return;
+            }
+
+            // Send to each subscriber in parallel
+            for (Map.Entry<Long, SseSubscriber> e : emitters.entrySet()) {
+                Long clientId = e.getKey();
+                SseSubscriber subscriber = e.getValue();
+                List<LogEntry> filtered = filterLogs(batch, subscriber.filters);
+                if (!filtered.isEmpty()) {
+                    senderPool.submit(() -> sendToSubscriber(clientId, subscriber.emitter, filtered));
+                }
+            }
+        } catch (Exception e) {
+            log.error("Error in flushBatch: {}", e.getMessage(), e);
+        }
+    }
+
+    private void sendToSubscriber(Long clientId, SseEmitter emitter, List<LogEntry> logs) {
+        try {
+            long batchTimestamp = System.currentTimeMillis();
+            int sequenceNumber = 0;
+            for (LogEntry logEntry : logs) {
+                String eventId = batchTimestamp + "-" + sequenceNumber++;
+                emitter.send(SseEmitter.event()
+                        .id(eventId)
+                        .name("LOG_EVENT")
+                        .data(logEntry));
+            }
+        } catch (IOException | IllegalStateException e) {
+            safeComplete(clientId, emitter);
+        } catch (Exception e) {
+            log.error("Failed to send to client {}: {}", clientId, e.getMessage());
+            safeComplete(clientId, emitter);
+        }
+    }
+
+    private void safeComplete(Long clientId, SseEmitter emitter) {
+        try {
+            emitter.complete();
+        } catch (Exception ignored) {
+        }
+        removeEmitter(clientId);
+    }
+
+    private List<LogEntry> filterLogs(List<LogEntry> logs, LogSseFilterCriteria filters) {
+        if (filters == null) {
+            return logs;
+        }
+        List<LogEntry> filtered = new ArrayList<>();
+        for (LogEntry log : logs) {
+            if (filters.matches(log)) {
+                filtered.add(log);
+            }
+        }
+        return filtered;
+    }
+
+    private void removeEmitter(Long clientId) {
+        emitters.remove(clientId);
+    }
+
+    public long getQueueSize() {
+        return queueSize.get();
+    }
+
+    /**
+     * SseSubscriber for SseEmitter and LogSseFilterCriteria
+     */
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class SseSubscriber {
+        private SseEmitter emitter;
+        private LogSseFilterCriteria filters;
+    }
+}

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/service/LogProtocolAdapter.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.service;
+
+/**
+ * Adapter interface for ingesting logs pushed via different protocols
+ * (e.g. OTLP, Loki, Filebeat, Vector).
+ * Implementations should:
+ *   1. Parse raw HTTP payload of their protocol.
+ *   2. Convert data to LogEntry.
+ *   3. Forward / persist it to downstream pipeline.
+ */
+public interface LogProtocolAdapter {
+
+    /**
+     * Ingest log payload pushed from external system.
+     *
+     * @param content raw request body string
+     */
+    void ingest(String content);
+
+    /**
+     * Identifier of the protocol this adapter supports ("otlp", "vector", etc.)
+     */
+    String supportProtocol();
+} 

hertzbeat-log/src/main/java/org/apache/hertzbeat/log/service/impl/OtlpLogProtocolAdapter.java

@@ -0,0 +1,281 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.service.impl;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.JsonFormat;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
+import io.opentelemetry.proto.common.v1.AnyValue;
+import io.opentelemetry.proto.common.v1.KeyValue;
+import io.opentelemetry.proto.common.v1.KeyValueList;
+import io.opentelemetry.proto.logs.v1.LogRecord;
+import io.opentelemetry.proto.logs.v1.ResourceLogs;
+import io.opentelemetry.proto.logs.v1.ScopeLogs;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.common.entity.log.LogEntry;
+import org.apache.hertzbeat.common.queue.CommonDataQueue;
+import org.apache.hertzbeat.log.notice.LogSseManager;
+import org.apache.hertzbeat.log.service.LogProtocolAdapter;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Adapter for OpenTelemetry OTLP/HTTP log ingestion.
+ * Supports both JSON-encoded and binary-encoded Protobuf formats.
+ * 
+ * @see <a href="https://opentelemetry.io/docs/specs/otlp/#otlphttp">OTLP/HTTP Specification</a>
+ */
+@Slf4j
+@Service
+public class OtlpLogProtocolAdapter implements LogProtocolAdapter {
+
+    private static final String PROTOCOL_NAME = "otlp";
+
+    private final CommonDataQueue commonDataQueue;
+    private final LogSseManager logSseManager;
+
+    public OtlpLogProtocolAdapter(CommonDataQueue commonDataQueue, LogSseManager logSseManager) {
+        this.commonDataQueue = commonDataQueue;
+        this.logSseManager = logSseManager;
+    }
+
+    @Override
+    public void ingest(String content) {
+        if (content == null || content.isEmpty()) {
+            log.warn("Received empty OTLP JSON log payload - skip processing.");
+            return;
+        }
+        ExportLogsServiceRequest.Builder builder = ExportLogsServiceRequest.newBuilder();
+        try {
+            JsonFormat.parser().ignoringUnknownFields().merge(content, builder);
+            ExportLogsServiceRequest request = builder.build();
+            processLogsRequest(request, "JSON");
+        } catch (InvalidProtocolBufferException e) {
+            log.error("Failed to parse OTLP JSON log payload: {}", e.getMessage());
+            throw new IllegalArgumentException("Invalid OTLP JSON log content", e);
+        }
+    }
+
+    /**
+     * Ingest binary-encoded Protobuf log payload (OTLP-specific).
+     *
+     * @param content binary-encoded ExportLogsServiceRequest
+     */
+    public void ingestBinary(byte[] content) {
+        if (content == null || content.length == 0) {
+            log.warn("Received empty OTLP binary log payload - skip processing.");
+            return;
+        }
+        try {
+            ExportLogsServiceRequest request = ExportLogsServiceRequest.parseFrom(content);
+            processLogsRequest(request, "binary");
+        } catch (InvalidProtocolBufferException e) {
+            log.error("Failed to parse OTLP binary log payload: {}", e.getMessage());
+            throw new IllegalArgumentException("Invalid OTLP binary log content", e);
+        }
+    }
+
+    private void processLogsRequest(ExportLogsServiceRequest request, String format) {
+        List<LogEntry> logEntries = extractLogEntries(request);
+        log.debug("Successfully extracted {} log entries from OTLP {} payload", logEntries.size(), format);
+        commonDataQueue.sendLogEntryToStorageBatch(logEntries);
+        commonDataQueue.sendLogEntryToAlertBatch(logEntries);
+        logEntries.forEach(logSseManager::broadcast);
+    }
+
+    /**
+     * Extract LogEntry instances from ExportLogsServiceRequest.
+     * 
+     * @param request the OTLP export logs service request
+     * @return list of extracted log entries
+     */
+    private List<LogEntry> extractLogEntries(ExportLogsServiceRequest request) {
+        List<LogEntry> logEntries = new ArrayList<>();
+        
+        for (ResourceLogs resourceLogs : request.getResourceLogsList()) {
+            // Extract resource attributes
+            Map<String, Object> resourceAttributes = extractAttributes(
+                resourceLogs.getResource().getAttributesList()
+            );
+            
+            for (ScopeLogs scopeLogs : resourceLogs.getScopeLogsList()) {
+                // Extract instrumentation scope information
+                LogEntry.InstrumentationScope instrumentationScope = extractInstrumentationScope(scopeLogs);
+                
+                for (LogRecord logRecord : scopeLogs.getLogRecordsList()) {
+                    LogEntry logEntry = convertLogRecordToLogEntry(
+                        logRecord, 
+                        resourceAttributes, 
+                        instrumentationScope
+                    );
+                    logEntries.add(logEntry);
+                }
+            }
+        }
+        
+        return logEntries;
+    }
+
+    /**
+     * Convert OpenTelemetry LogRecord to LogEntry.
+     */
+    private LogEntry convertLogRecordToLogEntry(
+            LogRecord logRecord, 
+            Map<String, Object> resourceAttributes,
+            LogEntry.InstrumentationScope instrumentationScope) {
+        
+        return LogEntry.builder()
+            .timeUnixNano(logRecord.getTimeUnixNano())
+            .observedTimeUnixNano(logRecord.getObservedTimeUnixNano())
+            .severityNumber(logRecord.getSeverityNumberValue())
+            .severityText(logRecord.getSeverityText())
+            .body(extractBody(logRecord.getBody()))
+            .attributes(extractAttributes(logRecord.getAttributesList()))
+            .droppedAttributesCount(logRecord.getDroppedAttributesCount())
+            .traceId(bytesToHex(logRecord.getTraceId().toByteArray()))
+            .spanId(bytesToHex(logRecord.getSpanId().toByteArray()))
+            .traceFlags(logRecord.getFlags())
+            .resource(resourceAttributes)
+            .instrumentationScope(instrumentationScope)
+            .build();
+    }
+
+    /**
+     * Extract instrumentation scope information from ScopeLogs.
+     */
+    private LogEntry.InstrumentationScope extractInstrumentationScope(ScopeLogs scopeLogs) {
+        if (!scopeLogs.hasScope()) {
+            return null;
+        }
+        
+        var scope = scopeLogs.getScope();
+        return LogEntry.InstrumentationScope.builder()
+            .name(scope.getName())
+            .version(scope.getVersion())
+            .attributes(extractAttributes(scope.getAttributesList()))
+            .droppedAttributesCount(scope.getDroppedAttributesCount())
+            .build();
+    }
+
+    /**
+     * Extract attributes from a list of KeyValue pairs.
+     */
+    private Map<String, Object> extractAttributes(List<KeyValue> keyValueList) {
+        if (keyValueList == null || keyValueList.isEmpty()) {
+            return new HashMap<>();
+        }
+
+        AnyValue anyValue = AnyValue.newBuilder()
+                .setKvlistValue(KeyValueList.newBuilder()
+                        .addAllValues(keyValueList)
+                        .build())
+                .build();
+        Object extractedAnyValue = extractAnyValue(anyValue);
+        if (extractedAnyValue instanceof Map<?, ?> genericMap) {
+            Map<String, Object> resultMap = new HashMap<>();
+            for (Map.Entry<?, ?> entry : genericMap.entrySet()) {
+                if (entry.getKey() instanceof String) {
+                    resultMap.put((String) entry.getKey(), entry.getValue());
+                }
+            }
+            return resultMap;
+        } else {
+            return new HashMap<>();
+        }
+    }
+
+    /**
+     * Extract body content from AnyValue.
+     */
+    private Object extractBody(AnyValue body) {
+        return extractAnyValue(body);
+    }
+
+    /**
+     * Extract value from OpenTelemetry AnyValue.
+     */
+    private Object extractAnyValue(AnyValue anyValue) {
+        switch (anyValue.getValueCase()) {
+            case STRING_VALUE:
+                return anyValue.getStringValue();
+            case BOOL_VALUE:
+                return anyValue.getBoolValue();
+            case INT_VALUE:
+                return anyValue.getIntValue();
+            case DOUBLE_VALUE:
+                return anyValue.getDoubleValue();
+            case ARRAY_VALUE:
+                List<Object> arrayList = new ArrayList<>();
+                for (AnyValue item : anyValue.getArrayValue().getValuesList()) {
+                    arrayList.add(extractAnyValue(item));
+                }
+                return arrayList;
+            case KVLIST_VALUE:
+                Map<String, Object> kvMap = new HashMap<>();
+                for (KeyValue kv : anyValue.getKvlistValue().getValuesList()) {
+                    kvMap.put(normalizeKey(kv.getKey()), extractAnyValue(kv.getValue()));
+                }
+                return kvMap;
+            case BYTES_VALUE:
+                return anyValue.getBytesValue().toByteArray();
+            case VALUE_NOT_SET:
+            default:
+                return null;
+        }
+    }
+
+    /**
+     * Normalize key by replacing dots and spaces with underscores.
+     * 
+     * @param key the original key
+     * @return normalized key with dots and spaces replaced by underscores
+     */
+    private String normalizeKey(String key) {
+        if (key == null) {
+            return null;
+        }
+        return key.replace(".", "_").replace(" ", "_");
+    }
+
+    /**
+     * Convert byte array to hex string.
+     */
+    private String bytesToHex(byte[] bytes) {
+        if (bytes == null || bytes.length == 0) {
+            return null;
+        }
+        StringBuilder hexString = new StringBuilder();
+        for (byte b : bytes) {
+            String hex = Integer.toHexString(0xff & b);
+            if (hex.length() == 1) {
+                hexString.append('0');
+            }
+            hexString.append(hex);
+        }
+        return hexString.toString();
+    }
+
+    @Override
+    public String supportProtocol() {
+        return PROTOCOL_NAME;
+    }
+} 

hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogIngestionControllerTest.java

@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.log.service.LogProtocolAdapter;
+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.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+
+/**
+ * Unit test for {@link LogIngestionController}
+ */
+@ExtendWith(MockitoExtension.class)
+class LogIngestionControllerTest {
+
+    private MockMvc mockMvc;
+
+    @Mock
+    private LogProtocolAdapter vectorAdapter;
+
+    private LogIngestionController logIngestionController;
+
+    @BeforeEach
+    void setUp() {
+        List<LogProtocolAdapter> adapters = Arrays.asList(vectorAdapter);
+        this.logIngestionController = new LogIngestionController(adapters);
+        this.mockMvc = MockMvcBuilders.standaloneSetup(logIngestionController).build();
+    }
+
+    @Test
+    void testIngestLogWithKnownProtocol() throws Exception {
+        String logContent = "{\"message\":\"Test log message\"}";
+
+        when(vectorAdapter.supportProtocol()).thenReturn("vector");
+        doNothing().when(vectorAdapter).ingest(anyString());
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .post("/api/logs/ingest/vector")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(logContent)
+        )
+                .andDo(print())
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
+                .andExpect(jsonPath("$.msg").value("Add extern log success"))
+                .andReturn();
+    }
+
+    @Test
+    void testIngestLogWithUnsupportedProtocol() throws Exception {
+        String logContent = "{\"message\":\"Unsupported protocol log\"}";
+
+        when(vectorAdapter.supportProtocol()).thenReturn("vector");
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .post("/api/logs/ingest/unsupported")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(logContent)
+        )
+                .andExpect(status().isBadRequest())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.FAIL_CODE))
+                .andExpect(jsonPath("$.msg").value("Not support the unsupported protocol log"));
+    }
+
+    @Test
+    void testIngestLogWithAdapterException() throws Exception {
+        String logContent = "{\"message\":\"Log message that will cause exception\"}";
+
+        when(vectorAdapter.supportProtocol()).thenReturn("vector");
+        doThrow(new IllegalArgumentException("Invalid log format")).when(vectorAdapter).ingest(anyString());
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .post("/api/logs/ingest/vector")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(logContent)
+        )
+                .andExpect(status().isBadRequest())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.FAIL_CODE))
+                .andExpect(jsonPath("$.msg").value("Add extern log failed: Invalid log format"));
+    }
+}

hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogManagerControllerTest.java

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
+
+import java.util.Arrays;
+
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.warehouse.store.history.tsdb.HistoryDataWriter;
+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.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+
+/**
+ * Unit test for {@link LogManagerController}
+ */
+@ExtendWith(MockitoExtension.class)
+class LogManagerControllerTest {
+
+    private MockMvc mockMvc;
+
+    @Mock
+    private HistoryDataWriter historyDataWriter;
+
+    private LogManagerController logManagerController;
+
+    @BeforeEach
+    void setUp() {
+        this.logManagerController = new LogManagerController(historyDataWriter);
+        this.mockMvc = MockMvcBuilders.standaloneSetup(logManagerController).build();
+    }
+
+    @Test
+    void testBatchDeleteLogsSuccess() throws Exception {
+        when(historyDataWriter.batchDeleteLogs(anyList())).thenReturn(true);
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .delete("/api/logs")
+                        .param("timeUnixNanos", "1734005477630000000", "1734005477640000000")
+        )
+                .andDo(print())
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
+                .andExpect(jsonPath("$.msg").value("Logs deleted successfully"))
+                .andReturn();
+    }
+
+    @Test
+    void testBatchDeleteLogsFailure() throws Exception {
+        when(historyDataWriter.batchDeleteLogs(anyList())).thenReturn(false);
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .delete("/api/logs")
+                        .param("timeUnixNanos", "1734005477630000000", "1734005477640000000")
+        )
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.FAIL_CODE))
+                .andExpect(jsonPath("$.msg").value("Failed to delete logs"));
+    }
+
+    @Test
+    void testBatchDeleteLogsWithSingleTimestamp() throws Exception {
+        when(historyDataWriter.batchDeleteLogs(Arrays.asList(1734005477630000000L))).thenReturn(true);
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .delete("/api/logs")
+                        .param("timeUnixNanos", "1734005477630000000")
+        )
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
+                .andExpect(jsonPath("$.msg").value("Logs deleted successfully"));
+    }
+}

hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogQueryControllerTest.java

@@ -0,0 +1,277 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.log.LogEntry;
+import org.apache.hertzbeat.warehouse.store.history.tsdb.HistoryDataReader;
+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.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+
+/**
+ * Unit test for {@link LogQueryController}
+ */
+@ExtendWith(MockitoExtension.class)
+class LogQueryControllerTest {
+
+    private MockMvc mockMvc;
+
+    @Mock
+    private HistoryDataReader historyDataReader;
+
+    private LogQueryController logQueryController;
+
+    @BeforeEach
+    void setUp() {
+        this.logQueryController = new LogQueryController(historyDataReader);
+        this.mockMvc = MockMvcBuilders.standaloneSetup(logQueryController).build();
+    }
+
+    @Test
+    void testListLogsWithAllFilters() throws Exception {
+        // Mock data
+        LogEntry logEntry1 = LogEntry.builder()
+                .timeUnixNano(1734005477630000000L)
+                .severityNumber(9)
+                .severityText("INFO")
+                .body("Test log message 1")
+                .traceId("trace123")
+                .spanId("span456")
+                .attributes(new HashMap<>())
+                .build();
+
+        LogEntry logEntry2 = LogEntry.builder()
+                .timeUnixNano(1734005477640000000L)
+                .severityNumber(17)
+                .severityText("ERROR")
+                .body("Test log message 2")
+                .traceId("trace123")
+                .spanId("span789")
+                .attributes(new HashMap<>())
+                .build();
+
+        List<LogEntry> mockLogs = Arrays.asList(logEntry1, logEntry2);
+
+        when(historyDataReader.countLogsByMultipleConditions(anyLong(), anyLong(), any(),
+                any(), any(), any(), any())).thenReturn(2L);
+        when(historyDataReader.queryLogsByMultipleConditionsWithPagination(anyLong(), anyLong(),
+                any(), any(), any(), any(), any(), 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")
+        )
+                .andDo(print())
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
+                .andExpect(jsonPath("$.data.content").isArray())
+                .andExpect(jsonPath("$.data.content.length()").value(2))
+                .andExpect(jsonPath("$.data.totalElements").value(2))
+                .andExpect(jsonPath("$.data.size").value(20))
+                .andExpect(jsonPath("$.data.number").value(0));
+    }
+
+    @Test
+    void testListLogsWithoutFilters() throws Exception {
+        List<LogEntry> mockLogs = Arrays.asList(
+                LogEntry.builder()
+                        .timeUnixNano(1734005477630000000L)
+                        .severityNumber(9)
+                        .severityText("INFO")
+                        .body("Test log message")
+                        .attributes(new HashMap<>())
+                        .build()
+        );
+
+        when(historyDataReader.countLogsByMultipleConditions(any(), any(), any(),
+                any(), any(), any(), any())).thenReturn(1L);
+        when(historyDataReader.queryLogsByMultipleConditionsWithPagination(any(), any(),
+                any(), any(), any(), any(), any(), eq(0), eq(20)))
+                .thenReturn(mockLogs);
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .get("/api/logs/list")
+        )
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
+                .andExpect(jsonPath("$.data.content").isArray())
+                .andExpect(jsonPath("$.data.content.length()").value(1));
+    }
+
+    @Test
+    void testOverviewStatsWithMixedSeverityLogs() throws Exception {
+        // Create logs with different severity levels according to OpenTelemetry standard
+        List<LogEntry> mockLogs = Arrays.asList(
+                // TRACE (1-4)
+                LogEntry.builder().severityNumber(2).build(),
+                // DEBUG (5-8)
+                LogEntry.builder().severityNumber(6).build(),
+                // INFO (9-12)
+                LogEntry.builder().severityNumber(9).build(),
+                LogEntry.builder().severityNumber(10).build(),
+                // WARN (13-16)
+                LogEntry.builder().severityNumber(14).build(),
+                // ERROR (17-20)
+                LogEntry.builder().severityNumber(17).build(),
+                LogEntry.builder().severityNumber(18).build(),
+                // FATAL (21-24)
+                LogEntry.builder().severityNumber(21).build()
+        );
+
+        when(historyDataReader.queryLogsByMultipleConditions(any(), any(), any(),
+                any(), any(), any(), any())).thenReturn(mockLogs);
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .get("/api/logs/stats/overview")
+        )
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
+                .andExpect(jsonPath("$.data.totalCount").value(8))
+                .andExpect(jsonPath("$.data.traceCount").value(1))
+                .andExpect(jsonPath("$.data.debugCount").value(1))
+                .andExpect(jsonPath("$.data.infoCount").value(2))
+                .andExpect(jsonPath("$.data.warnCount").value(1))
+                .andExpect(jsonPath("$.data.errorCount").value(2))
+                .andExpect(jsonPath("$.data.fatalCount").value(1));
+    }
+
+    @Test
+    void testOverviewStatsWithTimeRange() throws Exception {
+        List<LogEntry> mockLogs = Arrays.asList(
+                LogEntry.builder().severityNumber(9).build(),
+                LogEntry.builder().severityNumber(17).build()
+        );
+
+        when(historyDataReader.queryLogsByMultipleConditions(eq(1734005477000L), eq(1734005478000L),
+                any(), any(), any(), any(), any())).thenReturn(mockLogs);
+
+        mockMvc.perform(
+                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));
+    }
+
+    @Test
+    void testTraceCoverageStats() throws Exception {
+        List<LogEntry> mockLogs = Arrays.asList(
+                // With both trace and span
+                LogEntry.builder().traceId("trace1").spanId("span1").build(),
+                LogEntry.builder().traceId("trace2").spanId("span2").build(),
+                // With trace only
+                LogEntry.builder().traceId("trace3").spanId("").build(),
+                // With span only
+                LogEntry.builder().traceId("").spanId("span4").build(),
+                // Without trace info
+                LogEntry.builder().traceId("").spanId("").build(),
+                LogEntry.builder().build() // null values
+        );
+
+        when(historyDataReader.queryLogsByMultipleConditions(any(), any(), any(),
+                any(), any(), any(), any())).thenReturn(mockLogs);
+
+        mockMvc.perform(
+                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))
+                .andExpect(jsonPath("$.data.traceCoverage.withoutTrace").value(3))
+                .andExpect(jsonPath("$.data.traceCoverage.withSpan").value(3))
+                .andExpect(jsonPath("$.data.traceCoverage.withBothTraceAndSpan").value(2));
+    }
+
+    @Test
+    void testTrendStats() throws Exception {
+        // Create logs with timestamps that fall into different hours
+        List<LogEntry> mockLogs = Arrays.asList(
+                // 2023-12-12 10:00 (1734005477630000000L nano = 1734005477630L ms)
+                LogEntry.builder().timeUnixNano(1734005477630000000L).build(),
+                // Same hour
+                LogEntry.builder().timeUnixNano(1734005477640000000L).build(),
+                // Next hour: 2023-12-12 11:00 (1734009077630000000L nano = 1734009077630L ms)
+                LogEntry.builder().timeUnixNano(1734009077630000000L).build()
+        );
+
+        when(historyDataReader.queryLogsByMultipleConditions(any(), any(), any(),
+                any(), any(), any(), any())).thenReturn(mockLogs);
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .get("/api/logs/stats/trend")
+        )
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
+                .andExpect(jsonPath("$.data.hourlyStats").isMap());
+    }
+
+    @Test
+    void testTrendStatsWithNullTimestamp() throws Exception {
+        List<LogEntry> mockLogs = Arrays.asList(
+                LogEntry.builder().timeUnixNano(1734005477630000000L).build(),
+                LogEntry.builder().timeUnixNano(null).build() // This should be filtered out
+        );
+
+        when(historyDataReader.queryLogsByMultipleConditions(any(), any(), any(),
+                any(), any(), any(), any())).thenReturn(mockLogs);
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .get("/api/logs/stats/trend")
+        )
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
+                .andExpect(jsonPath("$.data.hourlyStats").isMap());
+    }
+
+}

hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogSseControllerTest.java

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.verify;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+import org.apache.hertzbeat.log.notice.LogSseFilterCriteria;
+import org.apache.hertzbeat.log.notice.LogSseManager;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+
+/**
+ * Unit tests for the {@link LogSseController}.
+ */
+@ExtendWith(MockitoExtension.class)
+class LogSseControllerTest {
+
+    private MockMvc mockMvc;
+
+    @Mock
+    private LogSseManager emitterManager;
+
+    @Captor
+    private ArgumentCaptor<LogSseFilterCriteria> filterCriteriaCaptor;
+
+    @BeforeEach
+    void setUp() {
+        // Initialize the controller and MockMvc instance before each test
+        LogSseController logSseController = new LogSseController(emitterManager);
+        this.mockMvc = MockMvcBuilders.standaloneSetup(logSseController).build();
+    }
+
+    @Test
+    void testSubscribeWithoutFilters() throws Exception {
+        // When: A request is made to the subscribe endpoint without any parameters
+        mockMvc.perform(get("/api/logs/sse/subscribe")
+                        .accept(MediaType.TEXT_EVENT_STREAM_VALUE))
+                .andExpect(status().isOk());
+
+        // Then: The emitter manager is called with an empty filter criteria
+        verify(emitterManager).createEmitter(anyLong(), filterCriteriaCaptor.capture());
+        LogSseFilterCriteria capturedCriteria = filterCriteriaCaptor.getValue();
+
+        Assertions.assertNull(capturedCriteria.getSeverityText());
+        Assertions.assertNull(capturedCriteria.getSeverityNumber());
+        Assertions.assertNull(capturedCriteria.getTraceId());
+        Assertions.assertNull(capturedCriteria.getSpanId());
+    }
+
+    @Test
+    void testSubscribeWithMultipleFilters() throws Exception {
+        // Given: Multiple filter parameters
+        String severityText = "ERROR";
+        String severityNumber = "17";
+        String traceId = "abcdef1234567890abcdef1234567890";
+        String spanId = "abcdef1234567890";
+
+        // When: A request is made with all filter parameters
+        mockMvc.perform(get("/api/logs/sse/subscribe")
+                        .param("severityText", severityText)
+                        .param("severityNumber", severityNumber)
+                        .param("traceId", traceId)
+                        .param("spanId", spanId)
+                        .accept(MediaType.TEXT_EVENT_STREAM_VALUE))
+                .andExpect(status().isOk());
+
+        // Then: The emitter manager is called with a criteria object containing all filter values
+        verify(emitterManager).createEmitter(anyLong(), filterCriteriaCaptor.capture());
+        LogSseFilterCriteria capturedCriteria = filterCriteriaCaptor.getValue();
+
+        Assertions.assertEquals(capturedCriteria.getSeverityText(), severityText);
+        Assertions.assertEquals(capturedCriteria.getSeverityNumber(), Integer.parseInt(severityNumber));
+        Assertions.assertEquals(capturedCriteria.getTraceId(), traceId);
+        Assertions.assertEquals(capturedCriteria.getSpanId(), spanId);
+    }
+
+    @Test
+    void testSubscribeWithInvalidSeverityNumber() throws Exception {
+        // When: A request is made with a non-integer value for severityNumber
+        mockMvc.perform(get("/api/logs/sse/subscribe")
+                        .param("severityNumber", "not-a-number")
+                        .accept(MediaType.TEXT_EVENT_STREAM_VALUE))
+                .andExpect(status().is4xxClientError());
+    }
+}

hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/OtlpLogControllerTest.java

@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.controller;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
+
+import org.apache.hertzbeat.log.service.impl.OtlpLogProtocolAdapter;
+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.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+
+/**
+ * Unit test for {@link OtlpLogController}
+ */
+@ExtendWith(MockitoExtension.class)
+class OtlpLogControllerTest {
+
+    private static final String CONTENT_TYPE_PROTOBUF = "application/x-protobuf";
+
+    private MockMvc mockMvc;
+
+    @Mock
+    private OtlpLogProtocolAdapter otlpLogProtocolAdapter;
+
+    private OtlpLogController otlpLogController;
+
+    @BeforeEach
+    void setUp() {
+        this.otlpLogController = new OtlpLogController(otlpLogProtocolAdapter);
+        this.mockMvc = MockMvcBuilders.standaloneSetup(otlpLogController).build();
+    }
+
+    @Test
+    void testIngestJsonLogsSuccess() throws Exception {
+        String jsonContent = "{\"resourceLogs\":[]}";
+        
+        doNothing().when(otlpLogProtocolAdapter).ingest(anyString());
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .post("/api/logs/otlp/v1/logs")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(jsonContent)
+        )
+                .andDo(print())
+                .andExpect(status().isOk())
+                .andExpect(content().contentType(MediaType.APPLICATION_JSON))
+                .andReturn();
+    }
+
+    @Test
+    void testIngestJsonLogsFailure() throws Exception {
+        String jsonContent = "{\"invalid\":\"content\"}";
+        
+        doThrow(new IllegalArgumentException("Invalid OTLP JSON log content"))
+                .when(otlpLogProtocolAdapter).ingest(anyString());
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .post("/api/logs/otlp/v1/logs")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(jsonContent)
+        )
+                .andExpect(status().isBadRequest())
+                .andExpect(content().contentType(MediaType.APPLICATION_JSON))
+                .andReturn();
+    }
+
+    @Test
+    void testIngestBinaryLogsSuccess() throws Exception {
+        byte[] binaryContent = new byte[]{0x0a, 0x0b, 0x0c};
+        
+        doNothing().when(otlpLogProtocolAdapter).ingestBinary(any(byte[].class));
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .post("/api/logs/otlp/v1/logs")
+                        .contentType(CONTENT_TYPE_PROTOBUF)
+                        .content(binaryContent)
+        )
+                .andDo(print())
+                .andExpect(status().isOk())
+                .andExpect(content().contentType(CONTENT_TYPE_PROTOBUF))
+                .andReturn();
+    }
+
+    @Test
+    void testIngestBinaryLogsFailure() throws Exception {
+        byte[] binaryContent = new byte[]{0x0a, 0x0b, 0x0c};
+        
+        doThrow(new IllegalArgumentException("Invalid OTLP binary log content"))
+                .when(otlpLogProtocolAdapter).ingestBinary(any(byte[].class));
+
+        mockMvc.perform(
+                MockMvcRequestBuilders
+                        .post("/api/logs/otlp/v1/logs")
+                        .contentType(CONTENT_TYPE_PROTOBUF)
+                        .content(binaryContent)
+        )
+                .andExpect(status().isBadRequest())
+                .andExpect(content().contentType(CONTENT_TYPE_PROTOBUF))
+                .andReturn();
+    }
+}

hertzbeat-log/src/test/java/org/apache/hertzbeat/log/notice/LogSseFilterCriteriaTest.java

@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.notice;
+
+import org.apache.hertzbeat.common.entity.log.LogEntry;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
+/**
+ * Unit tests for LogSseFilterCriteria.
+ */
+class LogSseFilterCriteriaTest {
+
+    private LogEntry testLogEntry;
+    private LogSseFilterCriteria filterCriteria;
+
+    @BeforeEach
+    void setUp() {
+        // Create test LogEntry
+        testLogEntry = LogEntry.builder()
+                .severityNumber(9)
+                .severityText("INFO")
+                .traceId("1234567890abcdef1234567890abcdef")
+                .spanId("1234567890abcdef")
+                .body("Test log message")
+                .timeUnixNano(System.currentTimeMillis() * 1_000_000L)
+                .build();
+        
+        filterCriteria = new LogSseFilterCriteria();
+    }
+
+    @Test
+    void testMatchesWithNoFilters() {
+        // Should match all logs when no filters are set
+        assertTrue(filterCriteria.matches(testLogEntry));
+    }
+
+    @Test
+    void testMatchesWithSeverityTextFilter() {
+        // Test severity text filter - match
+        filterCriteria.setSeverityText("INFO");
+        assertTrue(filterCriteria.matches(testLogEntry));
+        
+        // Test severity text filter - no match
+        filterCriteria.setSeverityText("ERROR");
+        assertFalse(filterCriteria.matches(testLogEntry));
+        
+        // Test severity text filter - case insensitive
+        filterCriteria.setSeverityText("info");
+        assertTrue(filterCriteria.matches(testLogEntry));
+    }
+
+    @Test
+    void testMatchesWithSeverityNumberFilter() {
+        // Test severity number filter - match
+        filterCriteria.setSeverityNumber(9);
+        assertTrue(filterCriteria.matches(testLogEntry));
+        
+        // Test severity number filter - no match
+        filterCriteria.setSeverityNumber(1);
+        assertFalse(filterCriteria.matches(testLogEntry));
+        
+        // Test severity number filter - null value
+        filterCriteria.setSeverityNumber(null);
+        assertTrue(filterCriteria.matches(testLogEntry));
+    }
+
+    @Test
+    void testMatchesWithTraceIdFilter() {
+        // Test Trace ID filter - match
+        filterCriteria.setTraceId("1234567890abcdef1234567890abcdef");
+        assertTrue(filterCriteria.matches(testLogEntry));
+        
+        // Test Trace ID filter - no match
+        filterCriteria.setTraceId("abcdef1234567890abcdef1234567890");
+        assertFalse(filterCriteria.matches(testLogEntry));
+        
+        // Test Trace ID filter - case insensitive
+        filterCriteria.setTraceId("1234567890ABCDEF1234567890ABCDEF");
+        assertTrue(filterCriteria.matches(testLogEntry));
+    }
+
+    @Test
+    void testMatchesWithSpanIdFilter() {
+        // Test Span ID filter - match
+        filterCriteria.setSpanId("1234567890abcdef");
+        assertTrue(filterCriteria.matches(testLogEntry));
+
+        // Test Span ID filter - no match
+        filterCriteria.setSpanId("abcdef1234567890");
+        assertFalse(filterCriteria.matches(testLogEntry));
+
+        // Test Span ID filter - case insensitive
+        filterCriteria.setSpanId("1234567890ABCDEF");
+        assertTrue(filterCriteria.matches(testLogEntry));
+    }
+
+    @Test
+    void testMatchesWithLogContentFilter() {
+        // Test log content filter - match
+        filterCriteria.setLogContent("Test log");
+        assertTrue(filterCriteria.matches(testLogEntry));
+
+        // Test log content filter - no match
+        filterCriteria.setLogContent("Error message");
+        assertFalse(filterCriteria.matches(testLogEntry));
+
+        // Test log content filter - case insensitive
+        filterCriteria.setLogContent("test log");
+        assertTrue(filterCriteria.matches(testLogEntry));
+
+        // Test log content filter - partial match
+        filterCriteria.setLogContent("message");
+        assertTrue(filterCriteria.matches(testLogEntry));
+
+        // Test log content filter with null body
+        LogEntry nullBodyLog = LogEntry.builder()
+                .severityNumber(9)
+                .severityText("INFO")
+                .body(null)
+                .build();
+        filterCriteria.setLogContent("test");
+        assertFalse(filterCriteria.matches(nullBodyLog));
+    }
+
+    @Test
+    void testMatchesWithMultipleFilters() {
+        // Test multiple filter combinations - all match
+        filterCriteria.setSeverityText("INFO");
+        filterCriteria.setSeverityNumber(9);
+        filterCriteria.setTraceId("1234567890abcdef1234567890abcdef");
+        filterCriteria.setSpanId("1234567890abcdef");
+        assertTrue(filterCriteria.matches(testLogEntry));
+        
+        // Test multiple filter combinations - partial no match
+        filterCriteria.setSeverityText("ERROR");
+        assertFalse(filterCriteria.matches(testLogEntry));
+    }
+
+    @Test
+    void testMatchesWithEmptyStringFilters() {
+        // Test empty string filters
+        filterCriteria.setSeverityText("");
+        filterCriteria.setTraceId("");
+        filterCriteria.setSpanId("");
+        assertTrue(filterCriteria.matches(testLogEntry));
+    }
+
+    @Test
+    void testMatchesWithLogEntryHavingNullValues() {
+        // Create log entry with null values
+        LogEntry logWithNulls = LogEntry.builder()
+                .severityNumber(null)
+                .severityText(null)
+                .traceId(null)
+                .spanId(null)
+                .body("Test log with nulls")
+                .build();
+        
+        // Set filter criteria
+        filterCriteria.setSeverityNumber(9);
+        filterCriteria.setSeverityText("INFO");
+        filterCriteria.setTraceId("1234567890abcdef1234567890abcdef");
+        filterCriteria.setSpanId("1234567890abcdef");
+        
+        // Should not match because log entry values are null
+        assertFalse(filterCriteria.matches(logWithNulls));
+    }
+
+    @Test
+    void testConstructorWithAllParameters() {
+        // Test constructor with all parameters
+        LogSseFilterCriteria criteria = new LogSseFilterCriteria(
+                9, "INFO", null, "1234567890abcdef1234567890abcdef", "1234567890abcdef"
+        );
+
+        assertEquals(9, criteria.getSeverityNumber());
+        assertEquals("INFO", criteria.getSeverityText());
+        assertEquals(null, criteria.getLogContent());
+        assertEquals("1234567890abcdef1234567890abcdef", criteria.getTraceId());
+        assertEquals("1234567890abcdef", criteria.getSpanId());
+
+        // Test matching
+        assertTrue(criteria.matches(testLogEntry));
+    }
+
+    @Test
+    void testNoArgsConstructorAndSetters() {
+        // Test no-args constructor and setter methods
+        LogSseFilterCriteria criteria = new LogSseFilterCriteria();
+
+        criteria.setSeverityNumber(9);
+        criteria.setSeverityText("INFO");
+        criteria.setLogContent("Test log");
+        criteria.setTraceId("1234567890abcdef1234567890abcdef");
+        criteria.setSpanId("1234567890abcdef");
+
+        assertEquals(9, criteria.getSeverityNumber());
+        assertEquals("INFO", criteria.getSeverityText());
+        assertEquals("Test log", criteria.getLogContent());
+        assertEquals("1234567890abcdef1234567890abcdef", criteria.getTraceId());
+        assertEquals("1234567890abcdef", criteria.getSpanId());
+
+        // Test matching
+        assertTrue(criteria.matches(testLogEntry));
+    }
+}

hertzbeat-log/src/test/java/org/apache/hertzbeat/log/notice/LogSseManagerTest.java

@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hertzbeat.log.notice;
+
+import org.apache.hertzbeat.common.entity.log.LogEntry;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Unit tests for {@link LogSseManager}.
+ */
+class LogSseManagerTest {
+
+    private LogSseManager logSseManager;
+    private static final Long CLIENT_ID = 1L;
+
+    @BeforeEach
+    void setUp() {
+        logSseManager = new LogSseManager();
+    }
+
+    @AfterEach
+    void tearDown() {
+        logSseManager.shutdown();
+    }
+
+    @Test
+    void shouldCreateAndStoreEmitter() {
+        // When: Creating a new emitter for a client
+        SseEmitter emitter = logSseManager.createEmitter(CLIENT_ID, new LogSseFilterCriteria());
+
+        // Then: The emitter should be created and stored
+        assertNotNull(emitter);
+        assertEquals(Long.MAX_VALUE, emitter.getTimeout());
+        assertTrue(logSseManager.getEmitters().containsKey(CLIENT_ID));
+    }
+
+    @Test
+    void shouldBroadcastLogWhenFilterMatches() throws IOException {
+        // Given: A client with a filter for "INFO" logs
+        LogSseFilterCriteria filters = new LogSseFilterCriteria();
+        filters.setSeverityText("INFO");
+        SseEmitter mockEmitter = mock(SseEmitter.class);
+        subscribeClient(CLIENT_ID, filters, mockEmitter);
+
+        LogEntry infoLog = createLogEntry("INFO", "An informational message");
+
+        // When: An "INFO" log is broadcast
+        logSseManager.broadcast(infoLog);
+
+        // Then: The log should be sent to the client (wait for batch processing)
+        await().atMost(500, TimeUnit.MILLISECONDS).untilAsserted(() -> 
+            verify(mockEmitter, atLeastOnce()).send(any(SseEmitter.SseEventBuilder.class))
+        );
+    }
+
+    @Test
+    void shouldSendBatchOnVirtualThread() throws IOException, InterruptedException {
+        SseEmitter mockEmitter = mock(SseEmitter.class);
+        AtomicBoolean virtualThread = new AtomicBoolean(false);
+        CountDownLatch latch = new CountDownLatch(1);
+        doAnswer(invocation -> {
+            virtualThread.set(Thread.currentThread().isVirtual());
+            latch.countDown();
+            return null;
+        }).when(mockEmitter).send(any(SseEmitter.SseEventBuilder.class));
+        subscribeClient(CLIENT_ID, null, mockEmitter);
+
+        logSseManager.broadcast(createLogEntry("INFO", "virtual-thread-send"));
+
+        assertTrue(latch.await(1, TimeUnit.SECONDS));
+        assertTrue(virtualThread.get());
+    }
+
+    @Test
+    void shouldNotBroadcastLogWhenFilterDoesNotMatch() throws IOException, InterruptedException {
+        // Given: A client with a filter for "ERROR" logs
+        LogSseFilterCriteria filters = new LogSseFilterCriteria();
+        filters.setSeverityText("ERROR");
+        SseEmitter mockEmitter = mock(SseEmitter.class);
+        subscribeClient(CLIENT_ID, filters, mockEmitter);
+
+        LogEntry infoLog = createLogEntry("INFO", "An informational message");
+
+        // When: An "INFO" log is broadcast
+        logSseManager.broadcast(infoLog);
+
+        // Wait for batch processing
+        Thread.sleep(300);
+
+        // Then: The log should NOT be sent to the client
+        verify(mockEmitter, never()).send(any(SseEmitter.SseEventBuilder.class));
+    }
+
+    @Test
+    void shouldBroadcastLogWhenSubscriberHasNoFilters() throws IOException {
+        // Given: A client subscribed with no filters (null)
+        SseEmitter mockEmitter = mock(SseEmitter.class);
+        subscribeClient(CLIENT_ID, null, mockEmitter);
+
+        LogEntry anyLog = createLogEntry("DEBUG", "A debug message");
+
+        // When: Any log is broadcast
+        logSseManager.broadcast(anyLog);
+
+        // Then: The log should be sent to the client (wait for batch processing)
+        await().atMost(500, TimeUnit.MILLISECONDS).untilAsserted(() -> 
+            verify(mockEmitter, atLeastOnce()).send(any(SseEmitter.SseEventBuilder.class))
+        );
+    }
+
+    @Test
+    void shouldBroadcastOnlyToMatchingSubscribers() throws IOException, InterruptedException {
+        // Given: Two clients with different filters
+        LogSseFilterCriteria infoFilter = new LogSseFilterCriteria();
+        infoFilter.setSeverityText("INFO");
+        SseEmitter infoEmitter = mock(SseEmitter.class);
+        subscribeClient(1L, infoFilter, infoEmitter);
+
+        LogSseFilterCriteria errorFilter = new LogSseFilterCriteria();
+        errorFilter.setSeverityText("ERROR");
+        SseEmitter errorEmitter = mock(SseEmitter.class);
+        subscribeClient(2L, errorFilter, errorEmitter);
+
+        LogEntry infoLog = createLogEntry("INFO", "An informational message");
+
+        // When: An "INFO" log is broadcast
+        logSseManager.broadcast(infoLog);
+
+        // Wait for batch processing
+        await().atMost(500, TimeUnit.MILLISECONDS).untilAsserted(() -> 
+            verify(infoEmitter, atLeastOnce()).send(any(SseEmitter.SseEventBuilder.class))
+        );
+
+        // Then: The log is sent only to the client subscribed to "INFO" logs
+        verify(errorEmitter, never()).send(any(SseEmitter.SseEventBuilder.class));
+    }
+
+    @Test
+    void shouldRemoveEmitterWhenBroadcastFails() throws IOException {
+        // Given: A client whose emitter will throw an exception on send
+        SseEmitter mockEmitter = mock(SseEmitter.class);
+        doThrow(new IOException("Connection closed")).when(mockEmitter).send(any(SseEmitter.SseEventBuilder.class));
+        subscribeClient(CLIENT_ID, null, mockEmitter);
+        assertTrue(logSseManager.getEmitters().containsKey(CLIENT_ID));
+
+        LogEntry log = createLogEntry("ERROR", "An error occurred");
+
+        // When: A log is broadcast, causing an exception
+        logSseManager.broadcast(log);
+
+        // Then: The failing emitter should be completed and removed
+        await().atMost(500, TimeUnit.MILLISECONDS).untilAsserted(() -> {
+            verify(mockEmitter).complete();
+            assertFalse(logSseManager.getEmitters().containsKey(CLIENT_ID));
+        });
+    }
+
+    @Test
+    void shouldDropLogsWhenQueueSizeLimitReached() {
+        for (int i = 0; i < 10_001; i++) {
+            logSseManager.broadcast(createLogEntry("INFO", "log-" + i));
+        }
+
+        assertEquals(10_000, logSseManager.getQueueSize());
+        assertEquals(10_000, logSseManager.getLogQueue().size());
+    }
+
+    /**
+     * Helper method to create a subscriber and inject a mock emitter for testing
+     */
+    private void subscribeClient(Long clientId, LogSseFilterCriteria filters, SseEmitter mockEmitter) {
+        logSseManager.createEmitter(clientId, filters);
+        LogSseManager.SseSubscriber subscriber = logSseManager.getEmitters().get(clientId);
+        subscriber.setEmitter(mockEmitter);
+    }
+
+    /**
+     * Helper method to create LogEntry instances
+     */
+    private LogEntry createLogEntry(String severityText, String body) {
+        return LogEntry.builder()
+                .severityText(severityText)
+                .body(body)
+                .build();
+    }
+}

hertzbeat-log/src/test/java/org/apache/hertzbeat/log/service/impl/OtlpLogProtocolAdapterTest.java

@@ -0,0 +1,387 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.log.service.impl;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.util.JsonFormat;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
+import io.opentelemetry.proto.common.v1.AnyValue;
+import io.opentelemetry.proto.common.v1.KeyValue;
+import io.opentelemetry.proto.logs.v1.LogRecord;
+import io.opentelemetry.proto.logs.v1.ResourceLogs;
+import io.opentelemetry.proto.logs.v1.ScopeLogs;
+import io.opentelemetry.proto.resource.v1.Resource;
+import io.opentelemetry.proto.common.v1.InstrumentationScope;
+import org.apache.hertzbeat.common.entity.log.LogEntry;
+import org.apache.hertzbeat.common.queue.CommonDataQueue;
+import org.apache.hertzbeat.log.notice.LogSseManager;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+import java.util.Map;
+
+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.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+/**
+ * Unit tests for OtlpLogProtocolAdapter.
+ */
+@ExtendWith(MockitoExtension.class)
+class OtlpLogProtocolAdapterTest {
+
+    @Mock
+    private CommonDataQueue commonDataQueue;
+
+    @Mock
+    private LogSseManager logSseManager;
+
+    private OtlpLogProtocolAdapter adapter;
+
+    @BeforeEach
+    void setUp() {
+        adapter = new OtlpLogProtocolAdapter(commonDataQueue, logSseManager);
+    }
+
+    @Test
+    void testIngestWithNullContent() {
+        adapter.ingest(null);
+        verifyNoInteractions(commonDataQueue, logSseManager);
+    }
+
+    @Test
+    void testIngestWithEmptyContent() {
+        adapter.ingest("");
+        verifyNoInteractions(commonDataQueue, logSseManager);
+    }
+
+    @Test
+    void testIngestBinaryWithNullContent() {
+        adapter.ingestBinary(null);
+        verifyNoInteractions(commonDataQueue, logSseManager);
+    }
+
+    @Test
+    void testIngestBinaryWithEmptyContent() {
+        adapter.ingestBinary(new byte[0]);
+        verifyNoInteractions(commonDataQueue, logSseManager);
+    }
+
+    @Test
+    void testIngestBinaryWithValidOtlpLogData() throws Exception {
+        byte[] binaryPayload = createValidOtlpLogBinaryPayload();
+        
+        adapter.ingestBinary(binaryPayload);
+        
+        ArgumentCaptor<List<LogEntry>> listCaptor = ArgumentCaptor.forClass(List.class);
+        verify(commonDataQueue, times(1)).sendLogEntryToStorageBatch(listCaptor.capture());
+        verify(commonDataQueue, times(1)).sendLogEntryToAlertBatch(anyList());
+        verify(logSseManager, times(1)).broadcast(any(LogEntry.class));
+        
+        List<LogEntry> capturedList = listCaptor.getValue();
+        assertNotNull(capturedList);
+        assertEquals(1, capturedList.size());
+        
+        LogEntry capturedEntry = capturedList.get(0);
+        assertEquals("binary-test-service", capturedEntry.getResource().get("service_name"));
+        assertEquals("binary log message", capturedEntry.getBody());
+        assertEquals("INFO", capturedEntry.getSeverityText());
+    }
+
+    @Test
+    void testIngestBinaryWithInvalidContent() {
+        byte[] invalidBinary = "not a valid protobuf".getBytes();
+        
+        assertThrows(IllegalArgumentException.class, () -> adapter.ingestBinary(invalidBinary));
+        verifyNoInteractions(commonDataQueue, logSseManager);
+    }
+
+    @Test
+    void testIngestWithValidOtlpLogData() throws Exception {
+        String otlpPayload = createValidOtlpLogPayload();
+        
+        adapter.ingest(otlpPayload);
+        
+        ArgumentCaptor<List<LogEntry>> listCaptor = ArgumentCaptor.forClass(List.class);
+        verify(commonDataQueue, times(1)).sendLogEntryToStorageBatch(listCaptor.capture());
+        verify(commonDataQueue, times(1)).sendLogEntryToAlertBatch(anyList());
+        verify(logSseManager, times(1)).broadcast(any(LogEntry.class));
+        
+        List<LogEntry> capturedList = listCaptor.getValue();
+        assertNotNull(capturedList);
+        assertEquals(1, capturedList.size());
+        
+        LogEntry capturedEntry = capturedList.get(0);
+        assertEquals("test-service", capturedEntry.getResource().get("service_name"));
+        assertEquals("test-version", capturedEntry.getResource().get("service_version"));
+        assertEquals("test-scope", capturedEntry.getInstrumentationScope().getName());
+        assertEquals("1.0.0", capturedEntry.getInstrumentationScope().getVersion());
+        assertEquals("test log message", capturedEntry.getBody());
+        assertEquals("INFO", capturedEntry.getSeverityText());
+        assertEquals(9, capturedEntry.getSeverityNumber());
+    }
+
+    @Test
+    void testIngestWithMultipleLogRecords() throws Exception {
+        String otlpPayload = createOtlpPayloadWithMultipleLogs();
+        
+        adapter.ingest(otlpPayload);
+        
+        ArgumentCaptor<List<LogEntry>> listCaptor = ArgumentCaptor.forClass(List.class);
+        verify(commonDataQueue, times(1)).sendLogEntryToStorageBatch(listCaptor.capture());
+        verify(commonDataQueue, times(1)).sendLogEntryToAlertBatch(anyList());
+        verify(logSseManager, times(2)).broadcast(any(LogEntry.class));
+        
+        List<LogEntry> capturedList = listCaptor.getValue();
+        assertEquals(2, capturedList.size());
+    }
+
+    @Test
+    void testIngestWithComplexAttributes() throws Exception {
+        String otlpPayload = createOtlpPayloadWithComplexAttributes();
+        
+        adapter.ingest(otlpPayload);
+        
+        ArgumentCaptor<List<LogEntry>> listCaptor = ArgumentCaptor.forClass(List.class);
+        verify(commonDataQueue, times(1)).sendLogEntryToStorageBatch(listCaptor.capture());
+        verify(commonDataQueue, times(1)).sendLogEntryToAlertBatch(anyList());
+        verify(logSseManager, times(1)).broadcast(any(LogEntry.class));
+        
+        List<LogEntry> capturedList = listCaptor.getValue();
+        assertNotNull(capturedList);
+        assertEquals(1, capturedList.size());
+        
+        LogEntry capturedEntry = capturedList.get(0);
+        Map<String, Object> attributes = capturedEntry.getAttributes();
+        assertEquals("string_value", attributes.get("string_attr"));
+        assertEquals(true, attributes.get("bool_attr"));
+        assertEquals(42L, attributes.get("int_attr"));
+        assertEquals(3.14, attributes.get("double_attr"));
+        
+        List<Object> arrayAttr = (List<Object>) attributes.get("array_attr");
+        assertNotNull(arrayAttr);
+        assertEquals(3, arrayAttr.size());
+        assertEquals("item1", arrayAttr.get(0));
+        assertEquals("item2", arrayAttr.get(1));
+        assertEquals("item3", arrayAttr.get(2));
+    }
+
+    @Test
+    void testIngestWithTraceAndSpanIds() throws Exception {
+        String otlpPayload = createOtlpPayloadWithTraceSpanIds();
+        
+        adapter.ingest(otlpPayload);
+        
+        ArgumentCaptor<List<LogEntry>> listCaptor = ArgumentCaptor.forClass(List.class);
+        verify(commonDataQueue, times(1)).sendLogEntryToStorageBatch(listCaptor.capture());
+        verify(commonDataQueue, times(1)).sendLogEntryToAlertBatch(anyList());
+        verify(logSseManager, times(1)).broadcast(any(LogEntry.class));
+        
+        List<LogEntry> capturedList = listCaptor.getValue();
+        assertEquals(1, capturedList.size());
+        
+        LogEntry capturedEntry = capturedList.get(0);
+        assertEquals("1234567890abcdef1234567890abcdef", capturedEntry.getTraceId());
+        assertEquals("1234567890abcdef", capturedEntry.getSpanId());
+        assertEquals(1, capturedEntry.getTraceFlags());
+    }
+
+    @Test
+    void testIngestWithInvalidJsonContent() {
+        String invalidJson = "{ invalid json content }";
+        
+        assertThrows(IllegalArgumentException.class, () -> adapter.ingest(invalidJson));
+        verifyNoInteractions(commonDataQueue, logSseManager);
+    }
+
+    @Test
+    void testIngestWithEmptyResourceLogs() throws Exception {
+        String otlpPayload = createEmptyResourceLogsPayload();
+        
+        adapter.ingest(otlpPayload);
+        
+        ArgumentCaptor<List<LogEntry>> listCaptor = ArgumentCaptor.forClass(List.class);
+        verify(commonDataQueue, times(1)).sendLogEntryToStorageBatch(listCaptor.capture());
+        verify(commonDataQueue, times(1)).sendLogEntryToAlertBatch(anyList());
+        
+        List<LogEntry> capturedList = listCaptor.getValue();
+        assertNotNull(capturedList);
+        assertEquals(0, capturedList.size());
+        
+        verifyNoInteractions(logSseManager);
+    }
+
+    private String createValidOtlpLogPayload() throws Exception {
+        ExportLogsServiceRequest request = ExportLogsServiceRequest.newBuilder()
+            .addResourceLogs(ResourceLogs.newBuilder()
+                .setResource(Resource.newBuilder()
+                    .addAttributes(KeyValue.newBuilder()
+                        .setKey("service.name")
+                        .setValue(AnyValue.newBuilder().setStringValue("test-service").build())
+                        .build())
+                    .addAttributes(KeyValue.newBuilder()
+                        .setKey("service.version")
+                        .setValue(AnyValue.newBuilder().setStringValue("test-version").build())
+                        .build())
+                    .build())
+                .addScopeLogs(ScopeLogs.newBuilder()
+                    .setScope(InstrumentationScope.newBuilder()
+                        .setName("test-scope")
+                        .setVersion("1.0.0")
+                        .build())
+                    .addLogRecords(LogRecord.newBuilder()
+                        .setTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                        .setObservedTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                        .setSeverityNumberValue(9)
+                        .setSeverityText("INFO")
+                        .setBody(AnyValue.newBuilder().setStringValue("test log message").build())
+                        .build())
+                    .build())
+                .build())
+            .build();
+        
+        return JsonFormat.printer().print(request);
+    }
+
+    private String createOtlpPayloadWithMultipleLogs() throws Exception {
+        ExportLogsServiceRequest request = ExportLogsServiceRequest.newBuilder()
+            .addResourceLogs(ResourceLogs.newBuilder()
+                .setResource(Resource.newBuilder().build())
+                .addScopeLogs(ScopeLogs.newBuilder()
+                    .setScope(InstrumentationScope.newBuilder().build())
+                    .addLogRecords(LogRecord.newBuilder()
+                        .setTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                        .setBody(AnyValue.newBuilder().setStringValue("first log").build())
+                        .build())
+                    .addLogRecords(LogRecord.newBuilder()
+                        .setTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                        .setBody(AnyValue.newBuilder().setStringValue("second log").build())
+                        .build())
+                    .build())
+                .build())
+            .build();
+        
+        return JsonFormat.printer().print(request);
+    }
+
+    private String createOtlpPayloadWithComplexAttributes() throws Exception {
+        ExportLogsServiceRequest request = ExportLogsServiceRequest.newBuilder()
+            .addResourceLogs(ResourceLogs.newBuilder()
+                .setResource(Resource.newBuilder().build())
+                .addScopeLogs(ScopeLogs.newBuilder()
+                    .setScope(InstrumentationScope.newBuilder().build())
+                    .addLogRecords(LogRecord.newBuilder()
+                        .setTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                        .setBody(AnyValue.newBuilder().setStringValue("complex attributes test").build())
+                        .addAttributes(KeyValue.newBuilder()
+                            .setKey("string.attr")
+                            .setValue(AnyValue.newBuilder().setStringValue("string_value").build())
+                            .build())
+                        .addAttributes(KeyValue.newBuilder()
+                            .setKey("bool.attr")
+                            .setValue(AnyValue.newBuilder().setBoolValue(true).build())
+                            .build())
+                        .addAttributes(KeyValue.newBuilder()
+                            .setKey("int.attr")
+                            .setValue(AnyValue.newBuilder().setIntValue(42).build())
+                            .build())
+                        .addAttributes(KeyValue.newBuilder()
+                            .setKey("double.attr")
+                            .setValue(AnyValue.newBuilder().setDoubleValue(3.14).build())
+                            .build())
+                        .addAttributes(KeyValue.newBuilder()
+                            .setKey("array.attr")
+                            .setValue(AnyValue.newBuilder()
+                                .setArrayValue(io.opentelemetry.proto.common.v1.ArrayValue.newBuilder()
+                                    .addValues(AnyValue.newBuilder().setStringValue("item1").build())
+                                    .addValues(AnyValue.newBuilder().setStringValue("item2").build())
+                                    .addValues(AnyValue.newBuilder().setStringValue("item3").build())
+                                    .build())
+                                .build())
+                            .build())
+                        .build())
+                    .build())
+                .build())
+            .build();
+        
+        return JsonFormat.printer().print(request);
+    }
+
+    private String createOtlpPayloadWithTraceSpanIds() throws Exception {
+        ExportLogsServiceRequest request = ExportLogsServiceRequest.newBuilder()
+            .addResourceLogs(ResourceLogs.newBuilder()
+                .setResource(Resource.newBuilder().build())
+                .addScopeLogs(ScopeLogs.newBuilder()
+                    .setScope(InstrumentationScope.newBuilder().build())
+                    .addLogRecords(LogRecord.newBuilder()
+                        .setTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                        .setBody(AnyValue.newBuilder().setStringValue("trace test").build())
+                        .setTraceId(ByteString.fromHex("1234567890abcdef1234567890abcdef"))
+                        .setSpanId(ByteString.fromHex("1234567890abcdef"))
+                        .setFlags(1)
+                        .build())
+                    .build())
+                .build())
+            .build();
+        
+        return JsonFormat.printer().print(request);
+    }
+
+    private String createEmptyResourceLogsPayload() throws Exception {
+        ExportLogsServiceRequest request = ExportLogsServiceRequest.newBuilder().build();
+        return JsonFormat.printer().print(request);
+    }
+
+    private byte[] createValidOtlpLogBinaryPayload() {
+        ExportLogsServiceRequest request = ExportLogsServiceRequest.newBuilder()
+            .addResourceLogs(ResourceLogs.newBuilder()
+                .setResource(Resource.newBuilder()
+                    .addAttributes(KeyValue.newBuilder()
+                        .setKey("service.name")
+                        .setValue(AnyValue.newBuilder().setStringValue("binary-test-service").build())
+                        .build())
+                    .build())
+                .addScopeLogs(ScopeLogs.newBuilder()
+                    .setScope(InstrumentationScope.newBuilder()
+                        .setName("binary-test-scope")
+                        .setVersion("1.0.0")
+                        .build())
+                    .addLogRecords(LogRecord.newBuilder()
+                        .setTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                        .setObservedTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                        .setSeverityNumberValue(9)
+                        .setSeverityText("INFO")
+                        .setBody(AnyValue.newBuilder().setStringValue("binary log message").build())
+                        .build())
+                    .build())
+                .build())
+            .build();
+        
+        return request.toByteArray();
+    }
+}