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

news2024/11/20 18:23:30

引言：上一期我们进行了对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 => 字段匹配 => 标准输出及es

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 }
}
（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 => true
    bootstrap_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端口将日志发送到logstash

2、数据被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 => true
        auto_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: log
  paths:
    -  /opt/logs/server/nginx.log
  json.keys_under_root: true
  json.add_error_key: true
  json.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: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
output.logstash:
  hosts: ["192.168.52.134:5000"]

filebeat 日志模板

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
output.kafka:
  hosts: ["192.168.52.129:9092","192.168.52.130:9092","192.168.52.131:9092"]
  topic: haha
  partition.round_robin:
    reachable_only: true
  required_acks: 1

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

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

相关文章

【DDD】学习笔记-领域驱动设计体系

【DDD】学习笔记-领域驱动设计体系

从统一语言到限界上下文，从限界上下文到上下文映射，从领域分析建模到领域设计建模，再从领域设计建模到领域实现建模，我将软件架构设计、面向对象设计、场景驱动设计和测试驱动开发有机地融合起来，贯穿于领域驱动设计的…
阅读更多...
mac打不开xxx软件， 因为apple 无法检查其是否包含恶意

mac打不开xxx软件， 因为apple 无法检查其是否包含恶意

1. 安全性与隐私下面的允许来源列表，有些版本中的‘任何来源’选项被隐藏了，有些从浏览器下载的软件需要勾选这个选项才能安装 打开‘任何来源’选项 sudo spctl --master-disable 关闭‘任何来源’选项 sudo spctl --master-enable
阅读更多...
UE4c++ ConvertActorsToStaticMesh

UE4c++ ConvertActorsToStaticMesh

UE4c ConvertActorsToStaticMesh ConvertActorsToStaticMesh UE4c ConvertActorsToStaticMesh创建Edior模块（最好是放Editor模块毕竟是编辑器代码）创建UBlueprintFunctionLibraryUTestFunctionLibrary.hUTestFunctionLibrary.cpp:.Build.cs 目标:为了大量…
阅读更多...
探索IP地址定位工具：解读IP数据云的功能与优势

探索IP地址定位工具：解读IP数据云的功能与优势

在当今数字化时代，IP地址定位工具成为了许多领域中不可或缺的技术支持，为网络安全、地理定位服务和个性化推荐等提供了重要数据支持。其中，IP数据云作为一种领先的IP地址定位工具，具有一系列功能和优势，本文将对其进行…
阅读更多...
k8s service的概念以及创建方法

k8s service的概念以及创建方法

Service 的功能： Service主要用于提供网络服务，通过Service的定义，能够为客户端应用提供稳定的访问地址（域名或IP地址）和负载均衡功能，以及屏蔽后端Endpoint的变化，是K8s实现微服务的核心资源。…
阅读更多...
如何选择科技公司或者技术团队来开发软件项目呢

如何选择科技公司或者技术团队来开发软件项目呢

最近有客户问我们为什么同样软件项目不同公司报价和工期差异很大，我们给他解释好久才讲清楚，今天整理一下打算写一篇文章来总结一下，有需要开发朋友可以参考，我们下次遇到客户也可以直接转发文章给客户自己看。 我们根据我们自己报…
阅读更多...
计算机网络：IP

计算机网络：IP

引言： IP协议是互联网协议族中的核心协议之一，负责为数据包在网络中传输提供路由寻址。它定义了数据包如何在互联网上从源地址传输到目的地址的规则和流程。IP协议使得各种不同类型的网络设备能够相互通信，实现了全球范围内的信息交换。 目录…
阅读更多...
人像背景分割SDK，智能图像处理

人像背景分割SDK，智能图像处理

美摄科技人像背景分割SDK解决方案：引领企业步入智能图像处理新时代 随着科技的不断进步，图像处理技术已成为许多行业不可或缺的一部分。为了满足企业对于高质量、高效率人像背景分割的需求，美摄科技推出了一款领先的人像背景分割SDK&#xf…
阅读更多...
Cluster-Level Contrastive Learning for Emotion Recognition in Conversations

Cluster-Level Contrastive Learning for Emotion Recognition in Conversations

对话情绪识别的聚类级对比学习 摘要一、介绍二、相关工作2.1 对话情感识别2.2 对比学习 三、方法3.1 任务定义和模型概述3.2 上下文感知的话语编码器3.3 使用适配器进行知识注入3.4 有监督的集群级对比学习3.4.1情感原型3.4.2集群级别对比学习 3.5 模型训练 四 实验设置4.1 数据…
阅读更多...
idea 更新maven java版本变化

idea 更新maven java版本变化

今天遇到个问题就是&#xff0c;点击maven的reload&#xff0c;会导致setting 里的java compiler 版本变化 这里的话&#xff0c;应该是settings.xml文件里面的这个限定死了&#xff0c;修改一下或者去掉就可以了 <profile><id>JDK-1.8</id><activatio…
阅读更多...
Qt下使用modbus-c库实现PLC线圈/保持寄存器的读写

