ElasticSearch实战系列七: Logstash实战使用-图文讲解

时间:2022-12-17 16:19:30

前言

在上一篇中我们介绍了Logstash快速入门,本文主要介绍的是ELK日志系统中的Logstash的实战使用。实战使用我打算从以下的几个场景来进行讲解。

时区问题解决方案

在我们使用logstash将采集的数据传输到ES中的时候,会发现采集的时间@timestamp的时间和我们本地的不一致,这个主要是因为时区的问题导致的,我们在计算时间的时候需要将这个时间增加8小时,但是这样会很不方便。为了永久解决这个问题,我们可以在logstash中的filter中对该字段进行转换,增加8小时。

添加的配置如下:


ruby {
code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
}
ruby {
code => "event.set('@timestamp',event.get('timestamp'))"
}
mutate {
remove_field => ["timestamp"]
}

原本示例图:

ElasticSearch实战系列七: Logstash实战使用-图文讲解

添加配置之后的示例图:

ElasticSearch实战系列七: Logstash实战使用-图文讲解

可以看到添加配置之后@timestamp时间已经和本地时间基本一致了。

日志内容切分

我们在进行采集日志到ES中的时候,有时需要对日志内容进行切割。比如得到日志内容的时间以及日志级别等等。这时我们就可以通过grok来对日志内容进行切分,比如将制定好的日志内容切割为日志时间、线程名称、日志级别、类名以及详细内容等等。我们只需要在logstash的filter中使用grok语法即可完成日志内容切割。

这里我们使用JAVA的Logback来制定日志输出格式,然后通过日志的格式编写grok语法,最后将grok配置添加到logstash的filter中。

Logback输出配置:

|%d{yyyy-MM-dd HH:mm:ss.SSS}|[%thread]|%-5level|%logger{50}|-%msg%n

日志样例数据:

|2020-07-24 17:08:33.159|[Thread-5]|INFO|com.pancm.Application|-测试示例三: All things in their being are good for something. 天生我才必有用3

grok模式:

|%{DATA:log_time}|%{DATA:thread}|%{DATA:log_level}|%{DATA:class_name}|-%{GREEDYDATA:content}

使用grok分析

ElasticSearch实战系列七: Logstash实战使用-图文讲解

可以看到以及分析匹配成功了。

然后我们在filter中添加如下配置:

grok {

match => { "message" =>"|%{DATA:log_time}|%{DATA:thread}|%{DATA:log_level}|%{DATA:class_name}|-%{GREEDYDATA:content}"

}

}

最终输出的日志到ES的示例图:

ElasticSearch实战系列七: Logstash实战使用-图文讲解

自定义模板

我们在使用Logstash采集日志的时候,如果没有指定索引库或模板,则会使用ElasticSearch默认自带的名字为”logstash”的模板,默认应用于Logstash写入数据到ElasticSearch使用。但是我们希望使用自定义的索引模板,将采集的日志按照我们自身的想法来写入,此时我们就需要用到自定义模板了。

主要有两种方式,一种是在logstash的output插件中使用template指定本机器上的一个模板json路径, 例如 template => "/home/logstash.json",json里面的内容为我们自定的索引mapping,虽然这种方式简单,但是分散在Logstash机器上,维护起来比较麻烦。还有一种是在elasticsearc服务端自定义配置模板,事先将模板设置好,然后在logstash的output输出中指定该模板即可,这种方式比较灵活方便,可动态更改,全局生效。

这里我们还是通过一个示例来进行说明,我们首先创建一个template_mylog的模板,配置这几个字段:

log_time、thread、log_level、class_name、content。

语句如下:


PUT _template/template_mylog
{
"index_patterns" : [
"mylog-*"
],
"order" : 10,
"settings": {
"index.number_of_shards": 3,
"number_of_replicas": 1
},
"mappings" : {
"properties" : {
"log_level" : { "type" : "keyword" },
"thread" : { "type" : "keyword" },
"class_name" : { "type" : "keyword" },
"content" : { "type" : "keyword" },
"log_time" : { "type" : "date","format" : "yyyy-MM-dd HH:mm:ss.SSS"}
} }
}

示例图:

ElasticSearch实战系列七: Logstash实战使用-图文讲解

注:上述的配置比其他mapping而言多了两个新配置,一个是index_patterns,该配置表明自动创建的索引开头以mylog-的索引库都会采用该模板;而order表示顺序级别,在有相同的索引模板中,该值越大,优先级越高。

