搭建流量分析工具elastiflow（基于elk）

一、*功能*

接收网络设备的netflow或sflow报文，对网络设备的数据进行分析，从而得到协议的流量排行、下载IP排行、通信对等信息。

二、*基础环境*

1、安装ELK和java

RHEL server 7，ELK 6.8.21

用rpm安装elasticsearch、logstash、kibana

下载地址：https://www.elastic.co/cn/downloads/past-releases#elasticsearch

rpm -ivh elasticserach-6.8.21.rpm

rpm -ivh logstash-6.8.21.rpm

rpm -ivh kibana-6.8.21-x86_64.rpm

安装java 1.8.0_171或以上（安装方法网上可找到）

2、kibana配置

编辑/etc/kibana/kibana.yml

server.port 5601
server.host: "192.168.11.105"
server.maxPayloadBytes: 8388608
elasticsearch.url: “http://192.168.11.105:9200”
i18n.locale: "zh-CN"

把kibana相关路径的权限修改

chown -R kibana:kibana /etc/kibana

chown -R kibana:kibana /usr/share/kibana

chown kibana:kibana /etc/default/kibana

启动kibana

systemctl enable kibana

systemctl start kibana

2、elasticsearch配置

编辑/etc/elasticsearch/elasticsearch.yml

node.name:net-pd-1
path.data:/data/elisticsearch/data
Path.logs:/data/elasticsearch/logs
bootstrap.memory_lock:true
network.host:192.168.11.105
http.port:9200

编辑/etc/elasticsearch/jvm.options，只改以下部分（大小为1/4 内存）

-Xms64g
-Xmx64g

编辑/usr/lib/systemd/system/elasticsearch.service（第一行下面添加第二行）

LimitFSIZE =infinity 
LimitMEMLOCK=infinity 

把elasticsearch相关路径的权限修改

chown -R elasticsearch:elasticsearch /etc/elasticsearch

chown -R elasticsearch:elasticsearch /usr/share/elasticsearch

chown -R elasticsearch:elasticsearch /data/elisticsearch/data

chown -R elasticsearch:elasticsearch /data/elisticsearch/logs

chown elasticsearch:elasticsearch /etc/sysconfig/elasticsearch

启动elasticsearch
systemctl daemon-reload

systemctl enable elasticsearch

systemctl start elasticsearch

3、logstash配置

编辑/etc/logstash/logstash.yml，data和logs路径是自定义

path.data:/data/logstash/data
config.reload.automatic:true
config.reload.interval:3600s
http.host: "192.168.11.105"
http.port: 9600-9700
path.logs:/data/logstash/logs

编辑/etc/logstash/jvm.options，只改以下部分（大小为1/4 内存）

-Xms64g 
-Xmx64g

编辑/etc/logstash/startup.options，只改以下部分（java 路径）

JAVACMD=/usr/bin/java

把logstash相关路径的权限修改

chown -R logstash:logstash /etc/logstash

chown -R logstash:logstash /usr/share/logstash

chown -R logstash:logstash /data/logstash/data

chown -R logstash:logstash /data/logstash/logs

chown logstash:logstash /etc/default/logstash

启动logstash

systemctl enable logstash

systemctl start logstash

三、*安装过程*

1、安装elastiflow

下载elastiflow：https://github.com/robcowart/elastiflow/releases/tag/v3.4.2 的tar.gz包

tar -zxvf v3.4.2.tar.gz

cd elastiflow-3.4.2

cp -r logstash/elastiflow /etc/logstash/

cp -r logstash.service.d /etc/systemd/system/

chown -R logstash:logstash /etc/logstash/elastiflow

2、elastiflow 配置

禁用/etc/logstash/elastiflow/conf.d/中不用的配置文件(文件名后添加.disabled)

10_input_ipfix_ipv4.logstash.conf.disabled

10_input_ipfix_ipv6.logstash.conf.disabled

10_input_netflow_ipv6.logstash.conf.disabled

10_input_sflow_jpv4.logstash.conf.disabled

10_input_sflow_ipv6.logstash.conf.disabled

20_filter_30_ipfix.logtsh.conf.disabled

20_filter_40_sflow logstash.conf.disabled

30_output_20_multi.logstash.conf.disabled

编辑/etc/systemd/system/logstash.service.d/elastiflow.conf，修改以下部分（NETFLOW的IPv6部分注释掉，IPFIX协议和SFLOW协议全部注释掉）

Environment= "ELASTIFLOW_GEOIP_CACHE_SIZE=12288"
Environment= "ELASTIFLOW_RESOLVE_IP2HOST=true"
Environment= "ELASTIFLOW_ES_HOST=192.168.11.105:9200"
Environment= "ELASTIFLOW_NETFLOW_IPV4_HOST=192.168.11.105"
Environment= "ELASTIFLOW_NETFLOW_IPV4_PORT=2055"

重载systemctl

systemctl daemon-reload

3、logstash 修改配置

编辑/etc/logstash/pipeline.yml (仅当logstash没有其他业务)

