Operator/ingest (#89735)

Add support for /_ingest/pipeline for file based settings.

Relates to #89183

docs/changelog/89735.yaml

@@ -0,0 +1,5 @@
+pr: 89735
+summary: Operator/ingest
+area: Infra/Core
+type: enhancement
+issues: []

server/src/internalClusterTest/java/org/elasticsearch/ingest/IngestFileSettingsIT.java

@@ -0,0 +1,273 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.ingest;
+
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.ingest.PutPipelineAction;
+import org.elasticsearch.action.ingest.PutPipelineRequest;
+import org.elasticsearch.action.ingest.ReservedPipelineAction;
+import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.cluster.metadata.ReservedStateErrorMetadata;
+import org.elasticsearch.cluster.metadata.ReservedStateHandlerMetadata;
+import org.elasticsearch.cluster.metadata.ReservedStateMetadata;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.core.Strings;
+import org.elasticsearch.core.Tuple;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.reservedstate.service.FileSettingsService;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.xcontent.XContentFactory;
+import org.elasticsearch.xcontent.XContentParserConfiguration;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.elasticsearch.xcontent.XContentType.JSON;
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.notNullValue;
+
+public class IngestFileSettingsIT extends ESIntegTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(CustomIngestTestPlugin.class);
+    }
+
+    private static AtomicLong versionCounter = new AtomicLong(1);
+
+    private static String testJSON = """
+        {
+             "metadata": {
+                 "version": "%s",
+                 "compatibility": "8.4.0"
+             },
+             "state": {
+                 "ingest_pipelines": {
+                   "my_ingest_pipeline": {
+                       "description": "_description",
+                       "processors": [
+                          {
+                            "test" : {
+                              "field": "pipeline",
+                              "value": "pipeline"
+                            }
+                          }
+                       ]
+                   },
+                   "my_ingest_pipeline_1": {
+                       "description": "_description",
+                       "processors": [
+                          {
+                            "test" : {
+                              "field": "pipeline",
+                              "value": "pipeline"
+                            }
+                          }
+                       ]
+                   }
+                 }
+             }
+        }""";
+
+    private static String testErrorJSON = """
+        {
+             "metadata": {
+                 "version": "%s",
+                 "compatibility": "8.4.0"
+             },
+             "state": {
+                 "ingest_pipelines": {
+                   "my_ingest_pipeline": {
+                       "description": "_description",
+                       "processors": [
+                          {
+                            "foo" : {
+                              "field": "pipeline",
+                              "value": "pipeline"
+                            }
+                          }
+                       ]
+                   }
+                 }
+             }
+        }""";
+
+    private void writeJSONFile(String node, String json) throws Exception {
+        long version = versionCounter.incrementAndGet();
+
+        FileSettingsService fileSettingsService = internalCluster().getInstance(FileSettingsService.class, node);
+        assertTrue(fileSettingsService.watching());
+
+        Files.deleteIfExists(fileSettingsService.operatorSettingsFile());
+
+        Files.createDirectories(fileSettingsService.operatorSettingsDir());
+        Path tempFilePath = createTempFile();
+
+        logger.info("--> writing JSON config to node {} with path {}", node, tempFilePath);
+        logger.info(Strings.format(json, version));
+        Files.write(tempFilePath, Strings.format(json, version).getBytes(StandardCharsets.UTF_8));
+        Files.move(tempFilePath, fileSettingsService.operatorSettingsFile(), StandardCopyOption.ATOMIC_MOVE);
+    }
+
+    private Tuple<CountDownLatch, AtomicLong> setupClusterStateListener(String node) {
+        ClusterService clusterService = internalCluster().clusterService(node);
+        CountDownLatch savedClusterState = new CountDownLatch(1);
+        AtomicLong metadataVersion = new AtomicLong(-1);
+        clusterService.addListener(new ClusterStateListener() {
+            @Override
+            public void clusterChanged(ClusterChangedEvent event) {
+                ReservedStateMetadata reservedState = event.state().metadata().reservedStateMetadata().get(FileSettingsService.NAMESPACE);
+                if (reservedState != null) {
+                    ReservedStateHandlerMetadata handlerMetadata = reservedState.handlers().get(ReservedPipelineAction.NAME);
+                    if (handlerMetadata != null && handlerMetadata.keys().contains("my_ingest_pipeline")) {
+                        clusterService.removeListener(this);
+                        metadataVersion.set(event.state().metadata().version());
+                        savedClusterState.countDown();
+                    }
+                }
+            }
+        });
+
+        return new Tuple<>(savedClusterState, metadataVersion);
+    }
+
+    private void assertPipelinesSaveOK(CountDownLatch savedClusterState, AtomicLong metadataVersion) throws Exception {
+        boolean awaitSuccessful = savedClusterState.await(20, TimeUnit.SECONDS);
+        assertTrue(awaitSuccessful);
+
+        final ClusterStateResponse clusterStateResponse = client().admin()
+            .cluster()
+            .state(new ClusterStateRequest().waitForMetadataVersion(metadataVersion.get()))
+            .get();
+
+        ReservedStateMetadata reservedState = clusterStateResponse.getState()
+            .metadata()
+            .reservedStateMetadata()
+            .get(FileSettingsService.NAMESPACE);
+
+        ReservedStateHandlerMetadata handlerMetadata = reservedState.handlers().get(ReservedPipelineAction.NAME);
+
+        assertThat(handlerMetadata.keys(), allOf(notNullValue(), containsInAnyOrder("my_ingest_pipeline", "my_ingest_pipeline_1")));
+
+        // Try using the REST API to update the my_autoscaling_policy policy
+        // This should fail, we have reserved certain autoscaling policies in operator mode
+        assertEquals(
+            "Failed to process request [org.elasticsearch.action.ingest.PutPipelineRequest/unset] with errors: "
+                + "[[my_ingest_pipeline] set as read-only by [file_settings]]",
+            expectThrows(
+                IllegalArgumentException.class,
+                () -> client().execute(PutPipelineAction.INSTANCE, sampleRestRequest("my_ingest_pipeline")).actionGet()
+            ).getMessage()
+        );
+    }
+
+    public void testPoliciesApplied() throws Exception {
+        ensureGreen();
+
+        var savedClusterState = setupClusterStateListener(internalCluster().getMasterName());
+        writeJSONFile(internalCluster().getMasterName(), testJSON);
+
+        assertPipelinesSaveOK(savedClusterState.v1(), savedClusterState.v2());
+    }
+
+    private Tuple<CountDownLatch, AtomicLong> setupClusterStateListenerForError(String node) {
+        ClusterService clusterService = internalCluster().clusterService(node);
+        CountDownLatch savedClusterState = new CountDownLatch(1);
+        AtomicLong metadataVersion = new AtomicLong(-1);
+        clusterService.addListener(new ClusterStateListener() {
+            @Override
+            public void clusterChanged(ClusterChangedEvent event) {
+                ReservedStateMetadata reservedState = event.state().metadata().reservedStateMetadata().get(FileSettingsService.NAMESPACE);
+                if (reservedState != null && reservedState.errorMetadata() != null) {
+                    clusterService.removeListener(this);
+                    metadataVersion.set(event.state().metadata().version());
+                    savedClusterState.countDown();
+                    assertEquals(ReservedStateErrorMetadata.ErrorKind.VALIDATION, reservedState.errorMetadata().errorKind());
+                    assertThat(reservedState.errorMetadata().errors(), allOf(notNullValue(), hasSize(1)));
+                    assertThat(
+                        reservedState.errorMetadata().errors().get(0),
+                        containsString("org.elasticsearch.ElasticsearchParseException: No processor type exists with name [foo]")
+                    );
+                }
+            }
+        });
+
+        return new Tuple<>(savedClusterState, metadataVersion);
+    }
+
+    private void assertPipelinesNotSaved(CountDownLatch savedClusterState, AtomicLong metadataVersion) throws Exception {
+        boolean awaitSuccessful = savedClusterState.await(20, TimeUnit.SECONDS);
+        assertTrue(awaitSuccessful);
+
+        // This should succeed, nothing was reserved
+        client().execute(PutPipelineAction.INSTANCE, sampleRestRequest("my_ingest_pipeline_bad")).get();
+    }
+
+    public void testErrorSaved() throws Exception {
+        ensureGreen();
+        var savedClusterState = setupClusterStateListenerForError(internalCluster().getMasterName());
+
+        writeJSONFile(internalCluster().getMasterName(), testErrorJSON);
+        assertPipelinesNotSaved(savedClusterState.v1(), savedClusterState.v2());
+    }
+
+    private PutPipelineRequest sampleRestRequest(String id) throws Exception {
+        var json = """
+            {
+               "description": "_description",
+               "processors": [
+                  {
+                    "test" : {
+                      "field": "_foo",
+                      "value": "_bar"
+                    }
+                  }
+               ]
+            }""";
+
+        try (
+            var bis = new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8));
+            var parser = JSON.xContent().createParser(XContentParserConfiguration.EMPTY, bis);
+            var builder = XContentFactory.contentBuilder(JSON)
+        ) {
+            builder.map(parser.map());
+            return new PutPipelineRequest(id, BytesReference.bytes(builder), JSON);
+        }
+    }
+
+    public static class CustomIngestTestPlugin extends IngestTestPlugin {
+        @Override
+        public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
+            Map<String, Processor.Factory> processors = new HashMap<>();
+            processors.put("test", (factories, tag, description, config) -> {
+                String field = (String) config.remove("field");
+                String value = (String) config.remove("value");
+                return new FakeProcessor("test", tag, description, (ingestDocument) -> ingestDocument.setFieldValue(field, value));
+            });
+
+            return processors;
+        }
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/cluster/node/info/NodesInfoRequest.java

