graylog ProcessBufferProcessor 简单说明

ProcessBufferProcessor 的核心是进行消息处理，会调用消息处理器，同时会将消息放到OutputBuffer 中
ProcessBufferProcessor 实际上是ProcessBuffer 中Disruptor 的一个handler

核心处理

handleMessage 方法

private void handleMessage(@Nonnull Message msg) {
        msg.addStream(defaultStreamProvider.get());
        Messages messages = msg;
       // 消息处理器，进行消息加工
        for (MessageProcessor messageProcessor : orderedMessageProcessors) {
            messages = messageProcessor.process(messages);
        }
 
        for (Message message : messages) {
            message.ensureValidTimestamp();
 
            if (!message.hasField(Message.FIELD_GL2_MESSAGE_ID) || isNullOrEmpty(message.getFieldAs(String.class, Message.FIELD_GL2_MESSAGE_ID))) {
                // Set the message ID once all message processors have finished
                // See documentation of Message.FIELD_GL2_MESSAGE_ID for details
                message.addField(Message.FIELD_GL2_MESSAGE_ID, ulid.nextULID());
            }
 
            // The processing time should only be set once all message processors have finished
            message.setProcessingTime(Tools.nowUTC());
           // 消息处理标记 
            processingStatusRecorder.updatePostProcessingReceiveTime(message.getReceiveTime());
 
            if(failureSubmissionService.submitProcessingErrors(message)) {
              // 消息放到 outputBuffer 中
                outputBuffer.insertBlocking(message);
            }
        }
    }

msg.addStream(defaultStreamProvider.get());

MessageProcessor

MessageProcessor 会调用Filter 进行消息的加工

参考实现

包含的filter 实现

参考资料

​​https://github.com/Graylog2/graylog2-server/blob/626be1f0d80506705b5ba41fbea33c2ec0164bc0/graylog2-server/src/main/java/org/graylog2/shared/buffers/processors/ProcessBufferProcessor.java​​​
​​​https://github.com/Graylog2/graylog2-server/blob/626be1f0d80506705b5ba41fbea33c2ec0164bc0/graylog2-server/src/main/java/org/graylog2/plugin/messageprocessors/MessageProcessor.java​​​
​​​https://github.com/Graylog2/graylog2-server/blob/626be1f0d80506705b5ba41fbea33c2ec0164bc0/graylog2-server/src/main/java/org/graylog2/plugin/filters/MessageFilter.java​​