feat: support otlp http binary protobuf format log data and update doc

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

@@ -17,6 +17,7 @@
 
 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;
@@ -25,24 +26,25 @@
 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.util.StringUtils;
 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;
 
 /**
- * Log Ingestion Controller
+ * 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 = "application/json")
+@RequestMapping(path = "/api/logs", produces = MediaType.APPLICATION_JSON_VALUE)
 @Slf4j
 public class LogIngestionController {
 
-    private static final String DEFAULT_PROTOCOL = "otlp";
     private final List<LogProtocolAdapter> protocolAdapters;
 
     public LogIngestionController(List<LogProtocolAdapter> protocolAdapters) {
@@ -51,26 +53,23 @@ public LogIngestionController(List<LogProtocolAdapter> protocolAdapters) {
 
     /**
      * Receive log payload pushed from external system specifying the log protocol.
-     * Examples:
-     *  - POST /api/logs/ingest/otlp         (content body is OTLP JSON)
      *
-     * @param protocol  log protocol identifier
-     * @param content raw request body
+     * @param protocol log protocol identifier (e.g., "vector", "loki")
+     * @param content  raw request body
      */
-    @PostMapping("/ingest/{protocol}")
-    public ResponseEntity<Message<Void>> ingestExternLog(@PathVariable("protocol") String protocol,
-                                                         @RequestBody String content) {
-        log.debug("Receive extern log from protocol: {}, content length: {}", protocol, content == null ? 0 : content.length());
-        if (!StringUtils.hasText(protocol)) {
-            protocol = DEFAULT_PROTOCOL; // Default to OTLP if no protocol specified
-        }
+    @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 extern log failed: {}", e.getMessage(), 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()));
                 }
@@ -80,30 +79,4 @@ public ResponseEntity<Message<Void>> ingestExternLog(@PathVariable("protocol") S
         return ResponseEntity.status(HttpStatus.BAD_REQUEST)
                 .body(Message.fail(CommonConstants.FAIL_CODE, "Not support the " + protocol + " protocol log"));
     }
-
-    /**
-     * Receive default log payload (when protocol is not specified).
-     * It will look for a service whose supportProtocol() returns "otlp".
-     */
-    @PostMapping("/ingest")
-    public ResponseEntity<Message<Void>> ingestDefaultExternLog(@RequestBody String content) {
-        log.info("Receive default extern log content, length: {}", content == null ? 0 : content.length());
-        LogProtocolAdapter adapter = protocolAdapters.stream()
-                .filter(item -> DEFAULT_PROTOCOL.equalsIgnoreCase(item.supportProtocol()))
-                .findFirst()
-                .orElse(null);
-        if (adapter != null) {
-            try {
-                adapter.ingest(content);
-                return ResponseEntity.ok(Message.success("Add extern log success"));
-            } catch (Exception e) {
-                log.error("Add extern log failed: {}", e.getMessage(), e);
-                return ResponseEntity.status(HttpStatus.BAD_REQUEST)
-                        .body(Message.fail(CommonConstants.FAIL_CODE, "Add extern log failed: " + e.getMessage()));
-            }
-        }
-        log.error("Not support default extern log protocol");
-        return ResponseEntity.status(HttpStatus.BAD_REQUEST)
-                .body(Message.fail(CommonConstants.FAIL_CODE, "Not support the default protocol log"));
-    }
 }

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

@@ -0,0 +1,145 @@
+/*
+ * 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.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 (Exception e) {
+            log.error("Failed to ingest OTLP JSON logs: {}", e.getMessage(), e);
+            return ResponseEntity.status(HttpStatus.BAD_REQUEST)
+                    .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 (Exception e) {
+            log.error("Failed to ingest OTLP binary logs: {}", e.getMessage(), e);
+            return ResponseEntity.status(HttpStatus.BAD_REQUEST)
+                    .body(createBinaryErrorResponse(e.getMessage()));
+        }
+    }
+
+    private String toJsonResponse(ExportLogsServiceResponse response) {
+        try {
+            return JsonFormat.printer().print(response);
+        } catch (InvalidProtocolBufferException 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) + "\"}";
+        }
+    }
+
+    private String escapeJson(String message) {
+        if (message == null) {
+            return "";
+        }
+        return message.replace("\\", "\\\\")
+                      .replace("\"", "\\\"")
+                      .replace("\n", "\\n")
+                      .replace("\r", "\\r")
+                      .replace("\t", "\\t");
+    }
+
+    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/service/LogProtocolAdapter.java

@@ -28,7 +28,7 @@
 public interface LogProtocolAdapter {
 
     /**
-     * Ingest raw log payload pushed from external system.
+     * Ingest log payload pushed from external system.
      *
      * @param content raw request body string
      */

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

@@ -39,7 +39,10 @@
 import java.util.Map;
 
 /**
- * Adapter for OpenTelemetry OTLP/HTTP JSON log ingestion.
+ * 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
@@ -58,26 +61,47 @@ public OtlpLogProtocolAdapter(CommonDataQueue commonDataQueue, LogSseManager log
     @Override
     public void ingest(String content) {
         if (content == null || content.isEmpty()) {
-            log.warn("Received empty OTLP log payload - skip processing.");
+            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();
-
-            // Extract LogEntry instances from the request
-            List<LogEntry> logEntries = extractLogEntries(request);
-            log.debug("Successfully extracted {} log entries from OTLP payload {}", logEntries.size(), content);
-            commonDataQueue.sendLogEntryToStorageBatch(logEntries);
-            commonDataQueue.sendLogEntryToAlertBatch(logEntries);
-            logEntries.forEach(logSseManager::broadcast);
+            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 log payload: {}", e.getMessage());
-            throw new IllegalArgumentException("Invalid OTLP log content", 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.
      *