Support OTLP/HTTP receiver for traces, logs, and metrics

.github/workflows/skywalking.yaml

@@ -769,7 +769,7 @@ jobs:
         if: matrix.test.docker != null
         run: docker build -t ${{ matrix.test.docker.name }} -f ${{ matrix.test.docker.base }}/${{ matrix.test.docker.file }} ${{ matrix.test.docker.base }}
       - name: ${{ matrix.test.name }}
-        uses: apache/skywalking-infra-e2e@530f9976e4bf12b65a6604ea0efbeafabeeea2a3
+        uses: apache/skywalking-infra-e2e@0d91769411db83f6329633df810a36c6ea1a9b02
         with:
           e2e-file: $GITHUB_WORKSPACE/${{ matrix.test.config }}
       - if: ${{ failure() }}
@@ -843,7 +843,7 @@ jobs:
           username: ${{ github.repository_owner }}
           password: ${{ secrets.GITHUB_TOKEN }}
       - name: ALS ${{ matrix.storage }}, ${{ matrix.analyzer }}, istio-${{ matrix.versions.istio }}, k8s-${{ matrix.versions.kubernetes }}
-        uses: apache/skywalking-infra-e2e@530f9976e4bf12b65a6604ea0efbeafabeeea2a3
+        uses: apache/skywalking-infra-e2e@0d91769411db83f6329633df810a36c6ea1a9b02
         env:
           ISTIO_VERSION: ${{ matrix.versions.istio }}
           KUBERNETES_VERSION: ${{ matrix.versions.kubernetes }}
@@ -915,7 +915,7 @@ jobs:
           username: ${{ github.repository_owner }}
           password: ${{ secrets.GITHUB_TOKEN }}
       - name: ${{ matrix.test.name }}
-        uses: apache/skywalking-infra-e2e@530f9976e4bf12b65a6604ea0efbeafabeeea2a3
+        uses: apache/skywalking-infra-e2e@0d91769411db83f6329633df810a36c6ea1a9b02
         env:
           ISTIO_VERSION: ${{ matrix.versions.istio }}
           KUBERNETES_VERSION: ${{ matrix.versions.kubernetes }}
@@ -979,7 +979,7 @@ jobs:
         shell: bash
         run: ./mvnw -B -q -f test/e2e-v2/java-test-service/pom.xml clean package
       - name: Java version ${{ matrix.java-version }}
-        uses: apache/skywalking-infra-e2e@530f9976e4bf12b65a6604ea0efbeafabeeea2a3
+        uses: apache/skywalking-infra-e2e@0d91769411db83f6329633df810a36c6ea1a9b02
         env:
           SW_AGENT_JDK_VERSION: ${{ matrix.java-version }}
         with:
@@ -1075,7 +1075,7 @@ jobs:
           fi
           docker compose -f ${BANYANDB_DATA_GENERATE_ROOT}/docker-compose.yml down -v
       - name: ${{ matrix.test.name }}
-        uses: apache/skywalking-infra-e2e@530f9976e4bf12b65a6604ea0efbeafabeeea2a3
+        uses: apache/skywalking-infra-e2e@0d91769411db83f6329633df810a36c6ea1a9b02
         with:
           e2e-file: $GITHUB_WORKSPACE/${{ matrix.test.config }}
       - if: ${{ failure() }}

docs/en/changes/changes.md

@@ -28,6 +28,7 @@
 * LAL: add `sourceAttribute()` function for non-persistent OTLP resource attribute access in LAL scripts.
 * LAL: add `layer: auto` mode for dynamic layer assignment when `service.layer` is absent.
 * Add two-phase `SpanListener` SPI mechanism for extensible trace span processing. Refactor GenAI from hardcoded `SpanForward.processGenAILogic()` to `GenAISpanListener`.
+* Add OTLP/HTTP receiver support for traces, logs, and metrics (`/v1/traces`, `/v1/logs`, `/v1/metrics`). Supports both `application/x-protobuf` and `application/json` content types.
 
 #### UI
 

docs/en/setup/backend/otlp-trace.md

@@ -1,10 +1,22 @@
 # OpenTelemetry Trace Format
 
