Logstash与FileBeat详解以及ELK整合

ELK架构

ELK架构分为两种，一种是经典的ELK，另外一种是加上消息队列（Redis或Kafka或RabbitMQ）和Nginx结构。
经典的ELK主要是由Filebeat + Logstash + Elasticsearch + Kibana组成，如下图：（早期的ELK只有Logstash + Elasticsearch + Kibana）

此架构主要适用于数据量小的开发环境，存在数据丢失的危险。
整合消息队列+Nginx架构:

Logstash介绍和使用

Logstash 是免费且开放的服务器端数据处理管道，能够从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的存储库中。
Logstash核心概念
Pipeline：

• 包含了input—filter-output三个阶段的处理流程
• 插件生命周期管理
• 队列管理

Logstash Event

• 数据在内部流转时的具体表现形式。数据在input 阶段被转换为Event，在 output被转化成目标格式数据
• Event 其实是一个Java Object，在配置文件中，对Event 的属性进行增删改查

Codec (Code / Decode)
将原始数据decode成Event;将Event encode成目标数据

插件
Logstash的配置文件主要分为三个部分input 、filter 、output 。每个部分都可以配置多个插件
Input Plugins：https://www.elastic.co/guide/en/logstash/7.17/input-plugins.html

常见的输入插件：
Stdin / File
Beats / Log4J /Elasticsearch / JDBC / Kafka /Rabbitmq /Redis
JMX/ HTTP / Websocket / UDP / TCP
Google Cloud Storage / S3
Github / Twitter

Output Plugins:https://www.elastic.co/guide/en/logstash/7.17/output-plugins.html

常见的输出插件
Elasticsearch
Email / Pageduty
Influxdb / Kafka / Mongodb / Opentsdb / Zabbix
Http / TCP / Websocket

Filter Plugins:https://www.elastic.co/guide/en/logstash/7.17/filter-plugins.html

事件处理插件
Mutate 一操作Event的字段
Metrics — Aggregate metrics
Ruby 一执行Ruby 代码‘’

Codec Plugins:https://www.elastic.co/guide/en/logstash/7.17/codec-plugins.html

事件编译插件
Line / Multiline
JSON / Avro / Cef (ArcSight Common Event Format)
Dots / Rubydebug

Logstash安装和使用

下载并解压logstash： https://www.elastic.co/cn/downloads/past-releases#logstash
选择版本：7.17.3
测试：运行最基本的logstash

cd logstash-7.17.3
#linux
#-e选项表示，直接把配置放在命令中，这样可以有效快速进行测试
bin/logstash -e 'input { stdin { } } output { stdout {} }'
#windows
.\bin\logstash.bat -e "input { stdin { } } output { stdout {} }"

Logstash导入数据到ES

测试数据集下载：https://grouplens.org/datasets/movielens/

下载好数据集放到对应的目录 解压

编写logstash-movie.conf配置文件