创建成功之后,我们只需在output中的添加如下配置即可。


elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "mylog-%{+YYYY.MM.dd}"
}

然后我们启动logstash进行日志的采集。

效果图:

ElasticSearch实战系列七: Logstash实战使用-图文讲解

写入多个索引库

我们在使用logstash采集日志的时候,有时有多种不同的日志并且需要采集到不同的索引库中,这时我们就可以通过标记来进行写入。比如采集/home/logs目录下的日志我定义一个标记为java,采集/home/logs2目录下的日志我定义一个标记为java2,那么在写入ElasticSearch的时候只需要根据该标记区分写入即可。

logstash input配置示例:


file {
path => ["/home/logs/mylog-2020-08-13.0.txt"]
type => "java"
start_position => "beginning"
sincedb_path => "/dev/null"
}
file {
path => ["/home/logs2/*.txt"]
type => "java2"
start_position => "beginning"
sincedb_path => "/dev/null"
}

logstash output配置示例:


if [type] == "java"{
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "mylog-%{+YYYY.MM.dd}"
}
} if [type] == "java2"{
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "mylog-%{+YYYY.MM}"
}
}

示例图在多行内容合并场景中。

多行内容合并

我们在采集日志的时候,经常会遇到异常日志,并且异常日志并非为一行内容,如果我们按照原有的方式采集,在ElasticSearch中显示的是一行一行的内容,这样的话我们排查问题会很头疼。幸好Logstash中支持多行日志合并,使用multiline.pattern、multiline.negate和multiline.what来实现配置实现。

下面的配置中,我们通过制定匹配规则将以空格开头的所有行合并到上一行,并把以Caused by开头的也追加到上一行。

在Logstash的input配置中添加如下配置:


codec => multiline {
pattern => "\s*\["
negate => "true"
what => "previous"
}

异常日志:

ElasticSearch实战系列七: Logstash实战使用-图文讲解

原异常日志在ElasticSearch中示例图:

ElasticSearch实战系列七: Logstash实战使用-图文讲解

多行合并之后的效果图:

ElasticSearch实战系列七: Logstash实战使用-图文讲解

完整配置

logstash-test.conf 配置

input{
file {
path => ["/home/logs/mylog-2020-08-13.0.txt"]
type => "java"
start_position => "beginning"
sincedb_path => "/dev/null"
}
file {
path => ["/home/logs2/*.txt"]
type => "java2"
codec => multiline {
pattern => "\s*\["
negate => "true"
what => "previous"
}
start_position => "beginning"
sincedb_path => "/dev/null"
}
} filter { grok {
match => { "message" =>"\|%{DATA:log_time}\|%{DATA:thread}\|%{DATA:log_level}\|%{DATA:class_name}\|-%{GREEDYDATA:content}" }
} ruby {
code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
}
ruby {
code => "event.set('@timestamp',event.get('timestamp'))"
}
mutate {
remove_field => ["timestamp"]
}
} output {
stdout {
codec => rubydebug
}
if [type] == "java"{
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "mylog-%{+YYYY.MM.dd}"
}
} if [type] == "java2"{
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "mylog-%{+YYYY.MM}"
}
} }

异常问题解决方案

1.logstash: Could not execute action: PipelineAction::Create, action_result: false

解决办法: 斜杆采用“/”

2, logstash: object mapping for [host] tried to parse field [host] as object, but found a concrete value

解决办法: 在filter里面添加如下配置:

 mutate {
rename => { "host" => "host.name" }
}

其它

ElasticSearch实战系列:

音乐推荐

原创不易,如果感觉不错,希望给个推荐!您的支持是我写作的最大动力!

版权声明:

作者:虚无境

博客园出处:http://www.cnblogs.com/xuwujing

CSDN出处:http://blog.csdn.net/qazwsxpcm    

个人博客出处:http://www.panchengming.com

