logstash插件简单介绍

news2024/11/27 22:33:47

logstash插件

输入插件(input)

Input：输入插件。

Input plugins | Logstash Reference [8.11] | Elastic

  • 所有输入插件都支持的配置选项

SettingInput typeRequiredDefaultDescription
add_fieldhashNo{}添加一个字段到一个事件
codeccodecNoplain用于输入数据的编解码器
enable_metricbooleanNotrue
idstringNo添加一个ID插件配置，如果没有指定ID，则Logstash将生成一个ID。强烈建议配置此ID，当两个或多个相同类型的插件时，这个非常有用的。例如，有两个文件输入，添加命名标识有助于监视
tagsarrayNo添加任意数量的标签，有助于后期处理
typestringNo为输入处理的所有事件添加一个字段，自已随便定义，比如linux系统日志，定义为syslog

stdin

  • 标准输入

# cat /etc/logstash/config.d/stdtest.conf
input {
  stdin {
​
  }
}
filter {
​
}
output {
  stdout {
    }
}
​

file

  • 从文件中读取内容

File input plugin | Logstash Reference [8.11] | Elastic

SettingInput typeRequiredDefaultDescription
close_oldernumberNo3600单位秒，打开文件多长时间关闭
delimiterstringNo\n每行分隔符
discover_intervalnumberNo15单位秒，多长时间检查一次path选项是否有新文件
excludearrayNo排除监听的文件，跟path一样，支持通配符
max_open_filesnumberNo打开文件最大数量
patharrayYES输入文件的路径，可以使用通配符 例如/var/log/*/.log，则会递归搜索
sincedb_pathstringNosincedb数据库文件的路径，用于记录被监控的日志文件当前位置
sincedb_write_intervalnumberNo15单位秒，被监控日志文件当前位置写入数据库的频率
start_positionstring, one of ["beginning", "end"]Noend指定从什么位置开始读取文件：开头或结尾。默认从结尾开始，如果要想导入旧数据，将其设置为begin。如果sincedb记录了此文件位置，那么此选项不起作用
stat_intervalnumberNo1单位秒，统计文件的频率，判断是否被修改。增加此值会减少系统调用次数。

# cat /etc/logstash/conf.d/filetest.conf
input {
    file {
        path => "/var/log/messages"
    }
}
filter {
}
output {
    stdout {
    }
}

TCP

  • 通过TCP套接字读取事件,即接收数据。与标准输入和文件输入一样，每个事件都被定位一行文本。

# cat /etc/logstash/conf.d/tcptest.conf
input {
  tcp {
     port => 12345
  }
}
filter {
​
}
output {
  stdout{
    }
}

在其他主机上安装nc工具，对logstash发送信息，即可被读取出来。

[root@vm4 ~]# yum -y install nc
[root@vm4 ~]# nc 10.1.1.13 12345
haha
​
​
在vm3上验证查看
{
      "@version" => "1",
    "@timestamp" => 2019-07-02T15:28:00.301Z,
          "port" => 33778,
          "type" => "nc",
       "message" => "haha",
          "host" => "vm4.cluster.com"
}

Beats

  • 从Elastic Beats框架接收事件

logstash配置文件
​
# cat /etc/logstash/conf.d/filebeattest.conf
input {
  beats {
    port => 5044
    host => "0.0.0.0"
  }
}
 
filter {
 
}
​
output {
  stdout { 
  }
}

filebeat配置文件
​
filebeat.prospectors:
  - type: log
    paths:
      - /var/log/messages
    tags: ["system-log","123"]
    fields:
      level: debug
​
output.logstash:
  hosts: ['127.0.0.1:5044']
​

过滤插件(filter)

参考: Filter plugins | Logstash Reference [8.11] | Elastic

Filter：过滤，将日志格式化。

有丰富的过滤插件：

  • Grok正则捕获

  • date时间处理

  • JSON编解码

  • 数据修改Mutate

  • geoip等。

所有的过滤器插件都支持以下配置选项：

SettingInput typeRequiredDefaultDescription
add_fieldhashNo{}如果过滤成功，添加任何field到这个事件。例如：add_field => [ "foo_%{somefield}", "Hello world, from %{host}" ]，如果这个事件有一个字段somefiled，它的值是hello，那么我们会增加一个字段foo_hello，字段值则用%{host}代替。
add_tagarrayNo[]过滤成功会增加一个任意的标签到事件例如：add_tag => [ "foo_%{somefield}" ]
enable_metricbooleanNotrue
idstringNo
periodic_flushbooleanNofalse定期调用过滤器刷新方法
remove_fieldarrayNo[]过滤成功从该事件中移除任意filed。例：remove_field => [ "foo_%{somefield}" ]
remove_tagarrayNo[]过滤成功从该事件中移除任意标签，例如：remove_tag => [ "foo_%{somefield}" ]

json(关注)

  • JSON解析过滤器，接收一个JSON的字段，将其展开为Logstash事件中的实际数据结构。

示例: 将原信息转成一个大字段，key-value做成大字段中的小字段

# cat /etc/logstash/conf.d/jsontest.conf
input {
  stdin {
  }
}
​
filter {
  json {
    source => "message"
    target => "content"
  }
}
​
output {
  stdout {
  
  }
}
​
​
对标准输入的内容进行json格式输出
把输出内容定向到target指定的content
​
[root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/jsontest.conf
输入测试数据
​
{"ip":"10.1.1.1","hostname":"vm3.cluster.com"}
​
输出测试数据
​
{
       "content" => {
        "hostname" => "vm3.cluster.com",
              "ip" => "10.1.1.1"
    },
    "@timestamp" => 2019-07-02T11:57:36.398Z,
      "@version" => "1",
          "host" => "vm3.cluster.com",
       "message" => "{\"ip\":\"10.1.1.1\",\"hostname\":\"vm3.cluster.com\"}"
}

示例: 直接将原信息转成各个字段

# cat /etc/logstash/conf.d/jsontest.conf
input {
  stdin {
  }
}
​
filter {
  json {
    source => "message"
  }
}
​
output {
  stdout {
  
  }
}
​
​
​
[root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/jsontest.conf
​
输入测试数据
​
{"ip":"10.1.1.1","hostname":"vm3.cluster.com"}
​
输出测试数据
​
​
{
          "port" => 39442,
      "@version" => "1",
    "@timestamp" => 2019-09-19T09:07:03.800Z,
      "hostname" => "vm3.cluster.com",
          "host" => "vm4.cluster.com",
            "ip" => "10.1.1.1",
       "message" => "{\"ip\":\"10.1.1.1\",\"hostname\":\"vm3.cluster.com\"}"
}
​

kv

  • 自动解析为key=value。

  • 也可以任意字符串分割数据。

  • field_split 一串字符，指定分隔符分析键值对

URL查询字符串拆分参数示例
# cat /etc/logstash/conf.d/kvtest.conf
input {
        stdin {
        }
}
​
filter {
  kv {
     field_split => "&?"
  }
}
​
output {
        stdout {
​
        }
}
​
文件中的列以&或?进行分隔
​
执行
[root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/kvtest.conf

输入数据
address=www.abc.com?pid=123&user=abc
​
输出数据
{
           "user" => "abc",
    "@timestamp" => 2019-07-02T12:05:23.143Z,
          "host" => "vm3.cluster.com",
      "@version" => "1",
       "message" => "address=www.abc.com?pid=123&abc=user",
       "address" => "www.abc.com",
           "pid" => "123"
}

使用正则也可以匹配
​
[root@vm3 bin]# cat /etc/logstash/conf.d/kvtest.conf
input {
        stdin {
        }
}
​
filter {
  kv {
     field_split_pattern => ":+"
  }
}
​
output {
        stdout {
​
        }
}

grok(关注)

  • grok是将非结构化数据解析为结构化

  • 这个工具非常适于系统日志，mysql日志，其他Web服务器日志以及通常人类无法编写任何日志的格式。

  • 默认情况下，Logstash附带约120个模式。也可以添加自己的模式（patterns_dir）

  • 模式后面对应正则表达式

  • 查看模式地址：https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns

  • 包含字段如下

SettingInput typeRequiredDefaultDescription
break_on_matchbooleanNotrue
keep_empty_capturesNofalse如果true将空保留为事件字段
matchhashNo{}一个hash匹配字段=>值
named_captures_onlybooleanNotrue如果true，只存储
overwritearrayNo[]覆盖已存在的字段的值
pattern_definitionsNo{}
patterns_dirarrayNo[]自定义模式
patterns_files_globstringNo*Glob模式，用于匹配patterns_dir指定目录中的模式文件
tag_on_failurearrayNo_grokparsefailuretags没有匹配成功时，将值附加到字段
tag_on_timeoutstringNo_groktimeout如果Grok正则表达式超时，则应用标记
timeout_millisnumber30000正则表达式超时时间

grok模式语法

格式：%{SYNTAX:SEMANTIC}

  • SYNTAX 模式名称

  • SEMANTIC 匹配文本的标识符

例如：%{NUMBER:duration} %{IP:client}

# vim /etc/logstash/conf.d/groktest.conf
input {
        stdin {
        }
}
​
​
filter {
  grok {
    match => {
       "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"
    }
  }
}
​
​
output {
        stdout {
​
        }
}
​
​
虚构http请求日志抽出有用的字段
55.3.244.1 GET /index.html 15824 0.043
​
​
输出结果
{
        "client" => "55.3.244.1",
      "duration" => "0.043",
       "message" => "55.3.244.1 GET /index.html 15824 0.043",
        "method" => "GET",
         "bytes" => "15824",
      "@version" => "1",
    "@timestamp" => 2019-07-03T12:24:47.596Z,
          "host" => "vm3.cluster.com",
       "request" => "/index.html"
}
​

自定义模式

如果默认模式中没有匹配的，可以自己写正则表达式。

# vim /opt/patterns
ID [0-9]{3,5}
​
配置文件中应包含如下内容
filter {
  grok {
    patterns_dir =>"/opt/patterns"
    match => {
      "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{ID:id}"              
    }
  }
}
​
​
完整文件内容
[root@vm3 ~]# cat /etc/logstash/conf.d/groktest.conf
input {
        stdin {
        }
}
​
filter {
  grok {
    patterns_dir =>"/opt/patterns"
    match => {
      "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{ID:id}"
    }
  }
}
​
output {
        stdout {
        }
}
​
#执行
[root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/groktest.conf
​
输入测试数据
55.3.244.1 GET /index.html 15824 0.043 6666
​
输出测试数据
{
        "client" => "55.3.244.1",
          "host" => "vm3.cluster.com",
       "request" => "/index.html",
    "@timestamp" => 2019-07-02T12:34:11.906Z,
         "bytes" => "15824",
        "method" => "GET",
       "message" => "55.3.244.1 GET /index.html 15824 0.043 15BF7F3ABB",
      "@version" => "1",
            "id" => "666",
      "duration" => "0.043"
}

geoip(关注)

  • 开源IP地址库

  • GeoLite2 Free Geolocation Data | MaxMind Developer Portal

下载IP地址库
[root@vm3 ~]# wget https://geolite.maxmind.com/download/geoip/database/GeoLite2-City.tar.gz
​
[root@vm3 ~]# tar xf GeoLite2-City.tar.gz
​
[root@vm3 ~]# cp GeoLite2-City_20190625/GeoLite2-City.mmdb /opt

# cat /etc/logstash/conf.d/geoiptest.conf                          
input {
        stdin {
        }
}
​
filter {
  grok {
    match => {
      "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"
    }
  }
  geoip {
      source => "client"
      database => "/opt/GeoLite2-City.mmdb"
  }
}
​
​
output {
        stdout {
​
        }
}
​
​
​
执行
[root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/geoiptest.conf
​
​
输入测试数据
​
202.106.0.20 GET /index.html 123 0.331
​
输出结果
​
{
        "method" => "GET",
        "client" => "202.106.0.20",
         "bytes" => "123",
       "request" => "/index.html",
         "geoip" => {
         "country_code2" => "CN",
          "country_name" => "China",
           "region_code" => "BJ",
             "longitude" => 116.3883,
              "latitude" => 39.9289,
              "timezone" => "Asia/Shanghai",
              "location" => {
            "lon" => 116.3883,
            "lat" => 39.9289
        },
         "country_code3" => "CN",
                    "ip" => "202.106.0.20",
        "continent_code" => "AS",
           "region_name" => "Beijing"
    },
      "duration" => "0.331",
          "host" => "vm3.cluster.com",
       "message" => "202.106.0.20 GET /index.html 123 0.331",
    "@timestamp" => 2019-07-02T12:15:29.384Z,
      "@version" => "1"
}
​
​

[root@vm3 bin]# cat /etc/logstash/conf.d/geoiptest2.conf
input {
        stdin {
        }
}
​
filter {
  grok {
    match => {
      "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"
​
    }
  }
  geoip {
      source => "client"
      database => "/opt/GeoLite2-City.mmdb"
      target => "geoip"
      fields => ["city_name", "country_code2", "country_name","region_name"]
  }
}
​
​
output {
        stdout {
                codec => rubydebug
        }
}
​
​
​
执行
[root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/geoiptest2.conf
​
​
输入测试数据
​
110.226.4.6 GET /home.html 518 0.247
​
输出结果
​
{
          "host" => "vm3.cluster.com",
      "duration" => "0.247",
       "request" => "/home.html",
      "@version" => "1",
        "client" => "110.226.4.6",
       "message" => "110.226.4.6 GET /home.html 518 0.247",
        "method" => "GET",
         "bytes" => "518",
    "@timestamp" => 2019-07-02T12:22:22.458Z,
         "geoip" => {
         "country_name" => "India",
        "country_code2" => "IN"
    }
}
​

输出插件(output)

Output：输出，输出目标可以是Stdout、ES、Redis、File、TCP等。

ES

SettingInput typeRequiredDefaultDescription
hostsURLNo
indexstringNologstash-%{+YYYY.MM.dd}将事件写入索引。默认按日期划分。
userstringNoES集群用户
passwordpasswordNoES集群密码

input {
    file {
        path => ["/var/log/messages"]
        type => "system"
        tags => ["syslog","test"]
        start_position => "beginning"
    }
    file {
        path => ["/var/log/audit/audit.log"]
        type => "system"
        tags => ["auth","test"]
        start_position => "beginning"
    }
}
​
​
filter {
​
}
​
output {
    if [type] == "system" {
        if [tags][0] == "syslog" {
            elasticsearch {
                hosts  => ["http://es1:9200","http://es2:9200","http://es3:9200"]
                index  => "logstash-system-syslog-%{+YYYY.MM.dd}"
            }
            stdout { codec=> rubydebug }
        }
        else if [tags][0] == "auth" {
            elasticsearch {
                hosts  => ["http://es1:9200","http://es2:9200","http://es3:9200"]
                index  => "logstash-system-auth-%{+YYYY.MM.dd}"
            }
            stdout { codec=> rubydebug }
        }
    }
}
​

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1300525.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

代码随想录二刷 ｜二叉树 ｜ 二叉树的右视图

代码随想录二刷 ｜二叉树 ｜ 二叉树的右视图

代码随想录二刷 ｜二叉树 ｜ 二叉树的右视图 题目描述解题思路代码实现 题目描述 199.二叉树的右视图 给定一个二叉树的 根节点 root，想象自己站在它的右侧，按照从顶部到底部的顺序，返回从右侧所能看到的节点值。 示例…
阅读更多...
【vtkWidgetRepresentation】第八期 vtkImplicitCylinderRepresentation

【vtkWidgetRepresentation】第八期 vtkImplicitCylinderRepresentation

很高兴在雪易的CSDN遇见你 前言 本文分享vtkImplicitCylinderRepresentation，主要从源码解析、和实际应用方面展开，希望对各位小伙伴有所帮助！ 感谢各位小伙伴的点赞关注，小易会继续努力分享，一起进步！ …
阅读更多...
基于PLC的电梯控制系统（论文+源码）

基于PLC的电梯控制系统（论文+源码）

1.系统设计 电梯采用了PLC控制方式，通过对PLC进行逻辑程序设计，电梯不仅在控制水平上得到了质的提升，同时在安全性上也得到了大大提高。控制系统在构造上实现了简洁化，不仅优化了硬件接线方便了线路施工，同时对控制要…
阅读更多...
Java设计模式分类

Java设计模式分类

java的设计模式大体上分为三大类： 创建型模式（5种）：工厂方法模式，抽象工厂模式，单例模式，建造者模式，原型模式。 结构型模式（7种）：适配器模式&am…
阅读更多...
51单片机的内核架构组成 介绍

51单片机的内核架构组成 介绍

对于51单片机相信很多电子信息或者相关专业的朋友应该都不会感觉陌生，很多专业在大学课程中开设的单片机课程就是使用的51单片机进行授课和学习的。51单片机的内容相较于其他高性能复杂的单片机来说，架构相对简单一些，寄存器也少很多&#xf…
阅读更多...
【FMCW毫米波雷达设计 】 — FMCW波形

【FMCW毫米波雷达设计 】 — FMCW波形

原书：FMCW Radar Design 1 引言 本章研究驱动FMCW雷达的主要波形:线性调频(LFM)波形。我们研究信号的行为及其性质。随后，本章讨论了匹配滤波理论，并研究了压缩这种波形的技术，特别是所谓的拉伸处理，它赋予FMCW雷达极…
阅读更多...
【STM32】电机驱动

【STM32】电机驱动

一、电机分类 二、直流电机的分类 1.有刷电机 2.无刷电机 3.直流减速电机 三、H桥电路 正向旋转 驱动Q1和Q4 反向旋转 驱动Q2和Q3 四、MC3386电机驱动芯片 1.基本原理图 1）前进/后退：IN1和IN2的电平顺序决定电机的正反转 2）调节速度&#…
阅读更多...
深入了解对象与内置构造函数

深入了解对象与内置构造函数

1. 深入对象 1.1 创建对象的三种方式 1.2 构造函数 语法约定： 总结 构造函数可以快速创建多个对象大写字母开头的函数使用new关键字将对象实例化构造函数不需要返回值自动返回新的对象 new实例化的执行过程 创建空对象this指向对象执行代码，追加新…
阅读更多...
AX和A(T)X的区别是？

AX和A(T)X的区别是？

目录 1.快速了解的例子： （1）假设所有节点的初始特征都是[1, 0, 0] ，那么AX的结果是： （2） 的结果是： (3) 总结： 2.计算结构系数的例子 （1&#xff09…
阅读更多...
Spring IoC和DI

Spring IoC和DI

目录 一. Spring是什么 IoC DI 二. IoC&DI的使用 IoC 1.Controller（控制器存储） 2.Service（服务存储） 3.Repository（仓库存储） 4.Componemt（组件存储） 5.Configuratio…
阅读更多...
【网络安全】网络设备可能面临哪些攻击？

【网络安全】网络设备可能面临哪些攻击？

网络设备通常是网络基础设施的核心，并控制着整个网络的通信和安全，同样面临着各种各样的攻击威胁。 对网络设备的攻击一旦成功，并进行暴力破坏，将会导致网络服务不可用，且可以对网络流量进行控制，利用被攻陷…
阅读更多...
找不到WMVCORE.dll怎么办?一键解决WMVCORE.dll缺失的详细方法分享

找不到WMVCORE.dll怎么办?一键解决WMVCORE.dll缺失的详细方法分享

当打开软件时提示wmvcore.dll丢失，这可能是由于以下几个原因导致的： 系统文件损坏：wmvcore.dll是系统文件，可能会因为各种原因（如病毒感染、系统错误、软件卸载等）而损坏。 软件依赖问题：某些…
阅读更多...
Nginx性能调优实战 1

Nginx性能调优实战 1

Nginx性能调优实战指南 1 Nginx作为一款高性能的Web服务器和反向代理服务器，在处理大量请求和并发连接时表现出色。然而，在实际应用中，为了更好地适应不同的负载和提高系统性能，进行Nginx性能调优是至关重要的。深入探讨Nginx性能…
阅读更多...
mysql数据库文件丢失恢复---惜分飞

mysql数据库文件丢失恢复---惜分飞

客户服务器重启,mysql相关数据文件丢失 通过底层工具进行分析,无法正确恢复数据库名字,一个个单个ibd文件(而且很多本身是错误的) 对于这种情况,通过mysql block扫描恢复出来page文件 恢复出来客户需要数据 这个客户出现该故障的原因大概率是由于文件系统损坏导致.最终…
阅读更多...
SSD基础架构与NAND IO并发问题探讨

SSD基础架构与NAND IO并发问题探讨

在我们的日常生活中，我们经常会遇到一些“快如闪电”的事物：比如那场突如其来的雨、那个突然出现在你眼前的前任、还有就是今天我们要聊的——固态硬盘（SSD）。 如果你是一个技术宅，或者对速度有着近乎偏执的追求&…
阅读更多...
程序的机器代码表示--函数调用

程序的机器代码表示--函数调用

call和ret指令 如何访问栈帧、如何切换栈帧、如何传递参数和返回值 call、ret指令作用： call：1）将IP（即PC）旧值压栈保存（保存在函数的栈帧顶部）；2）设置IP新值&#xff0…
阅读更多...
Module build failed : Error : Vue packages version mismatch:

Module build failed : Error : Vue packages version mismatch:

Vue packages version mismatch: - vue2.7.15 (E:\Workspace_ce\erp\erp-web\node_modules\vue\dist\vue.runtime.common.js) - vue-template-compiler2.6.11 (E:\Workspace_ce\erp\erp-web\node_modules\vue-template-compiler\package.json) 【问题解决了，我很不…
阅读更多...
MyBatis 四大核心组件之 ParameterHandler 源码解析

MyBatis 四大核心组件之 ParameterHandler 源码解析

🚀 作者主页： 有来技术 🔥 开源项目： youlai-mall 🍃 vue3-element-admin 🍃 youlai-boot 🌺 仓库主页： Gitee 💫 Github 💫 GitCode 💖 欢迎点赞…
阅读更多...
55.手写实现grpc连接池以及gin和grpc交互

55.手写实现grpc连接池以及gin和grpc交互

文章目录 一、简介前置说明 二、敏感词过滤服务1、定义sensitive.proto文件2、protoc生成pb.go文件3、sensitive服务端实现 三、关键词匹配服务1、编写keywords.proto文件2、生成pb.go文件3、keywords服务端实现 四、gin web 路由服务1、新建grpcpool服务作为gin web服务2、根据…
阅读更多...
.net core提示The xx field is required，One or more validation errors occurred

.net core提示The xx field is required，One or more validation errors occurred

访问接口时缺少model中的参数时，会提示： The xx field is required One or more validation errors occurred原因是.net core webapi默认参数为不可空，因此会验证并报错。 解决方案： 在项目的.csproj中，修改Nullable…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023