Kibana dashboard showing various NetFlow metrics

I've heard a lot about ElasticSearch lately, was trying to create some time to get a lab set up for the new trio on the block : ELK. For those who hasn't heard about the term ELK, it is an acronym for ElasticSearch + Logstash and Kibana. ELK stack is continuing tradition LAMP stack created a while ago by tightly integrating to each other, albeit on a completely different dimension, and becoming new invaluable tool for DevOps people.
Over the course of years, I've developed a lot of tools/guis, both small and big, to make metrics and data meaningful for my pleasure/business/troubleshooting purposes. But none was as much fun and as quick as I had with ElasticSearch and Kibana.

One of the most important advices I can give to anyone who is building, maintaining or operating an IT infrastructure is having situational awareness on every single angle possible. This means collecting a lot of metrics from all systems, including IPfix/NetFlows from network. Main focus of this tutorial is to show how ES and Kibana can be a valuable tool in assessing issues at network layer using Netflow on a real life scenario: On February 4th, 2014 a network issue caused 1 hour disruption to the services provied to a customer, an RCA requested by management. All logs gathered in one place, with very few reliable explanation as to what really happened. We've started looking at the issue deeper, this time utilizing NetFlows captured at various devices and using ES & Kibana to do analytics & drill down. ES & Kibana helped a lot to better understand and grasp what happened during the disruption and nailing down root cause.

I won't be going into details of setting up ElasticSearch and Kibana, as there are a lot of blog posts on the net on how to perform those steps. You'll see steps to get NFdump data ready, importing into ElasticSearch using ElasticSearch's Python & Bulk API, preparing Kibana for analytics, and discovering what NetFlow records show about the issue I'm after.
In addition to explaining steps to setup similar NFdump/ES/Kibana environment for yourself, I've also provided some analysis/benchmark on storage requirement of ES and explained different approaches on how to reduce foot print of ES while increasing performance.

First, some background information.

At our customer sites, we deploy various collectors and probes to collect and store network traffic metadata, namely NetFlow, for forensic/security and troubleshooting purposes. Probes are deployed at critical network edges to record and analyze activity passing through by using SPAN/RSPAN/ERSPAN, emmitting NetFlow metadata to collectors which save and store NetFlows on hard drive for later use. NetFlows can be costly to generate and store, especially if you're trying to capture traffic on outside/untrust/public interfaces, which face traffic/flows, both in and out, from all around the globe. Probing outside interfaces means recording traffic from spoofed IPs, ICMP pings, BGP announcements and every other bits that travels on Layer 3. For this blog post, I've used 3 flows collected at the same device, one from outside/public interface on the firewal, one for traffic filtered by firewall corresponding to the traffic passing through outside interface, and one for internal VLAN relevant to the issue I'm analysing for. All flows corresponding to the same 24 hour period, differed in size and characteristics widely:

1. FNF1 : Corresponding to an internal VLAN traffic, have 4.6 million flows, 228MB in size
2. FNF2 : Corresponding to the internal/trust interface on the firewall, having 8.6 million flows, 426MB in size.
3. FNF3 : Corresponding to the public facing interface on the firewall, having 33.7 million flows, 1.671MB in size.

Netflow captures following fields that can be used to analyze various aspects of the network :

• Timestamp
• Duration
• Protocol
• Source IP
• Source port
• Destination IP
• Destination port
• Bytes
• Packets
• TCP Flags
• Interface

There are a lot of other information that can be captured and shown, a sample 5 line output from FNF1 dump file is shown below :

1:  # nfdump -r fnf1.dump -oextended -c5  
2:  Date flow start     Duration Proto   Src IP Addr:Port     Dst IP Addr:Port  Flags Tos Packets  Bytes   pps   bps  Bpp Flows  
3:  2014-01-01 00:04:23.980  25.000 TCP     172.16.0.156:33510 ->    172.16.0.94:58781 .AP...  0   296  237484    11  75994  802   1  
4:  2014-01-01 00:04:23.980  34.000 TCP    192.168.0.133:20022 ->    172.16.0.121:64327 .AP...  0   375  24206    11   5695   64   1  
5:  2014-01-01 00:04:23.980  25.000 TCP     172.16.0.156:5910 ->    172.16.0.132:63880 .AP...  0   319  173587    12  55547  544   1  
6:  2014-01-01 00:04:23.980  21.000 TCP     172.16.0.22:39621 ->    172.16.2.135:80  .AP...  0    35   9504    1   3620  271   1  
7:  2014-01-01 00:04:23.980  11.000 TCP     172.16.0.99:65192 ->    172.16.9.132:80  .AP..F  0    39   6800    3   4945  174   1  