Qt下使用modbus-c库实现PLC线圈/保持寄存器的读写

系列文章目录 提示&#xff1a;这里是该系列文章的所有文章的目录 第一章&#xff1a;Qt下使用ModbusTcp通信协议进行PLC线圈/保持寄存器的读写&#xff08;32位有符号数&#xff09; 第二章&#xff1a;Qt下使用modbus-c库实现PLC线圈/保持寄存器的读写 文章目录 系列文章目录…
阅读更多...
如何使用程序通过OCR识别解析PDF中的表格

如何使用程序通过OCR识别解析PDF中的表格

https://github.com/PaddlePaddle/PaddleOCR/blob/release/2.7/ppstructure/table/README_ch.md#41-%E5%BF%AB%E9%80%9F%E5%BC%80%E5%A7%8B Paddle-structure是目前我们能找到的可以做中英文版面分析较好的一个基础模型&#xff0c;其开源版可以识别十类页面元素。这篇文章介绍…
阅读更多...
VL817-Q7 USB3.0 HUB芯片 适用于扩展坞 工控机 显示器

VL817-Q7 USB3.0 HUB芯片 适用于扩展坞 工控机 显示器

VL817-Q7 USB3.1 GEN1 HUB芯片 VL817-Q7 USB3.1 GEN1 HUB芯片 VIA Lab的VL817是一款现代USB 3.1 Gen 1集线器控制器&#xff0c;具有优化的成本结构和完全符合USB标准3.1 Gen 1规范&#xff0c;包括ecn和2017年1月的合规性测试更新。VL817提供双端口和双端口4端口配置&…
阅读更多...
windows安装部署node.js并搭建Vue项目

windows安装部署node.js并搭建Vue项目

一、官网下载安装包 官网地址&#xff1a;https://nodejs.org/zh-cn/download/ 二、安装程序 1、安装过程 如果有C/C编程的需求&#xff0c;勾选一下下图所示的部分&#xff0c;没有的话除了选择一下node.js安装路径&#xff0c;直接一路next 2、测试安装是否成功 【winR】…
阅读更多...
VUE从0到1创建项目及基本路由、页面配置

VUE从0到1创建项目及基本路由、页面配置

一、创建项目：（前提已经安装好vue和npm） 目录：E:\personal\project_pro\ windows下，win+R 输入cmd进入命令行： cd E:\personal\project_pro E：# 创建名为test的项目 vue create test# 用上下键选择vue2或vue3，回车确认创建本次选择VUE3 创建好项目后，使用…
阅读更多...
css transform 会影响position 定位

css transform 会影响position 定位

比如通过以下代码.实现导航条上的每个li栏目,以不同的时间间隔,从上向下移动进来并显示 .my-navbar ul li {position: relative;opacity: 0;transform: translateY(-30px);transition: transform .6s cubic-bezier(.165,.84,.44,1),opacity .6s cubic-bezier(.165,.84,.44,1);…
阅读更多...
书生·浦语大模型全链路开源体系介绍

书生·浦语大模型全链路开源体系介绍

背景介绍 随着人工智能技术的迅猛发展&#xff0c;大模型技术已成为当今人工智能领域的热门话题。2022 年 11 月 30 日&#xff0c;美国 OpenAI 公司发布了 ChatGPT 通用型对话系统 并引发了全球 的极大关注&#xff0c;上线仅 60 天月活用户数便超过 1 亿&#xff0c;成为历史…
阅读更多...
租赁系统|租赁软件提供免押金等多种租赁方式

租赁系统|租赁软件提供免押金等多种租赁方式

租赁小程序是一种全新的租赁模式&#xff0c;为用户提供了免押金、多种租赁方式、定制化服务等多项优势&#xff0c;让用户的租赁体验更加美好。让我们来了解一下它的特点和功能。 首先&#xff0c;租赁小程序支持租完即送&#xff0c;无需等待固定租期。它提供了多种租赁方式&…
阅读更多...
Rocky Linux 运维工具 ls

Rocky Linux 运维工具 ls

一、ls 的简介 ​​ls​ 用于列出当前目录下的文件和目录&#xff0c;以及它们的属性信息。通过 ​ls​命令可以查看文件名、文件大小、创建时间等信息&#xff0c;并方便用户浏览和管理文件。 二、ls 的参数说明 序号参数描述1-a显示所有文件&#xff0c;包括以 ​.​开头的…
阅读更多...
辽宁博学优晨教育：视频剪辑培训的领航者

辽宁博学优晨教育：视频剪辑培训的领航者

在当今数字化时代&#xff0c;视频剪辑已经成为一项炙手可热的技能。无论是短视频的火爆&#xff0c;还是影视行业的蓬勃发展&#xff0c;都离不开优秀的视频剪辑人才。辽宁博学优晨教育&#xff0c;作为视频剪辑培训的佼佼者&#xff0c;以其正规、专业的教学态度和高质量的培…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023