Skip to content

Commit

Permalink
addressed PR comments and added SSL tests
Browse files Browse the repository at this point in the history 
Signed-off-by: Jeremy Michael <jsusanto@amazon.com>
  • Loading branch information
Jeremy Michael committed Dec 4, 2024
1 parent 95ff177 commit 8d3499e
Show file tree
Hide file tree
Showing 2 changed files with 70 additions and 59 deletions.
61 changes: 34 additions & 27 deletions ...est/java/org/opensearch/dataprepper/plugins/source/otelmetrics/OTelMetricsSourceTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
import com.linecorp.armeria.client.Clients;
import com.linecorp.armeria.client.WebClient;
import com.linecorp.armeria.common.AggregatedHttpResponse;
import com.linecorp.armeria.common.ClosedSessionException;
import com.linecorp.armeria.common.HttpData;
import com.linecorp.armeria.common.HttpHeaderNames;
import com.linecorp.armeria.common.HttpMethod;
Expand All @@ -35,6 +36,7 @@
import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceResponse;
import io.opentelemetry.proto.collector.metrics.v1.MetricsServiceGrpc;
import io.opentelemetry.proto.collector.trace.v1.TraceServiceGrpc;
import io.opentelemetry.proto.metrics.v1.NumberDataPoint;

import io.opentelemetry.proto.common.v1.InstrumentationLibrary;
Expand Down Expand Up @@ -96,13 +98,15 @@
import java.util.Map;
import java.util.StringJoiner;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.zip.GZIPOutputStream;

import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.equalTo;
Expand Down Expand Up @@ -480,45 +484,48 @@ void testGrpcRequestWithInvalidCredentials_with_unsuccessful_response() throws E
}

