Fix extractor date converters (#11750) (4.2 backport)

graylog2-server/src/main/java/org/graylog2/plugin/Message.java

@@ -443,7 +443,7 @@ public Map<String, Object> toElasticSearchObject(ObjectMapper objectMapper, @Non
             }
             obj.put(FIELD_GL2_PROCESSING_ERROR,
                     processingErrors.stream()
-                            .map(ProcessingError::getDetails)
+                            .map(pe -> pe.getMessage() + " - " + pe.getDetails())
                             .collect(Collectors.joining(", ")));
         }
 
@@ -533,6 +533,16 @@ private void addRequiredField(final String key, final Object value) {
         addField(key, value, true);
     }
 
+    // Set the timestamp field without performing a conversion or fallback to the current time.
+    // This is needed by Extractors so they can use their own DateConverter. (cf #11495)
+    public void setTimeFieldAsString(final String value) {
+        final String str = value.trim();
+        if (!str.isEmpty()) {
+            final Object previousValue = fields.put(FIELD_TIMESTAMP, str);
+            updateSize(FIELD_TIMESTAMP, str, previousValue);
+        }
+    }
+
     private void addField(final String key, final Object value, final boolean isRequiredField) {
         final String trimmedKey = key.trim();
 

graylog2-server/src/main/java/org/graylog2/plugin/inputs/Extractor.java

@@ -27,6 +27,7 @@
 import org.graylog2.plugin.Message;
 import org.graylog2.plugin.database.EmbeddedPersistable;
 import org.graylog2.shared.utilities.ExceptionUtils;
+import org.joda.time.DateTime;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,6 +43,7 @@
 import java.util.stream.Collectors;
 
 import static com.codahale.metrics.MetricRegistry.name;
+import static org.graylog2.plugin.Message.FIELD_TIMESTAMP;
 
 public abstract class Extractor implements EmbeddedPersistable {
     private static final Logger LOG = LoggerFactory.getLogger(Extractor.class);
@@ -235,10 +237,10 @@ public void runExtractor(Message msg) {
                     return;
                 } else if (results.length == 1 && results[0].target == null) {
                     // results[0].target is null if this extractor cannot produce multiple fields use targetField in that case
-                    msg.addField(targetField, results[0].getValue());
+                    addField(msg, targetField, results[0].getValue());
                 } else {
                     for (final Result result : results) {
-                        msg.addField(result.getTarget(), result.getValue());
+                        addField(msg, result.getTarget(), result.getValue());
                     }
                 }
 
@@ -264,10 +266,29 @@ public void runExtractor(Message msg) {
                 }
 
                 runConverters(msg);
+
+                // The extractor / converter might have failed to build a valid timestamp.
+                // In this case we run msg#addField() to log this error and fallback to a current date.
+                if (targetField.equals(FIELD_TIMESTAMP)) {
+                    final Object timestampValue = msg.getField(FIELD_TIMESTAMP);
+                    if (!(timestampValue instanceof DateTime)) {
+                        msg.addField(FIELD_TIMESTAMP, timestampValue);
+                    }
+                }
             }
         }
     }
 
+    // Don't use addField when assigning timestamps.
+    // We have to delay the conversion, because the converters expect this field as a String. (cf #11495)
+    private void addField(Message msg, final String key, final Object value) {
+        if (key.trim().equals(FIELD_TIMESTAMP) && value instanceof String) {
+            msg.setTimeFieldAsString((String) value);
+        } else {
+            msg.addField(key, value);
+        }
+    }
+
     private void runConverters(Message msg) {
         try(final Timer.Context ignored = converterTimer.time()) {
             for (Converter converter : converters) {
@@ -279,8 +300,11 @@ private void runConverters(Message msg) {
                     final Object convertedValue = converter.convert((String) msg.getField(targetField));
                     if (!converter.buildsMultipleFields()) {
                         // We have arrived here if no exception was thrown and can safely replace the original field.
-                        msg.removeField(targetField);
-                        msg.addField(targetField, convertedValue);
+                        if (convertedValue == null) {
+                            msg.removeField(targetField);
+                        } else {
+                            msg.addField(targetField, convertedValue);
+                        }
                     } else if (convertedValue instanceof Map) {
                         @SuppressWarnings("unchecked")
                         final Map<String, Object> additionalFields = new HashMap<>((Map<String, Object>) convertedValue);

graylog2-server/src/test/java/org/graylog2/plugin/MessageTest.java

@@ -751,7 +751,7 @@ public void toElasticSearchObject_processingErrorDetailsAreJoinedInOneStringAndR
 
         // then
         assertThat(esObject.get(Message.FIELD_GL2_PROCESSING_ERROR))
-                .isEqualTo("Failure Details #1, Failure Details #2");
+                .isEqualTo("Failure Message #1 - Failure Details #1, Failure Message #2 - Failure Details #2");
     }
 
     @Test

graylog2-server/src/test/java/org/graylog2/plugin/inputs/ExtractorTest.java

@@ -23,6 +23,7 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import org.graylog.failure.ProcessingFailureCause;
+import org.graylog2.inputs.converters.DateConverter;
 import org.graylog2.inputs.extractors.ExtractorException;
 import org.graylog2.plugin.Message;
 import org.graylog2.plugin.inputs.Extractor.Result;
@@ -958,6 +959,54 @@ public Object apply(Object input) {
         assertThat(extractor.getConverterExceptionCount()).isEqualTo(0L);
     }
 
+    @Test
+    // Test for https://github.com/Graylog2/graylog2-server/issues/11495
+    // The Extractor returns a string that is not directly assignable to the timestamp field.
+    // The converter however, will parse that string and everything is fine.
+    public void testConvertersWithTimestamp() throws Exception {
+        final Converter converter = new DateConverter(ImmutableMap.of(
+                "date_format", "yyyy-MM-dd HH:mm:ss,SSS"
+        ));
+
+        final TestExtractor extractor = new TestExtractor.Builder()
+                .targetField("timestamp")
+                .converters(Collections.singletonList(converter))
+                .callback(() -> new Result[]{new Result("2021-10-20 09:05:39,892", -1, -1)})
+                .build();
+
+        final Message msg = createMessage("the message");
+
+        extractor.runExtractor(msg);
+
+        assertThat(msg.getTimestamp()).isEqualTo(new DateTime(2021, 10, 20, 9, 5, 39, 892, UTC));
+    }
+
+    @Test
+    public void testTimestampIsFixedWhenConverterHasFailed() throws Exception {
+        final Converter converter = new TestConverter.Builder()
+                .multiple(true)
+                .callback(new Function<Object, Object>() {
+                    @Nullable
+                    @Override
+                    public Object apply(Object input) {
+                        throw new IllegalArgumentException("Invalid format: FOO");
+                    }
+                })
+                .build();
+
+        final TestExtractor extractor = new TestExtractor.Builder()
+                .targetField("timestamp")
+                .converters(Collections.singletonList(converter))
+                .callback(() -> new Result[]{new Result("FOO", -1, -1)})
+                .build();
+
+        final Message msg = createMessage("the message");
+
+        extractor.runExtractor(msg);
+
+        assertThat(msg.getTimestamp()).isInstanceOf(DateTime.class);
+    }
+
     private Message createMessage(String message) {
         return new Message(message, "localhost", DateTime.now(UTC));
     }