From fd07effe6b5950df52576ad7be64920b10fffd9d Mon Sep 17 00:00:00 2001 From: Ajimelec Gonzalez Date: Thu, 8 Oct 2026 22:23:06 -0700 Subject: [PATCH] Resolve cross-cluster index schemas through the remote cluster's GetMappings (#4551) PPL removed the "cluster:" prefix from a cross-cluster index name and read its mapping from the local cluster, which cannot see remote clusters (OpenSearch#6573). A query such as source=remote:logs-* failed with "no such index" unless the same index existed locally. The node client now groups index names by cluster, the same way search does, and reads the mapping and settings of remote indices from each remote cluster with its own client. Local indices are unchanged, and callers no longer split or strip names. RemoteClusterService is passed into the client through a provider that the plugin creates and TransportPPLQueryAction fills at startup. Errors match a search: a missing index, an unreachable remote, and a missing permission each report their own error, and skip_unavailable is applied per cluster. If every remote cluster in a source is skipped and no local index is named, there are no fields to query, so the query fails and names the skipped clusters. Remote schemas are cached for 60 seconds per user, so field-level security is respected. Resolves #4551 Signed-off-by: Ajimelec Gonzalez --- docs/user/ppl/admin/cross_cluster_search.md | 43 ++- ...iteCrossClusterRemoteSchemaCoverageIT.java | 16 + .../CalciteCrossClusterRemoteSchemaIT.java | 16 + .../CrossClusterRemoteSchemaCoverageIT.java | 364 ++++++++++++++++++ .../security/CrossClusterRemoteSchemaIT.java | 312 +++++++++++++++ .../sql/security/CrossClusterSearchIT.java | 2 +- .../opensearch/client/IndexMappingErrors.java | 47 +++ .../client/OpenSearchNodeClient.java | 108 +++++- .../client/RemoteClusterServiceProvider.java | 32 ++ .../client/RemoteMappingsReader.java | 232 +++++++++++ .../OpenSearchDescribeIndexRequest.java | 22 +- .../client/OpenSearchNodeClientTest.java | 159 +++++++- .../client/RemoteMappingsReaderTest.java | 296 ++++++++++++++ .../OpenSearchDescribeIndexRequestTest.java | 5 +- .../org/opensearch/sql/plugin/SQLPlugin.java | 14 +- .../plugin/config/OpenSearchPluginModule.java | 6 +- .../transport/TransportPPLQueryAction.java | 7 +- 17 files changed, 1629 insertions(+), 52 deletions(-) create mode 100644 integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterRemoteSchemaCoverageIT.java create mode 100644 integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterRemoteSchemaIT.java create mode 100644 integ-test/src/test/java/org/opensearch/sql/security/CrossClusterRemoteSchemaCoverageIT.java create mode 100644 integ-test/src/test/java/org/opensearch/sql/security/CrossClusterRemoteSchemaIT.java create mode 100644 opensearch/src/main/java/org/opensearch/sql/opensearch/client/IndexMappingErrors.java create mode 100644 opensearch/src/main/java/org/opensearch/sql/opensearch/client/RemoteClusterServiceProvider.java create mode 100644 opensearch/src/main/java/org/opensearch/sql/opensearch/client/RemoteMappingsReader.java create mode 100644 opensearch/src/test/java/org/opensearch/sql/opensearch/client/RemoteMappingsReaderTest.java diff --git a/docs/user/ppl/admin/cross_cluster_search.md b/docs/user/ppl/admin/cross_cluster_search.md index 4acdd41e354..d116dea9728 100644 --- a/docs/user/ppl/admin/cross_cluster_search.md +++ b/docs/user/ppl/admin/cross_cluster_search.md @@ -44,13 +44,34 @@ fetched rows / total rows = 4/4 +----------------+-----------+----------------------+---------+--------+--------+----------+-------+-----+-----------------------+----------+ ``` +To search every connected remote cluster, use `*` as the cluster name, for example `source=*:accounts`. +Local and remote indices can be combined in one query, for example `source=accounts,my_remote_cluster:accounts`. + +## Remote Index Fields + +The fields of a remote cluster index can be queried directly. No index needs to exist on the local cluster. +To keep queries fast, the fields of a remote index are remembered for up to 60 seconds after a query reads +them. If a field is added on the remote cluster during that time, a query that uses it can fail with a +"field not found" error until the 60 seconds pass. The same applies to a new index that matches the pattern, +for example after a rollover. Fields are remembered separately for each user and on each node, so the wait +only affects queries on the same index pattern by the same user. + +## Unavailable Remote Clusters + +If a remote cluster cannot be reached, the query fails with the same error as a search request, for example +`Unable to open any proxy connections to remote cluster [my_remote_cluster]`. If the remote cluster is +configured with `skip_unavailable: true`, it is skipped and the other clusters in the query are still searched. +If every remote cluster in the query is skipped and no local index is named, the query fails with +`Remote cluster [my_remote_cluster] is unavailable and was skipped (skip_unavailable is true)`. + +If the remote cluster is reachable but the index does not exist there, the query fails with `no such index`. + ## Limitations -Since OpenSearch does not support cross-cluster index metadata retrieval, field mapping of a remote cluster index is not available to the local cluster. -([[Feature] Cross-cluster field mappings query #6573](https://github.com/opensearch-project/OpenSearch/issues/6573)) -Therefore, the query engine requires that for any remote cluster index that the users need to search, -the local cluster keep a field mapping system index with the same index name. -This can be done by creating an index on the local cluster with the same name and schema as the remote cluster index. +* Parts of a query can run on the remote cluster as scripts (for example, an aggregation on a field whose type + differs between indices). Run the same OpenSearch and SQL plugin version on the local and remote clusters: a + remote cluster on an older version may not support scripts created by a newer local cluster. + ## Authentication and Permission 1. The security plugin authenticates the user on the local cluster. @@ -60,8 +81,12 @@ This can be done by creating an index on the local cluster with the same name an Check [Cross-cluster search access control](https://opensearch.org/docs/latest/security/access-control/cross-cluster-search/) for more details. Example: Create the ppl_role for test_user on local cluster and the ccs_role for test_user on remote cluster. Then test_user could use PPL to query `ppl-security-demo` index on remote cluster. -1. On the local cluster, refer to [Security Settings](security.md) to create role and user for PPL plugin and index access permission. -2. On the remote cluster, create a new role and grant permission to access index. Create a user with the same name and credentials as the local cluster, and map the user to this role +1. On the local cluster, refer to [Security Settings](security.md) to create role and user for PPL plugin and index access permission. + If the role lists specific index patterns, also include the cluster-prefixed name (for example `*:ppl-security-demo`), + because some requests made on the local cluster, such as point-in-time creation, are checked against it. +2. On the remote cluster, create a new role and grant permission to access index. Create a user with the same name and credentials as the local cluster, and map the user to this role. + The role needs the same index permissions as PPL on a local cluster (see [Security Settings](security.md)), + as in the example below. Without them, the query fails with a permission error. ```bash PUT _plugins/_security/api/roles/ccs_role @@ -71,7 +96,9 @@ PUT _plugins/_security/api/roles/ccs_role "index_patterns":["ppl-security-demo"], "allowed_actions":[ "indices:admin/shards/search_shards", - "indices:data/read/search" + "indices:data/read/search", + "indices:admin/mappings/get", + "indices:monitor/settings/get" ] } ] diff --git a/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterRemoteSchemaCoverageIT.java b/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterRemoteSchemaCoverageIT.java new file mode 100644 index 00000000000..63a9b04e3c4 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterRemoteSchemaCoverageIT.java @@ -0,0 +1,16 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.security; + +/** {@link CrossClusterRemoteSchemaCoverageIT} on the Calcite engine. */ +public class CalciteCrossClusterRemoteSchemaCoverageIT extends CrossClusterRemoteSchemaCoverageIT { + + @Override + protected void init() throws Exception { + super.init(); + enableCalcite(); + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterRemoteSchemaIT.java b/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterRemoteSchemaIT.java new file mode 100644 index 00000000000..2a89bd40230 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterRemoteSchemaIT.java @@ -0,0 +1,16 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.security; + +/** {@link CrossClusterRemoteSchemaIT} on the Calcite engine. */ +public class CalciteCrossClusterRemoteSchemaIT extends CrossClusterRemoteSchemaIT { + + @Override + protected void init() throws Exception { + super.init(); + enableCalcite(); + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterRemoteSchemaCoverageIT.java b/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterRemoteSchemaCoverageIT.java new file mode 100644 index 00000000000..04fc0a07756 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterRemoteSchemaCoverageIT.java @@ -0,0 +1,364 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.security; + +import static org.opensearch.sql.security.SecurityTestBase.STRONG_PASSWORD; +import static org.opensearch.sql.util.MatcherUtils.rows; +import static org.opensearch.sql.util.MatcherUtils.verifyDataRows; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.List; +import java.util.Locale; +import org.json.JSONArray; +import org.json.JSONObject; +import org.junit.Test; +import org.opensearch.client.Request; +import org.opensearch.client.RequestOptions; +import org.opensearch.client.Response; +import org.opensearch.client.ResponseException; +import org.opensearch.client.RestClient; +import org.opensearch.sql.legacy.TestUtils; + +/** + * More cross-cluster cases on remote-only indices: the Log Explorer request shape (time bounds, + * which also turns on index pruning), wildcards over rollover indices, type conflicts, nested and + * multi-fields, the other entry points (SQL, describe, explain), and permissions. + */ +public class CrossClusterRemoteSchemaCoverageIT extends CrossClusterTestBase { + + /** Two remote rollover indices: September (2 docs) and October (2 docs). */ + private static final String ROLLOVER = "ccs_rollover"; + + /** Two indices that disagree on the type of "level" (keyword vs text), on both clusters. */ + private static final String CONFLICT = "ccs_conflict"; + + /** Local index whose "level" is text, mixed with a remote index whose "level" is keyword. */ + private static final String MIXED_LOCAL = "ccs_mixed_local"; + + private static final String NO_MAPPINGS_ROLE = "ccs_no_mappings_role"; + private static final String NO_MAPPINGS_USER = "ccs_no_mappings_user"; + + private static final String ROLLOVER_MAPPING = + """ + {"mappings": {"properties": { + "@timestamp": {"type": "date"}, + "level": {"type": "keyword"}, + "body": {"type": "text", "fields": {"keyword": {"type": "keyword"}}}, + "http": {"properties": {"status_code": {"type": "long"}}} + }}} + """; + + @Override + protected void init() throws Exception { + super.init(); + createIndex( + remoteClient(), + ROLLOVER + "-000001", + ROLLOVER_MAPPING, + doc("2026-09-01T10:00:00Z", "info", "ok", 200), + doc("2026-09-02T10:00:00Z", "error", "disk full", 507)); + createIndex( + remoteClient(), + ROLLOVER + "-000002", + ROLLOVER_MAPPING, + doc("2026-10-01T10:00:00Z", "error", "gateway down", 503), + doc("2026-10-02T10:00:00Z", "info", "ok", 200)); + createIndex( + remoteClient(), + CONFLICT + "-000001", + "{\"mappings\": {\"properties\": {\"level\": {\"type\": \"keyword\"}}}}", + "{\"level\": \"error\"}"); + createIndex( + remoteClient(), + CONFLICT + "-000002", + "{\"mappings\": {\"properties\": {\"level\": {\"type\": \"text\"}}}}", + "{\"level\": \"info\"}"); + createIndex( + client(), + CONFLICT + "-000001", + "{\"mappings\": {\"properties\": {\"level\": {\"type\": \"keyword\"}}}}", + "{\"level\": \"error\"}"); + createIndex( + client(), + CONFLICT + "-000002", + "{\"mappings\": {\"properties\": {\"level\": {\"type\": \"text\"}}}}", + "{\"level\": \"info\"}"); + createIndex( + client(), + MIXED_LOCAL, + "{\"mappings\": {\"properties\": {\"level\": {\"type\": \"text\"}}}}", + "{\"level\": \"warn\"}"); + } + + @Test + public void testLogExplorerRequestOverRemoteRolloverIndices() throws IOException { + JSONObject result = + executeWithTimeBounds( + String.format( + "source=%s:%s-* | where `@timestamp` >= '2026-09-01 00:00:00' and `@timestamp` <=" + + " '2026-10-31 23:59:59' | stats count() as c", + REMOTE_CLUSTER, ROLLOVER), + "2026-09-01 00:00:00", + "2026-10-31 23:59:59"); + + verifyDataRows(result, rows(4)); + } + + /** A time range covering only October lets index pruning drop the September index. */ + @Test + public void testTimeBoundsNarrowTheRemoteIndices() throws IOException { + JSONObject result = + executeWithTimeBounds( + String.format( + "source=%s:%s-* | where `@timestamp` >= '2026-10-01 00:00:00' and `@timestamp` <=" + + " '2026-10-31 23:59:59' | sort `@timestamp` | fields body", + REMOTE_CLUSTER, ROLLOVER), + "2026-10-01 00:00:00", + "2026-10-31 23:59:59"); + + verifyDataRows(result, rows("gateway down"), rows("ok")); + } + + @Test + public void testWildcardAcrossRemoteRolloverIndices() throws IOException { + JSONObject result = + executeQuery( + String.format("source=%s:%s-* | stats count() as c", REMOTE_CLUSTER, ROLLOVER)); + + verifyDataRows(result, rows(4)); + } + + @Test + public void testAllRemoteClustersWildcard() throws IOException { + JSONObject result = executeQuery(String.format("source=*:%s-* | stats count() as c", ROLLOVER)); + + verifyDataRows(result, rows(4)); + } + + /** + * Indices that disagree on a field's type behave the same remotely as locally (the v2 engine, for + * one, drops the text index's row when sorting on the conflicted field, on either side). + */ + @Test + public void testTypeConflictAcrossRemoteIndicesMatchesLocal() throws IOException { + for (String tail : + List.of("| fields level", "| sort level | fields level", "| stats count() as c")) { + JSONArray local = + executeQuery(String.format("source=%s-* %s", CONFLICT, tail)).getJSONArray("datarows"); + JSONArray remote = + executeQuery(String.format("source=%s:%s-* %s", REMOTE_CLUSTER, CONFLICT, tail)) + .getJSONArray("datarows"); + assertEquals(tail, sorted(local), sorted(remote)); + } + } + + private static List sorted(JSONArray rows) { + List out = new java.util.ArrayList<>(); + for (int i = 0; i < rows.length(); i++) { + out.add(rows.get(i).toString()); + } + java.util.Collections.sort(out); + return out; + } + + @Test + public void testNestedFieldFilterAndSort() throws IOException { + JSONObject result = + executeQuery( + String.format( + "source=%s:%s-* | where `http.status_code` >= 500 | sort - `http.status_code` |" + + " fields `http.status_code`, body", + REMOTE_CLUSTER, ROLLOVER)); + + verifyDataRows(result, rows(507, "disk full"), rows(503, "gateway down")); + } + + /** Grouping by a text field uses its ".keyword" sub-field from the remote mapping. */ + @Test + public void testGroupByTextFieldWithKeywordSubField() throws IOException { + JSONObject result = + executeQuery( + String.format( + "source=%s:%s-* | stats count() as c by body | sort body", + REMOTE_CLUSTER, ROLLOVER)); + + verifyDataRows(result, rows(1, "disk full"), rows(1, "gateway down"), rows(2, "ok")); + } + + @Test + public void testMixedLocalAndRemoteWithDifferentFieldTypes() throws IOException { + JSONObject result = + executeQuery( + String.format( + "source=%s,%s:%s-000001 | stats count() as c", + MIXED_LOCAL, REMOTE_CLUSTER, ROLLOVER)); + + verifyDataRows(result, rows(3)); + } + + @Test + public void testSqlOnRemoteIndex() throws IOException { + Request request = new Request("POST", "/_plugins/_sql"); + request.setJsonEntity( + new JSONObject() + .put( + "query", + String.format( + "SELECT level FROM `%s:%s-000001` ORDER BY level", REMOTE_CLUSTER, ROLLOVER)) + .toString()); + JSONObject result = toJson(client().performRequest(request)); + + verifyDataRows(result, rows("error"), rows("info")); + } + + @Test + public void testDescribeRemoteIndex() throws IOException { + JSONObject result = + executeQuery(String.format("describe %s:%s-000001", REMOTE_CLUSTER, ROLLOVER)); + + JSONArray rows = result.getJSONArray("datarows"); + String columns = rows.toString(); + assertTrue(columns, columns.contains("\"level\"")); + assertTrue(columns, columns.contains("\"http.status_code\"")); + } + + @Test + public void testExplainRemoteIndex() throws IOException { + Request request = new Request("POST", "/_plugins/_ppl/_explain"); + request.setJsonEntity( + new JSONObject() + .put("query", String.format("source=%s:%s-* | fields body", REMOTE_CLUSTER, ROLLOVER)) + .toString()); + Response response = client().performRequest(request); + + assertEquals(200, response.getStatusLine().getStatusCode()); + } + + /** Without mappings-get on the remote, the error is a permission error, not "no such index". */ + @Test + public void testMissingMappingsPermissionOnRemoteIsReported() throws IOException { + createUserOnBothClusters( + NO_MAPPINGS_ROLE, + NO_MAPPINGS_USER, + List.of("indices:data/read/search*", "indices:admin/shards/search_shards")); + + ResponseException e = + assertThrows( + ResponseException.class, + () -> + executeQueryAs( + String.format("source=%s:%s-* | fields body", REMOTE_CLUSTER, ROLLOVER), + NO_MAPPINGS_USER)); + + assertEquals(403, e.getResponse().getStatusLine().getStatusCode()); + assertTrue(e.getMessage(), e.getMessage().contains("indices:admin/mappings/get")); + assertFalse(e.getMessage(), e.getMessage().contains("no such index")); + } + + private static String doc(String timestamp, String level, String body, int status) { + return String.format( + Locale.ROOT, + "{\"@timestamp\": \"%s\", \"level\": \"%s\", \"body\": \"%s\", \"http\": {\"status_code\":" + + " %d}}", + timestamp, + level, + body, + status); + } + + private static void createIndex(RestClient cluster, String index, String mapping, String... docs) + throws IOException { + if (cluster.performRequest(new Request("HEAD", "/" + index)).getStatusLine().getStatusCode() + == 200) { + return; + } + Request create = new Request("PUT", "/" + index); + create.setJsonEntity(mapping); + cluster.performRequest(create); + StringBuilder bulk = new StringBuilder(); + for (String doc : docs) { + bulk.append("{\"index\": {}}\n").append(doc).append('\n'); + } + Request load = new Request("POST", "/" + index + "/_bulk?refresh=true"); + load.setJsonEntity(bulk.toString()); + cluster.performRequest(load); + } + + /** The request Dashboards sends: query plus request-level time bounds. */ + private JSONObject executeWithTimeBounds(String query, String start, String end) + throws IOException { + Request request = new Request("POST", "/_plugins/_ppl"); + request.setJsonEntity( + new JSONObject() + .put("query", query) + .put("time_field", "@timestamp") + .put("start_time", start) + .put("end_time", end) + .toString()); + return toJson(client().performRequest(request)); + } + + private void createUserOnBothClusters(String role, String user, List remoteActions) + throws IOException { + for (RestClient cluster : List.of(client(), remoteClient())) { + boolean remote = cluster == remoteClient(); + JSONObject indexPermissions = + new JSONObject() + .put("index_patterns", List.of(ROLLOVER + "-*", "*:" + ROLLOVER + "-*")) + .put( + "allowed_actions", + remote + ? remoteActions + : List.of( + "indices:data/read/search*", + "indices:data/read/field_caps*", + "indices:admin/shards/search_shards")); + put( + cluster, + "/_plugins/_security/api/roles/" + role, + new JSONObject() + .put("cluster_permissions", List.of("cluster:admin/opensearch/ppl")) + .put("index_permissions", List.of(indexPermissions)) + .toString()); + put( + cluster, + "/_plugins/_security/api/internalusers/" + user, + String.format( + Locale.ROOT, "{\"password\": \"%s\", \"backend_roles\": []}", STRONG_PASSWORD)); + put( + cluster, + "/_plugins/_security/api/rolesmapping/" + role, + String.format(Locale.ROOT, "{\"users\": [\"%s\"]}", user)); + } + } + + private static void put(RestClient cluster, String endpoint, String body) throws IOException { + Request request = new Request("PUT", endpoint); + request.setJsonEntity(body); + int status = cluster.performRequest(request).getStatusLine().getStatusCode(); + assertTrue(endpoint + " returned " + status, status == 200 || status == 201); + } + + private JSONObject executeQueryAs(String query, String username) throws IOException { + Request request = new Request("POST", "/_plugins/_ppl"); + request.setJsonEntity(new JSONObject().put("query", query).toString()); + RequestOptions.Builder options = RequestOptions.DEFAULT.toBuilder(); + options.addHeader( + "Authorization", + "Basic " + + Base64.getEncoder() + .encodeToString( + (username + ":" + STRONG_PASSWORD).getBytes(StandardCharsets.UTF_8))); + request.setOptions(options); + return toJson(client().performRequest(request)); + } + + private static JSONObject toJson(Response response) throws IOException { + return new JSONObject(TestUtils.getResponseBody(response, true)); + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterRemoteSchemaIT.java b/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterRemoteSchemaIT.java new file mode 100644 index 00000000000..bd49a015fb5 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterRemoteSchemaIT.java @@ -0,0 +1,312 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.security; + +import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK; +import static org.opensearch.sql.security.SecurityTestBase.STRONG_PASSWORD; +import static org.opensearch.sql.util.MatcherUtils.rows; +import static org.opensearch.sql.util.MatcherUtils.verifyDataRows; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.Locale; +import org.json.JSONArray; +import org.json.JSONObject; +import org.junit.Test; +import org.opensearch.client.Request; +import org.opensearch.client.RequestOptions; +import org.opensearch.client.Response; +import org.opensearch.client.ResponseException; +import org.opensearch.client.RestClient; + +/** + * Cross-cluster queries on indices that exist only on the remote cluster. Unlike {@link + * CrossClusterSearchIT}, nothing here is loaded on the local cluster, so the schema has to be read + * from the remote (via the remote cluster's GetMappings) rather than from a same-named local index. + */ +public class CrossClusterRemoteSchemaIT extends CrossClusterTestBase { + + /** Exists on the remote cluster only. */ + private static final String REMOTE_ONLY = "ccs_remote_only"; + + private static final String REMOTE_ONLY_REMOTE = REMOTE_CLUSTER + ":" + REMOTE_ONLY; + + private static final String FLS_ROLE = "ccs_remote_schema_fls_role"; + private static final String FLS_USER = "ccs_remote_schema_fls_user"; + + @Override + protected void init() throws Exception { + super.init(); + loadIndex(Index.BANK); + createRemoteOnlyIndex(); + } + + @Test + public void testRemoteOnlyIndexReturnsRows() throws IOException { + assertFalse("must not exist locally", indexExists(client(), REMOTE_ONLY)); + + JSONObject result = + executeQuery( + String.format( + "search source=%s | sort `http.status_code` | fields level, `http.status_code`", + REMOTE_ONLY_REMOTE)); + + verifyDataRows(result, rows("info", 200), rows("error", 503)); + } + + @Test + public void testRemoteOnlyIndexAggregation() throws IOException { + JSONObject result = + executeQuery(String.format("search source=%s | stats count() as c", REMOTE_ONLY_REMOTE)); + + verifyDataRows(result, rows(2)); + } + + @Test + public void testMixedLocalAndRemoteIndices() throws IOException { + JSONObject result = + executeQuery( + String.format( + "search source=%s,%s | stats count() as c", TEST_INDEX_BANK, REMOTE_ONLY_REMOTE)); + + // 7 bank documents locally plus 2 remote-only documents. + verifyDataRows(result, rows(9)); + } + + @Test + public void testMissingRemoteIndexIsIndexNotFound() { + ResponseException e = + assertThrows( + ResponseException.class, + () -> executeQuery(String.format("search source=%s:does_not_exist_*", REMOTE_CLUSTER))); + + assertEquals(404, e.getResponse().getStatusLine().getStatusCode()); + assertTrue(e.getMessage(), e.getMessage().contains("no such index")); + } + + @Test + public void testUnreachableRemoteReportsTheSearchError() throws IOException { + withUnreachableRemote( + "ccs_unreachable", + false, + () -> { + ResponseException e = + assertThrows( + ResponseException.class, + () -> executeQuery("search source=ccs_unreachable:logs-*")); + assertTrue( + e.getMessage(), e.getMessage().contains("Unable to open any proxy connections")); + }); + } + + /** + * With skip_unavailable, an unreachable remote is skipped. When it's the only cluster named, + * there are no fields to query, so the query fails and says which cluster was skipped. + */ + @Test + public void testUnreachableRemoteWithSkipUnavailableIsReportedAsSkipped() throws IOException { + withUnreachableRemote( + "ccs_skipped", + true, + () -> { + ResponseException e = + assertThrows( + ResponseException.class, () -> executeQuery("search source=ccs_skipped:logs-*")); + assertEquals(503, e.getResponse().getStatusLine().getStatusCode()); + assertTrue( + e.getMessage(), + e.getMessage() + .contains( + "Remote cluster [ccs_skipped] is unavailable and was skipped" + + " (skip_unavailable is true)")); + }); + } + + /** A skipped remote next to a local index adds nothing; the local index answers alone. */ + @Test + public void testSkippedRemoteWithLocalIndexReturnsTheLocalRows() throws IOException { + withUnreachableRemote( + "ccs_skipped", + true, + () -> + verifyDataRows( + executeQuery( + String.format( + "search source=%s,ccs_skipped:logs-* | stats count() as c", + TEST_INDEX_BANK)), + rows(7))); + } + + /** + * The remote schema is cached per user: a user whose role hides a field (field-level security) + * must not be served the full schema cached for an earlier, unrestricted user. + */ + @Test + public void testCachedRemoteSchemaIsNotSharedAcrossUsers() throws IOException { + String query = String.format("search source=%s", REMOTE_ONLY_REMOTE); + + // The admin sees every field; this also fills the cache for the admin. + assertTrue(schemaOf(executeQuery(query)).contains("body")); + + createFlsUserOnBothClusters(new String[] {"level", "@timestamp"}); + List restricted = schemaOf(executeQueryAs(query, FLS_USER)); + + assertTrue(restricted.toString(), restricted.contains("level")); + assertFalse(restricted.toString(), restricted.contains("body")); + assertFalse(restricted.toString(), restricted.contains("http.status_code")); + } + + private void createRemoteOnlyIndex() throws IOException { + if (indexExists(remoteClient(), REMOTE_ONLY)) { + return; + } + Request create = new Request("PUT", "/" + REMOTE_ONLY); + create.setJsonEntity( + """ + {"mappings": {"properties": { + "@timestamp": {"type": "date"}, + "level": {"type": "keyword"}, + "body": {"type": "text", "fields": {"keyword": {"type": "keyword"}}}, + "http": {"properties": {"status_code": {"type": "long"}}} + }}} + """); + remoteClient().performRequest(create); + + Request bulk = new Request("POST", "/" + REMOTE_ONLY + "/_bulk?refresh=true"); + bulk.setJsonEntity( + """ + {"index": {}} + {"@timestamp": "2026-09-20T10:00:00Z", "level": "info", "body": "ok", "http": {"status_code": 200}} + {"index": {}} + {"@timestamp": "2026-09-21T11:00:00Z", "level": "error", "body": "gateway down", "http": {"status_code": 503}} + """); + remoteClient().performRequest(bulk); + } + + /** Registers a remote alias pointing at a port nothing listens on, runs the check, removes it. */ + private void withUnreachableRemote(String alias, boolean skipUnavailable, Check check) + throws IOException { + putRemoteSetting(alias, "\"127.0.0.1:1\"", String.valueOf(skipUnavailable)); + try { + check.run(); + } finally { + putRemoteSetting(alias, "null", "null"); + } + } + + private void putRemoteSetting(String alias, String proxyAddress, String skipUnavailable) + throws IOException { + Request request = new Request("PUT", "/_cluster/settings"); + request.setJsonEntity( + String.format( + Locale.ROOT, + """ + {"persistent": { + "cluster.remote.%1$s.mode": %2$s, + "cluster.remote.%1$s.proxy_address": %3$s, + "cluster.remote.%1$s.skip_unavailable": %4$s + }} + """, + alias, + "null".equals(proxyAddress) ? "null" : "\"proxy\"", + proxyAddress, + skipUnavailable)); + adminClient().performRequest(request); + } + + /** + * Field-level security is evaluated on the remote cluster in CCS, so the role, user, and mapping + * are created on both clusters. The role also names the cluster-prefixed index, because the local + * cluster checks some actions (such as point-in-time creation) against "cluster:index". + */ + private void createFlsUserOnBothClusters(String[] allowedFields) throws IOException { + StringBuilder fls = new StringBuilder(); + for (String field : allowedFields) { + fls.append(fls.length() == 0 ? "" : ", ").append('"').append(field).append('"'); + } + for (RestClient cluster : List.of(client(), remoteClient())) { + put( + cluster, + "/_plugins/_security/api/roles/" + FLS_ROLE, + String.format( + Locale.ROOT, + """ + { + "cluster_permissions": ["cluster:admin/opensearch/ppl"], + "index_permissions": [{ + "index_patterns": ["%1$s", "*:%1$s"], + "allowed_actions": [ + "indices:data/read/search*", + "indices:data/read/field_caps*", + "indices:admin/shards/search_shards", + "indices:admin/mappings/get", + "indices:monitor/settings/get", + "indices:data/read/point_in_time/create", + "indices:data/read/point_in_time/delete" + ], + "fls": [%2$s] + }] + } + """, + REMOTE_ONLY, + fls)); + put( + cluster, + "/_plugins/_security/api/internalusers/" + FLS_USER, + String.format( + Locale.ROOT, "{\"password\": \"%s\", \"backend_roles\": []}", STRONG_PASSWORD)); + put( + cluster, + "/_plugins/_security/api/rolesmapping/" + FLS_ROLE, + String.format(Locale.ROOT, "{\"users\": [\"%s\"]}", FLS_USER)); + } + } + + private static void put(RestClient cluster, String endpoint, String body) throws IOException { + Request request = new Request("PUT", endpoint); + request.setJsonEntity(body); + Response response = cluster.performRequest(request); + int status = response.getStatusLine().getStatusCode(); + assertTrue(endpoint + " returned " + status, status == 200 || status == 201); + } + + private JSONObject executeQueryAs(String query, String username) throws IOException { + Request request = new Request("POST", "/_plugins/_ppl"); + request.setJsonEntity(new JSONObject().put("query", query).toString()); + RequestOptions.Builder options = RequestOptions.DEFAULT.toBuilder(); + options.addHeader( + "Authorization", + "Basic " + + Base64.getEncoder() + .encodeToString( + (username + ":" + STRONG_PASSWORD) + .getBytes(java.nio.charset.StandardCharsets.UTF_8))); + request.setOptions(options); + Response response = client().performRequest(request); + return new JSONObject(org.opensearch.sql.legacy.TestUtils.getResponseBody(response, true)); + } + + private static List schemaOf(JSONObject result) { + JSONArray schema = result.getJSONArray("schema"); + List names = new ArrayList<>(); + for (int i = 0; i < schema.length(); i++) { + names.add(schema.getJSONObject(i).getString("name")); + } + return names; + } + + private static boolean indexExists(RestClient cluster, String index) throws IOException { + return cluster.performRequest(new Request("HEAD", "/" + index)).getStatusLine().getStatusCode() + == 200; + } + + @FunctionalInterface + private interface Check { + void run() throws IOException; + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterSearchIT.java b/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterSearchIT.java index 86e98c1523f..8514d73cbac 100644 --- a/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterSearchIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterSearchIT.java @@ -43,7 +43,7 @@ public void testMatchAllCrossClusterSearchAllFields() throws IOException { } @Test - public void testCrossClusterSearchWithoutLocalFieldMappingShouldFail() throws IOException { + public void testCrossClusterSearchOfIndexMissingOnRemoteShouldFail() throws IOException { var exception = assertThrows( ResponseException.class, diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/IndexMappingErrors.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/IndexMappingErrors.java new file mode 100644 index 00000000000..8f4d6979c24 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/IndexMappingErrors.java @@ -0,0 +1,47 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.client; + +import org.opensearch.OpenSearchSecurityException; +import org.opensearch.OpenSearchStatusException; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.index.IndexNotFoundException; +import org.opensearch.sql.common.error.ErrorCode; +import org.opensearch.sql.common.error.ErrorReport; + +/** Errors from reading index mappings, the same for local and remote indices. */ +final class IndexMappingErrors { + + static final String LOCATION = "while fetching index mappings"; + + private IndexMappingErrors() {} + + /** Re-thrown directly to be treated as a client error. */ + static ErrorReport indexNotFound(IndexNotFoundException e, String indexName) { + return ErrorReport.wrap(e) + .code(ErrorCode.INDEX_NOT_FOUND) + .location(LOCATION) + .context("index_name", indexName) + .build(); + } + + /** Every remote cluster in the expression was unreachable and has skip_unavailable set. */ + static ErrorReport allClustersSkipped(String message, String indexName) { + return ErrorReport.wrap(new OpenSearchStatusException(message, RestStatus.SERVICE_UNAVAILABLE)) + .code(ErrorCode.EXECUTION_ERROR) + .location(LOCATION) + .context("index_name", indexName) + .build(); + } + + static ErrorReport permissionDenied(OpenSearchSecurityException e, String indexName) { + return ErrorReport.wrap(e) + .code(ErrorCode.PERMISSION_DENIED) + .location(LOCATION) + .context("index_name", indexName) + .build(); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java index a9da90500e6..15f7e8ce067 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java @@ -5,16 +5,21 @@ package org.opensearch.sql.opensearch.client; +import static org.opensearch.action.search.SearchRequest.DEFAULT_INDICES_OPTIONS; + import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import java.util.Arrays; import java.util.Collection; +import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.concurrent.ExecutionException; import java.util.function.Function; import java.util.function.Predicate; +import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; import org.opensearch.OpenSearchSecurityException; @@ -28,12 +33,11 @@ import org.opensearch.cluster.metadata.AliasMetadata; import org.opensearch.common.action.ActionFuture; import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; import org.opensearch.core.tasks.TaskId; import org.opensearch.index.IndexNotFoundException; import org.opensearch.index.IndexSettings; import org.opensearch.sql.calcite.CalcitePlanContext; -import org.opensearch.sql.common.error.ErrorCode; -import org.opensearch.sql.common.error.ErrorReport; import org.opensearch.sql.opensearch.executor.OpenSearchQueryManager; import org.opensearch.sql.opensearch.mapping.IndexMapping; import org.opensearch.sql.opensearch.request.OpenSearchRequest; @@ -41,6 +45,8 @@ import org.opensearch.sql.opensearch.response.OpenSearchResponse; import org.opensearch.sql.opensearch.response.ShardStats; import org.opensearch.tasks.CancellableTask; +import org.opensearch.transport.RemoteClusterAware; +import org.opensearch.transport.RemoteClusterService; import org.opensearch.transport.client.node.NodeClient; /** OpenSearch connection by node client. */ @@ -49,12 +55,25 @@ public class OpenSearchNodeClient implements OpenSearchClient { public static final Function> ALL_FIELDS = (anyIndex -> (anyField -> true)); + /** + * Thread-context key where the security plugin stores the request's user (the value of + * ConfigConstants.OPENSEARCH_SECURITY_USER_INFO_THREAD_CONTEXT in opensearch-commons). + */ + private static final String SECURITY_USER_INFO = "_opendistro_security_user_info"; + + /** Longest wait for a remote cluster's mappings or settings. */ + private static final TimeValue REMOTE_METADATA_TIMEOUT = TimeValue.timeValueSeconds(10); + /** Node client provided by OpenSearch container. */ private final NodeClient client; + /** Resolves "cluster:index" names; only asked for when a name has a cluster prefix. */ + private final Supplier remoteClusters; + /** Constructor of OpenSearchNodeClient. */ - public OpenSearchNodeClient(NodeClient client) { + public OpenSearchNodeClient(NodeClient client, Supplier remoteClusters) { this.client = client; + this.remoteClusters = remoteClusters; } @Override @@ -94,6 +113,19 @@ public void createIndex(String indexName, Map mappings) { */ @Override public Map getIndexMappings(String... indexExpression) { + IndexGroups groups = groupByCluster(indexExpression); + if (groups.remote().isEmpty()) { + return localIndexMappings(indexExpression); + } + Map mappings = new HashMap<>(); + if (groups.local().length > 0) { + mappings.putAll(localIndexMappings(groups.local())); + } + mappings.putAll(readRemote(groups).mappings()); + return mappings; + } + + private Map localIndexMappings(String... indexExpression) { try { GetMappingsResponse mappingsResponse = client.admin().indices().prepareGetMappings(indexExpression).setLocal(true).get(); @@ -105,19 +137,9 @@ public Map getIndexMappings(String... indexExpression) { Collectors.toUnmodifiableMap( Map.Entry::getKey, cursor -> new IndexMapping(cursor.getValue()))); } catch (IndexNotFoundException e) { - // Re-throw directly to be treated as client error finally - throw ErrorReport.wrap(e) - .code(ErrorCode.INDEX_NOT_FOUND) - .location("while fetching index mappings") - .context("index_name", indexExpression[0]) - .build(); + throw IndexMappingErrors.indexNotFound(e, indexExpression[0]); } catch (OpenSearchSecurityException e) { - // Re-throw with permission denied code - throw ErrorReport.wrap(e) - .code(ErrorCode.PERMISSION_DENIED) - .location("while fetching index mappings") - .context("index_name", indexExpression[0]) - .build(); + throw IndexMappingErrors.permissionDenied(e, indexExpression[0]); } catch (Exception e) { throw new IllegalStateException( "Failed to read mapping for index pattern [" @@ -136,6 +158,62 @@ public Map getIndexMappings(String... indexExpression) { */ @Override public Map getIndexMaxResultWindows(String... indexExpression) { + IndexGroups groups = groupByCluster(indexExpression); + if (groups.remote().isEmpty()) { + return localIndexMaxResultWindows(indexExpression); + } + Map windows = new HashMap<>(); + if (groups.local().length > 0) { + windows.putAll(localIndexMaxResultWindows(groups.local())); + } + // Read with the mappings and cached together, so this is a cache hit after getIndexMappings. + windows.putAll(readRemote(groups).windows()); + return windows; + } + + /** + * The remote indices' mappings and settings, read from each remote cluster and cached. If every + * remote cluster was skipped but local indices remain, the remotes add nothing and the local + * indices answer alone, as in a search. With no local indices there are no fields to query, so it + * is an error. + */ + private RemoteMappingsReader.RemoteMetadata readRemote(IndexGroups groups) { + try { + return new RemoteMappingsReader(client, remoteClusters.get(), REMOTE_METADATA_TIMEOUT) + .read( + client.threadPool().getThreadContext().getTransient(SECURITY_USER_INFO), + groups.remote()); + } catch (RemoteMappingsReader.AllClustersSkipped e) { + if (groups.local().length == 0) { + throw IndexMappingErrors.allClustersSkipped(e.getMessage(), e.indexNames); + } + return new RemoteMappingsReader.RemoteMetadata(Map.of(), Map.of()); + } + } + + /** Local index names, and remote cluster alias to the index names asked of it. */ + private record IndexGroups(String[] local, Map remote) {} + + /** + * Groups an expression by cluster. A name is remote only if its prefix is a registered remote + * cluster (or a pattern matching one), as for search, so a local date-math name containing ':' + * stays local. + */ + private IndexGroups groupByCluster(String... indexExpression) { + if (Arrays.stream(indexExpression) + .noneMatch(name -> name.indexOf(RemoteClusterAware.REMOTE_CLUSTER_INDEX_SEPARATOR) >= 0)) { + return new IndexGroups(indexExpression, Map.of()); + } + Map byCluster = new LinkedHashMap<>(); + remoteClusters + .get() + .groupIndices(DEFAULT_INDICES_OPTIONS, indexExpression, index -> false) + .forEach((clusterAlias, indices) -> byCluster.put(clusterAlias, indices.indices())); + String[] local = byCluster.remove(RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY); + return new IndexGroups(local == null ? new String[0] : local, byCluster); + } + + private Map localIndexMaxResultWindows(String... indexExpression) { try { GetSettingsResponse settingsResponse = client.admin().indices().prepareGetSettings(indexExpression).setLocal(true).get(); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/RemoteClusterServiceProvider.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/RemoteClusterServiceProvider.java new file mode 100644 index 00000000000..84a6107bb76 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/RemoteClusterServiceProvider.java @@ -0,0 +1,32 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.client; + +import java.util.function.Supplier; +import org.opensearch.transport.RemoteClusterService; + +/** + * The node's {@link RemoteClusterService}, for {@link OpenSearchNodeClient}. It is only reachable + * through {@code TransportService}, which plugin components are created before, so the plugin + * creates this provider with its components and a transport action fills it at startup. + */ +public class RemoteClusterServiceProvider implements Supplier { + + private volatile RemoteClusterService service; + + public void set(RemoteClusterService remoteClusterService) { + this.service = remoteClusterService; + } + + @Override + public RemoteClusterService get() { + RemoteClusterService current = service; + if (current == null) { + throw new IllegalStateException("Remote cluster service is not available yet"); + } + return current; + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/RemoteMappingsReader.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/RemoteMappingsReader.java new file mode 100644 index 00000000000..d91e33d623d --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/RemoteMappingsReader.java @@ -0,0 +1,232 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.client; + +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; +import com.google.common.util.concurrent.UncheckedExecutionException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.IdentityHashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import org.opensearch.ExceptionsHelper; +import org.opensearch.OpenSearchSecurityException; +import org.opensearch.action.admin.indices.mapping.get.GetMappingsResponse; +import org.opensearch.action.admin.indices.settings.get.GetSettingsRequest; +import org.opensearch.action.admin.indices.settings.get.GetSettingsResponse; +import org.opensearch.cluster.metadata.MappingMetadata; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.index.IndexNotFoundException; +import org.opensearch.index.IndexSettings; +import org.opensearch.sql.opensearch.mapping.IndexMapping; +import org.opensearch.transport.RemoteClusterAware; +import org.opensearch.transport.RemoteClusterService; +import org.opensearch.transport.client.Client; +import org.opensearch.transport.client.node.NodeClient; + +/** + * Reads remote index mappings and settings with each remote cluster's own GetMappings and + * GetSettings, through a per-cluster client. Both read the remote gateway node's cluster state + * ({@code local=true}), so no shard is touched. + * + *

Each remote cluster is asked on its own, so one that fails reports its own error: it is + * skipped when its {@code skip_unavailable} is true, and fails the query otherwise. + */ +class RemoteMappingsReader { + + /** Mappings and max result windows of the indices named, keyed by "cluster:index". */ + record RemoteMetadata(Map mappings, Map windows) {} + + private static final long TTL_SECONDS = 60; + + /** Most top-level fields kept across all cached entries. */ + private static final long MAX_TOTAL_FIELDS = 100000; + + private static final Cache, RemoteMetadata> CACHE = + CacheBuilder.newBuilder() + // One segment: Guava splits maximumWeight across segments and drops any entry heavier + // than one segment's share right after loading it. + .concurrencyLevel(1) + .maximumWeight(MAX_TOTAL_FIELDS) + .weigher((List key, RemoteMetadata metadata) -> weigh(metadata)) + .expireAfterWrite(TTL_SECONDS, TimeUnit.SECONDS) + .build(); + + private final NodeClient client; + private final RemoteClusterService remotes; + private final TimeValue timeout; + + RemoteMappingsReader(NodeClient client, RemoteClusterService remotes, TimeValue timeout) { + this.client = client; + this.remotes = remotes; + this.timeout = timeout; + } + + /** + * Cached per user (field-level security) and indices; an answer with a skipped cluster is never + * cached. + * + * @param user security user of the request, or null when security is off + * @param indicesByCluster remote cluster alias to the index names asked of it + */ + RemoteMetadata read(Object user, Map indicesByCluster) { + List key = List.of(Objects.toString(user, ""), qualifiedNames(indicesByCluster)); + try { + return CACHE.get(key, () -> fetch(indicesByCluster)); + } catch (UncheckedExecutionException | ExecutionException e) { + if (e.getCause() instanceof NotCached notCached) { + return notCached.metadata; + } + if (e.getCause() instanceof RuntimeException runtime) { + throw runtime; + } + throw new IllegalStateException(e.getCause()); + } + } + + private RemoteMetadata fetch(Map indicesByCluster) { + String indexNames = String.join(",", qualifiedNames(indicesByCluster)); + Map mappings = new LinkedHashMap<>(); + Map windows = new LinkedHashMap<>(); + // Rollover indices usually share one mapping: parse it once and share the result, so cost and + // memory follow the number of distinct mappings rather than the number of indices. + Map parsed = new HashMap<>(); + List skipped = new ArrayList<>(); + indicesByCluster.forEach( + (clusterAlias, indices) -> { + try { + Client remote = client.getRemoteClusterClient(clusterAlias); + GetMappingsResponse mappingsResponse = + remote.admin().indices().prepareGetMappings(indices).setLocal(true).get(timeout); + GetSettingsResponse settingsResponse = + remote + .admin() + .indices() + .getSettings( + new GetSettingsRequest().indices(indices).local(true).includeDefaults(true)) + .actionGet(timeout); + mappingsResponse + .mappings() + .forEach( + (index, mapping) -> + mappings.put( + RemoteClusterAware.buildRemoteIndexName(clusterAlias, index), + parsed.computeIfAbsent(mapping, IndexMapping::new))); + settingsResponse + .getIndexToSettings() + .forEach( + (index, settings) -> + windows.put( + RemoteClusterAware.buildRemoteIndexName(clusterAlias, index), + settings.getAsInt( + IndexSettings.MAX_RESULT_WINDOW_SETTING.getKey(), + IndexSettings.MAX_RESULT_WINDOW_SETTING.getDefault( + Settings.EMPTY)))); + } catch (Exception e) { + Throwable cause = ExceptionsHelper.unwrapCause(e); + // A missing index or a denied read is the remote's answer, not an outage. + if (cause instanceof IndexNotFoundException notFound) { + throw IndexMappingErrors.indexNotFound(notFound, indexNames); + } + if (cause instanceof OpenSearchSecurityException denied) { + throw IndexMappingErrors.permissionDenied(denied, indexNames); + } + if (!remotes.isSkipUnavailable(clusterAlias)) { + throw e instanceof RuntimeException runtime ? runtime : new IllegalStateException(e); + } + skipped.add(clusterAlias); + } + }); + + if (mappings.isEmpty()) { + if (!skipped.isEmpty()) { + throw new AllClustersSkipped(skippedMessage(skipped), indexNames); + } + throw IndexMappingErrors.indexNotFound(new IndexNotFoundException(indexNames), indexNames); + } + if (!skipped.isEmpty()) { + // Partial answer during an outage: serve it, but don't cache it. The search that follows + // reports the skipped cluster. + throw new NotCached(new RemoteMetadata(mappings, windows)); + } + return new RemoteMetadata(mappings, windows); + } + + /** Clusters named in a skipped-cluster message; the rest are counted. */ + private static final int MAX_NAMED_CLUSTERS = 5; + + /** Names up to {@link #MAX_NAMED_CLUSTERS} skipped clusters, sorted, and counts the rest. */ + static String skippedMessage(List clusters) { + List sorted = clusters.stream().sorted().toList(); + String named = + sorted.stream().limit(MAX_NAMED_CLUSTERS).collect(Collectors.joining(", ", "[", "]")); + int more = sorted.size() - MAX_NAMED_CLUSTERS; + String message = + sorted.size() == 1 + ? "Remote cluster " + named + " is unavailable and was skipped" + : "Remote clusters " + + named + + (more > 0 ? " and " + more + " more" : "") + + " are unavailable and were skipped"; + return message + " (skip_unavailable is true)"; + } + + /** + * Every remote cluster asked was unreachable and skipped, so there are no remote fields. The + * caller decides: local indices in the same query still answer, and otherwise it is an error. + */ + static final class AllClustersSkipped extends RuntimeException { + /** The "cluster:index" names asked, for the error's context. */ + final String indexNames; + + AllClustersSkipped(String message, String indexNames) { + super(message, null, false, false); + this.indexNames = indexNames; + } + } + + /** Carries a partial answer out of the cache loader without caching it. */ + static final class NotCached extends RuntimeException { + final transient RemoteMetadata metadata; + + NotCached(RemoteMetadata metadata) { + super(null, null, false, false); + this.metadata = metadata; + } + } + + /** Field count of the distinct mappings: indices sharing one mapping weigh it once. */ + static int weigh(RemoteMetadata metadata) { + Set distinct = Collections.newSetFromMap(new IdentityHashMap<>()); + distinct.addAll(metadata.mappings().values()); + return Math.max(1, distinct.stream().mapToInt(m -> m.getFieldMappings().size()).sum()); + } + + static void clear() { + CACHE.invalidateAll(); + } + + /** "cluster:index" for every index asked of every cluster, sorted, as the cache key. */ + private static List qualifiedNames(Map indicesByCluster) { + return indicesByCluster.entrySet().stream() + .flatMap( + entry -> + Arrays.stream(entry.getValue()) + .map(index -> RemoteClusterAware.buildRemoteIndexName(entry.getKey(), index))) + .sorted() + .toList(); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequest.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequest.java index 2b65b5c7af1..6847b6299a8 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequest.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequest.java @@ -9,7 +9,6 @@ import static org.opensearch.sql.opensearch.client.OpenSearchClient.META_CLUSTER_NAME; import java.util.ArrayList; -import java.util.Arrays; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -100,8 +99,7 @@ public List search() { // TODO possible collision if two indices have fields with the same name and different mappings public Map getFieldTypes() { Map fieldTypes = new HashMap<>(); - Map indexMappings = - client.getIndexMappings(getLocalIndexNames(indexName.getIndexNames())); + Map indexMappings = client.getIndexMappings(indexName.getIndexNames()); if (indexMappings.size() <= 1) { for (IndexMapping indexMapping : indexMappings.values()) { fieldTypes.putAll(indexMapping.getFieldMappings()); @@ -130,10 +128,7 @@ private static Map deepCopy( * @return max result window */ public Integer getMaxResultWindow() { - return client - .getIndexMaxResultWindows(getLocalIndexNames(indexName.getIndexNames())) - .values() - .stream() + return client.getIndexMaxResultWindows(indexName.getIndexNames()).values().stream() .min(Integer::compare) .get(); } @@ -158,19 +153,6 @@ private ExprTupleValue row(String fieldName, String fieldType, int position, Str return new ExprTupleValue(valueMap); } - /** - * Return index names without "{cluster}:" prefix. Without the prefix, they refer to the indices - * at the local cluster. - * - * @param indexNames a string array of index names - * @return local cluster index names - */ - private String[] getLocalIndexNames(String[] indexNames) { - return Arrays.stream(indexNames) - .map(name -> name.substring(name.indexOf(":") + 1)) - .toArray(String[]::new); - } - private String clusterName(Map meta) { return meta.getOrDefault(META_CLUSTER_NAME, DEFAULT_TABLE_CAT); } diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClientTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClientTest.java index 1671faedaf2..66a08bd9df7 100644 --- a/opensearch/src/test/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClientTest.java +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClientTest.java @@ -14,6 +14,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Answers.RETURNS_DEEP_STUBS; import static org.mockito.ArgumentMatchers.*; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -29,11 +30,13 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicBoolean; import lombok.SneakyThrows; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.lucene.search.TotalHits; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayNameGeneration; import org.junit.jupiter.api.DisplayNameGenerator; @@ -43,6 +46,7 @@ import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.junit.jupiter.MockitoExtension; +import org.opensearch.action.OriginalIndices; import org.opensearch.action.admin.indices.create.CreateIndexRequest; import org.opensearch.action.admin.indices.create.CreateIndexResponse; import org.opensearch.action.admin.indices.exists.indices.IndicesExistsRequest; @@ -51,6 +55,7 @@ import org.opensearch.action.admin.indices.mapping.get.GetMappingsResponse; import org.opensearch.action.admin.indices.settings.get.GetSettingsResponse; import org.opensearch.action.search.*; +import org.opensearch.action.search.SearchRequest; import org.opensearch.cluster.metadata.AliasMetadata; import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.cluster.metadata.MappingMetadata; @@ -81,6 +86,8 @@ import org.opensearch.sql.opensearch.request.OpenSearchRequest; import org.opensearch.sql.opensearch.request.OpenSearchScrollRequest; import org.opensearch.sql.opensearch.response.OpenSearchResponse; +import org.opensearch.transport.RemoteClusterService; +import org.opensearch.transport.client.Client; import org.opensearch.transport.client.node.NodeClient; @ExtendWith(MockitoExtension.class) @@ -104,9 +111,11 @@ class OpenSearchNodeClientTest { private OpenSearchClient client; + private final RemoteClusterServiceProvider remoteClusters = new RemoteClusterServiceProvider(); + @BeforeEach void setUp() { - this.client = new OpenSearchNodeClient(nodeClient); + this.client = new OpenSearchNodeClient(nodeClient, remoteClusters); } @Test @@ -582,6 +591,154 @@ void get_index_max_result_windows_error_message_includes_multiple_indices() { () -> assertTrue(exception.getMessage().contains(underlyingError))); } + /** Names without a cluster prefix take the local path and never need the remote service. */ + @Test + void get_index_mappings_of_local_names_does_not_resolve_remote_clusters() { + // The provider is never filled here: asking it would throw. + mockNodeClientIndicesMappings("logs", ""); + + assertEquals(Set.of("logs"), client.getIndexMappings("logs").keySet()); + } + + @Test + void get_index_mappings_merges_local_and_remote_indices() { + givenRemoteClusters(Map.of("", new String[] {"logs"}, "cape", new String[] {"logs-*"})); + mockNodeClientIndicesMappings("logs", ""); + givenRemoteIndex("cape", "logs-1", 50000); + + assertEquals( + Set.of("logs", "cape:logs-1"), client.getIndexMappings("logs", "cape:logs-*").keySet()); + } + + @Test + void get_index_max_result_windows_merges_local_and_remote_indices() throws IOException { + givenRemoteClusters(Map.of("", new String[] {"accounts"}, "cape", new String[] {"logs-*"})); + mockNodeClientSettings( + "accounts", + Resources.toString(Resources.getResource(TEST_MAPPING_SETTINGS_FILE), Charsets.UTF_8)); + givenRemoteIndex("cape", "logs-1", 50000); + + assertEquals( + Map.of("accounts", 100, "cape:logs-1", 50000), + client.getIndexMaxResultWindows("accounts", "cape:logs-*")); + } + + /** Skipped remotes add nothing when local indices are named too, as in a search. */ + @Test + void get_index_mappings_answers_from_local_indices_when_every_remote_is_skipped() { + RemoteClusterService remotes = + givenRemoteClusters(Map.of("", new String[] {"logs"}, "dead", new String[] {"logs-*"})); + mockNodeClientIndicesMappings("logs", ""); + Client remote = mock(Client.class, RETURNS_DEEP_STUBS); + when(nodeClient.getRemoteClusterClient("dead")).thenReturn(remote); + when(remote + .admin() + .indices() + .prepareGetMappings(any(String[].class)) + .setLocal(true) + .get(any(TimeValue.class))) + .thenThrow(new IllegalStateException("Unable to open any proxy connections")); + when(remotes.isSkipUnavailable("dead")).thenReturn(true); + + assertEquals(Set.of("logs"), client.getIndexMappings("logs", "dead:logs-*").keySet()); + } + + /** A name repeated in the expression is read as usual, as on the local-only path. */ + @Test + void get_index_mappings_accepts_a_repeated_local_name_with_remote_names() { + givenRemoteClusters(Map.of("", new String[] {"logs", "logs"}, "cape", new String[] {"logs-*"})); + MappingMetadata logs = mock(MappingMetadata.class); + when(logs.getSourceAsMap()).thenReturn(Map.of()); + // Matches any number of names, so the repeated "logs" reaches GetMappings as given. + when(nodeClient + .admin() + .indices() + .prepareGetMappings(any(String[].class)) + .setLocal(anyBoolean()) + .get() + .mappings()) + .thenReturn(Map.of("logs", logs)); + givenRemoteIndex("cape", "logs-1", 10000); + + assertEquals( + Set.of("logs", "cape:logs-1"), + client.getIndexMappings("logs", "logs", "cape:logs-*").keySet()); + } + + /** With no local indices, every remote skipped leaves no fields to query: an error. */ + @Test + void get_index_mappings_fails_when_every_remote_is_skipped_and_no_local_index_is_named() { + RemoteClusterService remotes = givenRemoteClusters(Map.of("dead", new String[] {"logs-*"})); + Client remote = mock(Client.class, RETURNS_DEEP_STUBS); + when(nodeClient.getRemoteClusterClient("dead")).thenReturn(remote); + when(remote + .admin() + .indices() + .prepareGetMappings(any(String[].class)) + .setLocal(true) + .get(any(TimeValue.class))) + .thenThrow(new IllegalStateException("Unable to open any proxy connections")); + when(remotes.isSkipUnavailable("dead")).thenReturn(true); + + ErrorReport report = + assertThrows(ErrorReport.class, () -> client.getIndexMappings("dead:logs-*")); + assertEquals( + "Remote cluster [dead] is unavailable and was skipped (skip_unavailable is true)", + report.getCause().getMessage()); + } + + /** A ':' whose prefix is not a registered cluster, as in date math, stays a local name. */ + @Test + void get_index_mappings_keeps_an_unregistered_prefix_local() { + String dateMath = ""; + givenRemoteClusters(Map.of("", new String[] {dateMath})); + mockNodeClientIndicesMappings("logs-2026.10.08", ""); + + assertEquals(Set.of("logs-2026.10.08"), client.getIndexMappings(dateMath).keySet()); + verify(nodeClient, never()).getRemoteClusterClient(any()); + } + + @AfterEach + void resetRemoteCache() { + RemoteMappingsReader.clear(); + } + + private RemoteClusterService givenRemoteClusters(Map indicesByCluster) { + RemoteClusterService remotes = mock(RemoteClusterService.class); + Map groups = new java.util.LinkedHashMap<>(); + indicesByCluster.forEach( + (alias, indices) -> + groups.put(alias, new OriginalIndices(indices, SearchRequest.DEFAULT_INDICES_OPTIONS))); + when(remotes.groupIndices(eq(SearchRequest.DEFAULT_INDICES_OPTIONS), any(), any())) + .thenReturn(groups); + remoteClusters.set(remotes); + return remotes; + } + + private void givenRemoteIndex(String alias, String index, int maxResultWindow) { + Client remote = mock(Client.class, RETURNS_DEEP_STUBS); + when(nodeClient.getRemoteClusterClient(alias)).thenReturn(remote); + when(remote + .admin() + .indices() + .prepareGetMappings(any(String[].class)) + .setLocal(true) + .get(any(TimeValue.class))) + .thenReturn( + new GetMappingsResponse( + Map.of( + index, + new MappingMetadata( + "_doc", Map.of("properties", Map.of("f", Map.of("type", "keyword"))))))); + when(remote.admin().indices().getSettings(any()).actionGet(any(TimeValue.class))) + .thenReturn( + new GetSettingsResponse( + Map.of( + index, + Settings.builder().put("index.max_result_window", maxResultWindow).build()), + Map.of())); + } + public void mockNodeClientIndicesMappings(String indexName, String mappings) { GetMappingsResponse mockResponse = mock(GetMappingsResponse.class); MappingMetadata emptyMapping = mock(MappingMetadata.class); diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/client/RemoteMappingsReaderTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/client/RemoteMappingsReaderTest.java new file mode 100644 index 00000000000..7d1a262f0a0 --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/client/RemoteMappingsReaderTest.java @@ -0,0 +1,296 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.LinkedHashMap; +import java.util.Map; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.opensearch.OpenSearchSecurityException; +import org.opensearch.action.admin.indices.mapping.get.GetMappingsResponse; +import org.opensearch.action.admin.indices.settings.get.GetSettingsResponse; +import org.opensearch.cluster.metadata.MappingMetadata; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.index.IndexNotFoundException; +import org.opensearch.sql.common.error.ErrorCode; +import org.opensearch.sql.common.error.ErrorReport; +import org.opensearch.transport.RemoteClusterService; +import org.opensearch.transport.client.Client; +import org.opensearch.transport.client.node.NodeClient; + +class RemoteMappingsReaderTest { + + private static final TimeValue TIMEOUT = TimeValue.timeValueSeconds(10); + + private final NodeClient client = mock(NodeClient.class); + private final RemoteClusterService remotes = mock(RemoteClusterService.class); + private final Map remoteClients = new LinkedHashMap<>(); + + @BeforeEach + void setUp() { + RemoteMappingsReader.clear(); + } + + @Test + void readsEachClusterAndPrefixesItsIndexNames() { + givenRemote("a", Map.of("logs-1", 10000)); + givenRemote("b", Map.of("logs-2", 10000)); + + RemoteMappingsReader.RemoteMetadata metadata = read("alice", "a:logs-*", "b:logs-*"); + + assertEquals(java.util.Set.of("a:logs-1", "b:logs-2"), metadata.mappings().keySet()); + assertEquals( + "keyword", + metadata + .mappings() + .get("a:logs-1") + .getFieldMappings() + .get("f") + .legacyTypeName() + .toLowerCase(java.util.Locale.ROOT)); + } + + /** Rollover indices share a mapping: it is parsed once and weighed once in the cache. */ + @Test + void indicesWithTheSameMappingShareOneParsedCopy() { + givenRemote("a", Map.of("logs-1", 10000, "logs-2", 10000, "logs-3", 10000)); + + RemoteMappingsReader.RemoteMetadata metadata = read("alice", "a:logs-*"); + + assertSame(metadata.mappings().get("a:logs-1"), metadata.mappings().get("a:logs-2")); + assertSame(metadata.mappings().get("a:logs-1"), metadata.mappings().get("a:logs-3")); + assertEquals(1, RemoteMappingsReader.weigh(metadata)); + } + + /** Below the cache's total limit, a schema of many distinct mappings is still cached. */ + @Test + void largeSchemaWithManyDistinctMappingsIsCached() { + int indices = 30; + int fieldsEach = 1000; + Map mappings = new LinkedHashMap<>(); + Map settings = new LinkedHashMap<>(); + for (int i = 0; i < indices; i++) { + Map properties = new LinkedHashMap<>(); + for (int f = 0; f < fieldsEach; f++) { + properties.put("i" + i + "_f" + f, Map.of("type", "keyword")); + } + mappings.put("logs-" + i, new MappingMetadata("_doc", Map.of("properties", properties))); + settings.put("logs-" + i, Settings.EMPTY); + } + Client remote = remoteClient("a"); + when(remote + .admin() + .indices() + .prepareGetMappings(any(String[].class)) + .setLocal(true) + .get(TIMEOUT)) + .thenReturn(new GetMappingsResponse(mappings)); + when(remote.admin().indices().getSettings(any()).actionGet(TIMEOUT)) + .thenReturn(new GetSettingsResponse(settings, Map.of())); + + read("alice", "a:logs-*"); + read("alice", "a:logs-*"); + + verify(client, times(1)).getRemoteClusterClient("a"); + } + + @Test + void readsTheRemoteMaxResultWindow() { + givenRemote("a", Map.of("logs-1", 50000)); + + assertEquals(50000, read("alice", "a:logs-1").windows().get("a:logs-1")); + } + + @Test + void secondReadBySameUserIsCached() { + givenRemote("a", Map.of("logs-1", 10000)); + + RemoteMappingsReader.RemoteMetadata first = read("alice", "a:logs-1"); + RemoteMappingsReader.RemoteMetadata second = read("alice", "a:logs-1"); + + assertSame(first, second); + verify(client, times(1)).getRemoteClusterClient("a"); + } + + /** Field-level security is per user, so users never share an entry. */ + @Test + void usersDoNotShareEntries() { + givenRemote("a", Map.of("logs-1", 10000)); + + read("alice", "a:logs-1"); + read("bob", "a:logs-1"); + + verify(client, times(2)).getRemoteClusterClient("a"); + } + + @Test + void skipsAnUnreachableClusterWhenSkipUnavailable() { + givenRemote("a", Map.of("logs-1", 10000)); + givenUnreachable("b", true); + + RemoteMappingsReader.RemoteMetadata metadata = read("alice", "a:logs-*", "b:logs-*"); + + assertEquals(java.util.Set.of("a:logs-1"), metadata.mappings().keySet()); + } + + /** An answer missing a skipped cluster is an outage answer, so it is never cached. */ + @Test + void partialAnswerIsNotCached() { + givenRemote("a", Map.of("logs-1", 10000)); + givenUnreachable("b", true); + + read("alice", "a:logs-*", "b:logs-*"); + read("alice", "a:logs-*", "b:logs-*"); + + verify(client, times(2)).getRemoteClusterClient("a"); + } + + @Test + void failsOnAnUnreachableClusterWithoutSkipUnavailable() { + givenRemote("a", Map.of("logs-1", 10000)); + givenUnreachable("b", false); + + IllegalStateException e = + assertThrows(IllegalStateException.class, () -> read("alice", "a:logs-*", "b:logs-*")); + assertTrue(e.getMessage().contains("Unable to open any proxy connections")); + } + + /** Every cluster skipped: no remote fields, signalled for the caller to decide. */ + @Test + void everyClusterSkippedIsSignalledWithItsMessage() { + givenUnreachable("b", true); + + RemoteMappingsReader.AllClustersSkipped e = + assertThrows( + RemoteMappingsReader.AllClustersSkipped.class, () -> read("alice", "b:logs-*")); + assertEquals( + "Remote cluster [b] is unavailable and was skipped (skip_unavailable is true)", + e.getMessage()); + } + + @Test + void messageNamesSeveralSkippedClustersInOrder() { + assertEquals( + "Remote clusters [a, b, c] are unavailable and were skipped (skip_unavailable is true)", + RemoteMappingsReader.skippedMessage(java.util.List.of("c", "a", "b"))); + } + + /** Many skipped clusters: the message names five and counts the rest. */ + @Test + void messageNamesFiveClustersAndCountsTheRest() { + assertEquals( + "Remote clusters [a, b, c, d, e] and 2 more are unavailable and were skipped" + + " (skip_unavailable is true)", + RemoteMappingsReader.skippedMessage(java.util.List.of("g", "f", "e", "d", "c", "b", "a"))); + } + + /** A missing index is the remote's answer, not an outage: skip_unavailable does not hide it. */ + @Test + void missingIndexIsReportedEvenWithSkipUnavailable() { + givenFailure("a", new IndexNotFoundException("nope")); + when(remotes.isSkipUnavailable("a")).thenReturn(true); + + ErrorReport report = assertThrows(ErrorReport.class, () -> read("alice", "a:nope")); + assertEquals(ErrorCode.INDEX_NOT_FOUND, report.getCode()); + } + + @Test + void deniedReadIsAPermissionError() { + givenFailure("a", new OpenSearchSecurityException("no permissions", RestStatus.FORBIDDEN)); + + ErrorReport report = assertThrows(ErrorReport.class, () -> read("alice", "a:logs-1")); + assertEquals(ErrorCode.PERMISSION_DENIED, report.getCode()); + } + + @Test + void patternMatchingNothingIsIndexNotFound() { + givenRemote("a", Map.of()); + + ErrorReport report = assertThrows(ErrorReport.class, () -> read("alice", "a:none-*")); + assertEquals(ErrorCode.INDEX_NOT_FOUND, report.getCode()); + } + + /** Reads "cluster:index" names, grouped by cluster as the node client passes them. */ + private RemoteMappingsReader.RemoteMetadata read(String user, String... names) { + Map> byCluster = new LinkedHashMap<>(); + for (String name : names) { + int separator = name.indexOf(':'); + byCluster + .computeIfAbsent(name.substring(0, separator), k -> new java.util.ArrayList<>()) + .add(name.substring(separator + 1)); + } + Map indicesByCluster = new LinkedHashMap<>(); + byCluster.forEach( + (cluster, indices) -> indicesByCluster.put(cluster, indices.toArray(String[]::new))); + return new RemoteMappingsReader(client, remotes, TIMEOUT).read(user, indicesByCluster); + } + + /** A reachable cluster whose indices each have one keyword field "f". */ + private void givenRemote(String alias, Map windowsByIndex) { + Map mappings = new LinkedHashMap<>(); + Map settings = new LinkedHashMap<>(); + windowsByIndex.forEach( + (index, window) -> { + mappings.put( + index, + new MappingMetadata( + "_doc", Map.of("properties", Map.of("f", Map.of("type", "keyword"))))); + settings.put(index, Settings.builder().put("index.max_result_window", window).build()); + }); + Client remote = remoteClient(alias); + when(remote + .admin() + .indices() + .prepareGetMappings(any(String[].class)) + .setLocal(true) + .get(TIMEOUT)) + .thenReturn(new GetMappingsResponse(mappings)); + when(remote.admin().indices().getSettings(any()).actionGet(TIMEOUT)) + .thenReturn(new GetSettingsResponse(settings, Map.of())); + } + + private void givenUnreachable(String alias, boolean skipUnavailable) { + givenFailure( + alias, + new IllegalStateException( + "Unable to open any proxy connections to remote cluster [" + alias + "]")); + when(remotes.isSkipUnavailable(alias)).thenReturn(skipUnavailable); + } + + private void givenFailure(String alias, RuntimeException failure) { + Client remote = remoteClient(alias); + when(remote + .admin() + .indices() + .prepareGetMappings(any(String[].class)) + .setLocal(true) + .get(TIMEOUT)) + .thenThrow(failure); + } + + private Client remoteClient(String alias) { + return remoteClients.computeIfAbsent( + alias, + a -> { + Client remote = mock(Client.class, RETURNS_DEEP_STUBS); + when(client.getRemoteClusterClient(a)).thenReturn(remote); + return remote; + }); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequestTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequestTest.java index 6e156e6d333..31733d616b5 100644 --- a/opensearch/src/test/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequestTest.java +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequestTest.java @@ -136,11 +136,12 @@ void testSearchWithLangSpec() { hasEntry("TYPE_NAME", stringValue("STRING")))); } + /** The client resolves "cluster:index" itself, so the name reaches it unchanged. */ @Test - void testCrossClusterShouldSearchLocal() { + void testCrossClusterNameReachesTheClientUnchanged() { when(mapping.getFieldMappings()) .thenReturn(Map.of("name", OpenSearchDataType.of(OpenSearchDataType.MappingType.Keyword))); - when(client.getIndexMappings("index")).thenReturn(ImmutableMap.of("test", mapping)); + when(client.getIndexMappings("ccs:index")).thenReturn(ImmutableMap.of("ccs:index", mapping)); final List results = new OpenSearchDescribeIndexRequest(client, "ccs:index").search(); diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/SQLPlugin.java b/plugin/src/main/java/org/opensearch/sql/plugin/SQLPlugin.java index 80d3c1c382f..53b4d47c02f 100644 --- a/plugin/src/main/java/org/opensearch/sql/plugin/SQLPlugin.java +++ b/plugin/src/main/java/org/opensearch/sql/plugin/SQLPlugin.java @@ -104,6 +104,7 @@ import org.opensearch.sql.legacy.plugin.RestSqlAction; import org.opensearch.sql.legacy.plugin.RestSqlStatsAction; import org.opensearch.sql.opensearch.client.OpenSearchNodeClient; +import org.opensearch.sql.opensearch.client.RemoteClusterServiceProvider; import org.opensearch.sql.opensearch.setting.OpenSearchSettings; import org.opensearch.sql.opensearch.storage.OpenSearchDataSourceFactory; import org.opensearch.sql.opensearch.storage.rest.CoreEndpointsProvider; @@ -200,6 +201,13 @@ private static void setDefaultIfAbsent(String key, String value) { private org.opensearch.sql.common.setting.Settings pluginSettings; private NodeClient client; + + /** + * Filled by {@link org.opensearch.sql.plugin.transport.TransportPPLQueryAction}, the first holder + * of a {@code TransportService}; returned as a component so it can be injected there. + */ + private final RemoteClusterServiceProvider remoteClusters = new RemoteClusterServiceProvider(); + private DataSourceServiceImpl dataSourceService; private OpenSearchAsyncQueryScheduler asyncQueryScheduler; private Injector injector; @@ -444,6 +452,7 @@ public Collection createComponents( b.bind(org.opensearch.sql.common.setting.Settings.class).toInstance(pluginSettings); b.bind(DataSourceService.class).toInstance(dataSourceService); b.bind(ClusterService.class).toInstance(clusterService); + b.bind(RemoteClusterServiceProvider.class).toInstance(remoteClusters); }); modules.add(new AsyncExecutorServiceModule()); modules.add(new DirectQueryModule()); @@ -487,7 +496,8 @@ public Collection createComponents( directQueryExecutorService, extensionsHolder, queryJobService, - securityAdapter); + securityAdapter, + remoteClusters); } @Override @@ -578,7 +588,7 @@ private DataSourceServiceImpl createDataSourceService() { new ImmutableSet.Builder() .add( new OpenSearchDataSourceFactory( - new OpenSearchNodeClient(this.client), pluginSettings)) + new OpenSearchNodeClient(this.client, remoteClusters), pluginSettings)) .add(new PrometheusStorageFactory(pluginSettings)) .add(new GlueDataSourceFactory(pluginSettings)) .add(new SecurityLakeDataSourceFactory(pluginSettings)) diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/config/OpenSearchPluginModule.java b/plugin/src/main/java/org/opensearch/sql/plugin/config/OpenSearchPluginModule.java index 594ac0dcfa9..4bd4384dbd4 100644 --- a/plugin/src/main/java/org/opensearch/sql/plugin/config/OpenSearchPluginModule.java +++ b/plugin/src/main/java/org/opensearch/sql/plugin/config/OpenSearchPluginModule.java @@ -34,6 +34,7 @@ import org.opensearch.sql.monitor.ResourceMonitor; import org.opensearch.sql.opensearch.client.OpenSearchClient; import org.opensearch.sql.opensearch.client.OpenSearchNodeClient; +import org.opensearch.sql.opensearch.client.RemoteClusterServiceProvider; import org.opensearch.sql.opensearch.executor.OpenSearchExecutionEngine; import org.opensearch.sql.opensearch.executor.OpenSearchQueryManager; import org.opensearch.sql.opensearch.executor.ThreadPoolExecutionDispatcher; @@ -71,8 +72,9 @@ public OpenSearchPluginModule() { protected void configure() {} @Provides - public OpenSearchClient openSearchClient(NodeClient nodeClient) { - return new OpenSearchNodeClient(nodeClient); + public OpenSearchClient openSearchClient( + NodeClient nodeClient, RemoteClusterServiceProvider remoteClusters) { + return new OpenSearchNodeClient(nodeClient, remoteClusters); } @Provides diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java b/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java index d5036da5023..ade37892e6c 100644 --- a/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java +++ b/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java @@ -51,6 +51,7 @@ import org.opensearch.sql.legacy.metrics.Metrics; import org.opensearch.sql.monitor.profile.ProfileScope; import org.opensearch.sql.monitor.profile.QueryProfiling; +import org.opensearch.sql.opensearch.client.RemoteClusterServiceProvider; import org.opensearch.sql.opensearch.executor.OpenSearchQueryManager; import org.opensearch.sql.opensearch.executor.tracing.TracingPhaseListener; import org.opensearch.sql.opensearch.setting.OpenSearchSettings; @@ -114,12 +115,15 @@ public TransportPPLQueryAction( EngineExtensionsHolder extensionsHolder, Tracer tracer, OpenSearchQueryJobService queryJobService, - OpenSearchSecurityAdapter securityAdapter) { + OpenSearchSecurityAdapter securityAdapter, + RemoteClusterServiceProvider remoteClusters) { super(PPLQueryAction.NAME, transportService, actionFilters, TransportPPLQueryRequest::new); this.clientRef = client; this.clusterServiceRef = clusterService; this.queryJobService = queryJobService; this.securityAdapter = securityAdapter; + // The plugin's components are created before any TransportService exists; fill it here. + remoteClusters.set(transportService.getRemoteClusterService()); ModulesBuilder modules = new ModulesBuilder(); modules.add(new OpenSearchPluginModule(extensionsHolder.engines(), tracer)); @@ -129,6 +133,7 @@ public TransportPPLQueryAction( modules.add( b -> { b.bind(NodeClient.class).toInstance(client); + b.bind(RemoteClusterServiceProvider.class).toInstance(remoteClusters); b.bind(org.opensearch.sql.common.setting.Settings.class).toInstance(pluginSettings); b.bind(DataSourceService.class).toInstance(dataSourceService); b.bind(ClusterService.class).toInstance(clusterService);