ElasticSearch实战系列七: Logstash实战使用-图文讲解的更多相关文章

  1. ElasticSearch实战系列六: Logstash快速入门和实战

    前言 本文主要介绍的是ELK日志系统中的Logstash快速入门和实战 ELK介绍 ELK是三个开源软件的缩写,分别表示:Elasticsearch , Logstash, Kibana , 它们都是 ...

  2. MP实战系列(七)之集成springboot

    springboot是现在比较流行的微服使用的框架,springboot本质上就是将spring+springmvc+mybatis零配置化,基本上springboot的默认配置符合我们的开发.当然有 ...

  3. ATT&CK实战系列 红队实战(一)————环境搭建

    首先感谢红日安全团队分享的靶机实战环境.红队实战系列,主要以真实企业环境为实例搭建一系列靶场,通过练习.视频教程.博客三位一体学习. 靶机下载地址:http://vulnstack.qiyuanxue ...

  4. memcached实战系列(七)理解Memcached的数据过期方式、新建过程、查找过程

    1.1.1. 新建Item分配内存过程 1:快速定位slab classid,先计算Item长度 key键长+flag+suffix(16字节)+value值长+结构大小(32字节),如90byte ...

  5. shiro实战系列(七)之Realm

    Realm 是一个能够访问应用程序特定的安全数据(如用户.角色及权限)的组件.Realm 将应用程序特定的数据转 换成一种 Shiro 能够理解的格式,这样 Shiro 能够提供一个单一的易理解的 S ...

  6. ElasticSearch优化系列七:优化建议

    尽量运行在Sun/Oracle JDK1.7以上环境中,低版本的jdk容易出现莫名的bug,ES性能体现在在分布式计算中,一个节点是不足以测试出其性能,一个生产系统至少在三个节点以上. ES集群节点规 ...

  7. ATT&CK实战系列——红队实战(一)

    一.环境搭建 1.环境搭建测试 最近想要开始学习内网渗透,搜集了一些教程,准备先实验一个vulnstack靶机,熟悉一下内网渗透操作再学习基础知识. 靶场下载地址:http://vulnstack.q ...

  8. 【红日安全-VulnStack】ATT&CK实战系列——红队实战(二)

    一.环境搭建 1.1 靶场下载 靶场下载地址:http://vulnstack.qiyuanxuetang.net/vuln/detail/3/ 靶机通用密码:  1qaz@WSX 1.2 环境配置 ...

  9. ATT&CK实战系列——红队实战(二)

    一.环境搭建 靶场下载地址:http://vulnstack.qiyuanxuetang.net/vuln/detail/3/ DC IP:10.10.10.10OS:Windows 2012应用:A ...

随机推荐

  1. Linq 单表城市级联

    var list = (from province in db.Areas && province.IsDel == join city in db.Areas on province ...

  2. C#设计模式——外观模式(Facade Pattern)

    一.概述 在系统设计中,某一个系统可能非常庞大,用户要使用该系统就不得不掌握大量的接口,造成使用的不便.这时可以考虑将该系统细分成一系列子系统并使子系统间的耦合降到最低,利用外观模式提供一个外观对象, ...

  3. HDU 1003 Max Sum(AC代码)

    #include <stdio.h> int main(){ int i,t,j,n,x; int start,end,temp,max,sum; scanf("%d" ...

  4. varnish状态引擎2

    如何让varnish支持虚拟主机: if (req.http.host = "www.nihao.com") { } 强制对某资源的请求,不检查缓存: 上图表示以/test1.ht ...

  5. Permutation Test 置换检验&lpar;转)

    Permutation Test 置换检验 显著性检验通常可以告诉我们一个观测值是否是有效的,例如检测两组样本均值差异的假设检验可以告诉我们这两组样本的均值是否相等(或者那个均值更大).我们在实验中经 ...

  6. 【转】 Android中退出程序的提示框

    原文网址:http://blog.csdn.net/jumping_android/article/details/7571309 @Override public boolean onKeyDown ...

  7. Eclipse将引用了第三方jar包的Java项目打包成jar文件

    第一步:建议手动 Eclipse插件fatjar 安装方法:1:下载地址:http://downloads.sourceforge.net/fjep/net.sf.fjep.fatjar_0.0.27 ...

  8. 如何在Linux中使用命令行卸载软件

    您可以使用“dpkg”命令来查看您的计算机,按“Ctrl + Alt + T”的所有已安装包的列表,打开一个终端窗口. 在提示符下键入以下命令,然后按Enter键.dpkg -- list 要卸载程序 ...

  9. 记录&period;net使用ueditor富文本编辑器

    UEditor是什么 最近在在项目的时候使用到了百度的富文本编辑器.官网有详细的操作流程文档.这里我只是记录项目中常用到的一些事件.以便日后可以方便查询. UEditor是百度的一个javascrip ...

  10. SVN创建分支主干策略

    本篇目录 前言 SVN分支管理策略 VisualSVN Server TortoiseSVN客户端 Repository的创建 Check out trunk创建新项目MyProject trunk更 ...