Docker安装Logstash，并结合logback实现ELK日志收集

2024-08-23 15:12
文章标签 实现 安装 日志 docker 结合 elk 收集 logback logstash

本文主要是介绍Docker安装Logstash，并结合logback实现ELK日志收集，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

拉取镜像

docker pull docker.elastic.co/logstash/logstash:8.14.3

创建文件夹

mkdir /mnt/data/logstash

创建默认文件

先不做目录挂载，run出一个容器

docker run -d --rm -it docker.elastic.co/logstash/logstash:8.14.3

将config和pipeline从容器cp到宿主机

docker cp confident_driscoll:/usr/share/logstash/config /mnt/data/logstash/docker cp confident_driscoll:/usr/share/logstash/pipeline /mnt/data/logstash/

停止容器(启动时配置了 --rm ，停止后会自动删除)

docker stop confident_driscoll

创建容器

docker run --name logstash \
--network es-net \
-p 5044:5044 \
-v /mnt/data/logstash/pipeline/:/usr/share/logstash/pipeline/ \
-v /mnt/data/logstash/config/:/usr/share/logstash/config/ \
--restart=always \
-d docker.elastic.co/logstash/logstash:8.14.3

修改配置文件

修改挂载出来的默认 logstash.yml 文件，这里位于 /mnt/data/logstash/config/ 目录，修改后配置如下：

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://es:9200" ]
xpack.monitoring.elasticsearch.username: "elastic"
xpack.monitoring.elasticsearch.password: "123456"

注意：这里明文的用户民和密码安全性较差，可改用使用 API 密钥的方式。

修改默认的 logstash.conf 文件，创建日志处理规则，位于 /mnt/data/logstash/pipeline/ 目录，修改后如下：

input {  tcp {  port => 5044  codec => json_lines  }  
}  output {  elasticsearch {  hosts => ["es:9200"]  index => "iov-%{+YYYY.MM.dd}"  document_type => "_doc"user => "elastic"password => "123456"  }
}

注：上面的各个属性值，根据实际环境进行配置。
修改后重启 logstash 服务。

配置logback

引入相关依赖包，这里使用的 gradle ，maven同理，如下：

// Logback Classicimplementation 'ch.qos.logback:logback-classic:1.2.3' // 使用适合你项目的版本// Logstash Logback Encoderimplementation 'net.logstash.logback:logstash-logback-encoder:6.6' // 使用适合你项目的版本

修改 logback-spring.xml 文件，若不存在则创建。位于 resources 下（这里是springboot项目），内容如下：

<configuration><appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender"><destination>192.168.1.110:5044</destination> <!-- Logstash的TCP端口 --><encoder class="net.logstash.logback.encoder.LogstashEncoder"/></appender><root level="info"><appender-ref ref="LOGSTASH" /><!-- 可以添加其他appender，如控制台或文件appender --></root>
</configuration>

根据实际情况将上面的192.168.1.110:5044部分改成真实 logstash 的IP和端口。

验证功能

确认 ES 和 Logstash 已经正常启动，然后启动 spring boot 服务。进入 Kibana 可以看到已经有日志文件写入到es中。如下：
image.png

格式化日志

未处理前日志信息统一在message字段中，不方便我们管理。
这里主要针对设备的上传的原始数据进行格式化。
比如iov中的日志格式如下：

// 正常的设备上报信息，用imei区分
logger.warn("IMEI: {},DATA: {}", imei, ByteBufUtil.hexDump(rawMessage));// 打印的错误日志
logger.error("测试报错日志");// 系统报错
throw new DecoderException();

修改logstash.conf

为了更方便管理日志，将 logstash.conf 修改如下：

input {tcp {port => 5044codec => json_lines}
}filter {# 使用grok过滤器来从message字段中提取imei和datagrok {match => { "message" => "IMEI: %{WORD:imei},DATA: %{GREEDYDATA:data}" }# 注意：如果imei可能包含特殊字符（如空格、冒号等），您可能需要调整WORD为正则表达式模式# 例如：(?<imei>[^,]+) 来匹配从IMEI:到逗号之前的所有内容tag_on_failure => ["_grok_failed"] # 当grok失败时添加此标签}# 检查grok是否成功以及level字段的值if "_grok_failed" in [tags] and [level] == "WARN" {# 如果grok失败且level为WARN，则丢弃事件（可以通过drop过滤器或在输出阶段排除）drop {}} else if [level] in ["ERROR", "FATAL", "SEVERE"] and "_grok_failed" in [tags] {# 如果level是ERROR及以上，但grok失败或没有imei和data，则设置imei为"error"，data为messagemutate {add_field => { "imei" => "error" }add_field => { "data" => "%{message}" }remove_field => ["tags"]}}# ruby {#  code => "#   require 'time'#    event.set('datetime', event.get('@timestamp').time.localtime().strftime('%Y-%m-%d %H:%M:%S'))#  "# }# 使用mutate过滤器来移除不需要的字段mutate {remove_field => ["level_value", "@version", "message"]}}output {elasticsearch {hosts => ["es:9200"]index => "iov-%{+YYYY.MM.dd}"document_type => "_doc"user => "elastic"password => "123456"}
}

