一、logstash简介

Logstash是一个开源的服务器端数据处理管道。
logstash拥有200多个插件，能够同时从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的 “存储库” 中。（大多都是 Elasticsearch。）
Logstash管道有两个必需的元素，输入和输出，以及一个可选元素过滤器。
logstash有input,filter以及output三个部分：
input：采集各种样式、大小和来源的数据
Logstash 支持各种输入选择 ，同时从众多常用来源捕捉事件。
能够以连续的流式传输方式，轻松地从您的日志、指标、Web 应用、数据存储以及各种 AWS 服务采集数据。
filter：实时解析和转换数据
数据从源传输到存储库的过程中，Logstash 过滤器能够解析各个事件，识别已命名的字段以构建结构，并将它们转换成通用格式，以便更轻松、更快速地分析和实现商业价值。
利用 Grok 从非结构化数据中派生出结构
从 IP 地址破译出地理坐标
将 PII 数据匿名化，完全排除敏感字段
简化整体处理，不受数据源、格式或架构的影响
output：选择您的存储库，导出您的数据
尽管 Elasticsearch 是我们的首选输出方向，能够为我们的搜索和分析带来无限可能，但它并非唯一选择。
Logstash 提供众多输出选择，您可以将数据发送到您要指定的地方，并且能够灵活地解锁众多下游用例

二、Logstash安装与配置

软件下载地址：https://elasticsearch.cn/download/
logstash安装：

rpm -ivh jdk-8u171-linux-x64.rpm
rpm -ivh logstash-7.6.1.rpm

示例1：标准输入到标准输出
命令行输入：/usr/share/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'

示例2：标准输入到文件

步骤1：编写文件

vim /etc/logstash/conf.d/file.conf
input {
        stdin { }
}
output {
        file {
                path => "/tmp/logstash.txt"        	#输出的文件路径
                codec => line { format => "custom format:%{message}"}				#定制数据格式
        }
}

步骤2：指定配置文件运行

 /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/file.conf

步骤3：在标准输入中键入"wang zhuangzhuang"后可以查看：

示例4：elasticsearch输出插件，标准输入到es主机

vim /etc/logstash/conf.d/es.conf
input {
        stdin { }
}
output {
        elasticsearch {
                hosts => "172.20.10.11:9200"    #输出到的ES主机与端口
                index => "logstash-%{+YYYY.MM.dd}"   #定制索引名称
        }
}

指定配置文件运行

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/es.conf

示例5：把文件内容输出到es主机

vim /etc/logstash/conf.d/apache.conf
input {
        file {
		path => "/var/log/httpd/access_log"
		start_position => "beginning"
	}
}
output {
        elasticsearch {
                hosts => "172.20.10.11:9200"
                index => "apachelog-%{+YYYY.MM.dd}"
        }
}

指定配置文件运行

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/apache.conf

示例6：logstash可以伪装成日志服务器，直接接受远程日志

vim /etc/logstash/conf.d/syslog.conf
input {
        syslog {
                type => "rsyslog"
                port => 514		#日志收集端口
        }
}
output {
        stdout { }
}

配置客户端传入日志

vim /etc/rsyslog.conf 
*.*         @@172.20.10.9:514    
systemctl restart rsyslog  

指定配置文件运行

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/syslog.conf

三、多行过滤插件

多行过滤可以把多行日志记录合并为一行事件

systemctl restart elasticsearch.service 
input {
	stdin {
	codec => multiline {
		pattern => "^EOF"
		negate => true
		what => previous
		}
	}
}

output {
	stdout {}
}

指定配置文件运行

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/multiline.conf

四、grok过滤插件

vim /etc/logstash/conf.d/grok.conf
input {
	stdin {}
}
filter {
	grok {
	match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
	}
}
output {
	stdout {}
}

指定配置文件运行

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf

查看输出：55.3.244.1 GET /index.html 15824 0.043

示例：apache服务日志过滤实战

vim /etc/logstash/conf.d/apache.conf
input {
        file {
		path => "/var/log/httpd/access_log"
		start_position => "beginning"
	}
}
filter{
	grok {
		match => { "message" => "%{HTTPD_COMBINEDLOG}" }
	}
}
output {
        elasticsearch {
                hosts => "172.25.0.13:9200"
                index => "apachelog-%{+YYYY.MM.dd}"
        }
}

没有过滤的日志显示格式如下：