Move SlowLogFieldProvider instantiation to node construction (elastic…

…#117949)

SPI from plugins should be created at node startup. This commit moves
creation of SlowLogFieldProvider into node construction and passes it in
to IndicesService so that it is not recreated on each index creation.

relates elastic#102103

docs/changelog/117949.yaml

@@ -0,0 +1,5 @@
+pr: 117949
+summary: Move `SlowLogFieldProvider` instantiation to node construction
+area: Infra/Logging
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/index/IndexModule.java

@@ -208,8 +208,9 @@ public IndexModule(
         this.engineFactory = Objects.requireNonNull(engineFactory);
         // Need to have a mutable arraylist for plugins to add listeners to it
         this.searchOperationListeners = new ArrayList<>(searchOperationListeners);
-        this.searchOperationListeners.add(new SearchSlowLog(indexSettings, slowLogFieldProvider));
-        this.indexOperationListeners.add(new IndexingSlowLog(indexSettings, slowLogFieldProvider));
+        SlowLogFields slowLogFields = slowLogFieldProvider.create(indexSettings);
+        this.searchOperationListeners.add(new SearchSlowLog(indexSettings, slowLogFields));
+        this.indexOperationListeners.add(new IndexingSlowLog(indexSettings, slowLogFields));
         this.directoryFactories = Collections.unmodifiableMap(directoryFactories);
         this.allowExpensiveQueries = allowExpensiveQueries;
         this.expressionResolver = expressionResolver;

server/src/main/java/org/elasticsearch/index/IndexingSlowLog.java

@@ -103,7 +103,7 @@ public final class IndexingSlowLog implements IndexingOperationListener {
      * <em>characters</em> of the source.
      */
     private int maxSourceCharsToLog;
-    private final SlowLogFieldProvider slowLogFieldProvider;
+    private final SlowLogFields slowLogFields;
 
     /**
      * Reads how much of the source to log. The user can specify any value they
@@ -125,8 +125,8 @@ public final class IndexingSlowLog implements IndexingOperationListener {
         Property.IndexScope
     );
 
-    IndexingSlowLog(IndexSettings indexSettings, SlowLogFieldProvider slowLogFieldProvider) {
-        this.slowLogFieldProvider = slowLogFieldProvider;
+    IndexingSlowLog(IndexSettings indexSettings, SlowLogFields slowLogFields) {
+        this.slowLogFields = slowLogFields;
         this.index = indexSettings.getIndex();
 
         indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_REFORMAT_SETTING, this::setReformat);
@@ -179,47 +179,19 @@ public void postIndex(ShardId shardId, Engine.Index indexOperation, Engine.Index
             final long tookInNanos = result.getTook();
             if (indexWarnThreshold >= 0 && tookInNanos > indexWarnThreshold) {
                 indexLogger.warn(
-                    IndexingSlowLogMessage.of(
-                        this.slowLogFieldProvider.indexSlowLogFields(),
-                        index,
-                        doc,
-                        tookInNanos,
-                        reformat,
-                        maxSourceCharsToLog
-                    )
+                    IndexingSlowLogMessage.of(this.slowLogFields.indexFields(), index, doc, tookInNanos, reformat, maxSourceCharsToLog)
                 );
             } else if (indexInfoThreshold >= 0 && tookInNanos > indexInfoThreshold) {
                 indexLogger.info(
-                    IndexingSlowLogMessage.of(
-                        this.slowLogFieldProvider.indexSlowLogFields(),
-                        index,
-                        doc,
-                        tookInNanos,
-                        reformat,
-                        maxSourceCharsToLog
-                    )
+                    IndexingSlowLogMessage.of(this.slowLogFields.indexFields(), index, doc, tookInNanos, reformat, maxSourceCharsToLog)
                 );
             } else if (indexDebugThreshold >= 0 && tookInNanos > indexDebugThreshold) {
                 indexLogger.debug(
-                    IndexingSlowLogMessage.of(
-                        this.slowLogFieldProvider.indexSlowLogFields(),
-                        index,
-                        doc,
-                        tookInNanos,
-                        reformat,
-                        maxSourceCharsToLog
-                    )
+                    IndexingSlowLogMessage.of(this.slowLogFields.indexFields(), index, doc, tookInNanos, reformat, maxSourceCharsToLog)
                 );
             } else if (indexTraceThreshold >= 0 && tookInNanos > indexTraceThreshold) {
                 indexLogger.trace(
-                    IndexingSlowLogMessage.of(
-                        this.slowLogFieldProvider.indexSlowLogFields(),
-                        index,
-                        doc,
-                        tookInNanos,
-                        reformat,
-                        maxSourceCharsToLog
-                    )
+                    IndexingSlowLogMessage.of(this.slowLogFields.indexFields(), index, doc, tookInNanos, reformat, maxSourceCharsToLog)
                 );
             }
         }

server/src/main/java/org/elasticsearch/index/SearchSlowLog.java

@@ -45,7 +45,7 @@ public final class SearchSlowLog implements SearchOperationListener {
     private static final Logger queryLogger = LogManager.getLogger(INDEX_SEARCH_SLOWLOG_PREFIX + ".query");
     private static final Logger fetchLogger = LogManager.getLogger(INDEX_SEARCH_SLOWLOG_PREFIX + ".fetch");
 
-    private final SlowLogFieldProvider slowLogFieldProvider;
+    private final SlowLogFields slowLogFields;
 
     public static final Setting<Boolean> INDEX_SEARCH_SLOWLOG_INCLUDE_USER_SETTING = Setting.boolSetting(
         INDEX_SEARCH_SLOWLOG_PREFIX + ".include.user",
@@ -126,9 +126,8 @@ public final class SearchSlowLog implements SearchOperationListener {
 
     private static final ToXContent.Params FORMAT_PARAMS = new ToXContent.MapParams(Collections.singletonMap("pretty", "false"));
 
-    public SearchSlowLog(IndexSettings indexSettings, SlowLogFieldProvider slowLogFieldProvider) {
-        slowLogFieldProvider.init(indexSettings);
-        this.slowLogFieldProvider = slowLogFieldProvider;
+    public SearchSlowLog(IndexSettings indexSettings, SlowLogFields slowLogFields) {
+        this.slowLogFields = slowLogFields;
         indexSettings.getScopedSettings()
             .addSettingsUpdateConsumer(INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_WARN_SETTING, this::setQueryWarnThreshold);
         this.queryWarnThreshold = indexSettings.getValue(INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_WARN_SETTING).nanos();
@@ -159,26 +158,26 @@ public SearchSlowLog(IndexSettings indexSettings, SlowLogFieldProvider slowLogFi
     @Override
     public void onQueryPhase(SearchContext context, long tookInNanos) {
         if (queryWarnThreshold >= 0 && tookInNanos > queryWarnThreshold) {
-            queryLogger.warn(SearchSlowLogMessage.of(this.slowLogFieldProvider.searchSlowLogFields(), context, tookInNanos));
+            queryLogger.warn(SearchSlowLogMessage.of(this.slowLogFields.searchFields(), context, tookInNanos));
         } else if (queryInfoThreshold >= 0 && tookInNanos > queryInfoThreshold) {
-            queryLogger.info(SearchSlowLogMessage.of(this.slowLogFieldProvider.searchSlowLogFields(), context, tookInNanos));
+            queryLogger.info(SearchSlowLogMessage.of(this.slowLogFields.searchFields(), context, tookInNanos));
         } else if (queryDebugThreshold >= 0 && tookInNanos > queryDebugThreshold) {
-            queryLogger.debug(SearchSlowLogMessage.of(this.slowLogFieldProvider.searchSlowLogFields(), context, tookInNanos));
+            queryLogger.debug(SearchSlowLogMessage.of(this.slowLogFields.searchFields(), context, tookInNanos));
         } else if (queryTraceThreshold >= 0 && tookInNanos > queryTraceThreshold) {
-            queryLogger.trace(SearchSlowLogMessage.of(this.slowLogFieldProvider.searchSlowLogFields(), context, tookInNanos));
+            queryLogger.trace(SearchSlowLogMessage.of(this.slowLogFields.searchFields(), context, tookInNanos));
         }
     }
 
     @Override
     public void onFetchPhase(SearchContext context, long tookInNanos) {
         if (fetchWarnThreshold >= 0 && tookInNanos > fetchWarnThreshold) {
-            fetchLogger.warn(SearchSlowLogMessage.of(this.slowLogFieldProvider.searchSlowLogFields(), context, tookInNanos));
+            fetchLogger.warn(SearchSlowLogMessage.of(this.slowLogFields.searchFields(), context, tookInNanos));
         } else if (fetchInfoThreshold >= 0 && tookInNanos > fetchInfoThreshold) {
-            fetchLogger.info(SearchSlowLogMessage.of(this.slowLogFieldProvider.searchSlowLogFields(), context, tookInNanos));
+            fetchLogger.info(SearchSlowLogMessage.of(this.slowLogFields.searchFields(), context, tookInNanos));
         } else if (fetchDebugThreshold >= 0 && tookInNanos > fetchDebugThreshold) {
-            fetchLogger.debug(SearchSlowLogMessage.of(this.slowLogFieldProvider.searchSlowLogFields(), context, tookInNanos));
+            fetchLogger.debug(SearchSlowLogMessage.of(this.slowLogFields.searchFields(), context, tookInNanos));
         } else if (fetchTraceThreshold >= 0 && tookInNanos > fetchTraceThreshold) {
-            fetchLogger.trace(SearchSlowLogMessage.of(this.slowLogFieldProvider.searchSlowLogFields(), context, tookInNanos));
+            fetchLogger.trace(SearchSlowLogMessage.of(this.slowLogFields.searchFields(), context, tookInNanos));
         }
     }
 

server/src/main/java/org/elasticsearch/index/SlowLogFieldProvider.java

@@ -9,28 +9,14 @@
 
 package org.elasticsearch.index;
 
-import java.util.Map;
-
 /**
  * Interface for providing additional fields to the slow log from a plugin.
  * Intended to be loaded through SPI.
  */
 public interface SlowLogFieldProvider {
     /**
-     * Initialize field provider with index level settings to be able to listen for updates and set initial values
+     * Create a field provider with index level settings to be able to listen for updates and set initial values
      * @param indexSettings settings for the index
      */
-    void init(IndexSettings indexSettings);
-
-    /**
-     * Slow log fields for indexing events
-     * @return map of field name to value
-     */
-    Map<String, String> indexSlowLogFields();
-
-    /**
-     * Slow log fields for search events
-     * @return map of field name to value
-     */
-    Map<String, String> searchSlowLogFields();
+    SlowLogFields create(IndexSettings indexSettings);
 }

server/src/main/java/org/elasticsearch/index/SlowLogFields.java

@@ -0,0 +1,30 @@
+/*
+ * 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.index;
+
+import java.util.Map;
+
+/**
+ * Fields for the slow log. These may be different each call depending on the state of the system.
+ */
+public interface SlowLogFields {
+
+    /**
+     * Slow log fields for indexing events
+     * @return map of field name to value
+     */
+    Map<String, String> indexFields();
+
+    /**
+     * Slow log fields for search events
+     * @return map of field name to value
+     */
+    Map<String, String> searchFields();
+}

server/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -267,6 +267,7 @@ public class IndicesService extends AbstractLifecycleComponent
     private final MapperMetrics mapperMetrics;
     private final List<SearchOperationListener> searchOperationListeners;
     private final QueryRewriteInterceptor queryRewriteInterceptor;
+    final SlowLogFieldProvider slowLogFieldProvider; // pkg-private for testingå
 
     @Override
     protected void doStart() {
@@ -385,6 +386,7 @@ public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, lon
 
         this.timestampFieldMapperService = new TimestampFieldMapperService(settings, threadPool, this);
         this.searchOperationListeners = builder.searchOperationListener;
+        this.slowLogFieldProvider = builder.slowLogFieldProvider;
     }
 
     private static final String DANGLING_INDICES_UPDATE_THREAD_NAME = "DanglingIndices#updateTask";
@@ -755,7 +757,7 @@ private synchronized IndexService createIndexService(
             () -> allowExpensiveQueries,
             indexNameExpressionResolver,
             recoveryStateFactories,
-            loadSlowLogFieldProvider(),
+            slowLogFieldProvider,
             mapperMetrics,
             searchOperationListeners
         );
@@ -835,7 +837,7 @@ public synchronized MapperService createIndexMapperServiceForValidation(IndexMet
             () -> allowExpensiveQueries,
             indexNameExpressionResolver,
             recoveryStateFactories,
-            loadSlowLogFieldProvider(),
+            slowLogFieldProvider,
             mapperMetrics,
             searchOperationListeners
         );
@@ -1437,31 +1439,6 @@ int numPendingDeletes(Index index) {
         }
     }
 
-    // pkg-private for testing
-    SlowLogFieldProvider loadSlowLogFieldProvider() {
-        List<? extends SlowLogFieldProvider> slowLogFieldProviders = pluginsService.loadServiceProviders(SlowLogFieldProvider.class);
-        return new SlowLogFieldProvider() {
-            @Override
-            public void init(IndexSettings indexSettings) {
-                slowLogFieldProviders.forEach(provider -> provider.init(indexSettings));
-            }
-
-            @Override
-            public Map<String, String> indexSlowLogFields() {
-                return slowLogFieldProviders.stream()
-                    .flatMap(provider -> provider.indexSlowLogFields().entrySet().stream())
-                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-            }
-
-            @Override
-            public Map<String, String> searchSlowLogFields() {
-                return slowLogFieldProviders.stream()
-                    .flatMap(provider -> provider.searchSlowLogFields().entrySet().stream())
-                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-            }
-        };
-    }
-
     /**
      * Checks if all pending deletes have completed. Used by tests to ensure we don't check directory contents
      * while deletion still ongoing. * The reason is that, on Windows, browsing the directory contents can interfere

server/src/main/java/org/elasticsearch/indices/IndicesServiceBuilder.java

@@ -23,6 +23,8 @@
 import org.elasticsearch.features.FeatureService;
 import org.elasticsearch.gateway.MetaStateService;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.SlowLogFieldProvider;
+import org.elasticsearch.index.SlowLogFields;
 import org.elasticsearch.index.analysis.AnalysisRegistry;
 import org.elasticsearch.index.engine.EngineFactory;
 import org.elasticsearch.index.mapper.MapperMetrics;
@@ -79,6 +81,22 @@ public class IndicesServiceBuilder {
     MapperMetrics mapperMetrics;
     List<SearchOperationListener> searchOperationListener = List.of();
     QueryRewriteInterceptor queryRewriteInterceptor = null;
+    SlowLogFieldProvider slowLogFieldProvider = new SlowLogFieldProvider() {
+        @Override
+        public SlowLogFields create(IndexSettings indexSettings) {
+            return new SlowLogFields() {
+                @Override
+                public Map<String, String> indexFields() {
+                    return Map.of();
+                }
+
+                @Override
+                public Map<String, String> searchFields() {
+                    return Map.of();
+                }
+            };
+        }
+    };
 
     public IndicesServiceBuilder settings(Settings settings) {
         this.settings = settings;
@@ -191,6 +209,11 @@ public IndicesServiceBuilder searchOperationListeners(List<SearchOperationListen
         return this;
     }
 
+    public IndicesServiceBuilder slowLogFieldProvider(SlowLogFieldProvider slowLogFieldProvider) {
+        this.slowLogFieldProvider = slowLogFieldProvider;
+        return this;
+    }
+
     public IndicesService build() {
         Objects.requireNonNull(settings);
         Objects.requireNonNull(pluginsService);
@@ -216,6 +239,7 @@ public IndicesService build() {
         Objects.requireNonNull(snapshotCommitSuppliers);
         Objects.requireNonNull(mapperMetrics);
         Objects.requireNonNull(searchOperationListener);
+        Objects.requireNonNull(slowLogFieldProvider);
 
         // collect engine factory providers from plugins
         engineFactoryProviders = pluginsService.filterPlugins(EnginePlugin.class)

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

@@ -113,6 +113,8 @@
 import org.elasticsearch.index.IndexSettingProvider;
 import org.elasticsearch.index.IndexSettingProviders;
 import org.elasticsearch.index.IndexingPressure;
+import org.elasticsearch.index.SlowLogFieldProvider;
+import org.elasticsearch.index.SlowLogFields;
 import org.elasticsearch.index.analysis.AnalysisRegistry;
 import org.elasticsearch.index.mapper.MapperMetrics;
 import org.elasticsearch.index.mapper.SourceFieldMetrics;
@@ -808,6 +810,31 @@ private void construct(
             new ShardSearchPhaseAPMMetrics(telemetryProvider.getMeterRegistry())
         );
 
+        List<? extends SlowLogFieldProvider> slowLogFieldProviders = pluginsService.loadServiceProviders(SlowLogFieldProvider.class);
+        // NOTE: the response of index/search slow log fields below must be calculated dynamically on every call
+        // because the responses may change dynamically at runtime
+        SlowLogFieldProvider slowLogFieldProvider = indexSettings -> {
+            final List<SlowLogFields> fields = new ArrayList<>();
+            for (var provider : slowLogFieldProviders) {
+                fields.add(provider.create(indexSettings));
+            }
+            return new SlowLogFields() {
+                @Override
+                public Map<String, String> indexFields() {
+                    return fields.stream()
+                        .flatMap(f -> f.indexFields().entrySet().stream())
+                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+                }
+
+                @Override
+                public Map<String, String> searchFields() {
+                    return fields.stream()
+                        .flatMap(f -> f.searchFields().entrySet().stream())
+                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+                }
+            };
+        };
+
         IndicesService indicesService = new IndicesServiceBuilder().settings(settings)
             .pluginsService(pluginsService)
             .nodeEnvironment(nodeEnvironment)
@@ -829,6 +856,7 @@ private void construct(
             .requestCacheKeyDifferentiator(searchModule.getRequestCacheKeyDifferentiator())
             .mapperMetrics(mapperMetrics)
             .searchOperationListeners(searchOperationListeners)
+            .slowLogFieldProvider(slowLogFieldProvider)
             .build();
 
         final var parameters = new IndexSettingProvider.Parameters(clusterService, indicesService::createIndexMapperServiceForValidation);