如何使用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://www.hqwc.cn/news/496206.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

智能化运维是新行业吗为什么

智能化运维是新行业吗为什么

智能运维产品线和服务线包括一系列自动化、智能化的技术和服务，为进一步提高IT运维的效率和有效性。智能运维（AIOps）利用大数据、分析技术和机器学习能力自动执行和简化业务流程，包括收集和总结多源IT系统架构组件的数据、业务需求…
阅读更多...
流模型 Flow 超详解，基于 Flow 的生成式模型，从思路到基础到公式推导到模型理解与应用（Flow-based Generative Model）

流模型 Flow 超详解，基于 Flow 的生成式模型，从思路到基础到公式推导到模型理解与应用（Flow-based Generative Model）

参考文献： [1] Dinh L, Krueger D, Bengio Y. Nice: Non-linear independent components estimation[J]. arXiv preprint arXiv:1410.8516, 2014. [2] Dinh L, Sohl-Dickstein J, Bengio S. Density estimation using real nvp[J]. arXiv preprint arXiv:1605.08803…
阅读更多...
express+mysql+vue,从零搭建一个商城管理系统3--user路由模块

express+mysql+vue,从零搭建一个商城管理系统3--user路由模块

提示：学习express，搭建管理系统 文章目录 前言一、新建routes文件夹二、新建routes/index.js和routes/user.js三、修改index.js四、修改routes/index.js五、修改routes/user.js六、启动项目预览总结 前言 需求：主要学习express，所…
阅读更多...
必看——HTTP怎么升级成HTTPS

必看——HTTP怎么升级成HTTPS

将HTTP升级为HTTPS主要涉及获取SSL/TLS证书并在您的服务器上配置它。这个过程可以增强网站的安全性，通过加密客户端和服务器之间的通信来保护数据。下面是一个基本的步骤指南： 1.购买SSL/TLS证书：您可以从许多证书颁发机构（CA&…
阅读更多...
Java Web(八）--Servlet（二）

Java Web(八）--Servlet（二）

Servlet API Servlet API 包含以下4个Java包： 1. javax.servlet：其中包含定义Servlet和Servlet容器之间契约的类和接口。 2. javax.servlet.http：主要定义了与HTTP协议相关的HttpServlet类，HttpServletRequest接口和HttpServl…
阅读更多...
Ubuntu Mysql Innodb cluster集群搭建+MaxScale负载均衡（读写分离）

Ubuntu Mysql Innodb cluster集群搭建+MaxScale负载均衡（读写分离）

Ubuntu系统版本 20.04.3 LTS (Focal Fossa) 、64位系统。 cat /etc/os-release查看Ubuntu系统是32位还是64位 uname -m如果显示“i686”，则表示安装了32位操作系统。如果显示“x86_64”，则表示安装了64位操作系统。 一、安装MySql 参考： https://blog.csdn.net/qq_3712…
阅读更多...
当Web3叙事寒冬到来，游戏是否是冬日里的“一把火”？

当Web3叙事寒冬到来，游戏是否是冬日里的“一把火”？

出品｜欧科云链研究院 作者｜Jason Jiang 以太坊创始人Vitalik在2019年曾说：金融与游戏会是区块链最先落地的场景。 在DeFi金融创新驱动上个周期后，沉寂近两年的Web3游戏板块，如今似乎也在复苏。无论是频繁获得融资&a…
阅读更多...
分享three.js和cannon.js构建Web 3D场景

分享three.js和cannon.js构建Web 3D场景

使用 three.js，您不再需要花哨的游戏PC或控制台来显示逼真的3D图形。 您甚至不需要下载特殊的应用程序。现在每个人都可以使用智能手机和网络浏览器体验令人惊叹的3D应用程序。 这个惊人的库和充满活力的社区是您在浏览器、笔记本电脑、平板电脑或智能手机上创建游…
阅读更多...
PHP+vue+mysql高校学生健康管理系统fe93x

PHP+vue+mysql高校学生健康管理系统fe93x

。高校学生健康管理平台采用系统设计遵循界面层、业务逻辑层和数据访问层的Web开发三层架构。采用B/S结构,使得系统更加容易维护。高校学生健康 管理平台主要实现角色有管理员和学生,医护人员,辅导员,管理员在后台管理诊断结果模块、医护咨询模块、医护人员模块、医护回复模块、…
阅读更多...
WPF 【十月的寒流】学习笔记(3):DataGrid分页

WPF 【十月的寒流】学习笔记(3):DataGrid分页

文章目录 前言相关链接代码仓库项目配置（省略）项目初始配置xamlviewModel Filter过滤详细代码展示结果问题 Linq过滤CollectionDataxamlviewModel sql，这里用到数据库，就不展开了 总结 前言 我们这次详细了解一下列表通知的底层是…
阅读更多...
团结引擎——DotNet Wasm方案

团结引擎——DotNet Wasm方案

参考：团结引擎 DotNet WebAssembly(Wasm) 介绍 一、当前编译流程 通过IL2CPP将C#转成C/C；通过Emscripen将C/C转成WebAssembly； 二、 当前存在问题 IL2CPP在处理类似泛型、反射结构时，由于缺少运行时信息，必须全量生…
阅读更多...
LACP——链路聚合控制协议

LACP——链路聚合控制协议

LACP——链路聚合控制协议 什么是LACP？ LACP（Link Aggregation Control Protocol，链路聚合控制协议）是一种基于IEEE802.3ad标准的实现链路动态聚合与解聚合的协议，它是链路聚合中常用的一种协议。 链路聚合组中启用了…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023