logstash学习

logstash学习

一、logstash学习

​​参考指南（logstash.yml）

3、startup.options

参数 用途JAVACMD=/usr/bin/java 　　　　　　　　　　　　　　 本地jdkLS_HOME=/opt/logstash 　　　　　　　　　　　　　　 logstash所在目录LS_SETTINGS_DIR="${LS_HOME}/config" 　　　　　　默认logstash配置文件目录LS_OPTS="–path.settings ${LS_SETTINGS_DIR}" logstash启动命令参数 指定配置文件目录LS_JAVA_OPTS="" 　　　　　　　　　　　　　　　　　　指定jdk目录LS_PIDFILE=/var/run/logstash.pid 　　　　　　　　logstash.pid所在目录LS_USER=logstash 　　　　　　　　　　　　　　　　　 logstash启动用户LS_GROUP=logstash 　　　　　　　　　　　　　　　　　logstash启动组LS_GC_LOG_FILE=/var/log/logstash/gc.log logstash jvm gc日志路径LS_OPEN_FILES=65534

二、修改配置文件

1、logstash kafka输入插件

2、logstash的filter

1、logstash处理json日志（嵌套json）

2、logstash 数据采集时间差8小时问题及解决

三种方法

3、logstash @timestamp 内容替换

filter { grok{ match => ["message","%{HTTPDATE:[@metadata][timestamp]}"] } date{ match=>["[@metadata][timestamp]","dd/MMM/yyyy:HH:mm:ss Z"] }}

4、logstash条件判断语句

使用条件来决定filter和output处理特定的事件。logstash条件类似于编程语言。条件支持if、else if、else语句，可以嵌套。 比较操作有： 相等: ==, !=, <, >, <=, >= 正则: =~(匹配正则), !~(不匹配正则) 包含: in(包含), not in(不包含) 布尔操作： and(与), or(或), nand(非与), xor(非或) 一元运算符： !(取反) ()(复合表达式), !()(对复合表达式结果取反) 2、if[foo] in "String"在执行这样的语句是出现错误原因是没有找到叫做foo的field，无法把该字段值转化成String类型。所以最好要加field if exist判断。判断字段是否存在，代码如下：if ["foo"] { mutate { add_field => { "bar" => "%{foo}"} }}

3、logstash中字符串的split，对每个子串进行json解析

4、配置logstash消费kafka多个topic,分别生成索引

​​{ kafka { bootstrap_servers => ["l-kafkacluster1.ops.test.com:9092,l-kafkacluster2.test.com:9092,l-kafkacluster3.ops.test.com:9092,l-kafkacluster4.ops.test.com:9092,l-kafkacluster5.ops.test.com:9092"] group_id => "elkstacki" #logstash集群的标志 topics => ["ng"] consumer_threads => 5 #多实列相加最好等于 topic分区数 auto_offset_reset => "latest" decorate_events => true codec => json }}filter { grok { match => { "message" => '%{IPV4:clientip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:%{NUMBER:response:int} (?:%{NUMBER:bytes:int}|-) "(?.*?)" %{QS:agent} "(?:%{IPV4:forwarded}|-)" "%{BASE10NUM:request_duration_front:float}" "(%{BASE10NUM:request_duration_back:float}|.*?)" "%{IPORHOST:host}" "(%{IPORHOST:upstream_host}:%{POSINT:upstream_port}|.*?)" "(?.*)" "(?.*)" "(?.*)" "(?.*)" "(?.*)" "(?.*)" "(?.*)" "(?.*)" "(?.*)" "(?.*)"' } } geoip { source => "clientip" database => "/etc/logstash/GeoLite2-City/GeoLite2-City.mmdb" } if [request] { ruby { init=> "@kname = ['uri','url_args']" code => " new_event = LogStash::Event.new(Hash[@kname.zip(event.get('request').split('?'))]) new_event.remove('@timestamp') event.append(new_event) " } }# date {# match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]# target => "@timestamp"# }# mutate {# remove_field => ["fields"]## }}output { stdout { codec => rubydebug } elasticsearch { hosts => ["elastic1.ops.test.com:9200"] index => "tengine-access-%{+YYYY.MM.dd}" user => "user" password => "password" template_overwrite => true }}

三、logstash安装geoip

GeoLite2 数据库是一个免费的 IP 地理定位数据库，与 MaxMind 的 GeoIP2 数据库有一定可比性，但不如后者准确。GeoLite2 Country 与 City 数据库在每月的第一个周二更新。GeoLite2 ASN 数据库的更新时间为每周二。

IP 地理定位在本质上缺乏准确性。地点通常接近人口的中心。 由 GeoIP 数据库提供的任何地点均不应用来确认某个具体的地址或住户。

针对我们为某个 IP 地址返回的经纬度坐标，用准确性半径作为其地理定位准确度的标识。IP 地址的实际地点很可能位于该半径和经纬度坐标界定的区域内。

四、logstash管理脚本

1、使用脚本管理服务

#!/bin/bashNAME=$1DATE=$(date +'%Y-%m-%d')BASEDIR=/opt/ytd_soft/logstashCONF=$NAME.confLOG=/opt/ytd_logs/logstash/$NAME.$DATE.logPROG=logstashOPTS="-w 5 -f ./config/$NAME.conf"#PSIGN="_kkkkk${NAME}kkkkk_"echo "config file : ./config/$NAME.conf"if [ ! -f "./config/$NAME.conf" ];then echo "ERROR: config file not exist ..." exit 1ficd $BASEDIRfunction do_status(){ PNUM=$(ps aux |grep $NAME |grep -v grep |awk '{print $2}') echo "process pid: $PNUM" if [ "$PNUM"x = "x" ];then echo "$PROG is stopped" else echo "$PROG is running" fi}function do_start(){ nohup ./bin/$PROG $OPTS &>$LOG &}function do_stop(){ PNUM=$(ps aux |grep $NAME |grep -v grep |awk '{print $2}') echo "process pid: $PNUM" kill -9 $PNUM}function do_restart(){ do_stop sleep 1 do_start}case $1 instatus) do_start ;;start) do_start ;;stop) do_stop ;;restart) do_restart ;;*) echo 'invalid arguments ...' ;;esacexit 0

logstash_run.sh

2、可以使用supervisor进行管理

# cat /etc/supervisor/conf.d/elk.conf[program:logstash-tengine1]user=rootenvironment=JAVA_HOME=/home/d/java/defaultcommand=/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/logstash-tengine1.conf --path.data=/data/logstash/tengine/tengine1directory=/autostart=Trueautorestart=Trueredirect_stderr=Truestopsignal=INTstopasgroup=True

logstash.conf

logstash6.3使用supervisor启动报错解决办法如下

​​ dotfile = File.join(ENV['HOME'].to_s, '.aprc')+ dotfile = File.join(ENV['HOME'], '.aprc')if dotfile_readable?(dotfile) end@@ -161,7 +161,7 @@ def merge_custom_defaults! load_dotfile merge_options!(AwesomePrint.defaults) if AwesomePrint.defaults.is_a?(Hash) rescue =>- $stderr.puts "Could not load #{dotfile}: #{e}"+ $stderr.puts "Could not load '~/.aprc': #{e}" end endend

五、logstash深入学习

1、logstash6.x多实例启动（指定--path.data=/data/logstash/logstash-tengine）

报错如下

Logstash could not be started because there is already another instance using the configured data directory. If you wish to run multiple instances, you must change the "path.data" setting.

新版本启动方式：

其中，--path.data是指存放数据的路径

${LOGSTASH_HOME}/bin/logstash -f logstash-tengine.conf --path.data=/data/logstash/logstash-tengine

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。