Skip to content

Commit d84b65d

Browse files
smalyshevsmalyshev
authored
Ensure the set of remote clusters is consistent over the life of ES|QL query (#126000)
* Ensure the set of remote clusters is consistent over the life of ES|QL query
1 parent 483f979 commit d84b65d

File tree

7 files changed

+82
-53
lines changed

7 files changed

+82
-53
lines changed

‎server/src/main/java/org/elasticsearch/indices/IndicesExpressionGrouper.java‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import org.elasticsearch.common.Strings;
1515

1616
import java.util.Map;
17+
import java.util.Set;
1718

1819
/**
1920
* Interface for grouping index expressions, along with IndicesOptions by cluster alias.
@@ -29,23 +30,36 @@
2930
public interface IndicesExpressionGrouper {
3031

3132
/**
33+
* @param remoteClusterNames Set of configured remote cluster names.
3234
* @param indicesOptions IndicesOptions to clarify how the index expression should be parsed/applied
3335
* @param indexExpressionCsv Multiple index expressions as CSV string (with no spaces), e.g., "logs1,logs2,cluster-a:logs1".
3436
* A single index expression is also supported.
3537
* @return Map where the key is the cluster alias (for "local" cluster, it is RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)
3638
* and the value for that cluster from the index expression is an OriginalIndices object.
3739
*/
38-
default Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String indexExpressionCsv) {
39-
return groupIndices(indicesOptions, Strings.splitStringByCommaToArray(indexExpressionCsv));
40+
default Map<String, OriginalIndices> groupIndices(
41+
Set<String> remoteClusterNames,
42+
IndicesOptions indicesOptions,
43+
String indexExpressionCsv
44+
) {
45+
return groupIndices(remoteClusterNames, indicesOptions, Strings.splitStringByCommaToArray(indexExpressionCsv));
4046
}
4147

4248
/**
4349
* Same behavior as the other groupIndices, except the incoming multiple index expressions must already be
4450
* parsed into a String array.
51+
* @param remoteClusterNames Set of configured remote cluster names.
4552
* @param indicesOptions IndicesOptions to clarify how the index expressions should be parsed/applied
4653
* @param indexExpressions Multiple index expressions as string[].
4754
* @return Map where the key is the cluster alias (for "local" cluster, it is RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)
4855
* and the value for that cluster from the index expression is an OriginalIndices object.
4956
*/
50-
Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indexExpressions);
57+
Map<String, OriginalIndices> groupIndices(Set<String> remoteClusterNames, IndicesOptions indicesOptions, String[] indexExpressions);
58+
59+
/**
60+
* Returns a set of currently configured remote clusters.
61+
*/
62+
default Set<String> getConfiguredClusters() {
63+
return Set.of();
64+
}
5165
}

