|
| 1 | +/* |
| 2 | + * Copyright OpenSearch Contributors |
| 3 | + * SPDX-License-Identifier: Apache-2.0 |
| 4 | + */ |
| 5 | + |
| 6 | +package org.opensearch.dataprepper.integration.log; |
| 7 | + |
| 8 | +import static org.awaitility.Awaitility.await; |
| 9 | +import static org.hamcrest.CoreMatchers.is; |
| 10 | +import static org.hamcrest.MatcherAssert.assertThat; |
| 11 | +import static org.hamcrest.Matchers.contains; |
| 12 | +import static org.hamcrest.Matchers.not; |
| 13 | +import static org.junit.jupiter.api.Assertions.assertEquals; |
| 14 | + |
| 15 | +import java.io.IOException; |
| 16 | +import java.util.ArrayList; |
| 17 | +import java.util.Collections; |
| 18 | +import java.util.List; |
| 19 | +import java.util.Map; |
| 20 | +import java.util.concurrent.TimeUnit; |
| 21 | +import java.util.stream.Collectors; |
| 22 | + |
| 23 | +import org.junit.Test; |
| 24 | +import org.opensearch.action.admin.indices.refresh.RefreshRequest; |
| 25 | +import org.opensearch.action.search.SearchRequest; |
| 26 | +import org.opensearch.action.search.SearchResponse; |
| 27 | +import org.opensearch.client.RequestOptions; |
| 28 | +import org.opensearch.client.RestHighLevelClient; |
| 29 | +import org.opensearch.dataprepper.plugins.sink.opensearch.ConnectionConfiguration; |
| 30 | +import org.opensearch.dataprepper.plugins.source.loggenerator.ApacheLogFaker; |
| 31 | +import org.opensearch.search.SearchHits; |
| 32 | +import org.opensearch.search.builder.SearchSourceBuilder; |
| 33 | + |
| 34 | +import com.google.protobuf.ByteString; |
| 35 | +import com.google.protobuf.InvalidProtocolBufferException; |
| 36 | +import com.google.protobuf.util.JsonFormat; |
| 37 | +import com.linecorp.armeria.client.WebClient; |
| 38 | +import com.linecorp.armeria.common.HttpData; |
| 39 | +import com.linecorp.armeria.common.HttpMethod; |
| 40 | +import com.linecorp.armeria.common.HttpStatus; |
| 41 | +import com.linecorp.armeria.common.MediaType; |
| 42 | +import com.linecorp.armeria.common.RequestHeaders; |
| 43 | +import com.linecorp.armeria.common.SessionProtocol; |
| 44 | + |
| 45 | +import io.netty.util.AsciiString; |
| 46 | +import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest; |
| 47 | +import io.opentelemetry.proto.common.v1.AnyValue; |
| 48 | +import io.opentelemetry.proto.logs.v1.LogRecord; |
| 49 | +import io.opentelemetry.proto.logs.v1.ResourceLogs; |
| 50 | +import io.opentelemetry.proto.logs.v1.ScopeLogs; |
| 51 | + |
| 52 | +public class EndToEndOtelLogsSourceTest { |
| 53 | + private static final int SOURCE_PORT = 2021; |
| 54 | + private static final String INDEX_NAME = "otel-logs-index"; |
| 55 | + |
| 56 | + private final ApacheLogFaker apacheLogFaker = new ApacheLogFaker(); |
| 57 | + |
| 58 | + @Test |
| 59 | + public void testOtelLogsSourcePipelineEndToEnd() throws InvalidProtocolBufferException { |
| 60 | + sendHttpRequestToSourceAndAssertResponse(SOURCE_PORT, createOtelLogsRequest()); |
| 61 | + final RestHighLevelClient restHighLevelClient = prepareOpenSearchRestHighLevelClient(); |
| 62 | + final List<Map<String, Object>> retrievedDocs = new ArrayList<>(); |
| 63 | + |
| 64 | + makeRequestAndAssertResponse(restHighLevelClient, retrievedDocs); |
| 65 | + } |
| 66 | + |
| 67 | + private HttpData createOtelLogsRequest() throws InvalidProtocolBufferException { |
| 68 | + ExportLogsServiceRequest exportLogsServiceRequest = ExportLogsServiceRequest.newBuilder().addResourceLogs( |
| 69 | + ResourceLogs.newBuilder() |
| 70 | + .addScopeLogs(ScopeLogs.newBuilder() |
| 71 | + .addLogRecords(LogRecord.newBuilder() |
| 72 | + .setBody(AnyValue.newBuilder().setStringValue(apacheLogFaker.generateRandomCommonApacheLog()).build()) |
| 73 | + .setSeverityNumberValue(1) |
| 74 | + .setTimeUnixNano(System.currentTimeMillis() * 1_000_000) |
| 75 | + .setTraceId(ByteString.copyFromUtf8("trace-id")) |
| 76 | + .setSpanId(ByteString.copyFromUtf8("span-id") |
| 77 | + )).build() |
| 78 | + ) |
| 79 | + ).build(); |
| 80 | + |
| 81 | + return HttpData.copyOf(JsonFormat.printer().print(exportLogsServiceRequest).getBytes()); |
| 82 | + } |
| 83 | + |
| 84 | + private void makeRequestAndAssertResponse(RestHighLevelClient restHighLevelClient, List<Map<String, Object>> retrievedDocs) { |
| 85 | + await().atMost(10, TimeUnit.SECONDS).untilAsserted( |
| 86 | + () -> { |
| 87 | + refreshIndices(restHighLevelClient); |
| 88 | + final SearchRequest searchRequest = new SearchRequest(INDEX_NAME); |
| 89 | + searchRequest.source( |
| 90 | + SearchSourceBuilder.searchSource().size(100) |
| 91 | + ); |
| 92 | + final SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); |
| 93 | + final List<Map<String, Object>> foundSources = getSourcesFromSearchHits(searchResponse.getHits()); |
| 94 | + assertEquals(1, foundSources.size()); |
| 95 | + retrievedDocs.addAll(foundSources); |
| 96 | + } |
| 97 | + ); |
| 98 | + } |
| 99 | + |
| 100 | + private RestHighLevelClient prepareOpenSearchRestHighLevelClient() { |
| 101 | + final ConnectionConfiguration.Builder builder = new ConnectionConfiguration.Builder( |
| 102 | + Collections.singletonList("https://127.0.0.1:9200")); |
| 103 | + builder.withUsername("admin"); |
| 104 | + builder.withPassword("admin"); |
| 105 | + builder.withInsecure(true); |
| 106 | + return builder.build().createClient(null); |
| 107 | + } |
| 108 | + |
| 109 | + private void sendHttpRequestToSourceAndAssertResponse(final int port, final HttpData httpData) { |
| 110 | + WebClient.of().execute(RequestHeaders.builder() |
| 111 | + .scheme(SessionProtocol.HTTP) |
| 112 | + .authority(String.format("127.0.0.1:%d", port)) |
| 113 | + .method(HttpMethod.POST) |
| 114 | + .path("/otel-logs-pipeline/logs") |
| 115 | + .contentType(MediaType.JSON_UTF_8) |
| 116 | + .build(), |
| 117 | + httpData) |
| 118 | + .aggregate() |
| 119 | + .whenComplete((i, ex) -> { |
| 120 | + assertThat(i.status(), is(HttpStatus.OK)); |
| 121 | + final List<String> headerKeys = i.headers() |
| 122 | + .stream() |
| 123 | + .map(Map.Entry::getKey) |
| 124 | + .map(AsciiString::toString) |
| 125 | + .collect(Collectors.toList()); |
| 126 | + assertThat("Response Header Keys", headerKeys, not(contains("server"))); |
| 127 | + }).join(); |
| 128 | + } |
| 129 | + |
| 130 | + private List<Map<String, Object>> getSourcesFromSearchHits(final SearchHits searchHits) { |
| 131 | + final List<Map<String, Object>> sources = new ArrayList<>(); |
| 132 | + searchHits.forEach(hit -> { |
| 133 | + Map<String, Object> source = hit.getSourceAsMap(); |
| 134 | + sources.add(source); |
| 135 | + }); |
| 136 | + return sources; |
| 137 | + } |
| 138 | + |
| 139 | + private void refreshIndices(final RestHighLevelClient restHighLevelClient) throws IOException { |
| 140 | + final RefreshRequest requestAll = new RefreshRequest(); |
| 141 | + restHighLevelClient.indices().refresh(requestAll, RequestOptions.DEFAULT); |
| 142 | + } |
| 143 | +} |
0 commit comments