#- pipeline.id:main
# path.config:/etc/logstash/conf.d/*.conf
- pipeline.id:elastiflow
  path.config: “/etc/logstash/elastiflow/conf.d/*.conf"

编辑/etc/logstash/elatilow/conf.d/30_output_10_single.logstash.conf，在output的elasticsearch中修改此行

hosts => [ "${ELASTIFLOW_ES_HOST:192.168.11.105:9200}" ]

重启logstash

systemctl restart logstash

（用netstat -ntulp验证是否监听udp 2055端口）

4、kibana 修改配置

将elastiflow-3.4.2/kibana/elastiflow.kibana.6.7.x.json上传到kibana界面(管理→已保存对象→导入)

新建索引(管理→索引模式→创建索引模式) ，取名"elastiflow-*" (必须在启动logstash之后再添加)

5、kibana仪表板

新建仪表板，添加自己惯用的图表（以下是应用排名、客户端流量排名、服务端流量排名、会话流量排名），同时使用筛选器可以过滤出指定ip的分析结果

6、elastiflow设置（如果discover界面中的@timestamp参数慢8小时，可按此方法改正）

编辑/etc/logstash/elastiflow/conf.d/20_filter_10_begin.logstash.conf，在filter中添加

# timezone
ruby {
  code => "event.set('index_date',event.get('@timestamp).time.localtime + 8*60*60)"
} 
mutate {
  convert => [index_date", "string"]
  gsub => ["index_date","T([\S\s]*?)Z",""]
  gsub => ["index_date","-", "."]
}

编辑/etc/logstash/elatilow/conf.d/30_output_10_single.logstash.conf，在output的elasticsearch中注释此行index => "elastiflow-3.4.2-%{index.date}"

#index => "elastiflow-3.4.2 -%{+YYY.MM.dd}"
index => "elastiflow-3.4.2-%{index.date}"

四、*网络设备netflow配置模板*

*思科:*

int GigabitEthernet0/0
 ip flow ingress
 ip flow egress
ip flow-export source GigabitEthernet0/0
ip flow-export version 5
ip flow-export destination 192.168.11.105 2055

*瞻博:*

set services flow- monitoring
set interfaces ge-0/0/0 unit 0 family inet sampling input
set interfaces ge-0/0/0 unit 0 family inet sampling output
set forwarding-options sampling input rate 1000
set forwarding-options sampling input run-length 0
set forwarding-options sampling input max-packets-per-second 2000
set forwarding-options sampling family inet output flow-server 192.168.11.105 port 2055
set forwarding-options sampling family inet output flow-server 192.168.11.105 source-address 192.168.11.106
set forwarding-options sampling family inet output flow-server 192.168.11.105 version 5

*华为/华三:*

sampler2 mode random packet-interval 2000
ip netstream export index-switch 32（部分华为设备默认接口索引是16位，故需要此设置）
ip netstream export version 5 origin-as
ip netstream export host 192.168.11.105 2055
ip netstream export source interface GigabitEthernet0/0
interface GigabitEthernet0/0
 ip netstream inbound
 ip netstream outbound
 ip netstream inbound sampler 2
 ip netstream outbound sampler 2

五、*网络设备sflow配置模板(仅针对不支持netflow的设备)*

1、logstash安装sflow插件

在 https://gems.ruby-china.com/gems/logstash-codec-sflow 下载logstash-codec-sflow插件，注意和logstash的版本适配（logstash 6.8.1需要sflow 2.1.3）。

用zip打包成logstash-codec-sflow.zip，上传到服务器的/tmp

cd /usr/share/logstash

bin/logstash-plugin install file:///tmp/logstash-codec-sflow.zip

安装完插件再次修改权限

chown -R logstash:logstash /usr/share/logstash

2、编辑/etc/systemd/system/logstash.service.d/elastiflow.conf，把sflow取消注释（除了ipv6部分）

Environment="ELASTIFLOW_SFLOW_IPV4_HOST=192.168.11.105"
Environment="ELASTIFLOW_SFLOW_IPV4_PORT=6343"
Environment="ELASTIFLOW_SFLOW_UDP_WORKERS=4"
Environment="ELASTIFLOW_SFLOW_UDP_QUEUE_SIZE=4096"
Environment="ELASTIFLOW_SFLOW_UDP_RCV_BUFF=33554432"

重载systemctl

systemctl daemon-reload

3、解禁/etc/logstash/elastiflow/conf.d/中sflow配置文件(文件名后删除.disabled)

10_input_sflow_ipv4.logstash.conf

20_filter_40_sflow.logstash.conf

4、编辑/etc/logstash/elastiflow/conf.d/20_filter_40_sflow.logstash.conf (sflow的node.ipaddr默认是agent ip，要改成管理ip)，注释以下内容

    #mutate {
    #  id => "sflow_set_node_agent_ip"
    #  replace => {
    #    "[node][ipaddr]" => "%{[agent_ip]}"
    #    "[node][hostname]" => "%{[agent_ip]}"
    #  }
    #}

5、重启logstash

systemctl restart logstash

（用netstat -ntulp验证是否监听udp 2055和udp 6343端口）

瞻博sflow (例如EX4200) :

set protocols sflow collector 192.168.11.105

set protocols sflow collector udp-port 6343

set protocols sflow interfaces ge-0/0/0.0

set protocols sflow polling-interval 60

set protocols sflow sample-rate 1000

set protocols sflow source-ip 192.168.11.130

注意:

EX系列的sflow 包含的接口索引是物理接口索引，即使流量是子接口产生的!

六、*设备名和接口名映射*

1、设备名

编辑/etc/hosts， elastiflow 会根据node.ipaddr来解析node.hostname。格式：

192.168.11.106 RT4
192.168.11.108 vMx-1

2、接口名

编辑/etc/logstash/elastiflow/dictionaries/ifName.yml，elastiflow 会根据node.ipaddr和ifindex来获取ifname。格式：

"192.168.11.106::ifName.1": "Gi0/0"
"192.168.11.108::ifName.513": "ge-0/0/0"
"192.168.11.108::ifName.523": "ge-0/0/0.0"

设备名和接口名的效果图如下：

修改hosts文件和ifName.yml文件后要重启logstash生效