Streams - Log's Enable, Disable and Status endpoints #129474

lukewhiting · 2025-06-16T12:07:55Z

This pull request introduces a new module to Elasticsearch called streams.

This feature will provide "batteries included" ingest of bulk data with minimal user setup.

This initial PR introduces the new new ES module to support the function, including:

Build config
REST endpoints to enable/disable the functionality
REST endpoint to check if function is enabled
Appropriate actions to power those REST endpoints
A new cluster state object for storing feature state
REST and Integration tests.

Impliments ES-11330

elasticsearchmachine · 2025-06-16T12:08:21Z

Hi @lukewhiting, I've created a changelog YAML for you.

elasticsearchmachine · 2025-06-16T12:46:33Z

Pinging @elastic/es-data-management (Team:Data Management)

rest-api-spec/src/main/resources/rest-api-spec/api/streams.logs_disable.json

...treams/src/main/java/org/elasticsearch/rest/streams/logs/RestSetLogStreamsEnabledAction.java

rest-api-spec/src/main/resources/rest-api-spec/api/streams.status.json

nielsbauman

I didn't look much at the logic or tests, I just left some comments on the general wiring of the new APIs.

docs/changelog/129474.yaml

.../streams/src/main/java/org/elasticsearch/rest/streams/logs/TransportStreamsStatusAction.java

server/src/main/java/org/elasticsearch/TransportVersions.java

modules/streams/build.gradle

modules/streams/src/internalClusterTest/java/org/elasticsearch/rest/streams/TestToggleIT.java

...treams/src/main/java/org/elasticsearch/rest/streams/logs/RestSetLogStreamsEnabledAction.java

nielsbauman

Aside from the timeout comment, the API wiring stuff LGTM 👍

nielsbauman · 2025-06-17T14:43:35Z

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/RestStreamsStatusAction.java

+
+    @Override
+    protected BaseRestHandler.RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
+        StreamsStatusAction.Request statusRequest = new StreamsStatusAction.Request(RestUtils.getAckTimeout(request));


This should use the masterNodeTimeout (same goes for the supported params above). Even though the request runs on the local node, we decided to keep the query param name the same (master_timeout) for BwC reasons, so I think it makes sense to use that name here too for consistency.

Switched both to master node timeout 👍🏻

Don't forget to also update the spec :)

Too many moving parts 😅 Thanks for the reminder

…eam-enable-endpoints

Copilot

Pull Request Overview

This PR introduces a new streams module that provides REST endpoints to enable, disable, and report the status of the logs streams feature, backed by a cluster-state metadata custom object and tested via both YAML and Java integration tests.

Add REST API specs, handlers, and transport actions for /_streams/logs/_enable, /_streams/logs/_disable, and /_streams/status.
Persist feature state in a new StreamsMetadata cluster-state custom object using a sequential, acking task executor.
Register the module in the build and multi-project test configurations, and add both YAML and internal-cluster tests.

Reviewed Changes

Copilot reviewed 20 out of 20 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
x-pack/qa/multi-project/core-rest-tests-with-multiple-projects/build.gradle	Include `:modules:streams` in YAML REST multi-project tests
x-pack/plugin/security/qa/operator-privileges-tests/.../Constants.java	Grant operator privileges for streams toggle and status
server/src/main/java/org/elasticsearch/cluster/SequentialTaskAckingTaskExecutor.java	New executor for sequential acked cluster-state update tasks
server/src/main/java/org/elasticsearch/TransportVersions.java	Add `STREAMS_LOGS_SUPPORT` transport version constant
rest-api-spec/src/main/resources/rest-api-spec/api/streams.*.json	Define API specs for streams enable, disable, and status
modules/streams/src/yamlRestTest/.../streams/logs/10_basic.yml	Add YAML tests for toggling and status checks
modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/Transport*Action.java	Implement transport actions and REST handlers
modules/streams/src/main/java/org/elasticsearch/rest/streams/StreamsPlugin.java	Register REST handlers, actions, and named writeables
modules/streams/src/main/java/org/elasticsearch/rest/streams/StreamsMetadata.java	Define custom cluster-state metadata for streams feature
modules/streams/src/main/java/module-info.java	Module descriptor for the `streams` plugin
modules/streams/src/internalClusterTest/.../TestToggleIT.java	Add Java integration test for toggle behavior