-SkyWalking can receive traces from Traces in OTLP format and convert them to Zipkin Trace format eventually. 
+SkyWalking can receive traces in OTLP format and convert them to Zipkin Trace format eventually.
 For data analysis and queries related to Zipkin Trace, please [refer to the relevant documentation](./zipkin-trace.md#zipkin-query).
 
 OTLP Trace handler references the [Zipkin Exporter in the OpenTelemetry Collector](https://opentelemetry.io/docs/specs/otel/trace/sdk_exporters/zipkin/#summary) to convert the data format.
 
+## Supported Protocols
+
+Both **OTLP/gRPC** and **OTLP/HTTP** are supported for traces, logs, and metrics:
+
+| Signal  | OTLP/gRPC (port 11800)       | OTLP/HTTP (port 12800)  |
+|---------|------------------------------|-------------------------|
+| Traces  | gRPC `TraceService/Export`    | `POST /v1/traces`       |
+| Logs    | gRPC `LogsService/Export`     | `POST /v1/logs`         |
+| Metrics | gRPC `MetricsService/Export`  | `POST /v1/metrics`      |
+
+OTLP/HTTP supports both `application/x-protobuf` and `application/json` content types.
+
 ## Set up backend receiver
 
 1. Make sure to enable **otlp-traces** handler in OTLP receiver of `application.yml`.

oap-server/server-receiver-plugin/aws-firehose-receiver/src/main/java/org/apache/skywalking/oap/server/receiver/aws/firehose/FirehoseHTTPHandler.java

@@ -19,6 +19,7 @@
 
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.server.annotation.Blocking;
 import com.linecorp.armeria.common.HttpStatus;
 import com.linecorp.armeria.server.annotation.ConsumesJson;
 import com.linecorp.armeria.server.annotation.Default;
@@ -35,6 +36,7 @@
 
 @Slf4j
 @AllArgsConstructor
+@Blocking
 public class FirehoseHTTPHandler {
     private final OpenTelemetryMetricRequestProcessor openTelemetryMetricRequestProcessor;
     private final String firehoseAccessKey;

oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHTTPHandler.java

@@ -0,0 +1,84 @@
+/*
+ * 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.skywalking.oap.server.receiver.otel.otlp;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.JsonFormat;
+import com.linecorp.armeria.common.AggregatedHttpRequest;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.HttpStatus;
+import com.linecorp.armeria.common.MediaType;
+import com.linecorp.armeria.server.annotation.Blocking;
+import com.linecorp.armeria.server.annotation.Consumes;
+import com.linecorp.armeria.server.annotation.ConsumesJson;
+import com.linecorp.armeria.server.annotation.Post;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceResponse;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * OTLP/HTTP handler for log data. Supports both protobuf and JSON encoding.
+ * Delegates processing to {@link OpenTelemetryLogHandler#processExport}.
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class OpenTelemetryLogHTTPHandler {
+    private static final byte[] EMPTY_RESPONSE =
+        ExportLogsServiceResponse.getDefaultInstance().toByteArray();
+
+    private final OpenTelemetryLogHandler logHandler;
+
+    @Blocking
+    @Post("/v1/logs")
+    @Consumes("application/x-protobuf")
+    public HttpResponse collectProtobuf(AggregatedHttpRequest request) {
+        try {
+            final ExportLogsServiceRequest exportRequest =
+                ExportLogsServiceRequest.parseFrom(request.content().array());
+            logHandler.processExport(exportRequest);
+            return HttpResponse.of(HttpStatus.OK, MediaType.PROTOBUF, EMPTY_RESPONSE);
+        } catch (InvalidProtocolBufferException e) {
+            log.warn("Failed to parse OTLP/HTTP log request", e);
+            return HttpResponse.of(HttpStatus.BAD_REQUEST);
+        } catch (Exception e) {
+            log.error("Failed to process OTLP/HTTP log request", e);
+            return HttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR);
+        }
+    }
+
+    @Blocking
+    @Post("/v1/logs")
+    @ConsumesJson
+    public HttpResponse collectJson(AggregatedHttpRequest request) {
+        try {
+            final ExportLogsServiceRequest.Builder builder =
+                ExportLogsServiceRequest.newBuilder();
+            JsonFormat.parser().ignoringUnknownFields().merge(
+                request.contentUtf8(), builder);
+            logHandler.processExport(builder.build());
+            return HttpResponse.of(HttpStatus.OK, MediaType.JSON_UTF_8, "{}");
+        } catch (InvalidProtocolBufferException e) {
+            log.warn("Failed to parse OTLP/HTTP JSON log request", e);
+            return HttpResponse.of(HttpStatus.BAD_REQUEST);
+        } catch (Exception e) {
+            log.error("Failed to process OTLP/HTTP JSON log request", e);
+            return HttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR);
+        }
+    }
+}

oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java

@@ -36,7 +36,10 @@
 import org.apache.skywalking.oap.server.core.source.LogMetadataUtils;
 import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule;
 import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService;
+import com.linecorp.armeria.common.HttpMethod;
+import java.util.Collections;
 import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
+import org.apache.skywalking.oap.server.core.server.HTTPHandlerRegister;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
 import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.apache.skywalking.oap.server.receiver.otel.Handler;
@@ -88,10 +91,26 @@ public void active() throws ModuleStartException {
                                                          .provider()
                                                          .getService(GRPCHandlerRegister.class);
         grpcHandlerRegister.addHandler(this);
+
+        HTTPHandlerRegister httpHandlerRegister = manager.find(SharingServerModule.NAME)
+                                                         .provider()
+                                                         .getService(HTTPHandlerRegister.class);
+        httpHandlerRegister.addHandler(
+            new OpenTelemetryLogHTTPHandler(this),
+            Collections.singletonList(HttpMethod.POST));
     }
 
     @Override
     public void export(ExportLogsServiceRequest request, StreamObserver<ExportLogsServiceResponse> responseObserver) {
+        processExport(request);
+        responseObserver.onNext(ExportLogsServiceResponse.getDefaultInstance());
+        responseObserver.onCompleted();
+    }
+
+    /**
+     * Process an OTLP log export request. Shared by both gRPC and HTTP handlers.
+     */
+    void processExport(ExportLogsServiceRequest request) {
         request.getResourceLogsList().forEach(resourceLogs -> {
             final var resource = resourceLogs.getResource();
             final var attributes = resource
@@ -122,8 +141,6 @@ public void export(ExportLogsServiceRequest request, StreamObserver<ExportLogsSe
                     }
                 });
         });
-        responseObserver.onNext(ExportLogsServiceResponse.getDefaultInstance());
-        responseObserver.onCompleted();
     }
 
     private void doAnalysisQuietly(String service, String layer, String serviceInstance,

oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryMetricHTTPHandler.java

@@ -0,0 +1,84 @@
+/*
+ * 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.skywalking.oap.server.receiver.otel.otlp;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.JsonFormat;
+import com.linecorp.armeria.common.AggregatedHttpRequest;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.HttpStatus;
+import com.linecorp.armeria.common.MediaType;
+import com.linecorp.armeria.server.annotation.Blocking;
+import com.linecorp.armeria.server.annotation.Consumes;
+import com.linecorp.armeria.server.annotation.ConsumesJson;
+import com.linecorp.armeria.server.annotation.Post;
+import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
+import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceResponse;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * OTLP/HTTP handler for metric data. Supports both protobuf and JSON encoding.
+ * Delegates processing to {@link OpenTelemetryMetricRequestProcessor#processMetricsRequest}.
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class OpenTelemetryMetricHTTPHandler {
+    private static final byte[] EMPTY_RESPONSE =
+        ExportMetricsServiceResponse.getDefaultInstance().toByteArray();
+
+    private final OpenTelemetryMetricRequestProcessor metricProcessor;
+
+    @Blocking
+    @Post("/v1/metrics")
+    @Consumes("application/x-protobuf")
+    public HttpResponse collectProtobuf(AggregatedHttpRequest request) {
+        try {
+            final ExportMetricsServiceRequest exportRequest =
+                ExportMetricsServiceRequest.parseFrom(request.content().array());
+            metricProcessor.processMetricsRequest(exportRequest);
+            return HttpResponse.of(HttpStatus.OK, MediaType.PROTOBUF, EMPTY_RESPONSE);
+        } catch (InvalidProtocolBufferException e) {
+            log.warn("Failed to parse OTLP/HTTP metric request", e);
+            return HttpResponse.of(HttpStatus.BAD_REQUEST);
+        } catch (Exception e) {
+            log.error("Failed to process OTLP/HTTP metric request", e);
+            return HttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR);
+        }
+    }
+
+    @Blocking
+    @Post("/v1/metrics")
+    @ConsumesJson
+    public HttpResponse collectJson(AggregatedHttpRequest request) {
+        try {
+            final ExportMetricsServiceRequest.Builder builder =
+                ExportMetricsServiceRequest.newBuilder();
+            JsonFormat.parser().ignoringUnknownFields().merge(
+                request.contentUtf8(), builder);
+            metricProcessor.processMetricsRequest(builder.build());
+            return HttpResponse.of(HttpStatus.OK, MediaType.JSON_UTF_8, "{}");
+        } catch (InvalidProtocolBufferException e) {
+            log.warn("Failed to parse OTLP/HTTP JSON metric request", e);
+            return HttpResponse.of(HttpStatus.BAD_REQUEST);
+        } catch (Exception e) {
+            log.error("Failed to process OTLP/HTTP JSON metric request", e);
+            return HttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR);
+        }
+    }
+}

oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryMetricHandler.java

@@ -23,7 +23,10 @@
 import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceResponse;
 import io.opentelemetry.proto.collector.metrics.v1.MetricsServiceGrpc;
 import lombok.extern.slf4j.Slf4j;
+import com.linecorp.armeria.common.HttpMethod;
+import java.util.Collections;
 import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
+import org.apache.skywalking.oap.server.core.server.HTTPHandlerRegister;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
 import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.apache.skywalking.oap.server.receiver.otel.Handler;
@@ -58,6 +61,13 @@ public void active() throws ModuleStartException {
                                                          .provider()
                                                          .getService(GRPCHandlerRegister.class);
         grpcHandlerRegister.addHandler(this);
+
+        HTTPHandlerRegister httpHandlerRegister = manager.find(SharingServerModule.NAME)
+                                                         .provider()
+                                                         .getService(HTTPHandlerRegister.class);
+        httpHandlerRegister.addHandler(
+            new OpenTelemetryMetricHTTPHandler(metricRequestProcessor),
+            Collections.singletonList(HttpMethod.POST));
     }
 
     @Override