简介

　　ELK技术栈中的logstash起到了管道的作用，其主要的功能可以看做input-filter-output这几样，因此可以看住logstash在数据传输中的作用就是对数据进行一定程度的修改。一般我们是用此来使数据格式更加规范，符合我们期望的格式，从而在kibana或者ES检索更加方便。

主要组成

　　logstash下载即用，对环境的要求是jre，11以上就能完全使用了。其中bin目录是主文件，config目录是配置目录，有关属性配置，性能调优都在这一块。一般主要用到的是logstash-sample.conf，input-filter-output都写在这一块。
　　顾名思义，input就是获取数据，filter就是过滤器，也是我们处理数据的地方，output则是输出到指定地区。本笔记以从kafka获取数据，输出到ES为示例。

input

　　input区域较为简洁，我们从kafka获取数据可以这样按下示例子写。具体配置参见这里

input{    kafka{ #一个input中可以写多个kafka        bootstrap_servers => "localhost:9092" #这里是kafka的地址，如果要从多个kafka获取数据的话可以以逗号分隔        topic => "nginx" #指定从kafka的哪个topic获取数据        codec => json{            charset => "UTF-8"        } #如果数据是json格式可直接在这里进行转换        consumer_threads => 2 #使用多少个消费者线程，尽量与分区数一致        enable_auto_commit => true #是否启用自动告诉kafka成功消费的偏移量(或索引)，下次会自动以该偏移量为消耗的初始偏移量        auto_offset_reset => "latest" #初始时将偏移量移到最新偏移量        auto_commit_interval_ms => "1000" #定时提交偏移量的时间间隔        group_id => "logstash" #指定该消费者的group_id。可用于负载均衡        add_field => {            "[@metadata][type]" => "nginx" #以@metadata来存放数据的是不会被output的，除某些被配置了metadata => true的配置项        }#增加数据的字段    }}

filter

　　filter区是我们处理数据的主力区域，数据的处理流程都在这里写就。filter是通过一个个插件组合实现的，具体的插件以及参数可以从这里看。目前插件数量有46种，常用的有date drop grok json mutate。filter区域的写法如下，其格式与input的大致相同的。

filter{    plugin{        ...some param    }}

　　需要重视的是各个插件的组合使用。下文涉及的所有tag都会放在最外层的tags的列表里。插件通用继承了的且常用方法有add_field add_tag id remove_field remove_tag这些方法在任意一个filter中都可以使用。

date

filter{    date{        match=>["client_time","yyyy-MM-dd HH:mm:ss"] #若client_time字段与后者匹配则替换target，会变成UTC时间，标准时区        target=>"real time" #target默认为@timestamp    }}

drop

filter{    if [Level] == "DEBUG"{#利用[filed]的语法可访问字段，可嵌套        drop {} #drop后日志就会被丢弃    }}

grok

　　grok是十分重要的插件，用于按正则提取信息后自动赋值到字段。更进一步的使用规则可以点击这里。

filter{    match =>{        prtterns_dir => ["./patterns"] #如果你有需要可以自己定义一个规则        "message" => "%{IP:client} %{Number:times:int}" #match会按字段=>正则的形式提取信息，其%{IP:client}中，IP是内置规则，client是欲赋值的字段名，默认值均为字符串，如果已知类型可以加以转换，如在后面加上:int来转化为整数型    }}

　　内置的规则你可以从这里查看。

json

filter{    json {        source => "message" #指定需要解码的字段        target => "jsoncontent" #指定存放解码后内容的字段，如果不写将会把所有字段释放在最外层        remove_field => "test" #删除名为test的字段    }#如果解码失败会产生一个"_jsonparsefailure"的tag，当然你也可以用tag_on_failure来指定}

mutate

　　mutate是用于数据修改、字符串处理、类型转换等重要功能的插件。

filter{    mutate {        split => ["hostname","."] #会被原地替换        add_field => { "shortHostname" => "%{hostname[0]}" }    }    mutate {        rename => ["shortHostname","hostname"] #替换    }    mutate {        convert => {            "fieldname" => "integer"            "booleanfield" => "boolean"        }    }}

　　由于功能繁多，建议一定要去看官方文档。

output

　　output的格式与input的格式类似。

output{    elasticsearch{        hosts => ["localhost:9200"]    }}

更多的设置，如修改index等，可参考官方文档使用。

性能调优

　　logstash的性能与内存占用向来为人所诟病，因此我们要关注的是单独抽离logstash，使其独立成为集群。轻量级的beat的为此提供了解决方案，不需要依赖，内存与cpu消耗极低，而kafka则是缓解了生产者与消费者时间-速率不一致的矛盾。从而我们可以通过拓展logstash集群从而达到提高处理速率的目的。其中，我们也可以通过调整logstash的参数来提高其性能。

• worker 调整worker数量，可以提高filter的效率，一般可设置为略高于CPU数
• batch_size 调整worker一次批量处理的条数，默认为150，其与ES的bulk写入次数相关。
• Output.flush_size&Output.idle_flush_time 控制lush的大小与频率
• Filter.grok 尽量用合适的类型分割字段，尽量避免使用DATA，增加一定的锚点可提高grok性能