forked from elastic/elasticsearch
-
Notifications
You must be signed in to change notification settings - Fork 0
/
SystemIndices.java
186 lines (164 loc) · 9.79 KB
/
SystemIndices.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
/*
* 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 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 or the Server
* Side Public License, v 1.
*/
package org.elasticsearch.indices;
import org.apache.lucene.util.automaton.Automata;
import org.apache.lucene.util.automaton.Automaton;
import org.apache.lucene.util.automaton.CharacterRunAutomaton;
import org.apache.lucene.util.automaton.MinimizationOperations;
import org.apache.lucene.util.automaton.Operations;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.index.Index;
import org.elasticsearch.tasks.TaskResultsService;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import static java.util.stream.Collectors.toUnmodifiableList;
import static org.elasticsearch.tasks.TaskResultsService.TASKS_DESCRIPTOR;
/**
* This class holds the {@link SystemIndexDescriptor} objects that represent system indices the
* node knows about. Methods for determining if an index should be a system index are also provided
* to reduce the locations within the code that need to deal with {@link SystemIndexDescriptor}s.
*/
public class SystemIndices {
private static final Map<String, Collection<SystemIndexDescriptor>> SERVER_SYSTEM_INDEX_DESCRIPTORS = Map.of(
TaskResultsService.class.getName(), List.of(TASKS_DESCRIPTOR)
);
private final CharacterRunAutomaton runAutomaton;
private final Collection<SystemIndexDescriptor> systemIndexDescriptors;
public SystemIndices(Map<String, Collection<SystemIndexDescriptor>> pluginAndModulesDescriptors) {
final Map<String, Collection<SystemIndexDescriptor>> descriptorsMap = buildSystemIndexDescriptorMap(pluginAndModulesDescriptors);
checkForOverlappingPatterns(descriptorsMap);
this.systemIndexDescriptors = descriptorsMap.values().stream().flatMap(Collection::stream).collect(Collectors.toUnmodifiableList());
checkForDuplicateAliases(this.systemIndexDescriptors);
this.runAutomaton = buildCharacterRunAutomaton(systemIndexDescriptors);
}
private void checkForDuplicateAliases(Collection<SystemIndexDescriptor> descriptors) {
final Map<String, Integer> aliasCounts = new HashMap<>();
for (SystemIndexDescriptor descriptor : descriptors) {
final String aliasName = descriptor.getAliasName();
if (aliasName != null) {
aliasCounts.compute(aliasName, (alias, existingCount) -> 1 + (existingCount == null ? 0 : existingCount));
}
}
final List<String> duplicateAliases = aliasCounts.entrySet()
.stream()
.filter(entry -> entry.getValue() > 1)
.map(Map.Entry::getKey)
.sorted()
.collect(Collectors.toList());
if (duplicateAliases.isEmpty() == false) {
throw new IllegalStateException("Found aliases associated with multiple system index descriptors: " + duplicateAliases + "");
}
}
/**
* Determines whether a given index is a system index by comparing its name to the collection of loaded {@link SystemIndexDescriptor}s
* @param index the {@link Index} object to check against loaded {@link SystemIndexDescriptor}s
* @return true if the {@link Index}'s name matches a pattern from a {@link SystemIndexDescriptor}
*/
public boolean isSystemIndex(Index index) {
return isSystemIndex(index.getName());
}
/**
* Determines whether a given index is a system index by comparing its name to the collection of loaded {@link SystemIndexDescriptor}s
* @param indexName the index name to check against loaded {@link SystemIndexDescriptor}s
* @return true if the index name matches a pattern from a {@link SystemIndexDescriptor}
*/
public boolean isSystemIndex(String indexName) {
return runAutomaton.run(indexName);
}
/**
* Finds a single matching {@link SystemIndexDescriptor}, if any, for the given index name.
* @param name the name of the index
* @return The matching {@link SystemIndexDescriptor} or {@code null} if no descriptor is found
* @throws IllegalStateException if multiple descriptors match the name
*/
public @Nullable SystemIndexDescriptor findMatchingDescriptor(String name) {
final List<SystemIndexDescriptor> matchingDescriptors = systemIndexDescriptors.stream()
.filter(descriptor -> descriptor.matchesIndexPattern(name))
.collect(toUnmodifiableList());
if (matchingDescriptors.isEmpty()) {
return null;
} else if (matchingDescriptors.size() == 1) {
return matchingDescriptors.get(0);
} else {
// This should be prevented by failing on overlapping patterns at startup time, but is here just in case.
StringBuilder errorMessage = new StringBuilder()
.append("index name [")
.append(name)
.append("] is claimed as a system index by multiple system index patterns: [")
.append(matchingDescriptors.stream()
.map(descriptor -> "pattern: [" + descriptor.getIndexPattern() +
"], description: [" + descriptor.getDescription() + "]").collect(Collectors.joining("; ")));
// Throw AssertionError if assertions are enabled, or a regular exception otherwise:
assert false : errorMessage.toString();
throw new IllegalStateException(errorMessage.toString());
}
}
private static CharacterRunAutomaton buildCharacterRunAutomaton(Collection<SystemIndexDescriptor> descriptors) {
Optional<Automaton> automaton = descriptors.stream()
.map(descriptor -> SystemIndexDescriptor.buildAutomaton(descriptor.getIndexPattern(), descriptor.getAliasName()))
.reduce(Operations::union);
return new CharacterRunAutomaton(MinimizationOperations.minimize(automaton.orElse(Automata.makeEmpty()), Integer.MAX_VALUE));
}
/**
* Given a collection of {@link SystemIndexDescriptor}s and their sources, checks to see if the index patterns of the listed
* descriptors overlap with any of the other patterns. If any do, throws an exception.
*
* @param sourceToDescriptors A map of source (plugin) names to the SystemIndexDescriptors they provide.
* @throws IllegalStateException Thrown if any of the index patterns overlaps with another.
*/
static void checkForOverlappingPatterns(Map<String, Collection<SystemIndexDescriptor>> sourceToDescriptors) {
List<Tuple<String, SystemIndexDescriptor>> sourceDescriptorPair = sourceToDescriptors.entrySet().stream()
.flatMap(entry -> entry.getValue().stream().map(descriptor -> new Tuple<>(entry.getKey(), descriptor)))
.sorted(Comparator.comparing(d -> d.v1() + ":" + d.v2().getIndexPattern())) // Consistent ordering -> consistent error message
.collect(Collectors.toUnmodifiableList());
// This is O(n^2) with the number of system index descriptors, and each check is quadratic with the number of states in the
// automaton, but the absolute number of system index descriptors should be quite small (~10s at most), and the number of states
// per pattern should be low as well. If these assumptions change, this might need to be reworked.
sourceDescriptorPair.forEach(descriptorToCheck -> {
List<Tuple<String, SystemIndexDescriptor>> descriptorsMatchingThisPattern = sourceDescriptorPair.stream()
.filter(d -> descriptorToCheck.v2() != d.v2()) // Exclude the pattern currently being checked
.filter(d -> overlaps(descriptorToCheck.v2(), d.v2()))
.collect(Collectors.toUnmodifiableList());
if (descriptorsMatchingThisPattern.isEmpty() == false) {
throw new IllegalStateException("a system index descriptor [" + descriptorToCheck.v2() + "] from [" +
descriptorToCheck.v1() + "] overlaps with other system index descriptors: [" +
descriptorsMatchingThisPattern.stream()
.map(descriptor -> descriptor.v2() + " from [" + descriptor.v1() + "]")
.collect(Collectors.joining(", ")));
}
});
}
private static boolean overlaps(SystemIndexDescriptor a1, SystemIndexDescriptor a2) {
Automaton a1Automaton = SystemIndexDescriptor.buildAutomaton(a1.getIndexPattern(), null);
Automaton a2Automaton = SystemIndexDescriptor.buildAutomaton(a2.getIndexPattern(), null);
return Operations.isEmpty(Operations.intersection(a1Automaton, a2Automaton)) == false;
}
private static Map<String, Collection<SystemIndexDescriptor>> buildSystemIndexDescriptorMap(
Map<String, Collection<SystemIndexDescriptor>> pluginAndModulesMap) {
final Map<String, Collection<SystemIndexDescriptor>> map =
new HashMap<>(pluginAndModulesMap.size() + SERVER_SYSTEM_INDEX_DESCRIPTORS.size());
map.putAll(pluginAndModulesMap);
// put the server items last since we expect less of them
SERVER_SYSTEM_INDEX_DESCRIPTORS.forEach((source, descriptors) -> {
if (map.putIfAbsent(source, descriptors) != null) {
throw new IllegalArgumentException("plugin or module attempted to define the same source [" + source +
"] as a built-in system index");
}
});
return Map.copyOf(map);
}
Collection<SystemIndexDescriptor> getSystemIndexDescriptors() {
return this.systemIndexDescriptors;
}
}