具体更改内容，看上面的注解。
更改后重启logstash服务。

添加es动态模板

根据上面的 logstash.conf 配置文件，动态生成的索引。数据结构除了 @timestamp 字段为date类型，其他的都将默认声明为 text 类型，即都将分词。如下：

"iov-2024.07.30": {"aliases": {},"mappings": {"properties": {"@timestamp": {"type": "date"},"data": {"type": "text","fields": {"keyword": {"type": "keyword","ignore_above": 256}}},"imei": {"type": "text","fields": {"keyword": {"type": "keyword","ignore_above": 256}}},"level": {"type": "text","fields": {"keyword": {"type": "keyword","ignore_above": 256}}},"logger_name": {"type": "text","fields": {"keyword": {"type": "keyword","ignore_above": 256}}},"thread_name": {"type": "text","fields": {"keyword": {"type": "keyword","ignore_above": 256}}}}},"settings": {"index": {"routing": {"allocation": {"include": {"_tier_preference": "data_content"}}},"number_of_shards": "1","provided_name": "iov-2024.07.30","creation_date": "1722317697099","number_of_replicas": "1","uuid": "5OoDo8eITpqdhBMKPMBaeg","version": {"created": "8505000"}}}}
}

这会导致浪费大量的空间和性能，比如我们的imei、logger_name、thread_name和level字段，完全没有必要进行分词，精准查找即可，所以现在我们需要在logstash上传日志到es时，精准的建立字段的类型。这里采用es动态索引模板的方式。
使用 Kibana 开发者工具执行如下指令，创建索引模板：

PUT /_index_template/iov_template  
{  "index_patterns": ["iov-*"],   "template": {  "mappings": {  "properties": {  "@timestamp": {  "type": "date"  },  "data": {  "type": "text" ,"fields": {  "keyword": {  "type": "keyword",  "ignore_above": 256  }  }  },"imei": {  "type": "keyword"  },  "level": {  "type": "keyword"  },  "logger_name": {  "type": "keyword"  },  "thread_name": {  "type": "keyword"  }  }  }}  
}

这个模板将自动应用于任何以iov-开头的新索引。当让适用于以iov-开头，后跟日期（如iov-2024.07.30）的索引名称。如果字段名称可以和模板对应上，那么字段类型将跟模板一致。
创建模板后，再次测试日志插入，查看类型，发现模板已经生效：

{"iov-2024.07.30": {"aliases": {},"mappings": {"properties": {"@timestamp": {"type": "date"},"data": {"type": "text","fields": {"keyword": {"type": "keyword","ignore_above": 256}}},"imei": {"type": "keyword"},"level": {"type": "keyword"},"logger_name": {"type": "keyword"},"thread_name": {"type": "keyword"}}},"settings": {"index": {"routing": {"allocation": {"include": {"_tier_preference": "data_content"}}},"number_of_shards": "1","provided_name": "iov-2024.07.30","creation_date": "1722320800301","number_of_replicas": "1","uuid": "pGrsRAN4Rg6HsP2mmhrGpA","version": {"created": "8505000"}}}}
}

好了，现在日志数据已经有了需要的字段，并且正确的设置了字段类型。

设置索引声明周期

创建ILM配置文件：

