文档中心

上云无忧 > 文档中心 > 百度智能云Elasticsearch - Logstash使用指南
百度智能云
Elasticsearch
简介/价格/文档
百度智能云Elasticsearch - Logstash使用指南

文档简介：
简介： Logstash是一款开源的实时数据采集工具。通过配置多种类型的input、fliter、output插件，Logstash支持多类数据源的数据采集、转换，并存储到目的端。
*此产品及展示信息均由百度智能云官方提供。免费试用 咨询热线：400-826-7010，为您提供专业的售前咨询，让您快速了解云产品，助您轻松上云！ 微信咨询
  免费试用、价格特惠

简介

Logstash是一款开源的实时数据采集工具。通过配置多种类型的input、fliter、output插件，Logstash支持多类数据源的数据采集、转换，并存储到目的端。

安装

  1. 检查JDK版本是不是JAVA 8以上，检查方式： 
java -version
  1. 登录Elasticsearch官网，选择版本并下载Logstash。如果用户集群是百度智能云Elasticsearch 7.4.2版本，使用以下链接可直接下载logstash-oss-7.4.2.zip。 
wget https://artifacts.elastic.co/downloads/logstash/logstash-oss-7.4.2.zip

快速使用

下边以用户在BCC上解析日志数据并写入百度智能云Elasticsearch为例，描述Logstash的使用方式。

安装Logstash

  1. 登录BCC并配置JAVA环境
  2. 下载并解压Logstash
  3. 在config目录下新建一个配置文件，命名为logstash-es.conf，格式如下： 
input {
  ...
}
filter {
  ...
}
output {
  ...
}

配置input

日志目录路径为/{LOG_PATH}/logs/*.log，日志格式为： 

[2021-01-22T10:21:02,837][INFO ][c.b.e.i.s.TestClass] [test_node1]test log.
[2021-01-22T10:21:02,838][INFO ][c.b.e.i.s.TestClass] [test_node1]test log.
[2021-01-22T10:21:02,838][INFO ][c.b.e.i.s.TestClass] [test_node1]test log.
[2021-01-22T10:26:02,839][ERROR][c.b.e.i.s.TestClass] [test_node1]throws a ERROR:
c.b.e.i.s.TestException: this a ERROR
        at c.b.e.i.s.TestClass.test(TestClass.java:387) ~[baidu-test-logstash-7.4.2.jar:?]
        at c.b.e.i.s.TestClass.test(TestClass.java:385) ~[baidu-test-logstash-7.4.2.jar:?]

则input配置为： 

input {
  file {
    path => "/{LOG_PATH}/logs/*.log"  #日志数据文件路径
    start_position => "beginning"     #读取文件位置，默认为 end            
    codec => multiline {              #multiline插件。匹配不是以“[”开头的日志数据，并入上一行数据。
      pattern => "^\["
      negate => true
      what => previous
    }
  }
}

配置filter

针对采集的日志，数据需要解析出日志时间time、日志级别level、日志所在类class、节点名称node、日志详细信息log_data，则可以使用grok插件配置filter，使用方式如下： 

filter {
  grok {
    match => { "message" => "\[%{TIMESTAMP_ISO8601:time}\]\[%{LOGLEVEL:level}( )*\]\[%{JAVACLASS:class}
( )*\] \[%{DATA:node}\]%{GREEDYDATA:log_data}" }
    remove_field => ["tags","path","@version","host"]
  }
}

Grok

简介

grok是一种采用组合多个预定义的正则表达式，用来匹配分割文本并映射到关键字的工具。通常用来对日志数据进行预处理。

Logstash内置grok的120个预定义匹配字段，可参考：grok默认预定义匹配字段。grok支持正则表达式和自定义匹配字段规则，可以灵活满足扩展的需求。grok同时支持多条匹配，因此如果有不同格式的日志数据，可以使用多个正则表达式语句同时匹配。

示例

下文是一条http请求日志： 

55.3.244.1 GET /index.html 15824 0.043

可以使用如下grok pattern来匹配这种记录： 

%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

以下是结果： 

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

配置output

输出目的端为百度智能云Elasticsearch，索引名称格式为test_logstash_index_{YYYY.MM.dd}，用户可以创建模板文件template.json，格式如下： 

{
  "template": "test_logstash_index_*",
  "settings": {
    "index.number_of_shards": 5,
    "number_of_replicas": 1,
    "index.refresh_interval": "60s"
  },
  "mappings": {
    "properties": {
      "@timestamp": {
        "type": "date"
      },
      "class": {
        "type": "keyword"
      },
      "log_data": {
        "type": "keyword"
      },
      "level": {
        "type": "keyword"
      },
      "message": {
        "type": "keyword"
      },
      "node": {
        "type": "keyword"
      },
      "time": {
        "type": "date"
      }
    }
  }
}

output配置如下： 

output {
  elasticsearch {
    hosts => ["http://{ES_URL}"]
    index => "test_logstash_index_%{+YYYY.MM.dd}"
    template => "/{PATH}/template.json"
    template_name => "my_template_es"
    user => "{USER}"
    password => "{PASSWORD}"
  }
}

启动Logstash 

bin/logstash -f config/logstash-es.conf

启动后，通过以下方式查询ES验证日志写入情况： 

POST /test_logstash_index*/_search
{
   "query": {
      "match_all": {}
   }
}

