如何使用Logstash搜集日志传输到es集群并使用kibana检测

引言：上一期我们进行了对Elasticsearch和kibana的部署，今天我们来解决如何使用Logstash搜集日志传输到es集群并使用kibana检测

目录

Logstash部署

1.安装配置Logstash

（1）安装

（2）测试文件

（3）配置

grok

1、手动输入日志数据

数据链路

2、手动输入数据，并存储到 es

数据链路

3、自定义日志1

数据链路

5、nginx access 日志

数据链路

6、nginx error日志

数据链路

7、filebate 传输给 logstash

filebeat 日志模板

Logstash部署

  • 服务器

安装软件主机名IP地址系统版本配置
LogstashElk10.12.153.71centos7.5.18042核4G

  • 软件版本：logstash-7.13.2.tar.gz

1.安装配置Logstash

Logstash运行同样依赖jdk，本次为节省资源，故将Logstash安装在了10.12.153.71节点。

（1）安装
tar zxf /usr/local/package/logstash-7.13.2.tar.gz -C /usr/local/
（2）测试文件

标准输入=>标准输出

1、启动logstash

2、logstash启动后，直接进行数据输入

3、logstash处理后，直接进行返回

input {stdin {}
}
output {stdout {codec => rubydebug}
}

标准输入=>标准输出及es集群

1、启动logstash

2、启动后直接在终端输入数据

3、数据会由logstash处理后返回并存储到es集群中

input {stdin {}
}
output {stdout {codec => rubydebug}elasticsearch {hosts => ["10.12.153.71","10.12.153.72","10.12.153.133"]index => 'logstash-debug-%{+YYYY-MM-dd}'}
}

端口输入=>字段匹配=>标准输出及es集群

1、由tcp 的8888端口将日志发送到logstash

2、数据被grok进行正则匹配处理

3、处理后，数据将被打印到终端并存储到es

input {tcp {port => 8888}
}
filter {grok {match => {"message" => "%{DATA:key} %{NUMBER:value:int}"} }
}
output {stdout {codec => rubydebug}elasticsearch {hosts => ["10.12.153.71","10.12.153.72","10.12.153.133"]index => 'logstash-debug-%{+YYYY-MM-dd}'}
}
# yum install -y nc
# free -m |awk 'NF==2{print $1,$3}' |nc logstash_ip 8888

文件输入=>字段匹配及修改时间格式修改=>es集群

1、直接将本地的日志数据拉去到logstash当中

2、将日志进行处理后存储到es

