Improved NetFlowV9 support (#21)

* move template caching and flow buffering in custom message aggregator

much of this is still wip

* remove template cache

* wip for codec aggregation and parsing custom format of v9

currently fixing tests

* wip

* migrate tests

* fix v9 parsing by preserving the complete packets during buffering

aggregating the data flowsets does not work, because all records are based on the packet's timestamp
to simplify parsing in the codec, the aggregator now collects all templates/option templates into a protobuf
and adds received and buffered data flows to be parsed.
the netflow packets are preserved completely and can also contain templates. the codec will not use them, though, only the aggregator does

* remove duplicated license header

* update protobuf comment

* update comment

* tweak license header to avoid diff

* fix guice setup for transport

* fix handler setup

the codec-aggregator is null in super class, so that the put call added it after the raw-message handler
thus the code never ran and screwed up parsing

* change how the packet cache works

the previous implementation only checked for a single template id to be present for each packet, which in general is wrong if not all templates arrive at the same time (which might happen for large numbers of active templates)
the new implementation manually checks each packet's template requirements agains the ids of received templates, for the current remoteaddress/source id combination

* prefix netflow v9 fields with nf_

* remove unused optional template atomic ref

* don't prefix unknown fields names, that is done centrally now

fixes double prefixing of fields

* don't forget to fix test

* add flow timestamp fields from ipfix

* fix test after field definition list update

Author: kroepke

pom.xml

@@ -72,6 +72,13 @@
             <artifactId>jackson-dataformat-yaml</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>${protobuf.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

src/main/java/org/graylog/plugins/netflow/NetFlowPluginModule.java

@@ -21,6 +21,7 @@
 
 import org.graylog.plugins.netflow.codecs.NetFlowCodec;
 import org.graylog.plugins.netflow.inputs.NetFlowUdpInput;
+import org.graylog.plugins.netflow.transport.NetFlowUdpTransport;
 import org.graylog2.plugin.PluginConfigBean;
 import org.graylog2.plugin.PluginModule;
 
@@ -37,5 +38,6 @@ public Set<? extends PluginConfigBean> getConfigBeans() {
     protected void configure() {
         addMessageInput(NetFlowUdpInput.class);
         addCodec("netflow", NetFlowCodec.class);
+        addTransport("netflow-udp", NetFlowUdpTransport.class);
     }
 }

src/main/java/org/graylog/plugins/netflow/codecs/NetFlowCodec.java

@@ -16,74 +16,74 @@
  */
 package org.graylog.plugins.netflow.codecs;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
 import com.google.inject.assistedinject.Assisted;
-import org.apache.commons.lang3.SystemUtils;
+import com.google.protobuf.InvalidProtocolBufferException;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
 import org.graylog.plugins.netflow.flows.FlowException;
-import org.graylog.plugins.netflow.flows.NetFlowParser;
+import org.graylog.plugins.netflow.flows.NetFlowFormatter;
+import org.graylog.plugins.netflow.v5.NetFlowV5Packet;
+import org.graylog.plugins.netflow.v5.NetFlowV5Parser;
 import org.graylog.plugins.netflow.v9.NetFlowV9FieldTypeRegistry;
-import org.graylog.plugins.netflow.v9.NetFlowV9TemplateCache;
+import org.graylog.plugins.netflow.v9.NetFlowV9Journal;
+import org.graylog.plugins.netflow.v9.NetFlowV9OptionTemplate;
+import org.graylog.plugins.netflow.v9.NetFlowV9Packet;
+import org.graylog.plugins.netflow.v9.NetFlowV9Parser;
+import org.graylog.plugins.netflow.v9.NetFlowV9Record;
+import org.graylog.plugins.netflow.v9.NetFlowV9Template;
 import org.graylog2.plugin.Message;
+import org.graylog2.plugin.ResolvableInetSocketAddress;
 import org.graylog2.plugin.configuration.Configuration;
 import org.graylog2.plugin.configuration.ConfigurationRequest;
 import org.graylog2.plugin.configuration.fields.ConfigurationField;
-import org.graylog2.plugin.configuration.fields.NumberField;
 import org.graylog2.plugin.configuration.fields.TextField;
 import org.graylog2.plugin.inputs.annotations.Codec;
 import org.graylog2.plugin.inputs.annotations.ConfigClass;
 import org.graylog2.plugin.inputs.annotations.FactoryClass;
 import org.graylog2.plugin.inputs.codecs.AbstractCodec;
+import org.graylog2.plugin.inputs.codecs.CodecAggregator;
 import org.graylog2.plugin.inputs.codecs.MultiMessageCodec;
 import org.graylog2.plugin.inputs.transports.NettyTransport;
 import org.graylog2.plugin.journal.RawMessage;
+import org.graylog2.shared.utilities.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import javax.inject.Inject;
-import javax.inject.Named;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.nio.file.Path;
-import java.nio.file.Paths;
+import java.net.InetSocketAddress;
 import java.util.Collection;
-import java.util.concurrent.ScheduledExecutorService;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 @Codec(name = "netflow", displayName = "NetFlow")
 public class NetFlowCodec extends AbstractCodec implements MultiMessageCodec {
-    private static final Logger LOG = LoggerFactory.getLogger(NetFlowCodec.class);
-
-    @VisibleForTesting
-    static final String CK_CACHE_SIZE = "cache_size";
-    @VisibleForTesting
-    static final String CK_CACHE_PATH = "cache_path";
-    @VisibleForTesting
-    static final String CK_CACHE_SAVE_INTERVAL = "cache_save_interval";
+    /**
+     * Marker byte which signals that the contained netflow packet should be parsed as is.
+     */
+    public static final byte PASSTHROUGH_MARKER = 0x00;
+    /**
+     * Marker byte which signals that the contained netflow v9 packet is non-RFC:
+     * It contains all necessary template flows before any data flows and can be completely parsed without a template cache.
+     */
+    public static final byte ORDERED_V9_MARKER = 0x01;
     @VisibleForTesting
     static final String CK_NETFLOW9_DEFINITION_PATH = "netflow9_definitions_Path";
-
-    private static final int DEFAULT_CACHE_SIZE = 1000;
-    private static final String DEFAULT_CACHE_PATH = SystemUtils.getJavaIoTmpDir().toPath().resolve("netflow-templates.json").toString();
-    private static final int DEFAULT_CACHE_SAVE_INTERVAL = 15 * 60;
-
-    private final NetFlowV9TemplateCache templateCache;
+    private static final Logger LOG = LoggerFactory.getLogger(NetFlowCodec.class);
     private final NetFlowV9FieldTypeRegistry typeRegistry;
 
     @Inject
-    protected NetFlowCodec(@Assisted Configuration configuration,
-                           @Named("daemonScheduler") ScheduledExecutorService scheduler,
-                           ObjectMapper objectMapper) throws IOException {
+    protected NetFlowCodec(@Assisted Configuration configuration) throws IOException {
         super(configuration);
 
-        final int cacheSize = configuration.getInt(CK_CACHE_SIZE, DEFAULT_CACHE_SIZE);
-        final int cacheSaveInterval = configuration.getInt(CK_CACHE_SAVE_INTERVAL, DEFAULT_CACHE_SAVE_INTERVAL);
-        final String configCachePath = configuration.getString(CK_CACHE_PATH, DEFAULT_CACHE_PATH);
-        final Path cachePath = Paths.get(configCachePath);
-        this.templateCache = new NetFlowV9TemplateCache(cacheSize, cachePath, cacheSaveInterval, scheduler, objectMapper);
-
         final String netFlow9DefinitionsPath = configuration.getString(CK_NETFLOW9_DEFINITION_PATH);
         if (netFlow9DefinitionsPath == null || netFlow9DefinitionsPath.trim().isEmpty()) {
             this.typeRegistry = NetFlowV9FieldTypeRegistry.create();
@@ -94,6 +94,13 @@ protected NetFlowCodec(@Assisted Configuration configuration,
         }
     }
 
+    @Nullable
+    @Override
+    public CodecAggregator getAggregator() {
+        // this is intentional: we replace the entire channel handler in NetFlowUdpTransport because we need a different signature
+        return null;
+    }
+
     @Nullable
     @Override
     public Message decode(@Nonnull RawMessage rawMessage) {
@@ -104,13 +111,82 @@ public Message decode(@Nonnull RawMessage rawMessage) {
     @Override
     public Collection<Message> decodeMessages(@Nonnull RawMessage rawMessage) {
         try {
-            return NetFlowParser.parse(rawMessage, templateCache, typeRegistry);
+            final ResolvableInetSocketAddress remoteAddress = rawMessage.getRemoteAddress();
+            final InetSocketAddress sender = remoteAddress != null ? remoteAddress.getInetSocketAddress() : null;
+
+            final byte[] payload = rawMessage.getPayload();
+            if (payload.length < 3) {
+                LOG.debug("NetFlow message (source: {}) doesn't even fit the NetFlow version (size: {} bytes)",
+                        sender, payload.length);
+                return null;
+            }
+
+            final ByteBuf buffer = Unpooled.wrappedBuffer(payload);
+            switch (buffer.readByte()) {
+                case PASSTHROUGH_MARKER:
+                    final NetFlowV5Packet netFlowV5Packet = NetFlowV5Parser.parsePacket(buffer);
+
+                    return netFlowV5Packet.records().stream()
+                            .map(record -> NetFlowFormatter.toMessage(netFlowV5Packet.header(), record, sender))
+                            .collect(Collectors.toList());
+                case ORDERED_V9_MARKER:
+                    // our "custom" netflow v9 that has all the templates in the same packet
+                    return decodeV9(sender, buffer);
+                default:
+                    final List<RawMessage.SourceNode> sourceNodes = rawMessage.getSourceNodes();
+                    final RawMessage.SourceNode sourceNode = sourceNodes.isEmpty() ? null : sourceNodes.get(sourceNodes.size() - 1);
+                    final String inputId = sourceNode == null ? "<unknown>" : sourceNode.inputId;
+                    LOG.warn("Unsupported NetFlow packet on input {} (source: {})", inputId, sender);
+                    return null;
+            }
         } catch (FlowException e) {
             LOG.error("Error parsing NetFlow packet <{}> received from <{}>", rawMessage.getId(), rawMessage.getRemoteAddress(), e);
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("NetFlow packet hexdump:\n{}", ByteBufUtil.prettyHexDump(Unpooled.wrappedBuffer(rawMessage.getPayload())));
+            }
+            return null;
+        } catch (InvalidProtocolBufferException e) {
+            LOG.error("Invalid NetFlowV9 entry found, cannot parse the messages", ExceptionUtils.getRootCause(e));
             return null;
         }
     }
 
+    @VisibleForTesting
+    Collection<Message> decodeV9(InetSocketAddress sender, ByteBuf buffer) throws InvalidProtocolBufferException {
+        final List<NetFlowV9Packet> netFlowV9Packets = decodeV9Packets(buffer);
+
+        return netFlowV9Packets.stream().map(netFlowV9Packet -> netFlowV9Packet.records().stream()
+                .filter(record -> record instanceof NetFlowV9Record)
+                .map(record -> NetFlowFormatter.toMessage(netFlowV9Packet.header(), record, sender))
+                .collect(Collectors.toList())
+        ).flatMap(Collection::stream)
+         .collect(Collectors.toList());
+    }
+
+    @VisibleForTesting
+    List<NetFlowV9Packet> decodeV9Packets(ByteBuf buffer) throws InvalidProtocolBufferException {
+        byte[] v9JournalEntry = new byte[buffer.readableBytes()];
+        buffer.readBytes(v9JournalEntry);
+        final NetFlowV9Journal.RawNetflowV9 rawNetflowV9 = NetFlowV9Journal.RawNetflowV9.parseFrom(v9JournalEntry);
+
+        // parse all templates used in the packet
+        final Map<Integer, NetFlowV9Template> templateMap = Maps.newHashMap();
+        rawNetflowV9.getTemplatesMap().forEach((templateId, byteString) -> {
+            final NetFlowV9Template netFlowV9Template = NetFlowV9Parser.parseTemplate(
+                    Unpooled.wrappedBuffer(byteString.toByteArray()), typeRegistry);
+            templateMap.put(templateId, netFlowV9Template);
+        });
+        final NetFlowV9OptionTemplate[] optionTemplate = {null};
+        rawNetflowV9.getOptionTemplateMap().forEach((templateId, byteString) -> {
+            optionTemplate[0] = NetFlowV9Parser.parseOptionTemplate(Unpooled.wrappedBuffer(byteString.toByteArray()), typeRegistry);
+        });
+
+        return rawNetflowV9.getPacketsList().stream()
+                .map(bytes -> Unpooled.wrappedBuffer(bytes.toByteArray()))
+                .map(buf -> NetFlowV9Parser.parsePacket(buf, typeRegistry, templateMap, optionTemplate[0]))
+                .collect(Collectors.toList());
+    }
+
     @FactoryClass
     public interface Factory extends AbstractCodec.Factory<NetFlowCodec> {
         @Override
@@ -132,12 +208,7 @@ public void overrideDefaultValues(@Nonnull ConfigurationRequest cr) {
         @Override
         public ConfigurationRequest getRequestedConfiguration() {
             final ConfigurationRequest configuration = super.getRequestedConfiguration();
-
-            configuration.addField(new NumberField(CK_CACHE_SIZE, "Maximum cache size", DEFAULT_CACHE_SIZE, "Maximum number of elements in the NetFlow 9 template cache", ConfigurationField.Optional.OPTIONAL));
-            configuration.addField(new TextField(CK_CACHE_PATH, "Cache file path", DEFAULT_CACHE_PATH, "Path to the file persisting the the NetFlow 9 template cache", ConfigurationField.Optional.OPTIONAL));
-            configuration.addField(new NumberField(CK_CACHE_SAVE_INTERVAL, "Cache save interval (seconds)", DEFAULT_CACHE_SAVE_INTERVAL, "Interval in seconds for persisting the cache contents", ConfigurationField.Optional.OPTIONAL));
             configuration.addField(new TextField(CK_NETFLOW9_DEFINITION_PATH, "Netflow 9 field definitions", "", "Path to the YAML file containing Netflow 9 field definitions", ConfigurationField.Optional.OPTIONAL));
-
             return configuration;
         }
     }