Comments suppressed due to low confidence (3)

modules/streams/src/main/java/module-info.java:10

The module name org.elasticsearch.rest.root is misleading; consider renaming it to org.elasticsearch.rest.streams to match the plugin package and functionality.

module org.elasticsearch.rest.root {

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/TransportLogsStreamsToggleActivation.java:85

[nitpick] The task name prefix enable-streams-logs can be confusing when disabling; consider using a neutral prefix like streams-logs-toggle-[%s] so both enable and disable modes read clearly.

            String taskName = String.format(Locale.ROOT, "enable-streams-logs-[%s]", shouldEnable ? "enable" : "disable");

modules/streams/src/main/java/module-info.java:12

The @Inject annotation from org.elasticsearch.injection.guice is used throughout the code but the module descriptor does not require the injection module; add requires org.elasticsearch.injection.guice (or the correct module) to avoid compilation errors.

    requires org.elasticsearch.xcontent;

masseyke · 2025-06-17T20:20:50Z

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/StreamsStatusAction.java

+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeBoolean(logs_enabled);


This ought to never be called, right? You could just add an assert false or something.

Good catch! You can use TransportAction.localOnly() for that, if you want.

Replaced this with a call to TransportAction.localOnly() and removed the redundant constructor.

masseyke · 2025-06-17T20:21:17Z

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/StreamsStatusAction.java

+            logs_enabled = logsEnabled;
+        }
+
+        public Response(StreamInput in) throws IOException {


This is never called, right? And this is always local? So I think you could just delete it.

Removed :-)

masseyke · 2025-06-17T20:48:21Z

server/src/main/java/org/elasticsearch/cluster/SequentialTaskAckingTaskExecutor.java