input {
  file {
    path => "/usr/local/es/dataset/moves/movies.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}
filter {
  csv {
    separator => ","
    columns => ["id","content","genre"]
  }

  mutate {
    split => { "genre" => "|" }
    remove_field => ["path", "host","@timestamp","message"]
  }

  mutate {

    split => ["content", "("]
    add_field => { "title" => "%{[content][0]}"}
    add_field => { "year" => "%{[content][1]}"}
  }

  mutate {
    convert => {
      "year" => "integer"
    }
    strip => ["title"]
    remove_field => ["path", "host","@timestamp","message","content"]
  }

}
output {
   elasticsearch {
     hosts => "http://192.168.10.114:9200"
     index => "movies"
     document_id => "%{id}"
     user => "elastic"
     password => "123456"
   }
  stdout {}
}

运行logstash

# linux
bin/logstash -f logstash-movie.conf

可以看到原始数据按照这个格式输出。然后来看下es上有没有对应的数据

Logstash从数据库中导入数据到ES

测试数据：

#user表
CREATE TABLE `user` (
  `id` int NOT NULL AUTO_INCREMENT,
  `name` varchar(50) DEFAULT NULL,
  `address` varchar(50) CHARACTER DEFAULT NULL,
  `last_updated` bigint DEFAULT NULL,
  `is_deleted` int DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
#插入数据
INSERT INTO user(name,address,last_updated,is_deleted) VALUES("张三","广州天河",unix_timestamp(NOW()),0)

拷贝jdbc依赖到logstash-7.17.3/driver目录下
Logstash配置文件

input {
  jdbc {
    jdbc_driver_library => "/usr/local/es/logstash-7.17.3/driver/mysql-connector-java-5.1.48.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://xxx:3306/user?useSSL=false"
    jdbc_user => "root"
    jdbc_password => "xxx"
     #启用追踪，如果为true，则需要指定tracking_column
    use_column_value => true
    tracking_column => "last_updated"
     #追踪字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型
    tracking_column_type => "numeric"
     #记录最后一次运行的结果
    record_last_run => true
      #上面运行结果的保存位置
    last_run_metadata_path => "jdbc-position.txt"
    statement => "SELECT * FROM user_elk where last_updated >:sql_last_value;"
     #定时执行
    schedule => " * * * * * *"
  }
}
output {
  elasticsearch {
    document_id => "%{id}"
    document_type => "_doc"
    index => "users"
    hosts => ["http://192.168.10.114:9200"]
    user => "elastic"
    password => "123456"
  }
  stdout{
    codec => rubydebug
  }
}

运行Logstash
bin/logstash -f config/mysql_data.conf

说明成功，查看es数据

当对数据进行增删改查的时候 Logstash回同步的把数据库的变动更新到ES当中，因为 Logstash会定时去执行配置文件中的sql，SELECT * FROM user_elk where last_updated >:sql_last_value :sql_last_value表示最后一次执行成功的时间戳的值，这个值记录在jdbc-position.txt 也是在配置文件中配好的。
而且删除必须是 逻辑删除才会奏效。
测试

# 更新
update user_elk set address="广州白云山",last_updated=unix_timestamp(NOW()) where name="张三"
#删除
update user_elk set is_deleted=1,last_updated=unix_timestamp(NOW()) where name="张三"

当我们在数据中进行删除以后，就不能在查出数据，所以要在es中进行过滤

# 添加别名 进行过滤
POST _aliases
{
"actions": [
  {
    "add": {
      "index": "users",
      "alias": "show_users",
      "filter": {
        "term": {
          "is_deleted": "0"
        }
      }
    }
  }
]  
}
# 查询
GET show_users/_search
{
  "query": {
    "term": {
      "name.keyword": {
        "value": "张三"
      }
    }
  }
}

Beats介绍和使用

Beats 是一个免费且开放的平台，集合了多种单一用途的数据采集器。它们从成百上千或成千上万台机器和系统向 Logstash 或 Elasticsearch 发送数据

FileBeat简介
FileBeat专门用于转发和收集日志数据的轻量级采集工具。它可以作为代理安装在服务器上，FileBeat监视指定路径的日志文件，收集日志数据，并将收集到的日志转发到Elasticsearch或者Logstash。
FileBeat的工作原理
启动FileBeat时，会启动一个或者多个输入（Input），这些Input监控指定的日志数据位置。FileBeat会针对每一个文件启动一个Harvester。Harvester读取每一个文件的日志，将新的日志发送到libbeat，libbeat将数据收集到一起，并将数据发送给输出（Output）。
logstash vs FileBeat

• Logstash是在jvm上运行的，资源消耗比较大。而FileBeat是基于golang编写的，功能较少但资源消耗也比较小，更轻量级。
• Logstash 和Filebeat都具有日志收集功能，Filebeat更轻量，占用资源更少
• Logstash 具有Filter功能，能过滤分析日志
• 一般结构都是Filebeat采集日志，然后发送到消息队列、Redis、MQ中，然后Logstash去获取，利用Filter功能过滤分析，然后存储到Elasticsearch中
• FileBeat和Logstash配合，实现背压机制。当将数据发送到Logstash或 Elasticsearch时，Filebeat使用背压敏感协议，以应对更多的数据量。如果Logstash正在忙于处理数据，则会告诉Filebeat 减慢读取速度。一旦拥堵得到解决，Filebeat就会恢复到原来的步伐并继续传输数据。

FileBeat的安装和使用

下载并解压Filebeat
下载地址：https://www.elastic.co/cn/downloads/past-releases#filebeat

编辑配置
修改 filebeat.yml 以设置连接信息：

output.elasticsearch:
  hosts: ["192.168.10.111:9200","192.168.10.112:9200","192.168.10.114:9200"]
  username: "elastic"
  password: "123456"
setup.kibana:
  host: "192.168.10.111:5601"

启用和配置数据收集模块

# 查看可以模块列表
./filebeat modules list
#启用 Logstash 模块
./filebeat modules enable logstash
#在 modules.d/logstash.yml 文件中修改设置
- module: logstash
  log:
    enabled: true
    var.paths: ["/usr/local/es/logstash-7.17.3/logs/*.log"]

启动 Filebeat

# setup命令加载Kibana仪表板。 如果仪表板已经设置，则忽略此命令。 
./filebeat setup
# 启动Filebeat
./filebeat -e 

ELK整合收集项目日志

有两台服务器 ip分别为 xxx.217和xxx.108。其中217部署项目，108部署Logstash、kibana、es。由于filebeat需要收集日志，所以filebeat部署在271服务器
kibana、es按照之前博客配置能启动就好 不需要额外配置。本次案例不使用密码认证，如果es配置了密码的话，相应的Logstash、kibana也要配置响应密码。
项目的日志路径
/usr/local/springcloud/xxx/logs
filebeat部署
安装下载安装就不再说了，具体配置
filebeat.yml

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      # 当前目录下的所有.log文件
      - /usr/local/springcloud/xxx/logs/*.log
    multiline.pattern: ^\[
    multiline.negate: true
    multiline.match: after
  - type: tcp
    enabled: true
    max_message_size: 10MiB
    # 监听9000端口
    host: "0.0.0.0:9000"
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 1
setup.dashboards.enabled: false
setup.kibana:
  host: "http://xxx.108:5601"
output.logstash:
  # The Logstash hosts
  hosts: ["xxx.108:5044"]
processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - add_docker_metadata: ~
  - add_kubernetes_metadata: ~

执行./filebeat setup没报错即可退出

Logstash部署
下载安装也不再说。看配置
logstash.yml
http.host: "0.0.0.0"
编写filebeat输入到Logstash的配置文件
xxx-log-logstash.conf

input {
    # 来源beats
    beats {
        # 端口
        port => "5044"
    }
}
# 分析、过滤插件，可以多个
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    # 选择elasticsearch
    elasticsearch {
        hosts => ["http://xxx.108:9200"]
        index => "xxx-log-%{+YYYY.MM.dd}"
    }
}

日志收集
按顺序启动es，启动kibana，启动Logstash，启动filebeat
记得开放对应的端口。
filebeat的数据来源：1. 从9000端口监听数据 2. 扫描日志路径下的日志文件，所以需要开启9000端口
Logstash的监听来自5044端口打的数据，因此需要开放5044端口。
启动成功后从kibana上查看数据



出现对应xxx-log-日期的索引。再看下具体数据
首先创建索引模式


我这里是已经创建好了的。
创建好后点击


就可以从这里更直观的看到数据