feat(ppl): Report completed PPL queries to Query Insights Top N

docs/user/ppl/admin/settings.md

@@ -472,4 +472,36 @@ Expected output:
   "transient": {}
 }
 ```
-
+
+
+## search.insights.top_queries.ppl.enabled
+
+### Version
+
+3.10
+
+### Description
+
+This setting controls whether completed PPL queries are recorded into Query Insights Top N. When enabled, the PPL plugin reports each finished PPL query (with its query text, latency, CPU, memory, and — on a security-enabled cluster — the requesting user) to the Query Insights plugin, and tags the child DSL searches a PPL query spawns so they appear as sub-queries.
+
+1. The default value is false (opt-in).
+2. This setting is node scope, matching the rest of the `search.insights.top_queries.*` family.
+3. It is dynamic: update it through the cluster settings API (as in the example below), and the value applies per node.
+
+Note on query text and user identity: the record stores the query as it was submitted, including any literal values in it, the same way Query Insights stores the source of a DSL search. When a security plugin is installed, the record also includes the requesting user (username, roles, backend roles). Who can see these in Top N results is governed by the Query Insights plugin's own RBAC filter, `search.insights.top_queries.filter_by_mode` — default `none` (no per-user filtering), `username` (a user sees only their own queries; admins see all), or `backend_roles`. Set it to `username` or `backend_roles` if other users' queries and identities should not be visible to every Top N reader.
+
+Note on CPU and memory: the record reads them from the PPL task's resource stats (the same values `_tasks` shows), which OpenSearch records only while the cluster setting `task_resource_tracking.enabled` is on (the default). With it off, a PPL record reports 0 CPU and 0 memory.
+
+Note: This setting is **registered by the Query Insights plugin** (it belongs to the `search.insights.top_queries.*` family), not by the PPL plugin. The PPL plugin reads its value to decide whether to report. It therefore only takes effect when the Query Insights plugin is installed; when Query Insights is absent the setting is unregistered and PPL reporting is a no-op.
+
+### Example
+
+With the Query Insights plugin installed, you can enable PPL recording like this:
+
+```shell
+curl -sS -H 'Content-Type: application/json' \
+-X PUT localhost:9200/_cluster/settings \
+-d '{"persistent" : {"search.insights.top_queries.ppl.enabled" : "true"}}'
+```
+
+This example is not executed as part of the documentation tests because the setting is registered by the Query Insights plugin, which is not installed in the PPL documentation test cluster.

integ-test/build.gradle

@@ -205,6 +205,11 @@ configurations.all {
 
 configurations {
     zipArchive
+    // Resolved only by integTestWithQueryInsights, so no other task depends on a Query Insights
+    // snapshot being published.
+    queryInsightsZip {
+        transitive = false
+    }
 }
 
 dependencies {
@@ -238,6 +243,9 @@ dependencies {
     // For GeoIP PPL functions
     zipArchive group: 'org.opensearch.plugin', name:'geospatial', version: "${opensearch_build}"
 
+    // Installed alongside the plugin by integTestWithQueryInsights
+    queryInsightsZip group: 'org.opensearch.plugin', name:'query-insights', version: "${opensearch_build}"
+
 }
 
 java {
@@ -281,6 +289,17 @@ def getGeoSpatialPlugin() {
     }
 }
 
+// The published Query Insights snapshot, or a local build via -PqueryInsightsZip=/path/to.zip.
+def getQueryInsightsPlugin() {
+    provider { (RegularFile) (() ->
+        project.findProperty('queryInsightsZip')
+            ? file(project.findProperty('queryInsightsZip'))
+            : configurations.queryInsightsZip.asFileTree.matching {
+                include '**/query-insights*'
+            }.singleFile )
+    }
+}
+
 // fetch from the feature-build artifact for now (linux/x64 only; for local dev pass -PanalyticsEngineZip=/path instead).
 ext.pluginVersion = opensearch_version.tokenize('-')[0]
 ext.featureBuildBase = "https://ci.opensearch.org/ci/dbc/feature-build-opensearch/feature-datafusion/latest/linux/x64/tar/builds/opensearch/plugins"
@@ -548,6 +567,23 @@ testClusters {
         plugin(getJobSchedulerPlugin())
         plugin ":opensearch-sql-plugin"
     }
+    // Query Insights installed, so the PPL -> Query Insights reporting path actually runs.
+    integTestWithQueryInsights {
+        testDistribution = 'archive'
+        plugin(getJobSchedulerPlugin())
+        plugin(getQueryInsightsPlugin())
+        plugin ":opensearch-sql-plugin"
+        setting "plugins.query.datasources.encryption.masterkey", "1234567812345678"
+    }
+    // The same, plus the security plugin (added by configureSecurityPlugin below), so records carry
+    // the requesting user and Top N reads go through Query Insights' RBAC filter.
+    integTestWithQueryInsightsAndSecurity {
+        testDistribution = 'archive'
+        plugin(getJobSchedulerPlugin())
+        plugin(getQueryInsightsPlugin())
+        plugin ":opensearch-sql-plugin"
+        setting "plugins.query.datasources.encryption.masterkey", "1234567812345678"
+    }
     // Smoke test: verify sql loads cleanly alongside analytics-engine.
     analyticsEngineCompat {
         testDistribution = 'archive'
@@ -803,6 +839,58 @@ task asyncMultiNodeIT(type: RestIntegTestTask) {
 // (needs a 2-node cluster) and would never actually run.
 check.dependsOn asyncMultiNodeIT
 
+// Asserts that PPL queries are recorded into Query Insights. Not part of `check` yet: the published
+// Query Insights snapshot does not include the PPL consumer until that change merges there, so run
+// it against a local build until then:
+//   ./gradlew :integ-test:integTestWithQueryInsights -PqueryInsightsZip=/path/to/query-insights.zip
+task integTestWithQueryInsights(type: RestIntegTestTask) {
+    useCluster testClusters.integTestWithQueryInsights
+
+    dependsOn ':opensearch-sql-plugin:bundlePlugin'
+
+    systemProperty 'tests.security.manager', 'false'
+    systemProperty 'project.root', project.projectDir.absolutePath
+    systemProperty 'defaultQuerySizeLimit', '10000'
+    systemProperty 'defaultQueryBucketSize', '1000'
+    systemProperty 'defaultMaxResultWindow', '10000'
+    doFirst {
+        systemProperty 'cluster.debug', getDebug()
+    }
+
+    filter {
+        includeTestsMatching 'org.opensearch.sql.ppl.QueryInsightsRecordingIT'
+    }
+}
+
+// The reporting path on a secured cluster: whose identity a record carries and who can read it
+// back. Not part of `check` yet, for the same reason as integTestWithQueryInsights:
+//   ./gradlew :integ-test:integTestWithQueryInsightsAndSecurity -PqueryInsightsZip=/path/to.zip
+task integTestWithQueryInsightsAndSecurity(type: RestIntegTestTask) {
+    useCluster testClusters.integTestWithQueryInsightsAndSecurity
+
+    dependsOn ':opensearch-sql-plugin:bundlePlugin'
+
+    // This task installs the security plugin, so its absence is a breakage rather than a skip.
+    systemProperty 'tests.required.plugins', 'opensearch-security'
+    systemProperty 'tests.security.manager', 'false'
+    systemProperty 'project.root', project.projectDir.absolutePath
+    systemProperty 'defaultQuerySizeLimit', '10000'
+    systemProperty 'defaultQueryBucketSize', '1000'
+    systemProperty 'defaultMaxResultWindow', '10000'
+    doFirst {
+        systemProperty 'cluster.debug', getDebug()
+        systemProperty "https", "false"
+        systemProperty "user", "admin"
+        systemProperty "password", "admin"
+    }
+
+    filter {
+        includeTestsMatching 'org.opensearch.sql.security.QueryInsightsRecordingSecurityIT'
+    }
+}
+
+configureSecurityPlugin(testClusters.integTestWithQueryInsightsAndSecurity)
+
 task integTestWithSecurity(type: RestIntegTestTask) {
     useCluster testClusters.integTestWithSecurity
     useCluster testClusters.remoteIntegTestWithSecurity
@@ -832,6 +920,8 @@ task integTestWithSecurity(type: RestIntegTestTask) {
     // Exclude tests that require the analytics engine plugin stack (run separately via analyticsEngineSecurityIT)
     filter {
         excludeTestsMatching 'org.opensearch.sql.security.AnalyticsEngineSecurityIT'
+        // Needs Query Insights installed; runs via integTestWithQueryInsightsAndSecurity.
+        excludeTestsMatching 'org.opensearch.sql.security.QueryInsightsRecordingSecurityIT'
     }
 
     systemProperty 'tests.security.manager', 'false'
@@ -966,6 +1056,9 @@ integTest {
     // Run on dedicated two-node cluster via :integ-test:asyncMultiNodeIT.
     exclude 'org/opensearch/sql/ppl/AsyncPPLMultiNodeRoutingIT.class'
 
+    // Needs Query Insights installed; runs via :integ-test:integTestWithQueryInsights.
+    exclude 'org/opensearch/sql/ppl/QueryInsightsRecordingIT.class'
+
     // Exclude tests that require the analytics engine plugin stack (run separately via dedicated tasks)
     exclude 'org/opensearch/sql/analytics/AnalyticsEngineProfileIT.class'
 
@@ -1317,6 +1410,10 @@ task integTestRemote(type: RestIntegTestTask) {
             excludeTestsMatching "org.opensearch.sql.bwc.*IT"
         }
 
+        // Need Query Insights installed; run via the integTestWithQueryInsights* tasks.
+        excludeTestsMatching 'org.opensearch.sql.ppl.QueryInsightsRecordingIT'
+        excludeTestsMatching 'org.opensearch.sql.security.QueryInsightsRecordingSecurityIT'
+
         if (analyticsEnabled) {
             // Full-text search (search-filter syntax -> Lucene query_string) needs an inverted-index
             // reader that parquet-backed analytics indices lack, so it's unsupported on this route.

integ-test/src/test/java/org/opensearch/sql/ppl/QueryInsightsRecordingIT.java

@@ -0,0 +1,206 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ppl;
+
+import static org.opensearch.sql.plugin.rest.RestPPLQueryAction.QUERY_API_ENDPOINT;
+import static org.opensearch.sql.util.QueryInsightsRecords.PPL_ENABLED;
+import static org.opensearch.sql.util.QueryInsightsRecords.awaitChildren;
+import static org.opensearch.sql.util.QueryInsightsRecords.indicesOf;
+import static org.opensearch.sql.util.QueryInsightsRecords.isPpl;
+import static org.opensearch.sql.util.QueryInsightsRecords.measurement;
+import static org.opensearch.sql.util.QueryInsightsRecords.pplRecordFor;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
+import org.json.JSONObject;
+import org.junit.Test;
+import org.opensearch.client.Request;
+import org.opensearch.client.RequestOptions;
+import org.opensearch.client.ResponseException;
+import org.opensearch.sql.util.QueryInsightsRecords;
+
+/**
+ * End to end: PPL queries are recorded into Query Insights Top N on a cluster that has Query
+ * Insights installed.
+ *
+ * <p>Runs only in {@code :integ-test:integTestWithQueryInsights}, whose cluster installs Query
+ * Insights; the default integTest cluster does not, and {@link QueryInsightsReportingIT} covers
+ * that case.
+ *
+ * <p>Each scenario reads its own index, so a record is identified by the index its query text
+ * reads. Absence is only asserted after a later query's record has arrived, so a missing record
+ * means it was never sent rather than not yet processed.
+ */
+public class QueryInsightsRecordingIT extends PPLIntegTestCase {
+
+  /** Records are stamped with their query's start, so everything this class records is later. */
+  private static final long CLASS_START_MILLIS = System.currentTimeMillis();
+
+  private static final AtomicInteger BARRIER_SEQ = new AtomicInteger();
+
+  @Override
+  public void init() throws Exception {
+    super.init();
+    enableCalcite(); // join runs on the Calcite path
+    QueryInsightsRecords.assertSupportsPpl(client());
+    QueryInsightsRecords.enableRecording(client());
+  }
+
+  @Test
+  public void successfulQueryIsRecorded() throws Exception {
+    indexWithDocs("qi_it_success");
+    executeQuery("source=qi_it_success | where v > 1 | fields v");
+
+    JSONObject record = awaitPplRecord("qi_it_success");
+    assertFalse(record.getBoolean("failed"));
+    // As written, literal included, the way a DSL search's source is recorded.
+    assertEquals("source=qi_it_success | where v > 1 | fields v", record.getString("source"));
+    assertTrue(
+        "parent marker must be <source>:<nodeId>:<taskId>, was "
+            + record.optString("parent_marker"),
+        record.optString("parent_marker").matches("PPL:[^:]+:\\d+"));
+    assertTrue(record.getLong("timestamp") >= CLASS_START_MILLIS);
+    // The plugin measures its own threads; zero here means the accounting never ran.
+    assertTrue(measurement(record, "cpu") > 0);
+    assertTrue(measurement(record, "memory") > 0);
+  }
+
+  @Test
+  public void queryOnMissingIndexIsRecordedAsFailed() throws Exception {
+    expectThrows(ResponseException.class, () -> executeQuery("source=qi_it_missing | fields v"));
+
+    // Fails in analysis, before any search runs; the query parsed, so it is still recorded.
+    JSONObject record = awaitPplRecord("qi_it_missing");
+    assertTrue(record.getBoolean("failed"));
+  }
+
+  @Test
+  public void queryOnUnknownFieldIsRecordedAsFailed() throws Exception {
+    indexWithDocs("qi_it_bad_field");
+    expectThrows(
+        ResponseException.class,
+        () -> executeQuery("source=qi_it_bad_field | where no_such_field > 1 | fields v"));
+
+    JSONObject record = awaitPplRecord("qi_it_bad_field");
+    assertTrue(record.getBoolean("failed"));
+  }
+
+  @Test
+  public void explainIsNotRecorded() throws Exception {
+    indexWithDocs("qi_it_explain");
+    explainQueryToString("source=qi_it_explain | fields v");
+    // explain is also valid inside the query text, which reaches the execute route.
+    executeQuery("explain source=qi_it_explain | fields v");
+
+    awaitBarrier();
+    assertNoPplRecord("qi_it_explain");
+  }
+
+  @Test
+  public void syntaxErrorIsNotRecorded() throws Exception {
+    expectThrows(ResponseException.class, () -> executeQuery("source=qi_it_syntax | wherr v > 1"));
+
+    awaitBarrier();
+    // A query that never parses is not reported.
+    assertNoPplRecord("qi_it_syntax");
+  }
+
+  @Test
+  public void nothingIsRecordedWhenTheGateIsOff() throws Exception {
+    indexWithDocs("qi_it_gated");
+    QueryInsightsRecords.putPersistentSettings(client(), Map.of(PPL_ENABLED, "false"));
+    try {
+      executeQuery("source=qi_it_gated | where v > 1 | fields v");
+    } finally {
+      QueryInsightsRecords.putPersistentSettings(client(), Map.of(PPL_ENABLED, "true"));
+    }
+
+    awaitBarrier();
+    assertNoPplRecord("qi_it_gated");
+    // The search it ran is still recorded as an ordinary query: with the gate off no parent marker
+    // is stamped, so it is nobody's child and stays in the Top N overview.
+    JSONObject search =
+        awaitRecord(r -> !isPpl(r) && indicesOf(r).contains("qi_it_gated"), "native search");
+    assertFalse(search.optBoolean("is_child", false));
+  }
+
+  @Test
+  public void asyncSubmissionIsNotRecorded() throws Exception {
+    indexWithDocs("qi_it_async");
+    Request request = new Request("POST", QUERY_API_ENDPOINT);
+    request.setJsonEntity(
+        "{\"query\": \"source=qi_it_async | fields v\", \"wait_for_completion_timeout\": \"30s\"}");
+    client().performRequest(request);
+
+    // The query outlives its task on this route, so a record built when the task ends would carry
+    // near-zero latency; the route is left out of Top N.
+    awaitBarrier();
+    assertNoPplRecord("qi_it_async");
+  }
+
+  @Test
+  public void childSearchesCarryTheParentMarker() throws Exception {
+    indexWithDocs("qi_it_join_left");
+    indexWithDocs("qi_it_join_right");
+    executeQuery(
+        "source=qi_it_join_left | join left=l right=r on l.k = r.k qi_it_join_right"
+            + " | fields l.v, r.v");
+
+    JSONObject parent = awaitPplRecord("qi_it_join_left");
+
+    List<JSONObject> children = awaitChildren(client(), parent.getString("parent_marker"), 2);
+    Set<String> childIndices = new HashSet<>();
+    for (JSONObject child : children) {
+      assertTrue(child.getBoolean("is_child"));
+      assertEquals("DSL", child.getString("query_source"));
+      // Stamped with the query's start, the record never sorts after the searches it spawned; a
+      // completion-time stamp could put it in a later Top N window than its children.
+      assertTrue(
+          "parent " + parent.getLong("timestamp") + " after child " + child.getLong("timestamp"),
+          parent.getLong("timestamp") <= child.getLong("timestamp"));
+      childIndices.addAll(indicesOf(child));
+    }
+    // The record names no indices; its searches carry the join's two sides.
+    assertEquals(Set.of("qi_it_join_left", "qi_it_join_right"), childIndices);
+  }
+
+  // ---------------------------------------------------------------------------------------------
+
+  private static void indexWithDocs(String index) throws Exception {
+    QueryInsightsRecords.indexWithDocs(client(), index);
+  }
+
+  /** Runs a recorded query and waits for its record, so everything sent before it has arrived. */
+  private void awaitBarrier() throws Exception {
+    String index = "qi_it_barrier_" + BARRIER_SEQ.incrementAndGet();
+    indexWithDocs(index);
+    executeQuery("source=" + index + " | fields v");
+    awaitPplRecord(index);
+  }
+
+  private static JSONObject awaitPplRecord(String index) throws Exception {
+    return awaitRecord(pplRecordFor(index), "PPL record for " + index);
+  }
+
+  private static void assertNoPplRecord(String index) throws Exception {
+    List<JSONObject> found = topQueries(pplRecordFor(index));
+    assertTrue("unexpected PPL record for " + index + ": " + found, found.isEmpty());
+  }
+
+  private static JSONObject awaitRecord(Predicate<JSONObject> match, String what) throws Exception {
+    return QueryInsightsRecords.awaitRecord(
+        client(), RequestOptions.DEFAULT, CLASS_START_MILLIS, match, what);
+  }
+
+  private static List<JSONObject> topQueries(Predicate<JSONObject> match) throws Exception {
+    return QueryInsightsRecords.topQueries(
+        client(), RequestOptions.DEFAULT, CLASS_START_MILLIS, match);
+  }
+}