查询结果如下： 

[
  {
    "_index": "test_logstash_index_2021.01.01",
    "_type": "_doc",
    "_id": "3Sf3KHcBRfCVcyxEVdqW",
    "_score": 1,
    "_source": {
      "message": "[2021-01-22T10:21:02,837][INFO ][c.b.e.i.s.TestClass] [test_node1]test log.",
      "node": "test_node1",
      "level": "INFO",
      "@timestamp": "2021-01-22T07:32:24.927Z",
      "time": "2021-01-22T10:21:02,837",
      "class": "c.b.e.i.s.TestClass",
      "log_data": "test log."
    }
  },
  {
    "_index": "test_logstash_index_2021.01.01",
    "_type": "_doc",
    "_id": "4ScCKXcBRfCVcyxEL9rz",
    "_score": 1,
    "_source": {
      "message": "[2021-01-22T10:21:02,838][INFO ][c.b.e.i.s.TestClass] [test_node1]test log.",
      "node": "test_node1",
      "level": "INFO",
      "@timestamp": "2021-01-22T07:32:24.927Z",
      "time": "2021-01-22T10:21:02,838",
      "class": "c.b.e.i.s.TestClass",
      "log_data": "test log."
    }
  },
  {
    "_index": "test_logstash_index_2021.01.01",
    "_type": "_doc",
    "_id": "3yf3KHcBRfCVcyxEdNqY",
    "_score": 1,
    "_source": {
      "message": "[2021-01-22T10:21:02,838][INFO ][c.b.e.i.s.TestClass] [test_node1]test log.",
      "node": "test_node1",
      "level": "INFO",
      "@timestamp": "2021-01-22T07:32:24.927Z",
      "time": "2021-01-22T10:21:02,838",
      "class": "c.b.e.i.s.TestClass",
      "log_data": "test log."
    }
  },
  {
    "_index": "test_logstash_index_2021.01.01",
    "_type": "_doc",
    "_id": "4Cf4KHcBRfCVcyxEDtpl",
    "_score": 1,
    "_source": {
      "@timestamp": "2021-01-22T07:21:21.141Z",
      "message": "[2021-01-22T10:26:02,839][ERROR][c.b.e.i.s.TestClass] [test_node1]throws a 
ERROR:\nc.b.e.i.s.TestException: this a ERROR\n        at c.b.e.i.s.TestClass.test(TestClass.java:387) 
~[baidu-test-logstash-7.4.2.jar:?]\n        at c.b.e.i.s.TestClass.test(TestClass.java:385) ~[baidu-test-logstash-7.4.2.jar:?]",
      "time": "2021-01-22T10:26:02,839",
      "level": "ERROR",
      "node": "test_node1",
      "log_data": "throws a ERROR:\nc.b.e.i.s.TestException: this a ERROR\n        at c.b.e.i.s.TestClass.test(TestClass.java:387) 
~[baidu-test-logstash-7.4.2.jar:?]\n        at c.b.e.i.s.TestClass.test(TestClass.java:385) ~[baidu-test-logstash-7.4.2.jar:?]",
      "class": "c.b.e.i.s.TestClass"
    }
  }
]

Logstash常用插件

输入插件input

插件名称 说明 详细信息
beats 从Elastic Beats框架中接收事件。 Beats input plugin
elasticsearch 从Elasticsearch集群中读取数据。 Elasticsearch input plugin
generator 生成随机测试数据。 Generator input plugin
file 从文件中读取数据 file input plugin
http 通过HTTP或HTTPS接收单行或多行事件。 Http input plugin
jdbc 通过JDBC，将任一数据库数据读取到Logstash中。 Jdbc input plugin
kafka 从Kafka主题读取事件。 Kafka input plugin
rabbitmq 从RabbitMQ队列中读取事件。 Rabbitmq input plugin
redis 从Redis实例中读取事件。 Redis input plugin
stdin 从标准输入读取事件。 Stdin input plugin
syslog 在网络上将Syslog消息作为事件读取。 Syslog input plugin
tcp 通过TCP套接字读取事件。 Tcp input plugin
udp 在网络上通过UDP，将消息作为事件读取。 Udp input plugin

更多input插件可参考Logstash官网。

编码插件codec

常用编码插件：

