logstash基本操作及常用日志收集插件

运行一个最基本的logstash管道

Logstash管道有两个必需的元素，input和output，以及一个可选的元素filter。输入插件使用来自源的数据，过滤器插件在您指定时修改数据，输出插件将数据写入目标。

在logstash中使用-e 参数可以在命令行中指定配置

logstash -e 'input {stdin {} } output { stdout { codec => rubydebug } }'

stdin和stdout是logstash内置的插件，他们可以在终端上显示输入输出的结果而方便我们测试

当然也可以将上述配置写在一个配置文件里

vim test.conf
input {
        stdin { }
}
output {
        stdout {
         codec => rubydebug
        }
}

然后执行命令logstash -f test.conf，使用-f选项用来指定配置文件，效果是与在命令行中使用-e选项是一样的，当您使用-e或-f时，Logstash会忽略该pipelines.yml文件并记录有关它的警告。如果不加任何参数的话，那么logstash会读取pipelines.yml文件里指定的目录，pipelines.yml默认存在于/etc/logsatsh/pipelines.yml目录

logstash数值类型

• 数组

match =>["datetime", "UNIX", "ISO8601"]

• 布尔

   必须是一个true或false

ssl_enable => true

• 字节

   一个字段是字节字符串字段表示有效字节的单元。它是一种方便的方式在特定尺寸的插件选项。 支持SI (k M G T P E Z Y)和Binary (TiKimigipiziyiei)单位。二进制单元在基座单元和Si-1024在基底1000。这个字段是大小写敏感的。如果未指定单位,则整数表示的字符串的字节数。

my_bytes => "1113"   # 1113 bytes   my_bytes => "10MiB"  # 10485760 bytes   my_bytes => "100kib" # 102400bytes   my_bytes => "180 mb"# 180000000 bytes

• 编解码器

codec => "json"

• 哈希

   哈希是一个键值对的集合中指定的格式，多个键值对的条目以空格分隔而不是逗号。

match => {  "field1" => "value1"   "field2" =>"value2"   ... }

• 数字

   数字必须有效的数字值(浮点或整数)。

   port => 33

• 密码

   密码是一个字符串的单个值，则不对其进行记录或打印。

my_password => "password"

   URI

my_uri =>"http://foo:bar@example.net"

• 路径

   一个路径是一个字符串，表示系统运行的有效路径。

my_path =>"/tmp/logstash"

• 转义序列

   默认地，转义字符没有被启用。如果你希望使用转义字符串序列，您需要在你的logstash.yml中设置config.support_escapes: true

条件判断

有时您只想在特定条件下过滤或输出事件。为此，您可以使用条件。

Logstash中的条件查看和行为与编程语言中的条件相同。条件语句支持if，else if以及else报表和可以被嵌套。

条件语法

if EXPRESSION{   ... } else if EXPRESSION {   ... } else {   ... }

条件表达式支持的比较运算符

比较：==, !=, <, >, <=, >=

正则：=~, !~(是否使用正则匹配)

包含：in,not in (是否包含)

支持的布尔运算符

and，or，nand，xor

支持的一元运算符

！

全局模式支持

只要允许glob模式，Logstash就支持以下模式：

匹配任何文件。您还可以使用a *来限制glob中的其他值。例如，*conf匹配所有结尾的文件conf。*apache*匹配apache名称中的任何文件。

递归匹配目录。

匹配任何一个角色。

[set]

匹配集合中的任何一个字符。例如，[a-z]。还支持排除集合中的任意字符（[^a-z]）。

{p,q}

匹配文字p或文字q。匹配的文字可以是多个字符，您可以指定两个以上的文字。此模式相当于在正则表达式（foo|bar）中使用垂直条的交替。

转义字符。

正则匹配插件grok

描述

grok可以将非结构化日志数据解析为结构化和可查询的内容。此工具非常适用于syslog日志，apache和其他Web服务器日志，mysql日志，以及通常为人类而非计算机使用而编写的任何日志格式。经过grok过滤之后日志会被分成多个字段

