Logstash 部署文档

一、文档介绍

本文档详细描述 Logstash 部署步骤，以及描述常用日志类型的日志格式转换和存储过程。

二、部署说明

操作系统：Linux（CentOS 7.6）
安装包版本：Logstash 8.15.2

三、部署

(一) 传统部署

1. 下载

官网下载地址：https://www.elastic.co/downloads/logstash

cd /usr/local/srcwget https://artifacts.elastic.co/downloads/logstash/logstash-8.15.2-linux-x86_64.tar.gz

2. 安装

cd /usr/local/srctar xzvf logstash-8.15.2-linux-x86_64.tar.gzmv logstash-8.15.2-linux-x86_64.tar.gz /usr/local/logstash-8.15.2cd /usr/local/ln -s logstash-8.15.2 logstash

3. 创建目录

mkdir -p /data/logstash_data/{data,tmp}cd /usr/local/logstash# 配置文件目录mv config /data/logstash_data/confln -s /data/logstash_data/conf /usr/local/logstash/config# 日志目录mkdir -p /data/logs/logstashln -s /data/logs/logstash /usr/local/logstash/logs# 数据目录ln -s /data/logstash_data/data /usr/local/logstash/data

4. 基础配置

配置文件路径：/usr/local/logstash/logstash.yml。

http.host: "0.0.0.0"http.port: 5044pipeline.batch.size: 1250pipeline.batch.delay: 10xpack.monitoring.enabled: truexpack.monitoring.elasticsearch.username: 'logstash_system'xpack.monitoring.elasticsearch.password: '2023.logstash.elastic.co'xpack.monitoring.elasticsearch.hosts: [ "http://elasticsearch:9200" ]config.reload.automatic: trueconfig.reload.interval: 10s

5. 创建服务

cat <<"EOF" | tee /usr/lib/systemd/system/logstash.service[Unit]Description=Logstash is an open source data collection engine with real-time pipelining capabilities.Documentation=https://www.elastic.co/logstashWants=network-online.targetAfter=network-online.target[Service]RuntimeDirectory=logstashPrivateTmp=trueEnvironment=JAVA_HOME=/usr/local/logstash/jdkEnvironment=PID_DIR=/data/logstash_data/tmpWorkingDirectory=/usr/local/logstashUser=aiuserGroup=aiuserExecStart=/usr/local/logstash/bin/logstash "--path.settings" "/usr/local/logstash/config"StandardOutput=journalStandardError=inheritLimitNOFILE=655350LimitNPROC=4096LimitAS=infinityLimitFSIZE=infinityLimitMEMLOCK=infinityTimeoutStopSec=0KillSignal=SIGTERMKillMode=processSendSIGKILL=noSuccessExitStatus=143[Install]WantedBy=multi-user.targetEOF

6. 启动服务

systemctl daemon-reloadsystemctl enable logstash --now

(二) docker-compose 部署

version: "3.9"networks:  elastic:    external: trueservices:  kibana:    image: "logstash:8.15.2"    container_name: "logstash"    hostname: "logstash"    restart: always    networks:      - elastic    command:       - --config.reload.automatic    environment:      - TZ=Asia/Shanghai      - "LS_JAVA_OPTS=-Xms2048m -Xmx2048m"    volumes:      - /data/logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml      - /data/logstash/config/pipelines.yml:/usr/share/logstash/config/pipelines.yml      - /data/logstash/config/pipelines:/usr/share/logstash/config/pipelines      - /data/logstash/config/patterns:/usr/share/logstash/config/patterns      - /data/logstash/data:/usr/share/logstash/data    ports:      - "5044:5044"

四、pipeline 配置

Logstash Pipeline（管道）是 Logstash 中独立的运行单元，每个管道都包含两个必需的元素输入（input）和输出（output），和一个可选的元素过滤器（filter），事件处理管道负责协调它们的执行。输入和输出支持编解码器，使您可以在数据进入或退出管道时对其进行编码或解码，而不必使用单独的过滤器。

Logstash Pipeline 中每个输入阶段都运行在自己的线程中，输入将事件写入到内存或磁盘的中央队列。每个管道工作线程（pipeline worker）从队列中获取一批事件，通过配置的过滤器运行这批事件，然后将过滤的事件运行到所有输出。

配置文件路径：/usr/local/logstash/pipelines.yml。例如：