@Test
void testHttpsFullJsonWithCustomPathAndAuthHeaderAndSsl_with_unsuccessful_response() throws InvalidProtocolBufferException {
void testHttpWithoutSslFailsWhenSslIsEnabled() throws InvalidProtocolBufferException {
when(oTelMetricsSourceConfig.isSsl()).thenReturn(true);
when(oTelMetricsSourceConfig.getSslKeyCertChainFile()).thenReturn("data/certificate/test_cert.crt");
when(oTelMetricsSourceConfig.getSslKeyFile()).thenReturn("data/certificate/test_decrypted_key.key");
when(oTelMetricsSourceConfig.getAuthentication()).thenReturn(new PluginModel("http_basic",
Map.of(
"username", USERNAME,
"password", PASSWORD
)));
when(httpBasicAuthenticationConfig.getUsername()).thenReturn(USERNAME);
when(httpBasicAuthenticationConfig.getPassword()).thenReturn(PASSWORD);
configureObjectUnderTest();
SOURCE.start(buffer);

final GrpcAuthenticationProvider authenticationProvider = new GrpcBasicAuthenticationProvider(httpBasicAuthenticationConfig);
when(pluginFactory.loadPlugin(eq(GrpcAuthenticationProvider.class), any(PluginSetting.class)))
.thenReturn(authenticationProvider);
WebClient client = WebClient.builder("http://127.0.0.1:21891")
.build();

when(oTelMetricsSourceConfig.enableUnframedRequests()).thenReturn(true);
when(oTelMetricsSourceConfig.getPath()).thenReturn(TEST_PATH);
CompletionException exception = assertThrows(CompletionException.class, () -> client.execute(RequestHeaders.builder()
.scheme(SessionProtocol.HTTP)
.authority("127.0.0.1:21891")
.method(HttpMethod.POST)
.path("/opentelemetry.proto.collector.metrics.v1.MetricsService/Export")
.contentType(MediaType.JSON_UTF_8)
.build(),
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);

final String transformedPath = "/" + TEST_PIPELINE_NAME + "/v1/metrics";

WebClient client = WebClient.builder("https://127.0.0.1:21891")
.factory(ClientFactory.builder().tlsNoVerify().build())
.build();
MetricsServiceGrpc.MetricsServiceBlockingStub client = Clients.builder(GRPC_ENDPOINT)
.build(MetricsServiceGrpc.MetricsServiceBlockingStub.class);

client.prepare()
.post(transformedPath)
.content(MediaType.JSON_UTF_8, JsonFormat.printer().print(createExportMetricsRequest()).getBytes())
.execute()
.aggregate()
.whenComplete((response, throwable) -> {
assertSecureResponseWithStatusCode(response, HttpStatus.UNAUTHORIZED, throwable);
})
.join();
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)) {
Expand Down
68 changes: 36 additions & 32 deletions ...rc/test/java/org/opensearch/dataprepper/plugins/source/oteltrace/OTelTraceSourceTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import com.linecorp.armeria.client.Clients;
import com.linecorp.armeria.client.WebClient;
import com.linecorp.armeria.common.AggregatedHttpResponse;
import com.linecorp.armeria.common.ClosedSessionException;
import com.linecorp.armeria.common.HttpData;
import com.linecorp.armeria.common.HttpHeaderNames;
import com.linecorp.armeria.common.HttpMethod;
Expand Down Expand Up @@ -93,13 +94,15 @@
import java.util.StringJoiner;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.zip.GZIPOutputStream;

import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.equalTo;
Expand Down Expand Up @@ -523,46 +526,47 @@ void testGrpcRequestWithoutAuthentication_with_unsuccessful_response() throws Ex
}

@Test
void testHttpsFullJsonWithCustomPathAndAuthHeaderAndSsl_with_unsuccessful_response() throws Exception {
void testHttpWithoutSslFailsWhenSslIsEnabled() throws InvalidProtocolBufferException {
when(oTelTraceSourceConfig.isSsl()).thenReturn(true);
when(oTelTraceSourceConfig.getSslKeyCertChainFile()).thenReturn("data/certificate/test_cert.crt");
when(oTelTraceSourceConfig.getSslKeyFile()).thenReturn("data/certificate/test_decrypted_key.key");
when(oTelTraceSourceConfig.getAuthentication()).thenReturn(new PluginModel("http_basic",
Map.of(
"username", USERNAME,
"password", PASSWORD
)));
when(httpBasicAuthenticationConfig.getUsername()).thenReturn(USERNAME);
when(httpBasicAuthenticationConfig.getPassword()).thenReturn(PASSWORD);

final GrpcAuthenticationProvider authenticationProvider = new GrpcBasicAuthenticationProvider(httpBasicAuthenticationConfig);
when(pluginFactory.loadPlugin(eq(GrpcAuthenticationProvider.class), any(PluginSetting.class)))
.thenReturn(authenticationProvider);

when(oTelTraceSourceConfig.enableUnframedRequests()).thenReturn(true);
when(oTelTraceSourceConfig.getPath()).thenReturn(TEST_PATH);

configureObjectUnderTest();
SOURCE.start(buffer);

final String transformedPath = "/" + TEST_PIPELINE_NAME + "/v1/traces";

WebClient client = WebClient.builder("https://127.0.0.1:21890")
.factory(ClientFactory.builder().tlsNoVerify().build())

WebClient client = WebClient.builder("http://127.0.0.1:21890")
.build();

client.prepare()
.post(transformedPath)
.content(MediaType.JSON_UTF_8, JsonFormat.printer().print(createExportTraceRequest()).getBytes())
.execute()

CompletionException exception = assertThrows(CompletionException.class, () -> client.execute(RequestHeaders.builder()
.scheme(SessionProtocol.HTTP)
.authority("127.0.0.1:21890")
.method(HttpMethod.POST)
.path("/opentelemetry.proto.collector.trace.v1.TraceService/Export")
.contentType(MediaType.JSON_UTF_8)
.build(),
HttpData.copyOf(JsonFormat.printer().print(createExportTraceRequest()).getBytes()))
.aggregate()
.whenComplete((response, throwable) -> {
assertSecureResponseWithStatusCode(response, HttpStatus.UNAUTHORIZED, throwable);
})
.join();
.join());

assertThat(exception.getCause(), instanceOf(ClosedSessionException.class));
}



@Test
void testGrpcFailsIfSslIsEnabledAndNoTls() {
when(oTelTraceSourceConfig.isSsl()).thenReturn(true);
when(oTelTraceSourceConfig.getSslKeyCertChainFile()).thenReturn("data/certificate/test_cert.crt");
when(oTelTraceSourceConfig.getSslKeyFile()).thenReturn("data/certificate/test_decrypted_key.key");
configureObjectUnderTest();
SOURCE.start(buffer);

TraceServiceGrpc.TraceServiceBlockingStub client = Clients.builder(GRPC_ENDPOINT)
.build(TraceServiceGrpc.TraceServiceBlockingStub.class);

StatusRuntimeException actualException = assertThrows(StatusRuntimeException.class, () -> client.export(createExportTraceRequest()));

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)) {
Expand Down

0 comments on commit 8d3499e

Please sign in to comment.