日志审计Graylog 使用教程-kafka收取消息
Graylog 是一个开源的日志管理和分析平台。它重要用于网络、存储、处理和分析大量的日志数据。Graylog 的焦点功能包括：[*]日志网络：Graylog 可以从各种来源（如服务器、网络装备、应用程序等）网络日志数据。它支持多种输入方式，包括 Syslog、GELF、Beats、Kafka 等。
[*]日志存储：Graylog 使用 Elasticsearch 作为其后端存储体系，因此它可以高效地存储和检索大量的日志数据。
[*]日志处理和分析：Graylog 提供了强大的搜索和分析功能，用户可以通过查询语言对日志举行过滤、排序、聚合等操作，资助发现问题或监控体系状态。
[*]警报和通知：Graylog 可以根据预设条件触发警报，并通过电子邮件、Slack 等渠道发送通知。
[*]可视化和仪表板：用户可以在 Graylog 中创建可视化的仪表板，用于及时监控和分析日志数据。
Graylog 常用于 IT 运维、安全监控、故障排查等场景，通过对日志数据的集中管理和分析，资助企业进步体系的可观测性和问题办理能力。
一、日志网络
1.1、配置日志服务器
https://i-blog.csdnimg.cn/direct/886832a7e2d44c399ca5d72f7809ae31.png
选择inputhttps://i-blog.csdnimg.cn/direct/f98cb090af994630988f9ac4080936d2.png
这里给了很多日志接收方法https://i-blog.csdnimg.cn/direct/dc236265bd1b47c39d0e56bf11412248.png
试验使用kafka获取日志，https://i-blog.csdnimg.cn/direct/32736aabee9c496b9e0818c379b7c6a9.png
1.2、配置映射表
他作为动态数据获取，比如IP 地理位置等，根据日志某个字段的信息从XX库找到对应的数据https://i-blog.csdnimg.cn/direct/d748774fde9a421a9a4767b453a90db8.png
选择数据源https://i-blog.csdnimg.cn/direct/488548c19a3640a6aa3c88a21d0e56b8.png
创建链接源数据实例https://i-blog.csdnimg.cn/direct/40f0e638c8ec4e3b8fcbd6c15408283d.png
有很多差异格式的数据源https://i-blog.csdnimg.cn/direct/ae091ac230c8400eb158be6acf9e0c06.png
选择csv文件https://i-blog.csdnimg.cn/direct/2421e3f8b42d42129915bc6486e75a45.png
https://i-blog.csdnimg.cn/direct/3bd53a85f1ea4291a125ed50776df706.png
假如我们的内容为
username ,department,email
jdoe,IT,jdoe@example.com
asmith,HR,asmith@example.com
bwhite,Finance,bwhite@example.com
johndoe,jiushi,a.com
将内容编写为https://i-blog.csdnimg.cn/direct/4728bfe8f71e4333875e6351c9525c9d.png
不区分巨细写，更新https://i-blog.csdnimg.cn/direct/2afc7b2131a949ddba9bdd2b93cb4839.png
https://i-blog.csdnimg.cn/direct/2af89e33a804483799ec4fab8cd8864a.png
创建索引表实例https://i-blog.csdnimg.cn/direct/76cce5e2a1e04b37a92256ad730d71bc.png
选择刚刚创建的数据源实例并创建https://i-blog.csdnimg.cn/direct/30fe3a8073d2468daa593df6797ae615.png
1.3、创建字段提取器
创建，内容填写自己构建的kafka服务器地址和topichttps://i-blog.csdnimg.cn/direct/b7049c63fe364bcc8d3ada73268e0d9a.png
生存乐成，他会自己运行的，假如没有问题就会酿成RUNNING的状态https://i-blog.csdnimg.cn/direct/718c15daed624e9a8771bfbd9cc3e6bd.png
先生产一个消息，因为要配置字段提取器
生产内容
{ "version": "1.1", "host": "webserver.local", "short_message": "User login failed", "full_message": "User johndoe failed to login to the web application", "level": 3, "_username": "johndoe", "_src_ip": "192.168.1.2"}
检察消息https://i-blog.csdnimg.cn/direct/8225c3043b5146d8b1b31ce1944ce464.png
新产生的这个就是https://i-blog.csdnimg.cn/direct/281634282990494eb3ef3e10c68caffd.png
构造字段提取https://i-blog.csdnimg.cn/direct/caf5bab7eecb492e98a8101ec96a59a2.png
新增字段提取器https://i-blog.csdnimg.cn/direct/aed0fdfcd3be4c9ca8acbed598d8a69b.png
加载最近的消息https://i-blog.csdnimg.cn/direct/5b16e2dbc8a4412ea84ce8d53d439f7a.png
这里内容是kafka生产的消息，只作为日志，可以对每个字段配置提取。https://i-blog.csdnimg.cn/direct/7952a42a709a45fc9c2354c7a39ec5f0.png
将username 用于 Look UP配置https://i-blog.csdnimg.cn/direct/6d91cc42245c4c5e88f83fc466b7ba09.png
忘记设置映射表了。先在1,2小节创建映射表实例
选择映射表https://i-blog.csdnimg.cn/direct/d1bdce645c30412ebe7399e97f4fcde7.png
https://i-blog.csdnimg.cn/direct/db6208ce8ce744d690bc3c3992ca0c14.png
1.4、创建告警
1.4.1、创建上报
https://i-blog.csdnimg.cn/direct/ea445c66df7c46e3bfbc0b8c5d734ede.png
使用简朴的API上报https://i-blog.csdnimg.cn/direct/36ce025526664e588cf3b77404d7dde5.png
填充URL，并尝试发送测试数据https://i-blog.csdnimg.cn/direct/738c934cb07f4485907caf7ebaa05f1a.png
https://i-blog.csdnimg.cn/direct/30dfa0752fed4004b3c79e08adda3e76.png
提示乐成https://i-blog.csdnimg.cn/direct/f6f9b04433f7406a97a1ec0e9c273a04.png
检察api服务器https://i-blog.csdnimg.cn/direct/cd209d9068aa4d56ba54a340117cf26d.png
1.4.2、定义变乱
https://i-blog.csdnimg.cn/direct/c0fc49b5ca4e417182365fe56cb6fb47.png
https://i-blog.csdnimg.cn/direct/4da216f7be6647e4b21e5e4a305447e3.png
https://i-blog.csdnimg.cn/direct/b86de73da96f42b791963fcf3f7c6472.png
https://i-blog.csdnimg.cn/direct/377528a735274f31a470746e7ef3dc0f.png
https://i-blog.csdnimg.cn/direct/dcee1316f0a244bab1a3c04ffe80a2e2.png
https://i-blog.csdnimg.cn/direct/5f1b867225644554a09876d9e512cdbd.png
设置上报方式，刚才创建的API方式https://i-blog.csdnimg.cn/direct/f13eec5e318543f48f528180a26c5907.png
创建告警变乱完成
1.5、测试
1.5.1、产生关于登录失败的日志
{ "version": "1.1", "host": "webserver.local", "short_message": "User login failed", "full_message": "User johndoe failed to login to the web application", "level": 3, "_username": "johndoe", "_src_ip": "192.168.1.2"}
此中username的值为johndoe，记得之前映射表给定文件内容https://i-blog.csdnimg.cn/direct/10eb721ac8364eceb8e2a104670f171d.png
我们创建映射表时，通过username查找department内容，并添加到告警自定义字段中。https://i-blog.csdnimg.cn/direct/b4490176a98b4e2b983c6f7de481916c.png
产生的告警内容，我们自定义字段“renyuan”他的value通过映射表username关联到的。日志内容jdoe对应表格IT内容
附录
Input kafka差异格式示例
1. CEF Kafka
示例消息：
CEF:0|SecurityVendor|SecurityProduct|1.0|100|User Login|5|dvc=192.168.1.1 duser=johndoe src=10.0.0.1 spt=443
解释：
[*]CEF:0: CEF 版本号。
[*]SecurityVendor: 装备供应商名称。
[*]SecurityProduct: 装备产品名称。
[*]1.0: 装备版本。
[*]100: 变乱 ID。
[*]User Login: 变乱名称或描述。
[*]5: 严肃级别（1-10）。
[*]dvc=192.168.1.1: 装备 IP 地址。
[*]duser=johndoe: 目标用户名。
[*]src=10.0.0.1: 源 IP 地址。
[*]spt=443: 源端口。
2. GELF Kafka
示例消息：
{
"version": "1.1",
"host": "webserver.local",
"short_message": "User login failed",
"full_message": "User johndoe failed to login to the web application",
"level": 3,
"_username": "johndoe",
"_src_ip": "192.168.1.2"
}
解释：
[*]version: GELF 版本号。
[*]host: 主机名或 IP 地址。
[*]short_message: 简短的变乱描述。
[*]full_message: 变乱的详细描述。
[*]level: 日志级别（比方：3 表示错误）。
[*]_username 和 _src_ip: 自定义字段。
3. RAW Kafka
示例消息：
User johndoe failed to login from IP 192.168.1.2 via web application.
解释：
[*]RAW 格式是原始日志消息的简朴文本，没有布局化的字段或元数据。消息内容完全取决于发送方。
4. Syslog Kafka
示例消息：
<134>webserver.local User login failed: johndoe from 192.168.1.2
解释：
[*]<134>: Syslog 优先级，盘算方式为 (Facility * 8) + Severity。
[*]webserver.local: 发送日志的主机名。
[*]User login failed: johndoe from 192.168.1.2: 日志消息内容，描述发生的变乱。
这些示例展示了差异格式在 Kafka 中传输的方式，供 Graylog 等日志管理体系剖析和处理。
问题
1、接收到消息，但是没有输出https://i-blog.csdnimg.cn/direct/5c8f50ef5448479e876aa3b940c723e1.png
存在下行数据变动
检察体系日志tail -f /var/log/graylog-server/server.log &
root@a:/home/install/kafka_2.12-2.8.2# 2024-08-19T03:49:22.949Z ERROR Unable to decode raw message RawMessage{id=0593cd30-5dde-11ef-b23f-ac1f6b7e6aea, journalOffset=70, codec=CEF, payloadSize=9, timestamp=2024-08-19T03:49:22.947Z} on input <66c2a27b4c33336f079032fc>.
2024-08-19T03:49:22.949Z ERROR Error processing message RawMessage{id=0593cd30-5dde-11ef-b23f-ac1f6b7e6aea, journalOffset=70, codec=CEF, payloadSize=9, timestamp=2024-08-19T03:49:22.947Z}
java.lang.NullPointerException: null
at org.graylog.plugins.cef.parser.MappedMessage.<init>(MappedMessage.java:37) ~
at org.graylog.plugins.cef.codec.CEFCodec.decodeCEF(CEFCodec.java:128) ~
at org.graylog.plugins.cef.codec.CEFCodec.decode(CEFCodec.java:117) ~
at org.graylog2.shared.buffers.processors.DecodingProcessor.processMessage(DecodingProcessor.java:149) ~
at org.graylog2.shared.buffers.processors.DecodingProcessor.onEvent(DecodingProcessor.java:90)
at org.graylog2.shared.buffers.processors.ProcessBufferProcessor.onEvent(ProcessBufferProcessor.java:90)
at org.graylog2.shared.buffers.processors.ProcessBufferProcessor.onEvent(ProcessBufferProcessor.java:47)
at com.lmax.disruptor.WorkProcessor.run(WorkProcessor.java:143)
at com.codahale.metrics.InstrumentedThreadFactory$InstrumentedRunnable.run(InstrumentedThreadFactory.java:66)
at java.lang.Thread.run(Thread.java:830) [?:?]
创建了CEF格式解码器，缘故原由为kafka生产的消息不符及格式
示例CEF格式内容
CEF:0|Security|Graylog|1.0|100|Test event|5|src=10.0.0.1 dst=10.0.0.2 spt=1232
可以通过重新创建一个sysloghttps://i-blog.csdnimg.cn/direct/e03d4720bc34472bb1b1120282d37278.png
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]