​序言：本文搭建日志审计系统，8g内存+1T硬盘，单机实现每秒3000eps的吞吐量，主要架构如下：

客户端日志监听工具：
    evtsys：监听主机系统，网络设备的日志
    filebeat：监听应用日志
日志服务器接收日志：
    logstash：监听客户机发送的日志，做简单的处理，规整日志
日志服务器存储日志：
    elasticsearch：存储日志，类似于nosql数据库，按天建立索引
日志缓存队列：
    根据日志吞吐量进行选择：首选kafka，activeMQ、rabbitMQ次之，使用ZeroMQ技术成本偏高  

整体框架如下：

一 logstash安装配置

logstash依赖JaVA环境，首先安装jdk，略下载logstash：进入es官网下载最新版本logstash。
elasticsearch官网：https://www.elastic.co/cn/downloads 
（1）下载logstash压缩文件：logstash-5.4.1.tar.gz
（2）赋权：chmod +x logstash-5.4.1.tar.gz
（3）解压缩：tar -zxvf logstash-5.4.1.tar.gz
（4）在config目录下创建：test.conf文件
（5）修改test.conf配置文件，如下：
input {
  # 接收syslog日志
  syslog {
    port => "514"
    type => "syslog"
    codec=>plain{charset=>"ISO-8859-1"}
  }
  # 接收snmptrap日志
  snmptrap {
    port => "162"
    type => "snmpTrap"
  }
  #接收filebeat的日志
  beats {
   port => "5044"
   type => "app"
  }
}
​
filter {
  #去掉换行符
  mutate{
    gsub => [ "message", "\r", "" ]
    gsub => [ "message", "\n", "" ]
  }
  # 由于应用日志是非标准syslog日志，故针对filebeat的应用日志type（例：tomcat,ip）格式做处理
  if [type] =~ /(,)/ {
      mutate {
        split => ["type",","]
        # 新增字段，原字段转换
        add_field => {
            "hostname" => "%{[host]}"
            "appType" => "%{[type][0]}"
            "isApp" => 1
            "timestamp" => "yyyy-MM-dd HH:mm:ss"
        }
        # 删除不规则字段
        remove_field => [
            "tags",
            "offset",
            "host"
        ] 
        # 转换字段的类型
        convert => { "isApp" => "integer" }
      }   
      mutate {
          add_field => {
            "host" => "%{[type][1]}"
          }  
          remove_field => [
            "type",
            "beat"
          ] 
      }
  }else{ #非应用日志的处理
      mutate {
          rename => [ "logsource", "hostname" ]
      }
      mutate { 
        add_field => {
            "isApp" => 0
        }
        remove_field => [
            "tags",
            "priority"
        ]
        convert => {
           "isApp" => "integer"
        }
      }
  }
  # 规整时间，加8个小时
  ruby {
    code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
  }
  mutate {
        remove_field => [
            "@version",
            "tags"
        ]
  }
}
output {
     # 使用kafka作为队列
     kafka{
        topic_id => "loginfoTopic"  #kafka处理接收日志的topic
        # kafka的地址
        bootstrap_servers => "localhost:9092" 
        # 一定要注明输出格式
        codec => "json"
     }
}   
（6）启动logstash
创建日志路径并制定日志目录
nohup /usr/local/wutongyu/tools/logstash/bin/logstash -f /usr/local/wutongyu/tools/logstash/config/test.conf 1>>/opt/log/siem/logstash.log 2>>/opt/log/siem/logstash.log &
二 elasticsearch 安装配置

详见：https://blog.csdn.net/wutongyuWxc/article/details/78888190

三 kafka 安装配置

1.下载最新版kafka 
    Apache官网: http://kafka.apache.org/downloads.html
2.解压kafka
    tar -zxvf kafka_2.12-1.0.0.tgz
3.重命名解压目录名  
    mv kafka_2.12-1.0.0 kafka
4.启动
    进入解压目录：cd kafka
    修改config/zookeeper.properties的日志路径
    启动zookeeper：bin/zookeeper-server-start.sh config/zookeeper.properties &
    修改config/server.properties的日志路径
    启动kafka：bin/kafka-server-start.sh config/server.properties &

kafka结合spring的使用细则详见：http://blog.csdn.net/wutongyuwxc/article/details/79148011

 

四 客户端日志监听配置

evtsys配置

evtsys用于监听Windows、Linux、交换机、防火墙等系统日志，通过514端口将日志发送到logstash
(1)下载最新版 evtsys;
(2)将evtsys.exe文件拷贝到C:\Windows\System32
(3)以管理员身份打开命令窗口执行如下命令：
    net evtsys stop
    evtsys -i ip:端口
    net evtsys start

filebeat配置

    filebeat用于监听应用系统产生的日志，通过5044端口将日志发送到logstash,根据主机系统的类别和位数下载对应版本filebeat。
(1)进入es官网：https://www.elastic.co/cn/products/beats 下载 filebeat
(2)解压文件：tar -zxvf filebeat-5.4.1-linux-x86_64.tar.gz
(3)修改配置文件filebeat.yml
filebeat.prospectors:
 -
    input_type: log
    paths:
        - /opt/log/imp/localhost_access_log*.txt  #tomcat访问日志路径
    exclude_lines: ["(200 )"]   #过滤正常请求
    close_older: 5h  #连续5h没有日志产生，就不监听该日志文件
    scan_frequency: 10s #每10s检查一次是否有新日志产生
    document_type: "tomcat,192.168.211.68"
​
 - 
    input_type: log
    paths:
        - /opt/log/imp/catalina*.log  #tomcat错误日志
    exclude_lines: ["^DBG"]    #过滤debug日志
    close_older: 5h
    scan_frequency: 10s
    document_type: "tomcat,192.168.211.68"
​
 -
    input_type: log
    paths:
        - /dom/loglink/mysql/mysqld.log  #mysql错误日志
    exclude_lines: ["([Note])"]    
    scan_frequency: 60s
    document_type: "mysql-error,192.168.211.68"
# 日志通过514端口输出到logstash
output.logstash:
  hosts: ["192.168.211.50:5044"]
​
(4)启动：
    ./filebeat -e -c filebeat.yml -d "Publish"
或者:
    nohup ./filebeat -e -c filebeat.yml >/dev/null 2>&1 &