Logstash部署

Edited on In ,

介绍

Logstash 是一个自由和开放的服务器端数据处理管道吸入来自多种数据源的数据,转换,然后将其发送给你最喜欢的“储备”。

工作原理就是一种 输入->过滤->输出 的模式。

安装

下载

下载最新版本的Logstash压缩包

1
2
3
4
cd ~/
wget https://artifacts.elastic.co/downloads/logstash/logstash-8.4.1-linux-x86_64.tar.gz
tar -zxf logstash-8.4.1-linux-x86_64.tar.gz
mv logstash-8.4.1 logstash

启动

1
/opt/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'

输出 Successfully started 表示启动成功。

命令里的-e表示后面跟着的是一个配置信息字符串，字符串的功能就是让 Logstash
接受控制台里的输入，并输出到控制台，比如在控制台输入 hello world ：

如果用最简化的方式执行：

1
bin/logstash -e ""

输出

1
2
3
4
5
6
7
hello word!
{
          "host" => "MAXZHAO-PC",
       "message" => "hello word!\r",
      "@version" => "1",
    "@timestamp" => 2021-07-02T05:35:18.533Z
}

访问 端口默认为 9600

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
{
"host": "MAXZHAO-PC",
"version": "7.13.2",
"http_address": "127.0.0.1:9600",
"id": "60d59feb-80eb-466a-97e0-d46d3fa75587",
"name": "MAXZHAO-PC",
"ephemeral_id": "758e9967-2f6f-49be-83b3-8e261e09446f",
"status": "green",
"snapshot": false,
"pipeline": {
"workers": 12,
"batch_size": 125,
"batch_delay": 50
},
"build_date": "2021-06-10T19:51:49Z",
"build_sha": "6d32f7df79a7d10d821b4cbff51c47f46d8c67b1",
"build_snapshot": false
}

配置

配置文件

实际运维中配置信息通常会放配置文件中来管理，这种情况下 Logstash 提供了 -f 参数来指定配置信息所在的配置文件。在 Logstash 的根目录下执行命令：

1
~/logstash/bin/logstash -f ~/logstash/config/logstash.conf

参考 config/logstash-sample.conf 创建config/logstash.conf , 内容:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
input {
  stdin { }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200", "http://127.0.0.1:9200"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }
  stdout { }
  file {
	path => "D:\develop\logstash-7.13.2\data-demo\%{type}.%{+yyyy.MM.dd.HH}.txt"
  }
}

在启动的控制台输入 hello，则控制台会输出 hello，并且文件也会写入 hello。

使用

logstash 在输入、过滤、输出都是由很多插件配合。

官方的插件文档

插件配置

value types

Array

: list = > true 

1
users => [ {id => 1, name => bob}, {id => 2, name => jane} ]
List

不是类型本身，而是属性类型可以具有。这使得输入检查多个值成为可能。插件作者可以通过在声明参数时指定: list = > true 来启用列表检查。

1
2
path => [ "/var/log/messages", "/var/log/*.log" ]
uris => [ "http://elastic.co", "http://example.net" ]

此示例将 path 配置为一个包含三个字符串中每个字符串的元素的列表。它还将 URIs 参数配置为 uri 列表，如果提供的任何 uri 无效，则失败。

Boolean
1
ssl_enable => true
Bytes
1
2
3
4
my_bytes => "1113"   # 1113 bytes
my_bytes => "10MiB"  # 10485760 bytes
my_bytes => "100kib" # 102400 bytes
my_bytes => "180 mb" # 180000000 bytes
Codec

编码解码器是用来表示数据的 Logstash 编码解码器的名称。编解码器可以用于输入和输出。

输入解码器提供了一种在数据进入输入之前对其进行解码的方便方法。输出解码器提供了一种在数据离开输出之前对其进行编码的方便方法。使用输入或输出编码解码器可以避免在 Logstash 管道中使用单独的过滤器。

1
codec => "json"

Hash

多个键值项之间用空格而不是逗号分隔。

