allows field caps to be cross project (#137530)

* allows fc to be cross project * Update docs/changelog/137530.yaml * accept projectRouting * more overrides * iter * iter

Author: piergm

docs/changelog/137530.yaml

@@ -0,0 +1,5 @@
+pr: 137530
+summary: Allows field caps to be cross project
+area: Search
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequest.java

@@ -14,11 +14,13 @@
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.LegacyActionRequest;
+import org.elasticsearch.action.ResolvedIndexExpressions;
 import org.elasticsearch.action.ValidateActions;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
@@ -51,6 +53,8 @@ public final class FieldCapabilitiesRequest extends LegacyActionRequest implemen
  private String[] types = Strings.EMPTY_ARRAY;
  private boolean includeUnmapped = false;
  private boolean includeEmptyFields = true;
+ @Nullable
+ private ResolvedIndexExpressions resolvedIndexExpressions = null;
  /**
  * Controls whether the field caps response should always include the list of indices
  * where a field is defined. This flag is only used locally on the coordinating node,
@@ -60,6 +64,7 @@ public final class FieldCapabilitiesRequest extends LegacyActionRequest implemen
  private transient boolean includeIndices = false;
 
  private boolean includeResolvedTo = false;
+ private String projectRouting;
 
  /**
  * Controls whether all local indices should be returned if no remotes matched
@@ -259,6 +264,31 @@ public boolean allowsRemoteIndices() {
  return true;
  }
 
+ @Override
+ public boolean allowsCrossProject() {
+ return true;
+ }
+
+ @Override
+ public void setResolvedIndexExpressions(ResolvedIndexExpressions expressions) {
+ this.resolvedIndexExpressions = expressions;
+ }
+
+ @Override
+ @Nullable
+ public ResolvedIndexExpressions getResolvedIndexExpressions() {
+ return resolvedIndexExpressions;
+ }
+
+ public void projectRouting(String projectRouting) {
+ this.projectRouting = projectRouting;
+ }
+
+ @Override
+ public String getProjectRouting() {
+ return projectRouting;
+ }
+
  @Override
  public boolean includeDataStreams() {
  return true;

server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesResponse.java

@@ -321,6 +321,11 @@ public Builder withResolvedRemotelyBuilder(Map<String, ResolvedIndexExpressions.
  return this;
  }
 
+ public Builder withResolvedRemotely(Map<String, ResolvedIndexExpressions> resolvedRemotely) {
+ this.resolvedRemotely = resolvedRemotely;
+ return this;
+ }
+
  public Builder withResolvedLocally(ResolvedIndexExpressions resolvedLocally) {
  this.resolvedLocally = resolvedLocally;
  return this;

server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesAction.java

@@ -55,6 +55,8 @@
 import org.elasticsearch.logging.LogManager;
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.search.SearchService;
+import org.elasticsearch.search.crossproject.CrossProjectIndexResolutionValidator;
+import org.elasticsearch.search.crossproject.CrossProjectModeDecider;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -88,6 +90,7 @@
 
 import static org.elasticsearch.action.fieldcaps.FieldCapabilitiesRequest.RESOLVED_FIELDS_CAPS;
 import static org.elasticsearch.action.search.TransportSearchHelper.checkCCSVersionCompatibility;
+import static org.elasticsearch.search.crossproject.CrossProjectIndexResolutionValidator.indicesOptionsForCrossProjectFanout;
 
 public class TransportFieldCapabilitiesAction extends HandledTransportAction<FieldCapabilitiesRequest, FieldCapabilitiesResponse> {
  public static final String EXCLUSION = "-";
@@ -110,6 +113,7 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
  private final boolean ccsCheckCompatibility;
  private final ThreadPool threadPool;
  private final TimeValue forceConnectTimeoutSecs;
+ private final CrossProjectModeDecider crossProjectModeDecider;
 
  @Inject
  public TransportFieldCapabilitiesAction(
@@ -138,6 +142,7 @@ public TransportFieldCapabilitiesAction(
  this.ccsCheckCompatibility = SearchService.CCS_VERSION_CHECK_SETTING.get(clusterService.getSettings());
  this.threadPool = threadPool;
  this.forceConnectTimeoutSecs = clusterService.getSettings().getAsTime("search.ccs.force_connect_timeout", null);
+ this.crossProjectModeDecider = new CrossProjectModeDecider(clusterService.getSettings());
  }
 
  @Override
@@ -196,8 +201,14 @@ private <R extends ActionResponse> void doExecuteForked(
  long nowInMillis = request.nowInMillis() == null ? System.currentTimeMillis() : request.nowInMillis();
  final ProjectState projectState = projectResolver.getProjectState(clusterService.state());
  final var minTransportVersion = new AtomicReference<>(clusterService.state().getMinTransportVersion());
+ final IndicesOptions originalIndicesOptions = request.indicesOptions();
+ final boolean resolveCrossProject = crossProjectModeDecider.resolvesCrossProject(request);
  final Map<String, OriginalIndices> remoteClusterIndices = transportService.getRemoteClusterService()
- .groupIndices(request.indicesOptions(), request.indices(), request.returnLocalAll());
+ .groupIndices(
+ resolveCrossProject ? indicesOptionsForCrossProjectFanout(originalIndicesOptions) : originalIndicesOptions,
+ request.indices(),
+ request.returnLocalAll()
+ );
  final OriginalIndices localIndices = remoteClusterIndices.remove(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY);
 
  final String[] concreteLocalIndices;
@@ -270,9 +281,9 @@ private <R extends ActionResponse> void doExecuteForked(
  indexResponses.clear();
  indexMappingHashToResponses.clear();
  };
- Map<String, ResolvedIndexExpressions.Builder> resolvedRemotely = new ConcurrentHashMap<>();
+ Map<String, ResolvedIndexExpressions.Builder> resolvedRemotelyBuilder = new ConcurrentHashMap<>();
  for (String clusterAlias : remoteClusterIndices.keySet()) {
- resolvedRemotely.put(clusterAlias, ResolvedIndexExpressions.builder());
+ resolvedRemotelyBuilder.put(clusterAlias, ResolvedIndexExpressions.builder());
  }
  final Consumer<FieldCapabilitiesIndexResponse> handleIndexResponse = resp -> {
  if (fieldCapTask.isCancelled()) {
@@ -335,12 +346,28 @@ private <R extends ActionResponse> void doExecuteForked(
  if (fieldCapTask.notifyIfCancelled(listener)) {
  releaseResourcesOnCancel.run();
  } else {
+ Map<String, ResolvedIndexExpressions> resolvedRemotely = resolvedRemotelyBuilder.entrySet()
+ .stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().build()));
+ ResolvedIndexExpressions resolvedLocally = new ResolvedIndexExpressions(resolvedLocallyList);
+ if (resolveCrossProject) {
+ final Exception ex = CrossProjectIndexResolutionValidator.validate(
+ request.indicesOptions(),
+ request.getProjectRouting(),
+ resolvedLocally,
+ resolvedRemotely
+ );
+ if (ex != null) {
+ listener.onFailure(ex);
+ return;
+ }
+ }
  mergeIndexResponses(
  request,
  fieldCapTask,
  indexResponses,
  indexFailures,
- resolvedLocallyList,
+ resolvedLocally,
  resolvedRemotely,
  minTransportVersion,
  listener.map(linkedRequestExecutor::wrapPrimary)
@@ -370,15 +397,21 @@ private <R extends ActionResponse> void doExecuteForked(
  for (Map.Entry<String, OriginalIndices> remoteIndices : remoteClusterIndices.entrySet()) {
  String clusterAlias = remoteIndices.getKey();
  OriginalIndices originalIndices = remoteIndices.getValue();
- FieldCapabilitiesRequest remoteRequest = prepareRemoteRequest(clusterAlias, request, originalIndices, nowInMillis);
+ FieldCapabilitiesRequest remoteRequest = prepareRemoteRequest(
+ clusterAlias,
+ request,
+ originalIndices,
+ nowInMillis,
+ resolveCrossProject
+ );
  ActionListener<FieldCapabilitiesResponse> remoteListener = ActionListener.wrap(response -> {
 
  if (request.includeResolvedTo() && response.getResolvedLocally() != null) {
  ResolvedIndexExpressions resolvedOnRemoteProject = response.getResolvedLocally();
  // for bwc we need to check that resolvedOnRemoteProject Exists in the response
  if (resolvedOnRemoteProject != null) {
  for (ResolvedIndexExpression remoteResolvedExpression : resolvedOnRemoteProject.expressions()) {
- resolvedRemotely.computeIfPresent(clusterAlias, (k, v) -> {
+ resolvedRemotelyBuilder.computeIfPresent(clusterAlias, (k, v) -> {
  v.addExpression(remoteResolvedExpression);
  return v;
  });
@@ -413,7 +446,7 @@ private <R extends ActionResponse> void doExecuteForked(
  ),
  Set.of()
  );
- resolvedRemotely.computeIfPresent(clusterAlias, (k, v) -> {
+ resolvedRemotelyBuilder.computeIfPresent(clusterAlias, (k, v) -> {
  v.addExpression(err);
  return v;
  });
@@ -439,7 +472,7 @@ private <R extends ActionResponse> void doExecuteForked(
  ),
  Set.of()
  );
- resolvedRemotely.computeIfPresent(clusterAlias, (k, v) -> {
+ resolvedRemotelyBuilder.computeIfPresent(clusterAlias, (k, v) -> {
  v.addExpression(err);
  return v;
  });
@@ -552,12 +585,11 @@ private static void mergeIndexResponses(
  CancellableTask task,
  Map<String, FieldCapabilitiesIndexResponse> indexResponses,
  FailureCollector indexFailures,
- List<ResolvedIndexExpression> resolvedLocallyList,
- Map<String, ResolvedIndexExpressions.Builder> resolvedRemotely,
+ ResolvedIndexExpressions resolvedLocally,
+ Map<String, ResolvedIndexExpressions> resolvedRemotely,
  AtomicReference<TransportVersion> minTransportVersion,
  ActionListener<FieldCapabilitiesResponse> listener
  ) {
- ResolvedIndexExpressions resolvedLocally = new ResolvedIndexExpressions(resolvedLocallyList);
  List<FieldCapabilitiesFailure> failures = indexFailures.build(indexResponses.keySet());
  if (indexResponses.isEmpty() == false) {
  if (request.isMergeResults()) {
@@ -570,7 +602,7 @@ private static void mergeIndexResponses(
  FieldCapabilitiesResponse.builder()
  .withIndexResponses(new ArrayList<>(indexResponses.values()))
  .withResolvedLocally(resolvedLocally)
- .withResolvedRemotelyBuilder(resolvedRemotely)
+ .withResolvedRemotely(resolvedRemotely)
  .withMinTransportVersion(minTransportVersion.get())
  .withFailures(failures)
  .build()
@@ -590,7 +622,7 @@ private static void mergeIndexResponses(
  FieldCapabilitiesResponse.builder()
  .withFailures(failures)
  .withResolvedLocally(resolvedLocally)
- .withResolvedRemotelyBuilder(resolvedRemotely)
+ .withResolvedRemotely(resolvedRemotely)
  .withMinTransportVersion(minTransportVersion.get())
  .build()
  );
@@ -607,12 +639,18 @@ private static FieldCapabilitiesRequest prepareRemoteRequest(
  String clusterAlias,
  FieldCapabilitiesRequest request,
  OriginalIndices originalIndices,
- long nowInMillis
+ long nowInMillis,
+ boolean resolveCrossProject
  ) {
+ IndicesOptions indicesOptions = originalIndices.indicesOptions();
+ if (indicesOptions.resolveCrossProjectIndexExpression()) {
+ // if is a CPS request reset CrossProjectModeOptions to Default and use lenient IndicesOptions.
+ indicesOptions = indicesOptionsForCrossProjectFanout(indicesOptions);
+ }
  FieldCapabilitiesRequest remoteRequest = new FieldCapabilitiesRequest();
  remoteRequest.clusterAlias(clusterAlias);
  remoteRequest.setMergeResults(false); // we need to merge on this node
- remoteRequest.indicesOptions(originalIndices.indicesOptions());
+ remoteRequest.indicesOptions(indicesOptions);
  remoteRequest.indices(originalIndices.indices());
  remoteRequest.fields(request.fields());
  remoteRequest.filters(request.filters());
@@ -621,7 +659,7 @@ private static FieldCapabilitiesRequest prepareRemoteRequest(
  remoteRequest.indexFilter(request.indexFilter());
  remoteRequest.nowInMillis(nowInMillis);
  remoteRequest.includeEmptyFields(request.includeEmptyFields());
- remoteRequest.includeResolvedTo(request.includeResolvedTo());
+ remoteRequest.includeResolvedTo(request.includeResolvedTo() || resolveCrossProject);
  return remoteRequest;
  }
 
@@ -634,7 +672,7 @@ private static boolean hasSameMappingHash(FieldCapabilitiesIndexResponse r1, Fie
  private static FieldCapabilitiesResponse merge(
  Map<String, FieldCapabilitiesIndexResponse> indexResponsesMap,
  ResolvedIndexExpressions resolvedLocally,
- Map<String, ResolvedIndexExpressions.Builder> resolvedRemotely,
+ Map<String, ResolvedIndexExpressions> resolvedRemotely,
  CancellableTask task,
  FieldCapabilitiesRequest request,
  List<FieldCapabilitiesFailure> failures,
@@ -703,7 +741,7 @@ private static FieldCapabilitiesResponse merge(
  .withMinTransportVersion(minTransportVersion.get());
  if (request.includeResolvedTo() && minTransportVersion.get().supports(RESOLVED_FIELDS_CAPS)) {
  // add resolution to response iff includeResolvedTo and all the nodes in the cluster supports it
- responseBuilder.withResolvedLocally(new ResolvedIndexExpressions(collect)).withResolvedRemotelyBuilder(resolvedRemotely);
+ responseBuilder.withResolvedLocally(new ResolvedIndexExpressions(collect)).withResolvedRemotely(resolvedRemotely);
  }
  return responseBuilder.build();
  }

server/src/main/java/org/elasticsearch/rest/action/RestFieldCapabilitiesAction.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
+import org.elasticsearch.search.crossproject.CrossProjectModeDecider;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 
@@ -33,9 +34,11 @@
 public class RestFieldCapabilitiesAction extends BaseRestHandler {
 
  private final Settings settings;
+ private final CrossProjectModeDecider crossProjectModeDecider;
 
  public RestFieldCapabilitiesAction(Settings settings) {
  this.settings = settings;
+ this.crossProjectModeDecider = new CrossProjectModeDecider(settings);
  }
 
  @Override
@@ -55,15 +58,26 @@ public String getName() {
 
  @Override
  public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
- if (settings != null && settings.getAsBoolean("serverless.cross_project.enabled", false)) {
- // accept but drop project_routing param until fully supported
- request.param("project_routing");
+ final FieldCapabilitiesRequest fieldRequest = new FieldCapabilitiesRequest();
+
+ final boolean crossProjectEnabled = crossProjectModeDecider.crossProjectEnabled();
+ if (crossProjectModeDecider.crossProjectEnabled()) {
+ fieldRequest.projectRouting(request.param("project_routing", null));
+ // Setting includeResolvedTo to always include index resolution data structure in the linked project responses,
+ // in order to allow the coordinating node to call CrossProjectIndexResolutionValidator#validate
+ fieldRequest.includeResolvedTo(true);
  }
 
  final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
- final FieldCapabilitiesRequest fieldRequest = new FieldCapabilitiesRequest();
  fieldRequest.indices(indices);
 
+ if (crossProjectEnabled && fieldRequest.allowsCrossProject()) {
+ var cpsIdxOpts = IndicesOptions.builder(fieldRequest.indicesOptions())
+ .crossProjectModeOptions(new IndicesOptions.CrossProjectModeOptions(true))
+ .build();
+ fieldRequest.indicesOptions(cpsIdxOpts);
+ }
+
  fieldRequest.indicesOptions(IndicesOptions.fromRequest(request, fieldRequest.indicesOptions()));
  fieldRequest.includeUnmapped(request.paramAsBoolean("include_unmapped", false));
  fieldRequest.includeEmptyFields(request.paramAsBoolean("include_empty_fields", true));