Logstash的简介、安装、配置、Pipeline、插件

一. 简介

官方文档：https://www.elastic.co/guide/en/logstash/current/getting-started-with-logstash.html

• Logstash是一个开源数据收集引擎，具有实时管道功能。
• Logstash可以动态地将来自不同数据源的数据统一起来，并将数据标准化到你所选择的目的地
• Logstash 是一款强大的数据处理工具，它可以实现数据传输，格式处理，格式化输出，还有强大的插件功能，常用于日志处理。
• Logstash耗资源较大，运行占用CPU和内存高。另外没有消息队列缓存，存在数据丢失隐患
• Logstash使用Ruby语言编写的运行在Java虚拟机上的具有收集、分析和转发数据流功能的工具
• Logstash使用Pipeline方式进行日志的搜集，处理和输出

Event：logstash将数据流中的每一条数据在input处被转换为event，在output处event再被转换为目标格式的数据

• Inputs：用于从数据源获取Event。每个Input启动一个线程，从对应数据源获取数据，将数据写入一个队列
• Filters：用于过滤、修改Event
• Outputs：负责输出Event到其他系统中

Logstash使用Pipeline流水线的形式来处理数据Event事件，大致流程如下

其中inputs和outputs支持codecs（coder&decoder）在1.3.0 版之前，logstash 只支持纯文本形式输入，然后用filter处理它。但现在，我们可以在输入期间处理不同类型的数据。所以现在的数据处理流程

箭头代表数据流向。可以有多个input。中间的queue负责将数据分发到不通的pipline中，每个pipline由batcher，filter和output构成。batcher的作用是批量从queue中取数据（可配置）。

logstash数据流历程

1. 首先有一个输入数据，例如是一个web.log文件，其中每一行都是一条数据。file imput会从文件中取出数据，然后通过json codec将数据转换成logstash event。
2. 这条event会通过queue流入某一条pipline处理线程中，首先会存放在batcher中。当batcher达到处理数据的条件（如一定时间或event一定规模）后，batcher会把数据发送到filter中，filter对event数据进行处理后转到output，output就把数据输出到指定的输出位置。
3. 输出后还会返回ACK给queue，包含已经处理的event，queue会将已处理的event进行标记。

queue分类

• In Memory： 在内存中，固定大小，无法处理进程crash. 机器宕机等情况，会导致数据丢失。
• Persistent Queue：可处理进程crash情况，保证数据不丢失。保证数据至少消费一次；充当缓冲区，可代替kafka等消息队列作用。
• Dead Letter Queues：存放logstash因数据类型错误等原因无法处理的Event

Persistent Queue（PQ）处理流程

1. 一条数据经由input进入PQ，PQ将数据备份在disk，然后PQ响应input表示已收到数据；
2. 数据从PQ到达filter/output，其处理到事件后返回ACK到PQ；
3. PQ收到ACK后删除磁盘的备份数据；

二. 安装

1. 安装Java环境

在一些Linux环境下，必须设置JAVA_HOME环境变量，否则Logstash在安装期间没有检测到JAVA_HOME环境变量，会报错并且启动不起来服务。如果JDK目录在/opt下，则 在/usr/bin/下建立软连接指向JAVA_HOME/bin路径下的java

2. 安装Logstash

YUM/RPM

[elasticsearch-7.x]
name=Elasticsearch repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

yum install -y logstash-7.2.0

手动下载RPM安装，官方下载链接：https://www.elastic.co/downloads/logstash

yum localinstall -y logstash-7*.rpm

RPM包安装后各个配置文件的位置

二进制包

二进制包中各个配置文件的位置

3. 启动

以服务形式或命令启动Logstash

systemctl start logstash 
#后台会起一个名叫org.jruby.Main的Java后台进程，用jps -l查看
jps -l

使用二进制执行文件启动

/user/share/logstash/bin/logstash -f logstash.conf --config.reload.automatic

#-f 指定配置文件路径
#--config.reload.automatic 自动检测加载配置文件，该参数在有-e参数是不生效  
#--config.reload.interval <seconds> 设置多少秒检测一次配置文件  如果Logstash启动时没有配置自动加载配置文件，重启进程时加上。

4. 验证

/usr/share/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'
#参数-e：直接从命令行定义配置信息
#配置从标准输入读取输入，然后输出到标准输出

stdin > hello world
stdout> 2013-11-21T01:22:14.405+0000 0.0.0.0 hello world

#Logstash会在消息上添加时间戳和IP地址
#Ctrl+D 退出Logstash

5. 命令行参数

三. Docker镜像

docker pull docker.elastic.co/logstash/logstash:7.4.0