input {file {type => "nginx-log"path => "/var/log/nginx/error.log"start_position => "beginning" # 此参数表示在第一次读取日志时从头读取# sincedb_path => "自定义位置"  # 此参数记录了读取日志的位置，默认在 data/plugins/inputs/file/.sincedb*}
}
filter {grok {match => { "message" => '%{DATESTAMP:date} [%{WORD:level}] %{DATA:msg} client: %{IPV4:cip},%{DATA}"%{DATA:url}"%{DATA}"%{IPV4:host}"'}    }    date {match => [ "timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]    }    
}
​
output {if [type] == "nginx-log" {elasticsearch {hosts => ["10.12.153.71","10.12.153.72","10.12.153.133"]index => 'logstash-audit_log-%{+YYYY-MM-dd}'}}}filebeat => 字段匹配 => 标准输出及esinput {beats {port => 5000}
}
filter {grok {match => {"message" => "%{IPV4:cip}"}   }
}
output {elasticsearch {hosts => ["192.168.249.139:9200","192.168.249.149:9200","192.168.249.159:9200"]index => 'test-%{+YYYY-MM-dd}'}stdout { codec => rubydebug }
}
（3）配置

创建目录，我们将所有input、filter、output配置文件全部放到该目录中。

mkdir -p /usr/local/logstash-7.13.2/etc/conf.d
vim /usr/local/logstash-7.13.2/etc/conf.d/input.conf
input { 
kafka {type => "audit_log"codec => "json"topics => "nginx"decorate_events => truebootstrap_servers => "10.12.153.71","10.12.153.72","10.12.153.133"}
}
​
vim /usr/local/logstash-7.13.2/etc/conf.d/filter.conf
filter {json { # 如果日志原格式是json的，需要用json插件处理source => "message"target => "nginx" # 组名}
}
​
vim /usr/local/logstash-7.13.2/etc/conf.d/output.conf
output {if [type] == "audit_log" {elasticsearch {hosts => ["10.12.153.71","10.12.153.72","10.12.153.133"]index => 'logstash-audit_log-%{+YYYY-MM-dd}'}}}

（3）启动

cd /usr/local/logstash-7.13.2
nohup bin/logstash -f etc/conf.d/  --config.reload.automatic &

grok

1、手动输入日志数据

一般为debug 方式，检测 ELK 集群是否健康，这种方法在 logstash 启动后可以直接手动数据数据，并将格式化后的数据打印出来。

数据链路

1、启动logstash

2、logstash启动后，直接进行数据输入

3、logstash处理后，直接进行返回

input {stdin {}
}
output {stdout {codec => rubydebug}
}

2、手动输入数据，并存储到 es

数据链路

1、启动logstash

2、启动后直接在终端输入数据

3、数据会由logstash处理后返回并存储到es集群中

input {stdin {}
}
output {stdout {codec => rubydebug}elasticsearch {hosts => ["10.12.153.71","10.12.153.72","10.12.153.133"]index => 'logstash-debug-%{+YYYY-MM-dd}'}
}

3、自定义日志1

数据链路

1、由tcp 的8888端口将日志发送到logstash

2、数据被grok进行正则匹配处理

3、处理后，数据将被打印到终端并存储到es

input {tcp {port => 8888}
}
filter {grok {match => {"message" => "%{DATA:key} %{NUMBER:value:int}"}   }
}
output {stdout {codec => rubydebug}elasticsearch {hosts => [""10.12.153.71","10.12.153.72","10.12.153.133""]index => 'logstash-debug-%{+YYYY-MM-dd}'}
}
# yum install -y nc
# free -m |awk 'NF==2{print $1,$3}' |nc logstash_ip 8888
​
4、自定义日志2
数据链路
1、由tcp 的8888端口将日志发送到logstash2、数据被grok进行正则匹配处理3、处理后，数据将被打印到终端input {tcp {port => 8888}
}
filter {grok {match => {"message" => "%{WORD:username}\:%{WORD:passwd}\:%{INT:uid}\:%{INT:gid}\:%{DATA:describe}\:%{DATA:home}\:%{GREEDYDATA:shell}"}}
}
output {stdout {codec => rubydebug}
}
​
# cat /etc/passwd | nc logstash_ip 8888

5、nginx access 日志

数据链路

1、在filebeat配置文件中，指定kafka集群ip [output.kafka] 的指定topic当中

2、在logstash配置文件中，input区域内指定kafka接口，并指定集群ip和相应topic

3、logstash 配置filter 对数据进行清洗

4、将数据通过 output 存储到es指定index当中

5、kibana 添加es 索引，展示数据

input {kafka {type => "audit_log"codec => "json"topics => "haha"#decorate_events => true#enable_auto_commit => trueauto_offset_reset => "earliest"bootstrap_servers => ["192.168.52.129:9092,192.168.52.130:9092,192.168.52.131:9092"]}
}
​
filter {grok {match => { "message" => "%{COMBINEDAPACHELOG} %{QS:x_forwarded_for}"}    }    date {match => [ "timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]    }    geoip {source => "lan_ip"    }
}
​
output {if [type] == "audit_log" {stdout {codec => rubydebug}elasticsearch {hosts => ["192.168.52.129","192.168.52.130","192.168.52.131"]index => 'tt-%{+YYYY-MM-dd}'}}}
​#filebeat 配置filebeat.prospectors:
- input_type: logpaths:-  /opt/logs/server/nginx.logjson.keys_under_root: truejson.add_error_key: truejson.message_key: log
​
output.kafka:   hosts: [""10.12.153.71","10.12.153.72","10.12.153.133""]topic: 'nginx'
​# nginx 配置log_format main        '{"user_ip":"$http_x_real_ip","lan_ip":"$remote_addr","log_time":"$time_iso8601","user_req":"$request","http_code":"$status","body_bytes_sents":"$body_bytes_sent","req_time":"$request_time","user_ua":"$http_user_agent"}';access_log  /var/log/nginx/access.log  main;
​

6、nginx error日志

数据链路

1、直接将本地的日志数据拉去到logstash当中

2、将日志进行处理后存储到es

input {file {type => "nginx-log"path => "/var/log/nginx/error.log"start_position => "beginning"}
}
filter {grok {match => { "message" => '%{DATESTAMP:date} [%{WORD:level}] %{DATA:msg} client: %{IPV4:cip},%{DATA}"%{DATA:url}"%{DATA}"%{IPV4:host}"'}    }    date {match => [ "timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]    }    
}
​
output {if [type] == "nginx-log" {elasticsearch {hosts => [""10.12.153.71:9200","10.12.153.72:9200","10.12.153.133:9200""]index => 'logstash-audit_log-%{+YYYY-MM-dd}'}}}

7、filebate 传输给 logstash

input {beats {port => 5000}
}
filter {grok {match => {"message" => "%{IPV4:cip}"}   }
}
output {elasticsearch {hosts => ["192.168.249.139:9200","192.168.249.149:9200","192.168.249.159:9200"]index => 'test-%{+YYYY-MM-dd}'}stdout { codec => rubydebug }
}
​
filebeat.inputs:
- type: logenabled: truepaths:- /var/log/nginx/access.log
output.logstash:hosts: ["192.168.52.134:5000"]

filebeat 日志模板

filebeat.inputs:
- type: logenabled: truepaths:- /var/log/nginx/access.log
output.kafka:hosts: ["192.168.52.129:9092","192.168.52.130:9092","192.168.52.131:9092"]topic: hahapartition.round_robin:reachable_only: truerequired_acks: 1

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://xiahunao.cn/news/2813404.html

如若内容造成侵权/违法违规/事实不符，请联系瞎胡闹网进行投诉反馈，一经查实，立即删除！

相关文章

Java+SpringBoot+Vue+MySQL：美食推荐系统的技术革新

Java+SpringBoot+Vue+MySQL：美食推荐系统的技术革新

✍✍计算机编程指导师 ⭐⭐个人介绍：自己非常喜欢研究技术问题！专业做Java、Python、微信小程序、安卓、大数据、爬虫、Golang、大屏等实战项目。 ⛽⛽实战项目：有源码或者技术上的问题欢迎在评论区一起讨论交流！ ⚡⚡ Java实战 |…
阅读更多...
2024-02-26（Spark，kafka）

2024-02-26（Spark，kafka）

1.Spark SQL是Spark的一个模块，用于处理海量结构化数据 限定：结构化数据处理 RDD的数据开发中，结构化，非结构化，半结构化数据都能处理。 2.为什么要学习SparkSQL SparkSQL是非常成熟的海量结构化数据处理框架。 学…
阅读更多...
docker小知识：linux环境安装docker

docker小知识：linux环境安装docker

安装必要软件包，执行如下命令 yum install -y yum-utils device-mapper-persistent-data lvm2目的是确保在安装 Docker 之前，系统已经安装了必要的软件包和服务，以支持 Docker 的正常运行。设置yum源，添加Docker官方的CentOS存储…
阅读更多...
阿里云服务器2024年优惠价格表曝光，太可怕了！

阿里云服务器2024年优惠价格表曝光，太可怕了！

2024阿里云服务器优惠活动政策整理，轻量2核2G3M服务器61元一年、2核4G4M带宽165元1年，云服务器4核16G10M带宽26元1个月、149元半年，阿里云ECS云服务器2核2G3M新老用户均可99元一年续费不涨价，企业用户2核4G5M带宽199元一年&#x…
阅读更多...
微服务-微服务链路追踪组件Skywalking实战

微服务-微服务链路追踪组件Skywalking实战

自动化监控系统Prometheus&Grafana实战： 4 trem APM-性能监控项目班： https://vip.tulingxueyuan.cn/detail/p_602e574ae4b035d3cdb8f8fe/6 1. skywalking是什么 1.1 Skywalking主要功能特性 1.2 Skywalking整体架构 1.3 SkyWalking 环境搭建部…
阅读更多...
2024热门UI设计工具推荐

2024热门UI设计工具推荐

最好的UI设计工具可以适应几乎每一个设计过程，并且有望满足您的创造性要求。但既然UI设计工具那么多，应该用什么工具呢？ 在过去的几年里，Sketch和figma的结合一直是许多设计师的选择，但其他工具也提供了有竞争力的特点…
阅读更多...
Redis 16种妙用

Redis 16种妙用

1、缓存 2、数据共享分布式 3、分布式锁 4、全局ID 5、计数器 6、限流 7、位统计 8、购物车 9、用户消息时间线timeline 10、消息队列 11、抽奖 12、点赞、签到、打卡 13、商品标签 14、商品筛选 15、用户关注、推荐模型 16、排行榜 1、缓存 String类型 例如：热点…
阅读更多...
QT信号槽实现分析

QT信号槽实现分析

1.宏定义 qt中引入了MOC来反射，编译阶段变成 MOC–>预处理–>编译–>汇编–>链接 1-1、Q_OBJECT 这个宏定义了一系列代码，包括元对象和处理的函数 #define Q_OBJECT \public: \QT_WARNING_PUSH \Q_OBJECT_NO_OVERRIDE_WARNING \static c…
阅读更多...
【学海无涯】嵌入式工程师的100本专业书籍

【学海无涯】嵌入式工程师的100本专业书籍

001《大话数据结构》 002《鸟哥的 linux 私房菜》 003《疯狂 android 讲义》 004《第一行代码》 005《linux 内核设计与实现》 006《驱动设计开发》 007《linux 内核解密》 008《unix 环境高级编程》 009《linux 内核设计与实现》 010《essential C》 011《嵌入式 li…
阅读更多...
CTFHub技能树web之XSS

CTFHub技能树web之XSS

在XSS系列的题目中，由于需要使用能够接受XSS数据的平台，并且由于使用的是CTFHub的模拟机器人点击我们的虚假URL，因此使用的XSS平台不能是自己本地搭建的，如果是本地的模拟点击的机器人将无法访问我们给的这个URL地址，也…
阅读更多...
移动端学习：如何把exe转换成apk

移动端学习：如何把exe转换成apk

exe转换成apk是怎么实现的呢？-电脑端-一门科技将exe文件转换成apk文件是一个比较常见的需求，尤其是对于一些开发者和用户来说。但是，这个过程并不是简单的复制和粘贴。在本文中，我们将介绍exe转换成apk的原理和详细介绍。首先，我们需要了解什么https://www.yimenapp.net/k…
阅读更多...
Windows常用协议

Windows常用协议

LLMNR 1. LLMNR 简介 链路本地多播名称解析(LLMNR)是一个基于域名系统(DNS)数据包格式的协议,可用于解析局域网中本地链路上的主机名称。它可以很好地支持IPv4和IPv6，是仅次于DNS 解析的名称解析协议。 2.LLMNR 解析过程 当本地hosts 和 DNS解析 当本地hosts 和 …
阅读更多...
pytorch --反向传播和优化器

pytorch --反向传播和优化器

1. 反向传播 计算当前张量的梯度 Tensor.backward(gradientNone, retain_graphNone, create_graphFalse, inputsNone)计算当前张量相对于图中叶子节点的梯度。 使用反向传播，每个节点的梯度，根据梯度进行参数优化，最后使得损失最小化 代码…
阅读更多...
【MATLAB】小波 MFE_SVM_LSTM 神经网络时序预测算法

【MATLAB】小波 MFE_SVM_LSTM 神经网络时序预测算法

有意向获取代码，请转文末观看代码获取方式~也可转原文链接获取~ 1 基本定义 小波MFE_SVM_LSTM神经网络时序预测算法是一种结合了小波变换、多尺度特征提取（MFE）、支持向量机（SVM）和长短期记忆神经网络（LST…
阅读更多...
2024年食品饮料市场格局预测-2024年食品饮料行业未来发展趋势

2024年食品饮料市场格局预测-2024年食品饮料行业未来发展趋势

娃哈哈创始人、董事长宗庆后离世，未来将由其女宗馥莉正式接班。 目前，娃哈哈电商的品类结构如下：饮料市场占比整体第一，占比超过55%，代表性产品为“AD钙奶”；其次是饮用水，占比在20%左右&#…
阅读更多...
python 基础知识点（蓝桥杯python科目个人复习计划51）

python 基础知识点（蓝桥杯python科目个人复习计划51）

今日复习计划：做复习题 例题1：大石头的搬运工 问题描述： 在一款名为“大石头的搬运工”的游戏中，玩家需要 操作一排n堆石头，进行n - 1轮游戏。 每一轮，玩家可以选择一堆石头，并将其移动到任…
阅读更多...
mapbox高德地图与相机

mapbox高德地图与相机

mapbox高德地图与相机 本案例使用Mapbox GL JavaScript库创建高德地图。 演示效果引入 CDN 链接地图显示 创建地图实例定义地图数据源配置地图图层 设置地图样式实现代码 1. 演示效果 2. 引入 CDN 链接 <script src"https://api.mapbox.com/mapbox-gl-js/v2.12.0/mapb…
阅读更多...
YOLOv9-Openvino和ONNXRuntime推理【CPU】

YOLOv9-Openvino和ONNXRuntime推理【CPU】

1 环境&#xff1a; CPU&#xff1a;i5-12500 Python&#xff1a;3.8.18 2 安装Openvino和ONNXRuntime 2.1 Openvino简介 Openvino是由Intel开发的专门用于优化和部署人工智能推理的半开源的工具包&#xff0c;主要用于对深度推理做优化。 Openvino内部集成了Opencv、Tens…
阅读更多...
SQL注入漏洞解析--less-46

SQL注入漏洞解析--less-46

我们先看一下46关 他说让我们先输入一个数字作为sort,那我们就先输入数字看一下 当我们分别输入1&#xff0c;2&#xff0c;3可以看到按照字母顺序进行了排序&#xff0c;所以它便是一个使用了order by语句进行排序的查询的一种查询输出方式 当输入时出现报错提示&#xff0c;说…
阅读更多...
编曲学习：和声小调 终止式 离调和弦 转调应用

编曲学习：和声小调 终止式 离调和弦 转调应用

和声小调 音阶 大调音程关系排列：全 全 半 全 全 全 半 小调音程关系排列：全 半 全 全 半 全 全 C大调音阶： 1 2 3 4 5 6 7 1 C小调音阶： 1 2 b3 4 5 b6 b7 1 C大调基本音级构成的和弦： Cmaj7 Dmin7 Emin7 Fmaj7 G7 Amin7 Bm7-5 C小调基本音级构成的和弦： Cmin7 D…
阅读更多...
最新文章

Copyright @ 2022~2023