Skip to content

Commit 1173763

Browse files
oeyhoeyh
authored
Add codec to http source (#5694)
* Add codec to http source Signed-off-by: Hai Yan <oeyh@amazon.com> * Add a test with sample data Signed-off-by: Hai Yan <oeyh@amazon.com> --------- Signed-off-by: Hai Yan <oeyh@amazon.com>
1 parent 5d04c25 commit 1173763

8 files changed

Lines changed: 222 additions & 43 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/jmh/java/org/opensearch/dataprepper/plugins/source/loghttp/LogHTTPServiceMeasure.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ public void setUp() throws IOException {
5151
when(buffer.getOptimalRequestSize()).thenReturn(Optional.of(256 * 1024));
5252

5353
serviceRequestContext = mock(ServiceRequestContext.class);
54-
logHTTPService = new LogHTTPService((int) Duration.ofSeconds(10).toMillis(), buffer, PluginMetrics.fromPrefix("testing"));
54+
logHTTPService = new LogHTTPService((int) Duration.ofSeconds(10).toMillis(), buffer, PluginMetrics.fromPrefix("testing"), null);
5555

5656
requestHeaders = RequestHeaders.builder()
5757
.method(HttpMethod.POST)

‎data-prepper-plugins/http-source/src/main/java/org/opensearch/dataprepper/plugins/source/loghttp/HTTPSource.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
1616
import org.opensearch.dataprepper.model.buffer.Buffer;
1717
import org.opensearch.dataprepper.model.codec.ByteDecoder;
18+
import org.opensearch.dataprepper.model.codec.InputCodec;
1819
import org.opensearch.dataprepper.model.codec.JsonDecoder;
1920
import org.opensearch.dataprepper.model.configuration.PipelineDescription;
2021
import org.opensearch.dataprepper.model.configuration.PluginModel;
@@ -48,6 +49,7 @@ public class HTTPSource implements Source<Record<Log>> {
4849
private final PluginMetrics pluginMetrics;
4950
private static final String HTTP_HEALTH_CHECK_PATH = "/health";
5051
private ByteDecoder byteDecoder;
52+
private final InputCodec codec;
5153

5254
@DataPrepperPluginConstructor
5355
public HTTPSource(final HTTPSourceConfig sourceConfig, final PluginMetrics pluginMetrics, final PluginFactory pluginFactory,
@@ -75,6 +77,13 @@ public HTTPSource(final HTTPSourceConfig sourceConfig, final PluginMetrics plugi
7577
authenticationPluginSetting.setPipelineName(pipelineName);
7678
authenticationProvider = pluginFactory.loadPlugin(ArmeriaHttpAuthenticationProvider.class, authenticationPluginSetting);
7779
httpRequestExceptionHandler = new HttpRequestExceptionHandler(pluginMetrics);
80+
final PluginModel codecConfiguration = sourceConfig.getCodec();
81+
if (codecConfiguration == null) {
82+
codec = null;
83+
} else {
84+
final PluginSetting codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(), codecConfiguration.getPluginSettings());
85+
codec = pluginFactory.loadPlugin(InputCodec.class, codecPluginSettings);
86+
}
7887
}
7988

8089
@Override
@@ -85,7 +94,7 @@ public void start(final Buffer<Record<Log>> buffer) {
8594
if (server == null) {
8695
ServerConfiguration serverConfiguration = ConvertConfiguration.convertConfiguration(sourceConfig);
8796
CreateServer createServer = new CreateServer(serverConfiguration, LOG, pluginMetrics, PLUGIN_NAME, pipelineName);
88-
final LogHTTPService logHTTPService = new LogHTTPService(serverConfiguration.getBufferTimeoutInMillis(), buffer, pluginMetrics);
97+
final LogHTTPService logHTTPService = new LogHTTPService(serverConfiguration.getBufferTimeoutInMillis(), buffer, pluginMetrics, codec);
8998
server = createServer.createHTTPServer(buffer, certificateProviderFactory, authenticationProvider, httpRequestExceptionHandler, logHTTPService);
9099
pluginMetrics.gauge(SERVER_CONNECTIONS, server, Server::numConnections);
91100
}

‎data-prepper-plugins/http-source/src/main/java/org/opensearch/dataprepper/plugins/source/loghttp/HTTPSourceConfig.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,9 @@
55

66
package org.opensearch.dataprepper.plugins.source.loghttp;
77

8+
import com.fasterxml.jackson.annotation.JsonProperty;
89
import org.opensearch.dataprepper.http.BaseHttpServerConfig;
10+
import org.opensearch.dataprepper.model.configuration.PluginModel;
911

1012
public class HTTPSourceConfig extends BaseHttpServerConfig {
1113

@@ -21,4 +23,11 @@ public int getDefaultPort() {
2123
public String getDefaultPath() {
2224
return DEFAULT_LOG_INGEST_URI;
2325
}
26+
27+
@JsonProperty("codec")
28+
private PluginModel codec;
29+
30+
public PluginModel getCodec() {
31+
return codec;
32+
}
2433
}

‎data-prepper-plugins/http-source/src/main/java/org/opensearch/dataprepper/plugins/source/loghttp/LogHTTPService.java‎

Lines changed: 36 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -5,26 +5,28 @@
55

66
package org.opensearch.dataprepper.plugins.source.loghttp;
77

8-
import com.linecorp.armeria.server.ServiceRequestContext;
9-
import org.opensearch.dataprepper.metrics.PluginMetrics;
10-
import org.opensearch.dataprepper.model.buffer.Buffer;
11-
import org.opensearch.dataprepper.model.log.JacksonLog;
12-
import org.opensearch.dataprepper.model.log.Log;
13-
import org.opensearch.dataprepper.model.record.Record;
148
import com.linecorp.armeria.common.AggregatedHttpRequest;
159
import com.linecorp.armeria.common.HttpData;
1610
import com.linecorp.armeria.common.HttpResponse;
1711
import com.linecorp.armeria.common.HttpStatus;
12+
import com.linecorp.armeria.server.ServiceRequestContext;
1813
import com.linecorp.armeria.server.annotation.Blocking;
1914
import com.linecorp.armeria.server.annotation.Post;
2015
import io.micrometer.core.instrument.Counter;
2116
import io.micrometer.core.instrument.DistributionSummary;
2217
import io.micrometer.core.instrument.Timer;
2318
import org.opensearch.dataprepper.http.codec.JsonCodec;
19+
import org.opensearch.dataprepper.metrics.PluginMetrics;
20+
import org.opensearch.dataprepper.model.buffer.Buffer;
21+
import org.opensearch.dataprepper.model.codec.InputCodec;
22+
import org.opensearch.dataprepper.model.log.JacksonLog;
23+
import org.opensearch.dataprepper.model.log.Log;
24+
import org.opensearch.dataprepper.model.record.Record;
2425
import org.slf4j.Logger;
2526
import org.slf4j.LoggerFactory;
2627

2728
import java.io.IOException;
29+
import java.util.ArrayList;
2830
import java.util.List;
2931
import java.util.UUID;
3032
import java.util.stream.Collectors;
@@ -48,6 +50,7 @@ public class LogHTTPService {
4850
// TODO: support other data-types as request body, e.g. json_lines, msgpack
4951
private final JsonCodec jsonCodec = new JsonCodec();
5052
private final Buffer<Record<Log>> buffer;
53+
private final InputCodec codec;
5154
private final int bufferWriteTimeoutInMillis;
5255
private final Counter requestsReceivedCounter;
5356
private final Counter successRequestsCounter;
@@ -60,11 +63,13 @@ public class LogHTTPService {
6063

6164
public LogHTTPService(final int bufferWriteTimeoutInMillis,
6265
final Buffer<Record<Log>> buffer,
63-
final PluginMetrics pluginMetrics) {
66+
final PluginMetrics pluginMetrics,
67+
final InputCodec codec) {
6468
this.buffer = buffer;
6569
this.bufferWriteTimeoutInMillis = bufferWriteTimeoutInMillis;
6670
this.bufferMaxRequestLength = buffer.getMaxRequestSize().isPresent() ? buffer.getMaxRequestSize().get(): null;
6771
this.bufferOptimalRequestLength = buffer.getOptimalRequestSize().isPresent() ? buffer.getOptimalRequestSize().get(): null;
72+
this.codec = codec;
6873
requestsReceivedCounter = pluginMetrics.counter(REQUESTS_RECEIVED);
6974
successRequestsCounter = pluginMetrics.counter(SUCCESS_REQUESTS);
7075
requestsOverOptimalSizeCounter = pluginMetrics.counter(REQUESTS_OVER_OPTIMAL_SIZE);
@@ -108,17 +113,32 @@ HttpResponse processRequest(final AggregatedHttpRequest aggregatedHttpRequest) t
108113
}
109114
} else {
110115
final List<String> jsonList;
116+
final List<Record<Log>> records = new ArrayList<>();
111117

112-
try {
113-
jsonList = jsonCodec.parse(content);
114-
} catch (IOException e) {
115-
LOG.error("Failed to parse the request of size {} due to: {}", content.length(), e.getMessage());
116-
throw new IOException("Bad request data format. Needs to be json array.", e.getCause());
117-
}
118+
if (codec != null) {
119+
try {
120+
codec.parse(content.toInputStream(), record -> {
121+
records.add(new Record<>((Log) record.getData()));
122+
});
123+
} catch (IOException e) {
124+
LOG.error("Failed to parse the request of size {} using specified input codec {} due to: {}", content.length(), codec.getClass(), e.getMessage());
125+
throw new IOException("Bad request data format. ", e.getCause());
126+
}
127+
} else {
128+
129+
try {
130+
jsonList = jsonCodec.parse(content);
131+
} catch (IOException e) {
132+
LOG.error("Failed to parse the request of size {} due to: {}", content.length(), e.getMessage());
133+
throw new IOException("Bad request data format. Needs to be json array.", e.getCause());
134+
}
118135

119-
final List<Record<Log>> records = jsonList.stream()
120-
.map(this::buildRecordLog)
121-
.collect(Collectors.toList());
136+
records.addAll(
137+
jsonList.stream()
138+
.map(this::buildRecordLog)
139+
.collect(Collectors.toList())
140+
);
141+
}
122142

123143
try {
124144
buffer.writeAll(records, bufferWriteTimeoutInMillis);

‎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
}

0 commit comments

Comments
 (0)