Fix extractors being deleted when an input is started or stopped

changelog/unreleased/issue-26009.toml

@@ -0,0 +1,5 @@
+type = "fixed"
+message = "Fix extractors being deleted when an input is started, stopped, or updated."
+
+issues = ["26009"]
+pulls = ["26198"]

graylog2-server/src/main/java/org/graylog2/inputs/InputImpl.java

@@ -120,6 +120,15 @@ public Map<String, String> getStaticFields() {
         return result;
     }
 
+    /**
+     * The embedded extractor documents. They are modified through targeted update operations (see
+     * {@code InputServiceImpl#addExtractor} etc.) and only modeled here so that they survive full document
+     * replacements when saving an input.
+     */
+    @Nullable
+    @JsonProperty(EMBEDDED_EXTRACTORS)
+    public abstract List<Map<String, Object>> getEmbeddedExtractors();
+
     @NotNull
     @JsonProperty(FIELD_TYPE)
     public abstract String getType();
@@ -179,6 +188,9 @@ public static Builder create() {
         @JsonProperty(EMBEDDED_STATIC_FIELDS)
         public abstract Builder setEmbeddedStaticFields(List<Map<String, String>> staticFields);
 
+        @JsonProperty(EMBEDDED_EXTRACTORS)
+        public abstract Builder setEmbeddedExtractors(List<Map<String, Object>> extractors);
+
         @JsonProperty(FIELD_TYPE)
         public abstract Builder setType(String type);
 
@@ -226,6 +238,11 @@ public Map<String, Object> getFields() {
             doc.put(EMBEDDED_STATIC_FIELDS, getEmbeddedStaticFields());
         }
 
+        final List<Map<String, Object>> extractors = getEmbeddedExtractors();
+        if (extractors != null && !extractors.isEmpty()) {
+            doc.put(EMBEDDED_EXTRACTORS, extractors);
+        }
+
         if (getContentPack() != null) {
             doc.put(FIELD_CONTENT_PACK, getContentPack());
         }

graylog2-server/src/main/java/org/graylog2/inputs/InputServiceImpl.java

@@ -296,6 +296,11 @@ private InputImpl.Builder buildFromMap(Map<String, Object> fields) {
             builder.setEmbeddedStaticFields(staticFields);
         }
 
+        final List<Map<String, Object>> extractors = (List<Map<String, Object>>) fields.get(InputImpl.EMBEDDED_EXTRACTORS);
+        if (extractors != null && !extractors.isEmpty()) {
+            builder.setEmbeddedExtractors(extractors);
+        }
+
         if (!isGlobal) {
             builder.setNodeId((String) fields.get(MessageInput.FIELD_NODE_ID));
         }
@@ -762,8 +767,12 @@ private InputImpl withEncryptedFields(InputImpl input) {
 
     @Override
     public void persistDesiredState(Input input, IOState.Type desiredState) throws ValidationException {
-        final Input updatedInput = input.withDesiredState(desiredState);
-        saveWithoutEvents(updatedInput);
+        // Use a targeted update instead of saving the whole input to avoid overwriting concurrent changes
+        // to other parts of the input document.
+        collection.updateOne(
+                MongoUtils.idEq(input.getId()),
+                Updates.set(InputImpl.FIELD_DESIRED_STATE, desiredState.name())
+        );
     }
 
     @Override

graylog2-server/src/test/java/org/graylog2/inputs/InputServiceImplTest.java

@@ -448,6 +448,7 @@ public void persistedDocumentContainsOnlyExpectedFields(MongoCollections mongoCo
                 .setEmbeddedStaticFields(List.of(
                         Map.of(InputImpl.FIELD_STATIC_FIELD_KEY, "static_key",
                                 InputImpl.FIELD_STATIC_FIELD_VALUE, "static_value")))
+                .setEmbeddedExtractors(List.of(createCopyInputExtractor().getPersistedFields()))
                 .build();
 
         final String id = inputService.save(input);
@@ -466,12 +467,64 @@ public void persistedDocumentContainsOnlyExpectedFields(MongoCollections mongoCo
                 InputImpl.FIELD_GLOBAL,
                 InputImpl.FIELD_CONFIGURATION,
                 InputImpl.EMBEDDED_STATIC_FIELDS,
+                InputImpl.EMBEDDED_EXTRACTORS,
                 InputImpl.FIELD_DESIRED_STATE,
                 InputImpl.FIELD_CONTENT_PACK,
                 InputImpl.FIELD_NODE_ID
         );
     }
 
+    /**
+     * Regression test for <a href="https://github.com/Graylog2/graylog2-server/issues/26009">#26009</a>:
+     * persisting the desired state when starting or stopping an input must not delete its extractors.
+     */
+    @Test
+    void persistDesiredStateKeepsExtractors() throws Exception {
+        final Input input = inputService.find(inputService.save(createTestInput()));
+        inputService.addExtractor(input, createCopyInputExtractor());
+
+        inputService.persistDesiredState(input, IOState.Type.STOPPED);
+
+        assertThat(inputService.extractorCountByInputId(List.of(input.getId())))
+                .containsEntry(input.getId(), 1);
+        assertThat(inputService.find(input.getId()).getDesiredState()).isEqualTo(IOState.Type.STOPPED);
+    }
+
+    /**
+     * Regression test for <a href="https://github.com/Graylog2/graylog2-server/issues/26009">#26009</a>:
+     * saving an input that was loaded from the database must round-trip the embedded extractors.
+     */
+    @Test
+    void saveKeepsExtractors() throws Exception {
+        final Input input = inputService.find(inputService.save(createTestInput()));
+        inputService.addExtractor(input, createCopyInputExtractor());
+
+        inputService.save(inputService.find(input.getId()));
+
+        assertThat(inputService.extractorCountByInputId(List.of(input.getId())))
+                .containsEntry(input.getId(), 1);
+    }
+
+    /**
+     * Regression test for <a href="https://github.com/Graylog2/graylog2-server/issues/26009">#26009</a>:
+     * the input update flow in {@code InputsResource} merges via the {@code getFields()} Map view and
+     * re-creates the input from the merged map. Extractors must survive this round-trip, too.
+     */
+    @Test
+    void updateViaFieldsMapKeepsExtractors() throws Exception {
+        final Input input = inputService.find(inputService.save(createTestInput()));
+        inputService.addExtractor(input, createCopyInputExtractor());
+
+        final Input reloaded = inputService.find(input.getId());
+        final Map<String, Object> mergedFields = new HashMap<>(reloaded.getFields());
+        mergedFields.put(MessageInput.FIELD_TITLE, "updated title");
+        inputService.update(inputService.create(reloaded.getId(), mergedFields));
+
+        assertThat(inputService.extractorCountByInputId(List.of(input.getId())))
+                .containsEntry(input.getId(), 1);
+        assertThat(inputService.find(input.getId()).getTitle()).isEqualTo("updated title");
+    }
+
     @Test
     void totalExtractorCountByTypeCountsExtractorsAcrossInputs() throws Exception {
         final Input input1 = inputService.find(inputService.save(createTestInput()));