ES|QL CCS uses skip_unavailable setting for handling disconnected remote clusters #115266
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,6 @@ | ||
| pr: 115266 | ||
| summary: ES|QL CCS uses `skip_unavailable` setting for handling disconnected remote | ||
| clusters | ||
| area: ES|QL | ||
| type: enhancement | ||
| issues: [ 114531 ] |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,6 +7,7 @@ | |
|
|
||
| package org.elasticsearch.xpack.esql.enrich; | ||
|
|
||
| import org.elasticsearch.ExceptionsHelper; | ||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.action.ActionListenerResponseHandler; | ||
| import org.elasticsearch.action.search.SearchRequest; | ||
|
|
@@ -50,6 +51,7 @@ | |
| import java.util.ArrayList; | ||
| import java.util.Arrays; | ||
| import java.util.Collection; | ||
| import java.util.Collections; | ||
| import java.util.HashMap; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
|
|
@@ -113,12 +115,27 @@ public void resolvePolicies( | |
| final boolean includeLocal = remoteClusters.remove(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY); | ||
| lookupPolicies(remoteClusters, includeLocal, unresolvedPolicies, listener.map(lookupResponses -> { | ||
| final EnrichResolution enrichResolution = new EnrichResolution(); | ||
|
|
||
| Map<String, LookupResponse> lookupResponsesToProcess = new HashMap<>(); | ||
|
|
||
| for (Map.Entry<String, LookupResponse> entry : lookupResponses.entrySet()) { | ||
| String clusterAlias = entry.getKey(); | ||
| if (entry.getValue().connectionError != null) { | ||
| enrichResolution.addUnavailableCluster(clusterAlias, entry.getValue().connectionError); | ||
| // remove unavailable cluster from the list of clusters which is used below to create the ResolvedEnrichPolicy | ||
| remoteClusters.remove(clusterAlias); | ||
| } else { | ||
| lookupResponsesToProcess.put(clusterAlias, entry.getValue()); | ||
| } | ||
| } | ||
|
|
||
| for (UnresolvedPolicy unresolved : unresolvedPolicies) { | ||
| Tuple<ResolvedEnrichPolicy, String> resolved = mergeLookupResults( | ||
| unresolved, | ||
| calculateTargetClusters(unresolved.mode, includeLocal, remoteClusters), | ||
| lookupResponsesToProcess | ||
| ); | ||
|
|
||
| if (resolved.v1() != null) { | ||
| enrichResolution.addResolvedPolicy(unresolved.name, unresolved.mode, resolved.v1()); | ||
| } else { | ||
|
|
@@ -149,13 +166,16 @@ private Tuple<ResolvedEnrichPolicy, String> mergeLookupResults( | |
| Collection<String> targetClusters, | ||
| Map<String, LookupResponse> lookupResults | ||
| ) { | ||
| String policyName = unresolved.name; | ||
| if (targetClusters.isEmpty()) { | ||
| return Tuple.tuple(null, "enrich policy [" + policyName + "] cannot be resolved since remote clusters are unavailable"); | ||
|
There was a problem hiding this comment. I couldn't find a test for this situation. Does it make sense to add one now? (I know there are several other follow up PRs in this area) There was a problem hiding this comment. This code was added in response to one of the IT tests I added failing: In that case invariants in the code below are not met and assertion failures are thrown, so I added this to just leave the code immediately when there are no targetClusters, which can happen when the Enrich Mode type is "REMOTE" and the local cluster was not included in the query. |
||
| } | ||
| final Map<String, ResolvedEnrichPolicy> policies = new HashMap<>(); | ||
| final List<String> failures = new ArrayList<>(); | ||
| for (String cluster : targetClusters) { | ||
| LookupResponse lookupResult = lookupResults.get(cluster); | ||
| if (lookupResult != null) { | ||
| assert lookupResult.connectionError == null : "Should never have a non-null connectionError here"; | ||
| ResolvedEnrichPolicy policy = lookupResult.policies.get(policyName); | ||
| if (policy != null) { | ||
| policies.put(cluster, policy); | ||
|
|
@@ -261,22 +281,34 @@ private void lookupPolicies( | |
| if (remotePolicies.isEmpty() == false) { | ||
| for (String cluster : remoteClusters) { | ||
|
There was a problem hiding this comment. By this point, as I understand, we have already run field caps resolution. So we may have some skipped clusters. Do we exclude them somewhere from the list of clusters for which we're going to resolve policies? Because it looks like it's pointless to try to resolve those policies if the cluster will be skipped anyway. There was a problem hiding this comment. No, the enrich policy resolution is done before the field-caps. I'm not sure why or if that's necessary for some reason or just historical accident. IMO, the field-caps call seems the logical one to do first, but that's not how it is, so we need to track unavailable clusters here and then pass that list to the callback listener which is the one that does the field-caps lookup: https://github.com/elastic/elasticsearch/pull/115266/files#diff-40060e2ec9003953a228c4a03bdc80a301949b0d4e3dccc1978798e33a992a73R345 There was a problem hiding this comment. Oh ok, I was for some reason assuming field caps go first... then of course yes, enrich should be telling field caps which clusters are skipped, not the other way around. |
||
| ActionListener<LookupResponse> lookupListener = refs.acquire(resp -> lookupResponses.put(cluster, resp)); | ||
| getRemoteConnection(cluster, new ActionListener<Transport.Connection>() { | ||
| @Override | ||
| public void onResponse(Transport.Connection connection) { | ||
| transportService.sendRequest( | ||
| connection, | ||
| RESOLVE_ACTION_NAME, | ||
| new LookupRequest(cluster, remotePolicies), | ||
| TransportRequestOptions.EMPTY, | ||
| new ActionListenerResponseHandler<>(lookupListener.delegateResponse((l, e) -> { | ||
| if (ExceptionsHelper.isRemoteUnavailableException(e) | ||
| && remoteClusterService.isSkipUnavailable(cluster)) { | ||
| l.onResponse(new LookupResponse(e)); | ||
| } else { | ||
| l.onFailure(e); | ||
| } | ||
| }), LookupResponse::new, threadPool.executor(ThreadPool.Names.SEARCH)) | ||
| ); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Exception e) { | ||
| if (ExceptionsHelper.isRemoteUnavailableException(e) && remoteClusterService.isSkipUnavailable(cluster)) { | ||
| lookupListener.onResponse(new LookupResponse(e)); | ||
| } else { | ||
| lookupListener.onFailure(e); | ||
| } | ||
| } | ||
| }); | ||
|
Comment on lines
+292
to
+311
There was a problem hiding this comment. Since the code is similar between onFailure and onResponse, pull it out in a separate method. |
||
| } | ||
| } | ||
| // local cluster | ||
|
|
@@ -323,16 +355,30 @@ public void writeTo(StreamOutput out) throws IOException { | |
| private static class LookupResponse extends TransportResponse { | ||
| final Map<String, ResolvedEnrichPolicy> policies; | ||
| final Map<String, String> failures; | ||
| // does not need to be Writable since this indicates a failure to contact a remote cluster, so only set on querying cluster | ||
| final transient Exception connectionError; | ||
|
|
||
| LookupResponse(Map<String, ResolvedEnrichPolicy> policies, Map<String, String> failures) { | ||
| this.policies = policies; | ||
| this.failures = failures; | ||
| this.connectionError = null; | ||
| } | ||
|
|
||
| /** | ||
| * Use this constructor when the remote cluster is unavailable to indicate inability to do the enrich policy lookup | ||
| * @param connectionError Exception received when trying to connect to a remote cluster | ||
| */ | ||
| LookupResponse(Exception connectionError) { | ||
| this.policies = Collections.emptyMap(); | ||
| this.failures = Collections.emptyMap(); | ||
| this.connectionError = connectionError; | ||
| } | ||
|
|
||
| LookupResponse(StreamInput in) throws IOException { | ||
| PlanStreamInput planIn = new PlanStreamInput(in, in.namedWriteableRegistry(), null); | ||
| this.policies = planIn.readMap(StreamInput::readString, ResolvedEnrichPolicy::new); | ||
| this.failures = planIn.readMap(StreamInput::readString, StreamInput::readString); | ||
| this.connectionError = null; | ||
| } | ||
|
|
||
| @Override | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this.failures = failures == null ? emptyList() : failures