-
Notifications
You must be signed in to change notification settings - Fork 24.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Expose timestamp field type on coordinator node (#65873)
Today a coordinating node does not have (easy) access to the mappings for the indices for the searches it wishes to coordinate. This means it can't properly interpret a timestamp range filter in a query and must involve a copy of every shard in at least the `can_match` phase. It therefore cannot cope with cases when shards are temporarily not started even if those shards are irrelevant to the search. This commit captures the mapping of the `@timestamp` field for indices which expose a timestamp range in their index metadata.
- Loading branch information
1 parent
ef6fb59
commit e1e1974
Showing
3 changed files
with
269 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
174 changes: 174 additions & 0 deletions
174
server/src/main/java/org/elasticsearch/indices/TimestampFieldMapperService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,174 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.indices; | ||
|
|
||
| import com.carrotsearch.hppc.cursors.ObjectCursor; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.apache.logging.log4j.message.ParameterizedMessage; | ||
| import org.elasticsearch.action.support.PlainActionFuture; | ||
| import org.elasticsearch.cluster.ClusterChangedEvent; | ||
| import org.elasticsearch.cluster.ClusterStateApplier; | ||
| import org.elasticsearch.cluster.metadata.DataStream; | ||
| import org.elasticsearch.cluster.metadata.IndexMetadata; | ||
| import org.elasticsearch.cluster.metadata.Metadata; | ||
| import org.elasticsearch.common.Nullable; | ||
| import org.elasticsearch.common.component.AbstractLifecycleComponent; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.util.concurrent.AbstractRunnable; | ||
| import org.elasticsearch.common.util.concurrent.ConcurrentCollections; | ||
| import org.elasticsearch.common.util.concurrent.EsExecutors; | ||
| import org.elasticsearch.index.Index; | ||
| import org.elasticsearch.index.IndexService; | ||
| import org.elasticsearch.index.mapper.DateFieldMapper; | ||
| import org.elasticsearch.index.mapper.MappedFieldType; | ||
| import org.elasticsearch.index.mapper.MapperService; | ||
| import org.elasticsearch.index.shard.IndexLongFieldRange; | ||
| import org.elasticsearch.node.Node; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.Objects; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| import static org.elasticsearch.common.util.concurrent.EsExecutors.daemonThreadFactory; | ||
|
|
||
| /** | ||
| * Tracks the mapping of the {@code @timestamp} field of immutable indices that expose their timestamp range in their index metadata. | ||
| * Coordinating nodes do not have (easy) access to mappings for all indices, so we extract the type of this one field from the mapping here. | ||
| */ | ||
| public class TimestampFieldMapperService extends AbstractLifecycleComponent implements ClusterStateApplier { | ||
|
|
||
| private static final Logger logger = LogManager.getLogger(TimestampFieldMapperService.class); | ||
|
|
||
| private final IndicesService indicesService; | ||
| private final ExecutorService executor; // single thread to construct mapper services async as needed | ||
|
|
||
| /** | ||
| * The type of the {@code @timestamp} field keyed by index. Futures may be completed with {@code null} to indicate that there is | ||
| * no usable {@code @timestamp} field. | ||
| */ | ||
| private final Map<Index, PlainActionFuture<DateFieldMapper.DateFieldType>> fieldTypesByIndex = ConcurrentCollections.newConcurrentMap(); | ||
|
|
||
| public TimestampFieldMapperService(Settings settings, ThreadPool threadPool, IndicesService indicesService) { | ||
| this.indicesService = indicesService; | ||
|
|
||
| final String nodeName = Objects.requireNonNull(Node.NODE_NAME_SETTING.get(settings)); | ||
| final String threadName = "TimestampFieldMapperService#updateTask"; | ||
| executor = EsExecutors.newScaling(nodeName + "/" + threadName, 0, 1, 0, TimeUnit.MILLISECONDS, | ||
| daemonThreadFactory(nodeName, threadName), threadPool.getThreadContext()); | ||
| } | ||
|
|
||
| @Override | ||
| protected void doStart() { | ||
| } | ||
|
|
||
| @Override | ||
| protected void doStop() { | ||
| ThreadPool.terminate(executor, 10, TimeUnit.SECONDS); | ||
| } | ||
|
|
||
| @Override | ||
| protected void doClose() { | ||
| } | ||
|
|
||
| @Override | ||
| public void applyClusterState(ClusterChangedEvent event) { | ||
| final Metadata metadata = event.state().metadata(); | ||
|
|
||
| // clear out mappers for indices that no longer exist or whose timestamp range is no longer known | ||
| fieldTypesByIndex.keySet().removeIf(index -> hasUsefulTimestampField(metadata.index(index)) == false); | ||
|
|
||
| // capture mappers for indices that do exist | ||
| for (ObjectCursor<IndexMetadata> cursor : metadata.indices().values()) { | ||
| final IndexMetadata indexMetadata = cursor.value; | ||
| final Index index = indexMetadata.getIndex(); | ||
|
|
||
| if (hasUsefulTimestampField(indexMetadata) && fieldTypesByIndex.containsKey(index) == false) { | ||
| logger.trace("computing timestamp mapping for {}", index); | ||
| final PlainActionFuture<DateFieldMapper.DateFieldType> future = new PlainActionFuture<>(); | ||
| fieldTypesByIndex.put(index, future); | ||
|
|
||
| final IndexService indexService = indicesService.indexService(index); | ||
| if (indexService == null) { | ||
| logger.trace("computing timestamp mapping for {} async", index); | ||
| executor.execute(new AbstractRunnable() { | ||
| @Override | ||
| public void onFailure(Exception e) { | ||
| logger.debug(new ParameterizedMessage("failed to compute mapping for {}", index), e); | ||
| future.onResponse(null); // no need to propagate a failure to create the mapper service to searches | ||
| } | ||
|
|
||
| @Override | ||
| protected void doRun() throws Exception { | ||
| try (MapperService mapperService = indicesService.createIndexMapperService(indexMetadata)) { | ||
| mapperService.merge(indexMetadata, MapperService.MergeReason.MAPPING_RECOVERY); | ||
| future.onResponse(fromMapperService(mapperService)); | ||
| } | ||
| } | ||
| }); | ||
| } else { | ||
| logger.trace("computing timestamp mapping for {} using existing index service", index); | ||
| try { | ||
| future.onResponse(fromMapperService(indexService.mapperService())); | ||
| } catch (Exception e) { | ||
| assert false : e; | ||
| future.onResponse(null); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static boolean hasUsefulTimestampField(IndexMetadata indexMetadata) { | ||
| if (indexMetadata == null) { | ||
| return false; | ||
| } | ||
| final IndexLongFieldRange timestampMillisRange = indexMetadata.getTimestampMillisRange(); | ||
| return timestampMillisRange.isComplete() && timestampMillisRange != IndexLongFieldRange.UNKNOWN; | ||
| } | ||
|
|
||
| private static DateFieldMapper.DateFieldType fromMapperService(MapperService mapperService) { | ||
| final MappedFieldType mappedFieldType = mapperService.fieldType(DataStream.TimestampField.FIXED_TIMESTAMP_FIELD); | ||
| if (mappedFieldType instanceof DateFieldMapper.DateFieldType) { | ||
| return (DateFieldMapper.DateFieldType) mappedFieldType; | ||
| } else { | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * @return the field type of the {@code @timestamp} field of the given index, or {@code null} if: | ||
| * - the index is not found, | ||
| * - the field is not found, | ||
| * - the mapping is not known yet, or | ||
| * - the field is not a timestamp field. | ||
| */ | ||
| @Nullable | ||
| public DateFieldMapper.DateFieldType getTimestampFieldType(Index index) { | ||
| final PlainActionFuture<DateFieldMapper.DateFieldType> future = fieldTypesByIndex.get(index); | ||
| if (future == null || future.isDone() == false) { | ||
| return null; | ||
| } | ||
| return future.actionGet(); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters