diff --git a/common/src/main/java/org/opensearch/sql/common/setting/Settings.java b/common/src/main/java/org/opensearch/sql/common/setting/Settings.java index bf3e65d8741..6ca131d9d13 100644 --- a/common/src/main/java/org/opensearch/sql/common/setting/Settings.java +++ b/common/src/main/java/org/opensearch/sql/common/setting/Settings.java @@ -36,6 +36,7 @@ public enum Key { PPL_SYNTAX_LEGACY_PREFERRED("plugins.ppl.syntax.legacy.preferred"), PPL_SUBSEARCH_MAXOUT("plugins.ppl.subsearch.maxout"), PPL_JOIN_SUBSEARCH_MAXOUT("plugins.ppl.join.subsearch_maxout"), + PPL_REST_ALLOWED_ENDPOINTS("plugins.ppl.rest.allowed_endpoints"), /** Enable Calcite as execution engine */ CALCITE_ENGINE_ENABLED("plugins.calcite.enabled"), diff --git a/core/src/main/java/org/opensearch/sql/ast/tree/RestRelation.java b/core/src/main/java/org/opensearch/sql/ast/tree/RestRelation.java new file mode 100644 index 00000000000..d0a0e34d3c9 --- /dev/null +++ b/core/src/main/java/org/opensearch/sql/ast/tree/RestRelation.java @@ -0,0 +1,25 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.ast.tree; + +import java.util.Collections; +import lombok.EqualsAndHashCode; +import lombok.ToString; +import org.opensearch.sql.ast.expression.UnresolvedExpression; + +/** + * Extend Relation to mark a {@code rest} leading command. The single table name is a reserved, + * encoded token (produced by {@link org.opensearch.sql.utils.SystemIndexUtils#restTable}) that + * carries the validated REST endpoint spec; it resolves through the storage engine to a REST source + * table on the Calcite path, exactly as {@link DescribeRelation} resolves to a system index. + */ +@ToString +@EqualsAndHashCode(callSuper = false) +public class RestRelation extends Relation { + public RestRelation(UnresolvedExpression tableName) { + super(Collections.singletonList(tableName)); + } +} diff --git a/core/src/main/java/org/opensearch/sql/utils/SystemIndexUtils.java b/core/src/main/java/org/opensearch/sql/utils/SystemIndexUtils.java index c9fa35d7068..7589cf522f6 100644 --- a/core/src/main/java/org/opensearch/sql/utils/SystemIndexUtils.java +++ b/core/src/main/java/org/opensearch/sql/utils/SystemIndexUtils.java @@ -5,6 +5,9 @@ package org.opensearch.sql.utils; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.Map; import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.experimental.UtilityClass; @@ -34,6 +37,121 @@ public static Boolean isSystemIndex(String indexName) { return indexName.endsWith(SYS_TABLES_SUFFIX); } + /** + * Reserved suffix marking a {@code rest} source table. Distinct from {@link #SYS_TABLES_SUFFIX} + * so {@link #isSystemIndex} and {@link #isRestSource} never overlap. The whole reserved name is a + * single Calcite identifier (REST + lowercase hex + this suffix), so it survives name resolution + * the same way the uppercase system-mapping names do. + */ + private static final String REST_SOURCE_SUFFIX = "__REST_SOURCE"; + + private static final String REST_SOURCE_PREFIX = "REST"; + + /** True if the resolved table name is a {@code rest} source token. */ + public static boolean isRestSource(String indexName) { + return indexName.endsWith(REST_SOURCE_SUFFIX); + } + + /** + * Encode a validated {@link RestSpec} into a single reserved table name. Mirrors {@link + * #mappingTable}: structured metadata travels inside a reserved name rather than a side channel. + * The endpoint/args have already been allow-list-validated before this is called. + */ + public static String restTable(RestSpec spec) { + StringBuilder sb = new StringBuilder(); + sb.append("endpoint=").append(spec.getEndpoint()); + if (spec.getCount() != null) { + sb.append('\n').append("count=").append(spec.getCount()); + } + if (spec.getTimeout() != null) { + sb.append('\n').append("timeout=").append(spec.getTimeout()); + } + if (spec.getArgs() != null) { + for (Map.Entry e : spec.getArgs().entrySet()) { + sb.append('\n').append("arg.").append(e.getKey()).append('=').append(e.getValue()); + } + } + return REST_SOURCE_PREFIX + toHex(sb.toString()) + REST_SOURCE_SUFFIX; + } + + /** Decode a reserved {@code rest} table name back into its {@link RestSpec}. */ + public static RestSpec decodeRestSpec(String indexName) { + // Validate the token shape before slicing it: callers gate on isRestSource today, but a + // public decoder must not assume its precondition, otherwise a malformed token would throw an + // opaque StringIndexOutOfBoundsException from substring rather than a clear input error. + if (!isRestSource(indexName)) { + throw new IllegalArgumentException("not a valid rest source token: " + indexName); + } + String body = + indexName.substring( + REST_SOURCE_PREFIX.length(), indexName.length() - REST_SOURCE_SUFFIX.length()); + String decoded = fromHex(body); + String endpoint = null; + Integer count = null; + String timeout = null; + LinkedHashMap args = new LinkedHashMap<>(); + for (String line : decoded.split("\n")) { + if (line.isEmpty()) { + continue; + } + int eq = line.indexOf('='); + if (eq < 0) { + continue; + } + String k = line.substring(0, eq); + String v = line.substring(eq + 1); + if (k.equals("endpoint")) { + endpoint = v; + } else if (k.equals("count")) { + count = Integer.parseInt(v); + } else if (k.equals("timeout")) { + timeout = v; + } else if (k.startsWith("arg.")) { + args.put(k.substring("arg.".length()), v); + } + } + if (endpoint == null) { + throw new IllegalArgumentException("rest source token is missing the endpoint: " + indexName); + } + return new RestSpec(endpoint, args, count, timeout); + } + + private static String toHex(String s) { + StringBuilder h = new StringBuilder(); + for (byte b : s.getBytes(StandardCharsets.UTF_8)) { + h.append(Character.forDigit((b >> 4) & 0xF, 16)).append(Character.forDigit(b & 0xF, 16)); + } + return h.toString(); + } + + private static String fromHex(String h) { + if (h.length() % 2 != 0) { + throw new IllegalArgumentException("not a valid rest source token: odd-length hex body"); + } + byte[] bytes = new byte[h.length() / 2]; + for (int i = 0; i < bytes.length; i++) { + bytes[i] = + (byte) + ((Character.digit(h.charAt(2 * i), 16) << 4) + + Character.digit(h.charAt(2 * i + 1), 16)); + } + return new String(bytes, StandardCharsets.UTF_8); + } + + /** + * The validated spec for a {@code rest} command: an allow-listed read-only endpoint plus optional + * count/timeout/query args. Lives in core so the parser (encode) and the storage engine (decode) + * share it without a cross-module dependency. + */ + @Getter + @RequiredArgsConstructor + public static class RestSpec { + private final String endpoint; + private final Map args; + private final Integer count; + private final String timeout; + } + /** * Compose system mapping table. * diff --git a/docs/category.json b/docs/category.json index 7296b089650..a8665b82eac 100644 --- a/docs/category.json +++ b/docs/category.json @@ -35,6 +35,7 @@ "user/ppl/cmd/rename.md", "user/ppl/cmd/multisearch.md", "user/ppl/cmd/replace.md", + "user/ppl/cmd/rest.md", "user/ppl/cmd/rex.md", "user/ppl/cmd/search.md", "user/ppl/cmd/showdatasources.md", diff --git a/docs/user/ppl/cmd/rest.md b/docs/user/ppl/cmd/rest.md new file mode 100644 index 00000000000..fdbc9fc7935 --- /dev/null +++ b/docs/user/ppl/cmd/rest.md @@ -0,0 +1,60 @@ +# rest + +The `rest` command is a leading command that reads an allow-listed, read-only in-cluster management endpoint and emits the response as PPL rows. Its rows come from the endpoint dispatch, not from an index, so `rest` appears at the start of a query. + +> **Note**: The `rest` command is supported only on the Calcite query engine (`plugins.calcite.enabled=true`). Each endpoint has a fixed output schema, and the dispatch runs under the caller's security context, so a user who cannot call an endpoint directly cannot call it through `rest`. The command is read-only; mutating and non-allow-listed endpoints are rejected. Each endpoint requires the same cluster-monitor privilege as calling it natively, so `rest` grants no extra access. + +The `rest` command is a generic, extensible framework: a plugin contributes additional read-only endpoints through the `RestEndpointProvider` extension point without changing the grammar. This first version ships a single built-in endpoint, `/_cluster/health`. Additional endpoints (for example `/_cat/nodes`, `/_cat/shards`, `/_cluster/state`, `/_cluster/settings`) can be added in follow-ups, together with optional response redaction applied centrally at the endpoint choke point. + +## Enabling the command + +`/_cluster/health` is **enabled by default**: `plugins.ppl.rest.allowed_endpoints` defaults to `["/_cluster/health"]`. Any other endpoint is rejected until a deployment adds it to the allow-list (a node-level setting, applied at node startup and not changeable at runtime): + +```yaml +plugins.ppl.rest.allowed_endpoints: ["/_cluster/health"] +``` + +Use `["*"]` to allow every endpoint in the curated list below. Set an empty list to disable the command entirely. + +## Syntax + +```syntax +rest [count=] [= ...] +``` + +## Parameters + +| Parameter | Required/Optional | Description | +| --- | --- | --- | +| `` | Required | An allow-listed, read-only endpoint path (see the allow-list below), for example `/_cluster/health`. | +| `count=` | Optional | Caps the number of emitted rows. | +| `=` | Optional | Endpoint query arguments, validated per endpoint by both key and value (for example `local=true` for `/_cluster/health`). | + +## Allow-list + +`rest` resolves only an explicit, curated set of read-only endpoints. Anything outside the list, including any mutating endpoint, is rejected with a clear error. + +| Endpoint | Output columns | Accepted args | +| --- | --- | --- | +| `/_cluster/health` | `cluster_name` (string), `status` (string), `number_of_nodes` (integer), `number_of_data_nodes` (integer), `active_primary_shards` (integer), `active_shards` (integer), `relocating_shards` (integer), `initializing_shards` (integer), `unassigned_shards` (integer), `timed_out` (boolean) | `local` | + +## Example: Counting the nodes in the cluster + +The following query reads cluster health and projects a column that is deterministic on a single-node cluster: + +```ppl +| rest '/_cluster/health' | fields number_of_nodes +``` + +The query returns the following results: + +```text +fetched rows / total rows = 1/1 ++-----------------+ +| number_of_nodes | +|-----------------| +| 1 | ++-----------------+ +``` + +`/_cluster/health` also exposes `status`, `active_shards`, and the other columns listed in the allow-list. The `rest` row source composes with downstream `where`, `sort`, `stats`, and `fields` exactly like an index scan, for example `| rest '/_cluster/health' | where status = 'green' | fields status, active_shards`. diff --git a/docs/user/ppl/index.md b/docs/user/ppl/index.md index 647c4568446..3e5a08d990d 100644 --- a/docs/user/ppl/index.md +++ b/docs/user/ppl/index.md @@ -75,6 +75,7 @@ source=accounts | [lookup command](cmd/lookup.md) | 3.0 | experimental (since 3.0) | Add or replace data from a lookup index. | | [multisearch command](cmd/multisearch.md) | 3.4 | experimental (since 3.4) | Execute multiple search queries and combine their results. | | [union command](cmd/union.md) | 3.7 | experimental (since 3.7) | Combine results from multiple datasets using UNION ALL semantics. | +| [rest command](cmd/rest.md) | 3.9 | experimental (since 3.9) | Read an allow-listed, read-only in-cluster management endpoint (cluster/cat/nodes) as rows. Calcite engine only. | | [ml command](cmd/ml.md) | 2.5 | stable (since 2.5) | Apply machine learning algorithms to analyze data. | | [kmeans command](cmd/kmeans.md) | 1.3 | stable (since 1.3) | Apply the kmeans algorithm on the search result returned by a PPL command. | | [ad command](cmd/ad.md) | 1.3 | deprecated (since 2.5) | Apply Random Cut Forest algorithm on the search result returned by a PPL command. | diff --git a/doctest/build.gradle b/doctest/build.gradle index cce64170f56..45f4b768aef 100644 --- a/doctest/build.gradle +++ b/doctest/build.gradle @@ -205,6 +205,8 @@ testClusters { plugin(getJobSchedulerPlugin()) plugin ':opensearch-sql-plugin' testDistribution = 'archive' + // Only /_cluster/health is registered; pin the allow-list to it so the rest.md examples run. + setting 'plugins.ppl.rest.allowed_endpoints', '/_cluster/health' } } tasks.register("runRestTestCluster", RunTask) { diff --git a/integ-test/build.gradle b/integ-test/build.gradle index 1435d1d499d..29627ff7b60 100644 --- a/integ-test/build.gradle +++ b/integ-test/build.gradle @@ -387,6 +387,8 @@ testClusters { plugin(getGeoSpatialPlugin()) plugin ":opensearch-sql-plugin" setting "plugins.query.datasources.encryption.masterkey", "1234567812345678" + // Only /_cluster/health is registered; pin the allow-list to it for the rest ITs. + setting 'plugins.ppl.rest.allowed_endpoints', '/_cluster/health' } yamlRestTest { testDistribution = 'archive' @@ -405,6 +407,8 @@ testClusters { testDistribution = 'archive' plugin(getJobSchedulerPlugin()) plugin ":opensearch-sql-plugin" + // Only /_cluster/health is registered; pin the allow-list to it for RestCommandSecurityIT. + setting 'plugins.ppl.rest.allowed_endpoints', '/_cluster/health' } remoteIntegTestWithSecurity { testDistribution = 'archive' @@ -419,6 +423,8 @@ testClusters { plugin(getArrowFlightRpcPlugin()) plugin(getAnalyticsEnginePlugin()) plugin ":opensearch-sql-plugin" + // Composite-default cluster: PPL queries route to the analytics engine unless excluded. + setting 'cluster.pluggable.dataformat', 'composite' } } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java index 6d0c88bf773..e7c024e49fa 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java @@ -59,6 +59,7 @@ CalciteObjectFieldOperateIT.class, CalciteOperatorIT.class, CalciteParseCommandIT.class, + CalcitePPLRestIT.class, CalcitePPLAggregationIT.class, CalcitePPLAppendcolIT.class, CalcitePPLAppendCommandIT.class, diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java index 04dc2b0e74b..858f4cb5066 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java @@ -66,6 +66,15 @@ public void init() throws Exception { loadIndex(Index.GRAPH_EMPLOYEES); } + // Only for Calcite: the rest row source explains as a CalciteScannableCatalogScan. + @Test + public void explainRestCommand() throws IOException { + String result = explainQueryToString("| rest '/_cluster/health' | fields status"); + Assert.assertTrue( + "Expected a rest scan node in the explain output, got: " + result, + result.contains("CatalogScan")); + } + @Override @Ignore("test only in v2") public void testExplainModeUnsupportedInV2() throws IOException {} diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLRestIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLRestIT.java new file mode 100644 index 00000000000..e779e6c8c12 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLRestIT.java @@ -0,0 +1,94 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import static org.opensearch.sql.util.MatcherUtils.rows; +import static org.opensearch.sql.util.MatcherUtils.schema; +import static org.opensearch.sql.util.MatcherUtils.verifyDataRows; +import static org.opensearch.sql.util.MatcherUtils.verifySchema; + +import java.io.IOException; +import org.json.JSONObject; +import org.junit.jupiter.api.Test; +import org.opensearch.client.ResponseException; +import org.opensearch.sql.ppl.PPLIntegTestCase; + +/** + * Integration tests for the {@code rest} leading command on the Calcite path. This first version + * ships a single built-in endpoint, {@code /_cluster/health} (a deterministic single-row endpoint + * that carries no network identifiers and needs no redaction). These tests exercise it end to end + * and verify the allow-list and per-arg gates. + */ +public class CalcitePPLRestIT extends PPLIntegTestCase { + + @Override + public void init() throws Exception { + super.init(); + enableCalcite(); + } + + @Test + public void testRestClusterHealthSchema() throws IOException { + JSONObject result = executeQuery("| rest '/_cluster/health' | fields status, number_of_nodes"); + verifySchema(result, schema("status", "string"), schema("number_of_nodes", "int")); + } + + @Test + public void testRestClusterHealthDataRows() throws IOException { + // Single-node test cluster: exactly one node. + JSONObject result = executeQuery("| rest '/_cluster/health' | fields number_of_nodes"); + verifyDataRows(result, rows(1)); + } + + @Test + public void testRestClusterHealthComposesDownstream() throws IOException { + // The rest row source composes with downstream stats exactly like an index scan. + JSONObject result = executeQuery("| rest '/_cluster/health' | stats count() as cnt"); + verifyDataRows(result, rows(1)); + } + + @Test + public void testRestClusterHealthLocalArg() throws IOException { + // local=true reads health from the local node; on a single-node cluster the row is unchanged. + JSONObject result = + executeQuery("| rest '/_cluster/health' local='true' | fields number_of_nodes"); + verifyDataRows(result, rows(1)); + } + + @Test + public void testRestRejectsNonAllowListedEndpoint() { + // /_cat/nodes is not registered in this version; it is refused before any transport call. + assertRestBadRequest("| rest '/_cat/nodes'", "allow-list"); + } + + @Test + public void testRestRejectsEmptyEndpoint() { + assertRestBadRequest("| rest ''", "non-empty path"); + } + + @Test + public void testRestRejectsDisallowedArg() { + assertRestBadRequest("| rest '/_cluster/health' h='name'", "does not accept arg"); + } + + @Test + public void testRestRejectsNegativeCount() { + assertRestBadRequest("| rest '/_cluster/health' count=-1", "non-negative"); + } + + /** + * Assert a {@code rest} query is refused as a client error: HTTP 400 (not a 500 system error) + * with the given substring in the response body. Covers allow-list and bad-argument rejection. + */ + private void assertRestBadRequest(String query, String expectedSubstring) { + ResponseException e = + org.junit.Assert.assertThrows(ResponseException.class, () -> executeQuery(query)); + org.junit.Assert.assertEquals(400, e.getResponse().getStatusLine().getStatusCode()); + org.junit.Assert.assertTrue( + "expected [" + expectedSubstring + "] in response body: " + e.getMessage(), + e.getMessage().contains(expectedSubstring)); + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/plugin/AnalyticsEngineCompatIT.java b/integ-test/src/test/java/org/opensearch/sql/plugin/AnalyticsEngineCompatIT.java index f6ec903c395..b5c08cc8ad0 100644 --- a/integ-test/src/test/java/org/opensearch/sql/plugin/AnalyticsEngineCompatIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/plugin/AnalyticsEngineCompatIT.java @@ -5,9 +5,13 @@ package org.opensearch.sql.plugin; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assume.assumeTrue; import java.io.IOException; +import org.json.JSONArray; +import org.json.JSONObject; import org.junit.Before; import org.junit.Test; import org.opensearch.client.Request; @@ -56,4 +60,31 @@ public void testClusterStarted() { // If the cluster booted with analytics-engine present, all plugins loaded without classloader // errors. The assumption above guarantees we only assert this where it is meaningful. } + + /** + * The {@code rest} row source is a Calcite Enumerable/Scannable scan with no backing index, so it + * is never routed to the analytics (DataFusion) engine. This pins that {@code rest} returns its + * fixed schema and correct data unchanged when the analytics-engine plugin is present. + */ + @Test + public void testRestCommandUnaffectedByAnalyticsEngine() throws IOException { + Request request = new Request("POST", "/_plugins/_ppl"); + request.setJsonEntity( + "{\"query\": \"| rest '/_cluster/health' | fields status, number_of_nodes\"}"); + Response response = client().performRequest(request); + assertEquals(200, response.getStatusLine().getStatusCode()); + + JSONObject result = new JSONObject(TestUtils.getResponseBody(response, true)); + + JSONArray schema = result.getJSONArray("schema"); + assertEquals(2, schema.length()); + assertEquals("status", schema.getJSONObject(0).getString("name")); + assertEquals("string", schema.getJSONObject(0).getString("type")); + assertEquals("number_of_nodes", schema.getJSONObject(1).getString("name")); + assertEquals("int", schema.getJSONObject(1).getString("type")); + + JSONArray datarows = result.getJSONArray("datarows"); + assertEquals(1, datarows.length()); + assertTrue(datarows.getJSONArray(0).getInt(1) >= 1); + } } diff --git a/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java b/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java index a32dd9fb990..011d3ed0141 100644 --- a/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java @@ -33,6 +33,19 @@ public void init() throws Exception { loadIndex(Index.GRAPH_EMPLOYEES); } + @Test + public void testRest() throws IOException { + JSONObject result; + try { + result = executeQuery("| rest '/_cluster/health' | fields status, number_of_nodes"); + } catch (ResponseException e) { + result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + } + if (isCalciteEnabled()) { + assertFalse(result.getJSONArray("datarows").isEmpty()); + } + } + @Test public void testJoin() throws IOException { JSONObject result; diff --git a/integ-test/src/test/java/org/opensearch/sql/security/RestCommandSecurityIT.java b/integ-test/src/test/java/org/opensearch/sql/security/RestCommandSecurityIT.java new file mode 100644 index 00000000000..091846097ca --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/security/RestCommandSecurityIT.java @@ -0,0 +1,90 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.security; + +import static org.opensearch.sql.util.MatcherUtils.columnName; +import static org.opensearch.sql.util.MatcherUtils.verifyColumn; + +import java.io.IOException; +import org.json.JSONObject; +import org.junit.Test; +import org.opensearch.client.ResponseException; +import org.opensearch.sql.legacy.TestUtils; + +/** + * Integration tests that the rest command is subject to the security plugin fine grained access + * control. The command dispatches a standard transport action under the caller identity, so the + * security ActionFilter authorizes it by action name. This version ships only {@code + * /_cluster/health}, which requires the {@code cluster:monitor/health} privilege: a caller holding + * cluster monitor can run it, a caller without it is denied. The command therefore grants no access + * beyond calling the endpoint natively. + */ +public class RestCommandSecurityIT extends SecurityTestBase { + + private static final String MONITOR_USER = "rest_monitor_user"; + private static final String MONITOR_ROLE = "rest_monitor_role"; + + private static final String NO_MONITOR_USER = "rest_no_monitor_user"; + private static final String NO_MONITOR_ROLE = "rest_no_monitor_role"; + + @Override + protected void init() throws Exception { + super.init(); + setupRolesAndUsers(); + enableCalcite(); + // rest is Calcite only, so a V2 fallback would replace the security denial with an unsupported + // command error. Disable fallback so the denial reason surfaces to the caller. + disallowCalciteFallback(); + } + + private void setupRolesAndUsers() throws IOException { + createRoleWithPermissions( + MONITOR_ROLE, + "*", + new String[] {"cluster:admin/opensearch/ppl", "cluster:monitor/health"}, + new String[] {}); + createUser(MONITOR_USER, MONITOR_ROLE); + + createRoleWithPermissions( + NO_MONITOR_ROLE, "*", new String[] {"cluster:admin/opensearch/ppl"}, new String[] {}); + createUser(NO_MONITOR_USER, NO_MONITOR_ROLE); + } + + @Test + public void monitorUserCanRunClusterHealth() throws IOException { + JSONObject result = + executeQueryAsUser("| rest '/_cluster/health' | fields status", MONITOR_USER); + verifyColumn(result, columnName("status")); + } + + @Test + public void userWithoutClusterMonitorCannotRunClusterHealth() throws IOException { + assertDenied( + "| rest '/_cluster/health' | fields status", NO_MONITOR_USER, "cluster:monitor/health"); + } + + /** + * Asserts the query is rejected for a caller lacking the privilege. A denied transport action on + * the Calcite only rest path surfaces as a client or server error whose body carries the security + * denial reason, so this checks the denial signal rather than a fixed status code. + */ + private void assertDenied(String query, String user, String deniedAction) throws IOException { + try { + executeQueryAsUser(query, user); + fail("Expected a permission denial for user without privilege: " + user); + } catch (ResponseException e) { + int status = e.getResponse().getStatusLine().getStatusCode(); + String body = TestUtils.getResponseBody(e.getResponse(), false); + assertTrue("Expected an error status, got " + status, status >= 400); + assertTrue( + "Response should indicate a permission denial. Status " + status + ", body: " + body, + body.contains("no permissions") + || body.contains("Forbidden") + || body.contains("security_exception") + || body.contains(deniedAction)); + } + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/sql/VectorSearchIT.java b/integ-test/src/test/java/org/opensearch/sql/sql/VectorSearchIT.java index f3c3046f41e..c10bc474bae 100644 --- a/integ-test/src/test/java/org/opensearch/sql/sql/VectorSearchIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/sql/VectorSearchIT.java @@ -28,6 +28,7 @@ public class VectorSearchIT extends SQLIntegTestCase { @Override protected void init() throws Exception { + super.init(); loadIndex(Index.ACCOUNT); } diff --git a/opensearch/build.gradle b/opensearch/build.gradle index 5ca10c7091c..03d16cad083 100644 --- a/opensearch/build.gradle +++ b/opensearch/build.gradle @@ -32,6 +32,7 @@ plugins { dependencies { api project(':core') + api project(':ppl-rest-spi') api group: 'org.opensearch', name: 'opensearch', version: "${opensearch_version}" implementation "io.github.resilience4j:resilience4j-retry:${resilience4j_version}" implementation group: 'com.fasterxml.jackson.core', name: 'jackson-core', version: "${versions.jackson}" diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableCatalogScanRule.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableCatalogScanRule.java new file mode 100644 index 00000000000..9905ca564ab --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableCatalogScanRule.java @@ -0,0 +1,63 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.planner.rules; + +import org.apache.calcite.adapter.enumerable.EnumerableConvention; +import org.apache.calcite.plan.Convention; +import org.apache.calcite.plan.RelOptRuleCall; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.convert.ConverterRule; +import org.opensearch.sql.opensearch.storage.system.CalciteEnumerableCatalogScan; +import org.opensearch.sql.opensearch.storage.system.CalciteLogicalCatalogScan; +import org.opensearch.sql.opensearch.storage.system.CalciteScannableCatalogScan; + +/** + * Rule to convert a {@link CalciteLogicalCatalogScan} into an enumerable scan: a {@link + * CalciteScannableCatalogScan} when the source opts into {@code Scannable}, otherwise a plain + * {@link CalciteEnumerableCatalogScan}. + */ +public class EnumerableCatalogScanRule extends ConverterRule { + /** Default configuration. */ + public static final Config DEFAULT_CONFIG = + Config.INSTANCE + .as(Config.class) + .withConversion( + CalciteLogicalCatalogScan.class, + s -> s.getCatalogTable() != null, + Convention.NONE, + EnumerableConvention.INSTANCE, + "EnumerableCatalogScanRule") + .withRuleFactory(EnumerableCatalogScanRule::new); + + protected EnumerableCatalogScanRule(Config config) { + super(config); + } + + @Override + public boolean matches(RelOptRuleCall call) { + CalciteLogicalCatalogScan scan = call.rel(0); + return scan.getVariablesSet().isEmpty(); + } + + @Override + public RelNode convert(RelNode rel) { + final CalciteLogicalCatalogScan scan = (CalciteLogicalCatalogScan) rel; + if (scan.getCatalogTable().getSource().isScannable()) { + return new CalciteScannableCatalogScan( + scan.getCluster(), + scan.getHints(), + scan.getTable(), + scan.getCatalogTable(), + scan.getSchema()); + } + return new CalciteEnumerableCatalogScan( + scan.getCluster(), + scan.getHints(), + scan.getTable(), + scan.getCatalogTable(), + scan.getSchema()); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableSystemIndexScanRule.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableSystemIndexScanRule.java deleted file mode 100644 index 616d1178873..00000000000 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableSystemIndexScanRule.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * Copyright OpenSearch Contributors - * SPDX-License-Identifier: Apache-2.0 - */ - -package org.opensearch.sql.opensearch.planner.rules; - -import org.apache.calcite.adapter.enumerable.EnumerableConvention; -import org.apache.calcite.plan.Convention; -import org.apache.calcite.plan.RelOptRuleCall; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.convert.ConverterRule; -import org.opensearch.sql.opensearch.storage.system.CalciteEnumerableSystemIndexScan; -import org.opensearch.sql.opensearch.storage.system.CalciteLogicalSystemIndexScan; - -/** - * Rule to convert a {@link CalciteLogicalSystemIndexScan} to a {@link - * CalciteEnumerableSystemIndexScan}. - */ -public class EnumerableSystemIndexScanRule extends ConverterRule { - /** Default configuration. */ - public static final Config DEFAULT_CONFIG = - Config.INSTANCE - .as(Config.class) - .withConversion( - CalciteLogicalSystemIndexScan.class, - s -> s.getSysIndex() != null, - Convention.NONE, - EnumerableConvention.INSTANCE, - "EnumerableSystemIndexScanRule") - .withRuleFactory(EnumerableSystemIndexScanRule::new); - - /** Creates an EnumerableProjectRule. */ - protected EnumerableSystemIndexScanRule(Config config) { - super(config); - } - - @Override - public boolean matches(RelOptRuleCall call) { - CalciteLogicalSystemIndexScan scan = call.rel(0); - return scan.getVariablesSet().isEmpty(); - } - - @Override - public RelNode convert(RelNode rel) { - final CalciteLogicalSystemIndexScan scan = (CalciteLogicalSystemIndexScan) rel; - return new CalciteEnumerableSystemIndexScan( - scan.getCluster(), scan.getHints(), scan.getTable(), scan.getSysIndex(), scan.getSchema()); - } -} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java index 3c8508cc455..c200ffa8909 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java @@ -12,8 +12,8 @@ public class OpenSearchIndexRules { private static final RelOptRule INDEX_SCAN_RULE = EnumerableIndexScanRule.DEFAULT_CONFIG.toRule(); - private static final RelOptRule SYSTEM_INDEX_SCAN_RULE = - EnumerableSystemIndexScanRule.DEFAULT_CONFIG.toRule(); + private static final RelOptRule CATALOG_SCAN_RULE = + EnumerableCatalogScanRule.DEFAULT_CONFIG.toRule(); private static final RelOptRule NESTED_AGGREGATE_RULE = EnumerableNestedAggregateRule.DEFAULT_CONFIG.toRule(); private static final RelOptRule GRAPH_LOOKUP_RULE = @@ -26,7 +26,7 @@ public class OpenSearchIndexRules { public static final List OPEN_SEARCH_NON_PUSHDOWN_RULES = ImmutableList.of( INDEX_SCAN_RULE, - SYSTEM_INDEX_SCAN_RULE, + CATALOG_SCAN_RULE, NESTED_AGGREGATE_RULE, GRAPH_LOOKUP_RULE, RELEVANCE_FUNCTION_RULE); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java index b596c7bc47a..1a4e6daa279 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java @@ -71,6 +71,13 @@ public class OpenSearchSettings extends Settings { Setting.Property.NodeScope, Setting.Property.Dynamic); + public static final Setting> PPL_REST_ALLOWED_ENDPOINTS_SETTING = + Setting.listSetting( + Key.PPL_REST_ALLOWED_ENDPOINTS.getKeyValue(), + List.of("/_cluster/health"), + Function.identity(), + Setting.Property.NodeScope); + public static final Setting PPL_QUERY_TIMEOUT_SETTING = Setting.positiveTimeSetting( Key.PPL_QUERY_TIMEOUT.getKeyValue(), @@ -380,6 +387,11 @@ public OpenSearchSettings(ClusterSettings clusterSettings) { Key.PPL_ENABLED, PPL_ENABLED_SETTING, new Updater(Key.PPL_ENABLED)); + registerNonDynamicSettings( + settingBuilder, + clusterSettings, + Key.PPL_REST_ALLOWED_ENDPOINTS, + PPL_REST_ALLOWED_ENDPOINTS_SETTING); register( settingBuilder, clusterSettings, @@ -640,7 +652,9 @@ private void registerNonDynamicSettings( Settings.Key key, Setting setting) { settingBuilder.put(key, setting); - latestSettings.put(key, clusterSettings.get(setting)); + if (clusterSettings.get(setting) != null) { + latestSettings.put(key, clusterSettings.get(setting)); + } } /** @@ -711,6 +725,7 @@ public static List> pluginNonDynamicSettings() { return new ImmutableList.Builder>() .add(DATASOURCE_MASTER_SECRET_KEY) .add(DATASOURCE_CONFIG) + .add(PPL_REST_ALLOWED_ENDPOINTS_SETTING) .build(); } diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngine.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngine.java index 1b7de315fb6..5b64442d795 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngine.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngine.java @@ -5,6 +5,8 @@ package org.opensearch.sql.opensearch.storage; +import static org.opensearch.sql.utils.SystemIndexUtils.decodeRestSpec; +import static org.opensearch.sql.utils.SystemIndexUtils.isRestSource; import static org.opensearch.sql.utils.SystemIndexUtils.isSystemIndex; import java.util.Collection; @@ -15,9 +17,16 @@ import org.opensearch.sql.common.setting.Settings; import org.opensearch.sql.expression.function.FunctionResolver; import org.opensearch.sql.opensearch.client.OpenSearchClient; -import org.opensearch.sql.opensearch.storage.system.OpenSearchSystemIndex; +import org.opensearch.sql.opensearch.storage.rest.RedactorHolder; +import org.opensearch.sql.opensearch.storage.rest.RestCatalogSource; +import org.opensearch.sql.opensearch.storage.rest.RestEndpointRegistry; +import org.opensearch.sql.opensearch.storage.rest.RestEndpointRegistryHolder; +import org.opensearch.sql.opensearch.storage.system.OpenSearchCatalogTable; +import org.opensearch.sql.opensearch.storage.system.SystemIndexCatalogSource; +import org.opensearch.sql.spi.rest.Redactor; import org.opensearch.sql.storage.StorageEngine; import org.opensearch.sql.storage.Table; +import org.opensearch.sql.utils.SystemIndexUtils.RestSpec; /** OpenSearch storage engine implementation. */ @RequiredArgsConstructor @@ -35,10 +44,37 @@ public Collection getFunctions() { @Override public Table getTable(DataSourceSchemaName dataSourceSchemaName, String name) { - if (isSystemIndex(name)) { - return new OpenSearchSystemIndex(client, settings, name); + if (isRestSource(name)) { + return restTable(name); + } else if (isSystemIndex(name)) { + return new OpenSearchCatalogTable(new SystemIndexCatalogSource(client, name), settings); } else { return new OpenSearchIndex(client, settings, name); } } + + private Table restTable(String name) { + RestSpec spec = decodeRestSpec(name); + RestEndpointRegistry registry = RestEndpointRegistryHolder.get(); + if (registry == null) { + throw new IllegalStateException( + "the rest command endpoint registry is not initialized on this node"); + } + registry.resolve(spec.getEndpoint()); + List allowed = settings.getSettingValue(Settings.Key.PPL_REST_ALLOWED_ENDPOINTS); + if (allowed == null || !(allowed.contains("*") || allowed.contains(spec.getEndpoint()))) { + throw new IllegalArgumentException( + allowed == null || allowed.isEmpty() + ? "the rest command is disabled on this cluster" + : "rest endpoint [" + + spec.getEndpoint() + + "] is not enabled on this cluster. Enabled endpoints: " + + allowed); + } + // Redaction is applied by the single Redactor at the row-shaping choke point; the default is + // Redactor.NONE (no-op). + Redactor redactor = RedactorHolder.get(); + return new OpenSearchCatalogTable( + new RestCatalogSource(registry, spec, client, redactor), settings); + } } diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/CoreEndpointsProvider.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/CoreEndpointsProvider.java new file mode 100644 index 00000000000..25f9b036b0c --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/CoreEndpointsProvider.java @@ -0,0 +1,84 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import static org.opensearch.sql.spi.rest.ColumnType.BOOLEAN; +import static org.opensearch.sql.spi.rest.ColumnType.INTEGER; +import static org.opensearch.sql.spi.rest.ColumnType.STRING; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import org.opensearch.action.admin.cluster.health.ClusterHealthRequest; +import org.opensearch.action.admin.cluster.health.ClusterHealthResponse; +import org.opensearch.sql.spi.rest.ArgSpec; +import org.opensearch.sql.spi.rest.Column; +import org.opensearch.sql.spi.rest.RestEndpointContext; +import org.opensearch.sql.spi.rest.RestEndpointDefinition; +import org.opensearch.sql.spi.rest.RestEndpointProvider; +import org.opensearch.transport.client.node.NodeClient; + +/** + * The built-in {@link RestEndpointProvider}. It ships a single read-only, in-cluster endpoint, + * {@code /_cluster/health}, expressed as a {@link RestEndpointDefinition}. It is a uniform client + * of the same SPI an external plugin uses; it holds no privileged position in {@link + * RestEndpointRegistry}. Additional endpoints are left to follow-up changes. + * + *

Like any external provider, it fetches at execution time through the transport node client the + * context carries ({@link RestEndpointContext#client()}), so it holds no reference to the sql + * storage client and runs under the caller's security thread-context. + */ +public final class CoreEndpointsProvider implements RestEndpointProvider { + + @Override + public List getEndpoints() { + return List.of( + RestEndpointDefinition.builder() + .name("/_cluster/health") + .schema( + List.of( + Column.of("cluster_name", STRING), + Column.of("status", STRING), + Column.of("number_of_nodes", INTEGER), + Column.of("number_of_data_nodes", INTEGER), + Column.of("active_primary_shards", INTEGER), + Column.of("active_shards", INTEGER), + Column.of("relocating_shards", INTEGER), + Column.of("initializing_shards", INTEGER), + Column.of("unassigned_shards", INTEGER), + Column.of("timed_out", BOOLEAN))) + .argSpec(ArgSpec.builder().arg("local", Set.of("true", "false")).build()) + .handler(CoreEndpointsProvider::clusterHealth) + .build()); + } + + private static List> clusterHealth(RestEndpointContext ctx) { + NodeClient client = ctx.client(); + if (client == null) { + throw new IllegalStateException( + "the /_cluster/health rest endpoint requires an in-cluster node client"); + } + ClusterHealthRequest request = new ClusterHealthRequest(); + if (Boolean.parseBoolean(ctx.args().get("local"))) { + request.local(true); + } + ClusterHealthResponse response = client.admin().cluster().health(request).actionGet(); + Map row = new LinkedHashMap<>(); + row.put("cluster_name", response.getClusterName()); + row.put("status", response.getStatus().name().toLowerCase(Locale.ROOT)); + row.put("number_of_nodes", response.getNumberOfNodes()); + row.put("number_of_data_nodes", response.getNumberOfDataNodes()); + row.put("active_primary_shards", response.getActivePrimaryShards()); + row.put("active_shards", response.getActiveShards()); + row.put("relocating_shards", response.getRelocatingShards()); + row.put("initializing_shards", response.getInitializingShards()); + row.put("unassigned_shards", response.getUnassignedShards()); + row.put("timed_out", response.isTimedOut()); + return List.of(row); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RedactorHolder.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RedactorHolder.java new file mode 100644 index 00000000000..9b7a7b10958 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RedactorHolder.java @@ -0,0 +1,34 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import org.opensearch.sql.spi.rest.Redactor; + +/** + * Bridge for sharing the platform {@link Redactor} between plugin bootstrap (where the SQL plugin + * publishes it) and {@code OpenSearchStorageEngine.getTable} (which applies it at query time). + * Mirrors {@link RestEndpointRegistryHolder}. + * + *

Defaults to {@link Redactor#NONE} so the OSS choke point is a pure no-op even before bootstrap + * runs and so {@code getTable} never has to null-check; a provider calls {@link #set} during {@code + * createComponents} to install its own implementation. This is intentionally a single central hook, + * not a {@code loadExtensions} SPI, so redaction has one owner rather than an open registration + * surface. + */ +public final class RedactorHolder { + + private static volatile Redactor redactor = Redactor.NONE; + + private RedactorHolder() {} + + public static void set(Redactor instance) { + redactor = instance == null ? Redactor.NONE : instance; + } + + public static Redactor get() { + return redactor; + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestCatalogSource.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestCatalogSource.java new file mode 100644 index 00000000000..66b5bacb333 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestCatalogSource.java @@ -0,0 +1,65 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import java.util.Map; +import lombok.Getter; +import org.opensearch.sql.data.type.ExprType; +import org.opensearch.sql.opensearch.client.OpenSearchClient; +import org.opensearch.sql.opensearch.request.system.OpenSearchSystemRequest; +import org.opensearch.sql.opensearch.storage.system.CatalogSource; +import org.opensearch.sql.planner.logical.LogicalPlan; +import org.opensearch.sql.planner.physical.PhysicalPlan; +import org.opensearch.sql.spi.rest.Redactor; +import org.opensearch.sql.utils.SystemIndexUtils.RestSpec; + +/** + * {@link CatalogSource} for the {@code rest} command: an allow-listed, read-only management + * endpoint resolved against a {@link RestEndpointRegistry} instance, exposing the fixed endpoint + * schema. Calcite only (no V2 path) and {@code Scannable} for the {@code collect} short-circuit. + */ +@Getter +public class RestCatalogSource implements CatalogSource { + + private final OpenSearchClient client; + private final RestSpec spec; + private final RestEndpointRegistry.Endpoint endpoint; + private final Redactor redactor; + + public RestCatalogSource(RestEndpointRegistry registry, RestSpec spec, OpenSearchClient client) { + this(registry, spec, client, Redactor.NONE); + } + + public RestCatalogSource( + RestEndpointRegistry registry, RestSpec spec, OpenSearchClient client, Redactor redactor) { + this.client = client; + this.spec = spec; + this.redactor = redactor; + // Allow-list enforced here: unknown or mutating endpoints and disallowed args are rejected. + this.endpoint = registry.resolve(spec.getEndpoint()); + registry.validate(spec); + } + + @Override + public Map getFieldTypes() { + return endpoint.getSchema(); + } + + @Override + public OpenSearchSystemRequest createRequest() { + return new RestRequest(client, endpoint, spec, redactor); + } + + @Override + public boolean isScannable() { + return true; + } + + @Override + public PhysicalPlan implementV2(LogicalPlan plan) { + throw new UnsupportedOperationException("rest command is supported only on the Calcite engine"); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistry.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistry.java new file mode 100644 index 00000000000..a37438ae3bd --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistry.java @@ -0,0 +1,251 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import static org.opensearch.sql.data.model.ExprValueUtils.booleanValue; +import static org.opensearch.sql.data.model.ExprValueUtils.doubleValue; +import static org.opensearch.sql.data.model.ExprValueUtils.integerValue; +import static org.opensearch.sql.data.model.ExprValueUtils.longValue; +import static org.opensearch.sql.data.model.ExprValueUtils.stringValue; +import static org.opensearch.sql.data.type.ExprCoreType.BOOLEAN; +import static org.opensearch.sql.data.type.ExprCoreType.DOUBLE; +import static org.opensearch.sql.data.type.ExprCoreType.INTEGER; +import static org.opensearch.sql.data.type.ExprCoreType.LONG; +import static org.opensearch.sql.data.type.ExprCoreType.STRING; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import lombok.Getter; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.sql.data.model.ExprNullValue; +import org.opensearch.sql.data.model.ExprTupleValue; +import org.opensearch.sql.data.model.ExprValue; +import org.opensearch.sql.data.type.ExprType; +import org.opensearch.sql.spi.rest.ArgSpec; +import org.opensearch.sql.spi.rest.Column; +import org.opensearch.sql.spi.rest.ColumnType; +import org.opensearch.sql.spi.rest.Redactor; +import org.opensearch.sql.spi.rest.RestEndpointContext; +import org.opensearch.sql.spi.rest.RestEndpointDefinition; +import org.opensearch.sql.spi.rest.RestEndpointHandler; +import org.opensearch.sql.spi.rest.RestEndpointProvider; +import org.opensearch.sql.utils.SystemIndexUtils.RestSpec; + +/** + * The read-only endpoint allow-list, built by merging every {@link RestEndpointProvider} (the + * built-in {@link CoreEndpointsProvider} plus any externally contributed providers) into one map of + * endpoint name to an internal {@link Endpoint}. A built-in and an externally contributed endpoint + * are uniform entries here; the built-in provider holds no privileged position. + * + *

This is the single place the read-only allow-list is enforced. An endpoint that no provider + * registered, including every mutating endpoint, is rejected by {@link #resolve} with a clear + * exception, and an arg the endpoint's {@link ArgSpec} does not accept is rejected by {@link + * #validate}. Adding an endpoint is a reviewed change to a provider, never arbitrary pass-through. + */ +public final class RestEndpointRegistry { + + private static final Logger LOG = LogManager.getLogger(RestEndpointRegistry.class); + + private final Map registry; + + public RestEndpointRegistry(List providers) { + Map m = new LinkedHashMap<>(); + for (RestEndpointProvider provider : providers) { + for (RestEndpointDefinition definition : provider.getEndpoints()) { + if (m.containsKey(definition.name())) { + LOG.warn( + "rest endpoint [{}] is already registered; ignoring the duplicate from provider [{}]", + definition.name(), + provider.getClass().getName()); + continue; + } + m.put(definition.name(), new Endpoint(definition)); + } + } + this.registry = m; + } + + public java.util.Set endpointNames() { + return registry.keySet(); + } + + /** A single allow-listed endpoint, adapted from a provider's {@link RestEndpointDefinition}. */ + @Getter + public static final class Endpoint { + private final String path; + private final LinkedHashMap schema; + private final ArgSpec argSpec; + private final RestEndpointHandler handler; + + Endpoint(RestEndpointDefinition definition) { + this.path = definition.name(); + this.schema = toExprSchema(definition.schema()); + this.argSpec = definition.argSpec(); + this.handler = definition.handler(); + } + + /** + * Invoke the provider's handler, apply the platform {@link Redactor} to each raw row (with the + * endpoint name as scope), then shape it into fixed-schema rows. Runs at execution time (scan + * open). {@link Redactor#NONE} (the OSS default) passes rows through unchanged. + */ + public List toRows(RestEndpointContext ctx, Redactor redactor) { + List out = new ArrayList<>(); + for (Map raw : handler.fetch(ctx)) { + Map redacted = redactor.redact(path, raw); + LinkedHashMap tuple = new LinkedHashMap<>(); + for (Map.Entry col : schema.entrySet()) { + tuple.put(col.getKey(), coerce(col.getKey(), col.getValue(), redacted.get(col.getKey()))); + } + out.add(new ExprTupleValue(tuple)); + } + return out; + } + } + + /** + * Resolve an allow-listed endpoint. Anything no provider registered (unknown path, mutating verb, + * {@code /services/*}, plugin admin endpoints) is refused here. + */ + public Endpoint resolve(String path) { + if (path == null || path.isBlank()) { + throw new IllegalArgumentException( + "rest endpoint must be a non-empty path. Only the following endpoints are supported: " + + registry.keySet()); + } + Endpoint endpoint = registry.get(path); + if (endpoint == null) { + throw new IllegalArgumentException( + "rest endpoint [" + + path + + "] is not allow-listed. Only the following endpoints are supported: " + + registry.keySet()); + } + return endpoint; + } + + /** Validate the count, the reserved timeout token, and every supplied query arg. */ + public void validate(RestSpec spec) { + Endpoint endpoint = resolve(spec.getEndpoint()); + if (spec.getCount() != null && spec.getCount() < 0) { + throw new IllegalArgumentException( + "rest endpoint [" + + spec.getEndpoint() + + "] count must be a non-negative integer, got [" + + spec.getCount() + + "]"); + } + if (spec.getTimeout() != null) { + // The timeout token is reserved in the grammar for forward compatibility, but a single + // uniform timeout cannot map cleanly across the endpoints (wait-for-status vs + // cluster-manager vs client socket timeouts differ per action). Reject it with a clear + // client error rather than silently ignoring it. + throw new IllegalArgumentException( + "rest endpoint [" + spec.getEndpoint() + "] does not support the timeout argument yet"); + } + if (spec.getArgs() != null) { + ArgSpec argSpec = endpoint.getArgSpec(); + for (String arg : spec.getArgs().keySet()) { + if (!argSpec.allows(arg)) { + throw new IllegalArgumentException( + "rest endpoint [" + + spec.getEndpoint() + + "] does not accept arg [" + + arg + + "]. Allowed args: " + + argSpec.allowedArgs()); + } + argSpec.validateValue(spec.getEndpoint(), arg, spec.getArgs().get(arg)); + } + } + } + + private static LinkedHashMap toExprSchema(List columns) { + LinkedHashMap schema = new LinkedHashMap<>(); + for (Column column : columns) { + schema.put(column.name(), toExprType(column.type())); + } + return schema; + } + + private static ExprType toExprType(ColumnType type) { + return switch (type) { + case STRING -> STRING; + case INTEGER -> INTEGER; + case LONG -> LONG; + case DOUBLE -> DOUBLE; + case BOOLEAN -> BOOLEAN; + }; + } + + private static ExprValue coerce(String column, ExprType type, Object value) { + if (value == null) { + return ExprNullValue.of(); + } + try { + if (type == INTEGER) { + return integerValue(toNumber(value).intValue()); + } + if (type == LONG) { + return longValue(toNumber(value).longValue()); + } + if (type == DOUBLE) { + return doubleValue(toNumber(value).doubleValue()); + } + if (type == BOOLEAN) { + return booleanValue(toBoolean(value)); + } + } catch (IllegalArgumentException | ClassCastException e) { + // Surface a clear client error (HTTP 400) instead of a raw HTTP 500 when an endpoint + // returns an unexpected value shape. NumberFormatException extends IllegalArgumentException, + // so toNumber parse failures and toBoolean's "not a boolean" are both caught here; genuinely + // unexpected faults (NPE, etc.) are left to propagate. + throw new IllegalArgumentException( + "rest endpoint value for column [" + + column + + "] could not be coerced to " + + type + + ": [" + + value + + "]"); + } + return stringValue(String.valueOf(value)); + } + + private static Number toNumber(Object value) { + if (value instanceof Number n) { + return n; + } + String s = String.valueOf(value).trim(); + if (s.isEmpty()) { + throw new NumberFormatException("empty string"); + } + if (s.indexOf('.') >= 0 || s.indexOf('e') >= 0 || s.indexOf('E') >= 0) { + return Double.parseDouble(s); + } + return Long.parseLong(s); + } + + private static boolean toBoolean(Object value) { + if (value instanceof Boolean b) { + return b; + } + String s = String.valueOf(value).trim(); + if (s.isEmpty()) { + throw new IllegalArgumentException("empty string is not a boolean"); + } + if (s.equalsIgnoreCase("true")) { + return true; + } + if (s.equalsIgnoreCase("false")) { + return false; + } + throw new IllegalArgumentException("not a boolean: " + value); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistryHolder.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistryHolder.java new file mode 100644 index 00000000000..18669c05187 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistryHolder.java @@ -0,0 +1,32 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +/** + * Bridge for sharing the merged {@link RestEndpointRegistry} between plugin bootstrap (where the + * SQL plugin builds it from the built-in provider plus every provider discovered via {@code + * ExtensiblePlugin.loadExtensions}) and {@code OpenSearchStorageEngine.getTable} (which resolves a + * {@code rest} endpoint at query time). + * + *

Why a static holder: {@code loadExtensions} runs during node bootstrap, before the Node-level + * Guice injector exists, so the merged registry cannot be injected into the storage engine. + * Publishing it here once at bootstrap lets the storage engine read the same instance without going + * through the injector. Mirrors {@code AnalyticsExecutorHolder}. + */ +public final class RestEndpointRegistryHolder { + + private static volatile RestEndpointRegistry registry; + + private RestEndpointRegistryHolder() {} + + public static void set(RestEndpointRegistry instance) { + registry = instance; + } + + public static RestEndpointRegistry get() { + return registry; + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestRequest.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestRequest.java new file mode 100644 index 00000000000..1c2036380d1 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/rest/RestRequest.java @@ -0,0 +1,61 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import java.util.List; +import org.opensearch.sql.data.model.ExprValue; +import org.opensearch.sql.opensearch.client.OpenSearchClient; +import org.opensearch.sql.opensearch.request.system.OpenSearchSystemRequest; +import org.opensearch.sql.spi.rest.Redactor; +import org.opensearch.sql.spi.rest.RestEndpointContext; +import org.opensearch.sql.utils.SystemIndexUtils.RestSpec; +import org.opensearch.transport.client.node.NodeClient; + +/** + * Dispatches an allow-listed, read-only management endpoint through the endpoint's handler under + * the caller's security thread-context and returns the response shaped to the endpoint's fixed + * schema. The {@code rest} analogue of {@code OpenSearchCatIndicesRequest}; it implements {@link + * OpenSearchSystemRequest} so the enumerator pattern (resource-monitored iteration) is identical to + * the system-index scan family. This is the lazy scan: the handler runs here at {@link #search} + * (execution), never at planning time. + */ +public class RestRequest implements OpenSearchSystemRequest { + + private final OpenSearchClient client; + private final RestEndpointRegistry.Endpoint endpoint; + private final RestSpec spec; + private final Redactor redactor; + + public RestRequest( + OpenSearchClient client, + RestEndpointRegistry.Endpoint endpoint, + RestSpec spec, + Redactor redactor) { + this.client = client; + this.endpoint = endpoint; + this.spec = spec; + this.redactor = redactor; + } + + @Override + public List search() { + // The node transport client every provider handler fetches through, sourced from the same + // OpenSearchClient the storage engine uses so it runs under the caller's security + // thread-context. + NodeClient nodeClient = client == null ? null : client.getNodeClient().orElse(null); + RestEndpointContext ctx = RestEndpointContext.of(spec.getArgs(), nodeClient); + List rows = endpoint.toRows(ctx, redactor); + if (spec.getCount() != null && spec.getCount() >= 0 && rows.size() > spec.getCount()) { + return rows.subList(0, spec.getCount()); + } + return rows; + } + + @Override + public String toString() { + return "RestRequest{endpoint=" + endpoint.getPath() + "}"; + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/AbstractCalciteSystemIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/AbstractCalciteCatalogScan.java similarity index 68% rename from opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/AbstractCalciteSystemIndexScan.java rename to opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/AbstractCalciteCatalogScan.java index fa543ab266b..19d585ea684 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/AbstractCalciteSystemIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/AbstractCalciteCatalogScan.java @@ -16,21 +16,21 @@ import org.apache.calcite.rel.hint.RelHint; import org.apache.calcite.rel.type.RelDataType; -/** An abstract relational operator representing a scan of an OpenSearchSystemIndex type. */ +/** An abstract relational operator representing a scan of an {@link OpenSearchCatalogTable}. */ @Getter -public abstract class AbstractCalciteSystemIndexScan extends TableScan { - public final OpenSearchSystemIndex sysIndex; +public abstract class AbstractCalciteCatalogScan extends TableScan { + public final OpenSearchCatalogTable catalogTable; protected final RelDataType schema; - protected AbstractCalciteSystemIndexScan( + protected AbstractCalciteCatalogScan( RelOptCluster cluster, RelTraitSet traitSet, List hints, RelOptTable table, - OpenSearchSystemIndex sysIndex, + OpenSearchCatalogTable catalogTable, RelDataType schema) { super(cluster, traitSet, hints, table); - this.sysIndex = requireNonNull(sysIndex, "OpenSearch system index"); + this.catalogTable = requireNonNull(catalogTable, "OpenSearch catalog table"); this.schema = schema; } diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteEnumerableSystemIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteEnumerableCatalogScan.java similarity index 80% rename from opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteEnumerableSystemIndexScan.java rename to opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteEnumerableCatalogScan.java index b0c92dce8f9..58f86874c73 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteEnumerableSystemIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteEnumerableCatalogScan.java @@ -26,17 +26,22 @@ import org.apache.calcite.util.Pair; import org.checkerframework.checker.nullness.qual.Nullable; -/** The physical relational operator representing a scan of an OpenSearchSystemIndex type. */ -public class CalciteEnumerableSystemIndexScan extends AbstractCalciteSystemIndexScan +/** The physical relational operator representing a scan of an {@link OpenSearchCatalogTable}. */ +public class CalciteEnumerableCatalogScan extends AbstractCalciteCatalogScan implements EnumerableRel { - public CalciteEnumerableSystemIndexScan( + public CalciteEnumerableCatalogScan( RelOptCluster cluster, List hints, RelOptTable table, - OpenSearchSystemIndex sysIndex, + OpenSearchCatalogTable catalogTable, RelDataType schema) { super( - cluster, cluster.traitSetOf(EnumerableConvention.INSTANCE), hints, table, sysIndex, schema); + cluster, + cluster.traitSetOf(EnumerableConvention.INSTANCE), + hints, + table, + catalogTable, + schema); } @Override @@ -60,7 +65,7 @@ public Result implement(EnumerableRelImplementor implementor, Prefer pref) { PhysType physType = PhysTypeImpl.of(implementor.getTypeFactory(), getRowType(), pref.preferArray()); - Expression scanOperator = implementor.stash(this, CalciteEnumerableSystemIndexScan.class); + Expression scanOperator = implementor.stash(this, CalciteEnumerableCatalogScan.class); return implementor.result(physType, Blocks.toBlock(Expressions.call(scanOperator, "scan"))); } @@ -68,10 +73,10 @@ public Result implement(EnumerableRelImplementor implementor, Prefer pref) { return new AbstractEnumerable<>() { @Override public Enumerator enumerator() { - return new OpenSearchSystemIndexEnumerator( + return new OpenSearchCatalogEnumerator( getFieldPath(), - sysIndex.getSystemIndexBundle().getRight(), - sysIndex.createOpenSearchResourceMonitor()); + catalogTable.getSource().createRequest(), + catalogTable.createOpenSearchResourceMonitor()); } }; } diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteLogicalSystemIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteLogicalCatalogScan.java similarity index 59% rename from opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteLogicalSystemIndexScan.java rename to opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteLogicalCatalogScan.java index 012ceec8c13..4278539c2b0 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteLogicalSystemIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteLogicalCatalogScan.java @@ -14,35 +14,35 @@ import org.apache.calcite.plan.RelTraitSet; import org.apache.calcite.rel.hint.RelHint; import org.apache.calcite.rel.type.RelDataType; -import org.opensearch.sql.opensearch.planner.rules.EnumerableSystemIndexScanRule; +import org.opensearch.sql.opensearch.planner.rules.EnumerableCatalogScanRule; -/** The logical relational operator representing a scan of an OpenSearchSystemIndex type. */ -public class CalciteLogicalSystemIndexScan extends AbstractCalciteSystemIndexScan { +/** The logical relational operator representing a scan of an {@link OpenSearchCatalogTable}. */ +public class CalciteLogicalCatalogScan extends AbstractCalciteCatalogScan { - public CalciteLogicalSystemIndexScan( - RelOptCluster cluster, RelOptTable table, OpenSearchSystemIndex sysIndex) { + public CalciteLogicalCatalogScan( + RelOptCluster cluster, RelOptTable table, OpenSearchCatalogTable catalogTable) { this( cluster, cluster.traitSetOf(Convention.NONE), ImmutableList.of(), table, - sysIndex, + catalogTable, table.getRowType()); } - protected CalciteLogicalSystemIndexScan( + protected CalciteLogicalCatalogScan( RelOptCluster cluster, RelTraitSet traitSet, List hints, RelOptTable table, - OpenSearchSystemIndex sysIndex, + OpenSearchCatalogTable catalogTable, RelDataType schema) { - super(cluster, traitSet, hints, table, sysIndex, schema); + super(cluster, traitSet, hints, table, catalogTable, schema); } @Override public void register(RelOptPlanner planner) { super.register(planner); - planner.addRule(EnumerableSystemIndexScanRule.DEFAULT_CONFIG.toRule()); + planner.addRule(EnumerableCatalogScanRule.DEFAULT_CONFIG.toRule()); } } diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteScannableCatalogScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteScannableCatalogScan.java new file mode 100644 index 00000000000..1d021cda354 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CalciteScannableCatalogScan.java @@ -0,0 +1,29 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.system; + +import java.util.List; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rel.hint.RelHint; +import org.apache.calcite.rel.type.RelDataType; +import org.opensearch.sql.calcite.plan.Scannable; + +/** + * A {@link CalciteEnumerableCatalogScan} that additionally carries the {@link Scannable} marker, + * enabling the {@code collect} short-circuit. Produced when the {@link CatalogSource} opts in via + * {@link CatalogSource#isScannable()}. + */ +public class CalciteScannableCatalogScan extends CalciteEnumerableCatalogScan implements Scannable { + public CalciteScannableCatalogScan( + RelOptCluster cluster, + List hints, + RelOptTable table, + OpenSearchCatalogTable catalogTable, + RelDataType schema) { + super(cluster, hints, table, catalogTable, schema); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CatalogSource.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CatalogSource.java new file mode 100644 index 00000000000..85dabe06b9a --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/CatalogSource.java @@ -0,0 +1,39 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.system; + +import java.util.Map; +import org.opensearch.sql.data.type.ExprType; +import org.opensearch.sql.opensearch.request.system.OpenSearchSystemRequest; +import org.opensearch.sql.planner.logical.LogicalPlan; +import org.opensearch.sql.planner.physical.PhysicalPlan; + +/** + * Strategy supplying the schema and row source for an {@link OpenSearchCatalogTable} backed by a + * read-only OpenSearch admin API. Each endpoint family plugs in its own {@code CatalogSource} + * rather than defining a separate table and Calcite scan hierarchy. + */ +public interface CatalogSource { + + /** Fixed schema mapping each column name to its type. */ + Map getFieldTypes(); + + /** + * Builds the read-only request whose {@link OpenSearchSystemRequest#search()} yields the rows. + */ + OpenSearchSystemRequest createRequest(); + + /** + * Whether the enumerable scan should carry the {@link org.opensearch.sql.calcite.plan.Scannable} + * marker for the {@code collect} short-circuit. Defaults to {@code false}. + */ + default boolean isScannable() { + return false; + } + + /** The V2 physical plan path, or throws when the source is Calcite only. */ + PhysicalPlan implementV2(LogicalPlan plan); +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndexEnumerator.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogEnumerator.java similarity index 90% rename from opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndexEnumerator.java rename to opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogEnumerator.java index 1bb8f9d5293..dff9b47265a 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndexEnumerator.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogEnumerator.java @@ -16,8 +16,11 @@ import org.opensearch.sql.monitor.ResourceMonitor; import org.opensearch.sql.opensearch.request.system.OpenSearchSystemRequest; -/** Supports a simple iteration over a collection for OpenSearch system index */ -public class OpenSearchSystemIndexEnumerator implements Enumerator { +/** + * Resource-monitored iteration over the rows produced by a read-only catalog {@link + * OpenSearchSystemRequest}. + */ +public class OpenSearchCatalogEnumerator implements Enumerator { /** How many moveNext() calls to perform resource check once. */ private static final long NUMBER_OF_NEXT_CALL_TO_CHECK = 1000; @@ -35,7 +38,7 @@ public class OpenSearchSystemIndexEnumerator implements Enumerator { /** ResourceMonitor. */ private final ResourceMonitor monitor; - public OpenSearchSystemIndexEnumerator( + public OpenSearchCatalogEnumerator( List fields, OpenSearchSystemRequest request, ResourceMonitor monitor) { this.fields = fields; this.request = request; diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogTable.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogTable.java new file mode 100644 index 00000000000..9bcb3a3ff0c --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogTable.java @@ -0,0 +1,67 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.system; + +import java.util.Map; +import lombok.Getter; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rel.RelNode; +import org.opensearch.sql.calcite.plan.AbstractOpenSearchTable; +import org.opensearch.sql.common.setting.Settings; +import org.opensearch.sql.data.type.ExprType; +import org.opensearch.sql.opensearch.monitor.OpenSearchMemoryHealthy; +import org.opensearch.sql.opensearch.monitor.OpenSearchResourceMonitor; +import org.opensearch.sql.planner.logical.LogicalPlan; +import org.opensearch.sql.planner.physical.PhysicalPlan; + +/** + * A single generic Calcite and V2 table over a read-only OpenSearch catalog endpoint. Per-endpoint + * behavior (schema, row source, V2 support, and the {@code Scannable} marker) is supplied by a + * pluggable {@link CatalogSource}. + */ +@Getter +public class OpenSearchCatalogTable extends AbstractOpenSearchTable { + + private final CatalogSource source; + private final Settings settings; + + public OpenSearchCatalogTable(CatalogSource source, Settings settings) { + this.source = source; + this.settings = settings; + } + + @Override + public boolean exists() { + return true; + } + + @Override + public void create(Map schema) { + throw new UnsupportedOperationException( + "OpenSearch catalog table is predefined and cannot be created"); + } + + @Override + public Map getFieldTypes() { + return source.getFieldTypes(); + } + + @Override + public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) { + final RelOptCluster cluster = context.getCluster(); + return new CalciteLogicalCatalogScan(cluster, relOptTable, this); + } + + @Override + public PhysicalPlan implement(LogicalPlan plan) { + return source.implementV2(plan); + } + + public OpenSearchResourceMonitor createOpenSearchResourceMonitor() { + return new OpenSearchResourceMonitor(settings, new OpenSearchMemoryHealthy(settings)); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndex.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndex.java deleted file mode 100644 index 6bae4d21aba..00000000000 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndex.java +++ /dev/null @@ -1,105 +0,0 @@ -/* - * Copyright OpenSearch Contributors - * SPDX-License-Identifier: Apache-2.0 - */ - -package org.opensearch.sql.opensearch.storage.system; - -import static org.opensearch.sql.utils.SystemIndexUtils.systemTable; - -import com.google.common.annotations.VisibleForTesting; -import java.util.Map; -import lombok.Getter; -import lombok.RequiredArgsConstructor; -import org.apache.calcite.plan.RelOptCluster; -import org.apache.calcite.plan.RelOptTable; -import org.apache.calcite.rel.RelNode; -import org.apache.commons.lang3.tuple.Pair; -import org.opensearch.sql.calcite.plan.AbstractOpenSearchTable; -import org.opensearch.sql.common.setting.Settings; -import org.opensearch.sql.data.type.ExprType; -import org.opensearch.sql.opensearch.client.OpenSearchClient; -import org.opensearch.sql.opensearch.monitor.OpenSearchMemoryHealthy; -import org.opensearch.sql.opensearch.monitor.OpenSearchResourceMonitor; -import org.opensearch.sql.opensearch.request.system.OpenSearchCatIndicesRequest; -import org.opensearch.sql.opensearch.request.system.OpenSearchDescribeIndexRequest; -import org.opensearch.sql.opensearch.request.system.OpenSearchSystemRequest; -import org.opensearch.sql.planner.DefaultImplementor; -import org.opensearch.sql.planner.logical.LogicalPlan; -import org.opensearch.sql.planner.logical.LogicalRelation; -import org.opensearch.sql.planner.physical.PhysicalPlan; -import org.opensearch.sql.utils.SystemIndexUtils; - -/** OpenSearch System Index Table Implementation. */ -@Getter -public class OpenSearchSystemIndex extends AbstractOpenSearchTable { - /** System Index Name. */ - private final Pair systemIndexBundle; - - @Getter private final Settings settings; - - public OpenSearchSystemIndex(OpenSearchClient client, Settings settings, String indexName) { - this.systemIndexBundle = buildIndexBundle(client, indexName); - this.settings = settings; - } - - @Override - public boolean exists() { - return true; // TODO: implement for system index later - } - - @Override - public void create(Map schema) { - throw new UnsupportedOperationException( - "OpenSearch system index is predefined and cannot be created"); - } - - @Override - public Map getFieldTypes() { - return systemIndexBundle.getLeft().getMapping(); - } - - @Override - public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) { - final RelOptCluster cluster = context.getCluster(); - return new CalciteLogicalSystemIndexScan(cluster, relOptTable, this); - } - - @Override - public PhysicalPlan implement(LogicalPlan plan) { - return plan.accept(new OpenSearchSystemIndexDefaultImplementor(), null); - } - - public OpenSearchResourceMonitor createOpenSearchResourceMonitor() { - return new OpenSearchResourceMonitor(getSettings(), new OpenSearchMemoryHealthy(settings)); - } - - @VisibleForTesting - @RequiredArgsConstructor - public class OpenSearchSystemIndexDefaultImplementor extends DefaultImplementor { - - @Override - public PhysicalPlan visitRelation(LogicalRelation node, Object context) { - return new OpenSearchSystemIndexScan(systemIndexBundle.getRight()); - } - } - - /** - * Constructor of ElasticsearchSystemIndexName. - * - * @param indexName index name; - */ - private Pair buildIndexBundle( - OpenSearchClient client, String indexName) { - SystemIndexUtils.SystemTable systemTable = systemTable(indexName); - if (systemTable.isSystemInfoTable()) { - return Pair.of( - OpenSearchSystemIndexSchema.SYS_TABLE_TABLES, new OpenSearchCatIndicesRequest(client)); - } else { - return Pair.of( - OpenSearchSystemIndexSchema.SYS_TABLE_MAPPINGS, - new OpenSearchDescribeIndexRequest( - client, systemTable.getTableName(), systemTable.getLangSpec())); - } - } -} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/SystemIndexCatalogSource.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/SystemIndexCatalogSource.java new file mode 100644 index 00000000000..3541e745e9d --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/SystemIndexCatalogSource.java @@ -0,0 +1,70 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.system; + +import static org.opensearch.sql.utils.SystemIndexUtils.systemTable; + +import java.util.Map; +import org.apache.commons.lang3.tuple.Pair; +import org.opensearch.sql.data.type.ExprType; +import org.opensearch.sql.opensearch.client.OpenSearchClient; +import org.opensearch.sql.opensearch.request.system.OpenSearchCatIndicesRequest; +import org.opensearch.sql.opensearch.request.system.OpenSearchDescribeIndexRequest; +import org.opensearch.sql.opensearch.request.system.OpenSearchSystemRequest; +import org.opensearch.sql.planner.DefaultImplementor; +import org.opensearch.sql.planner.logical.LogicalPlan; +import org.opensearch.sql.planner.logical.LogicalRelation; +import org.opensearch.sql.planner.physical.PhysicalPlan; +import org.opensearch.sql.utils.SystemIndexUtils; + +/** + * {@link CatalogSource} for the SHOW TABLES and DESCRIBE system tables, backed by index listing and + * index field mappings. + */ +public class SystemIndexCatalogSource implements CatalogSource { + + private final Pair bundle; + + public SystemIndexCatalogSource(OpenSearchClient client, String indexName) { + this.bundle = buildBundle(client, indexName); + } + + @Override + public Map getFieldTypes() { + return bundle.getLeft().getMapping(); + } + + @Override + public OpenSearchSystemRequest createRequest() { + return bundle.getRight(); + } + + @Override + public PhysicalPlan implementV2(LogicalPlan plan) { + return plan.accept( + new DefaultImplementor() { + @Override + public PhysicalPlan visitRelation(LogicalRelation node, Object context) { + return new OpenSearchSystemIndexScan(bundle.getRight()); + } + }, + null); + } + + private static Pair buildBundle( + OpenSearchClient client, String indexName) { + SystemIndexUtils.SystemTable systemTable = systemTable(indexName); + if (systemTable.isSystemInfoTable()) { + return Pair.of( + OpenSearchSystemIndexSchema.SYS_TABLE_TABLES, new OpenSearchCatIndicesRequest(client)); + } else { + return Pair.of( + OpenSearchSystemIndexSchema.SYS_TABLE_MAPPINGS, + new OpenSearchDescribeIndexRequest( + client, systemTable.getTableName(), systemTable.getLangSpec())); + } + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java index 5024d416086..4d77df2c992 100644 --- a/opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java @@ -9,12 +9,14 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.AdditionalMatchers.not; import static org.mockito.AdditionalMatchers.or; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.when; import static org.opensearch.sql.opensearch.setting.OpenSearchSettings.ASYNC_QUERY_EXTERNAL_SCHEDULER_ENABLED_SETTING; import static org.opensearch.sql.opensearch.setting.OpenSearchSettings.ASYNC_QUERY_EXTERNAL_SCHEDULER_INTERVAL_SETTING; +import static org.opensearch.sql.opensearch.setting.OpenSearchSettings.PPL_REST_ALLOWED_ENDPOINTS_SETTING; import static org.opensearch.sql.opensearch.setting.OpenSearchSettings.QUERY_MEMORY_LIMIT_SETTING; import static org.opensearch.sql.opensearch.setting.OpenSearchSettings.SPARK_EXECUTION_ENGINE_CONFIG; @@ -74,6 +76,13 @@ void pluginNonDynamicSettings() { assertFalse(settings.isEmpty()); } + @Test + void restSettingsAreNonDynamic() { + assertFalse(PPL_REST_ALLOWED_ENDPOINTS_SETTING.isDynamic()); + List> nonDynamic = OpenSearchSettings.pluginNonDynamicSettings(); + assertTrue(nonDynamic.contains(PPL_REST_ALLOWED_ENDPOINTS_SETTING)); + } + @Test void getSettings() { when(clusterSettings.get(ClusterName.CLUSTER_NAME_SETTING)).thenReturn(ClusterName.DEFAULT); diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngineTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngineTest.java index fa04395e065..cae3858a83d 100644 --- a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngineTest.java +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngineTest.java @@ -7,11 +7,16 @@ import static org.junit.jupiter.api.Assertions.assertAll; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.when; import static org.opensearch.sql.analysis.DataSourceSchemaIdentifierNameResolver.DEFAULT_DATASOURCE_NAME; import static org.opensearch.sql.utils.SystemIndexUtils.TABLE_INFO; import java.util.Collection; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; @@ -20,8 +25,12 @@ import org.opensearch.sql.common.setting.Settings; import org.opensearch.sql.expression.function.FunctionResolver; import org.opensearch.sql.opensearch.client.OpenSearchClient; -import org.opensearch.sql.opensearch.storage.system.OpenSearchSystemIndex; +import org.opensearch.sql.opensearch.storage.rest.CoreEndpointsProvider; +import org.opensearch.sql.opensearch.storage.rest.RestEndpointRegistry; +import org.opensearch.sql.opensearch.storage.rest.RestEndpointRegistryHolder; +import org.opensearch.sql.opensearch.storage.system.OpenSearchCatalogTable; import org.opensearch.sql.storage.Table; +import org.opensearch.sql.utils.SystemIndexUtils; @ExtendWith(MockitoExtension.class) class OpenSearchStorageEngineTest { @@ -30,6 +39,13 @@ class OpenSearchStorageEngineTest { @Mock private Settings settings; + @BeforeEach + void publishRestRegistry() { + // restTable() reads the merged registry from the holder (published by SQLPlugin in production); + // publish a built-in-only registry here so the rest endpoints resolve in this unit test. + RestEndpointRegistryHolder.set(new RestEndpointRegistry(List.of(new CoreEndpointsProvider()))); + } + @Test public void getTable() { OpenSearchStorageEngine engine = new OpenSearchStorageEngine(client, settings); @@ -52,6 +68,67 @@ public void getSystemTable() { OpenSearchStorageEngine engine = new OpenSearchStorageEngine(client, settings); Table table = engine.getTable(new DataSourceSchemaName(DEFAULT_DATASOURCE_NAME, "default"), TABLE_INFO); - assertAll(() -> assertNotNull(table), () -> assertTrue(table instanceof OpenSearchSystemIndex)); + assertAll( + () -> assertNotNull(table), () -> assertTrue(table instanceof OpenSearchCatalogTable)); + } + + @Test + public void getRestTableAllowedByWildcard() { + when(settings.getSettingValue(Settings.Key.PPL_REST_ALLOWED_ENDPOINTS)) + .thenReturn(List.of("*")); + OpenSearchStorageEngine engine = new OpenSearchStorageEngine(client, settings); + String name = + SystemIndexUtils.restTable( + new SystemIndexUtils.RestSpec("/_cluster/health", Map.of(), null, null)); + Table table = + engine.getTable(new DataSourceSchemaName(DEFAULT_DATASOURCE_NAME, "default"), name); + assertTrue(table instanceof OpenSearchCatalogTable); + } + + @Test + public void getRestTableAllowedBySubset() { + when(settings.getSettingValue(Settings.Key.PPL_REST_ALLOWED_ENDPOINTS)) + .thenReturn(List.of("/_cluster/health")); + OpenSearchStorageEngine engine = new OpenSearchStorageEngine(client, settings); + String name = + SystemIndexUtils.restTable( + new SystemIndexUtils.RestSpec("/_cluster/health", Map.of(), null, null)); + assertTrue( + engine.getTable(new DataSourceSchemaName(DEFAULT_DATASOURCE_NAME, "default"), name) + instanceof OpenSearchCatalogTable); + } + + @Test + public void getRestTableRejectedWhenEndpointNotInSubset() { + // The endpoint resolves in the registry but is absent from the (non-empty) enabled subset. + when(settings.getSettingValue(Settings.Key.PPL_REST_ALLOWED_ENDPOINTS)) + .thenReturn(List.of("/_some/other")); + OpenSearchStorageEngine engine = new OpenSearchStorageEngine(client, settings); + String name = + SystemIndexUtils.restTable( + new SystemIndexUtils.RestSpec("/_cluster/health", Map.of(), null, null)); + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> + engine.getTable( + new DataSourceSchemaName(DEFAULT_DATASOURCE_NAME, "default"), name)); + assertTrue(e.getMessage().contains("is not enabled on this cluster")); + } + + @Test + public void getRestTableDisabledWhenListEmpty() { + when(settings.getSettingValue(Settings.Key.PPL_REST_ALLOWED_ENDPOINTS)).thenReturn(List.of()); + OpenSearchStorageEngine engine = new OpenSearchStorageEngine(client, settings); + String name = + SystemIndexUtils.restTable( + new SystemIndexUtils.RestSpec("/_cluster/health", Map.of(), null, null)); + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> + engine.getTable( + new DataSourceSchemaName(DEFAULT_DATASOURCE_NAME, "default"), name)); + assertTrue(e.getMessage().contains("disabled on this cluster")); } } diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/CoreEndpointsProviderTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/CoreEndpointsProviderTest.java new file mode 100644 index 00000000000..cb36a3253fb --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/CoreEndpointsProviderTest.java @@ -0,0 +1,64 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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.when; + +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.opensearch.action.admin.cluster.health.ClusterHealthRequest; +import org.opensearch.action.admin.cluster.health.ClusterHealthResponse; +import org.opensearch.cluster.health.ClusterHealthStatus; +import org.opensearch.sql.data.model.ExprValue; +import org.opensearch.sql.spi.rest.Redactor; +import org.opensearch.sql.spi.rest.RestEndpointContext; +import org.opensearch.transport.client.node.NodeClient; + +/** + * Proves the built-in {@code /_cluster/health} provider fetches through the transport node client + * the context carries (the same seam an external provider uses) and flattens the response to the + * fixed schema, holding no reference to the sql storage client. + */ +class CoreEndpointsProviderTest { + + @Test + void clusterHealthFetchesViaContextNodeClient() { + ClusterHealthResponse response = mock(ClusterHealthResponse.class); + when(response.getClusterName()).thenReturn("test-cluster"); + when(response.getStatus()).thenReturn(ClusterHealthStatus.GREEN); + when(response.getNumberOfNodes()).thenReturn(3); + when(response.getNumberOfDataNodes()).thenReturn(2); + when(response.getActivePrimaryShards()).thenReturn(5); + when(response.getActiveShards()).thenReturn(10); + when(response.getRelocatingShards()).thenReturn(0); + when(response.getInitializingShards()).thenReturn(1); + when(response.getUnassignedShards()).thenReturn(4); + when(response.isTimedOut()).thenReturn(false); + + NodeClient nodeClient = mock(NodeClient.class, RETURNS_DEEP_STUBS); + when(nodeClient.admin().cluster().health(any(ClusterHealthRequest.class)).actionGet()) + .thenReturn(response); + + RestEndpointRegistry registry = new RestEndpointRegistry(List.of(new CoreEndpointsProvider())); + List rows = + registry + .resolve("/_cluster/health") + .toRows(RestEndpointContext.of(Map.of(), nodeClient), Redactor.NONE); + + assertEquals(1, rows.size()); + Map row = rows.get(0).tupleValue(); + assertEquals("test-cluster", row.get("cluster_name").stringValue()); + assertEquals("green", row.get("status").stringValue()); + assertEquals(3, row.get("number_of_nodes").integerValue()); + assertFalse(row.get("timed_out").booleanValue()); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RedactorSeamTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RedactorSeamTest.java new file mode 100644 index 00000000000..1f134486a02 --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RedactorSeamTest.java @@ -0,0 +1,65 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.opensearch.sql.data.model.ExprValue; +import org.opensearch.sql.spi.rest.Column; +import org.opensearch.sql.spi.rest.ColumnType; +import org.opensearch.sql.spi.rest.Redactor; +import org.opensearch.sql.spi.rest.RestEndpointContext; +import org.opensearch.sql.spi.rest.RestEndpointDefinition; +import org.opensearch.sql.spi.rest.RestEndpointProvider; + +/** + * Proves the {@code rest} choke point applies the single {@link Redactor} to each raw response row + * (with the endpoint name as scope) before coercion, and that {@link Redactor#NONE} passes rows + * through unchanged. + */ +class RedactorSeamTest { + + private static RestEndpointRegistry.Endpoint probe() { + RestEndpointProvider provider = + () -> + List.of( + RestEndpointDefinition.builder() + .name("/_test/probe") + .schema( + List.of( + Column.of("ip", ColumnType.STRING), + Column.of("count", ColumnType.INTEGER))) + .handler(ctx -> List.of(Map.of("ip", "10.0.0.7", "count", 3))) + .build()); + return new RestEndpointRegistry(List.of(provider)).resolve("/_test/probe"); + } + + @Test + void chokePointAppliesRedactorToRawRowBeforeCoercion() { + Redactor maskIp = + (endpoint, row) -> { + Map masked = new HashMap<>(row); + masked.put("ip", "x.x.x.x"); + return masked; + }; + + List rows = probe().toRows(RestEndpointContext.of(Map.of(), null), maskIp); + + assertEquals("x.x.x.x", rows.get(0).tupleValue().get("ip").stringValue()); + // A column the redactor did not touch is still coerced to its schema type. + assertEquals(3, rows.get(0).tupleValue().get("count").integerValue()); + } + + @Test + void noneRedactorPassesThrough() { + List rows = probe().toRows(RestEndpointContext.of(Map.of(), null), Redactor.NONE); + assertEquals("10.0.0.7", rows.get(0).tupleValue().get("ip").stringValue()); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestCatalogSourceTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestCatalogSourceTest.java new file mode 100644 index 00000000000..b8111fadadc --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestCatalogSourceTest.java @@ -0,0 +1,157 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.hasEntry; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.when; +import static org.opensearch.sql.data.type.ExprCoreType.INTEGER; +import static org.opensearch.sql.data.type.ExprCoreType.STRING; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.opensearch.sql.data.model.ExprValue; +import org.opensearch.sql.data.type.ExprType; +import org.opensearch.sql.opensearch.client.OpenSearchClient; +import org.opensearch.sql.spi.rest.Column; +import org.opensearch.sql.spi.rest.ColumnType; +import org.opensearch.sql.spi.rest.RestEndpointDefinition; +import org.opensearch.sql.spi.rest.RestEndpointProvider; +import org.opensearch.sql.utils.SystemIndexUtils.RestSpec; + +/** + * Covers the {@code rest} {@link RestCatalogSource} against the PR1 endpoint set (only {@code + * /_cluster/health}): fixed endpoint schema, allow-list enforcement, response row shaping and + * truncation, the {@code Scannable} opt-in, and the Calcite only (no V2) path. Schema and gating + * resolve against a registry built from the built-in {@link CoreEndpointsProvider}; row shaping and + * truncation use a fake provider returning canned rows, independent of the health transport fetch. + */ +@ExtendWith(MockitoExtension.class) +class RestCatalogSourceTest { + + @Mock private OpenSearchClient client; + + private RestEndpointRegistry registry; + + @BeforeEach + void buildRegistry() { + registry = new RestEndpointRegistry(List.of(new CoreEndpointsProvider())); + } + + private RestSpec healthSpec() { + return new RestSpec("/_cluster/health", Map.of(), null, null); + } + + private static RestEndpointRegistry fakeHealthRegistry(List> rows) { + RestEndpointProvider provider = + () -> + List.of( + RestEndpointDefinition.builder() + .name("/_cluster/health") + .schema( + List.of( + Column.of("status", ColumnType.STRING), + Column.of("number_of_nodes", ColumnType.INTEGER))) + .handler(ctx -> rows) + .build()); + return new RestEndpointRegistry(List.of(provider)); + } + + @Test + void getFieldTypesReturnsFixedEndpointSchema() { + RestCatalogSource source = new RestCatalogSource(registry, healthSpec(), client); + Map fieldTypes = source.getFieldTypes(); + assertThat(fieldTypes, hasEntry("status", STRING)); + assertThat(fieldTypes, hasEntry("number_of_nodes", INTEGER)); + } + + @Test + void isScannable() { + assertTrue(new RestCatalogSource(registry, healthSpec(), client).isScannable()); + } + + @Test + void implementV2IsUnsupported() { + RestCatalogSource source = new RestCatalogSource(registry, healthSpec(), client); + assertThrows(UnsupportedOperationException.class, () -> source.implementV2(null)); + } + + @Test + void constructorRejectsNonAllowListedEndpoint() { + assertThrows( + IllegalArgumentException.class, + () -> + new RestCatalogSource( + registry, new RestSpec("/_cluster/reroute", Map.of(), null, null), client)); + } + + @Test + void constructorRejectsDisallowedArg() { + assertThrows( + IllegalArgumentException.class, + () -> + new RestCatalogSource( + registry, + new RestSpec("/_cluster/health", Map.of("bad", "x"), null, null), + client)); + } + + @Test + void constructorRejectsNegativeCount() { + assertThrows( + IllegalArgumentException.class, + () -> + new RestCatalogSource( + registry, new RestSpec("/_cluster/health", Map.of(), -1, null), client)); + } + + @Test + void constructorRejectsTimeoutArg() { + assertThrows( + IllegalArgumentException.class, + () -> + new RestCatalogSource( + registry, new RestSpec("/_cluster/health", Map.of(), null, "5s"), client)); + } + + @Test + void restRequestShapesResponseRows() { + when(client.getNodeClient()).thenReturn(Optional.empty()); + Map health = new LinkedHashMap<>(); + health.put("status", "green"); + health.put("number_of_nodes", 1); + RestCatalogSource source = + new RestCatalogSource(fakeHealthRegistry(List.of(health)), healthSpec(), client); + List rows = source.createRequest().search(); + assertEquals(1, rows.size()); + assertEquals("green", rows.get(0).tupleValue().get("status").stringValue()); + assertEquals(1, rows.get(0).tupleValue().get("number_of_nodes").integerValue()); + } + + @Test + void countTruncatesRows() { + // count=0 exercises the truncation path (subList to empty) over a single-row response. + when(client.getNodeClient()).thenReturn(Optional.empty()); + Map health = new LinkedHashMap<>(); + health.put("status", "green"); + RestCatalogSource source = + new RestCatalogSource( + fakeHealthRegistry(List.of(health)), + new RestSpec("/_cluster/health", Map.of(), 0, null), + client); + assertTrue(source.createRequest().search().isEmpty()); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointExtensibilityTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointExtensibilityTest.java new file mode 100644 index 00000000000..f8db570cb29 --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointExtensibilityTest.java @@ -0,0 +1,106 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.opensearch.sql.spi.rest.ColumnType.STRING; + +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.opensearch.sql.data.model.ExprValue; +import org.opensearch.sql.spi.rest.ArgSpec; +import org.opensearch.sql.spi.rest.Column; +import org.opensearch.sql.spi.rest.Redactor; +import org.opensearch.sql.spi.rest.RestEndpointContext; +import org.opensearch.sql.spi.rest.RestEndpointDefinition; +import org.opensearch.sql.spi.rest.RestEndpointProvider; +import org.opensearch.sql.utils.SystemIndexUtils.RestSpec; + +/** + * Proves the {@code rest} framework treats the built-in {@link CoreEndpointsProvider} and an + * externally contributed {@link RestEndpointProvider} as uniform clients of one registry: endpoints + * from BOTH resolve, validate against the same allow-list, and fetch rows the same way. The + * built-in provider holds no privileged position. + */ +class RestEndpointExtensibilityTest { + + /** A stand-in external plugin provider: contributes one endpoint that echoes a query arg. */ + private static final class FakeEchoProvider implements RestEndpointProvider { + @Override + public List getEndpoints() { + return List.of( + RestEndpointDefinition.builder() + .name("/_plugin/echo") + .schema(List.of(Column.of("message", STRING))) + .argSpec(ArgSpec.builder().arg("text").build()) + .handler( + ctx -> List.of(Map.of("message", ctx.args().getOrDefault("text", "default")))) + .build()); + } + } + + private RestEndpointRegistry mergedRegistry() { + return new RestEndpointRegistry(List.of(new CoreEndpointsProvider(), new FakeEchoProvider())); + } + + @Test + void bothBuiltInAndExternalEndpointsResolve() { + RestEndpointRegistry registry = mergedRegistry(); + + assertEquals("/_cluster/health", registry.resolve("/_cluster/health").getPath()); + assertEquals("/_plugin/echo", registry.resolve("/_plugin/echo").getPath()); + + assertTrue(registry.endpointNames().contains("/_cluster/health")); + assertTrue(registry.endpointNames().contains("/_plugin/echo")); + } + + @Test + void externalEndpointFetchesThroughTheSamePath() { + RestEndpointRegistry.Endpoint echo = mergedRegistry().resolve("/_plugin/echo"); + List rows = + echo.toRows(RestEndpointContext.of(Map.of("text", "hi"), null), Redactor.NONE); + assertEquals(1, rows.size()); + assertEquals("hi", rows.get(0).tupleValue().get("message").stringValue()); + } + + @Test + void externalEndpointArgsValidatedByTheSameAllowList() { + RestEndpointRegistry registry = mergedRegistry(); + // Declared arg is accepted. + registry.validate(new RestSpec("/_plugin/echo", Map.of("text", "x"), null, null)); + // Undeclared arg is rejected by the same validation path that guards built-in endpoints. + IllegalArgumentException ex = + assertThrows( + IllegalArgumentException.class, + () -> + registry.validate( + new RestSpec("/_plugin/echo", Map.of("not_allowed", "x"), null, null))); + assertTrue(ex.getMessage().contains("does not accept arg")); + } + + @Test + void duplicateEndpointNameAcrossProvidersIsSkippedFirstWins() { + // A second provider that re-declares /_cluster/health is ignored rather than failing the + // build; the first (built-in) registration wins. + RestEndpointProvider shadowsCore = + () -> + List.of( + RestEndpointDefinition.builder() + .name("/_cluster/health") + .schema(List.of(Column.of("shadow", STRING))) + .handler(ctx -> List.of()) + .build()); + RestEndpointRegistry registry = + new RestEndpointRegistry(List.of(new CoreEndpointsProvider(), shadowsCore)); + + assertEquals("/_cluster/health", registry.resolve("/_cluster/health").getPath()); + assertFalse(registry.resolve("/_cluster/health").getSchema().containsKey("shadow")); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistryTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistryTest.java new file mode 100644 index 00000000000..fc93877a5b4 --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/rest/RestEndpointRegistryTest.java @@ -0,0 +1,192 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.rest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.opensearch.sql.data.type.ExprCoreType.INTEGER; +import static org.opensearch.sql.data.type.ExprCoreType.STRING; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.opensearch.sql.data.model.ExprValue; +import org.opensearch.sql.spi.rest.Column; +import org.opensearch.sql.spi.rest.ColumnType; +import org.opensearch.sql.spi.rest.Redactor; +import org.opensearch.sql.spi.rest.RestEndpointContext; +import org.opensearch.sql.spi.rest.RestEndpointDefinition; +import org.opensearch.sql.spi.rest.RestEndpointProvider; +import org.opensearch.sql.utils.SystemIndexUtils.RestSpec; + +/** + * Covers the {@code rest} {@link RestEndpointRegistry} against the PR1 endpoint set (only {@code + * /_cluster/health}): allow-list resolution, arg validation, count/timeout gating, value coercion, + * and fixed-schema row shaping. Coercion and shaping are exercised through a fake provider that + * returns canned rows, so they stay independent of how any one endpoint fetches; rows are shaped + * with {@link Redactor#NONE} (the no-op redactor). The centralized redactor seam is covered + * separately in {@link RedactorSeamTest}. + */ +class RestEndpointRegistryTest { + + private RestEndpointRegistry registry; + + @BeforeEach + void buildRegistry() { + registry = new RestEndpointRegistry(List.of(new CoreEndpointsProvider())); + } + + private static RestEndpointContext ctx(Map args) { + return RestEndpointContext.of(args, null); + } + + private static RestEndpointRegistry.Endpoint fakeEndpoint( + List schema, Map row) { + RestEndpointProvider provider = + () -> + List.of( + RestEndpointDefinition.builder() + .name("/_test/probe") + .schema(schema) + .handler(context -> List.of(row)) + .build()); + return new RestEndpointRegistry(List.of(provider)).resolve("/_test/probe"); + } + + @Test + void resolveAllowListedEndpoint() { + RestEndpointRegistry.Endpoint endpoint = registry.resolve("/_cluster/health"); + assertEquals("/_cluster/health", endpoint.getPath()); + assertEquals(STRING, endpoint.getSchema().get("status")); + assertEquals(INTEGER, endpoint.getSchema().get("number_of_nodes")); + } + + @Test + void resolveRejectsNonAllowListedEndpoint() { + // A mutating endpoint, and any endpoint deferred out of PR1, is simply absent and refused here. + assertThrows(IllegalArgumentException.class, () -> registry.resolve("/_cluster/reroute")); + assertThrows(IllegalArgumentException.class, () -> registry.resolve("/_cat/nodes")); + assertThrows(IllegalArgumentException.class, () -> registry.resolve("/services/server/info")); + } + + @Test + void resolveRejectsBlankEndpoint() { + IllegalArgumentException emptyEx = + assertThrows(IllegalArgumentException.class, () -> registry.resolve("")); + assertTrue(emptyEx.getMessage().contains("non-empty path")); + assertThrows(IllegalArgumentException.class, () -> registry.resolve(" ")); + assertThrows(IllegalArgumentException.class, () -> registry.resolve(null)); + } + + @Test + void validateRejectsUnknownArg() { + RestSpec spec = new RestSpec("/_cluster/health", Map.of("not_allowed", "x"), null, null); + assertThrows(IllegalArgumentException.class, () -> registry.validate(spec)); + } + + @Test + void validateAcceptsAllowedArg() { + RestSpec spec = new RestSpec("/_cluster/health", Map.of("local", "true"), null, null); + registry.validate(spec); // no throw + } + + @Test + void validateRejectsDroppedLevelArg() { + // level was dropped (no-op against the fixed cluster-level health schema); now unknown. + RestSpec spec = new RestSpec("/_cluster/health", Map.of("level", "indices"), null, null); + IllegalArgumentException ex = + assertThrows(IllegalArgumentException.class, () -> registry.validate(spec)); + assertTrue(ex.getMessage().contains("does not accept arg")); + } + + @Test + void validateRejectsBadArgValue() { + IllegalArgumentException local = + assertThrows( + IllegalArgumentException.class, + () -> + registry.validate( + new RestSpec("/_cluster/health", Map.of("local", "maybe"), null, null))); + assertTrue(local.getMessage().contains("unsupported value")); + } + + @Test + void validateRejectsNegativeCount() { + RestSpec spec = new RestSpec("/_cluster/health", Map.of(), -1, null); + IllegalArgumentException ex = + assertThrows(IllegalArgumentException.class, () -> registry.validate(spec)); + assertTrue(ex.getMessage().contains("non-negative")); + } + + @Test + void validateAcceptsZeroCount() { + RestSpec spec = new RestSpec("/_cluster/health", Map.of(), 0, null); + registry.validate(spec); // no throw: 0 is a valid limit + } + + @Test + void validateRejectsTimeoutArg() { + RestSpec spec = new RestSpec("/_cluster/health", Map.of(), null, "5s"); + IllegalArgumentException ex = + assertThrows(IllegalArgumentException.class, () -> registry.validate(spec)); + assertTrue(ex.getMessage().contains("timeout")); + } + + @Test + void coerceParsesNumericStringValues() { + // The transport/JSON layer may return numeric columns as strings; coerce must parse them. + RestEndpointRegistry.Endpoint endpoint = + fakeEndpoint( + List.of( + Column.of("status", ColumnType.STRING), + Column.of("number_of_nodes", ColumnType.INTEGER)), + Map.of("status", "green", "number_of_nodes", "3")); + List rows = endpoint.toRows(ctx(Map.of()), Redactor.NONE); + assertEquals(3, rows.get(0).tupleValue().get("number_of_nodes").integerValue()); + } + + @Test + void coerceThrowsClearErrorOnUncoercibleValue() { + // A non-numeric value for an INTEGER column must surface a clear client error (HTTP 400), + // not a raw ClassCastException / NumberFormatException (HTTP 500). + RestEndpointRegistry.Endpoint endpoint = + fakeEndpoint( + List.of( + Column.of("status", ColumnType.STRING), + Column.of("number_of_nodes", ColumnType.INTEGER)), + Map.of("status", "green", "number_of_nodes", "not-a-number")); + IllegalArgumentException ex = + assertThrows( + IllegalArgumentException.class, () -> endpoint.toRows(ctx(Map.of()), Redactor.NONE)); + assertTrue(ex.getMessage().contains("number_of_nodes")); + assertTrue(ex.getMessage().contains("not-a-number")); + } + + @Test + void rowsAreShapedToFixedSchema() { + Map raw = new LinkedHashMap<>(); + raw.put("cluster_name", "test-cluster"); + raw.put("status", "green"); + raw.put("number_of_nodes", 1); + RestEndpointRegistry.Endpoint endpoint = + fakeEndpoint( + List.of( + Column.of("cluster_name", ColumnType.STRING), + Column.of("status", ColumnType.STRING), + Column.of("number_of_nodes", ColumnType.INTEGER), + Column.of("relocating_shards", ColumnType.INTEGER)), + raw); + List rows = endpoint.toRows(ctx(Map.of()), Redactor.NONE); + assertEquals(1, rows.size()); + assertEquals("green", rows.get(0).tupleValue().get("status").stringValue()); + assertEquals(1, rows.get(0).tupleValue().get("number_of_nodes").integerValue()); + // a declared column the handler did not return becomes null, never absent. + assertTrue(rows.get(0).tupleValue().get("relocating_shards").isNull()); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndexTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogTableTest.java similarity index 76% rename from opensearch/src/test/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndexTest.java rename to opensearch/src/test/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogTableTest.java index 0b0aa1ec521..df81225afbe 100644 --- a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndexTest.java +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/system/OpenSearchCatalogTableTest.java @@ -32,8 +32,12 @@ import org.opensearch.sql.planner.physical.ProjectOperator; import org.opensearch.sql.storage.Table; +/** + * Covers the generic {@link OpenSearchCatalogTable} through a {@link SystemIndexCatalogSource}: + * schema delegation, the predefined-table contract, and the V2 physical path. + */ @ExtendWith(MockitoExtension.class) -class OpenSearchSystemIndexTest { +class OpenSearchCatalogTableTest { @Mock private OpenSearchClient client; @@ -41,36 +45,37 @@ class OpenSearchSystemIndexTest { @Mock private Settings settings; + private OpenSearchCatalogTable systemTable(String name) { + return new OpenSearchCatalogTable(new SystemIndexCatalogSource(client, name), settings); + } + @Test void testGetFieldTypesOfMetaTable() { - OpenSearchSystemIndex systemIndex = new OpenSearchSystemIndex(client, settings, TABLE_INFO); - final Map fieldTypes = systemIndex.getFieldTypes(); + final Map fieldTypes = systemTable(TABLE_INFO).getFieldTypes(); assertThat(fieldTypes, anyOf(hasEntry("TABLE_CAT", STRING))); } @Test void testGetFieldTypesOfMappingTable() { - OpenSearchSystemIndex systemIndex = - new OpenSearchSystemIndex(client, settings, mappingTable("test_index")); - final Map fieldTypes = systemIndex.getFieldTypes(); + final Map fieldTypes = + systemTable(mappingTable("test_index")).getFieldTypes(); assertThat(fieldTypes, anyOf(hasEntry("COLUMN_NAME", STRING))); } @Test void testIsExist() { - Table systemIndex = new OpenSearchSystemIndex(client, settings, TABLE_INFO); - assertTrue(systemIndex.exists()); + assertTrue(systemTable(TABLE_INFO).exists()); } @Test void testCreateTable() { - Table systemIndex = new OpenSearchSystemIndex(client, settings, TABLE_INFO); + Table systemIndex = systemTable(TABLE_INFO); assertThrows(UnsupportedOperationException.class, () -> systemIndex.create(ImmutableMap.of())); } @Test void implement() { - OpenSearchSystemIndex systemIndex = new OpenSearchSystemIndex(client, settings, TABLE_INFO); + OpenSearchCatalogTable systemIndex = systemTable(TABLE_INFO); NamedExpression projectExpr = named("TABLE_NAME", ref("TABLE_NAME", STRING)); final PhysicalPlan plan = 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 e1278aa75e8..a62432bbd5f 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,10 @@ import org.opensearch.sql.opensearch.client.OpenSearchNodeClient; import org.opensearch.sql.opensearch.setting.OpenSearchSettings; import org.opensearch.sql.opensearch.storage.OpenSearchDataSourceFactory; +import org.opensearch.sql.opensearch.storage.rest.CoreEndpointsProvider; +import org.opensearch.sql.opensearch.storage.rest.RedactorHolder; +import org.opensearch.sql.opensearch.storage.rest.RestEndpointRegistry; +import org.opensearch.sql.opensearch.storage.rest.RestEndpointRegistryHolder; import org.opensearch.sql.opensearch.storage.script.CompoundedScriptEngine; import org.opensearch.sql.plugin.config.EngineExtensionsHolder; import org.opensearch.sql.plugin.config.OpenSearchPluginModule; @@ -135,6 +139,8 @@ import org.opensearch.sql.spark.transport.model.CancelAsyncQueryActionResponse; import org.opensearch.sql.spark.transport.model.CreateAsyncQueryActionResponse; import org.opensearch.sql.spark.transport.model.GetAsyncQueryResultActionResponse; +import org.opensearch.sql.spi.rest.Redactor; +import org.opensearch.sql.spi.rest.RestEndpointProvider; import org.opensearch.sql.sql.domain.SQLQueryRequest; import org.opensearch.sql.storage.DataSourceFactory; import org.opensearch.threadpool.ExecutorBuilder; @@ -154,6 +160,7 @@ public class SQLPlugin extends Plugin private static final Logger LOGGER = LogManager.getLogger(SQLPlugin.class); private List executionEngineExtensions = List.of(); + private List restEndpointProviders = List.of(); private ClusterService clusterService; /** Settings should be inited when bootstrap the plugin. */ @@ -182,6 +189,17 @@ public void loadExtensions(ExtensionLoader loader) { executionEngineExtensions.size(), executionEngineExtensions.stream().map(e -> e.getClass().getSimpleName()).toList()); } + + List restProviders = loader.loadExtensions(RestEndpointProvider.class); + this.restEndpointProviders = restProviders != null ? List.copyOf(restProviders) : List.of(); + } + + private void publishRestCommandRegistries() { + List providers = new ArrayList<>(); + providers.add(new CoreEndpointsProvider()); + providers.addAll(this.restEndpointProviders); + RestEndpointRegistryHolder.set(new RestEndpointRegistry(providers)); + RedactorHolder.set(Redactor.NONE); } @Override @@ -372,6 +390,9 @@ public Collection createComponents( this.clusterService = clusterService; this.pluginSettings = new OpenSearchSettings(clusterService.getClusterSettings()); this.client = (NodeClient) client; + + publishRestCommandRegistries(); + this.dataSourceService = createDataSourceService(); dataSourceService.createDataSource(defaultOpenSearchDataSourceMetadata()); LocalClusterState.state().setClusterService(clusterService); diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java b/plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java index f8214096e41..5685d539541 100644 --- a/plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java +++ b/plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java @@ -107,13 +107,14 @@ public boolean isAnalyticsIndex(String query, QueryType queryType) { .equals( IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_VALUE_SETTING.get( clusterService.getSettings()))) { - // Analytics engine can't serve system catalog; SHOW/DESCRIBE fall back to default pipeline + // Analytics engine serves neither the system catalog nor the rest command's reserved + // in-cluster source; both fall back to the default (Calcite) pipeline. try (UnifiedQueryContext context = buildParsingContext(queryType)) { - boolean systemCatalog = + boolean defaultPipeline = extractIndexName(query, queryType, context) - .map(RestUnifiedQueryAction::isSystemCatalog) + .map(name -> isSystemCatalog(name) || SystemIndexUtils.isRestSource(name)) .orElse(false); - return !systemCatalog; + return !defaultPipeline; } catch (Exception e) { // Check legacy-syntax SHOW/DESCRIBE; otherwise let AE handle and surface the error. return !isLegacySystemCatalogQuery(query); diff --git a/plugin/src/test/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryActionTest.java b/plugin/src/test/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryActionTest.java index 111597bb587..0cf87f0604e 100644 --- a/plugin/src/test/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryActionTest.java +++ b/plugin/src/test/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryActionTest.java @@ -178,6 +178,12 @@ public void describeStatementNotRoutedToAnalyticsEngineUnderClusterComposite() { assertFalse(action.isAnalyticsIndex("DESCRIBE TABLES LIKE 'parquet_logs'", QueryType.SQL)); } + @Test + public void restCommandNotRoutedToAnalyticsEngineUnderClusterComposite() { + enableClusterComposite(); + assertFalse(action.isAnalyticsIndex("| rest '/_cluster/health'", QueryType.PPL)); + } + @Test public void dataQueryStillRoutesToAnalyticsUnderClusterComposite() { enableClusterComposite(); diff --git a/ppl-rest-spi/README.md b/ppl-rest-spi/README.md new file mode 100644 index 00000000000..f51dab19f78 --- /dev/null +++ b/ppl-rest-spi/README.md @@ -0,0 +1,80 @@ +# ppl-rest-spi + +The extension point for the PPL `rest` command. Any OpenSearch plugin can contribute read-only, +fixed-schema management endpoints that become queryable as `rest ''`, without touching the +PPL grammar and without depending on the sql plugin's internals. The built-in `/_cluster/health` +endpoint is just one client of this same contract. + +This module is intentionally thin: it depends only on OpenSearch core, and the row values it +exchanges are plain `java.lang` types (String / Number / Boolean / nested Map), so there is no +cross-classloader type-identity problem. + +## Contract + +| Type | Role | +|---|---| +| `RestEndpointProvider` | Your entry point: `List getEndpoints()`. | +| `RestEndpointDefinition` | One endpoint: `name()`, `schema()`, `argSpec()`, `handler()`. Build with `RestEndpointDefinition.builder()`. | +| `Column` / `ColumnType` | A fixed output column. `ColumnType` is `STRING`, `INTEGER`, `LONG`, `DOUBLE`, or `BOOLEAN`. The schema is pinned before execution. | +| `ArgSpec` | The query args the endpoint accepts and each arg's allowed value domain; an unknown or out-of-domain arg is rejected. `ArgSpec.NONE` accepts none. | +| `RestEndpointHandler` | `List> fetch(RestEndpointContext)`, run at scan execution (not planning), so the scan is lazy and `EXPLAIN` is side-effect free. | +| `RestEndpointContext` | The validated `args()` plus an optional core `NodeClient` (`client()`) for a handler that issues its own read-only transport action. | + +## Add an endpoint + +1. Depend on this module `compileOnly`, and declare the sql plugin as an extended plugin so + `loadExtensions` discovers your provider: + + ```gradle + dependencies { compileOnly project(':ppl-rest-spi') } // or the published artifact + opensearchplugin { extendedPlugins = ['opensearch-sql'] } + ``` + +2. Implement `RestEndpointProvider`: + + ```java + public final class MyRestProvider implements RestEndpointProvider { + @Override + public List getEndpoints() { + return List.of( + RestEndpointDefinition.builder() + .name("/_my/thing") + .schema(List.of( + Column.of("name", ColumnType.STRING), + Column.of("count", ColumnType.LONG))) + .argSpec(ArgSpec.builder().arg("verbose", Set.of("true", "false")).build()) + .handler(MyRestProvider::fetch) + .build()); + } + + private static List> fetch(RestEndpointContext ctx) { + // ctx.args() is already validated; ctx.client() is a NodeClient for transport calls (may be null). + return List.of(Map.of("name", "example", "count", 1L)); + } + } + ``` + +3. Register the provider as a service in + `META-INF/services/org.opensearch.sql.spi.rest.RestEndpointProvider`, containing your class's + fully-qualified name. + +4. Enable the endpoint on the cluster. `plugins.ppl.rest.allowed_endpoints` is a node-level setting + listing the endpoint names a deployment permits; it defaults to `["/_cluster/health"]`. Add your + name to run it: + + ```yaml + plugins.ppl.rest.allowed_endpoints: ["/_cluster/health", "/_my/thing"] + ``` + +Then `rest '/_my/thing' verbose=true | where count > 0 | stats sum(count)` composes like any scan. + +## Rules and guarantees + +- Endpoints are read-only. The handler produces rows; the sql side coerces each value to the + declared `ColumnType` and rejects an uncoercible value with a clear error. +- Endpoint names are global. If two providers register the same name, the first registration wins + and the duplicate is ignored with a logged warning. +- An endpoint must be both registered by a provider and present in `allowed_endpoints`; anything + else is rejected before any transport call. +- Redaction of sensitive values is applied centrally at the row-shaping choke point; a provider + does not implement it. diff --git a/ppl-rest-spi/build.gradle b/ppl-rest-spi/build.gradle new file mode 100644 index 00000000000..ce91f160f7a --- /dev/null +++ b/ppl-rest-spi/build.gradle @@ -0,0 +1,60 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +/* + * The `rest` command extension SPI. + * + * This module is intentionally thin: it declares ONLY the interfaces a plugin implements to + * contribute read-only `rest` endpoints, plus the JDK-typed row/column/arg value objects those + * interfaces exchange. It depends on the OpenSearch core artifact ONLY (for the transport client + * handed to a handler) and on NO sql module, so external plugins can compileOnly it without a + * dependency cycle and without a cross-classloader type-identity problem (row values are plain + * java.lang types). + */ + +plugins { + id 'java-library' + id "io.freefair.lombok" + id 'com.diffplug.spotless' +} + +dependencies { + // Core only. Deliberately no `project(':core')` / `project(':opensearch')` dependency so the + // sql plugin (which depends on this module) never forms a cycle, and so an external plugin can + // depend on this module alone. + compileOnly group: 'org.opensearch', name: 'opensearch', version: "${opensearch_version}" + + testImplementation group: 'org.opensearch', name: 'opensearch', version: "${opensearch_version}" + testImplementation('org.junit.jupiter:junit-jupiter-api:5.9.3') + testImplementation('org.junit.jupiter:junit-jupiter-params:5.9.3') + testRuntimeOnly('org.junit.jupiter:junit-jupiter-engine') + testRuntimeOnly('org.junit.platform:junit-platform-launcher') +} + +test { + useJUnitPlatform() + testLogging { + events "passed", "skipped", "failed" + exceptionFormat "full" + } +} + +spotless { + java { + target fileTree('.') { + include '**/*.java' + exclude '**/build/**', '**/build-*/**' + } + importOrder() + licenseHeader("/*\n" + + " * Copyright OpenSearch Contributors\n" + + " * SPDX-License-Identifier: Apache-2.0\n" + + " */\n\n") + removeUnusedImports() + trimTrailingWhitespace() + endWithNewline() + googleJavaFormat('1.32.0').reflowLongStrings().groupArtifact('com.google.googlejavaformat:google-java-format') + } +} diff --git a/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/ArgSpec.java b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/ArgSpec.java new file mode 100644 index 00000000000..003c56cca6a --- /dev/null +++ b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/ArgSpec.java @@ -0,0 +1,83 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.spi.rest; + +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** + * The query args a {@code rest} endpoint accepts, with the allowed value domain of each. An + * undeclared arg is rejected; a declared arg whose value is outside its domain is rejected; an + * empty domain accepts any value. + */ +public final class ArgSpec { + + public static final ArgSpec NONE = builder().build(); + + // arg name -> allowed values (empty set means "any value allowed"). + private final Map> valueDomains; + + private ArgSpec(Map> valueDomains) { + this.valueDomains = valueDomains; + } + + public Set allowedArgs() { + return valueDomains.keySet(); + } + + public boolean allows(String arg) { + return valueDomains.containsKey(arg); + } + + /** Validate an accepted arg's value against its domain (no-op if the domain is empty). */ + public void validateValue(String endpoint, String arg, String value) { + Set domain = valueDomains.get(arg); + if (domain == null || domain.isEmpty()) { + return; + } + if (value == null || !domain.contains(value.toLowerCase(Locale.ROOT))) { + throw unsupported(endpoint, arg, value, domain); + } + } + + private static IllegalArgumentException unsupported( + String endpoint, String arg, String value, Set domain) { + return new IllegalArgumentException( + "rest endpoint [" + + endpoint + + "] arg [" + + arg + + "] has an unsupported value [" + + value + + "]. Allowed values: " + + domain); + } + + public static Builder builder() { + return new Builder(); + } + + /** Builder for {@link ArgSpec}. Declaration order is preserved for stable error messages. */ + public static final class Builder { + private final LinkedHashMap> valueDomains = new LinkedHashMap<>(); + + public Builder arg(String name) { + valueDomains.put(name, Set.of()); + return this; + } + + public Builder arg(String name, Set domain) { + valueDomains.put(name, Set.copyOf(domain)); + return this; + } + + public ArgSpec build() { + return new ArgSpec(new LinkedHashMap<>(valueDomains)); + } + } +} diff --git a/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/Column.java b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/Column.java new file mode 100644 index 00000000000..da91ed6ab0c --- /dev/null +++ b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/Column.java @@ -0,0 +1,17 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.spi.rest; + +/** + * One column of a {@code rest} endpoint's fixed output schema: a name and a {@link ColumnType}. + * Column order is output-field order. + */ +public record Column(String name, ColumnType type) { + + public static Column of(String name, ColumnType type) { + return new Column(name, type); + } +} diff --git a/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/ColumnType.java b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/ColumnType.java new file mode 100644 index 00000000000..b3f0cf7c2cf --- /dev/null +++ b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/ColumnType.java @@ -0,0 +1,18 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.spi.rest; + +/** + * The output type of a {@link Column}. A small, dependency-free enum so the SPI does not leak the + * sql engine's {@code ExprType} across the module boundary. + */ +public enum ColumnType { + STRING, + INTEGER, + LONG, + DOUBLE, + BOOLEAN +} diff --git a/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/Redactor.java b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/Redactor.java new file mode 100644 index 00000000000..6773a9ed416 --- /dev/null +++ b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/Redactor.java @@ -0,0 +1,31 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.spi.rest; + +import java.util.Map; + +/** + * The single redaction hook for the {@code rest} command: given one raw response row (column name + * to value) for an endpoint, it returns the row to emit with sensitive values masked in one pass. + * Not a {@code loadExtensions} SPI (one central owner, not an open registration surface); the + * default is {@link #NONE} (no-op), applied once per row at the choke point before values are + * coerced to the schema types. + */ +@FunctionalInterface +public interface Redactor { + + /** A no-op that returns the row unchanged (the OSS default). */ + Redactor NONE = (endpoint, row) -> row; + + /** + * Redact one raw response row. + * + * @param endpoint the endpoint name the row came from, a scope for endpoint-aware masking + * @param row column name to raw value for one row (never null) + * @return the row to emit, with sensitive values masked + */ + Map redact(String endpoint, Map row); +} diff --git a/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointContext.java b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointContext.java new file mode 100644 index 00000000000..3cf78911deb --- /dev/null +++ b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointContext.java @@ -0,0 +1,38 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.spi.rest; + +import java.util.Map; +import org.opensearch.transport.client.node.NodeClient; + +/** + * Per-invocation context handed to {@link RestEndpointHandler#fetch} at execution time (scan open), + * carrying the validated query args and the node transport client a provider may use for its own + * read-only transport action. A provider that already holds a client may ignore {@link #client()}. + */ +public interface RestEndpointContext { + + /** Validated query args for this invocation (never null; empty when none supplied). */ + Map args(); + + /** Node transport client for a provider that issues its own transport action; may be null. */ + NodeClient client(); + + static RestEndpointContext of(Map args, NodeClient client) { + Map safeArgs = args == null ? Map.of() : args; + return new RestEndpointContext() { + @Override + public Map args() { + return safeArgs; + } + + @Override + public NodeClient client() { + return client; + } + }; + } +} diff --git a/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointDefinition.java b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointDefinition.java new file mode 100644 index 00000000000..a36262db9a1 --- /dev/null +++ b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointDefinition.java @@ -0,0 +1,87 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.spi.rest; + +import java.util.List; +import java.util.Objects; + +/** + * One read-only {@code rest} endpoint from a {@link RestEndpointProvider}: a unique name (the token + * after {@code rest}, e.g. {@code /_cluster/health}), a fixed {@link Column} schema, the {@link + * ArgSpec} it accepts, and the {@link RestEndpointHandler} that produces its rows. Immutable; build + * with {@link #builder()}. + */ +public interface RestEndpointDefinition { + + String name(); + + List schema(); + + ArgSpec argSpec(); + + RestEndpointHandler handler(); + + static Builder builder() { + return new Builder(); + } + + final class Builder { + private String name; + private List schema = List.of(); + private ArgSpec argSpec = ArgSpec.NONE; + private RestEndpointHandler handler; + + private Builder() {} + + public Builder name(String name) { + this.name = name; + return this; + } + + public Builder schema(List schema) { + this.schema = schema; + return this; + } + + public Builder argSpec(ArgSpec argSpec) { + this.argSpec = argSpec; + return this; + } + + public Builder handler(RestEndpointHandler handler) { + this.handler = handler; + return this; + } + + public RestEndpointDefinition build() { + String endpointName = Objects.requireNonNull(name, "rest endpoint name is required"); + List cols = List.copyOf(Objects.requireNonNull(schema, "schema is required")); + ArgSpec spec = Objects.requireNonNull(argSpec, "argSpec is required"); + RestEndpointHandler endpointHandler = Objects.requireNonNull(handler, "handler is required"); + return new RestEndpointDefinition() { + @Override + public String name() { + return endpointName; + } + + @Override + public List schema() { + return cols; + } + + @Override + public ArgSpec argSpec() { + return spec; + } + + @Override + public RestEndpointHandler handler() { + return endpointHandler; + } + }; + } + } +} diff --git a/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointHandler.java b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointHandler.java new file mode 100644 index 00000000000..ca59fab727c --- /dev/null +++ b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointHandler.java @@ -0,0 +1,28 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.spi.rest; + +import java.util.List; +import java.util.Map; + +/** + * Produces the raw rows of a {@code rest} endpoint. Invoked at execution time (scan open), never at + * planning, so the scan stays lazy and EXPLAIN is side-effect free; a transport-backed provider may + * block on {@code ctx.client().execute(...).actionGet()} here. Each row is a map of output-column + * name to a plain {@code java.lang} value; the sql side coerces each to the declared {@link Column} + * type. + */ +@FunctionalInterface +public interface RestEndpointHandler { + + /** + * Fetch the rows for one invocation. + * + * @param ctx the validated query args and (optional) transport client for this invocation + * @return one map of column name to value per row (never null; empty when there are no rows) + */ + List> fetch(RestEndpointContext ctx); +} diff --git a/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointProvider.java b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointProvider.java new file mode 100644 index 00000000000..db04c016f8f --- /dev/null +++ b/ppl-rest-spi/src/main/java/org/opensearch/sql/spi/rest/RestEndpointProvider.java @@ -0,0 +1,21 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.spi.rest; + +import java.util.List; + +/** + * The extension point a plugin implements to contribute read-only {@code rest} endpoints, via + * {@code ExtensiblePlugin.loadExtensions(RestEndpointProvider.class)}. The sql plugin merges every + * discovered provider with its own built-in one into a single registry, so built-in and external + * endpoints are uniform clients of the same contract. A provider declares data (name, schema, args) + * and a handler; it never touches the PPL grammar. + */ +public interface RestEndpointProvider { + + /** The endpoints this provider contributes. Called once when the registry is built. */ + List getEndpoints(); +} diff --git a/ppl/src/main/antlr/OpenSearchPPLLexer.g4 b/ppl/src/main/antlr/OpenSearchPPLLexer.g4 index fb072ae134f..6f21942c9c2 100644 --- a/ppl/src/main/antlr/OpenSearchPPLLexer.g4 +++ b/ppl/src/main/antlr/OpenSearchPPLLexer.g4 @@ -13,6 +13,8 @@ options { caseInsensitive = true; } SEARCH: 'SEARCH'; DESCRIBE: 'DESCRIBE'; SHOW: 'SHOW'; +REST: 'REST'; +TIMEOUT: 'TIMEOUT'; MAKERESULTS: 'MAKERESULTS'; FORMAT: 'FORMAT'; CSV: 'CSV'; diff --git a/ppl/src/main/antlr/OpenSearchPPLParser.g4 b/ppl/src/main/antlr/OpenSearchPPLParser.g4 index a23a314537d..7ad26e09850 100644 --- a/ppl/src/main/antlr/OpenSearchPPLParser.g4 +++ b/ppl/src/main/antlr/OpenSearchPPLParser.g4 @@ -45,6 +45,7 @@ subSearch // commands pplCommands : describeCommand + | restCommand | showDataSourcesCommand | makeresultsCommand | searchCommand @@ -107,6 +108,7 @@ commands commandName : SEARCH | DESCRIBE + | REST | SHOW | WHERE | FIELDS @@ -214,6 +216,16 @@ describeCommand : DESCRIBE tableSourceClause ; + +restCommand + : REST stringLiteral (restArgument)* + ; + +restArgument + : COUNT EQUAL integerLiteral + | TIMEOUT EQUAL stringLiteral + | ident EQUAL literalValue + ; showDataSourcesCommand : SHOW DATASOURCES ; @@ -1871,5 +1883,7 @@ searchableKeyWord | MAX_DEPTH | DEPTH_FIELD | EDGE + // rest command token, also usable as a free-text search term / identifier + | TIMEOUT | SEP ; diff --git a/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java b/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java index efdbf26a205..7aca48a4666 100644 --- a/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java +++ b/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java @@ -114,6 +114,7 @@ import org.opensearch.sql.ast.tree.Rename; import org.opensearch.sql.ast.tree.Replace; import org.opensearch.sql.ast.tree.ReplacePair; +import org.opensearch.sql.ast.tree.RestRelation; import org.opensearch.sql.ast.tree.Reverse; import org.opensearch.sql.ast.tree.Rex; import org.opensearch.sql.ast.tree.SPath; @@ -149,6 +150,7 @@ import org.opensearch.sql.ppl.utils.ArgumentFactory; import org.opensearch.sql.ppl.utils.MakeResultsDataParser; import org.opensearch.sql.ppl.utils.UnresolvedPlanHelper; +import org.opensearch.sql.utils.SystemIndexUtils; /** Class of building the AST. Refines the visit path and build the AST nodes */ public class AstBuilder extends OpenSearchPPLParserBaseVisitor { @@ -260,6 +262,34 @@ public UnresolvedPlan visitShowDataSourcesCommand( return new DescribeRelation(qualifiedName(DATASOURCES_TABLE_NAME)); } + /** + * Rest command.
+ * Encodes the validated endpoint spec into a reserved table name (via {@link + * org.opensearch.sql.utils.SystemIndexUtils#restTable}) that resolves through the storage engine + * to a REST source table on the Calcite path, mirroring how DESCRIBE resolves to a system index. + */ + @Override + public UnresolvedPlan visitRestCommand(OpenSearchPPLParser.RestCommandContext ctx) { + String endpoint = StringUtils.unquoteText(ctx.stringLiteral().getText()); + LinkedHashMap args = new LinkedHashMap<>(); + Integer count = null; + String timeout = null; + for (OpenSearchPPLParser.RestArgumentContext arg : ctx.restArgument()) { + if (arg.COUNT() != null) { + count = Integer.parseInt(arg.integerLiteral().getText()); + } else if (arg.TIMEOUT() != null) { + timeout = StringUtils.unquoteText(arg.stringLiteral().getText()); + } else { + args.put( + StringUtils.unquoteIdentifier(arg.ident().getText()), + StringUtils.unquoteText(arg.literalValue().getText())); + } + } + String token = + SystemIndexUtils.restTable(new SystemIndexUtils.RestSpec(endpoint, args, count, timeout)); + return new RestRelation(new QualifiedName(token)); + } + /** makeresults command. */ @Override public UnresolvedPlan visitMakeresultsCommand(OpenSearchPPLParser.MakeresultsCommandContext ctx) { diff --git a/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java b/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java index 398361fcea7..21692bbda87 100644 --- a/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java +++ b/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java @@ -97,6 +97,7 @@ import org.opensearch.sql.ast.tree.Relation; import org.opensearch.sql.ast.tree.Rename; import org.opensearch.sql.ast.tree.Replace; +import org.opensearch.sql.ast.tree.RestRelation; import org.opensearch.sql.ast.tree.Reverse; import org.opensearch.sql.ast.tree.Rex; import org.opensearch.sql.ast.tree.SPath; @@ -125,6 +126,7 @@ import org.opensearch.sql.planner.logical.LogicalRemove; import org.opensearch.sql.planner.logical.LogicalRename; import org.opensearch.sql.planner.logical.LogicalSort; +import org.opensearch.sql.utils.SystemIndexUtils; /** Utility class to mask sensitive information in incoming PPL queries. */ public class PPLQueryDataAnonymizer extends AbstractNodeVisitor { @@ -168,6 +170,23 @@ public String visitExplain(Explain node, String context) { @Override public String visitRelation(Relation node, String context) { + if (node instanceof RestRelation) { + SystemIndexUtils.RestSpec spec = + SystemIndexUtils.decodeRestSpec(node.getTableQualifiedName().toString()); + StringBuilder sb = new StringBuilder("rest ").append(spec.getEndpoint()); + if (spec.getCount() != null) { + sb.append(" count=").append(MASK_LITERAL); + } + if (spec.getTimeout() != null) { + sb.append(" timeout=").append(MASK_LITERAL); + } + if (spec.getArgs() != null) { + for (String key : spec.getArgs().keySet()) { + sb.append(' ').append(key).append('=').append(MASK_LITERAL); + } + } + return sb.toString(); + } if (node instanceof DescribeRelation) { return StringUtils.format("describe %s", MASK_TABLE); } diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLRestTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLRestTest.java new file mode 100644 index 00000000000..367d1509e4a --- /dev/null +++ b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLRestTest.java @@ -0,0 +1,68 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.ppl.calcite; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; + +import org.junit.Test; +import org.opensearch.sql.ast.Node; +import org.opensearch.sql.ast.tree.Project; +import org.opensearch.sql.ast.tree.RestRelation; +import org.opensearch.sql.common.setting.Settings; +import org.opensearch.sql.ppl.antlr.PPLSyntaxParser; +import org.opensearch.sql.ppl.parser.AstBuilder; +import org.opensearch.sql.utils.SystemIndexUtils; + +/** + * Calcite-path coverage for the {@code rest} leading command at the parse / AST tier. + * + *

The {@code rest} row source resolves through {@code OpenSearchStorageEngine.getTable} -> + * {@code RestCatalogSource} -> {@code CalciteLogicalCatalogScan}, which lives in the {@code + * opensearch} module. This ppl-module Calcite harness binds a Calcite SCOTT schema rather than the + * OpenSearch storage engine, so the optimized {@code CalciteScannableCatalogScan} logical-plan + * assertion is exercised in {@code RestCatalogSourceTest} (source schema + request, unit) and + * {@code CalcitePPLRestIT} (schema + datarows on a live single-node cluster). This test pins the + * Calcite-facing contract that the ppl module owns: the grammar/AST rewrite of {@code rest} into a + * {@code RestRelation} carrying the validated, reserved-name-encoded endpoint spec that rides + * {@code visitRelation} exactly like {@code DESCRIBE}. + */ +public class CalcitePPLRestTest { + + private final PPLSyntaxParser parser = new PPLSyntaxParser(); + private final Settings settings = mock(Settings.class); + + private Node parse(String ppl) { + return new AstBuilder(ppl, settings).visit(parser.parse(ppl)); + } + + @Test + public void restHealthProjectsDeclaredColumns() { + Project project = + (Project) parse("| rest \"/_cluster/health\" | fields status, number_of_nodes"); + RestRelation rest = (RestRelation) project.getChild().get(0); + SystemIndexUtils.RestSpec spec = + SystemIndexUtils.decodeRestSpec(rest.getTableQualifiedName().toString()); + assertEquals("/_cluster/health", spec.getEndpoint()); + // downstream fields compose on top of the rest row source. + assertEquals(2, project.getProjectList().size()); + } + + @Test + public void restReservedNameRoundTrips() { + RestRelation rest = + (RestRelation) + parse("| rest \"/_cluster/health\" count=10 timeout=\"5s\" health=\"green\""); + String reserved = rest.getTableQualifiedName().toString(); + assertTrue(SystemIndexUtils.isRestSource(reserved)); + SystemIndexUtils.RestSpec spec = SystemIndexUtils.decodeRestSpec(reserved); + assertEquals("/_cluster/health", spec.getEndpoint()); + assertEquals(Integer.valueOf(10), spec.getCount()); + assertEquals("5s", spec.getTimeout()); + assertEquals("green", spec.getArgs().get("health")); + } +} diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java index 0650b4d1e4c..b16b44f1c92 100644 --- a/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java +++ b/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java @@ -2019,4 +2019,32 @@ public void testJoinNoPrefixComparisonStaysCondition() { public void testJoinPrefixWithoutCriteriaKeywordIsSyntaxError() { assertThrows(SyntaxCheckException.class, () -> plan("source=t1 | inner join a t2")); } + + // rest command tests + + @Test + public void testRestCommand() { + org.opensearch.sql.ast.tree.Project project = + (org.opensearch.sql.ast.tree.Project) + plan("| rest \"/_cluster/health\" | fields status, number_of_nodes"); + org.opensearch.sql.ast.tree.RestRelation rest = + (org.opensearch.sql.ast.tree.RestRelation) project.getChild().get(0); + SystemIndexUtils.RestSpec spec = + SystemIndexUtils.decodeRestSpec(rest.getTableQualifiedName().toString()); + assertEquals("/_cluster/health", spec.getEndpoint()); + assertTrue(spec.getArgs().isEmpty()); + } + + @Test + public void testRestCommandWithArgs() { + org.opensearch.sql.ast.tree.RestRelation rest = + (org.opensearch.sql.ast.tree.RestRelation) + plan("| rest \"/_cluster/health\" count=5 timeout=\"30s\" level=\"indices\""); + SystemIndexUtils.RestSpec spec = + SystemIndexUtils.decodeRestSpec(rest.getTableQualifiedName().toString()); + assertEquals("/_cluster/health", spec.getEndpoint()); + assertEquals(Integer.valueOf(5), spec.getCount()); + assertEquals("30s", spec.getTimeout()); + assertEquals("indices", spec.getArgs().get("level")); + } } diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java index 63f7dea81d7..723d2859891 100644 --- a/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java +++ b/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java @@ -53,6 +53,18 @@ public void testPrometheusPPLCommand() { assertEquals("source=table", anonymize("source=prometheus.http_requests_process")); } + @Test + public void testRestCommand() { + assertEquals("rest /_cluster/health", anonymize("| rest \"/_cluster/health\"")); + } + + @Test + public void testRestCommandMasksArgValues() { + assertEquals( + "rest /_cluster/health count=*** timeout=*** level=***", + anonymize("| rest \"/_cluster/health\" count=5 timeout=\"30s\" level=\"indices\"")); + } + @Test public void testWhereCommand() { assertEquals("source=table | where identifier = ***", anonymize("search source=t | where a=1")); diff --git a/settings.gradle b/settings.gradle index 7fc93fe1725..7546d0093cf 100644 --- a/settings.gradle +++ b/settings.gradle @@ -17,6 +17,7 @@ include 'opensearch-sql-plugin' project(':opensearch-sql-plugin').projectDir = file('plugin') include 'api' include 'ppl' +include 'ppl-rest-spi' include 'common' include 'opensearch' include 'core'