+ */
+public class SequentialTaskAckingTaskExecutor<Task extends AckedClusterStateUpdateTask> extends SimpleBatchedAckListenerTaskExecutor<Task> {
+    @Override
+    public Tuple<ClusterState, ClusterStateAckListener> executeTask(Task task, ClusterState clusterState) throws Exception {


Just talking myself through what you've probably already considered -- there's no need to use a AckedBatchedClusterStateUpdateTask and to batch (deduplicate) these, b/c this is just not an action that is commonly going to ever be called more than once in the lifetime of a cluster.

@masseyke This executor class extends SimpleBatchedAckListenerTaskExecutor, so these cluster state updates will be batched. Maybe the name (SequentialTaskAckingTaskExecutor) should include the word "batch" somewhere to indicate that?

Hmm I think I'm just confused between what AckedBatchedClusterStateUpdateTask and AckedClusterStateUpdateTask are for. But regardless, it looks fine.

Renamed this class to SequentialAckingBatchedTaskExecutor to avoid future confusion

masseyke · 2025-06-17T20:55:16Z

.../src/main/java/org/elasticsearch/rest/streams/logs/TransportLogsStreamsToggleActivation.java

+        ActionListener<AcknowledgedResponse> listener
+    ) throws Exception {
+        ProjectId projectId = projectResolver.getProjectId();
+        StreamsMetadata streamsState = state.projectState(projectId).metadata().custom(StreamsMetadata.TYPE, StreamsMetadata.EMPTY);


Since StreamsMetadata is in a module, and we need to be able to check this from server (within the bulk transport action), do we need to move StreamsMetadata to server? Or have some way to expose it in there?

Yep that's probably a good idea. Have moved the metadata object into server.

…eam-enable-endpoints

masseyke · 2025-06-18T13:49:18Z

x-pack/qa/multi-project/core-rest-tests-with-multiple-projects/build.gradle

  restTestConfig project(path: ':modules:data-streams', configuration: "basicRestSpecs")
  restTestConfig project(path: ':modules:ingest-common', configuration: "basicRestSpecs")
  restTestConfig project(path: ':modules:reindex', configuration: "basicRestSpecs")
+  restTestConfig project(path: ':modules:streams', configuration: "basicRestSpecs")


I think you've also got to have

clusterModules project(':modules:streams')

masseyke

LGTM once the build works.

…eam-enable-endpoints

* Enable And Disable Endpoint * Status Endpoint * Integration Tests * REST Spec * REST Spec tests * Some documentation * Update docs/changelog/129474.yaml * Fix failing security test * PR Fixes * PR Fixes - Add missing feature flag name to YAML spec * PR Fixes - Fix support for timeout and master_timeout parameters * PR Fixes - Make the REST handler validation happy with the new params * Delete docs/changelog/129474.yaml * PR Fixes - Switch to local metadata action type and improve request handling * PR Fixes - Make enable / disable endpoint cancellable * PR Fixes - Switch timeout param name for status endpoint * PR Fixes - Switch timeout param name for status endpoint in spec * PR Fixes - Enforce local only use for status action * PR Fixes - Refactor StreamsMetadata into server * PR Fixes - Add streams module to multi project YAML test suite * PR Fixes - Add streams cluster module to multi project YAML test suite

) (#129838) * Streams - Log's Enable, Disable and Status endpoints (#129474) * Enable And Disable Endpoint * Status Endpoint * Integration Tests * REST Spec * REST Spec tests * Some documentation * Update docs/changelog/129474.yaml * Fix failing security test * PR Fixes * PR Fixes - Add missing feature flag name to YAML spec * PR Fixes - Fix support for timeout and master_timeout parameters * PR Fixes - Make the REST handler validation happy with the new params * Delete docs/changelog/129474.yaml * PR Fixes - Switch to local metadata action type and improve request handling * PR Fixes - Make enable / disable endpoint cancellable * PR Fixes - Switch timeout param name for status endpoint * PR Fixes - Switch timeout param name for status endpoint in spec * PR Fixes - Enforce local only use for status action * PR Fixes - Refactor StreamsMetadata into server * PR Fixes - Add streams module to multi project YAML test suite * PR Fixes - Add streams cluster module to multi project YAML test suite * Null out reader in status transport action super constructor * Added comment about minimum version support

* Enable And Disable Endpoint * Status Endpoint * Integration Tests * REST Spec * REST Spec tests * Some documentation * Update docs/changelog/129474.yaml * Fix failing security test * PR Fixes * PR Fixes - Add missing feature flag name to YAML spec * PR Fixes - Fix support for timeout and master_timeout parameters * PR Fixes - Make the REST handler validation happy with the new params * Delete docs/changelog/129474.yaml * PR Fixes - Switch to local metadata action type and improve request handling * PR Fixes - Make enable / disable endpoint cancellable * PR Fixes - Switch timeout param name for status endpoint * PR Fixes - Switch timeout param name for status endpoint in spec * PR Fixes - Enforce local only use for status action * PR Fixes - Refactor StreamsMetadata into server * PR Fixes - Add streams module to multi project YAML test suite * PR Fixes - Add streams cluster module to multi project YAML test suite

lukewhiting added 6 commits June 16, 2025 12:56

Enable And Disable Endpoint

63f07aa

Status Endpoint

2fa713f

Integration Tests

5dc679c

REST Spec

54d5730

REST Spec tests

7b7d52a

Some documentation

6ed2fdd

lukewhiting requested a review from Copilot June 16, 2025 12:07

lukewhiting added >feature :Data Management/Data streams Data streams and their lifecycles v8.19.0 v9.1.0 labels Jun 16, 2025

Update docs/changelog/129474.yaml

130f5b8

This comment was marked as outdated.

Sign in to view

lukewhiting marked this pull request as ready for review June 16, 2025 12:46

elasticsearchmachine added the Team:Data Management Meta label for data/management team label Jun 16, 2025

Fix failing security test

2b7ad57

masseyke self-requested a review June 16, 2025 20:49

masseyke reviewed Jun 16, 2025

View reviewed changes

rest-api-spec/src/main/resources/rest-api-spec/api/streams.logs_disable.json Outdated Show resolved Hide resolved

masseyke reviewed Jun 16, 2025

View reviewed changes

...treams/src/main/java/org/elasticsearch/rest/streams/logs/RestSetLogStreamsEnabledAction.java Outdated Show resolved Hide resolved

masseyke reviewed Jun 16, 2025

View reviewed changes

rest-api-spec/src/main/resources/rest-api-spec/api/streams.status.json Outdated Show resolved Hide resolved

lukewhiting added 4 commits June 17, 2025 10:08

PR Fixes

e601c4d

PR Fixes - Add missing feature flag name to YAML spec

323caca

PR Fixes - Fix support for timeout and master_timeout parameters

c65a5fe

PR Fixes - Make the REST handler validation happy with the new params

8fa2ccd

nielsbauman reviewed Jun 17, 2025

View reviewed changes

lukewhiting added >non-issue and removed >feature labels Jun 17, 2025

Delete docs/changelog/129474.yaml

3f244d3

nielsbauman reviewed Jun 17, 2025

View reviewed changes

lukewhiting added 3 commits June 17, 2025 16:06

PR Fixes - Switch timeout param name for status endpoint

9991a0d

Merge branch 'main' of github.com:elastic/elasticsearch into logs-str…

5a95155

…eam-enable-endpoints

PR Fixes - Switch timeout param name for status endpoint in spec

faf7dab

lukewhiting requested a review from Copilot June 17, 2025 15:25

Copilot AI reviewed Jun 17, 2025

View reviewed changes

masseyke reviewed Jun 17, 2025

View reviewed changes

lukewhiting added 4 commits June 18, 2025 09:31

PR Fixes - Enforce local only use for status action

2204296

PR Fixes - Refactor StreamsMetadata into server

9d3d547

PR Fixes - Add streams module to multi project YAML test suite

9501023

Merge branch 'main' of github.com:elastic/elasticsearch into logs-str…

6815393

…eam-enable-endpoints

masseyke reviewed Jun 18, 2025

View reviewed changes

masseyke approved these changes Jun 18, 2025

View reviewed changes

lukewhiting added 2 commits June 18, 2025 15:02

PR Fixes - Add streams cluster module to multi project YAML test suite

22db6f9

Merge branch 'main' of github.com:elastic/elasticsearch into logs-str…

1e48edc

…eam-enable-endpoints

lukewhiting merged commit 1ccf1c6 into elastic:main Jun 19, 2025
27 checks passed

lukewhiting mentioned this pull request Jun 20, 2025

Reserve transport version for streams endpoint backport #129753

Merged

masseyke mentioned this pull request Jun 20, 2025

Streams - Log's Enable, Disable and Status endpoints (#129474) #129798

Closed

lukewhiting mentioned this pull request Jun 23, 2025

[Backport] Streams - Log's Enable, Disable and Status endpoints (#129474) #129838

Merged

lukewhiting deleted the logs-stream-enable-endpoints branch June 24, 2025 09:26

pquentin mentioned this pull request Sep 11, 2025

Add spec and docs for new logs streams endpoints elastic/elasticsearch-specification#5258

Merged

masseyke mentioned this pull request Sep 12, 2025

Removing the logs stream feature flag #134649

Merged

Streams - Log's Enable, Disable and Status endpoints #129474

Streams - Log's Enable, Disable and Status endpoints #129474

Uh oh!

Conversation

lukewhiting commented Jun 16, 2025

Uh oh!

elasticsearchmachine commented Jun 16, 2025

Uh oh!

This comment was marked as outdated.

Uh oh!

elasticsearchmachine commented Jun 16, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

nielsbauman left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

nielsbauman left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

masseyke left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants