logstash

概览

收集日志数据 开源

输入：采集各种样式、大小和来源的数据

数据往往以各种各样的形式，或分散或集中地存在于很多系统中。Logstash 支持各种输入选择 ，可以在同一时间从众多常用来源捕捉事件。能够以连续的流式传输方式，轻松地从您的日志、指标、Web 应用、数据存储以及各种 AWS 服务采集数据。

过滤器：实时解析和转换数据

数据从源传输到存储库的过程中，Logstash 过滤器能够解析各个事件，识别已命名的字段以构建结构，并将它们转换成通用格式，以便更轻松、更快速地分析和实现商业价值。

Logstash 能够动态地转换和解析数据，不受格式或复杂度的影响：

利用 Grok 从非结构化数据中派生出结构
从 IP 地址破译出地理坐标
将 PII 数据匿名化，完全排除敏感字段
整体处理不受数据源、格式或架构的影响

Grok过滤器

grok：解析任意文本数据，Grok 是 Logstash 最重要的插件。它的主要作用就是将文本格式的字符串，转换成为具体的结构化的数据，配合正则表达式使用。内置120多个解析语法。
官方提供的grok表达式：https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns
grok在线调试：https://grokdebug.herokuapp.com/
mutate：对字段进行转换。例如对字段进行删除、替换、修改、重命名等。
drop：丢弃一部分events不进行处理。
clone：拷贝 event，这个过程中也可以添加或移除字段。
geoip：添加地理信息(为前台kibana图形化展示使用)

对于不理想的日志格式。用过滤器来解析日志文件，
grok 过滤器插件允许你将非结构化日志数据解析为结构化和可查询的数据。
为了解析数据，你可以用 %{COMBINEDAPACHELOG} grok pattern ，这种模式（或者说格式）的schema如下：
应该是%{COMBINEDAPACHELOG} 这种模式就是一种正则处理，将信心处理后能将数据拆分成以下的各个字段，以供以后使用

在logstash的配置文件中加入grok filter过滤，

lostash启动了自动加载配置后，修改过配置就不需要再重启了，但是，对于已经读过的日志文件，需要强制指定filebeat从头读取，所以需要停掉filebeat，然后删除filebeat的注册文件，
rm data/registr

经过过滤的日志文件

在过滤中添加地址信息

输出：选择你的存储，导出你的数据

Logstash 提供众多输出选择，您可以将数据发送到您要指定的地方，并且能够灵活地解锁众多下游用例

安装

可以使用rpm方式安装，也可以下载源码解压使用，依赖于java环境

命令行使用

bin/logstash [options]

最基本的通道
bin/logstash -e 'input { stdin {} } output { stdout {} }'
-e 允许从命令指定配置

logstash 通常有一个或多个input filter output

logstash通畅搭配filebeat使用，这是一个轻量级的日志收集工具。它从服务器上的文件中收集日志，并将这些日志转发到你的Logstash实例以进行处理。Filebeat设计就是为了可靠性和低延迟。Filebeat在主机上占用的资源很少，而且Beats input插件将对Logstash实例的资源需求降到最低。

指定文件开启
bin/logstash -f mypipeline.conf

在命令行中设置的任何标志都会覆盖logstash.yml中的相应设置。但是文件本身的内容没有改变

Command-Line Flags
--node.name NAME
　　指定Logstash实例的名字。如果没有指定的话，默认是当前主机名。

-f, --path.config CONFIG_PATH
　　从指定的文件或者目录加载Logstash配置。如果给定的是一个目录，则该目录中的所有文件将以字典顺序连接，然后作为一个配置文件进行解析。

-e, --config.string CONFIG_STRING
用给定的字符串作为配置数据，语法和配置文件中是一样的。

--modules
运行的模块名字

-l, --path.logs PATH
Logstash内部日志输出目录

--log.level LEVEL
日志级别

-t, --config.test_and_exit
检查配置语法是否正确并退出

-r, --config.reload.automatic
监视配置文件的改变，并且当配置文件被修改以后自动重新加载配置文件。

-config.reload.interval RELOAD_INTERVAL
为了检查配置文件是否改变，而拉去配置文件的频率。默认3秒。

--http.host HTTP_HOST
Web API绑定的主机。REST端点绑定的地址。默认是"127.0.0.1"

--http.port HTTP_PORT
Web API http端口。REST端点绑定的端口。默认是9600-9700之间。

--log.format FORMAT
指定Logstash写它自身的使用JSON格式还是文本格式。默认是"plain"。

--path.settings SETTINGS_DIR
设置包含logstash.yml配置文件的目录，比如log4j日志配置。也可以设置LS_SETTINGS_DIR环境变量。默认的配置目录是在Logstash home目录下。

-h, --help

应用

搭配filebeat使用

配置filebeat

filebeat.inputs:    #指定filebeat输入路径
- type: log            #指定类型
  paths:                  #指定路径
    - /var/xxx/xxxxx.log

output.logstash:  #指定输出到logstash，有许多的输出方式
  hosts: ["localhost:5044"]   #

配置logstash

input {
  beats { #输入方式为beats客户端
    port => 5044   #端口为5044
  }
}

output {  #可以指定多个输出
  elasticsearch {
    hosts => ["192.168.1.222:9200"]
    index => "appache-log"
  }
  stdout { codec => rubydebug }
}

检查配置
bin/logstash -f first-pipeline.conf --config.test_and_exit
--config.test_and_exit选项的意思是解析配置文件并报告任何错误

bin/logstash -f first-pipeline.conf --config.reload.automatic
--config.reload.automatic选项的意思是启用自动配置加载，以至于每次你修改完配置文件以后无需停止然后重启Logstash

启动filebeat
./filebeat -e -c filebeat.yml -d "publish"

通过tcp端口接受rsyslog日志

Logstash 配置文件
input {
 syslog {
    type => "rsyslog"
    port => "514"
  }
}
output{
  stdout{
    codec => rubydebug
  }
  elasticsearch{
    hosts => ["192.168.1.222:9200"]
    index => "syslog"   # 在配置文件中指定好索引的话可自动的在es中生成索引
}
}


关闭logstash所在设备的rsyslog，释放514端口
[root@node1 config]# systemctl stop rsyslog

启动logstash
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/sys.conf

查看端口情况
[root@elk-master-2 ~]# netstat -anp|grep 514
tcp6       0      0 :::514                  :::*                    LISTEN      12168/java          
udp        0      0 0.0.0.0:514             0.0.0.0:*                           12168/java          
unix  3      [ ]         STREAM     CONNECTED     107269   6514/filebeat

514已经被logstash使用


在另一台设备修改rsyslog配置文件
vi /etc/rsyslog.conf
最后一行添加
*.* @@node1:514

重启syslog，在logstash端接受到数据
logstatsh命令行汇总显示
{
         "timestamp" => "Feb 24 16:16:05",
              "type" => "rsyslog",
              "host" => "192.168.1.221",
           "program" => "dbus-daemon",
    "facility_label" => "system",
    "severity_label" => "Informational",
          "severity" => 6,
        "@timestamp" => 2020-02-24T21:16:05.000Z,
          "facility" => 3,
          "@version" => "1",
           "message" => "'list' object has no attribute 'split'
",
         "logsource" => "red-hat-enterprise-linux",
          "priority" => 30
}