1
2
3
4
5
6
7
match => {
  "field1" => "value1"
  "field2" => "value2"
  ...
}
# or as a single line. No commas between entries:
match => { "field1" => "value1" "field2" => "value2" }

Number

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

1
port => 33

Password

密码是一个字符串，其中包含一个未记录或打印的值。

1
my_password => "password"

URI

一个 URI 可以是任何东西，从一个完整的 URL，如 http://elastic.co/地址，到一个简单的标识符，如 foobar。如果 URI 包含一个密码，比如 http://user:pass@example.net ，那么 URI 的密码部分将不会被记录或打印。

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

路径是表示有效操作系统路径的字符串。

1
my_path => "/tmp/logstash"
String

字符串必须是单个字符序列。注意，字符串值用引号括起来，可以是双引号，也可以是单引号。

Escape Sequences

转移符号

Text文字 Result结果
\r carriage return (ASCII 13)回车符
\n new line (ASCII 10)
\t tab (ASCII 9)
\\ backslash (ASCII 92)反斜杠
\" double quote (ASCII 34)
\' single quote (ASCII 39) 
1
2
name => "Hello world"
name => 'It\'s a beautiful day'

插件安装

插件安装

安装插件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 查看
/opt/logstash/bin/logstash-plugin list
# 在线安装
/opt/logstash/bin/logstash-plugin install logstash-filter-grok
vim /opt/logstash/Gemfile 
# 更新
/opt/logstash/bin/logstash-plugin update logstash-filter-grok
mkdir /opt/logstash/plugins
cd  /opt/logstash/plugins
wget https://github.com/logstash-plugins/logstash-filter-grok/archive/refs/tags/v4.4.2.tar.gz
tar -zxvf v4.4.2.tar.gz
vim /opt/logstash/Gemfile
# 写入
# gem "logstash-filter-grok", :path => "path"

input-file 插件

看下 input 的一个插件例子。上面讲的是从控制台输入，如果要从文件读取输入，则可以用 file 插件：

1
input { file { path => "/tmp/abc.log" }}

上面是这个插件的最小化配置， path 后面跟随的是文件的全路径名，如果要匹配一个目录下的所有文件，可以用通配符*：

1
input { file { path => "/tmp/data/*" } }

要匹配多个不同目录下的文件，则用中括号括起来，并逗号分隔不同文件路径：

1
input { file { path => ["/tmp/data/*","/log/abc/hello.log"] }}

注意文件的路径名需要时绝对路径。

output-elasticsearch 插件

如果要将数据输出到，则可以用 Elasticsearch 插件：

1
2
3
4
5
6
7
output {
        elasticsearch {
                action => "index"
                hosts  => "localhost:9200"
                index  => "log-example"
         }
}

上面配置的含义是，将数据输出到 Elasticsearch 服务器，hosts 是提供服务的地址和端口号，action 是指索引到一个文档，index 是指要被写进的索引名，当然还有其他参数配置，具体参见该插件的官方文档说明

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
input {
  stdin { }
}

