[ELK] ELK+Filebeat变成ELK stack

发布时间:2024年01月19日

一、Filebeat+ELK 部署

在 Node1 节点上操作

1.?安装 Filebeat
#上传软件包 filebeat-6.7.2-linux-x86_64.tar.gz 到/opt目录
tar zxvf filebeat-6.7.2-linux-x86_64.tar.gz
mv filebeat-6.7.2-linux-x86_64/ /usr/local/filebeat
2. 设置 filebeat 的主配置文件?
cd /usr/local/filebeat

vim filebeat.yml
filebeat.inputs:
- type: log         #指定 log 类型,从日志文件中读取消息
  enabled: true
  paths:
    - /var/log/messages       #指定监控的日志文件
    - /var/log/*.log
  tags: ["sys"]		#设置索引标签
  fields:           #可以使用 fields 配置选项设置一些参数字段添加到 output 中
    service_name: filebeat
    log_type: syslog
    from: 192.168.80.13

--------------Elasticsearch output-------------------
(全部注释掉)

----------------Logstash output---------------------
output.logstash:
  hosts: ["192.168.80.12:5044"]      #指定 logstash 的 IP 和端口

?

?

3. 启动 filebeat?
nohup ./filebeat -e -c filebeat.yml > filebeat.out &
#-e:输出到标准输出,禁用syslog/文件输出
#-c:指定配置文件
#nohup:在系统后台不挂断地运行命令,退出终端不会影响程序的运行

?

4.在 Logstash 组件所在节点上新建一个 Logstash 配置文件?
input {
   beats { port => "5044" }

}
#filter {}

output {
   elasticsearch {
        hosts => ["192.168.136.180:9200","192.168.136.190:9200","192.168.136.195:9200"]
        index => "%{[fields][service_name]}-%{+YYYY.MM.dd}"
   }
   stdout { codec => rubydebug }
}

#filebeat发送给logstash的日志内容会放到message字段里面,logstash使用grok插件正则匹配message字段内容进行字段分割
#Kibana自带grok的正则匹配的工具:http://<your kibana IP>:5601/app/kibana#/dev_tools/grokdebugger
# %{IPV6}|%{IPV4} 为 logstash 自带的 IP 常量


#启动 logstash
logstash -f filebeat.conf
?5.浏览器访问

?

二、grok 正则捕获插件

?作用:grok 使用文本片段切分的方式来切分日志事件

2.1?内置正则表达式调用?

格式:%{SYNTAX:SEMANTIC}

●SYNTAX代表匹配值的类型,例如,0.11可以NUMBER类型所匹配,10.222.22.25可以使用IP匹配。

●SEMANTIC表示存储该值的一个变量声明,它会存储在elasticsearch当中方便kibana做字段搜索和统计,你可以将一个IP定义为客户端IP地址client_ip_address,如%{IP:client_ip_address},所匹配到的值就会存储到client_ip_address这个字段里边,类似数据库的列名,也可以把 event log 中的数字当成数字类型存储在一个指定的变量当中,比如响应时间http_response_time,假设event log record如下:

message: 192.168.80.10 GET /index.html 15824 0.043

可以使用如下grok pattern来匹配这种记录
%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}

在logstash conf.d文件夹下面创建filter conf文件,内容如下
# /etc/logstash/conf.d/01-filter.conf
filter {
  grok {
    match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" }
  }
}

以下是filter结果
client_id_address: 192.168.80.10
method: GET
request: /index.html
bytes: 15824
http_response_time: 0.043

?logstash 官方也给了一些常用的常量来表达那些正则表达式,可以到这个 Github 地址查看有哪些常用的常量:
https://github.com/logstash-plugins/logstash-patterns-core/blob/main/patterns/ecs-v1/grok-patterns

?

?

?

?

2.2?自定义表达式调用?

语法:(?<field_name>pattern)
举例:捕获10或11和长度的十六进制数的queue_id可以使用表达式(?<queue_id>[0-9A-F]{10,11})

message: 192.168.80.10 GET /index.html 15824 0.043

(?<remote_addr>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) (?<http_method>[A-Z]+) (?<request_uri>/.*) (?<response_bytes>[0-9]+) (?<response_time>[0-9\.]+)

filter {
  grok {
    match => { "message" => "(?<remote_addr>%{IP}) (?<http_method>[A-Z]+) (?<request_uri>/.*) (?<response_bytes>[0-9]+) (?<response_time>[0-9\.]+)"}
  }
}


如果表达式匹配失败,会生成一个tags字段,字段值为 _grokparsefailure,需要重新检查上边的match配置解析是否正确。

三、multiline 多行合并插件

java错误日志一般都是一条日志很多行的,会把堆栈信息打印出来,当经过 logstash 解析后,每一行都会当做一条记录存放到 ES, 那这种情况肯定是需要处理的。 这里就需要使用 multiline 插件,对属于同一个条日志的记录进行拼接。

3.1?安装 multiline 插件?

在线安装插件
cd /usr/share/logstash
bin/logstash-plugin install logstash-filter-multiline

离线安装插件
先在有网的机器上在线安装插件,然后打包,拷贝到服务器,执行安装命令
bin/logstash-plugin prepare-offline-pack --overwrite --output logstash-filter-multiline.zip logstash-filter-multiline

bin/logstash-plugin install file:///usr/share/logstash/logstash-filter-multiline.zip

检查下插件是否安装成功,可以执行以下命令查看插件列表
bin/logstash-plugin list

3.2?使用 multiline 插件?

第一步:每一条日志的第一行开头都是一个时间,可以用时间的正则表达式匹配到第一行。
第二步:然后将后面每一行的日志与第一行合并。
第三步:当遇到某一行的开头是可以匹配正则表达式的时间的,就停止第一条日志的合并,开始合并第二条日志。
第四步:重复第二步和第三步。?

input {
    file {
       path => "/opt/java.log"
       start_position => "beginning"
       sincedb_path => "/etc/logstash/sincedb_path/log_progress"
    }
}

filter {
   multiline {
   pattern => "^\d{4}-\d{1,2}-\d{1,2} \d{1,2}:\d{1,2}:\d{1,2}.\d{1,3}"
   negate => true
   what => "previous"
   }

}

output {
   elasticsearch {
        hosts => ["192.168.136.180:9200","192.168.136.190:9200","192.168.136.195:9200"]
        index => "java-error-%{+yyy.MM.dd}"
   }
}


●pattern:用来匹配文本的表达式,也可以是grok表达式

●what:如果pattern匹配成功的话,那么匹配行是归属于上一个事件,还是归属于下一个事件。previous: 归属于上一个事件,向上合并。next: 归属于下一个事件,向下合并

●negate:是否对pattern的结果取反。false:不取反,是默认值。true:取反。将多行事件扫描过程中的行匹配逻辑取反(如果pattern匹配失败,则认为当前行是多行事件的组成部分)

?

四、mutate 数据修改插件?

它提供了丰富的基础类型数据处理能力。可以重命名,删除,替换和修改事件中的字段。

4.1?Mutate 过滤器常用的配置选项?

选项作用
add_field?? ?向事件添加新字段,也可以添加多个字段
remove_field ?? ? ?从事件中删除任意字段
add_tag向事件添加任意标签,在tag字段中添加一段自定义的内容,当tag字段中超过一个内容的时候会变成数组
remove_tag从事件中删除标签(如果存在)
convert将字段值转换为另一种数据类型
id向现场事件添加唯一的ID
lowercase将字符串字段转换为其小写形式
replace用新值替换字段
strip?? ??删除开头和结尾的空格
uppercase将字符串字段转换为等效的大写字母
update用新值更新现有字段
rename重命名事件中的字段
gsub通过正则表达式替换字段中匹配到的值
merge合并数组或 hash 事件
split通过指定的分隔符分割字段中的字符串为数组

4.2?将字段path重命名为new_path

filter {
    mutate {
	    #写法1,使用中括号括起来
        rename => ["path" => "new_path"]

        #写法2,使用大括号{}括起来
	    rename => { "path" => "new_path" }		
    }
}

4.3?添加字段?

filter {
    mutate {
        add_field => {
        	"f1" => "field1"
        	"f2" => "field2"
        }
    }
}

?4.4?将字段删除?

filter {
    mutate {
        remove_field  =>  ["message", "@version", "tags"]
    }
}

五、date 时间处理插件

用于分析字段中的日期,然后使用该日期或时间戳作为事件的logstash时间戳。

在Logstash产生了一个Event对象的时候,会给该Event设置一个时间,字段为“@timestamp”,同时,我们的日志内容一般也会有时间,但是这两个时间是不一样的,因为日志内容的时间是该日志打印出来的时间,而“@timestamp”字段的时间是input插件接收到了一条数据并创建Event的时间,所有一般来说的话“@timestamp”的时间要比日志内容的时间晚一点,因为Logstash监控数据变化,数据输入,创建Event导致的时间延迟。这两个时间都可以使用,具体要根据自己的需求来定。

filter {    
   grok{    
           match => {"message" => ".+\[(?<log_time>.+)\].+"}
      }    
      date{  
           match => ["log_time","dd/MMM/yyyy:HH:mm:ss Z"]
           target => "@timestamp"
           timezone => "Asia/Shanghai"
    }
}   


●match:用于配置具体的匹配内容规则,前半部分内容表示匹配实际日志当中的时间戳的名称,后半部分则用于匹配实际日志当中的时间戳格式,这个地方是整条配置的核心内容,如果此处规则匹配是无效的,则生成后的日志时间戳将会被input插件读取的时间替代。
如果时间格式匹配失败,会生成一个tags字段,字段值为 _dateparsefailure,需要重新检查上边的match配置解析是否正确。

●target:将匹配的时间戳存储到给定的目标字段中。如果未提供,则默认更新事件的@timestamp字段。

●timezone:当需要配置的date里面没有时区信息,而且不是UTC时间,需要设置timezone参数。

时间戳详解:
●年
yyyy ?#全年号码。 例如:2015。
yy ? ?#两位数年份。 例如:2015年的15。

●月
M ? ? #最小数字月份。 例如:1 for January and 12 for December.。
MM ? ?#两位数月份。 如果需要,填充零。 例如:01 for January ?and 12 for Decembe
MMM ? #缩短的月份文本。 例如: Jan for January。 注意:使用的语言取决于您的语言环境。 请参阅区域设置以了解如何更改语言。
MMMM ?#全月文本,例如:January。 注意:使用的语言取决于您的语言环境。

●日
d ? #最少数字的一天。 例如:1月份的第一天1。
dd ?#两位数的日子,如果需要的话可以填零.例如:01 for the 1st of the month。

●时
H ? #最小数字小时。 例如:0表示午夜。
HH ?#两位数小时,如果需要填零。 例如:午夜00。

●分
m ? #最小的数字分钟。 例如:0。
mm ?#两位数分钟,如果需要填零。 例如:00。

●秒
s ? ?#最小数字秒。 例如:0。
ss ? #两位数字,如果需要填零。 例如:00。

●毫秒( 秒的小数部分最大精度是毫秒(SSS)。除此之外,零附加。)
S ? ?#十分之一秒。例如:0为亚秒值012
SS ? #百分之一秒 例如:01为亚秒值01
SSS ?#千分之一秒 例如:012为亚秒值012

●时区偏移或身份
Z ? ?#时区偏移,结构为HHmm(Zulu/UTC的小时和分钟偏移量)。例如:-0700。
ZZ ? #时区偏移结构为HH:mm(小时偏移和分钟偏移之间的冒号)。 例如:-07:00。
ZZZ ?#时区身份。例如:America/Los_Angeles。 注意:有效的ID在列表中列出http://joda-time.sourceforge.net/timezones.html

文章来源:https://blog.csdn.net/Cnm_147258/article/details/135700072
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。