Introduce otlp/http support in OTelTraceSource

data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/GrpcRequestExceptionHandler.java

@@ -39,14 +39,14 @@ public class GrpcRequestExceptionHandler implements GoogleGrpcExceptionHandlerFu
     private final Counter badRequestsCounter;
     private final Counter requestsTooLargeCounter;
     private final Counter internalServerErrorCounter;
-    private final GrpcRetryInfoCalculator retryInfoCalculator;
+    private final RetryInfoCalculator retryInfoCalculator;
 
     public GrpcRequestExceptionHandler(final PluginMetrics pluginMetrics, Duration retryInfoMinDelay, Duration retryInfoMaxDelay) {
         requestTimeoutsCounter = pluginMetrics.counter(REQUEST_TIMEOUTS);
         badRequestsCounter = pluginMetrics.counter(BAD_REQUESTS);
         requestsTooLargeCounter = pluginMetrics.counter(REQUESTS_TOO_LARGE);
         internalServerErrorCounter = pluginMetrics.counter(INTERNAL_SERVER_ERROR);
-        retryInfoCalculator = new GrpcRetryInfoCalculator(retryInfoMinDelay, retryInfoMaxDelay);
+        retryInfoCalculator = new RetryInfoCalculator(retryInfoMinDelay, retryInfoMaxDelay);
     }
 
     @Override

data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/GrpcRetryInfoCalculator.java → data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/RetryInfoCalculator.java

@@ -6,15 +6,15 @@
 import java.time.Instant;
 import java.util.concurrent.atomic.AtomicReference;
 