Grok的工作原理是将文本模式组合成与日志匹配的内容

grok模式的语法是 %{PATTERN_NAME:capture_name:data_type}

data_type 目前只支持两个值：int 和 float。

PATTERN:NAME为之前定义的文本匹配的模式，而capture_name则是匹配文本的提供的标识符

举个官方文档的例子

如果日志如以下格式，

55.3.244.1 GET/index.html 15824 0.043

那么可以设置grok匹配以下模式   grok {     match => { "message" =>"%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes}%{NUMBER:duration}" }   }

在grok过滤之后，该事件将添加一些额外的字段：

• client: 55.3.244.1
  
• method: GET
  
• request: /index.html
  
• bytes: 15824
  
• l duration: 0.043

自定义匹配模式

有时logstash没有需要的模式。为此，可以有几个选择。

首先，您可以使用Oniguruma语法进行命名捕获，它可以匹配一段文本并将其保存为字段：

（？<field_name>此处的模式）

例如，后缀日志具有queue id10或11个字符的十六进制值。我可以像这样轻松捕获：

（？<queue_id>[0-9A-F] {10,11}）

或者，也可以创建自定义模式文件。

vim patterns
POSTFIX_QUEUEID[0-9A-F] {10,11}

可以设置如下过滤器规则

filter {   grok {    patterns_dir => ["./patterns"]     match => { "message" =>  %{POSTFIX_QUEUEID:queue_id}" }   } }

这样经过过滤之后会多一个标识符为queue_id的字段

• queue_id: BEF25A72965

   一般的正则表达式只能匹配单行文本，如果一个Event的内容为多行，可以在pattern前加“(?m)”

grok过滤器配置选项

match

match的值类型是哈希，默认值为空

示例

grok { match => {"message" => "Duration: %{NUMBER:duration}" } }

overwrite

overwrite的值类型是阵列，默认值为空

如果你把"message" 里所有的信息通过 grok匹配成不同的字段，数据实质上就相当于是重复存储了两份。所以你可以用 remove_field 参数来删除掉 message 字段，或者用 overwrite 参数来重写默认的 message 字段，只保留最重要的部分。

示例

filter {
        grok {
          match => {"message"=> "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes}%{NUMBER:duration}"}
          overwrite => [ "message"]
}
}

数据修改插件mutate

描述

mutate过滤器允许您对字段执行常规突变。您可以重命名，删除，替换和修改事件中的字段。它提供了丰富的基础类型数据处理能力。包括类型转换，字符串处理和字段处理等。

mutate可用的配置选项

convert

convert值的类型是哈希，convert没有默认值

convert可以将字段的值转换为其他类型，例如将字符串转换为整数。如果字段值是数组，则将转换所有成员。可以转换的类型有Boolean，integer，float，string

以下是转换类型的详细解析

• integer：
  

字符串被解析; 支持逗号分隔符（例如，字符串"1,000"生成一个值为1000的整数）; 当字符串有小数部分时，它们会被截断。

浮点数和小数被截断（例如，3.99变为3，-2.7变为-2）

布尔真和布尔假分别被转换为1和0

• integer_eu：
  

相同integer，除了字符串值支持点分隔符和逗号小数（例如，"1.000"生成一个值为1000的整数）

• float：
  

整数转换为浮点数

字符串被解析; 支持逗号分隔符和点小数（例如，"1,000.5"生成一个值为一千零一半的整数）

布尔真和布尔假被转换为1.0和0.0分别

• float_eu：

相同float，除了字符串值支持点分隔符和逗号小数（例如，"1.000,5"生成一个值为一千零一半的整数）

• string：
  

所有值都使用UTF-8进行字符串化和编码

• boolean：
  

串"true"，"t"，"yes"，"y"，和"1"被转换成布尔true

串"false"，"f"，"no"，"n"，和"0"被转换成布尔false

空字符串转换为布尔值 false

所有其他值直接通过而不进行转换并记录警告消息

convert可以写成数组格式，数组格式可以将多个字段转成多种类型，并且两个为一组，第一个表示字段，第二个为想转换的数据类型，也可以写成哈希格式，字段与类型一一对应。

示例：

mutate {
       convert => ["reqTime","integer","statusCode","integer","bytes","integer"]
       convert => {"port"=>"integer"}
}

copy

将现有字段覆盖到另一个字段，并覆盖现有的目标字段，copy的值类型是哈希

示例：

mutate {
copy => {“source_field”=>“dest_field”}
}

gsub

用于字符串的替换，替换的值可以用正则表达式和字符串，gsub配置的值类型为数组，三个为一组，分别表示：字段名称，待匹配的字符串（或正则表达式），待替换的字符串。

在使用正则表达式的时候要注意需要转义的字符用反斜杠转义。

mutate {
    gsub => [
      #将斜杠替换成_
      "fieldname", "/","_",
      # 将“\?#-”替换成“-”
      "fieldname2","[\\?#-]", "."
    ]
  }

split

将字符串以分隔符分割成数组，只能用于字符串字段，值类型为哈希

示例

mutate {
         split =>{"message"=>"," }
}

join

使用分隔符将数组连接成字符串，join的值类型是哈希

示例：

mutate {
    join => { "fieldname" => "," }
  }

lowercase和uppercase

将字符串转换成小写或大写，值类型默认是数组

mutate {
    lowercase => [ "fieldname" ]
   uppercase => [ "fieldname" ]
  }

merge

合并两个数组或者哈希的字段，若是字符串格式自动转化成数组，值类型是哈希，合并多个字段可以用数组格式

示例

mutate {
          merge => {"message"=> "host"}
        }
或
 mutate {
          merge =>["message","host","@version","@timestamp"]
        }

rename

重命名某个字段，值类型为哈希

示例

mutate {
         rename => {"host" =>"hostname"}
  #将host字段重命名为hostname
        }

update

更新一个已存在字段的内容，如果字段不存在则不会新建，值类型为哈希

示例

mutate {
    update => { "message" =>"asd" }
  }

replace

替换一个字段的内容，如果字段不存在会新建一个新的字段，值类型为哈希

示例

mutate {
          replace => {"type" =>"mutate"}
  #添加一个新的字段type
}

coerce

为一个值为空的字段添加默认值，可以配合add_field,值类型为哈希

示例

mutate {
         coerce =>{"field"=>"a123"}
         add_field =>{"field"=>"asd"}
}

strip

去掉字段首尾的空格，值类型是数组

示例

mutate {
     strip=>["field1","field2"]
}

mutate插件的执行顺序

rename(event) if @rename
    update(event) if @update
    replace(event) if @replace
    convert(event) if @convert
    gsub(event) if @gsub
    uppercase(event) if @uppercase
    lowercase(event) if @lowercase
    strip(event) if @strip
    remove(event) if @remove
    split(event) if @split
    join(event) if @join
    merge(event) if @merge

时间处理插件date

描述

date插件用于解析字段中的日期，然后使用该日期或时间戳作为事件的logstash时间戳。

日期过滤器对于排序事件和回填旧数据尤其重要，而在实时数据处理的时候同样有效，因为一般情况下数据流程中我们都会有缓冲区，导致最终的实际处理时间跟事件产生时间略有偏差。如果没有此过滤器，logstash将根据第一次看到事件（在输入时），如果事件中尚未设置时间戳，则选择时间戳。例如，对于文件输入，时间戳设置为每次读取的时间。

locale

值类型是字符串

使用IETF-BCP47或POSIX语言标记指定用于日期解析的语言环境。比如设置为en,en-US等.主要用于解析非数字的月,和天,比如Monday,May等.如果是时间日期都是数字的话,不用关心这个值。