Skip to content

Commit b50c01e

Browse files
divbokdivbok
committed
Passing http request headers as metadata in the event for http source
Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com>
1 parent 2f8c836 commit b50c01e

8 files changed

Lines changed: 443 additions & 14 deletions

File tree

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.slf4j.LoggerFactory;
3131

3232
import java.util.Collections;
33+
import java.util.List;
3334
import java.util.concurrent.ExecutionException;
3435

3536
@DataPrepperPlugin(name = "http", pluginType = Source.class, pluginConfigurationType = HTTPSourceConfig.class)
@@ -50,6 +51,8 @@ public class HTTPSource implements Source<Record<Log>> {
5051
private static final String HTTP_HEALTH_CHECK_PATH = "/health";
5152
private ByteDecoder byteDecoder;
5253
private final InputCodec codec;
54+
private final List<String> metadataHeaders;
55+
private final HttpHeaderExtractor httpHeaderExtractor;
5356

5457
@DataPrepperPluginConstructor
5558
public HTTPSource(final HTTPSourceConfig sourceConfig, final PluginMetrics pluginMetrics, final PluginFactory pluginFactory,
@@ -59,6 +62,7 @@ public HTTPSource(final HTTPSourceConfig sourceConfig, final PluginMetrics plugi
5962
this.pipelineName = pipelineDescription.getPipelineName();
6063
this.byteDecoder = new JsonDecoder();
6164
this.certificateProviderFactory = new CertificateProviderFactory(sourceConfig);
65+
this.metadataHeaders = sourceConfig.getMetadataHeaders();
6266
final PluginModel authenticationConfiguration = sourceConfig.getAuthentication();
6367
final PluginSetting authenticationPluginSetting;
6468

@@ -84,6 +88,7 @@ public HTTPSource(final HTTPSourceConfig sourceConfig, final PluginMetrics plugi
8488
final PluginSetting codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(), codecConfiguration.getPluginSettings());
8589
codec = pluginFactory.loadPlugin(InputCodec.class, codecPluginSettings);
8690
}
91+
httpHeaderExtractor = new HttpHeaderExtractor(metadataHeaders);
8792
}
8893

8994
@Override
@@ -94,7 +99,7 @@ public void start(final Buffer<Record<Log>> buffer) {
9499
if (server == null) {
95100
ServerConfiguration serverConfiguration = ConvertConfiguration.convertConfiguration(sourceConfig);
96101
CreateServer createServer = new CreateServer(serverConfiguration, LOG, pluginMetrics, PLUGIN_NAME, pipelineName);
97-
final LogHTTPService logHTTPService = new LogHTTPService(serverConfiguration.getBufferTimeoutInMillis(), buffer, pluginMetrics, codec);
102+
final LogHTTPService logHTTPService = new LogHTTPService(serverConfiguration.getBufferTimeoutInMillis(), buffer, pluginMetrics, codec, httpHeaderExtractor);
98103
server = createServer.createHTTPServer(buffer, certificateProviderFactory, authenticationProvider, httpRequestExceptionHandler, logHTTPService);
99104
pluginMetrics.gauge(SERVER_CONNECTIONS, server, Server::numConnections);
100105
}

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99
import org.opensearch.dataprepper.http.BaseHttpServerConfig;
1010
import org.opensearch.dataprepper.model.configuration.PluginModel;
1111

12+
import java.util.List;
13+
1214
public class HTTPSourceConfig extends BaseHttpServerConfig {
1315

1416
static final String DEFAULT_LOG_INGEST_URI = "/log/ingest";
@@ -27,7 +29,21 @@ public String getDefaultPath() {
2729
@JsonProperty("codec")
2830
private PluginModel codec;
2931

32+
@JsonProperty("metadata_headers")
33+
private List<String> metadataHeaders;
34+
35+
@JsonProperty("metadata_headers_regex")
36+
private List<String> metadataHeadersRegex;
37+
3038
public PluginModel getCodec() {
3139
return codec;
3240
}
41+
42+
public List<String> getMetadataHeaders() {
43+
return metadataHeaders;
44+
}
45+
46+
public List<String> getMetadataHeadersRegex() {
47+
return metadataHeadersRegex;
48+
}
3349
}

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

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.opensearch.dataprepper.plugins.source.loghttp;
7+
8+
import com.linecorp.armeria.common.AggregatedHttpRequest;
9+
10+
import java.util.ArrayList;
11+
import java.util.Collections;
12+
import java.util.HashMap;
13+
import java.util.LinkedHashSet;
14+
import java.util.List;
15+
import java.util.Map;
16+
import java.util.Set;
17+
import java.util.stream.Collectors;
18+
19+
public class HttpHeaderExtractor {
20+
21+
static final Set<String> SENSITIVE_HEADERS = Set.of(
22+
"authorization",
23+
"proxy-authorization",
24+
"cookie",
25+
"set-cookie",
26+
"www-authenticate",
27+
"proxy-authenticate",
28+
"x-api-key",
29+
"x-csrf-token",
30+
"x-xsrf-token",
31+
"x-auth-token",
32+
"x-amz-security-token",
33+
"x-amz-credential"
34+
);
35+
36+
private final List<String> metadataHeaders;
37+
38+
public HttpHeaderExtractor(final List<String> metadataHeaders) {
39+
this.metadataHeaders = metadataHeaders;
40+
}
41+
42+
public Map<String, Object> extractHeaders(final AggregatedHttpRequest aggregatedHttpRequest) {
43+
if (metadataHeaders == null || metadataHeaders.isEmpty()) {
44+
return Collections.emptyMap();
45+
}
46+
47+
final Set<String> headerNames = metadataHeaders.stream()
48+
.map(String::toLowerCase)
49+
.collect(Collectors.toCollection(LinkedHashSet::new));
50+
51+
final Map<String, Object> headers = new HashMap<>();
52+
for (String headerName : headerNames) {
53+
if (isSensitiveHeader(headerName)) {
54+
continue;
55+
}
56+
List<String> values = aggregatedHttpRequest.headers().getAll(headerName);
57+
if (!values.isEmpty()) {
58+
headers.put(headerName, values.size() == 1 ? values.get(0) : new ArrayList<>(values));
59+
}
60+
}
61+
62+
return headers;
63+
}
64+
65+
static boolean isSensitiveHeader(final String headerName) {
66+
return SENSITIVE_HEADERS.contains(headerName.toLowerCase());
67+
}
68+
}

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

Lines changed: 26 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -28,13 +28,10 @@
2828
import java.io.IOException;
2929
import java.util.ArrayList;
3030
import java.util.List;
31+
import java.util.Map;
3132
import java.util.UUID;
3233
import java.util.stream.Collectors;
3334

34-
35-
/*
36-
* A HTTP service for log ingestion to be executed by BlockingTaskExecutor.
37-
*/
3835
@Blocking
3936
public class LogHTTPService {
4037
private static final int SERIALIZATION_OVERHEAD = 1024;
@@ -60,16 +57,19 @@ public class LogHTTPService {
6057
private final Timer requestProcessDuration;
6158
private Integer bufferMaxRequestLength;
6259
private Integer bufferOptimalRequestLength;
60+
private final HttpHeaderExtractor httpHeaderExtractor;
6361

6462
public LogHTTPService(final int bufferWriteTimeoutInMillis,
6563
final Buffer<Record<Log>> buffer,
6664
final PluginMetrics pluginMetrics,
67-
final InputCodec codec) {
65+
final InputCodec codec,
66+
final HttpHeaderExtractor httpHeaderExtractor) {
6867
this.buffer = buffer;
6968
this.bufferWriteTimeoutInMillis = bufferWriteTimeoutInMillis;
7069
this.bufferMaxRequestLength = buffer.getMaxRequestSize().isPresent() ? buffer.getMaxRequestSize().get(): null;
7170
this.bufferOptimalRequestLength = buffer.getOptimalRequestSize().isPresent() ? buffer.getOptimalRequestSize().get(): null;
7271
this.codec = codec;
72+
this.httpHeaderExtractor = httpHeaderExtractor;
7373
requestsReceivedCounter = pluginMetrics.counter(REQUESTS_RECEIVED);
7474
successRequestsCounter = pluginMetrics.counter(SUCCESS_REQUESTS);
7575
requestsOverOptimalSizeCounter = pluginMetrics.counter(REQUESTS_OVER_OPTIMAL_SIZE);
@@ -78,6 +78,13 @@ public LogHTTPService(final int bufferWriteTimeoutInMillis,
7878
requestProcessDuration = pluginMetrics.timer(REQUEST_PROCESS_DURATION);
7979
}
8080

81+
public LogHTTPService(final int bufferWriteTimeoutInMillis,
82+
final Buffer<Record<Log>> buffer,
83+
final PluginMetrics pluginMetrics,
84+
final InputCodec codec) {
85+
this(bufferWriteTimeoutInMillis, buffer, pluginMetrics, codec, new HttpHeaderExtractor(null));
86+
}
87+
8188
@Post
8289
public HttpResponse doPost(final ServiceRequestContext serviceRequestContext, final AggregatedHttpRequest aggregatedHttpRequest) throws Exception {
8390
requestsReceivedCounter.increment();
@@ -92,6 +99,7 @@ public HttpResponse doPost(final ServiceRequestContext serviceRequestContext, fi
9299

93100
HttpResponse processRequest(final AggregatedHttpRequest aggregatedHttpRequest) throws Exception {
94101
final HttpData content = aggregatedHttpRequest.content();
102+
final Map<String, Object> extractedHeaders = httpHeaderExtractor.extractHeaders(aggregatedHttpRequest);
95103

96104
if (buffer.isByteBuffer()) {
97105
if (bufferMaxRequestLength != null && bufferOptimalRequestLength != null && content.array().length > bufferOptimalRequestLength) {
@@ -124,6 +132,11 @@ HttpResponse processRequest(final AggregatedHttpRequest aggregatedHttpRequest) t
124132
LOG.error("Failed to parse the request of size {} using specified input codec {} due to: {}", content.length(), codec.getClass(), e.getMessage());
125133
throw new IOException("Bad request data format. ", e.getCause());
126134
}
135+
if (!extractedHeaders.isEmpty()) {
136+
for (Record<Log> record : records) {
137+
record.getData().getMetadata().setAttribute("headers", extractedHeaders);
138+
}
139+
}
127140
} else {
128141

129142
try {
@@ -135,7 +148,7 @@ HttpResponse processRequest(final AggregatedHttpRequest aggregatedHttpRequest) t
135148

136149
records.addAll(
137150
jsonList.stream()
138-
.map(this::buildRecordLog)
151+
.map(json -> buildRecordLog(json, extractedHeaders))
139152
.collect(Collectors.toList())
140153
);
141154
}
@@ -171,13 +184,13 @@ private void writeChunkedBody(final String chunk) {
171184
}
172185
}
173186

174-
private Record<Log> buildRecordLog(String json) {
175-
176-
final JacksonLog log = JacksonLog.builder()
187+
private Record<Log> buildRecordLog(final String json, final Map<String, Object> headerAttributes) {
188+
final JacksonLog.Builder builder = JacksonLog.builder()
177189
.withData(json)
178-
.getThis()
179-
.build();
180-
181-
return new Record<>(log);
190+
.getThis();
191+
if (!headerAttributes.isEmpty()) {
192+
builder.withEventMetadataAttributes(Map.of("headers", headerAttributes));
193+
}
194+
return new Record<>(builder.build());
182195
}
183196
}

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,18 @@
55

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

8+
import com.fasterxml.jackson.databind.ObjectMapper;
89
import org.junit.jupiter.api.Test;
910

11+
import java.util.List;
12+
1013
import static org.junit.jupiter.api.Assertions.assertEquals;
14+
import static org.junit.jupiter.api.Assertions.assertNull;
1115

1216
public class HTTPSourceConfigTest {
1317

18+
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
19+
1420
@Test
1521
void testDefault() {
1622
// Prepare
@@ -21,5 +27,14 @@ void testDefault() {
2127
assertEquals(HTTPSourceConfig.DEFAULT_LOG_INGEST_URI, sourceConfig.getPath());
2228
assertEquals(HTTPSourceConfig.DEFAULT_PORT, sourceConfig.getDefaultPort());
2329
assertEquals(HTTPSourceConfig.DEFAULT_LOG_INGEST_URI, sourceConfig.getDefaultPath());
30+
assertNull(sourceConfig.getMetadataHeaders());
31+
}
32+
33+
@Test
34+
void testSetMetadataHeaders() throws Exception {
35+
final String json = "{\"metadata_headers\": [\"X-Tenant-Id\", \"X-Region\"]}";
36+
final HTTPSourceConfig sourceConfig = OBJECT_MAPPER.readValue(json, HTTPSourceConfig.class);
37+
38+
assertEquals(List.of("X-Tenant-Id", "X-Region"), sourceConfig.getMetadataHeaders());
2439
}
2540
}

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

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@
7575
import java.util.Map;
7676
import java.util.Random;
7777
import java.util.StringJoiner;
78+
import java.util.UUID;
7879
import java.util.concurrent.CompletableFuture;
7980
import java.util.concurrent.CompletionException;
8081
import java.util.concurrent.ExecutionException;
@@ -92,6 +93,7 @@
9293
import static org.junit.jupiter.api.Assertions.assertEquals;
9394
import static org.junit.jupiter.api.Assertions.assertFalse;
9495
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
96+
import static org.junit.jupiter.api.Assertions.assertNull;
9597
import static org.junit.jupiter.api.Assertions.assertThrows;
9698
import static org.junit.jupiter.api.Assertions.assertTrue;
9799
import static org.mockito.ArgumentMatchers.any;
@@ -1020,6 +1022,58 @@ public void testHTTPJsonCodec() throws IOException {
10201022
assertEquals(testPayloadSize, payloadSizeMax.getValue());
10211023
}
10221024

1025+
@Test
1026+
public void testHTTPJsonResponse200WithMetadataHeaders() throws JsonProcessingException {
1027+
final String tenantId = UUID.randomUUID().toString();
1028+
final String testData = "[{\"log\": \"somelog\"}]";
1029+
1030+
when(sourceConfig.getMetadataHeaders()).thenReturn(List.of("X-Tenant-Id"));
1031+
HTTPSourceUnderTest = new HTTPSource(sourceConfig, pluginMetrics, pluginFactory, pipelineDescription);
1032+
testBuffer = getBuffer(1, 1);
1033+
HTTPSourceUnderTest.start(testBuffer);
1034+
1035+
WebClient.of().execute(RequestHeaders.builder()
1036+
.scheme(SessionProtocol.HTTP)
1037+
.authority("127.0.0.1:2021")
1038+
.method(HttpMethod.POST)
1039+
.path("/log/ingest")
1040+
.contentType(MediaType.JSON_UTF_8)
1041+
.add("X-Tenant-Id", tenantId)
1042+
.build(),
1043+
HttpData.ofUtf8(testData))
1044+
.aggregate()
1045+
.whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.OK)).join();
1046+
1047+
final Map.Entry<Collection<Record<Log>>, CheckpointState> result = testBuffer.read(100);
1048+
List<Record<Log>> records = new ArrayList<>(result.getKey());
1049+
assertEquals(1, records.size());
1050+
assertEquals(tenantId, records.get(0).getData().getMetadata().getAttribute("headers/x-tenant-id"));
1051+
}
1052+
1053+
@Test
1054+
public void testHTTPJsonResponse200WithNoMetadataHeaders() {
1055+
final String testData = "[{\"log\": \"somelog\"}]";
1056+
1057+
HTTPSourceUnderTest.start(testBuffer);
1058+
1059+
WebClient.of().execute(RequestHeaders.builder()
1060+
.scheme(SessionProtocol.HTTP)
1061+
.authority("127.0.0.1:2021")
1062+
.method(HttpMethod.POST)
1063+
.path("/log/ingest")
1064+
.contentType(MediaType.JSON_UTF_8)
1065+
.add("X-Tenant-Id", UUID.randomUUID().toString())
1066+
.build(),
1067+
HttpData.ofUtf8(testData))
1068+
.aggregate()
1069+
.whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.OK)).join();
1070+
1071+
final Map.Entry<Collection<Record<Log>>, CheckpointState> result = testBuffer.read(100);
1072+
List<Record<Log>> records = new ArrayList<>(result.getKey());
1073+
assertEquals(1, records.size());
1074+
assertNull(records.get(0).getData().getMetadata().getAttribute("headers/x-tenant-id"));
1075+
}
1076+
10231077
private void assertCommonFields(Record<Log> record) {
10241078
assertEquals("111111111111", record.getData().get("owner", String.class));
10251079
assertEquals("CloudTrail/logs", record.getData().get("logGroup", String.class));

0 commit comments

Comments
 (0)