允许防火墙放行。
打开esxi客户端-选择主机-主机配置-安全配置文件-防火墙-编辑-勾选syslog服务器，点击确定。
3.1.1.2 参考链接
Vmware Esxi syslog配置
在Esxi上配置syslog
Monitoring VMWare ESXi with the ELK Stack
3.1.2 添加logstash 配置文件
3.1.2.1 创建配置测试文件
cat > /data/config/test-vmware.config << EOF
input{
udp {port => 514 type => "Hillstone"}
}
output {
stdout { codec=> rubydebug }
}
EOF
logstash -f test-vmware.config
3.1.2.2 选取其中一条日志进行分析调试。
<167>2019-12-03T07:36:11.689Z localhost.localdomain Vpxa: verbose vpxa[644C8B70] [Originator@6876 sub=VpxaHalCnxHostagent opID=WFU-2d14bc3d] [WaitForUpdatesDone] Completed callback\n
3.1.2.3 结合网上的综合案例对测试文件进行配置改造。
cat > vmware.conf < 514
type => "vmware"
}
}
filter {
if "vmware" in [type] {
grok {
break_on_match => true
match => [
"message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: (?(?(?:\[%{DATA:message_thread_id} %{DATA:syslog_level} \'%{DATA:message_service}\'\ ?%{DATA:message_opID}])) \[%{DATA:message_service_info}]\ (?(%{GREEDYDATA})))",
"message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: (?(?(?:\[%{DATA:message_thread_id} %{DATA:syslog_level} \'%{DATA:message_service}\'\ ?%{DATA:message_opID}])) (?(%{GREEDYDATA})))",
"message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: %{GREEDYDATA:syslog_message}"
]
}
date {
match => [ "syslog_timestamp", "YYYY-MM-ddHH:mm:ss", "ISO8601" ]
}
mutate {
replace => [ "@source_host", "%{syslog_hostname}" ]
}
mutate {
replace => [ "@message", "%{syslog_message}" ]
}
mutate {
remove_field => ["@source_host","program","@timestamp","syslog_hostname","@message"]
}
if "Device naa" in [message] {
grok {
break_on_match => false
match => [
"message", "Device naa.%{WORD:device_naa} performance has %{WORD:device_status}%{GREEDYDATA} of %{INT:datastore_latency_from}%{GREEDYDATA} to %{INT:datastore_latency_to}",
"message", "Device naa.%{WORD:device_naa} performance has %{WORD:device_status}%{GREEDYDATA} from %{INT:datastore_latency_from}%{GREEDYDATA} to %{INT:datastore_latency_to}"
]
}
}
if "connectivity issues" in [message] {
grok {
match => [
"message", "Hostd: %{GREEDYDATA} : %{DATA:device_access} to volume %{DATA:device_id} %{DATA:datastore} (following|due to)"
]
}
}
if "WARNING" in [message] {
grok {
match => [
"message", "WARNING: %{GREEDYDATA:vmware_warning_msg}"
]
}
}
}
}
output {
elasticsearch {
hosts => "192.168.20.18:9200" #elasticsearch服务地址
index => "logstash-vmware-%{+YYYY.MM.dd}"
}
# stdout { codec=> rubydebug }
}
EOF
3.1.2.3 参考文件
mutate基本用法
基本logstash配置文件参考
vmware and syslog
logstash VCSA6.0
filter plugins
3.2 Vcenter日志收集部分
3.2.1 Vcenter-syslog服务器配置
主要是收集vcsa机器日志，方便进行安全日志分析；
主要通过syslog进行日志收集，再通过ELK栈提供的logstash进行分析
VCSA-Syslog--Logstash--Elasticsearch
3.2.1.1 配置步骤
打开VCSA的管理后台URL: http://192.168.20.90:5480，输入账号和密码（开机root和密码)--点击syslog配置中心，输入syslog配置信息。
3.2.1.2 参考链接
Vmware Esxi syslog配置
VCSA 6.5 forward to multiple syslog
VCSA syslog
3.2.2 添加logstash 配置文件
3.2.2.1 创建测试配置
input{
udp {
port => 1514
type => "vcenter"
}
}
output {
stdout { codec=> rubydebug }
}
3.2.2.2 选取一条日志进行分析调试
<14>1 2019-12-05T02:44:17.640474+00:00 photon-machine vpxd 4035 - - Event [4184629] [1-1] [2019-12-05T02:44:00.017928Z] [vim.event.UserLoginSessionEvent] [info] [root] [Datacenter] [4184629] [User root@192.168.20.17 logged in as pyvmomi Python/3.6.8 (Linux; 3.10.0-957.el7.x86_64; x86_64)]\n
3.2.2.3 结合网上的综合案例对配置文件进行配置改造。
cat > vcenter.conf < 1514
type => "vcenter"
}
}
filter {
if "vcenter" in [type] {
}
grok {
break_on_match => true
match => [
"message", "<%{NONNEGINT:syslog_pri}>%{NONNEGINT:syslog_ver} +(?:%{TIMESTAMP_ISO8601:syslog_timestamp}|-) +(?:%{HOSTNAME:syslog_hostname}|-) +(-|%{SYSLOG5424PRINTASCII:syslog_program}) +(-|%{SYSLOG5424PRINTASCII:syslog_proc}) +(-|%{SYSLOG5424PRINTASCII:syslog_msgid}) +(?:%{SYSLOG5424SD:syslog_sd}|-|) +%{GREEDYDATA:syslog_msg}"
]
}
date {
match => [ "syslog_timestamp", "YYYY-MM-ddHH:mm:ss,SSS", "YYYY-MM-dd HH:mm:ss,SSS", "ISO8601" ]
#timezone => "UTC" #For vCenter Appliance
#timezone => "Asia/Shanghai"
}
mutate {
remove_field => ["syslog_ver", "syslog_pri"]
}
}
output {
elasticsearch {
hosts => "192.168.20.18:9200" #elasticsearch服务地址
index => "logstash-vcenter-%{+YYYY.MM.dd}"
}
# stdout { codec=> rubydebug }
}
EOF
3.2.2.4 参考文件
mutate基本用法
基本logstash配置文件参考
vmware and syslog
logstash VCSA6.0
4 Windows server部分
4.1 windows server日志收集部分
主要是收集AD域日志，方便进行安全日志分析；
主要通过ELK栈提供的winlogbeat进行收集
winlogbeat--logstash--elasticsearch
4.1.1 windows服务器配置
下载连接地址：https://www.elastic.co/cn/downloads/beats/winlogbeat
将解压后的文件放到“C:\Program Files”，重命名为winlogbeat
命令安装
编辑winlogbeat.yml文件
winlogbeat.event_logs:
- name: Application
- name: Security
- name: System
output.logstash:
enbaled: true
hosts: ["192.168.20.18:5044"]
logging.to_files: true
logging.files:
path: D:\ProgramData\winlogbeat\Logs
logging.level: info
PS C:\Program Files\Winlogbeat> .\winlogbeat.exe test config -c .\winlogbeat.yml -e
启动winlogbeat
powershell命令行启动：
PS C:\Program Files\Winlogbeat> Start-Service winlogbeat
powershell命令行关闭
PS C:\Program Files\Winlogbeat> Stop-Service winlogbeat
导入winlogindex模板,因为我们使用的logstash，所以需要手动导入。
PS > .\winlogbeat.exe setup --index-management -E output.logstash.enabled=false -E 'output.elasticsearch.hosts=["192.168.20.18"]'
导入kibana-dashboard，因为我们使用的logstash，所以需要手动导入。
PS > .\winlogbeat.exe setup -e -E output.logstash.enabled=false -E output.elasticsearch.hosts=['192.168.20.18:9200'] -E setup.kibana.host=192.168.20.18:5601
4.1.2 参考链接
windows 上winlogbeat安装
官方手册分析
4.2 添加logstash 配置文件
4.2.1 创建测试配置文件
cat > /data/config/test-windows.config << EOF
input {
beats {
port => 5044
}
}
output {
stdout { codec=> rubydebug }
}
EOF
logstash -f test-windows.config
4.2.2 创建配置文件
创建正式配置文件，查看内容（因为已有模板，所以不错其他修改处理）
cat > /data/config/windows.config << EOF
input {
beats {
port => 5044
}
}
output {
elasticsearch {
hosts => ["http:192.168.20.18:9200"]
index => "%{[@metadata][beat]}-%{[@metadata][version]}"
}
}
EOF
logstash -f windows.config
4.2.3 参考文件
beats input plugin
4.3查看Discover
因为已经winlogbeat是elk栈的标准模块，已经被定义，所以我们不再自行定义。
直接打开搜索winlogbaet*的index。
4.4 查看dashboard
因为已经winlogbeat是elk栈的标准模块，已经被定义，所以我们不再自行定义。
直接打开搜索winlogbaet*的dashboard
5 Linux server部分
5.1 linux 系统日志收集部分
主要是收集linux上的开关机日志，安全日志；
主要通过ELK栈提供的filebeat进行收集
filebeat-filebeat-module--elasticsearch
5.1.1 安装部署
官网下载，连接地址：https://www.elastic.co/cn/downloads/beats/filebeat
命令行安装
curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.5.0-linux-x86_64.tar.gz
tar xzvf filebeat-7.5.0-linux-x86_64.tar.gz
查看filebeat目录布局
编辑filebeat.yml文件
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/*.log
#- c:\programdata\elasticsearch\logs\*
output.elasticsearch:
hosts: ["192.168.20.18:9200"]
setup.kibana:
host: "192.168.20.18:5601"
运行filebeat文件
filebeat -c filebeat.yml -e
5.1.2 参考链接
windows 上winlogbeat安装
官方手册分析
5.2 修改filebeat配置文件
5.2.1索引名称
关闭ilm声明周期管理
setup.ilm.enabled: false
更改索引名称
setup.template.overwrite: true
output.elasticsearch.index: "systemlog-7.3.0-%{+yyyy.MM.dd}"
setup.template.name: "systemlog"
setup.template.pattern: "systemlog-*"
修改预先构建的kibana仪表盘
setup.dashboards.index: "systemlog-*"
5.2.1 开启system模块
./filebeat modules enable system
./filebeat modules list
5.2.3 重新初始化环境
./filebeat setup --template -e -c filebeat.yml
5.2.4 配置模块，修改配置文件
filebeat.modules:
- module: system
syslog:
enabled: true
#默认位置/var/log/messages* /var/log/syslog*
auth:
enabled: true
#默认位置/var/log/auth.log* /var/log/secure*
output.elasticsearch:
hosts: ["192.168.20.18:9200"]
setup.kibana:
host: "192.168.20.18:5601"
5.2.5 重新启动filebeats，初始化模块
./filebeat setup -e -c filebeat.yml
5.2.6 参考文件
beats input plugin
filebeat模块与配置
system module
5.3 查看discover
5.4 查看dashboard
6 整合logstash配置
6.1 整合logstash配置文件
input{
udp {
port => 516
type => "h4c"
}
}
input{
udp {
port => 518
type => "hillstone"
}
}
input{
udp {
port => 514
type => "vmware"
}
}
input{
udp {
port => 1514
type => "vcenter"
}
}
input {
beats {
port => 5044
type => "windows"
}
}
filter {
if [type] == "hillstone" {
grok {
match => { "message" => "\<%{BASE10NUM:syslog_pri}\>%{SYSLOGTIMESTAMP:timestamp}\ %{BASE10NUM:serial}\(%{WORD:ROOT}\) %{DATA:logid}\ %{DATA:Sort}@%{DATA:Class}\: %{DATA:module}\: %{IPV4:srcip}\:%{BASE10NUM:srcport}->%{IPV4:dstip}:%{WORD:dstport}\(%{DATA:protocol}\), application %{USER:app}\, interface %{DATA:interface}\, vr %{USER:vr}\, policy %{DATA:policy}\, user %{USERNAME:user}\@%{DATA:AAAserver}\, host %{USER:HOST}\, send packets %{BASE10NUM:sendPackets}\,send bytes %{BASE10NUM:sendBytes}\,receive packets %{BASE10NUM:receivePackets}\,receive bytes %{BASE10NUM:receiveBytes}\,start time %{TIMESTAMP_ISO8601:startTime}\,close time %{TIMESTAMP_ISO8601:closeTime}\,session %{WORD:state}\,%{GREEDYDATA:reason}"}
match => { "message" => "\<%{BASE10NUM:syslog_pri}\>%{SYSLOGTIMESTAMP:timestamp}\ %{BASE10NUM:serial}\(%{WORD:ROOT}\) %{DATA:logid}\ %{DATA:Sort}@%{DATA:Class}\: %{DATA:module}\: %{IPV4:srcip}\:%{BASE10NUM:srcport}->%{IPV4:dstip}:%{WORD:dstport}\(%{DATA:protocol}\), interface %{DATA:interface}\, vr %{DATA:vr}\, policy %{DATA:policy}\, user %{USERNAME:user}\@%{DATA:AAAserver}\, host %{USER:HOST}\, session %{WORD:state}%{GREEDYDATA:reason}"}
match => { "message" => "\<%{BASE10NUM:syslog_pri}\>%{SYSLOGTIMESTAMP:timestamp}\ %{BASE10NUM:serial}\(%{WORD:ROOT}\) %{DATA:logid}\ %{DATA:Sort}@%{DATA:Class}\: %{DATA:module}\: %{IPV4:srcip}\:%{BASE10NUM:srcport}->%{IPV4:dstip}:%{WORD:dstport}\(%{DATA:protocol}\), %{WORD:state} to %{IPV4:snatip}\:%{BASE10NUM:snatport}\, vr\ %{DATA:vr}\, user\ %{USERNAME:user}\@%{DATA:AAAserver}\, host\ %{DATA:HOST}\, rule\ %{BASE10NUM:rule}"} match => { "message" => "\<%{BASE10NUM:syslog_pri}\>%{SYSLOGTIMESTAMP:timestamp}\ %{BASE10NUM:serial}\(%{WORD:ROOT}\) %{DATA:logid}\ %{DATA:Sort}@%{DATA:Class}\: %{DATA:module}\: %{IPV4:srcip}\:%{BASE10NUM:srcport}->%{IPV4:dstip}:%{WORD:dstport}\(%{DATA:protocol}\), %{WORD:state} to %{IPV4:dnatip}\:%{BASE10NUM:dnatport}\, vr\ %{DATA:vr}\, user\ %{USERNAME:user}\@%{DATA:AAAserver}\, host\ %{DATA:HOST}\, rule\ %{BASE10NUM:rule}"}
}
mutate {
lowercase => [ "module" ]
remove_field => ["host", "message", "ROOT", "HOST", "serial", "syslog_pri", "timestamp", "mac", "AAAserver", "user"]
}
}
if [type] == "h4c" {
grok {
match => { "message" => "\<%{BASE10NUM:syslog_pri}\>%{SYSLOGTIMESTAMP:timestamp}\ %{DATA:year} %{DATA:hostname} \%\%%{DATA:ddModuleName}\/%{POSINT:severity}\/%{DATA:brief}\: %{GREEDYDATA:reason}" }
add_field => {"severity_code" => "%{severity}"}
}
mutate {
gsub => [
"severity", "0", "Emergency",
"severity", "1", "Alert",
"severity", "2", "Critical",
"severity", "3", "Error",
"severity", "4", "Warning",
"severity", "5", "Notice",
"severity", "6", "Informational",
"severity", "7", "Debug"
]
remove_field => ["message", "syslog_pri"]
}
}
if [type] == "vmware" {
grok {
break_on_match => true
match => [
"message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: (?(?(?:\[%{DATA:message_thread_id} %{DATA:syslog_level} \'%{DATA:message_service}\'\ ?%{DATA:message_opID}])) \[%{DATA:message_service_info}]\ (?(%{GREEDYDATA})))",
"message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: (?(?(?:\[%{DATA:message_thread_id} %{DATA:syslog_level} \'%{DATA:message_service}\'\ ?%{DATA:message_opID}])) (?(%{GREEDYDATA})))",
"message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: %{GREEDYDATA:syslog_message}"
]
}
date {
match => [ "syslog_timestamp", "YYYY-MM-ddHH:mm:ss", "ISO8601" ]
}
mutate {
replace => [ "@source_host", "%{syslog_hostname}" ]
}
mutate {
replace => [ "@message", "%{syslog_message}" ]
}
mutate {
remove_field => ["@source_host","program","syslog_hostname","@message"]
}
}
if [type] == "vcenter" {
grok {
break_on_match => true
match => [
"message", "<%{NONNEGINT:syslog_pri}>%{NONNEGINT:syslog_ver} +(?:%{TIMESTAMP_ISO8601:syslog_timestamp}|-) +(?:%{HOSTNAME:syslog_hostname}|-) +(-|%{SYSLOG5424PRINTASCII:syslog_program}) +(-|%{SYSLOG5424PRINTASCII:syslog_proc}) +(-|%{SYSLOG5424PRINTASCII:syslog_msgid}) +(?:%{SYSLOG5424SD:syslog_sd}|-|) +%{GREEDYDATA:syslog_msg}"
]
}
date {
match => [ "syslog_timestamp", "YYYY-MM-ddHH:mm:ss,SSS", "YYYY-MM-dd HH:mm:ss,SSS", "ISO8601" ]
}
mutate {
remove_field => ["syslog_ver", "syslog_pri"]
}
}
}
output {
if [type] == "hillstone" {
elasticsearch {
hosts => "192.168.20.18:9200"
index => "hillstone-%{module}-%{+YYYY.MM.dd}"
}
}
if [type] == "h4c" {
elasticsearch {
hosts => "192.168.20.18:9200"
index => "h4c-%{+YYYY.MM.dd}"
}
}
if [type] == "vmware" {
elasticsearch {
hosts => "192.168.20.18:9200"
index => "vmware-%{+YYYY.MM.dd}"
}
}
if [type] == "vcenter" {
elasticsearch {
hosts => "192.168.20.18:9200"
index => "vcenter-%{+YYYY.MM.dd}"
}
}
if [type] == "windows" {
elasticsearch {
hosts => "192.168.20.18:9200"
index => "%{[@metadata][beat]}-%{[@metadata][version]}"
}
}
}
6.2 配置logstash服务
启动logstash服务，并把mix.config 更改为logstash.conf ,放到/etc/logstash 目录下。
systemctl enable logstash
systemc start logstash
未完待续
推荐查看英文文档方式
找到一篇英文站点，将鼠标移动到url开始，添加icopy.site/回车。
"icopy.site/"+"https://www.elastic.co"
例如 ： 源网址：https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html
转译后网址：https://s0www0elastic0co.icopy.site/guide/en/logstash/current/plugins-filters-grok.html
推荐理由：
1.比谷歌全文翻译更准确，而且关键代码不翻译。
2.如果你英文不好，或者看英文文档太累，可以试下哦。