Step 1) Prepare NFdump files

NFdump rotates dump files every 5 minutes by default. Instead of dealing with multiple NFdump files, I've converted all dump files in each collector's directory to a single file, while sorting by timestamp using following command :


For the sake of performance and analytics I'm after, I'm only interested in timestamp, source IP, source port, destination IP, destination port, bytes, packets, interface and protocol. To do this, I've used following nfdump command to dump data into CSV file for later use :

Step 2) Prepare ElasticSearch Mapping

After having all 3 files dumped into respective CSV files, I've crafted a Python script utilizing ElasticSearch's official Python API to index each flow record in ElasticSearch. ElasticSearch provides schemaless storage and indexing, however just throwing Netflow data without providing a mapping (a schema or DDL some sort) is not smart for storage perspective. Before importing CSV into ElasticSearch, I've experimented with different schemas, one of which using IP mapping for source and destination IP addresses, however it didn't work well for some reason. I've switched storing IP information in string field using following schema definition :

1:  {  
2:  "fnf1x": {  
3:  "_all" : {"enabled" : false},  
4:  "_source" : {"enabled" : false},  
5:  "properties": {  
6:       "ts": {"type": "date", "format" : "YYYY-MM-dd HH:mm:ss"},  
7:       "sa": {"type": "string", "index": "not_analyzed"},  
8:       "sp": {"type": "integer", "index": "not_analyzed"},  
9:       "da": {"type": "string", "index": "not_analyzed"},  
10:       "dp": {"type": "integer", "index": "not_analyzed"},  
11:       "byte": {"type": "long"},  
12:       "pkt": {"type": "long"},  
13:       "out": {"type": "integer", "index": "not_analyzed"},  
14:       "pr": {"type": "string", "index" : "not_analyzed"}  
15:  }  
16:  }  
17:  }  

For those who are not familiar with ElasticSearch's mapping definition, here is a short definition of what this schema does. First, I didn't want to store NetFlow records in both index and in "_source" field as a JSON document, so it is disabled. For drilling down and analytics purposes, complete document, NetFlow record in this case, is rarely needed. Also, since NetFlow records are well defined and structured, I only search/filter using field names ie : protocol = TCP or destination port = 80. "_all" field is used for searching multiple fields at the same time, when no field name provided. Eliminating "_all" field also saves unnecessary I/O to disk and storage. I chose 'not_analyzed' for string fields, as there is no need to tokenize or stem any of the strings stored. IP information along with protocol fields can be stored and indexed as a whole. Please duplicate above mapping for each of the NetFlow collectors after changing "fnf1x" to the appropriate name you choose.

Step 3) Import CSV files into ElasticSearch

After this map is PUT on ElasticSearch, we can have our Python script to import CSV file created on step 1. Python script code is c/p here, as it is less than 30 lines :

1:  #!/usr/bin/python  
2:  import csv, sys, time, json, elasticsearch  
3:  from elasticsearch import helpers  
4:  es      = elasticsearch.Elasticsearch()  
5:  source    = 'fnf1'  
6:  csvfile   = source + '.csv'  
7:  jdata    = dict()  
8:  actions   = list()  
9:  i      = 0  
10:  proto    = {'0': '0', '1': 'ICMP', '2': 'IGMP', '6': 'TCP', '17': 'UDP', '112': 'VRRP', '3': 'GGP', '50': 'ESP'}  
11:  with open(csvfile, 'rb') as file :  
12:       line = csv.reader(file, delimiter = ',', skipinitialspace = True)  
13:       for row in line :  
14:            ptime = time.strptime(row[0][0:19], "%Y-%m-%d %H:%M:%S")  
15:            ctime = time.strftime('%Y-%m-%d %H:%M:%S', ptime)  
16:            jdata = { 'ts': ctime, 'byte': int(row[5]), 'sa': row[1], 'sp': int(float(row[2])), 'da': row[3], 'dp': int(float(row[4])), 'pkt': int(row[6]), 'out': int(row[7]), 'pr': proto[row[8].strip()] }  
17:            action = { '_index': 'netflowlab', '_type': 'fnf1x', '_source': json.dumps(jdata, separators=(',', ':'))}  
18:            actions.append(action)  
19:            i += 1  
20:            if i % 100000 == 0:  
21:                 elasticsearch.helpers.bulk(es, actions)  
22:                 print "Indexed %d, working on next 100000" %(i)  
23:                 actions = list()  
24:       elasticsearch.helpers.bulk(es, actions)  
25:       print "Indexed %d, finishing." %(i)  

