Use raw messages on simulator (#42)

* Use arrow function to keep 'this' binding
* Adapt simulator to use raw message loader
  At this point the pipeline simulator only makes sense for messages that were
  not already processed by Graylog, as it can be really tricky to see
  changes done in already processed messages. Therefore, we replace the message
  loader, using the raw message loader.

Author: edmundoa

src/main/java/org/graylog/plugins/pipelineprocessor/rest/SimulationRequest.java

@@ -21,30 +21,27 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.auto.value.AutoValue;
 
+import java.util.Map;
+
 @AutoValue
 @JsonAutoDetect
 public abstract class SimulationRequest {
     @JsonProperty
     public abstract String streamId();
 
     @JsonProperty
-    public abstract String index();
-
-    @JsonProperty
-    public abstract String messageId();
+    public abstract Map<String, Object> message();
 
     public static Builder builder() {
         return new AutoValue_SimulationRequest.Builder();
     }
 
     @JsonCreator
     public static SimulationRequest create (@JsonProperty("stream_id") String streamId,
-                                            @JsonProperty("index") String index,
-                                            @JsonProperty("message_id")  String messageId) {
+                                            @JsonProperty("message") Map<String, Object> message) {
         return builder()
                 .streamId(streamId)
-                .index(index)
-                .messageId(messageId)
+                .message(message)
                 .build();
     }
 
@@ -54,8 +51,6 @@ public abstract static class Builder {
 
         public abstract Builder streamId(String streamId);
 
-        public abstract Builder index(String index);
-
-        public abstract Builder messageId(String messageId);
+        public abstract Builder message(Map<String, Object> message);
     }
 }

src/main/java/org/graylog/plugins/pipelineprocessor/rest/SimulatorResource.java

@@ -24,15 +24,13 @@
 import org.graylog.plugins.pipelineprocessor.processors.PipelineInterpreter;
 import org.graylog.plugins.pipelineprocessor.simulator.PipelineInterpreterTracer;
 import org.graylog2.database.NotFoundException;
-import org.graylog2.indexer.messages.DocumentNotFoundException;
-import org.graylog2.indexer.messages.Messages;
-import org.graylog2.indexer.results.ResultMessage;
 import org.graylog2.messageprocessors.OrderedMessageProcessors;
 import org.graylog2.plugin.Message;
 import org.graylog2.plugin.messageprocessors.MessageProcessor;
 import org.graylog2.plugin.rest.PluginRestResource;
 import org.graylog2.plugin.streams.Stream;
 import org.graylog2.rest.models.messages.responses.ResultMessageSummary;
+import org.graylog2.rest.resources.messages.MessageResource;
 import org.graylog2.shared.rest.resources.RestResource;
 import org.graylog2.shared.security.RestPermissions;
 import org.graylog2.streams.StreamService;
@@ -54,45 +52,40 @@
 @RequiresAuthentication
 public class SimulatorResource extends RestResource implements PluginRestResource {
     private final OrderedMessageProcessors orderedMessageProcessors;
-    private final Messages messages;
+    private final MessageResource messageResource;
     private final StreamService streamService;
 
     @Inject
-    public SimulatorResource(OrderedMessageProcessors orderedMessageProcessors, Messages messages, StreamService streamService) {
+    public SimulatorResource(OrderedMessageProcessors orderedMessageProcessors, MessageResource messageResource, StreamService streamService) {
         this.orderedMessageProcessors = orderedMessageProcessors;
-        this.messages = messages;
+        this.messageResource = messageResource;
         this.streamService = streamService;
     }
 
     @ApiOperation(value = "Simulate the execution of the pipeline message processor")
     @POST
     @RequiresPermissions(PipelineRestPermissions.PIPELINE_RULE_READ)
     public SimulationResponse simulate(@ApiParam(name = "simulation", required = true) @NotNull SimulationRequest request) throws NotFoundException {
-        checkPermission(RestPermissions.MESSAGES_READ, request.messageId());
         checkPermission(RestPermissions.STREAMS_READ, request.streamId());
-        try {
-            final ResultMessage resultMessage = messages.get(request.messageId(), request.index());
-            final Message message = resultMessage.getMessage();
-            if (!request.streamId().equals("default")) {
-                final Stream stream = streamService.load(request.streamId());
-                message.addStream(stream);
-            }
 
-            final List<ResultMessageSummary> simulationResults = new ArrayList<>();
-            final PipelineInterpreterTracer pipelineInterpreterTracer = new PipelineInterpreterTracer();
+        final Message message = new Message(request.message());
+        if (!request.streamId().equals("default")) {
+            final Stream stream = streamService.load(request.streamId());
+            message.addStream(stream);
+        }
+
+        final List<ResultMessageSummary> simulationResults = new ArrayList<>();
+        final PipelineInterpreterTracer pipelineInterpreterTracer = new PipelineInterpreterTracer();
 
-            for (MessageProcessor messageProcessor : orderedMessageProcessors) {
-                if (messageProcessor instanceof PipelineInterpreter) {
-                    org.graylog2.plugin.Messages processedMessages = ((PipelineInterpreter)messageProcessor).process(message, pipelineInterpreterTracer.getSimulatorInterpreterListener());
-                    for (Message processedMessage : processedMessages) {
-                        simulationResults.add(ResultMessageSummary.create(null, processedMessage.getFields(), ""));
-                    }
+        for (MessageProcessor messageProcessor : orderedMessageProcessors) {
+            if (messageProcessor instanceof PipelineInterpreter) {
+                org.graylog2.plugin.Messages processedMessages = ((PipelineInterpreter) messageProcessor).process(message, pipelineInterpreterTracer.getSimulatorInterpreterListener());
+                for (Message processedMessage : processedMessages) {
+                    simulationResults.add(ResultMessageSummary.create(null, processedMessage.getFields(), ""));
                 }
             }
-
-            return SimulationResponse.create(simulationResults, pipelineInterpreterTracer.getExecutionTrace(), pipelineInterpreterTracer.took());
-        } catch (DocumentNotFoundException e) {
-            throw new NotFoundException(e);
         }
+
+        return SimulationResponse.create(simulationResults, pipelineInterpreterTracer.getExecutionTrace(), pipelineInterpreterTracer.took());
     }
 }

src/web/simulator/ProcessorSimulator.jsx

@@ -1,10 +1,11 @@
 import React from 'react';
 import { Col, Row } from 'react-bootstrap';
 
-import LoaderTabs from 'components/messageloaders/LoaderTabs';
+import RawMessageLoader from 'components/messageloaders/RawMessageLoader';
 import SimulationResults from './SimulationResults';
 
 import SimulatorActions from './SimulatorActions';
+// eslint-disable-next-line no-unused-vars
 import SimulatorStore from './SimulatorStore';
 
 const ProcessorSimulator = React.createClass({
@@ -24,14 +25,16 @@ const ProcessorSimulator = React.createClass({
   _onMessageLoad(message) {
     this.setState({ message: message, simulation: undefined, loading: true, error: undefined });
 
-    SimulatorActions.simulate.triggerPromise(this.props.stream, message.index, message.id).then(
-      response => {
-        this.setState({ simulation: response, loading: false });
-      },
-      error => {
-        this.setState({ loading: false, error: error });
-      }
-    );
+    SimulatorActions.simulate
+      .triggerPromise(this.props.stream, message.fields)
+      .then(
+        response => {
+          this.setState({ simulation: response, loading: false });
+        },
+        error => {
+          this.setState({ loading: false, error: error });
+        }
+      );
   },
 
   render() {
@@ -40,9 +43,11 @@ const ProcessorSimulator = React.createClass({
         <Row>
           <Col md={12}>
             <h1>Load a message</h1>
-            <p>Load a message to be used in the simulation. <strong>No changes will be done in your stored
-              messages.</strong></p>
-            <LoaderTabs onMessageLoaded={this._onMessageLoad} disableMessagePreview />
+            <p>
+              Load a message to be used in the simulation.{' '}
+              <strong>No changes will be done in your stored messages.</strong>
+            </p>
+            <RawMessageLoader onMessageLoaded={this._onMessageLoad} />
           </Col>
         </Row>
         <SimulationResults stream={this.props.stream}

src/web/simulator/SimulatorStore.js

@@ -12,18 +12,17 @@ const urlPrefix = '/plugins/org.graylog.plugins.pipelineprocessor';
 const SimulatorStore = Reflux.createStore({
   listenables: [SimulatorActions],
 
-  simulate(stream, index, messageId) {
+  simulate(stream, messageFields) {
     const url = URLUtils.qualifyUrl(`${urlPrefix}/system/pipelines/simulate`);
     const simulation = {
       stream_id: stream.id,
-      index: index,
-      message_id: messageId,
+      message: messageFields,
     };
 
     let promise = fetch('POST', url, simulation);
     promise = promise.then(response => {
       const formattedResponse = ObjectUtils.clone(response);
-      formattedResponse.messages = response.messages.map(MessageFormatter.formatMessageSummary);
+      formattedResponse.messages = response.messages.map(msg => MessageFormatter.formatMessageSummary(msg));
 
       return formattedResponse;
     });