一 Logstash介绍

Logstash是具有实时流水线能力的开源的数据收集引擎。Logstash可以动态统一不同来源的数据，并将数据标准化到您选择的目标输出。它提供了大量插件，可帮助我们解析，丰富，转换和缓冲任何类型的数据。
logstash常用于日志系统中做日志采集设备，最常用于ELK中作为日志收集器使用

二 Logstash作用

集中、转换和存储你的数据，是一个开源的服务器端数据处理管道，可以同时从多个数据源获取数据，并对其进行转换，然后将其发送到你最喜欢的存储

三 Logstash工作原理

管道（Logstash Pipeline）是Logstash中独立的运行单元，每个管道都包含两个必须的元素输入（input）和输出（output），和一个可选的元素过滤器（filter），事件处理管道负责协调它们的执行。输入和输出支持编解码器，使您可以在数据进入或退出管道时对其进行编码或解码，而不必使用单独的过滤器。
Logstash工作原理(三个阶段: inputs→filters→outputs )
- Input ( 数据输入阶段) :会把数据输入数据到logstash
- Filters (数据清洗) : 数据中间处理,对数据进行操作
- Outputs (数据输出) : outputs是logstash处理管道的最末端组件
Logstash-Input阶段常见的输入
- file :从文件系统的文件中读取,类似于tail -f命令
- syslog :在514端口上监听系统日志消息,并根据RFC3164标准进行解析
- beats :从Filebeat中读取
Logstash-Filter数据中间件处理插件grok (可解析任意文本数据)
- GROK基础语法如: %{SYNTAX:SEMANTIC}
- SYNTAX: 代表匹配值的类型
- SEMANTIC: 代表存储该值的一一个变量名称
- 例: %{ERROR|DEBUG |INFO |WARN:1og_level}
Logstash-Output数据输出
- 输出到kafka和ES也可以输出到redis

编解码器
- Codecs是基本的流过滤器，可以作为输入或输出的一部分进行操作，Codecs使你能够轻松地将消息的传输与序列化过程分开，流行的codecs包括json、msgpack和plain（text）。
- json：以JSON格式对数据进行编码或解码。
- multiline：将多行文本事件（如java异常和stacktrace消息）合并到单个事件中。
执行模型
- Logstash事件处理管道协调输入、过滤器和输出的执行。
- Logstash管道中的每个输入阶段都在自己的线程中运行，输入将事件写入位于内存（默认）或磁盘上的中央队列，每个管道工作线程从这个队列中取出一批事件，通过配置的过滤器运行事件批处理，然后通过任何输出运行过滤的事件，可以配置批处理的大小和管道工作线程的数量。

四 Logstash安装

1.拉取镜像

docker pull logstash

2.运行命令

docker run -d --name logstash -p 5044:5044 -p 5000:5000 -p 9600:9600 logstash

3.查看是否运行

docker ps

五 Logstash使用

为了方便同时使用ELK，新建一个目录docker-elk
在docker-elk目录下新建logstash/logstash.yml
在logstash.yml，输入以下代码：

---
http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://elasticsearch:9200" ]

xpack.monitoring.enabled: true
xpack.monitoring.elasticsearch.username: elastic
xpack.monitoring.elasticsearch.password: pwd

在docker-elk目录下新建logstash/logstash.conf
在logstash.conf，输入以下代码：

input {
    beats {
        port => 5044
    }
    tcp {
        port => 5000
    }
}

output {
    elasticsearch {
        hosts => "elasticsearch:9200"
        user => "elastic"
        password => "pwd"
        index => "%{[@metadata][-imooc]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    }
}

在docker-elk目录下建立一个docker-stack.yml，同时启动ELK
输入以下代码：

version: '3.3'
services:
  logstash:
    image: cap1573/logstash:7.9.3
    ports:
      - "5044:5044"
      - "5000:5000"
      - "9600:9600"
    volumes:
      - ./logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml
      - ./logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    environment:
      LS_JAVA_OPTS: "-Xmx256m -Xms256m"