output {
  stdout { }
  file {
	path => "D:\develop\logstash-7.13.2\data-demo\%{type}.%{+yyyy.MM.dd.HH}.txt"
  }
  elasticsearch {
    #action => "%{[@metadata][action]}"
    #document_id => "%{[@metadata][_id]}"
    hosts => ["http://localhost:9200"]
    index => "logstash"
    #protocol => "http"
    #hosts => ["http://localhost:9200", "http://127.0.0.1:9200"]
    #index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }
}

filter-grok 插件

不论是 input 的 stdin 还是 file 插件，都是从不同来源读取数据，但它们读取的都是一行一行的文本数据，输出的时候会发现它们都是作为一行放到输出的 message 属性中：

img

文本输出

很多情况下需要对这些文本做格式化的解析再输出，比如日志文件中数据格式是这样的：

1
[2017-04-03 23:30:11.020][INFO][logstash] Pipeline main started

上面的文本格式是：[时间][日志级别][模块名]日志信息描述
输出 json 时也希望将时间放到自定义的 TimeStamp 属性中，将日志级别信息放到
LogLevel 属性中，将模块名放到 ModuleName 属性中，将日志信息描述放到
LogDesc 属性中。

如果是这种需求就用到了 grok 插件，该插件是 Logstash 将普通文本解析成结构化数据的最好的方式。（这可不是我非要安利，是官网上说的：Grok is currently the best way in logstash to parse crappy unstructured log data into something structured and queryable.）

  1. grok 表达式
    grok 表达式语法是这样的：
1
%{SYNTAX:SEMANTIC}

SYNTAX 是要匹配的文本的模式名。SEMANTIC 是匹配出的内容的标识符，该标识符即输出到 output 中的属性名。看一个简单的 grok 表达式例子：

1
%{IP:source_ip}

它表示使用名为 IP 的正则表达式提取内容，并将匹配出的内容命名为 source_ip 。所以如果输入是 IP 地址的格式：11.22.33.123，则会将该内容放到 source_ip 中。

  1. grok 中预定义的模式
    上面的 IP 是 grok 里预定义的模式，其正则表达式的定义实际是：

    img

    IP 模式定义

    grok 已经预定义了很多模式，比如匹配一串数字可以用 NUMBER、匹配 MAC 地址用 MAC、匹配时间用 TIMESTAMP_ISO8601、匹配 LOG 日志级别用 LOGLEVEL

    等等。详细定义参见

    github 中的说明

  2. grok 中的自定义模式
    如果上面列的这些模式不能满足需求，grok 也允许自定义模式。
    先看下自定义模式的语法：

1
pattern_name regexp

pattern_name 就是自定义的模式名，regexp 就是模式的正则表达式定义，之间用一个空格分隔开。比如自定义一个队列 ID 模式，它的格式是10或11个数字、大写字母组成的字符串，其定义语法如下：

1
QUEUE_ID [0-9A-F]{10,11}

熟悉正则表达是的应该知道[0-9A-F]表示数字、大写字母组成的字符，后面的{10,11}表示有10或11个这样的字符。
接着在 Logstash 根目录下建一个 patterns 目录，该目录下新建一个文件叫
myPattern 。文件内容就是：

1
QUEUE_ID [0-9A-F]{10,11}

最后在 filter 配置中加上 patterns 目录，并引用上面定义的这个 QUEUE_ID ：

1
2
3
4
5
6
filter {
    grok {
      patterns_dir => ["./patterns"]
      match => { "message" => "%{TIMESTAMP_ISO8601:TimeStamp} %{QUEUE_ID:queue_id}: %{GREEDYDATA:syslog_message}" }
    }
}

patterns_dir 用于指定自定义模式的文件所在的目录，match 的内容就是 grok 表达式，TIMESTAMP_ISO8601 和 GREEDYDATA 预定义的模式，QUEUE_ID 是自定义的模式。它将匹配以下格式的内容：

1
2017-04-03 23:30:11.020 12345abcde hello world

匹配后的结果是：

1
{"@timestamp": "2017-04-12T11:57:40.186Z",  "queue_id": "12345abcde",  "@version": "1",  "syslog_message": "hello world",  "message": "2017-04-03 23:30:11.020 12345abcde hello world",  "TimeStamp": "2017-04-03 23:30:11.020"}
  1. 日志文件的 filter 配置
    所以上面的日志文件中数据的 filter 区段配置：
1
2
3
4
5
6
7
filter {
        grok {
          match => {
                "message" => "\[%{TIMESTAMP_ISO8601:TimeStamp}\]\[%{LOGLEVEL:LogLevel}\]\[%{WORD:ModuleName}\] %{GREEDYDATA:LogDesc}"
        }
    }
}

由于中括号是特殊字符，所以正则表达式中它的前面要加上反斜杠\。
加载文件配置后运行结果如下：

img

本文地址： https://github.com/maxzhao-it/blog/post/9074/