插件名称 说明 详细信息
json 解码（通过输入）和编码（通过输出）完整的JSON消息。 Json codec plugin
json_lines 解码以新行分隔的JSON流。 Json_lines codec plugin
line 读取面向行的文本数据。 Line codec plugin
multiline 将多行消息合并到单个事件中。 Multiline codec plugin

更多codec插件可参考Logstash官网。

过滤器插件filter

插件名称 说明 详细信息
date 解析字段中的日期，然后使用该日期或时间戳作为事件的logstash时间戳。 Date filter plugin
geoip 根据GeoIP库为数据中的IP地址提供对应的地理位置信息。 Geoip filter plugin
grok 通过grok正则匹配将数据结构化。 Grok filter plugin
json JSON解析过滤器，将包含JSON的已有字段扩展为Logstash事件中的实际数据结构 JSON filter plugin
kv 处理各种形如key-value格式的数据 Kv filter plugin
metrics 聚合指标 Metrics filter plugin
mutate mutate提供了丰富的基础类型数据处理能力，包括类型转换，字符串处理和字段处理等 Mutate filter plugin
ruby 执行Ruby代码 Ruby filter plugin
split 通过分解事件的一个字段，并将产生的每个值嵌入原始事件的克隆版，从而克隆事件。 被分解的字段可以是字符串或字符串数组。 Split filter plugin
xml XML过滤器。将包含XML的字段，扩展为实际数据结构 Xml filter plugin

更多input插件可参考Logstash官网。

输出插件output

插件名称 说明 详细信息
stdout 将数据向Linux控制台标准输出。 Stdout output plugin
tcp 通过TCP套接字写入数据。 Tcp output plugin
csv 以逗号分隔（CSV）或其他分隔的格式，将数据写入磁盘。 Csv output plugin
elasticsearch 向Elasticsearch写入数据。 Elasticsearch output plugin
email 收到输出后发送电子邮件。 Email output plugin
file 向磁盘文件写入数据。 File output plugin

更多output插件可参考Logstash官网。

Logstash命令行使用

Logstash 提供了可执行脚本 bin/logstash ，可以方便快速运行，支持以下主要参数：

  • -e

快速执行Logstash，一般用于测试使用。

  • --config 或 -f

通常会把配置写到配置文件里，然后通过 bin/logstash -f agent.conf 的形式运行Logstash。

此外，Logstash还可以直接用 bin/logstash -f /etc/logstash.d/ 来运行，Logstash 会自动读取 /etc/logstash.d/ 目录下所有的文本文件，并执行。

  • --log 或 -l

Logstash 默认输出日志到标准错误。生产环境下可以通过 bin/logstash -l logs/logstash.log 命令来统一存储日志。

  • --filterworkers 或 -w

Logstash会为过滤插件（filter）运行多个线程。例如： bin/logstash -w 5为Logstash的filter会运行 5 个线程同时工作。

注意：Logstash目前不支持输入插件的多线程，而输出插件的多线程需要在配置内部设置。

相似文档
  • 百度智能云Elasticsearch - 基于ELK构建日志分析系统
    介绍： 日志是一个系统不可缺少的组成部分，日志记录系统产生的各种行为，基于日志信息可以对系统进行问题排查，性能优化，安全分析等工作。 使用ELK(Elasticsearch、Logstash、Kibana) 结合Filebeat与Kafka构建日志分析系统，可以方便的对日志数据进行采集解析存储，以及近实时的查询分析。
  • 百度智能云Elasticsearch常见问题总览
    Elasticsearch系统常见问题： 如何查看Es安装了哪些插件？ 线程池队列满导致错误？ Too Many Open Files的错误？ Es 中一个分片一般设置多大？ 当集群为red或者yellow的时候怎么办？ 如何cancel掉慢查询？
  • 百度智能云Elasticsearch常见问题QA
    本文档主要目的： 用来积累用户常见的问题。 文档中的操作可通过 Kibana 或在chrome浏览器中安装sense插件完成。用户也可以通过curl的方式自行实现。
  • Spark访问百度智能云Elasticsearch常见问题
    说明： 本文档主要介绍了通过elasticsearch-hadoop中的Spark访问ES时常见配置项意义。本文中的es-spark是elasticsearch-hadoop中和Spark相关联的包，用户通过自己的Spark集群读写ES集群，elasticsearch-hadoop基本上兼容了目前ES所有的版本。
  • 百度智能云Elasticsearch服务等级协议SLA（V1.0）
    协议生效时间：2020 年 9 月 1 日。 本服务等级协议（Service Level Agreement，以下简称 “SLA”）规定了百度智能云向用户提供的Elasticsearch的服务可用性等级指标及赔偿方案。
官方微信
联系客服
400-826-7010
7x24小时客服热线
分享
  • QQ好友
  • QQ空间
  • 微信
  • 微博
返回顶部