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);