‎server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java‎

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,7 @@ boolean isRemoteNodeConnected(final String remoteCluster, final DiscoveryNode no
185185

186186
/**
187187
* Group indices by cluster alias mapped to OriginalIndices for that cluster.
188+
* @param remoteClusterNames Set of configured remote cluster names.
188189
* @param indicesOptions IndicesOptions to clarify how the index expressions should be parsed/applied
189190
* @param indices Multiple index expressions as string[].
190191
* @param returnLocalAll whether to support the _all functionality needed by _search
@@ -193,9 +194,14 @@ boolean isRemoteNodeConnected(final String remoteCluster, final DiscoveryNode no
193194
* If false, an empty map is returned when no indices are specified.
194195
* @return Map keyed by cluster alias having OriginalIndices as the map value parsed from the String[] indices argument
195196
*/
196-
public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indices, boolean returnLocalAll) {
197+
public Map<String, OriginalIndices> groupIndices(
198+
Set<String> remoteClusterNames,
199+
IndicesOptions indicesOptions,
200+
String[] indices,
201+
boolean returnLocalAll
202+
) {
197203
final Map<String, OriginalIndices> originalIndicesMap = new HashMap<>();
198-
final Map<String, List<String>> groupedIndices = groupClusterIndices(getRemoteClusterNames(), indices);
204+
final Map<String, List<String>> groupedIndices = groupClusterIndices(remoteClusterNames, indices);
199205
if (groupedIndices.isEmpty()) {
200206
if (returnLocalAll) {
201207
// search on _all in the local cluster if neither local indices nor remote indices were specified
@@ -214,12 +220,26 @@ public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions,
214220
/**
215221
* If no indices are specified, then a Map with one entry for the local cluster with an empty index array is returned.
216222
* For details see {@code groupIndices(IndicesOptions indicesOptions, String[] indices, boolean returnLocalAll)}
223+
* @param remoteClusterNames Set of configured remote cluster names.
217224
* @param indicesOptions IndicesOptions to clarify how the index expressions should be parsed/applied
218225
* @param indices Multiple index expressions as string[].
219226
* @return Map keyed by cluster alias having OriginalIndices as the map value parsed from the String[] indices argument
220227
*/
228+
public Map<String, OriginalIndices> groupIndices(Set<String> remoteClusterNames, IndicesOptions indicesOptions, String[] indices) {
229+
return groupIndices(remoteClusterNames, indicesOptions, indices, true);
230+
}
231+
232+
public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indices, boolean returnLocalAll) {
233+
return groupIndices(getRemoteClusterNames(), indicesOptions, indices, returnLocalAll);
234+
}
235+
221236
public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indices) {
222-
return groupIndices(indicesOptions, indices, true);
237+
return groupIndices(getRemoteClusterNames(), indicesOptions, indices, true);
238+
}
239+
240+
@Override
241+
public Set<String> getConfiguredClusters() {
242+
return getRemoteClusterNames();
223243
}
224244

225245
/**

‎x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -446,8 +446,8 @@ protected void getRemoteConnection(String cluster, ActionListener<Transport.Conn
446446
);
447447
}
448448

449-
public Map<String, List<String>> groupIndicesPerCluster(String[] indices) {
450-
return remoteClusterService.groupIndices(SearchRequest.DEFAULT_INDICES_OPTIONS, indices)
449+
public Map<String, List<String>> groupIndicesPerCluster(Set<String> remoteClusterNames, String[] indices) {
450+
return remoteClusterService.groupIndices(remoteClusterNames, SearchRequest.DEFAULT_INDICES_OPTIONS, indices)
451451
.entrySet()
452452
.stream()
453453
.collect(Collectors.toMap(Map.Entry::getKey, e -> Arrays.asList(e.getValue().indices())));

‎x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlCCSUtils.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -278,12 +278,13 @@ public static void checkForCcsLicense(
278278
EsqlExecutionInfo executionInfo,
279279
List<TableInfo> indices,
280280
IndicesExpressionGrouper indicesGrouper,
281+
Set<String> configuredClusters,
281282
XPackLicenseState licenseState
282283
) {
283284
for (TableInfo tableInfo : indices) {
284285
Map<String, OriginalIndices> groupedIndices;
285286
try {
286-
groupedIndices = indicesGrouper.groupIndices(IndicesOptions.DEFAULT, tableInfo.id().indexPattern());
287+
groupedIndices = indicesGrouper.groupIndices(configuredClusters, IndicesOptions.DEFAULT, tableInfo.id().indexPattern());
287288
} catch (NoSuchRemoteClusterException e) {
288289
if (EsqlLicenseChecker.isCcsAllowed(licenseState)) {
289290
throw e;

‎x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java‎

Lines changed: 13 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,7 @@ public interface PlanRunner {
120120
private final PhysicalPlanOptimizer physicalPlanOptimizer;
121121
private final PlanTelemetry planTelemetry;
122122
private final IndicesExpressionGrouper indicesExpressionGrouper;
123+
private Set<String> configuredClusters;
123124

124125
public EsqlSession(
125126
String sessionId,
@@ -343,6 +344,8 @@ public void analyzedPlan(
343344
plan.setAnalyzed();
344345
return plan;
345346
};
347+
// Capture configured remotes list to ensure consistency throughout the session
348+
configuredClusters = Set.copyOf(indicesExpressionGrouper.getConfiguredClusters());
346349

347350
PreAnalyzer.PreAnalysis preAnalysis = preAnalyzer.preAnalyze(parsed);
348351
var unresolvedPolicies = preAnalysis.enriches.stream()
@@ -355,9 +358,10 @@ public void analyzedPlan(
355358
.collect(Collectors.toSet());
356359
final List<TableInfo> indices = preAnalysis.indices;
357360

358-
EsqlCCSUtils.checkForCcsLicense(executionInfo, indices, indicesExpressionGrouper, verifier.licenseState());
361+
EsqlCCSUtils.checkForCcsLicense(executionInfo, indices, indicesExpressionGrouper, configuredClusters, verifier.licenseState());
359362

360363
final Set<String> targetClusters = enrichPolicyResolver.groupIndicesPerCluster(
364+
configuredClusters,
361365
indices.stream()
362366
.flatMap(t -> Arrays.stream(Strings.commaDelimitedListToStringArray(t.id().indexPattern())))
363367
.toArray(String[]::new)
@@ -378,7 +382,7 @@ public void analyzedPlan(
378382
// invalid index resolution to updateExecutionInfo
379383
if (result.indices.isValid()) {
380384
// CCS indices and skip_unavailable cluster values can stop the analysis right here
381-
if (analyzeCCSIndices(executionInfo, targetClusters, unresolvedPolicies, result, logicalPlanListener, l)) return;
385+
if (allCCSClustersSkipped(executionInfo, result, logicalPlanListener)) return;
382386
}
383387
// whatever tuple we have here (from CCS-special handling or from the original pre-analysis), pass it on to the next step
384388
l.onResponse(result);
@@ -442,6 +446,7 @@ private void preAnalyzeIndices(
442446
IndexPattern table = tableInfo.id();
443447

444448
Map<String, OriginalIndices> clusterIndices = indicesExpressionGrouper.groupIndices(
449+
configuredClusters,
445450
IndicesOptions.DEFAULT,
446451
table.indexPattern()
447452
);
@@ -501,13 +506,14 @@ private void preAnalyzeIndices(
501506
}
502507
}
503508

504-
private boolean analyzeCCSIndices(
509+
/**
510+
* Check if there are any clusters to search.
511+
* @return true if there are no clusters to search, false otherwise
512+
*/
513+
private boolean allCCSClustersSkipped(
505514
EsqlExecutionInfo executionInfo,
506-
Set<String> targetClusters,
507-
Set<EnrichPolicyResolver.UnresolvedPolicy> unresolvedPolicies,
508515
PreAnalysisResult result,
509-
ActionListener<LogicalPlan> logicalPlanListener,
510-
ActionListener<PreAnalysisResult> l
516+
ActionListener<LogicalPlan> logicalPlanListener
511517
) {
512518
IndexResolution indexResolution = result.indices;
513519
EsqlCCSUtils.updateExecutionInfoWithClustersWithNoMatchingIndices(executionInfo, indexResolution);
@@ -520,22 +526,6 @@ private boolean analyzeCCSIndices(
520526
return true;
521527
}
522528

523-
Set<String> newClusters = enrichPolicyResolver.groupIndicesPerCluster(
524-
indexResolution.get().concreteIndices().toArray(String[]::new)
525-
).keySet();
526-
// If new clusters appear when resolving the main indices, we need to resolve the enrich policies again
527-
// or exclude main concrete indices. Since this is rare, it's simpler to resolve the enrich policies again.
528-
// TODO: add a test for this
529-
if (targetClusters.containsAll(newClusters) == false
530-
// do not bother with a re-resolution if only remotes were requested and all were offline
531-
&& executionInfo.getClusterStates(EsqlExecutionInfo.Cluster.Status.RUNNING).findAny().isPresent()) {
532-
enrichPolicyResolver.resolvePolicies(
533-
newClusters,
534-
unresolvedPolicies,
535-
l.map(enrichResolution -> result.withEnrichResolution(enrichResolution))
536-
);
537-
return true;
538-
}
539529
return false;
540530
}
541531

‎x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/session/EsqlCCSUtilsTests.java‎

Lines changed: 24 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -712,22 +712,22 @@ public void testCheckForCcsLicense() {
712712
List<TableInfo> indices = new ArrayList<>();
713713
indices.add(new TableInfo(new IndexPattern(EMPTY, randomFrom("idx", "idx1,idx2*"))));
714714

715-
checkForCcsLicense(executionInfo, indices, indicesGrouper, enterpriseLicenseValid);
716-
checkForCcsLicense(executionInfo, indices, indicesGrouper, platinumLicenseValid);
717-
checkForCcsLicense(executionInfo, indices, indicesGrouper, goldLicenseValid);
718-
checkForCcsLicense(executionInfo, indices, indicesGrouper, trialLicenseValid);
719-
checkForCcsLicense(executionInfo, indices, indicesGrouper, basicLicenseValid);
720-
checkForCcsLicense(executionInfo, indices, indicesGrouper, standardLicenseValid);
721-
checkForCcsLicense(executionInfo, indices, indicesGrouper, missingLicense);
722-
checkForCcsLicense(executionInfo, indices, indicesGrouper, nullLicense);
723-
724-
checkForCcsLicense(executionInfo, indices, indicesGrouper, enterpriseLicenseInactive);
725-
checkForCcsLicense(executionInfo, indices, indicesGrouper, platinumLicenseInactive);
726-
checkForCcsLicense(executionInfo, indices, indicesGrouper, goldLicenseInactive);
727-
checkForCcsLicense(executionInfo, indices, indicesGrouper, trialLicenseInactive);
728-
checkForCcsLicense(executionInfo, indices, indicesGrouper, basicLicenseInactive);
729-
checkForCcsLicense(executionInfo, indices, indicesGrouper, standardLicenseInactive);
730-
checkForCcsLicense(executionInfo, indices, indicesGrouper, missingLicenseInactive);
715+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), enterpriseLicenseValid);
716+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), platinumLicenseValid);
717+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), goldLicenseValid);
718+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), trialLicenseValid);
719+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), basicLicenseValid);
720+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), standardLicenseValid);
721+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), missingLicense);
722+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), nullLicense);
723+
724+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), enterpriseLicenseInactive);
725+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), platinumLicenseInactive);
726+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), goldLicenseInactive);
727+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), trialLicenseInactive);
728+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), basicLicenseInactive);
729+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), standardLicenseInactive);
730+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), missingLicenseInactive);
731731
}
732732