docker pull logstash:7.4.0

镜像中各个配置文件的位置

Note：基于该镜像启动的容器，日志是直接输出到控制台的，无法直接输出到日志文件中

• docker镜像是基于.tar.gz格式的二进制包创建的

• 将pipeline文件挂载到/usr/share/logstash/pipeline/下启动

  docker run --rm -it \
  -v ./test.conf:/usr/share/logstash/pipeline/test.conf \
  docker.elastic.co/logstash/logstash:7.4.0
  

• 默认pipeline文件：/usr/share/logstash/pipeline/logstash.conf

  input {
    beats {
      port => 5044
    }
  }
  output {
    stdout {
      codec => rubydebug
    }
  }
  

  也就是说如果不配置挂载pipeline文件就直接启动容器，logstash将启动一个最小化的pipeline：Beat Input ---> Stdout Output

• 可通过设置环境变量的形式配置logstash。

  docker run --rm -it -e PIPELINE_WORKERS:2 docker.elastic.co/logstash/logstash:7.4.0。例如以下环境变量对应的logstash配置

• logstash docker 镜像中的默认配置

四. 配置

Logstash配置文件中配置项的格式是基于YAML语法，例如：

pipeline:
  batch:
    size: 125
    delay: 50

也可以使用平级格式

pipeline.batch.size: 125
pipeline.batch.delay: 50

配置项的值可以引用系统级别的环境变量

pipeline.batch.size: ${BATCH_SIZE}
pipeline.batch.delay: ${BATCH_DELAY:50}
node.name: "node_${LS_NODE_NAME}"
path.queue: "/tmp/${QUEUE_DIR:queue}"

如果设置多个自定义的配置项时，推荐使用以下格式

modules:
  - name: MODULE_NAME1
    var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY1: VALUE
    var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY2: VALUE
    var.PLUGIN_TYPE2.PLUGIN_NAME2.KEY1: VALUE
    var.PLUGIN_TYPE3.PLUGIN_NAME3.KEY1: VALUE
  - name: MODULE_NAME2
    var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY1: VALUE
    var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY2: VALUE

常见的logstash配置

五. Pipeline

1. 配置项结构

Logstash Pipeline文件的配置项分为三个部分：

input{
  input插件{
      插件配置项
  }
}
filter{
  filter插件{
      插件配置项
  }
}
output{
  output插件{
      插件配置项
  }
}

Note:

1. 如果在filter中添加了多种处理规则，则按照它的顺序一一处理，但是有一些插件并不是线程安全的。

2. 如果在filter中指定了两个一样的的插件，这两个任务并不能保证准确的按顺序执行，因此官方也推荐避免在filter中重复使用插件。

2. 插件的条件控制

官方文档：https://www.elastic.co/guide/en/logstash/6.7/event-dependent-configuration.html#conditionals

有时需要在特定条件下过滤或输出事件。为此，您可以使用条件（conditional）来决定filter和output处理特定的事件。比如在elk系统中想要添加一个type类型的关键字来根据不同的条件赋值，最后好做统计。条件语支持if，else if和else语句并且可以嵌套。

条件语法

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

操作符

