Fix extractors being deleted when an input is started or stopped (7.1)

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));
         }
@@ -749,8 +754,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,6 +467,7 @@ 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