- pipeline.id: main  path.config: "/usr/local/logstash/config/pipelines/main.conf"- pipeline.id: java  path.config: "/usr/local/logstash/config/pipelines/java.conf"- pipeline.id: k8s  path.config: "/usr/local/logstash/config/pipelines/k8s.conf"- pipeline.id: mysqlslow  path.config: "/usr/local/logstash/config/pipelines/mysql-slow.conf"- pipeline.id: nginx  path.config: "/usr/local/logstash/config/pipelines/nginx.conf"

(一) main 配置入口

1. filebeat 配置一：自定义字段

filebeat 配置

filebeat.inputs:- type: filestream  paths: ["/var/log/nginx/access.log"]  fields:    log_type: nginx       # 关键字段  fields_under_root: false- type: filestream  paths: ["/data/logs/app/*.log"]  fields:    log_type: java    project_name: app    app_name: order- type: filestream  paths: ["/var/log/mysql/slow.log"]  fields:    log_type: mysql-slow

main.conf 配置

input {  beats {    port => 5044  }}output {  if [fields][log_type] == "nginx" {    pipeline {      send_to => "nginx_http"    }  }  else if [fields][log_type] == "java" {    pipeline {      send_to => "java_log"    }  }  else if [fields][log_type] == "mysql-slow" {    pipeline {      send_to => "mysql_slow"    }  }  else if [fields][log_type] == "k8s" {    pipeline {      send_to => "k8s"    }  }  else {    # 兜底，防止丢数据（可改成 stdout debug）    stdout {      codec => rubydebug    }  }}

2. filebeat 配置二：[tags] 字段

filebeat 配置

filebeat.inputs:- type: filestream  paths:    - /var/log/nginx/access.log  tags: ["nginx"]        # 关键字段- type: filestream  paths:    - /data/logs/app/*.log  tags: ["java"]- type: filestream  paths:    - /var/log/mysql/slow.log  tags: ["mysql-slow"]

main.conf 配置

input {  beats {    port => 5044  }}output {  if "nginx" in [tags] {    pipeline {      send_to => "nginx_http"    }  }  else if "java" in [tags] {    pipeline {      send_to => "java_log"    }  }  else if "mysql-slow" in [tags] {    pipeline {      send_to => "mysql_slow"    }  }  else if "k8s" in [tags] {    pipeline {      send_to => "k8s_log"    }  }  else {    # debug 兜底    stdout {      codec => rubydebug    }  }}

(二) 配置：Java 日志解析

input {    pipeline {        address => java    }}filter {    grok {        patterns_dir => "/data/logstash/config/patterns"        match => { "message" => "%{JAVALOG}" }    }    date {        match => ["create_time", "yyyy-MM-dd HH:mm:ss"]        target => "@timestamp"    }    mutate {        remove_field => [ "create_time", "tags", "message" ]    }}output {    elasticsearch {        hosts => [ "http://192.168.0.22:9200" ]        user => "elastic"        password => "2023.elastic.co"        index => "java-%{[log][project_name]}-%{[log][app_name]}-%{+YYYY.MM.dd}"        manage_template => false    }}

(三) 配置：Nginx 日志解析

input {    pipeline {        address => nginx_http    }}filter {    grok {        patterns_dir => "/data/logstash/config/patterns"        match => { "message" => "%{NGINXHTTP}" }    }    date {        match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]    }    mutate {        add_field => { "[log][date]" => "%{+YYYY.MM.dd}" }        remove_field => [ "timestamp", "message" ]        convert => ["request_time", "float"]        convert => ["upstream_response_time", "float"]        convert => ["sent_bytes", "integer"]    }    useragent {        source => 'http_user_agent'    }}output {    elasticsearch {        hosts => [ "http://192.168.0.22:9200" ]        user => "elastic"        password => "2023.elastic.co"        index => "nginx-http-%{[log][project_name]}-%{[log][app_name]}-%{+YYYY.MM.dd}"        manage_template => false    }}

(四) 配置：K8S 日志解析

input {  kafka {    bootstrap_servers => "192.168.0.33:9092"    topics => ["k8slog-8.9.2"]    group_id => "ops_group"    consumer_threads => 1    auto_offset_reset => "latest"    codec => json { charset => "UTF-8" }  }}filter {  mutate {    remove_field => [ "event" ]  }}output {  elasticsearch {    hosts => [ "http://192.168.0.22:9200" ]    user => "elastic"    password => "2023.elastic.co"    index => "app-%{+YYYY.MM.dd}"    template_name => "app"  }}

(五) 配置：MySQL 慢查询日志解析

input {  kafka {    bootstrap_servers => "192.168.0.33:9092"    topics => ["mysql-slow-log"]    group_id => "ops_group"    consumer_threads => 1    auto_offset_reset => "latest"    codec => json { charset => "UTF-8" }  }}filter {  grok {    patterns_dir => "/data/logstash/config/patterns"    match => { "message" => "%{MYSQLSLOWLOG}" }  }  date {    match => [ "timestamp", "UNIX" ]    target => "@timestamp"  }  mutate {    remove_field => [ "event", "message" ]  }}output {  elasticsearch {    hosts => [ "http://192.168.0.22:9200" ]    user => "elastic"    password => "2023.elastic.co"    index => "mysql-slow-%{+YYYY.MM.dd}"    template_name => "mysql-slow"  }}

五、pipelines input 配置

(一) 输入源：filebeat

logstash 直接接收 filebeat 输出的日志。

input {  beats {    port => 5044  }}

(二) 输入源：logstash pipeline

logstash 直接接收经过 logstash pipeline 转换后的日志，如第四节中 pipeline main 配置入口转换后的日志。

input {    pipeline {        address => nginx_http    }}

(三) 输入源：kafka

logstash 订阅 kafka 输出的日志。

input {  kafka {    bootstrap_servers => "192.168.0.33:9092"    topics => ["k8slog-8.9.2"]    group_id => "ops_group"    consumer_threads => 1    auto_offset_reset => "latest"    codec => json { charset => "UTF-8" }  }}

六、pattern 配置

Logstash Pipeline 在过滤阶段，可以对特定字段（例如 message 字段）进行再解析，将字段拆分成可符合业务要求、再精细的字段。通过 grok 操作匹配 pattern 配置中的规则。

(一) 解析 nginx 日志

原始日志格式：

log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '                  '$status$body_bytes_sent "$http_referer" '                  '"$http_user_agent" "$http_x_forwarded_for" $request_time '                  '$upstream_response_time$upstream_addr$upstream_status';

pattern 规则：

NGINXHTTP %{IP:remote_addr} - (%{WORD:remote_user}|-) \[%{HTTPDATE:timestamp}\] \"%{WORD:request_method} %{NOTSPACE:request_uri} %{NOTSPACE:request_protocol}\" %{NUMBER:status_code} %{NUMBER:sent_bytes} \"%{DATA:http_referer}\" \"%{DATA:http_user_agent}\" \"(%{IPV4:http_x_forwarded_for}|-)\" %{NUMBER:request_time} (%{NUMBER:upstream_response_time}|-) (%{HOSTPORT:upstream_addr}|-) (%{NUMBER:upstream_status_code}|-)

(二) 解析 java 日志

原始日志格式：

%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger -%msg%n

pattern 规则：

JAVALOG %{TIMESTAMP_ISO8601:create_time} \[%{DATA:thread_name}\] %{LOGLEVEL:level}%{SPACE}%{DATA:java_class}%{SPACE}-%{SPACE}%{GREEDYDATA:content}

(三) 解析 mysql 慢查询日志

标准的 MySQL 慢查询日志。

pattern 规则：

MYSQLSLOWLOG ^# User@Host: %{USER:query_user}\[[^\]]+\] @ (?:(?<query_host>\S*) )?\[(?:%{IP:query_ip})?\]\s+Id:\s+%{NUMBER:query_id}\s# Query_time: %{NUMBER:query_time}\s+Lock_time: %{NUMBER:lock_time}\s+Rows_sent: %{NUMBER:rows_sent}\s+Rows_examined: %{NUMBER:rows_examined}\s*(?:use %{DATA:database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<sql>(?<action>\w+)\s+.*)

(四) 解析 linux history 日志

原始日志格式：

export PROMPT_COMMAND='history -a;timestamp=$(date +%FT%T%z);command=$(history 1| { read x y;echo $y; } );logger -p local1.notice -t bash -i "timestamp=[$timestamp],user=$USER,ppid=$PPID,conn=$SSH_CONNECTION,tty=$SSH_TTY,pwd=$PWD,command=$command"'

pattern 规则：

LINUXHISTORY %{GREEDYDATA:create_time} %{DATA:hostname} %{WORD:shell}\[%{NUMBER:pid}\]: timestamp=\[%{TIMESTAMP_ISO8601:timestamp}\],user=%{DATA:user},ppid=%{NUMBER:ppid},conn=%{IPV4:remote_ip} %{NUMBER:remote_port} %{IPV4:ssh_ip} %{NUMBER:ssh_port},tty=%{DATA:tty},pwd=%{DATA:pwd},command=%{GREEDYDATA:command}