• 比较操作：

  • 相等: ==, !=, <, >, <=, >=
  • 正则: `=~(匹配正则), !~(不匹配正则)
  • 包含:in(包含), not in(不包含)

• 布尔操作：

  • and(与), or(或), nand(非与), xor(非或)

• 一元运算符：

  • !(取反)
  • ()(复合表达式), !()(对复合表达式结果取反)

示例

filter {
  if [foo] in [foobar] {
    mutate { add_tag => "field in field" }
  }
  if [foo] in "foo" {
    mutate { add_tag => "field in string" }
  }
  if "hello" in [greeting] {
    mutate { add_tag => "string in field" }
  }
  if [foo] in ["hello", "world", "foo"] {
    mutate { add_tag => "field in list" }
  }
  if [missing] in [alsomissing] {
    mutate { add_tag => "shouldnotexist" }
  }
  if !("foo" in ["hello", "world"]) {
    mutate { add_tag => "shouldexist" }
  }
  if [message] =~ /\w+\s+\/\w+(\/learner\/course\/)/ {
    mutate {
      add_field => { "learner_type" => "course" }
    }
  }

  mutate { add_field => { "show" => "This data will be in the output" } }
  mutate { add_field => { "[@metadata][test]" => "Hello" } }
  mutate { add_field => { "[@metadata][no_show]" => "This data will not be in the output" } }
}

output {
  if "_grokparsefailure" not in [tags] {
    elasticsearch { ... }
  }
  if [@metadata][test] == "Hello" {
    stdout { codec => rubydebug }
  }
  if [loglevel] == "ERROR" and [deployment] == "production" {
    pagerduty {
    ...
    }
  }
}

注意：

如果if[foo] in "String"在执行这样的语句时无法把该字段值转化成String类型。所以最好要加field if exist判断

if ["foo"] {
  mutate {
    add_field => "bar" => "%{foo}"
  }
}

3. 引用event中的字段

• 直接引用字段，使用[]，嵌套字段使用多层[][]即可

  {
      "a": "1",
      "b": "2",
      "c": {
          "c1": "3"
      }
  }
  ----------Pipeline中引用Event中的字段--------------
  if [b] =~ "2" {
      ..........
  }
  if [c][c1] == "3" {
      ...........
  }
  

• 在字符串中以sprintf方式引用,使用%{}

  {
      "a": "1",
      "b": "2",
      "c": {
          "c1": "3"
      }
  }
  ----------Pipeline中引用Event中的字段--------------
  add_field => {
    "test" => "test: %{b}"
  }
  add_field => {
    "test" => "test: %{[c][c1]}"
  }
  

六. Input插件

插件一览表

插件通用配置项

七. Filter插件

插件一览表

插件通用配置项

八. Output插件

插件一览表

插件通用配置项

九. Codec插件

插件一览表

十. 插件管理

Logstash 插件是使用 Ruby开发的，Logstash 从很早的1.5.0+版开始，其插件模块和核心模块便分开维护，其插件使用的是 RubyGems包管理器来管理维护。所以 Logstash插件本质上就是自包含的RubyGems。

RubyGems（简称 gems）是一个用于对 Ruby组件进行打包的 Ruby 打包系统。 它提供一个分发 Ruby 程序和库的标准格式，还提供一个管理程序包安装的工具。

插件的名字格式：logstash-{input/output/filter}-插件名 示例：filter中的date插件：logstash-filter-date

1. 安装插件

#以安装dissect插件为例
/usr/share/logstash/bin/logstash-plugin install 插件名
#参数详解：
--path.plugins  指定安装路径

2. 查看已安装的插件

/usr/share/logstash/bin/logstash-plugin list
#参数详解：
--verbose  查看插件的版本
--verbose  查看组（input, filter, codec, output）下面的所有插件。例如查看filter下的所有插件

3. 更新插件

#更新某个插件
/usr/share/logstash/bin/logstash-plugin update 插件名
#更新全部插件
/usr/share/logstash/bin/logstash-plugin update

4. 卸载插件

/usr/share/logstash/bin/logstash-plugin remove  插件名

5. 给插件管理器设置代理

export HTTP_PROXY=http://127.0.0.1:3128

6. 修改插件仓库地址

Logstash插件默认仓库地址是：http://rubygems.org

有一些开源的插件仓库：

• Geminabox：https://github.com/geminabox/geminabox
• Gemirro：https://github.com/PierreRambaud/gemirro
• Gemfury：https://gemfury.com/
• Artifactory：http://www.jfrog.com/open-source/

编辑/usr/share/logstash/Gemfile，将source "https://rubygems.org"改为source "https://my.private.repository"

十一. 其他操作

1、output-elasticsearch的template模板

PUT _template/logstash
{
    "order" : 2,
    "version" : 60001,
    "index_patterns" : [
      "*"
    ],
    "settings" : {
      "index" : {
        "number_of_replicas" : "1",
        "number_of_shards" : "2",
        "refresh_interval" : "60s"
      }
    },
    "mappings" : {
      "dynamic_templates" : [
        {
          "message_field" : {
            "path_match" : "message",
            "mapping" : {
              "norms" : false,
              "type" : "text"
            },
            "match_mapping_type" : "string"
          }
        },
        {
          "string_fields" : {
            "mapping" : {
              "norms" : false,
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "match_mapping_type" : "string",
            "match" : "*"
          }
        }
      ],
      "properties" : {
        "@timestamp" : {
          "type" : "date"
        },
        "geoip" : {
          "dynamic" : true,
          "properties" : {
            "ip" : {
              "type" : "ip"
            },
            "latitude" : {
              "type" : "half_float"
            },
            "location" : {
              "type" : "geo_point"
            },
            "longitude" : {
              "type" : "half_float"
            }
          }
        },
        "@version" : {
          "type" : "keyword"
        }
      }
    },
    "aliases" : { }
}

十二. 监控信息

1、查看pipeline运行监控信息

curl -XGET 'http://logstash实例地址:9600/_node/stats/pipelines/pipeline实例名?pretty'