Prevent ClassCastException with invalid timestamp in clone_message()

plugin/src/main/java/org/graylog/plugins/pipelineprocessor/functions/messages/CloneMessage.java

@@ -23,10 +23,16 @@
 import org.graylog.plugins.pipelineprocessor.ast.functions.FunctionDescriptor;
 import org.graylog.plugins.pipelineprocessor.ast.functions.ParameterDescriptor;
 import org.graylog2.plugin.Message;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import static org.graylog.plugins.pipelineprocessor.ast.functions.ParameterDescriptor.type;
 
 public class CloneMessage extends AbstractFunction<Message> {
+    private static final Logger LOG = LoggerFactory.getLogger(CloneMessage.class);
+
     public static final String NAME = "clone_message";
 
     private final ParameterDescriptor<Message, Message> messageParam;
@@ -38,8 +44,25 @@ public CloneMessage() {
     @Override
     public Message evaluate(FunctionArgs args, EvaluationContext context) {
         final Message currentMessage = messageParam.optional(args, context).orElse(context.currentMessage());
-        final Message clonedMessage = new Message(currentMessage.getMessage(), currentMessage.getSource(), currentMessage.getTimestamp());
-        clonedMessage.addFields(currentMessage.getFields());
+
+        final Object tsField = currentMessage.getField(Message.FIELD_TIMESTAMP);
+        final Message clonedMessage;
+        if (tsField instanceof DateTime) {
+            clonedMessage = new Message(currentMessage.getMessage(), currentMessage.getSource(), currentMessage.getTimestamp());
+            clonedMessage.addFields(currentMessage.getFields());
+        } else {
+            LOG.warn("Invalid timestamp <{}> (type: {}) in message <{}>. Using current time instead.",
+                    tsField, tsField.getClass().getCanonicalName(), currentMessage.getId());
+
+            final DateTime now = DateTime.now(DateTimeZone.UTC);
+            clonedMessage = new Message(currentMessage.getMessage(), currentMessage.getSource(), now);
+            clonedMessage.addFields(currentMessage.getFields());
+
+            // Message#addFields() overwrites the "timestamp" field.
+            clonedMessage.addField("timestamp", now);
+            clonedMessage.addField("gl2_original_timestamp", String.valueOf(tsField));
+        }
+
         clonedMessage.addStreams(currentMessage.getStreams());
 
         // register in context so the processor can inject it later on

plugin/src/test/java/org/graylog/plugins/pipelineprocessor/functions/FunctionsSnippetsTest.java

@@ -478,6 +478,27 @@ public void clonedMessage() {
         assertThat(otherMessage.getSource()).isEqualTo("source");
     }
 
+    @Test
+    public void clonedMessageWithInvalidTimestamp() {
+        final Message message = new Message("test", "test", Tools.nowUTC());
+        message.addField("timestamp", "foobar");
+        final Rule rule = parser.parseRule(ruleForTest(), false);
+        final EvaluationContext context = contextForRuleEval(rule, message);
+
+        final Message origMessage = context.currentMessage();
+        final Message clonedMessage = Iterables.get(context.createdMessages(), 0);
+
+        assertThat(origMessage).isNotEqualTo(clonedMessage);
+        assertThat(origMessage.getField("timestamp")).isNotInstanceOf(DateTime.class);
+
+        assertThat(clonedMessage).isNotNull();
+        assertThat(clonedMessage.getMessage()).isEqualTo(origMessage.getMessage());
+        assertThat(clonedMessage.getSource()).isEqualTo(origMessage.getSource());
+        assertThat(clonedMessage.getStreams()).isEqualTo(origMessage.getStreams());
+        assertThat(clonedMessage.getTimestamp()).isNotNull();
+        assertThat(clonedMessage.getField("gl2_original_timestamp")).isEqualTo(origMessage.getField("timestamp"));
+    }
+
     @Test
     public void grok() {
         final Rule rule = parser.parseRule(ruleForTest(), false);

plugin/src/test/resources/org/graylog/plugins/pipelineprocessor/functions/clonedMessageWithInvalidTimestamp.txt

@@ -0,0 +1,5 @@
+rule "operate on cloned message"
+when true
+then
+  let cloned = clone_message();
+end