Otel logs source http service by TomasLongo · Pull Request #6250 · opensearch-project/data-prepper

TomasLongo · 2025-11-07T13:53:50Z

Description

This PR adds an HTTP-Service to the OtelLogsSource, which runs alongside the gRPC-Service under a separate endpoint.

Both Services share the same underlying Armeria-Server, running under the same port.

Breakdown

The PR is kind of heavy. The OtelLogsSource had to be restructured resulting in the refactoring of the present tests.

A quick breakdown of the changes will certainly help with the review

One Server, two services

OtelLogsSource is now inhabitetd by two services running under different paths behind the same port.
The path of the services can be configured by the two fields path and http_path

path is the already present field for configuring path of the grpc service and keeps its purpose
http_path will set the path for the new http service

Unframed Requests

Since unframed requests served the purpose of sending plain http requests to the grpc server, this functionality has been dropped. However, for backwardscompatibilty of the config, it is still present in the config data structure.

CreateServer

The class CreateServer deals with creating both, http and gRPC servers. However, in its current form, the intended use is to create a server with a single service. Two quick attempts to rewrite the Class for the purpose if this PR yielded no results.

So the final version of the server creation loosely resembles the functionality found in CreateServer.

Refactoring of tests

Many tests were making sure that the former gRPC server, with unframed requests enabled, would handle http requests correctly. I have dropped this tests completely, replacing them with a separate test class that handles tests regarding the http functionality. I added a few improvements along the way like

Getting rid of unnecessary mocks
Introducing parameterized tests
Using unified test data and fixtures

Decisions made and additional considerations

Getting completely rid of unframed requests

enable_unframed_requests has no effect any more, since its purpose has been replaced with the new http service

Configuration

A new field, http_path has been added to specify the path of the http service. However, one could extend the config to

enable the client to specify only on of both service by omitting one of both paths. Currently, both services will be instantiated.

CreateServer

I think that the intention of CreateServer, to provide a unified mechanism to bootstrap armeria servers, is absolutely the right thing todo. However, in its current form, it only supports creating servers with a single service.

It's worth thinking about how to refactor CreateServer to streamline server creation for future plugins. The challenge I encountered was how to best model the fact, that different services might need different server capabilities/configurations.

Figuring this out, would have gone well beyond the scope of this PR.

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

TomasLongo · 2025-11-07T13:54:08Z

@KarstenSchnitter

KarstenSchnitter

Just a start of a review. Can you check the tests for unused imports. This is failing the CI builds. I need more time for the review. Please expect more comments to come.

KarstenSchnitter · 2025-11-07T18:03:29Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/ArmeriaHttpService.java

+    // no path provided. Will be set by config.
+    @Post("")
+    @Consumes(value = "application/json")
+    public ExportTraceServiceResponse exportLog(ExportLogsServiceRequest request) {


Why is this returning ExportTraceServiceResponse? Shouldn't this return a ExportLogServiceResponse?

Right. I copied the Http-Service from the Trace-Source. This is the remnant from it. Fix is pushed.

KarstenSchnitter

I already have some comments. I still need to review the tests.

KarstenSchnitter · 2025-11-15T08:06:35Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/ArmeriaHttpService.java

+        try {
+            logs = oTelProtoDecoder.parseExportLogsServiceRequest(request, Instant.now());
+        } catch (Exception e) {
+            LOG.error("Failed to parse the request {} due to:", request, e);


Please use warning level and sensitive request marker.

Suggested change

LOG.error("Failed to parse the request {} due to:", request, e);

LOG.warn(DataPrepperMarkers.SENSITIVE, "Failed to parse request with error '{}'. Request body: {}.", e.getMessage(), request);

done.

Is this a general rule? Use WARN-level with SENSITIVE-Marker for errors?

Is there any documentation on how to logging should be done in DataPrepper? Could not find any in the README

KarstenSchnitter · 2025-11-15T08:07:58Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/ArmeriaHttpService.java

+            throw new BadRequestException(e.getMessage(), e);
+        }
+
+        final List<Record<Object>> records = logs.stream().map(log -> new Record<Object>(log)).collect(Collectors.toList());


These records are only required in the else block of the following condition. Please move this operation to this block.

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSource.java

KarstenSchnitter · 2025-11-15T08:32:34Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSource.java

+
+        if (oTelLogsSourceConfig.hasHealthCheck()) {
+            LOG.info("HTTP source health check is enabled");
+            serverBuilder.service("/health", HealthCheckService.builder().longPolling(0).build());


Please extract the String to a constant.

KarstenSchnitter · 2025-11-15T08:42:18Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSource.java

+            serverBuilder.maxRequestLength(oTelLogsSourceConfig.getMaxRequestLength().getBytes());
+        }
+        final int threadCount = oTelLogsSourceConfig.getThreadCount();
+        serverBuilder.blockingTaskExecutor(new ScheduledThreadPoolExecutor(threadCount), true);


Please use the same logic as in CreateServer to configure the task executor.

KarstenSchnitter

This looks good so far. I would like to see an end-to-end test with OTLP/HTTP if possible. Can you add it? Please promote this PR to ready for review.

@TomasLongo please also fix the unused imports in the tests that lead to the failing Gradle builds.

TomasLongo · 2025-11-21T13:49:11Z

Yes. I'll add an E2E Test.

KarstenSchnitter

Thanks for getting this PR ready. I like the end-to-end test. Can you please add another test case using protobuf message body and not JSON?

KarstenSchnitter · 2025-12-02T08:41:37Z

...egrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java

+                        .authority(String.format("127.0.0.1:%d", port))
+                        .method(HttpMethod.POST)
+                        .path("/otel-logs-pipeline/logs")
+                        .contentType(MediaType.JSON_UTF_8)


This sends a JSON message. What about protobuf?

KarstenSchnitter · 2025-12-02T08:42:04Z

...egrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java

+                        )
+        ).build();
+
+        return HttpData.copyOf(JsonFormat.printer().print(exportLogsServiceRequest).getBytes());