-class GrpcRetryInfoCalculator {
+public class RetryInfoCalculator {
 
     private final Duration minimumDelay;
     private final Duration maximumDelay;
 
     private final AtomicReference<Instant> lastTimeCalled;
     private final AtomicReference<Duration> nextDelay;
 
-    GrpcRetryInfoCalculator(Duration minimumDelay, Duration maximumDelay) {
+    public RetryInfoCalculator(Duration minimumDelay, Duration maximumDelay) {
         this.minimumDelay = minimumDelay;
         this.maximumDelay = maximumDelay;
         // Create a cushion so that the calculator treats a first quick exception (after prepper startup) as normal request (e.g. does not calculate a backoff)
@@ -34,7 +34,7 @@ private static com.google.protobuf.Duration.Builder mapDuration(Duration duratio
         return com.google.protobuf.Duration.newBuilder().setSeconds(duration.getSeconds()).setNanos(duration.getNano());
     }
 
-    RetryInfo createRetryInfo() {
+    public RetryInfo createRetryInfo() {
         Instant now = Instant.now();
         // Is the last time we got called longer ago than the next delay?
         if (lastTimeCalled.getAndSet(now).isBefore(now.minus(nextDelay.get()))) {

data-prepper-plugins/armeria-common/src/test/java/org/opensearch/dataprepper/GrpcRetryInfoCalculatorTest.java

@@ -12,16 +12,16 @@ public class GrpcRetryInfoCalculatorTest {
 
     @Test
     public void testMinimumDelayOnFirstCall() {
-        RetryInfo retryInfo = new GrpcRetryInfoCalculator(Duration.ofMillis(100), Duration.ofSeconds(1)).createRetryInfo();
+        RetryInfo retryInfo = new RetryInfoCalculator(Duration.ofMillis(100), Duration.ofSeconds(1)).createRetryInfo();
 
         assertThat(retryInfo.getRetryDelay().getNanos(), equalTo(100_000_000));
         assertThat(retryInfo.getRetryDelay().getSeconds(), equalTo(0L));
     }
 
     @Test
     public void testExponentialBackoff() {
-        GrpcRetryInfoCalculator calculator =
-                new GrpcRetryInfoCalculator(Duration.ofSeconds(1), Duration.ofSeconds(10));
+        RetryInfoCalculator calculator =
+                new RetryInfoCalculator(Duration.ofSeconds(1), Duration.ofSeconds(10));
         RetryInfo retryInfo1 = calculator.createRetryInfo();
         RetryInfo retryInfo2 = calculator.createRetryInfo();
         RetryInfo retryInfo3 = calculator.createRetryInfo();
@@ -35,8 +35,8 @@ public void testExponentialBackoff() {
 
     @Test
     public void testUsesMaximumAsLongestDelay() {
-        GrpcRetryInfoCalculator calculator =
-                new GrpcRetryInfoCalculator(Duration.ofSeconds(1), Duration.ofSeconds(2));
+        RetryInfoCalculator calculator =
+                new RetryInfoCalculator(Duration.ofSeconds(1), Duration.ofSeconds(2));
         RetryInfo retryInfo1 = calculator.createRetryInfo();
         RetryInfo retryInfo2 = calculator.createRetryInfo();
         RetryInfo retryInfo3 = calculator.createRetryInfo();
@@ -49,8 +49,8 @@ public void testUsesMaximumAsLongestDelay() {
     @Test
     public void testResetAfterDelayWearsOff() throws InterruptedException {
         int minDelayNanos = 1_000_000;
-        GrpcRetryInfoCalculator calculator =
-                new GrpcRetryInfoCalculator(Duration.ofNanos(minDelayNanos), Duration.ofSeconds(1));
+        RetryInfoCalculator calculator =
+                new RetryInfoCalculator(Duration.ofNanos(minDelayNanos), Duration.ofSeconds(1));
 
         RetryInfo retryInfo1 = calculator.createRetryInfo();
         RetryInfo retryInfo2 = calculator.createRetryInfo();
@@ -66,8 +66,8 @@ public void testResetAfterDelayWearsOff() throws InterruptedException {
 
     @Test
     public void testQuickFirstExceptionDoesNotTriggerBackoffCalculationEvenWithLongMinDelay() throws InterruptedException {
-        GrpcRetryInfoCalculator calculator =
-                new GrpcRetryInfoCalculator(Duration.ofSeconds(10), Duration.ofSeconds(20));
+        RetryInfoCalculator calculator =
+                new RetryInfoCalculator(Duration.ofSeconds(10), Duration.ofSeconds(20));
 
         RetryInfo retryInfo1 = calculator.createRetryInfo();
         RetryInfo retryInfo2 = calculator.createRetryInfo();

data-prepper-plugins/otel-metrics-source/src/test/java/org/opensearch/dataprepper/plugins/source/otelmetrics/OTelMetricsSourceTest.java

@@ -350,7 +350,6 @@ void testHttpFullJsonWithCustomPathAndUnframedRequests() throws InvalidProtocolB
                 .join();
     }
 
-
     @Test
     void testHttpFullJsonWithCustomPathAndAuthHeader_with_successful_response() throws InvalidProtocolBufferException {
         when(httpBasicAuthenticationConfig.getUsername()).thenReturn(USERNAME);
@@ -420,7 +419,7 @@ void testHttpRequestWithInvalidCredentials_with_unsuccessful_response() throws I
         when(httpBasicAuthenticationConfig.getUsername()).thenReturn(USERNAME);
         when(httpBasicAuthenticationConfig.getPassword()).thenReturn(PASSWORD);
         final GrpcAuthenticationProvider grpcAuthenticationProvider = new GrpcBasicAuthenticationProvider(httpBasicAuthenticationConfig);
-    
+
         when(pluginFactory.loadPlugin(eq(GrpcAuthenticationProvider.class), any(PluginSetting.class)))
                 .thenReturn(grpcAuthenticationProvider);
         when(oTelMetricsSourceConfig.getAuthentication()).thenReturn(new PluginModel("http_basic",
@@ -430,17 +429,17 @@ void testHttpRequestWithInvalidCredentials_with_unsuccessful_response() throws I
                 )));
         when(oTelMetricsSourceConfig.enableUnframedRequests()).thenReturn(true);
         when(oTelMetricsSourceConfig.getPath()).thenReturn(TEST_PATH);
-    
+
         configureObjectUnderTest();
         SOURCE.start(buffer);
-    
+
         final String invalidUsername = "wrong_user";
         final String invalidPassword = "wrong_password";
         final String invalidCredentials = Base64.getEncoder()
                 .encodeToString(String.format("%s:%s", invalidUsername, invalidPassword).getBytes(StandardCharsets.UTF_8));
-    
+
         final String transformedPath = "/" + TEST_PIPELINE_NAME + "/v1/metrics";
-    
+
         WebClient.of().prepare()
                 .post("http://127.0.0.1:21891" + transformedPath)
                 .content(MediaType.JSON_UTF_8, JsonFormat.printer().print(createExportMetricsRequest()).getBytes())
@@ -450,7 +449,7 @@ void testHttpRequestWithInvalidCredentials_with_unsuccessful_response() throws I
                 .whenComplete((response, throwable) -> assertSecureResponseWithStatusCode(response, HttpStatus.UNAUTHORIZED, throwable))
                 .join();
     }
-    
+
     @Test
     void testGrpcRequestWithInvalidCredentials_with_unsuccessful_response() throws Exception {
         when(httpBasicAuthenticationConfig.getUsername()).thenReturn(USERNAME);
@@ -489,10 +488,10 @@ void testHttpWithoutSslFailsWhenSslIsEnabled() throws InvalidProtocolBufferExcep
         when(oTelMetricsSourceConfig.getSslKeyFile()).thenReturn("data/certificate/test_decrypted_key.key");
         configureObjectUnderTest();
         SOURCE.start(buffer);
-    
+
         WebClient client = WebClient.builder("http://127.0.0.1:21891")
                 .build();
-    
+
         CompletionException exception = assertThrows(CompletionException.class, () -> client.execute(RequestHeaders.builder()
                         .scheme(SessionProtocol.HTTP)
                         .authority("127.0.0.1:21891")
@@ -503,28 +502,28 @@ void testHttpWithoutSslFailsWhenSslIsEnabled() throws InvalidProtocolBufferExcep
                 HttpData.copyOf(JsonFormat.printer().print(createExportMetricsRequest()).getBytes()))
                 .aggregate()
                 .join());
-    
+
         assertThat(exception.getCause(), instanceOf(ClosedSessionException.class));
     }
-    
+
     @Test
     void testGrpcFailsIfSslIsEnabledAndNoTls() {
         when(oTelMetricsSourceConfig.isSsl()).thenReturn(true);
         when(oTelMetricsSourceConfig.getSslKeyCertChainFile()).thenReturn("data/certificate/test_cert.crt");
         when(oTelMetricsSourceConfig.getSslKeyFile()).thenReturn("data/certificate/test_decrypted_key.key");
         configureObjectUnderTest();
         SOURCE.start(buffer);
-    
+
         MetricsServiceGrpc.MetricsServiceBlockingStub client = Clients.builder(GRPC_ENDPOINT)
                 .build(MetricsServiceGrpc.MetricsServiceBlockingStub.class);
-    
+
         StatusRuntimeException actualException = assertThrows(StatusRuntimeException.class, () -> client.export(createExportMetricsRequest()));
-  
+
         assertThat(actualException.getStatus(), notNullValue());
         assertThat(actualException.getStatus().getCode(), equalTo(Status.Code.UNKNOWN));
     }
-    
-    
+
+
     @Test
     void testServerStartCertFileSuccess() throws IOException {
         try (MockedStatic<Server> armeriaServerMock = Mockito.mockStatic(Server.class)) {

data-prepper-plugins/otel-trace-source/build.gradle

@@ -32,6 +32,7 @@ dependencies {
     testImplementation 'org.assertj:assertj-core:3.27.3'
     testImplementation testLibs.slf4j.simple
     testImplementation libs.commons.io
+    testImplementation 'com.jayway.jsonpath:json-path-assert:2.6.0'
     testImplementation 'org.skyscreamer:jsonassert:1.5.3'
 }