2.6 Logstash环境搭建

Logstash 是 Elastic Stack（ELK Stack）技术栈中的数据收集与处理引擎，采用管道式架构设计，专门处理流式数据。

1. 安装部署

# 1.下载
[root@204-web ~]# wget https://mirrors.aliyun.com/elasticstack/8.x/yum/8.19.7/logstash-8.19.7-x86_64.rpm
# 2.安装
[root@204-web ~]# rpm -ivh logstash-8.19.7-x86_64.rpm

2. Nginx日志采集

a. 整体架构设计

b. Filebeat配置

# /etc/filebeat/filebeat.yml
[root@204-web ~]# cat /etc/filebeat/filebeat.yml 
filebeat.inputs:
- type: filestream
  id: nginx-access
  enabled: true
  paths:
    - /home/deploy/nginx/logs/access.log
  fields:
    log_type: "nginx_access"
  fields_under_root: true

  _config:
    module:
      enabled: false

# 必须禁用自动模板创建
setup.template:
  enabled: false

setup.ilm:
  enabled: false


# 只配置Logstash输出
output.logstash:
  hosts: ["10.0.0.204:5044"]
  loadbalance: true
  max_retries: 3
  timeout: 30s
  bulk_max_size: 2048
  compression_level: 3

# 禁用所有其他输出
output.console:
  enabled: false

output.file:
  enabled: false

processors:
  - add_fields:
      target: ''
      fields:
        processed_by: "logstash"
        log_ingestor: "filebeat-to-logstash"
        
  - drop_fields:
      fields: ["ecs", "agent", "input"]

logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7
  permissions: 0644

c. Logstash配置

# 1.logstash.yml
[root@204-web ~]# cat /etc/logstash/logstash.yml
http.host: "10.0.0.204"
http.port: 9600

# 管道配置
pipeline.workers: 2
pipeline.batch.size: 125
pipeline.batch.delay: 50

# 显式配置buffer类型（修复弃用警告）
pipeline.buffer.type: "heap"
queue.type: "memory"
queue.max_bytes: 1gb

# 路径配置
path.data: /var/lib/logstash
path.logs: /var/log/logstash
path.config: /etc/logstash/conf.d


# 监控配置
monitoring.enabled: false 
monitoring.elasticsearch.hosts: ["http://10.0.0.91:9200", "http://10.0.0.92:9200", "http://10.0.0.93:9200"]

# 性能优化
config.reload.automatic: true
config.reload.interval: 30s
config.support_escapes: true

# 安全配置
api.ssl.enabled: false

# geoip设置
xpack.geoip.downloader.enabled: true
xpack.geoip.downloader.endpoint: "https://geoip.elastic.co/v1/database"

# 日志配置
log.level: info
path.logs: /var/log/logstash

# 2.conf.d/nginx-pipeline.conf
[root@204-web ~]# cat /etc/logstash/conf.d/nginx-pipeline.conf
input {
  beats {
    port => 5044
    host => "10.0.0.204"
    ssl_enabled => false
    codec => "plain"
    # 增加连接优化
    client_inactivity_timeout => 3600
  }
}

filter {
  # 解析Nginx日志格式
  grok {
    match => { 
      "message" => '%{IPORHOST:remote_ip} - %{USER:remote_user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{DATA:request} HTTP/%{NUMBER:http_version}" %{NUMBER:status} %{NUMBER:body_bytes_sent} "%{DATA:http_referer}" "%{DATA:http_user_agent}" %{GREEDYDATA:http_x_forwarded_for}'
    }
    tag_on_failure => ["_grokparsefailure"]
    ecs_compatibility => "disabled"
  }
  
  # 日期解析
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
    timezone => "Asia/Shanghai"
  }
  
  # 字段清理
  mutate {
    convert => {
      "status" => "integer"
      "body_bytes_sent" => "integer"
    }
    remove_field => ["message", "timestamp"]
  }
  
  # 简化的user_agent处理 - 只提取关键信息
  if [http_user_agent] {
    # 使用grok提取简单的user_agent信息，避免嵌套对象
    grok {
      match => { 
        "http_user_agent" => "^%{WORD:ua_name}/%{NUMBER:ua_version}"
      }
    }
    
    # 如果没有匹配到版本，尝试其他模式
    if ![ua_name] {
      grok {
        match => { 
          "http_user_agent" => "^%{WORD:ua_name}"
        }
      }
    }
    
    # 设置默认值
    if ![ua_name] {
      mutate {
        add_field => { "ua_name" => "Unknown" }
      }
    }
  }
  
  # 地理信息
  if [remote_ip] and [remote_ip] != "-" {
    geoip {
      source => "remote_ip"
      target => "geoip"
    }
  }
  
  # 添加处理标记
  mutate {
    add_field => { 
      "logstash_processed" => "true"
      "pipeline_version" => "3.0-simple"
    }
  }
}

output {
  elasticsearch {
    hosts => ["http://10.0.0.91:9200", "http://10.0.0.92:9200", "http://10.0.0.93:9200"]
    index => "logstash-nginx-access-%{+YYYY.MM.dd}"
    template => "/etc/logstash/nginx-template.json"
    template_name => "logstash-nginx-access"
    template_overwrite => false
  }
}

d. Elasticsearch索引模版

# /etc/logstash/nginx-template.json
[root@204-web ~]# cat /etc/logstash/nginx-template.json 
{
  "index_patterns": ["logstash-nginx-access-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "refresh_interval": "30s"
    },
    "mappings": {
      "dynamic": true,
      "properties": {
        "@timestamp": {"type": "date"},
        "@version": {"type": "keyword"},
        "remote_ip": {"type": "ip"},
        "remote_user": {"type": "keyword"},
        "method": {"type": "keyword"},
        "request": {
          "type": "text", 
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "status": {"type": "integer"},
        "body_bytes_sent": {"type": "long"},
        "http_referer": {"type": "keyword"},
        "http_user_agent": {"type": "text"},
        "http_x_forwarded_for": {"type": "keyword"},
        "ua_name": {"type": "keyword"},
        "ua_version": {"type": "keyword"},
        "geoip": {
          "properties": {
            "ip": {"type": "ip"},
            "country_name": {"type": "keyword"},
            "city_name": {"type": "keyword"},
            "location": {"type": "geo_point"}
          }
        },
        "logstash_processed": {"type": "boolean"},
        "log_type": {"type": "keyword"},
        "pipeline_version": {"type": "keyword"}
      }
    }
  },
  "priority": 100
}

e. 启动服务

[root@204-web ~]# systemctl start filebeat
[root@204-web ~]# systemctl enable logstash
[root@204-web ~]# systemctl start logstash
[root@204-web ~]# systemctl enable logstash

f. 验证查看

有问题查看日志即可。