Please change 'source' and '_type' fields above to reflect file names and "_type" in ElasticSearch index. Uploading total of 47 million rows, in 3 different CSF files took about 2 hours on my i7-3770 (QC 3.4GHz CPU). Most of the time spent was on Python parsing CSV file and converting into JSON format, I didn't have time to optimize code or profile it for performance tuning. I've used SSD drive to store ElasticSearch data files, which makes upload and analytics faster than traditional drives. Also after inserting 47 million rows, into 3 different types, there will be a lot of segments in ElasticSearch index directories. I suggest optimizing them by reducing number of segments to 1 using (netflowlab is the name of index I've used for this blog post) :

This resulted in 3.8GB of Index directory under /var/lib/elasticsearch/

Step 4) Using Kibana to analyze NetFlow data

After everything is ready, use enclosed Kibana dashboard schema file. Once you have uploaded dashboard schema, you'll have something similar to the image on the right. With my ElasticSearch Index, dashboard shows high level information about 46 million flows, accounting for 5.8TB of transferred data in 24 hours. In the next image, we see histograms showing both byte and PPS values :


Histogram shows some anomaly started happening around 14:30 and 20:00, especially around 14:30 and 15:30


Once we started zooming into the timeframe when the anomaly occured, we can see all other graphs updated according to the window we selected. In the mid section of the dashboard I have source IP, source address, destination IP and destination address pie charts, showing flow itself. In destination port pie chart, I immediately noticed that port 12201 is accounting for roughly 20% of the trafffic/flows happened at that time, which is way above normal characteristic of the traffic :

When I changed histogram properties to show data with 1s resolution, I can see that PPS values went up to 250K alone for FNF1x collector, congesting network switch with both PPS and Throughput (MBit/sec). If I want, I can also drill down to the Interfaces and see how much traffic passed through each interface on the switch. This information showed us that the root cause of the issue we were investigating, was actually app servers pumping huge amounts of logs towards GrayLog server. Whole issue was triggered by another issue, but it was start of chain reaction, causing apps to go crazy with logs, making log pumping the root cause, and trigger itself a contributing factor.

Storage Perspective

I've also captured some information on how much data is required for NetFlow storage when different file formats are in use. JSON, obviously, being one of the least storage efficient file formats requires most storage when it comes to NetFlow data with 46 millon flows. After disabling "_all" and "_source" fields in ElasticSearch, its storage requirements also went down. 900MB of gzip compressed Nfdump data consumes about 3.8 GB of Index space on ElasticSearch. I should add that, I didn't store all Netflow fields in this test scenario, only included ones that are relevant to my use case. To make comparison a little more accurate, I've also added uncompressed Nfdump storage requirements below. Once I compress ElasticSearch's Index directory with .tar.gz (default gzip compression level), same 3.8GB becomes 2.7GB. This tells me that I can also store same Index files on ZFS with LZ compression turned on to save some space without sacrificing too much performance.

	Rows	Nfdump (bz2)	Nfdump	Nfdump (-j)	CSV	Json	ES Index Size	Tar.gz
FNF1	4,616,895	38.33	228.96	85.77	484.33	781.21
FNF2	8,598,766	74.02	426.43	146.18	902.05	1,424.96
FNF3	33,710,008	402.99	1,671.73	667.57	3,536.32	5,854.58
Total	46,925,669	515.35	2,327.12	899.52	4,922.70	8,060.74	3,805.68	2,711.24

Kayra Otaner 11 Mart 2014

Posted In: ElasticSearch, Gezegen, NetFlow