Refactor log writer, update tests

Author: smalyshev

server/src/internalClusterTest/java/org/elasticsearch/search/SearchLoggingIT.java

@@ -12,8 +12,6 @@
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.core.LogEvent;
-import org.apache.logging.log4j.message.MapMessage;
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.admin.indices.create.AutoCreateAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
@@ -23,6 +21,7 @@
 import org.elasticsearch.action.search.OpenPointInTimeRequest;
 import org.elasticsearch.action.search.OpenPointInTimeResponse;
 import org.elasticsearch.action.search.SearchLogProducer;
+import org.elasticsearch.action.search.SearchPhaseExecutionException;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.TransportClosePointInTimeAction;
 import org.elasticsearch.action.search.TransportOpenPointInTimeAction;
@@ -65,6 +64,9 @@
 import static org.elasticsearch.index.query.QueryBuilders.matchQuery;
 import static org.elasticsearch.index.query.QueryBuilders.simpleQueryStringQuery;
 import static org.elasticsearch.test.AbstractSearchCancellationTestCase.ScriptedBlockPlugin.SEARCH_BLOCK_SCRIPT_NAME;
+import static org.elasticsearch.test.ActionLoggingUtils.assertMessageFailure;
+import static org.elasticsearch.test.ActionLoggingUtils.assertMessageSuccess;
+import static org.elasticsearch.test.ActionLoggingUtils.getMessageData;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertFailures;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
@@ -74,7 +76,6 @@
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.instanceOf;
 
 public class SearchLoggingIT extends AbstractSearchCancellationTestCase {
     static AccumulatingMockAppender appender;
@@ -119,14 +120,6 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
 
     private static final String INDEX_NAME = "test_index";
 
-    @SuppressWarnings("unchecked")
-    private Map<String, String> getMessageData(LogEvent event) {
-        assertNotNull(event);
-        assertThat(event.getMessage(), instanceOf(MapMessage.class));
-
-        return ((MapMessage<?, String>) event.getMessage()).getData();
-    }
-
     // Test _search
     public void testSearchLog() {
         setupIndex();
@@ -136,12 +129,8 @@ public void testSearchLog() {
             assertSearchHitsWithoutFailures(prepareSearch().setQuery(simpleQueryStringQuery("fox")), "1");
             var event = appender.getLastEventAndReset();
             Map<String, String> message = getMessageData(event);
-            assertThat(message.get("success"), equalTo("true"));
-            assertThat(message.get("type"), equalTo("search"));
+            assertMessageSuccess(message, "search", "fox");
             assertThat(message.get("hits"), equalTo("1"));
-            assertThat(Long.valueOf(message.get("took")), greaterThan(0L));
-            assertThat(Long.valueOf(message.get("took_millis")), greaterThanOrEqualTo(0L));
-            assertThat(message.get("query"), containsString("fox"));
             assertThat(message.get("indices"), equalTo(""));
         }
 
@@ -150,12 +139,8 @@ public void testSearchLog() {
             assertSearchHitsWithoutFailures(prepareSearch(INDEX_NAME).setQuery(matchQuery("field1", "quick")), "1", "2", "3");
             var event = appender.getLastEventAndReset();
             Map<String, String> message = getMessageData(event);
-            assertThat(message.get("success"), equalTo("true"));
-            assertThat(message.get("type"), equalTo("search"));
+            assertMessageSuccess(message, "search", "quick");
             assertThat(message.get("hits"), equalTo("3"));
-            assertThat(Long.valueOf(message.get("took")), greaterThan(0L));
-            assertThat(Long.valueOf(message.get("took_millis")), greaterThanOrEqualTo(0L));
-            assertThat(message.get("query"), containsString("quick"));
             assertThat(message.get("indices"), equalTo(INDEX_NAME));
         }
     }
@@ -175,15 +160,9 @@ public void testFailureLog() {
         );
         var event = appender.getLastEventAndReset();
         Map<String, String> message = getMessageData(event);
-        assertThat(message.get("success"), equalTo("false"));
-        assertThat(message.get("type"), equalTo("search"));
+        assertMessageFailure(message, "search", "quick brown", SearchPhaseExecutionException.class, "all shards failed");
         assertThat(message.get("hits"), equalTo("0"));
-        assertThat(Long.valueOf(message.get("took")), greaterThan(0L));
-        assertThat(Long.valueOf(message.get("took_millis")), greaterThanOrEqualTo(0L));
-        assertThat(message.get("query"), containsString("quick brown"));
         assertThat(message.get("indices"), equalTo(INDEX_NAME));
-        assertThat(message.get("error.type"), equalTo("org.elasticsearch.action.search.SearchPhaseExecutionException"));
-        assertThat(message.get("error.message"), equalTo("all shards failed"));
     }
 
     public void testSearchCancel() throws Exception {
@@ -201,14 +180,9 @@ public void testSearchCancel() throws Exception {
         ensureSearchWasCancelled(searchResponse);
         var event = appender.getLastEventAndReset();
         Map<String, String> message = getMessageData(event);
-        assertThat(message.get("success"), equalTo("false"));
-        assertThat(message.get("type"), equalTo("search"));
+        assertMessageFailure(message, "search", "mockscript", SearchPhaseExecutionException.class, null);
         assertThat(message.get("hits"), equalTo("0"));
-        assertThat(Long.valueOf(message.get("took")), greaterThan(0L));
-        assertThat(Long.valueOf(message.get("took_millis")), greaterThanOrEqualTo(0L));
-        assertThat(message.get("query"), containsString("mockscript"));
         assertThat(message.get("indices"), equalTo("test"));
-        assertThat(message.get("error.type"), equalTo("org.elasticsearch.action.search.SearchPhaseExecutionException"));
     }
 
     public void testMultiSearch() {
@@ -250,12 +224,8 @@ public void testPitSearch() {
             );
             var event = appender.getLastEventAndReset();
             Map<String, String> message = getMessageData(event);
-            assertThat(message.get("success"), equalTo("true"));
-            assertThat(message.get("type"), equalTo("search"));
+            assertMessageSuccess(message, "search", "fox");
             assertThat(message.get("hits"), equalTo("1"));
-            assertThat(Long.valueOf(message.get("took")), greaterThan(0L));
-            assertThat(Long.valueOf(message.get("took_millis")), greaterThanOrEqualTo(0L));
-            assertThat(message.get("query"), containsString("fox"));
             assertThat(message.get("indices"), equalTo(INDEX_NAME));
         } finally {
             response.decRef();

server/src/main/java/org/elasticsearch/action/search/SearchLogProducer.java

@@ -67,6 +67,11 @@ public Level logLevel(SearchLogContext context, Level defaultLevel) {
         return defaultLevel;
     }
 
+    @Override
+    public String loggerName() {
+        return LOGGER_NAME;
+    }
+
     public void setLogSystemSearches(boolean logSystemSearches) {
         this.logSystemSearches = logSystemSearches;
     }

server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java

@@ -60,8 +60,8 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.logging.DeprecationCategory;
 import org.elasticsearch.common.logging.DeprecationLogger;
+import org.elasticsearch.common.logging.action.ActionLogWriterProvider;
 import org.elasticsearch.common.logging.action.ActionLogger;
-import org.elasticsearch.common.logging.action.Log4jActionWriter;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
@@ -205,7 +205,8 @@ public TransportSearchAction(
         SearchResponseMetrics searchResponseMetrics,
         Client client,
         UsageService usageService,
-        ActionLoggingFieldsProvider fieldProvider
+        ActionLoggingFieldsProvider fieldProvider,
+        ActionLogWriterProvider logWriterProvider
     ) {
         super(TYPE.name(), transportService, actionFilters, SearchRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.threadPool = threadPool;
@@ -241,7 +242,7 @@ public TransportSearchAction(
             "search",
             clusterService.getClusterSettings(),
             new SearchLogProducer(clusterService.getClusterSettings(), indexNameExpressionResolver.getSystemNamePredicate()),
-            new Log4jActionWriter(SearchLogProducer.LOGGER_NAME),
+            logWriterProvider,
             fieldProvider
         );
     }

server/src/main/java/org/elasticsearch/common/logging/action/ActionLogWriter.java

@@ -13,5 +13,7 @@
  * Generic writer interface to record a log message.
  */
 public interface ActionLogWriter {
+    ActionLogWriter NOOP = (m) -> {};
+
     void write(ActionLogMessage message);
 }

server/src/main/java/org/elasticsearch/common/logging/action/ActionLogWriterProvider.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.common.logging.action;
+
+public interface ActionLogWriterProvider {
+    ActionLogWriterProvider NOOP = (n) -> ActionLogWriter.NOOP;
+
+    ActionLogWriter getWriter(String loggerName);
+}

server/src/main/java/org/elasticsearch/common/logging/action/ActionLogger.java

@@ -70,11 +70,11 @@ public ActionLogger(
         String name,
         ClusterSettings settings,
         ActionLoggerProducer<Context> producer,
-        ActionLogWriter writer,
+        ActionLogWriterProvider writerProvider,
         ActionLoggingFieldsProvider fieldsProvider
     ) {
         this.producer = producer;
-        this.writer = writer;
+        this.writer = writerProvider.getWriter(producer.loggerName());
         var context = new ActionLoggingFieldsContext(true);
         // Initialize
         this.additionalFields = fieldsProvider.create(context);

server/src/main/java/org/elasticsearch/common/logging/action/ActionLoggerProducer.java

@@ -27,6 +27,8 @@ default Level logLevel(Context context, Level defaultLevel) {
         return defaultLevel;
     }
 
+    String loggerName();
+
     /**
      * Produces a {@link ActionLogMessage} with common fields.
      */

server/src/main/java/org/elasticsearch/common/logging/action/Log4jActionWriter.java

@@ -19,6 +19,7 @@
  * TODO: Convert this class to use ES Logger API.
  */
 public class Log4jActionWriter implements ActionLogWriter {
+    public static final ActionLogWriterProvider PROVIDER = Log4jActionWriter::new;
 
     private final Logger logger;
 

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -73,6 +73,8 @@
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.logging.DynamicContextDataProvider;
 import org.elasticsearch.common.logging.HeaderWarning;
+import org.elasticsearch.common.logging.action.ActionLogWriterProvider;
+import org.elasticsearch.common.logging.action.Log4jActionWriter;
 import org.elasticsearch.common.network.NetworkModule;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.ClusterSettings;
@@ -1324,6 +1326,11 @@ public <T extends TransportResponse> void sendRequest(
         );
         dataStreamAutoShardingService.init();
 
+        ActionLogWriterProvider logWriterProvider = pluginsService.loadSingletonServiceProvider(
+            ActionLogWriterProvider.class,
+            () -> Log4jActionWriter.PROVIDER
+        );
+
         modules.add(b -> {
             b.bind(NodeService.class).toInstance(nodeService);
             b.bind(BigArrays.class).toInstance(bigArrays);
@@ -1367,6 +1374,7 @@ public <T extends TransportResponse> void sendRequest(
             b.bind(MergeMetrics.class).toInstance(mergeMetrics);
             b.bind(ProjectRoutingResolver.class).toInstance(projectRoutingResolver);
             b.bind(ActionLoggingFieldsProvider.class).toInstance(loggingFieldsProvider);
+            b.bind(ActionLogWriterProvider.class).toInstance(logWriterProvider);
         });
 
         if (ReadinessService.enabled(environment)) {

server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java

@@ -48,6 +48,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.logging.action.ActionLogWriterProvider;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
@@ -1827,7 +1828,8 @@ protected void doWriteTo(StreamOutput out) throws IOException {
                 new SearchResponseMetrics(TelemetryProvider.NOOP.getMeterRegistry()),
                 client,
                 new UsageService(),
-                new TestActionActionLoggingFieldsProvider()
+                new TestActionActionLoggingFieldsProvider(),
+                ActionLogWriterProvider.NOOP
             );
 
             CountDownLatch latch = new CountDownLatch(1);