Fix requestsTooLarge metric reporting when decompression buffer overflows on armeria in otel logs source

Signed-off-by: Taylor Gray <tylgry@amazon.com>

Author: graytaylor0

data-prepper-plugins/otel-logs-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSource.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.dataprepper.plugins.source.otellogs;
 
+import com.linecorp.armeria.common.ContentTooLargeException;
 import com.linecorp.armeria.common.SessionProtocol;
 import com.linecorp.armeria.common.grpc.GrpcExceptionHandlerFunction;
 import com.linecorp.armeria.server.Server;
@@ -14,7 +15,11 @@
 import com.linecorp.armeria.server.grpc.GrpcServiceBuilder;
 import com.linecorp.armeria.server.healthcheck.HealthCheckService;
 import com.linecorp.armeria.server.throttling.ThrottlingService;
-
+import io.grpc.ServerInterceptor;
+import io.grpc.ServerInterceptors;
+import io.grpc.protobuf.services.ProtoReflectionService;
+import io.micrometer.core.instrument.Counter;
+import io.opentelemetry.proto.collector.logs.v1.LogsServiceGrpc;
 import org.opensearch.dataprepper.GrpcRequestExceptionHandler;
 import org.opensearch.dataprepper.armeria.authentication.GrpcAuthenticationProvider;
 import org.opensearch.dataprepper.http.LogThrottlingRejectHandler;
@@ -57,11 +62,6 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 
-import io.grpc.ServerInterceptor;
-import io.grpc.ServerInterceptors;
-import io.grpc.protobuf.services.ProtoReflectionService;
-import io.opentelemetry.proto.collector.logs.v1.LogsServiceGrpc;
-
 @DataPrepperPlugin(name = "otel_logs_source", pluginType = Source.class, pluginConfigurationType = OTelLogsSourceConfig.class)
 public class OTelLogsSource implements Source<Record<Object>> {
     private static final Logger LOG = LoggerFactory.getLogger(OTelLogsSource.class);
@@ -75,6 +75,7 @@ public class OTelLogsSource implements Source<Record<Object>> {
     private final CertificateProviderFactory certificateProviderFactory;
     private final ByteDecoder byteDecoder;
     private final PluginFactory pluginFactory;
+    final Counter requestsTooLargeCounter;
     private Server server;
 
     @DataPrepperPluginConstructor
@@ -95,6 +96,7 @@ public OTelLogsSource(final OTelLogsSourceConfig oTelLogsSourceConfig,
         this.pipelineName = pipelineDescription.getPipelineName();
         this.pluginFactory = pluginFactory;
         this.byteDecoder = new OTelLogsDecoder(oTelLogsSourceConfig.getOutputFormat());
+        this.requestsTooLargeCounter = pluginMetrics.counter(HttpExceptionHandler.REQUESTS_TOO_LARGE);
     }
 
     @Override
@@ -144,6 +146,12 @@ private Server createServer(ServerBuilder serverBuilder, Buffer<Record<Object>>
             serverBuilder.http(oTelLogsSourceConfig.getPort());
         }
 
+        serverBuilder.accessLogWriter(log -> {
+            if (log.responseCause() instanceof ContentTooLargeException) {
+                requestsTooLargeCounter.increment();
+            }
+        }, true);
+
         final GrpcAuthenticationProvider authProvider = createGrpcAuthenticationProvider(pluginFactory);
         authProvider.getHttpAuthenticationService().ifPresent(serverBuilder::decorator);
 

data-prepper-plugins/otel-logs-source/src/test/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSourceHttpTest.java

@@ -46,6 +46,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.UUID;
 import java.util.Base64;
 import java.util.List;
 import java.util.Map;
@@ -355,6 +356,72 @@ void httpRequest_requestBodyIsTooLarge_returns413() throws InvalidProtocolBuffer
                 .join();
     }
 
+    @Test
+    void httpRequest_compressedPayloadDecompressesTooLarge_incrementsRequestsTooLargeMetric() throws Exception {
+        configureSource(createDefaultConfigBuilder()
+                .httpPath(CONFIG_HTTP_PATH)
+                .compression(CompressionOption.GZIP)
+                .maxRequestLength(ByteCount.ofBytes(250))
+                .build());
+        SOURCE.start(buffer);
+
+        // ~720 bytes uncompressed, compresses well — compressed fits under maxRequestLength
+        // but decompressed exceeds 250 bytes, triggering ContentTooLargeException
+        String largePayload = UUID.randomUUID().toString().repeat(20);
+        byte[] compressedPayload = createGZipCompressedPayload(largePayload);
+
+        WebClient.of().execute(
+                        getDefaultRequestHeadersBuilder()
+                                .add(HttpHeaderNames.CONTENT_ENCODING, "gzip")
+                                .build(),
+                        HttpData.copyOf(compressedPayload))
+                .aggregate()
+                .whenComplete((response, throwable) -> {
+                    assertThat("Http Response Throwable", throwable, is(nullValue()));
+                    assertThat("Expected 413 or 507 for too-large decompressed payload",
+                            response.status().code() == 413 || response.status().code() == 507, is(true));
+                })
+                .join();
+
+        Thread.sleep(200);
+
+        List<Measurement> measurements = MetricsTestUtil.getMeasurementList(
+                "pipeline.otel_logs.requestsTooLarge");
+        Measurement countMeasurement = MetricsTestUtil.getMeasurementFromList(
+                measurements, Statistic.COUNT);
+        assertEquals(1.0, countMeasurement.getValue());
+    }
+
+    @Test
+    void httpRequest_compressedPayloadWithinLimit_doesNotIncrementRequestsTooLargeMetric() throws Exception {
+        configureSource(createDefaultConfigBuilder()
+                .httpPath(CONFIG_HTTP_PATH)
+                .compression(CompressionOption.GZIP)
+                .build());
+        SOURCE.start(buffer);
+
+        byte[] compressedPayload = createGZipCompressedPayload(
+                JsonFormat.printer().print(createLogsServiceRequest()));
+
+        WebClient.of().execute(
+                        getDefaultRequestHeadersBuilder()
+                                .add(HttpHeaderNames.CONTENT_ENCODING, "gzip")
+                                .build(),
+                        compressedPayload)
+                .aggregate()
+                .whenComplete((response, throwable) ->
+                        assertSecureResponseWithStatusCode(response, HttpStatus.OK, throwable))
+                .join();
+
+        Thread.sleep(200);
+
+        List<Measurement> measurements = MetricsTestUtil.getMeasurementList(
+                "pipeline.otel_logs.requestsTooLarge");
+        Measurement countMeasurement = MetricsTestUtil.getMeasurementFromList(
+                measurements, Statistic.COUNT);
+        assertEquals(0.0, countMeasurement.getValue());
+    }
+
     static class BufferExceptionToStatusArgumentsProvider implements ArgumentsProvider {
         @Override
         public Stream<? extends Arguments> provideArguments(final ExtensionContext context) {
@@ -389,7 +456,7 @@ private ExportLogsServiceRequest createExportLogsRequest() {
     private void assertSecureResponseWithStatusCode(final AggregatedHttpResponse response,
                                                     final HttpStatus expectedStatus,
                                                     final Throwable throwable) {
-        assertThat("Http Status", response.status(), equalTo(expectedStatus));
+         assertThat("Http Status", response.status(), equalTo(expectedStatus));
         assertThat("Http Response Throwable", throwable, is(nullValue()));
 
         final List<String> headerKeys = response.headers()