ELK日志处理与实战配置
Logstash核心功能与架构解析
Logstash 是 Elastic Stack 中用于数据采集、转换和输出的关键组件，支持从多种来源接收原始数据，并通过插件机制完成格式化、过滤与路由。其工作流程分为三个阶段：
- Input：负责接入数据源，如文件、系统日志、Redis 队列或 Beats 客户端。
- Filter：对数据进行清洗与结构化处理，包括字段提取、类型转换、条件判断等。
- Output：将处理后的数据写入目标存储，如 Elasticsearch、MongoDB、文件系统或 Redis 缓存。
常用插件说明
grok：利用正则表达式解析非结构化文本，内置120+常见模式（如 Apache、Nginx 日志）。mutate：执行字段重命名、删除、替换或类型转换操作。geoip：根据客户端 IP 地址查询地理位置信息，增强日志上下文。drop：丢弃特定事件（如调试日志），减少冗余数据。
部署与启动方式
可通过 yum 安装 Logstash 并配置官方仓库：
# /etc/yum.repos.d/es.repo
[kibana-4.5]
name=Kibana repository for 4.5.x packages
baseurl=https://packages.elastic.co/kibana/4.5/centos
gpgcheck=1
gpgkey=https://packages.elastic.co/GPG-KEY-elasticsearch
enabled=1
[beats]
name=Elastic Beats Repository
baseurl=https://packages.elastic.co/beats/yum/el/$basearch
enabled=1
gpgkey=https://packages.elastic.co/GPG-KEY-elasticsearch
gpgcheck=1
# 安装命令
yum install logstash -y
快速测试运行
使用命令行直接运行一个简单管道，读取标准输入并输出调试信息：
logstash/bin/logstash -e 'input{stdin{}} output{stdout{codec=>rubydebug}}'
输入任意内容（如 `abc`），将输出包含时间戳、主机名及原始消息的 JSON 结构。
配置语法规范
Logstash 配置采用层级结构，由 input、filter、output 三段组成：
input {
stdin {}
}
filter {
# 可添加 grok、mutate 等处理逻辑
}
output {
stdout {
codec => rubydebug
}
}
支持多输入源与多输出目标，例如同时读取多个日志文件并分别写入控制台与压缩日志文件：
input {
stdin {}
file {
path => ["/var/log/messages"]
type => "system"
start_position => "beginning"
}
}
output {
stdout { codec => rubydebug }
file {
path => "/var/datalog/mysystem.log.gz"
gzip => true
}
}
Filebeat 轻量级日志采集器
Filebeat 是基于 Go 语言开发的日志收集工具，无需 Java 环境，资源占用极低，适合部署在应用服务器端。
配置示例
配置文件位于 /etc/filebeat/filebeat.yml，可定义多个采集任务（prospector）：
filebeat:
prospectors:
- paths: ["/alidata/log/nginx/access/access.log.json"]
input_type: log
document_type: nginxacclog
- paths: ["/alidata/www/storage/logs/laravel.log"]
input_type: log
document_type: larlog
- paths: ["/alidata/log/php/php-fpm.log.slow"]
input_type: log
document_type: phpslowlog
multiline:
pattern: '^\s'
negate: true
match: after
output:
logstash:
hosts: ["10.160.8.221:5044"]
redis:
host: "10.122.52.129"
port: 6379
password: "123456"
注意：若原始日志中已含 @timestamp 字段，Filebeat 会覆盖为本地采集时间（UTC）。否则自动添加当前时间戳。
典型应用场景实践
架构一：日志经 Filebeat → Logstash → Redis → Elasticsearch
适用于高吞吐量场景，中间层用 Redis 做缓冲，避免数据丢失。
Nginx 日志格式（JSON）
log_format json '{"@timestamp":"$time_iso8601",
"slbip":"$remote_addr",
"clientip":"$http_x_forwarded_for",
"serverip":"$server_addr",
"size":$body_bytes_sent,
"responsetime":$request_time,
"domain":"$host",
"method":"$request_method",
"requesturi":"$request_uri",
"url":"$uri",
"appversion":"$HTTP_APP_VERSION",
"referer":"$http_referer",
"agent":"$http_user_agent",
"status":"$status"}';
Logstash 接收 Filebeat 数据并转储至 Redis
input {
beats {
port => 5044
codec => "json"
}
}
filter {
if [document_type] == "nginxacclog" {
geoip {
source => "clientip"
target => "geoip"
database => "/u01/elk/logstash/GeoLiteCity.dat"
add_field => ["[geoip][coordinates]", "%{[geoip][longitude]}"]
add_field => ["[geoip][coordinates]", "%{[geoip][latitude]}"]
}
mutate {
convert => ["[geoip][coordinates]", "float"]
}
}
}
output {
if [document_type] == "nginxacclog" {
redis {
data_type => "list"
key => "nginxacclog"
host => "127.0.0.1"
port => "26379"
password => "123456"
db => "0"
}
}
if [document_type] == "messages" {
redis {
data_type => "list"
key => "messages"
host => "127.0.0.1"
port => "26379"
password => "123456"
db => "0"
}
}
}
Logstash 从 Redis 读取并写入 Elasticsearch
input {
redis {
host => "10.10.1.2"
port => "26379"
db => "0"
key => "nginxacclog"
threads => 300
password => "123456"
data_type => "list"
codec => "json"
}
redis {
host => "10.10.1.2"
port => "26379"
db => "0"
key => "messages"
password => "123456"
threads => 50
data_type => "list"
codec => "json"
}
}
output {
if [document_type] == "nginxacclog" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "logstash-nginxacclog-%{+YYYY.MM.dd}"
manage_template => true
flush_size => 50000
idle_flush_time => 10
workers => 2
}
}
if [document_type] == "messages" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "logstash-messages-%{+YYYY.MM.dd}"
manage_template => true
flush_size => 50000
idle_flush_time => 30
workers => 1
}
}
}
关键参数解释
- threads：Redis 输入线程数，需根据下游处理能力调整，过高可能导致丢包。
- flush_size：批量提交大小，达到后立即发送至 ES。
- idle_flush_time：未达批量阈值时，等待最大时间后强制提交。
- workers：Elasticsearch 输出线程数，建议设为 1~2。
架构二：Filebeat 直接写入 Redis → Logstash 写入 MongoDB
适用于需要长期归档且不依赖 Elasticsearch 的场景。
Filebeat 配置（输出到 Redis）
output:
redis:
host: "10.160.8.221"
port: 26379
password: "123456"
Logstash 读取 Redis 并写入 MongoDB
input {
redis {
host => "10.160.8.221"
port => "26379"
key => "filebeat"
data_type => "list"
password => "123456"
threads => 50
}
redis {
host => "10.160.8.221"
port => "26379"
key => "mycat"
data_type => "list"
password => "123456"
threads => 50
type => "mycat"
}
}
output {
if [type] == "mycat" {
mongodb {
collection => "mycat%{+yyyyMMdd}"
isodate => true
database => "logdb"
uri => "mongodb://log_user:123456@10.10.1.102:27017/logdb"
}
}
if [type] == "nginxacclog" {
mongodb {
collection => "nginx_accress%{years_dik3k}%{months_dik3k}%{days_dik3k}"
isodate => true
database => "logdb"
uri => "mongodb://log_user:123456@10.10.1.102:27017/logdb"
}
}
}
该方案实现日志分类型存储，便于后续分析与检索。