733733
// cross-cluster search requires a valid (active, non-expired) enterprise license OR a valid trial license
@@ -742,8 +742,8 @@ public void testCheckForCcsLicense() {
742742
}
743743

744744
// licenses that work
745-
checkForCcsLicense(executionInfo, indices, indicesGrouper, enterpriseLicenseValid);
746-
checkForCcsLicense(executionInfo, indices, indicesGrouper, trialLicenseValid);
745+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), enterpriseLicenseValid);
746+
checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), trialLicenseValid);
747747

748748
// all others fail ---
749749

@@ -812,7 +812,7 @@ private void assertLicenseCheckFails(
812812
EsqlExecutionInfo executionInfo = new EsqlExecutionInfo(true);
813813
ElasticsearchStatusException e = expectThrows(
814814
ElasticsearchStatusException.class,
815-
() -> checkForCcsLicense(executionInfo, indices, indicesGrouper, licenseState)
815+
() -> checkForCcsLicense(executionInfo, indices, indicesGrouper, Set.of(), licenseState)
816816
);
817817
assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
818818
assertThat(
@@ -825,7 +825,11 @@ private void assertLicenseCheckFails(
825825

826826
static class TestIndicesExpressionGrouper implements IndicesExpressionGrouper {
827827
@Override
828-
public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indexExpressions) {
828+
public Map<String, OriginalIndices> groupIndices(
829+
Set<String> remoteClusterNames,
830+
IndicesOptions indicesOptions,
831+
String[] indexExpressions
832+
) {
829833
final Map<String, OriginalIndices> originalIndicesMap = new HashMap<>();
830834
final String localKey = RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY;
831835

‎x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/telemetry/PlanExecutorMetricsTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ public void testFailedMetric() {
157157
// test a failed query: xyz field doesn't exist
158158
request.query("from test | stats m = max(xyz)");
159159
EsqlSession.PlanRunner runPhase = (p, r) -> fail("this shouldn't happen");
160-
IndicesExpressionGrouper groupIndicesByCluster = (indicesOptions, indexExpressions) -> Map.of(
160+
IndicesExpressionGrouper groupIndicesByCluster = (remoteClusterNames, indicesOptions, indexExpressions) -> Map.of(
161161
"",
162162
new OriginalIndices(new String[] { "test" }, IndicesOptions.DEFAULT)
163163
);

0 commit comments

Comments
 (0)