@@ -125,6 +125,20 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeStringArray(requestedMetrics.toArray(String[]::new));
     }
 
+    /**
+     * Helper method for creating NodesInfoRequests with desired metrics
+     * @param metrics the metrics to include in the request
+     * @return
+     */
+    public static NodesInfoRequest requestWithMetrics(Metric... metrics) {
+        NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
+        nodesInfoRequest.clear();
+        for (var metric : metrics) {
+            nodesInfoRequest.addMetric(metric.metricName());
+        }
+        return nodesInfoRequest;
+    }
+
     /**
      * An enumeration of the "core" sections of metrics that may be requested
      * from the nodes information endpoint. Eventually this list list will be

server/src/main/java/org/elasticsearch/action/ingest/DeletePipelineTransportAction.java

@@ -22,6 +22,9 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.util.Optional;
+import java.util.Set;
+
 public class DeletePipelineTransportAction extends AcknowledgedTransportMasterNodeAction<DeletePipelineRequest> {
 
     private final IngestService ingestService;
@@ -62,4 +65,13 @@ protected ClusterBlockException checkBlock(DeletePipelineRequest request, Cluste
         return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
     }
 
+    @Override
+    public Optional<String> reservedStateHandlerName() {
+        return Optional.of(ReservedPipelineAction.NAME);
+    }
+
+    @Override
+    public Set<String> modifiedKeys(DeletePipelineRequest request) {
+        return Set.of(request.getId());
+    }
 }

server/src/main/java/org/elasticsearch/action/ingest/PutPipelineTransportAction.java

@@ -25,10 +25,12 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.util.Optional;
+import java.util.Set;
+
 import static org.elasticsearch.ingest.IngestService.INGEST_ORIGIN;
 
 public class PutPipelineTransportAction extends AcknowledgedTransportMasterNodeAction<PutPipelineRequest> {
-
     private final IngestService ingestService;
     private final OriginSettingClient client;
 
@@ -73,4 +75,13 @@ protected ClusterBlockException checkBlock(PutPipelineRequest request, ClusterSt
         return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
     }
 
+    @Override
+    public Optional<String> reservedStateHandlerName() {
+        return Optional.of(ReservedPipelineAction.NAME);
+    }
+
+    @Override
+    public Set<String> modifiedKeys(PutPipelineRequest request) {
+        return Set.of(request.getId());
+    }
 }