This creates just the JSON representation. I think we also need the protobuf version.

KarstenSchnitter · 2025-12-02T08:42:39Z

...egrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java

+    private final ApacheLogFaker apacheLogFaker = new ApacheLogFaker();
+
+    @Test
+    public void testOtelLogsSourcePipelineEndToEnd() throws InvalidProtocolBufferException {


Can you please add another test case for HTTP/protobuf format?

KarstenSchnitter · 2025-12-05T21:16:40Z

please fix the unused imports warning.

KarstenSchnitter · 2025-12-20T13:05:09Z

@TomasLongo I think I am missing something here. I can see a test using HTTP/JSON and one for GRPC protocol. But I am missing one using HTTP/protobuf.

I would expect a different content type at https://github.com/sternadsoftware/data-prepper-second/blob/e9ebc47cba8877c99d4ec5d89cdd115ba4af7d59/e2e-test/log/src/integrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java#L110 accompanied with a different payload. The payload can be generated calling exportLogsServiceRequest.toByteArray() at https://github.com/sternadsoftware/data-prepper-second/blob/e9ebc47cba8877c99d4ec5d89cdd115ba4af7d59/e2e-test/log/src/integrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java#L78.

Can you add this as a second test case?

TomasLongo · 2025-12-22T07:12:23Z

Nope. All good. I simply was not getting it.

I've added a unit test for enabled unframed requests. Will add an E2E-Test today.

Signed-off-by: Tomas Longo <tlongo@sternad.de>

KarstenSchnitter

Please fix the license header issues. I think this is a rather new check. Also have a look at the media type in the integration test.

KarstenSchnitter · 2026-01-19T13:57:52Z

...egrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java

+                .authority(String.format("127.0.0.1:%d", SOURCE_PORT))
+                .method(HttpMethod.POST)
+                .path(path)
+                .contentType(MediaType.JSON_UTF_8)


Shouldn't this media type depend on the payload? There is also protobuf data being ingested.

Signed-off-by: Tomas Longo <tlongo@sternad.de>

...egrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java

KarstenSchnitter

I tried to address the license header checks with comments. Please review and merge. You can find the explanation at https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md#license-headers.

...rc/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/ArmeriaHttpService.java

.../main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/HttpExceptionHandler.java

...rg/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSourceGrpcServiceRequestsTest.java

...src/test/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSourceHttpTest.java

...va/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSourceUnframedRequestTest.java

...tionTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceGrpcTest.java

...egrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java

e2e-test/log/src/integrationTest/resources/otel-logs-source-pipeline.yml

e2e-test/log/src/integrationTest/resources/otel-logs-source-unframed-requests-pipeline.yml

Signed-off-by: Tomas Longo <tlongo@sternad.de>

KarstenSchnitter

Thanks for providing the PR and implementing all the changes for my comments. @dlvenable and @kkondaka this is good to go from our side. Do you still have change requests or can we merge this PR?

KarstenSchnitter

@TomasLongo #6494 shows a missing configuration in the last change. This needs to be applied here, too.

KarstenSchnitter · 2026-02-06T05:33:22Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/ArmeriaHttpService.java

+
+    public ArmeriaHttpService(Buffer<Record<Object>> buffer, final PluginMetrics pluginMetrics, final int bufferWriteTimeoutInMillis) {
+        this.buffer = buffer;
+        this.oTelProtoDecoder = new OTelProtoOpensearchCodec.OTelProtoDecoder();


This needs to respect the output format. See #6494 .

KarstenSchnitter · 2026-02-06T05:34:16Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/ArmeriaHttpService.java

+
+    public ArmeriaHttpService(Buffer<Record<Object>> buffer, final PluginMetrics pluginMetrics, final int bufferWriteTimeoutInMillis) {
+        this.buffer = buffer;
+        this.oTelProtoDecoder = new OTelProtoOpensearchCodec.OTelProtoDecoder();


This needs to respect the output format. See #6494

Signed-off-by: Tomas Longo <tlongo@sternad.de>

KarstenSchnitter

I think we should follow the idea of @dlvenable to have the logic which decoder should be used centralized in the OTelLogsSource.

KarstenSchnitter · 2026-02-06T17:54:58Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/ArmeriaHttpService.java

+            final OTelOutputFormat otelOutputFormat
+    ) {
+        this.buffer = buffer;
+        this.oTelProtoDecoder = (otelOutputFormat == OTelOutputFormat.OPENSEARCH) ? new OTelProtoOpensearchCodec.OTelProtoDecoder() : new OTelProtoStandardCodec.OTelProtoDecoder();


@dlvenable suggested to inject the oTelProtoDecoder directly and have just a single point, where the decision is taken in #6494 (comment) for the OtelTraceSource. I think this applies here as well. Let the OTelLogsSource provide the decoder directly and take the decision in that class.

KarstenSchnitter · 2026-02-06T17:56:42Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSource.java

+                .exceptionHandler(createGrpExceptionHandler(oTelLogsSourceConfig));
+        final OTelLogsGrpcService oTelLogsGrpcService = new OTelLogsGrpcService(
+                (int) (oTelLogsSourceConfig.getRequestTimeoutInMillis() * 0.8),
+                oTelLogsSourceConfig.getOutputFormat() == OTelOutputFormat.OPENSEARCH ? new OTelProtoOpensearchCodec.OTelProtoDecoder() : new OTelProtoStandardCodec.OTelProtoDecoder(),


Extract this logic into a private method and use the same method in line 183.

KarstenSchnitter · 2026-02-06T17:57:25Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSource.java

+        LOG.info("Configuring HTTP service under {} ", path);
+
+
+        final ArmeriaHttpService armeriaHttpService = new ArmeriaHttpService(buffer, pluginMetrics, 100, oTelLogsSourceConfig.getOutputFormat());


create the decoder directly as in line 213 and pass it to the ArmeriaHttpService.

KarstenSchnitter · 2026-02-17T14:04:52Z

@TomasLongo can you make the suggested change for the ProtoDecoder. Afterwards, I really think, this is good to go.

TomasLongo · 2026-02-18T07:52:11Z

@KarstenSchnitter Sorry for the delay. Your comments were burried in my mail box. I will implement your suggestion.

Signed-off-by: Tomas Longo <tlongo@sternad.de>

KarstenSchnitter

This looks good to me. Thanks for the PR.

KarstenSchnitter · 2026-02-18T14:17:35Z

@dlvenable any reservations or can we proceed and merge this PR?

KarstenSchnitter reviewed Nov 7, 2025

View reviewed changes

TomasLongo force-pushed the otel-logs-source-http-service branch 3 times, most recently from 1dd14dd to 965b93d Compare November 14, 2025 07:18

KarstenSchnitter reviewed Nov 15, 2025

View reviewed changes

TomasLongo force-pushed the otel-logs-source-http-service branch from e05da93 to 656475c Compare November 21, 2025 08:27

KarstenSchnitter reviewed Nov 21, 2025

View reviewed changes

TomasLongo force-pushed the otel-logs-source-http-service branch from f6c5b40 to 5f2aeb3 Compare November 26, 2025 07:12

TomasLongo marked this pull request as ready for review November 30, 2025 21:52

TomasLongo requested review from chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners November 30, 2025 21:52

KarstenSchnitter requested changes Dec 2, 2025

View reviewed changes

TomasLongo added 5 commits January 7, 2026 11:24

[WIP]

646b02b

Signed-off-by: Tomas Longo <tlongo@sternad.de>

[WIP]

c52d4c9

Signed-off-by: Tomas Longo <tlongo@sternad.de>

[WIP]

ec77c3b

Signed-off-by: Tomas Longo <tlongo@sternad.de>

[WIP] Integrate http service and make sure it works properly

5f863a4

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Integrate grpc and http service into a single server

7747436

Signed-off-by: Tomas Longo <tlongo@sternad.de>

TomasLongo added 11 commits January 7, 2026 11:25

Revert introduction of CertificatProvider interface

3f21bce

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Cleanup

8be400a

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Fix return value of http service

2efce64

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Re-introduce unframed request for the grpc service

be398bf

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Incorporate review suggestions

53e7126

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Add E2E test

64382fa

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Add E2E test for gRPC

bdac9f2

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Remove unused imports

ac428aa

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Add test for unframed requests

0b23b4d

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Add e2e test for unframed requests

4207c71

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Add e2e test for protobuf requests

7269fe9

Signed-off-by: Tomas Longo <tlongo@sternad.de>

TomasLongo force-pushed the otel-logs-source-http-service branch from 148ee32 to 7269fe9 Compare January 7, 2026 10:25

KarstenSchnitter requested changes Jan 19, 2026

View reviewed changes

Fix media type for protobuf payload

e1a839d

Signed-off-by: Tomas Longo <tlongo@sternad.de>

KarstenSchnitter reviewed Jan 21, 2026

View reviewed changes

...egrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java Show resolved Hide resolved

KarstenSchnitter reviewed Jan 21, 2026

View reviewed changes

KarstenSchnitter mentioned this pull request Jan 22, 2026

Expand OTLP Sink to support any compatible endpoint and support otlphttp protocol #6330

Open

Update license headers

5ad01cc

Signed-off-by: Tomas Longo <tlongo@sternad.de>

KarstenSchnitter approved these changes Feb 1, 2026

View reviewed changes

KarstenSchnitter requested changes Feb 6, 2026

View reviewed changes

TomasLongo mentioned this pull request Feb 6, 2026

Fix OTEL trace source broken by PR 5322 #6494

Merged

2 tasks

Adhere to config when it comes to chose a codec

302c619

Signed-off-by: Tomas Longo <tlongo@sternad.de>

KarstenSchnitter reviewed Feb 6, 2026

View reviewed changes

Inject OtelProtoCodec into ArmeriaHttpService

8285d1f

Signed-off-by: Tomas Longo <tlongo@sternad.de>

KarstenSchnitter approved these changes Feb 18, 2026

View reviewed changes

KarstenSchnitter merged commit 8d85609 into opensearch-project:main Feb 20, 2026
71 of 72 checks passed

oeyh mentioned this pull request Feb 27, 2026

Fix breaking changes in otel_logs_source #6572

Merged

4 tasks

	LOG.error("Failed to parse the request {} due to:", request, e);
	LOG.warn(DataPrepperMarkers.SENSITIVE, "Failed to parse request with error '{}'. Request body: {}.", e.getMessage(), request);

		LOG.info("Configuring HTTP service under {} ", path);


		final ArmeriaHttpService armeriaHttpService = new ArmeriaHttpService(buffer, pluginMetrics, 100, oTelLogsSourceConfig.getOutputFormat());

Conversation

TomasLongo commented Nov 7, 2025

Description

Breakdown

One Server, two services

Unframed Requests

CreateServer

Refactoring of tests

Decisions made and additional considerations

Issues Resolved

Check List

Uh oh!

TomasLongo commented Nov 7, 2025

Uh oh!

KarstenSchnitter left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

TomasLongo Nov 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

KarstenSchnitter left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

KarstenSchnitter left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

TomasLongo commented Nov 21, 2025

Uh oh!

KarstenSchnitter left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

KarstenSchnitter commented Dec 5, 2025

Uh oh!

KarstenSchnitter commented Dec 20, 2025

Uh oh!

TomasLongo commented Dec 22, 2025

Uh oh!

KarstenSchnitter left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

KarstenSchnitter left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

TomasLongo Nov 10, 2025 •

edited

Loading

KarstenSchnitter left a comment •

edited

Loading