You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@logging.apache.org by vy...@apache.org on 2023/10/18 08:24:28 UTC
[logging-log4j2] 01/06: Fix `LogstashIT` of JTL
This is an automated email from the ASF dual-hosted git repository.
vy pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/logging-log4j2.git
commit 884e010607703dcc3387d57fa53ff7c34771e5b8
Author: Volkan Yazıcı <vo...@yazi.ci>
AuthorDate: Thu Oct 12 14:25:09 2023 +0200
Fix `LogstashIT` of JTL
---
log4j-layout-template-json-test/pom.xml | 16 +-
.../log4j/layout/template/json/LogstashIT.java | 397 ++++++++++-----------
log4j-parent/pom.xml | 8 +-
3 files changed, 212 insertions(+), 209 deletions(-)
diff --git a/log4j-layout-template-json-test/pom.xml b/log4j-layout-template-json-test/pom.xml
index 4ac070b78b..885900129c 100644
--- a/log4j-layout-template-json-test/pom.xml
+++ b/log4j-layout-template-json-test/pom.xml
@@ -81,8 +81,8 @@
</dependency>
<dependency>
- <groupId>org.elasticsearch.client</groupId>
- <artifactId>elasticsearch-rest-high-level-client</artifactId>
+ <groupId>co.elastic.clients</groupId>
+ <artifactId>elasticsearch-java</artifactId>
<scope>test</scope>
</dependency>
@@ -181,6 +181,13 @@
<activeByDefault>false</activeByDefault>
</activation>
+ <properties>
+ <elastic.version>8.10.2</elastic.version>
+ <!-- `docker run elasticsearch:<TAG>` exists with code 137 due to insufficient memory.
+ We limit the used JVM memory to avoid this OOM failure. -->
+ <elastic.java-opts>-Xms750m -Xmx750m</elastic.java-opts>
+ </properties>
+
<build>
<plugins>
@@ -198,6 +205,8 @@
<run>
<env>
<discovery.type>single-node</discovery.type>
+ <xpack.security.enabled>false</xpack.security.enabled>
+ <ES_JAVA_OPTS>${elastic.java-opts}</ES_JAVA_OPTS>
</env>
<ports>
<port>9200:9200</port>
@@ -229,6 +238,9 @@
<name>log4j-layout-template-json-network</name>
<alias>logstash</alias>
</network>
+ <env>
+ <LS_JAVA_OPTS>${elastic.java-opts}</LS_JAVA_OPTS>
+ </env>
<ports>
<port>12222:12222</port>
<port>12345:12345</port>
diff --git a/log4j-layout-template-json-test/src/test/java/org/apache/logging/log4j/layout/template/json/LogstashIT.java b/log4j-layout-template-json-test/src/test/java/org/apache/logging/log4j/layout/template/json/LogstashIT.java
index c27396bf37..25f8393438 100644
--- a/log4j-layout-template-json-test/src/test/java/org/apache/logging/log4j/layout/template/json/LogstashIT.java
+++ b/log4j-layout-template-json-test/src/test/java/org/apache/logging/log4j/layout/template/json/LogstashIT.java
@@ -23,16 +23,23 @@ import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.Instant;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
+import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.elasticsearch._types.ElasticsearchException;
+import co.elastic.clients.elasticsearch._types.HealthStatus;
+import co.elastic.clients.elasticsearch.cluster.HealthResponse;
+import co.elastic.clients.elasticsearch.core.CountResponse;
+import co.elastic.clients.elasticsearch.core.SearchResponse;
+import co.elastic.clients.elasticsearch.core.search.SourceConfig;
+import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
+import co.elastic.clients.elasticsearch.indices.DeleteIndexResponse;
+import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import co.elastic.clients.transport.ElasticsearchTransport;
+import co.elastic.clients.transport.rest_client.RestClientTransport;
import co.elastic.logging.log4j2.EcsLayout;
import org.apache.http.HttpHost;
import org.apache.logging.log4j.Level;
@@ -50,21 +57,10 @@ import org.apache.logging.log4j.message.SimpleMessage;
import org.apache.logging.log4j.status.StatusLogger;
import org.assertj.core.api.Assertions;
import org.awaitility.Awaitility;
-import org.elasticsearch.ElasticsearchStatusException;
-import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
-import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.support.master.AcknowledgedResponse;
-import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.cluster.health.ClusterHealthStatus;
-import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.search.SearchHit;
-import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.Execution;
import org.junit.jupiter.api.parallel.ExecutionMode;
@@ -143,6 +139,12 @@ class LogstashIT {
private static final String ES_INDEX_MESSAGE_FIELD_NAME = "message";
+ private static RestClient REST_CLIENT;
+
+ private static ElasticsearchTransport ES_TRANSPORT;
+
+ private static ElasticsearchClient ES_CLIENT;
+
/**
* Constants hardcoded in docker-maven-plugin configuration, do not change!
*/
@@ -160,6 +162,47 @@ class LogstashIT {
}
+ @BeforeAll
+ public static void initClient() throws IOException {
+
+ LOGGER.info("instantiating the ES client");
+ REST_CLIENT = RestClient
+ .builder(HttpHost.create(String.format("http://%s:%d", HOST_NAME, MavenHardcodedConstants.ES_PORT)))
+ .build();
+ ES_TRANSPORT = new RestClientTransport(REST_CLIENT, new JacksonJsonpMapper());
+ ES_CLIENT = new ElasticsearchClient(ES_TRANSPORT);
+
+ LOGGER.info("verifying the ES connection");
+ HealthResponse healthResponse = ES_CLIENT.cluster().health();
+ Assertions
+ .assertThat(healthResponse.status())
+ .isNotEqualTo(HealthStatus.Red);
+
+ }
+
+ @BeforeEach
+ void deleteIndex() throws IOException {
+ LOGGER.info("deleting the ES index");
+ try {
+ DeleteIndexResponse deleteIndexResponse = ES_CLIENT
+ .indices()
+ .delete(DeleteIndexRequest.of(builder -> builder.index(MavenHardcodedConstants.ES_INDEX_NAME)));
+ Assertions
+ .assertThat(deleteIndexResponse.acknowledged())
+ .isTrue();
+ } catch (ElasticsearchException error) {
+ if (!error.getMessage().contains("index_not_found_exception")) {
+ throw new RuntimeException(error);
+ }
+ }
+ }
+
+ @AfterAll
+ public static void stopClient() throws Exception {
+ ES_TRANSPORT.close();
+ REST_CLIENT.close();
+ }
+
@Test
void test_lite_events() throws IOException {
final List<LogEvent> logEvents =
@@ -175,41 +218,39 @@ class LogstashIT {
}
private static void testEvents(final List<LogEvent> logEvents) throws IOException {
- try (final RestHighLevelClient client = createClient()) {
- final Appender appender = createStartedAppender(
- JSON_TEMPLATE_GELF_LAYOUT,
- MavenHardcodedConstants.LS_GELF_INPUT_PORT);
- try {
-
- // Append events.
- LOGGER.info("appending events");
- logEvents.forEach(appender::append);
- LOGGER.info("completed appending events");
-
- // Wait all messages to arrive.
- Awaitility
- .await()
- .atMost(Duration.ofSeconds(60))
- .pollDelay(Duration.ofSeconds(2))
- .until(() -> queryDocumentCount(client) == LOG_EVENT_COUNT);
-
- // Verify indexed messages.
- final Set<String> expectedMessages = logEvents
- .stream()
- .map(LogstashIT::expectedLogstashMessageField)
- .collect(Collectors.toSet());
- final Set<String> actualMessages = queryDocuments(client)
- .stream()
- .map(source -> (String) source.get(ES_INDEX_MESSAGE_FIELD_NAME))
- .filter(Objects::nonNull)
- .collect(Collectors.toSet());
- Assertions
- .assertThat(actualMessages)
- .isEqualTo(expectedMessages);
-
- } finally {
- appender.stop();
- }
+ final Appender appender = createStartedAppender(
+ JSON_TEMPLATE_GELF_LAYOUT,
+ MavenHardcodedConstants.LS_GELF_INPUT_PORT);
+ try {
+
+ // Append events.
+ LOGGER.info("appending events");
+ logEvents.forEach(appender::append);
+ LOGGER.info("completed appending events");
+
+ // Wait all messages to arrive.
+ Awaitility
+ .await()
+ .atMost(Duration.ofSeconds(60))
+ .pollDelay(Duration.ofSeconds(2))
+ .until(() -> checkDocumentCount(LOG_EVENT_COUNT));
+
+ // Verify indexed messages.
+ final Set<String> expectedMessages = logEvents
+ .stream()
+ .map(LogstashIT::expectedLogstashMessageField)
+ .collect(Collectors.toSet());
+ final Set<String> actualMessages = queryDocuments()
+ .stream()
+ .map(source -> (String) source.get(ES_INDEX_MESSAGE_FIELD_NAME))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toSet());
+ Assertions
+ .assertThat(actualMessages)
+ .isEqualTo(expectedMessages);
+
+ } finally {
+ appender.stop();
}
}
@@ -258,42 +299,40 @@ class LogstashIT {
.setTimeMillis(instantMillis2)
.build();
- try (final RestHighLevelClient client = createClient()) {
- final Appender appender = createStartedAppender(
- JSON_TEMPLATE_GELF_LAYOUT,
- MavenHardcodedConstants.LS_GELF_INPUT_PORT);
- try {
-
- // Append the event.
- LOGGER.info("appending events");
- appender.append(logEvent1);
- appender.append(logEvent2);
- LOGGER.info("completed appending events");
-
- // Wait the message to arrive.
- Awaitility
- .await()
- .atMost(Duration.ofSeconds(60))
- .pollDelay(Duration.ofSeconds(2))
- .until(() -> queryDocumentCount(client) == 2);
-
- // Verify indexed messages.
- final Set<String> expectedMessages = Stream
- .of(logEvent1, logEvent2)
- .map(LogstashIT::expectedLogstashMessageField)
- .collect(Collectors.toSet());
- final Set<String> actualMessages = queryDocuments(client)
- .stream()
- .map(source -> (String) source.get(ES_INDEX_MESSAGE_FIELD_NAME))
- .filter(Objects::nonNull)
- .collect(Collectors.toSet());
- Assertions
- .assertThat(actualMessages)
- .isEqualTo(expectedMessages);
-
- } finally {
- appender.stop();
- }
+ final Appender appender = createStartedAppender(
+ JSON_TEMPLATE_GELF_LAYOUT,
+ MavenHardcodedConstants.LS_GELF_INPUT_PORT);
+ try {
+
+ // Append the event.
+ LOGGER.info("appending events");
+ appender.append(logEvent1);
+ appender.append(logEvent2);
+ LOGGER.info("completed appending events");
+
+ // Wait the message to arrive.
+ Awaitility
+ .await()
+ .atMost(Duration.ofSeconds(60))
+ .pollDelay(Duration.ofSeconds(2))
+ .until(() -> checkDocumentCount(2));
+
+ // Verify indexed messages.
+ final Set<String> expectedMessages = Stream
+ .of(logEvent1, logEvent2)
+ .map(LogstashIT::expectedLogstashMessageField)
+ .collect(Collectors.toSet());
+ final Set<String> actualMessages = queryDocuments()
+ .stream()
+ .map(source -> (String) source.get(ES_INDEX_MESSAGE_FIELD_NAME))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toSet());
+ Assertions
+ .assertThat(actualMessages)
+ .isEqualTo(expectedMessages);
+
+ } finally {
+ appender.stop();
}
}
@@ -305,7 +344,7 @@ class LogstashIT {
final List<LogEvent> logEvents =
LogEventFixture.createFullLogEvents(LOG_EVENT_COUNT);
- // Append log events and collect persisted sources.
+ // Create a function to uniquely identify each document
final Function<Map<String, Object>, Integer> keyMapper =
(final Map<String, Object> source) -> {
final String timestamp = (String) source.get("timestamp");
@@ -313,6 +352,8 @@ class LogstashIT {
final String fullMessage = (String) source.get("full_message");
return Objects.hash(timestamp, shortMessage, fullMessage);
};
+
+ // Collect documents created by `GelfLayout`
final Map<Integer, Object> expectedSourceByKey =
appendAndCollect(
logEvents,
@@ -320,6 +361,11 @@ class LogstashIT {
MavenHardcodedConstants.LS_GELF_INPUT_PORT,
keyMapper,
Collections.emptySet());
+
+ // Reset the index
+ deleteIndex();
+
+ // Collect documents created by `JsonTemplateLayout`
final Map<Integer, Object> actualSourceByKey =
appendAndCollect(
logEvents,
@@ -340,7 +386,7 @@ class LogstashIT {
final List<LogEvent> logEvents =
LogEventFixture.createFullLogEvents(LOG_EVENT_COUNT);
- // Append log events and collect persisted sources.
+ // Create a function to uniquely identify each document
final Function<Map<String, Object>, Integer> keyMapper =
(final Map<String, Object> source) -> {
final String timestamp = (String) source.get("@timestamp");
@@ -348,6 +394,8 @@ class LogstashIT {
final String errorMessage = (String) source.get("error.message");
return Objects.hash(timestamp, message, errorMessage);
};
+
+ // Collect documents created by `EcsLayout`
final Set<String> excludedKeys = Collections.singleton("port");
final Map<Integer, Object> expectedSourceByKey =
appendAndCollect(
@@ -356,6 +404,11 @@ class LogstashIT {
MavenHardcodedConstants.LS_TCP_INPUT_PORT,
keyMapper,
excludedKeys);
+
+ // Reset the index
+ deleteIndex();
+
+ // Collect documents created by `JsonTemplateLayout`
final Map<Integer, Object> actualSourceByKey =
appendAndCollect(
logEvents,
@@ -375,76 +428,34 @@ class LogstashIT {
final int port,
final Function<Map<String, Object>, K> keyMapper,
final Set<String> excludedKeys) throws IOException {
- try (final RestHighLevelClient client = createClient()) {
- final Appender appender = createStartedAppender(layout, port);
- try {
-
- // Append the event.
- LOGGER.info("appending events");
- logEvents.forEach(appender::append);
- LOGGER.info("completed appending events");
-
- // Wait the message to arrive.
- Awaitility
- .await()
- .atMost(Duration.ofSeconds(60))
- .pollDelay(Duration.ofSeconds(2))
- .until(() -> queryDocumentCount(client) == LOG_EVENT_COUNT);
-
- // Retrieve the persisted messages.
- return queryDocuments(client)
- .stream()
- .collect(Collectors.toMap(
- keyMapper,
- (final Map<String, Object> source) -> {
- excludedKeys.forEach(source::remove);
- return source;
- }));
-
- } finally {
- appender.stop();
- }
- }
- }
-
- private static RestHighLevelClient createClient() throws IOException {
-
- // Instantiate the client.
- LOGGER.info("instantiating the ES client");
- final HttpHost httpHost = new HttpHost(HOST_NAME, MavenHardcodedConstants.ES_PORT);
- final RestClientBuilder clientBuilder =
- RestClient.builder(httpHost);
- final RestHighLevelClient client = new RestHighLevelClient(clientBuilder);
-
- // Verify the connection.
- LOGGER.info("verifying the ES connection");
- final ClusterHealthResponse healthResponse = client
- .cluster()
- .health(new ClusterHealthRequest(), RequestOptions.DEFAULT);
- Assertions
- .assertThat(healthResponse.getStatus())
- .isNotEqualTo(ClusterHealthStatus.RED);
-
- // Delete the index.
- LOGGER.info("deleting the ES index");
- final DeleteIndexRequest deleteRequest =
- new DeleteIndexRequest(MavenHardcodedConstants.ES_INDEX_NAME);
+ final Appender appender = createStartedAppender(layout, port);
try {
- final AcknowledgedResponse deleteResponse = client
- .indices()
- .delete(deleteRequest, RequestOptions.DEFAULT);
- Assertions
- .assertThat(deleteResponse.isAcknowledged())
- .isTrue();
- } catch (ElasticsearchStatusException error) {
- Assertions.assertThat(error)
- .satisfies(ignored -> Assertions
- .assertThat(error.status())
- .isEqualTo(RestStatus.NOT_FOUND));
- }
-
- return client;
+ // Append the event.
+ LOGGER.info("appending events");
+ logEvents.forEach(appender::append);
+ LOGGER.info("completed appending events");
+
+ // Wait the message to arrive.
+ Awaitility
+ .await()
+ .atMost(Duration.ofSeconds(60))
+ .pollDelay(Duration.ofSeconds(2))
+ .until(() -> checkDocumentCount(LOG_EVENT_COUNT));
+
+ // Retrieve the persisted messages.
+ return queryDocuments()
+ .stream()
+ .collect(Collectors.toMap(
+ keyMapper,
+ (final Map<String, Object> source) -> {
+ excludedKeys.forEach(source::remove);
+ return source;
+ }));
+
+ } finally {
+ appender.stop();
+ }
}
private static SocketAppender createStartedAppender(
@@ -467,51 +478,31 @@ class LogstashIT {
return appender;
}
- private static long queryDocumentCount(
- final RestHighLevelClient client)
- throws IOException {
- final SearchSourceBuilder searchSourceBuilder =
- new SearchSourceBuilder()
- .size(0)
- .fetchSource(false);
- final SearchRequest searchRequest =
- new SearchRequest(MavenHardcodedConstants.ES_INDEX_NAME)
- .source(searchSourceBuilder);
- try {
- final SearchResponse searchResponse =
- client.search(searchRequest, RequestOptions.DEFAULT);
- return searchResponse.getHits().getTotalHits().value;
- } catch (ElasticsearchStatusException error) {
- if (RestStatus.NOT_FOUND.equals(error.status())) {
- return 0L;
- }
- throw new IOException(error);
- }
+ private static boolean checkDocumentCount(int expectedCount) throws IOException {
+ final CountResponse countResponse = ES_CLIENT.count(builder -> builder.index(MavenHardcodedConstants.ES_INDEX_NAME));
+ final long actualCount = countResponse.count();
+ Assertions.assertThat(actualCount).isLessThanOrEqualTo(expectedCount);
+ return actualCount == expectedCount;
}
- private static List<Map<String, Object>> queryDocuments(
- final RestHighLevelClient client
- ) throws IOException {
- final SearchSourceBuilder searchSourceBuilder =
- new SearchSourceBuilder()
+ private static List<Map<String, Object>> queryDocuments() throws IOException {
+ @SuppressWarnings("rawtypes")
+ SearchResponse<Map> searchResponse = ES_CLIENT.search(
+ searchBuilder -> searchBuilder
+ .index(MavenHardcodedConstants.ES_INDEX_NAME)
.size(LOG_EVENT_COUNT)
- .fetchSource(true);
- final SearchRequest searchRequest =
- new SearchRequest(MavenHardcodedConstants.ES_INDEX_NAME)
- .source(searchSourceBuilder);
- try {
- final SearchResponse searchResponse =
- client.search(searchRequest, RequestOptions.DEFAULT);
- return Arrays
- .stream(searchResponse.getHits().getHits())
- .map(SearchHit::getSourceAsMap)
- .collect(Collectors.toList());
- } catch (ElasticsearchStatusException error) {
- if (RestStatus.NOT_FOUND.equals(error.status())) {
- return Collections.emptyList();
- }
- throw new IOException(error);
- }
+ .source(SourceConfig.of(sourceConfigBuilder -> sourceConfigBuilder.fetch(true))),
+ Map.class);
+ return searchResponse
+ .hits()
+ .hits()
+ .stream()
+ .map(hit -> {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> source = hit.source();
+ return source;
+ })
+ .collect(Collectors.toList());
}
}
diff --git a/log4j-parent/pom.xml b/log4j-parent/pom.xml
index 6ee725d8aa..d70a910627 100644
--- a/log4j-parent/pom.xml
+++ b/log4j-parent/pom.xml
@@ -80,7 +80,7 @@
<!-- `com.conversantmedia:disruptor` version 1.2.16 requires Java 9: -->
<conversant.disruptor.version>1.2.15</conversant.disruptor.version>
<disruptor.version>3.4.4</disruptor.version>
- <elasticsearch.version>7.17.13</elasticsearch.version>
+ <elasticsearch-java.version>8.10.3</elasticsearch-java.version>
<embedded-ldap.version>0.9.0</embedded-ldap.version>
<felix.version>7.0.5</felix.version>
<flapdoodle-embed.version>4.7.1</flapdoodle-embed.version>
@@ -426,9 +426,9 @@
</dependency>
<dependency>
- <groupId>org.elasticsearch.client</groupId>
- <artifactId>elasticsearch-rest-high-level-client</artifactId>
- <version>${elasticsearch.version}</version>
+ <groupId>co.elastic.clients</groupId>
+ <artifactId>elasticsearch-java</artifactId>
+ <version>${elasticsearch-java.version}</version>
</dependency>
<dependency>