Skip to content

Commit 3efd668

Browse files
oeyhoeyh
committed
Add a test with sample data
1 parent fe544ad commit 3efd668

3 files changed

Lines changed: 124 additions & 7 deletions

File tree

‎data-prepper-plugins/http-source/build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ dependencies {
2323
testImplementation 'org.assertj:assertj-core:3.27.0'
2424
testImplementation project(':data-prepper-api').sourceSets.test.output
2525
testImplementation project(':data-prepper-test-common')
26+
testImplementation project(':data-prepper-plugins:parse-json-processor')
2627
}
2728

2829
jacocoTestCoverageVerification {

‎data-prepper-plugins/http-source/src/test/java/org/opensearch/dataprepper/plugins/source/loghttp/HTTPSourceTest.java‎

Lines changed: 97 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.opensearch.dataprepper.metrics.MetricsTestUtil;
4444
import org.opensearch.dataprepper.metrics.PluginMetrics;
4545
import org.opensearch.dataprepper.model.CheckpointState;
46+
import org.opensearch.dataprepper.model.codec.InputCodec;
4647
import org.opensearch.dataprepper.model.configuration.PipelineDescription;
4748
import org.opensearch.dataprepper.model.configuration.PluginModel;
4849
import org.opensearch.dataprepper.model.configuration.PluginSetting;
@@ -56,6 +57,8 @@
5657
import org.opensearch.dataprepper.plugins.certificate.CertificateProvider;
5758
import org.opensearch.dataprepper.plugins.certificate.model.Certificate;
5859
import org.opensearch.dataprepper.plugins.codec.CompressionOption;
60+
import org.opensearch.dataprepper.plugins.codec.json.JsonInputCodec;
61+
import org.opensearch.dataprepper.plugins.codec.json.JsonInputCodecConfig;
5962

6063
import java.io.ByteArrayOutputStream;
6164
import java.io.File;
@@ -115,6 +118,7 @@ class HTTPSourceTest {
115118
private final String TEST_PIPELINE_NAME = "test_pipeline";
116119
private final String TEST_SSL_CERTIFICATE_FILE = getClass().getClassLoader().getResource("test_cert.crt").getFile();
117120
private final String TEST_SSL_KEY_FILE = getClass().getClassLoader().getResource("test_decrypted_key.key").getFile();
121+
private static final String CLOUDWATCH_LOGS_SAMPLE = "/cloudwatch-logs-sample.json";
118122

119123
@Mock
120124
private ServerBuilder serverBuilder;
@@ -152,10 +156,10 @@ class HTTPSourceTest {
152156
private PluginMetrics pluginMetrics;
153157
private PluginFactory pluginFactory;
154158

155-
private BlockingBuffer<Record<Log>> getBuffer() throws JsonProcessingException {
159+
private BlockingBuffer<Record<Log>> getBuffer(final int bufferSize, final int batchSize) throws JsonProcessingException {
156160
final HashMap<String, Object> integerHashMap = new HashMap<>();
157-
integerHashMap.put("buffer_size", 1);
158-
integerHashMap.put("batch_size", 1);
161+
integerHashMap.put("buffer_size", bufferSize);
162+
integerHashMap.put("batch_size", batchSize);
159163
ObjectMapper objectMapper = new ObjectMapper();
160164
String json = objectMapper.writeValueAsString(integerHashMap);
161165
BlockingBufferConfig blockingBufferConfig = objectMapper.readValue(json, BlockingBufferConfig.class);
@@ -233,7 +237,7 @@ public void setUp() throws JsonProcessingException {
233237
when(pluginFactory.loadPlugin(eq(ArmeriaHttpAuthenticationProvider.class), any(PluginSetting.class)))
234238
.thenReturn(authenticationProvider);
235239

236-
testBuffer = getBuffer();
240+
testBuffer = getBuffer(1, 1);
237241
when(pipelineDescription.getPipelineName()).thenReturn(TEST_PIPELINE_NAME);
238242
HTTPSourceUnderTest = new HTTPSource(sourceConfig, pluginMetrics, pluginFactory, pipelineDescription);
239243
}
@@ -710,7 +714,7 @@ void testHTTPSJsonResponse() throws JsonProcessingException {
710714
when(sourceConfig.getSslKeyFile()).thenReturn(TEST_SSL_KEY_FILE);
711715
HTTPSourceUnderTest = new HTTPSource(sourceConfig, pluginMetrics, pluginFactory, pipelineDescription);
712716

713-
testBuffer = getBuffer();
717+
testBuffer = getBuffer(1, 1);
714718
HTTPSourceUnderTest.start(testBuffer);
715719

716720
WebClient.builder().factory(ClientFactory.insecure()).build().execute(RequestHeaders.builder()
@@ -740,7 +744,7 @@ void testHTTPRequestWhenSSLRequiredNoResponse() throws JsonProcessingException {
740744
when(sourceConfig.getSslKeyFile()).thenReturn(TEST_SSL_KEY_FILE);
741745
HTTPSourceUnderTest = new HTTPSource(sourceConfig, pluginMetrics, pluginFactory, pipelineDescription);
742746

743-
testBuffer = getBuffer();
747+
testBuffer = getBuffer(1, 1);
744748
HTTPSourceUnderTest.start(testBuffer);
745749

746750
CompletableFuture<AggregatedHttpResponse> future = WebClient.builder()
@@ -777,7 +781,7 @@ void testHTTPSJsonResponse_with_custom_path_along_with_placeholder() throws Json
777781
when(sourceConfig.getSslKeyFile()).thenReturn(TEST_SSL_KEY_FILE);
778782
HTTPSourceUnderTest = new HTTPSource(sourceConfig, pluginMetrics, pluginFactory, pipelineDescription);
779783

780-
testBuffer = getBuffer();
784+
testBuffer = getBuffer(1, 1);
781785
HTTPSourceUnderTest.start(testBuffer);
782786

783787
final String path = "/" + TEST_PIPELINE_NAME + "/test";
@@ -935,4 +939,90 @@ public void request_that_exceeds_maxRequestLength_returns_413() {
935939
assertTrue(testBuffer.isEmpty());
936940
}
937941

942+
@Test
943+
public void testHTTPJsonCodec() throws IOException {
944+
// Prepare
945+
final String testData;
946+
try (InputStream inputStream = this.getClass().getResourceAsStream(CLOUDWATCH_LOGS_SAMPLE)) {
947+
testData = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
948+
}
949+
final int testPayloadSize = testData.getBytes().length;
950+
951+
when(sourceConfig.getCodec()).thenReturn(mock(PluginModel.class));
952+
when(sourceConfig.getMaxRequestLength()).thenReturn(ByteCount.ofBytes(2048000));
953+
954+
ObjectMapper mapper = new ObjectMapper();
955+
final String configString = "{\"key_name\": \"logEvents\",\"include_keys\":[\"owner\",\"logGroup\",\"logStream\"]}";
956+
final JsonInputCodecConfig codecConfig = mapper.readValue(configString, JsonInputCodecConfig.class);
957+
958+
final InputCodec codec = new JsonInputCodec(codecConfig);
959+
960+
when(pluginFactory.loadPlugin(eq(InputCodec.class), any(PluginSetting.class))).thenReturn(codec);
961+
962+
HTTPSourceUnderTest = new HTTPSource(sourceConfig, pluginMetrics, pluginFactory, pipelineDescription);
963+
testBuffer = getBuffer(5, 5);
964+
HTTPSourceUnderTest.start(testBuffer);
965+
refreshMeasurements();
966+
967+
// When
968+
WebClient.of().execute(RequestHeaders.builder()
969+
.scheme(SessionProtocol.HTTP)
970+
.authority("127.0.0.1:2021")
971+
.method(HttpMethod.POST)
972+
.path("/log/ingest")
973+
.contentType(MediaType.JSON_UTF_8)
974+
.build(),
975+
HttpData.ofUtf8(testData))
976+
.aggregate()
977+
.whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.OK)).join();
978+
979+
// Then
980+
assertFalse(testBuffer.isEmpty());
981+
982+
final Map.Entry<Collection<Record<Log>>, CheckpointState> result = testBuffer.read(100);
983+
List<Record<Log>> records = new ArrayList<>(result.getKey());
984+
assertEquals(3, records.size());
985+
986+
// Verify content
987+
final Record<Log> record = records.get(0);
988+
assertCommonFields(record);
989+
assertEquals("31953106606966983378809025079804211143289615424298221568", record.getData().get("id", String.class));
990+
assertEquals(1432826855000L, record.getData().get("timestamp", Long.class));
991+
assertEquals("{\"eventVersion\":\"1.01\",\"userIdentity\":{\"type\":\"Root\"}", record.getData().get("message", String.class));
992+
993+
final Record<Log> record2 = records.get(1);
994+
assertCommonFields(record2);
995+
assertEquals("31953106606966983378809025079804211143289615424298221569", record2.getData().get("id", String.class));
996+
assertEquals(1432826855001L, record2.getData().get("timestamp", Long.class));
997+
assertEquals("{\"eventVersion\":\"1.02\",\"userIdentity\":{\"type\":\"Root\"}", record2.getData().get("message", String.class));
998+
999+
final Record<Log> record3 = records.get(2);
1000+
assertCommonFields(record3);
1001+
assertEquals("31953106606966983378809025079804211143289615424298221570", record3.getData().get("id", String.class));
1002+
assertEquals(1432826855002L, record3.getData().get("timestamp", Long.class));
1003+
assertEquals("{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}", record3.getData().get("message", String.class));
1004+
1005+
// Verify metrics
1006+
final Measurement requestReceivedCount = MetricsTestUtil.getMeasurementFromList(
1007+
requestsReceivedMeasurements, Statistic.COUNT);
1008+
assertEquals(1.0, requestReceivedCount.getValue());
1009+
final Measurement successRequestsCount = MetricsTestUtil.getMeasurementFromList(
1010+
successRequestsMeasurements, Statistic.COUNT);
1011+
assertEquals(1.0, successRequestsCount.getValue());
1012+
final Measurement requestProcessDurationCount = MetricsTestUtil.getMeasurementFromList(
1013+
requestProcessDurationMeasurements, Statistic.COUNT);
1014+
assertEquals(1.0, requestProcessDurationCount.getValue());
1015+
final Measurement requestProcessDurationMax = MetricsTestUtil.getMeasurementFromList(
1016+
requestProcessDurationMeasurements, Statistic.MAX);
1017+
assertTrue(requestProcessDurationMax.getValue() > 0);
1018+
final Measurement payloadSizeMax = MetricsTestUtil.getMeasurementFromList(
1019+
payloadSizeSummaryMeasurements, Statistic.MAX);
1020+
assertEquals(testPayloadSize, payloadSizeMax.getValue());
1021+
}
1022+
1023+
private void assertCommonFields(Record<Log> record) {
1024+
assertEquals("111111111111", record.getData().get("owner", String.class));
1025+
assertEquals("CloudTrail/logs", record.getData().get("logGroup", String.class));
1026+
assertEquals("111111111111_CloudTrail/logs_us-east-1", record.getData().get("logStream", String.class));
1027+
}
9381028
}

‎data-prepper-plugins/http-source/src/test/resources/cloudwatch-logs-sample.json‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
{
2+
"owner": "111111111111",
3+
"logGroup": "CloudTrail/logs",
4+
"logStream": "111111111111_CloudTrail/logs_us-east-1",
5+
"subscriptionFilters": [
6+
"Destination"
7+
],
8+
"messageType": "DATA_MESSAGE",
9+
"logEvents": [
10+
{
11+
"id": "31953106606966983378809025079804211143289615424298221568",
12+
"timestamp": 1432826855000,
13+
"message": "{\"eventVersion\":\"1.01\",\"userIdentity\":{\"type\":\"Root\"}"
14+
},
15+
{
16+
"id": "31953106606966983378809025079804211143289615424298221569",
17+
"timestamp": 1432826855001,
18+
"message": "{\"eventVersion\":\"1.02\",\"userIdentity\":{\"type\":\"Root\"}"
19+
},
20+
{
21+
"id": "31953106606966983378809025079804211143289615424298221570",
22+
"timestamp": 1432826855002,
23+
"message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
24+
}
25+
]
26+
}

0 commit comments

Comments
 (0)