zoukankan      html  css  js  c++  java
  • elk示例-精简版

    作者:Danbo 2016-03-09

    1.Grok正则捕获

    input {stdin{}}
    filter {
        grok {
            match => {
                "message" => "s+(?<request_time>d+(?:.d+)?)s+"
            }
        }
    }
    output {stdout{ codec => rubydebug  }}

    运行logstash输出结果如下:

    [root@centos-linux logstash]# bin/logstash -f conf/grok.conf 
    Logstash startup completed
    begin 123.456 end
    {
             "message" => "begin 123.456 end",
            "@version" => "1",
          "@timestamp" => "2016-03-09T12:30:58.976Z",
                "host" => "centos-linux.shared",
        "request_time" => "123.456"
    }

    不过这里request_time应该是数值而不是字符串。

    grok表达式的打印复制格式的完整语法是下面这样的:

    %{PATTERN_NAME:capture_name:data_type}

    date_type 目前只支持两个值:int 和 float。

    我们将配置修改成下面这样:

    input {stdin{}}
    filter {
        grok {
            match => {
                "message" => "%{WORD} %{NUMBER:request_time:float} %{WORD}"
            }
        }
    }
    output {stdout{ codec => rubydebug  }}

    运行结果如下:

    [root@centos-linux logstash]# bin/logstash -f conf/grok1.conf 
    Logstash startup completed
    begin 123.456 end
    {
             "message" => "begin 123.456 end",
            "@version" => "1",
          "@timestamp" => "2016-03-09T12:46:29.913Z",
                "host" => "centos-linux.shared",
        "request_time" => 123.456

    此时request_time 变成数值类型了。

    实际运用中,我们需要处理各种各样的日志文件,如果都是在配置文件里各自写一行自己的表达式,就完全不可管理了。我们建议把所有的grok表达式统一写入到一个地方。然后用filter/grok的patterns_dir 选项来指明。

    重点:如果你把"message"里所有的信息都grok到不同的字段了,数据数值上就相当于是重复复制存储了两份。所以你可以用remove_field 参数来删除掉message字段,或者用overwrite参数来重写默认的message字段,只保留最重要的部分。

    重写参数示例:

    input { stdin {} }
    filter {
        grok {
            patterns_dir => "/path/to/your/own/patterns"
            match => {
                "message" => "%{SYSLOGBASE} %{DATA:message}"
            }
            overwrite => ["message"]
        }
    }

    2.多项选择

    在和codec/multiline搭配使用的时候,需要注意一个问题,grok正则和普通正则一样,默认是不支持匹配回车换行的。就像你需要 =~ //m一样也需要单独指定,具体写法是在表达式开始位置加上(?m)标记。

    具体如下所示:

    match => {
        "message" => "(?m)s+(?<reuqest_time>d+(?:.d+)?)s+"
    }

    有时我们会碰上一个日志有多种可能格式的情况。这时候要写成单一正则就比较困难,或者全用| 隔开比较丑陋。

    实际上logstash/filters/grok插件的match参数应该接受的是一个Hash值。但是因为早期的logstash语法中Hash值也是用数组这种方式书写的,因此我们传递Array值给match参数也是可以的,我们这里是可以传递多个正则来匹配同一个字段:

    match => [
        "message", "(?<request_time>d(?:.d+)?)",
        "message", "%{SYSLOGBASE} %{DATA:message}",
        "message", "(?m)%{WORD}"
    ]

    logstash会按照这个定义次序以此尝试匹配,到匹配成功为止。虽说效果跟用 | 大大的正则是一样的,但是阅读性提高了很多。

    3.GeoIP地址查询归类

    GeoIP是最常见的免费IP地址归类查询库。GeoIP库可以根据IP地址提供对应的地域信息,包括国别、省市、经纬度等,对于可视化地图和区域统计非常有用。

    配置信息:

    [root@centos-linux logstash]# cat conf/geoip.conf 
    input {stdin{}}
    filter {
        geoip {
            source => "message"
    }
    }
    output {stdout { codec => rubydebug }}

    运行结果:

    [root@centos-linux logstash]# bin/logstash -f conf/geoip.conf
    Logstash startup completed
    114.114.114.114
    {
    "message" => "114.114.114.114",
    "@version" => "1",
    "@timestamp" => "2016-03-09T13:57:36.182Z",
    "host" => "centos-linux.shared",
    "geoip" => {
    "ip" => "114.114.114.114",
    "country_code2" => "CN",
    "country_code3" => "CHN",
    "country_name" => "China",
    "continent_code" => "AS",
    "region_name" => "19",
    "city_name" => "Chaoyang",
    "latitude" => 41.5703,
    "longitude" => 120.45859999999999,
    "timezone" => "Asia/Harbin",
    "real_region_name" => "Liaoning",
    "location" => [
    [0] 120.45859999999999,
    [1] 41.5703
    ]
    }
    }

    GeoIP库数据较多,如果你不需要这么多内容,可以通过fields选项指定自己所需要的。

    注意geoip库内只存有公网上的IP信息,查询不到内网结果的,会直接返回null,

    4.Key-Value切分

    在很多情况下,日志内容本身都是一个类似于key-value的格式,但是格式具体的样式确实多样的。Logstash提供 filters/kv 插件,帮助处理不同样式的key-value日志,变成实际的LogStash::Event数据。

    配置示例:

    Nginx访问日志中的$request,通过这段配置,可以详细切分成method,url_path,verb,url_a,url_b

    我们看kv { } 这个字段。

    上例即表示,除了url_uid 和 url_cip 两个字段以外,其他url_* 都不保留。

    5.数据修改(Mutate)

    数据类型转化:(可以设置的转换类型包括:“integer”,“float”,“sting”)

    filter {
        mutate {
            convert => ["request_time", "float"]
        }
    }

    注意:mutate除了转换简单的字符值,还支持对数组类型的字段进行转换,即将["1", "2"]转换成[1,2]。但不支持对哈希类型字段做类似处理。

    字符串处理

    -gsub

    仅对字符串类型字段有效

    input { stdin { } }
    
    filter {
        mutate {
            gsub => ["message", "[\?#]", "_"]
        }
    }
    
    output {
        stdout { codec => rubydebug }
    }

    运行结果如下:

    fight?for?you
    {
           "message" => "fight_for_you",
          "@version" => "1",
        "@timestamp" => "2016-03-12T07:05:30.419Z",
              "host" => "cen

    -split

    input { stdin { } }
    
    filter {
        mutate {
            split => ["message", "|"]
        }
    }
    
    output { stdout { codec => rubydebug } }

    随意输入一串以 | 分隔的字符,比如“123|321|abc|??!”,可以看到如下输出:

    123|321|abc|??!
    {
           "message" => [
            [0] "123",
            [1] "321",
            [2] "abc",
            [3] "??!"
        ],
          "@version" => "1",
        "@timestamp" => "2016-03-12T07:10:26.352Z",
              "host" => "centos-linux.shared"
    }

    -join

    仅对数组类型字段有效,我们在之前已经用split割切的基础再join回去。配置改成:

    input { stdin { } }
    
    filter {
        mutate {
            split => ["message", "|"]
        }
        mutate {
            join => ["message", ","]
        }
    }
    
    output { stdout { codec => rubydebug } }

    filter区段之内,是顺序执行,我们看到最后的输出结果是:

    123|321|abc|??!
    {
           "message" => "123,321,abc,??!",
          "@version" => "1",
        "@timestamp" => "2016-03-12T07:18:32.835Z",
              "host" => "centos-linux.shared"
    }

    -merge

    合并两个数组或者哈希字段。依然在之前的split基础上继续:

    input { stdin { } }
    
    filter {
        mutate {
            split => ["message", "|"]
        }
        mutate {
            merge => ["message", "message"]
        }
    }
    
    output { stdout { codec => rubydebug } }

    当我们输入123|321|abc|??!*=123 之后我们会看到输出:

    123|321|abc|??!*=123
    {
           "message" => [
            [0] "123",
            [1] "321",
            [2] "abc",
            [3] "??!*=123",
            [4] "123",
            [5] "321",
            [6] "abc",
            [7] "??!*=123"
        ],
          "@version" => "1",
        "@timestamp" => "2016-03-12T07:27:34.886Z",
              "host" => "centos-linux.shared"
    }

    如果src字段是字符串,会自动先转换成一个单元素的数组再合并

    如果把上一示例中的来源字段改成“host”

    运行结果如下所示:

    123|321|abc|??!
    {
           "message" => [
            [0] "123",
            [1] "321",
            [2] "abc",
            [3] "??!",
            [4] "centos-linux.shared"
        ],
          "@version" => "1",
        "@timestamp" => "2016-03-12T07:32:16.231Z",
              "host" => "centos-linux.shared"
    }

    看到目的字段“message”确实多了一个元素,但是来源字段"host"本身也由字符串变成数组类型了。

    字段处理

    -rename

    顾名思义重命名某个字段,如果目的字段已经存在,会被覆盖掉:

    filter {
        mutate {
            rename => ["syslog_host", "host"]
        }
    }

    -update

    更新某个字段的内容。如果字段不存在,不会新建。

    -replace

    作用和update类似,但是当字段不存在的时候,它会起到add_field 参数一样的,自动添加新的字段。

    6.split拆分事件

    我们上章通过multiline插件将多行数据合并进一个事件里,那么反过来,也可以把一行数据,拆分成多个事件。这就是split插件。

    配置示例:

    input { stdin { } }
    
    filter {
        split {
            field => "message"
            terminator => "#"
        }
    }
    
    output { stdout { codec => rubydebug } }

    输入fuck#you#man 运行结果如下所示:

    fuck#you#man
    {
           "message" => "fuck",
          "@version" => "1",
        "@timestamp" => "2016-03-12T16:31:05.564Z",
              "host" => "centos-linux.shared"
    }
    {
           "message" => "you",
          "@version" => "1",
        "@timestamp" => "2016-03-12T16:31:05.564Z",
              "host" => "centos-linux.shared"
    }
    {
           "message" => "man",
          "@version" => "1",
        "@timestamp" => "2016-03-12T16:31:05.564Z",
              "host" => "centos-linux.shared"
    }

    重要提示

    split插件中使用的是yield功能,其结果是split 出来的新事件,会直接结束其在filter阶段历程,也就是说写在split后面的其他filter插件都不起作用,进入output阶段。所以,一定要保证split配置写在全部filter配置的最后。

    7.保存进Elasticsearch

    Logstash可以试用不同的协议实现完成将数据写入Elasticsearch的工作。在不同时期,也有不同的插件实现方式。

    P74

    **************

  • 相关阅读:
    2020.10.23 19级training 补题报告
    2020.10.17 天梯赛练习 补题报告
    2020.10.16 19级training 补题报告
    2020.10.9 19级training 补题报告
    2020.10.10 天梯赛练习 补题报告
    2020.10.3 天梯赛练习 补题报告
    2020.10.2 19级training 补题报告
    第十届山东省ACM省赛复现补题报告
    VVDI Key Tool Plus Adds VW Passat 2015 Key via OBD
    Xhorse VVDI Prog Software V5.0.3 Adds Many MCUs
  • 原文地址:https://www.cnblogs.com/danbo/p/5258730.html
Copyright © 2011-2022 走看看