PUT _ilm/policy/log_retention_policy  
{  "policy": {  "phases": {  "hot": {  "min_age": "0ms",  "actions": {  "rollover": {  "max_age": "2d",  // 这里设置为2天，但实际上在单机模式下，你可能需要外部脚本来控制滚动  "max_size": "30gb"  // 你可以设置一个足够大的大小限制，以避免基于大小的滚动  }  }  },  "delete": {  "min_age": "7d",  "actions": {  "delete": {}  }  }  }  }  
}

这个策略将包含两个阶段：hot 和 delete。hot 阶段将不执行任何特殊操作（除了可能的滚动），而 delete 阶段将在索引达到7天时删除它。
注意：在单机模式下，rollover 操作可能不会自动按 max_age 触发，因为Elasticsearch的ILM滚动通常依赖于索引的大小或索引中事件的时间戳（如果配置了max_docs_per_partition）。在这种情况下，你可能需要依赖外部脚本来控制索引的滚动。

修改索引模板

修改索引模板，使用上面定义的ILM策略。修改后如下：

PUT /_index_template/iov_template  
{  "index_patterns": ["iov-*"],   "template": {  "mappings": {  "properties": {  "@timestamp": {  "type": "date"  },  "data": {  "type": "text" ,"fields": {  "keyword": {  "type": "keyword",  "ignore_above": 256  }  }  },"imei": {  "type": "keyword"  },  "level": {  "type": "keyword"  },  "logger_name": {  "type": "keyword"  },  "thread_name": {  "type": "keyword"  }  }  },"settings": {  "number_of_shards": 1,  "number_of_replicas": 1,  "index.lifecycle.name": "log_retention_policy",  "index.lifecycle.rollover_alias": "iov_logs"  }}  
}

在索引模板中，指定了之前定义的ILM策略，并设置一个别名用于索引滚动。

查看日志数据

使用Kibana开发者工具

在 Kibana 开发者工具中执行如下语句可以查看到对应日志记录：

GET /iov-2024.07.30/_search

查询结果如下：

{"took": 1,"timed_out": false,"_shards": {"total": 1,"successful": 1,"skipped": 0,"failed": 0},"hits": {"total": {"value": 3,"relation": "eq"},"max_score": 1,"hits": [{"_index": "iov-2024.07.30","_id": "Lo9RApEBFb2jjdrpyxet","_score": 1,"_ignored": ["data.keyword"],"_source": {"data": """io.netty.handler.codec.DecoderExceptionat com.thgcgps.iovservice.network.codec.MsgFrameDecode.decode(MsgFrameDecode.java:109)at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:493)at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:432)at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:271)at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:377)at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:355)at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:377)at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:714)at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650)at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576)at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)at java.base/java.lang.Thread.run(Thread.java:829)
""","thread_name": "nioEventLoopGroup-5-1","@timestamp": "2024-07-30T06:26:38.415Z","level": "ERROR","logger_name": "com.thgcgps.iovservice.network.server.TcpServerHandler","imei": "error"}},{"_index": "iov-2024.07.30","_id": "L49RApEBFb2jjdrpyxet","_score": 1,"_source": {"data": "测试报错日志","thread_name": "nioEventLoopGroup-5-1","@timestamp": "2024-07-30T06:26:38.413Z","level": "ERROR","logger_name": "com.thgcgps.iovservice.network.codec.MsgFrameDecode","imei": "error"}},{"_index": "iov-2024.07.30","_id": "LY9RApEBFb2jjdrpyxet","_score": 1,"_source": {"data": "78780a1344050400020035c60e0d0a","thread_name": "nioEventLoopGroup-5-1","@timestamp": "2024-07-30T06:26:38.412Z","level": "WARN","logger_name": "com.thgcgps.iovservice.network.codec.MsgFrameDecode","imei": "0868755128633153"}}]}
}

可以看到，这里的数据即我们代码中测试的相关日志，即系统报错，打印的error日志和最关注的设备上报日志。但这里的结果并不方便观察和整理。并且时间格式也不理想，时区也并不是北京时间。

使用Kibana的Discover功能

进入 Kibana 管理界面，点击 Discover ，如图：
image.png

进入后创建一个数据视图：
image.png

创建时，输入名称，和索引模式即可：
image.png
这里的索引模式需要和匹配的源对应，支持通配符，上图即使用 * 通配符创建。填写好后，点击使用而不保存（即临时的数据视图，刷新页面或跳转页面后就销毁了）或者 点击保存数据视图到Kibana（下次进入可以直接选择无需重建）。这里因为日志日期会有变动，创建了一个临时的。

创建好的数据视图如下：
image.png

点击“可用字段”后的“+”号，即可查看指定字段。如图：
image.png
“选中字段”即当前产看的字段。

若想搜索指定的值，可在搜索框中使用KQL语法搜索数据（光标点击搜素框后即有提示），如图：
image.png
因为imei字段之前设置了字段类型，控制其不分词，所以当使用imei字段时，要精确搜索。而这里的data字段支持全文搜索，如下：
image.png
注意：这里因为设置模板时，没有指定分词器，所以当前时默认的 standard 分词器。分词效果并不好由于数据上报的日志由数字和字母组成，也不是很适合分词。可以考虑data也不进行分词。或者更改一下数据上报的日志，比如区分出心跳和定位等信息，然后即可根据特定类型进行查找。

上面的 @timestamp 字段，日期时间虽然是对的，但并不符合我们常见的表达方式，这里可以设置其格式以达到预期效果，点击该字段，选择“编辑数据视图字段”，如下：
image.png
打开“设置格式”选项，格式选择“日期”，然后根据指定规则，输入如下格式 “yy-MM-DD HH:mm:ss”,然后点击保存，如下：
image.png
保存后时间展示如下：
image.png

若需要导出，可点击“共享”，然后选择“CSV Reports”
image.png
然后点击 “Generate CSV”即可，然后右下角会弹出一个快速访问Reports的连接，点击即可跳转到下载页面
image.png
也可以点击“Management”中的“Stack Management”功能如下
image.png
然后点击“Reporting”即可看到可下载的CSV文件。
image.png

**

这篇关于Docker安装Logstash，并结合logback实现ELK日志收集的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1099735。

相关文章

C#借助Spire.XLS for .NET实现在Excel中添加文档属性

C#借助Spire.XLS for .NET实现在Excel中添加文档属性

《C#借助Spire.XLSfor.NET实现在Excel中添加文档属性》在日常的数据处理和项目管理中,Excel文档扮演着举足轻重的角色,本文将深入探讨如何在C#中借助强大的第三方库Spire.... 目录为什么需要程序化添加Excel文档属性使用Spire.XLS for .NET库实现文档属性管理Sp
阅读更多...
Python+FFmpeg实现视频自动化处理的完整指南

Python+FFmpeg实现视频自动化处理的完整指南

《Python+FFmpeg实现视频自动化处理的完整指南》本文总结了一套在Python中使用subprocess.run调用FFmpeg进行视频自动化处理的解决方案,涵盖了跨平台硬件加速、中间素材处理... 目录一、 跨平台硬件加速：统一接口设计1. 核心映射逻辑2. python 实现代码二、 中间素材处
阅读更多...
Java数组动态扩容的实现示例

Java数组动态扩容的实现示例

《Java数组动态扩容的实现示例》本文主要介绍了Java数组动态扩容的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录1 问题2 方法3 结语1 问题实现动态的给数组添加元素效果，实现对数组扩容，原始数组使用静态分配
阅读更多...
Python实现快速扫描目标主机的开放端口和服务

Python实现快速扫描目标主机的开放端口和服务

《Python实现快速扫描目标主机的开放端口和服务》这篇文章主要为大家详细介绍了如何使用Python编写一个功能强大的端口扫描器脚本,实现快速扫描目标主机的开放端口和服务,感兴趣的小伙伴可以了解下... 目录功能介绍场景应用1. 网络安全审计2. 系统管理维护3. 网络故障排查4. 合规性检查报错处理1.
阅读更多...
Python轻松实现Word到Markdown的转换

Python轻松实现Word到Markdown的转换

《Python轻松实现Word到Markdown的转换》在文档管理、内容发布等场景中,将Word转换为Markdown格式是常见需求,本文将介绍如何使用FreeSpire.DocforPython实现... 目录一、工具简介二、核心转换实现1. 基础单文件转换2. 批量转换Word文件三、工具特性分析优点局
阅读更多...
Python中4大日志记录库比较的终极PK

Python中4大日志记录库比较的终极PK

《Python中4大日志记录库比较的终极PK》日志记录框架是一种工具,可帮助您标准化应用程序中的日志记录过程,：本文主要介绍Python中4大日志记录库比较的相关资料,文中通过代码介绍的非常详细,... 目录一、logging库1、优点2、缺点二、LogAid库三、Loguru库四、Structlogphp
阅读更多...
Springboot3统一返回类设计全过程(从问题到实现)

Springboot3统一返回类设计全过程(从问题到实现)

《Springboot3统一返回类设计全过程(从问题到实现)》文章介绍了如何在SpringBoot3中设计一个统一返回类,以实现前后端接口返回格式的一致性,该类包含状态码、描述信息、业务数据和时间戳,... 目录Spring Boot 3 统一返回类设计：从问题到实现一、核心需求：统一返回类要解决什么问题？
阅读更多...
Java使用Spire.Doc for Java实现Word自动化插入图片

Java使用Spire.Doc for Java实现Word自动化插入图片

《Java使用Spire.DocforJava实现Word自动化插入图片》在日常工作中,Word文档是不可或缺的工具,而图片作为信息传达的重要载体,其在文档中的插入与布局显得尤为关键,下面我们就来... 目录1. Spire.Doc for Java库介绍与安装2. 使用特定的环绕方式插入图片3. 在指定位
阅读更多...
springboot中配置logback-spring.xml的方法

springboot中配置logback-spring.xml的方法

《springboot中配置logback-spring.xml的方法》文章介绍了如何在SpringBoot项目中配置logback-spring.xml文件来进行日志管理,包括如何定义日志输出方式、... 目录一、在src/main/resources目录下，也就是在classpath路径下创建logba
阅读更多...
Java使用Spire.Barcode for Java实现条形码生成与识别

Java使用Spire.Barcode for Java实现条形码生成与识别

《Java使用Spire.BarcodeforJava实现条形码生成与识别》在现代商业和技术领域,条形码无处不在,本教程将引导您深入了解如何在您的Java项目中利用Spire.Barcodefor... 目录1. Spire.Barcode for Java 简介与环境配置2. 使用 Spire.Barco
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2