情况一：普通格式日志

架构图

filebeat收集Nginx日志输出到Redis；

logstash读取Redis内的缓存日志经过Grok语法转换为json格式；

Kibana进行展示。

Redis

# 这里Redis不是重点，yum安装
yum install redis -y

# 修改Redis配置文件，允许外网访问
vim /etc/redis.conf
bind 0.0.0.0

# 查看
[root@002 ~]#  grep "^bind" /etc/redis.conf
bind 0.0.0.0

# 启动Redis
systemctl start redis

# 测试访问Redis
[root@002 ~]# redis-cli -h 172.16.49.130
172.16.49.130:6379>

Nginx

Nginx日志配置文件不做任何更改，为普通格式

filebeat配置

# 配置文件
cat > /etc/filebeat/filebeat.yml << 'EOF'
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log

output.redis:
  hosts: ["172.16.49.130:6379"]
  key: "nginx-access"

setup.ilm.enabled: false
setup.template.enabled: false
EOF

# 重启filebeat
systemctl restart filebeat
systemctl status filebeat

ab压测工具

# 安装ab压测工具
yum install httpd-tools -y

# 压测Nginx 1000次，产生访问日志
[root@002 ~]# ab -c 10 -n 1000 http://172.16.49.130/
Benchmarking 172.16.49.130 (be patient)
Completed 100 requests
Completed 200 requests
Completed 300 requests
Completed 400 requests
Completed 500 requests
Completed 600 requests
Completed 700 requests
Completed 800 requests
Completed 900 requests
Completed 1000 requests
Finished 1000 requests

查看Redis数据

此时查看Redis中的数据，已经有了自定义的“nginx-access”key，长度也是1000，日志也已经存储完成。

[root@002 filebeat]# redis-cli
127.0.0.1:6379> keys *
1) "nginx-access"
127.0.0.1:6379> LLEN nginx-access
(integer) 1000
127.0.0.1:6379> LRANGE nginx-access 1 2
1) "{\"@timestamp\":\"2024-05-22T10:36:46.852Z\",\"@metadata\":{\"beat\":\"filebeat\",\"type\":\"_doc\",\"version\":\"7.9.1\"},\"host\":{\"name\":\"002\"},\"agent\":{\"ephemeral_id\":\"ed8d6ea6-1108-4704-842d-bd79833e4d79\",\"id\":\"c1d7ab2b-3579-4917-8d9c-65266d75d7ca\",\"name\":\"002\",\"type\":\"filebeat\",\"version\":\"7.9.1\",\"hostname\":\"002\"},\"log\":{\"offset\":98,\"file\":{\"path\":\"/var/log/nginx/access.log\"}},\"message\":\"172.16.49.130 - - [22/May/2024:18:36:37 +0800] \\\"GET / HTTP/1.0\\\" 200 615 \\\"-\\\" \\\"ApacheBench/2.3\\\" \\\"-\\\"\",\"input\":{\"type\":\"log\"},\"ecs\":{\"version\":\"1.5.0\"}}"
2) "{\"@timestamp\":\"2024-05-22T10:36:46.852Z\",\"@metadata\":{\"beat\":\"filebeat\",\"type\":\"_doc\",\"version\":\"7.9.1\"},\"host\":{\"name\":\"002\"},\"log\":{\"offset\":196,\"file\":{\"path\":\"/var/log/nginx/access.log\"}},\"message\":\"172.16.49.130 - - [22/May/2024:18:36:37 +0800] \\\"GET / HTTP/1.0\\\" 200 615 \\\"-\\\" \\\"ApacheBench/2.3\\\" \\\"-\\\"\",\"input\":{\"type\":\"log\"},\"agent\":{\"ephemeral_id\":\"ed8d6ea6-1108-4704-842d-bd79833e4d79\",\"id\":\"c1d7ab2b-3579-4917-8d9c-65266d75d7ca\",\"name\":\"002\",\"type\":\"filebeat\",\"version\":\"7.9.1\",\"hostname\":\"002\"},\"ecs\":{\"version\":\"1.5.0\"}}"

Logstash准备

安装

# 下载Logstash RPM安装包
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.9.1.rpm

# 安装logstash
rpm -ivh logstash-7.9.1.rpm

配置文件

# 写入logstash子配置文件
# stdout {} 表示开启logstash的调试模式，初次使用添加调试，正式使用的时候不需要此配置
cat > /etc/logstash/conf.d/redis.conf << 'EOF'
input {
  redis {
    host => "172.16.49.130"
    port =>"6379"
    db => "0"
    key => "nginx-access"
    data_type => "list"
  }
}

output {
   stdout {}
   elasticsearch {
     hosts => "http://172.16.49.130:9200"
     manage_template => false
     index =>"nginx-access-%{+yyyy.MM}"
  }
}
EOF

启动测试

# 前台启动检测配置是否一切正常
[root@002 ~]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/redis.conf
。。。。
                  "id" => "c1d7ab2b-3579-4917-8d9c-65266d75d7ca",
                "type" => "filebeat",
            "hostname" => "002",
        "ephemeral_id" => "ed8d6ea6-1108-4704-842d-bd79833e4d79",
                "name" => "002"
    },
           "ecs" => {
        "version" => "1.5.0"
    },
          "host" => {
        "name" => "002"
    }
}

如上显示，已经读取到了Redis内的数据，表示成功。

此时，ES中也已经有了1000条数据。

查看Redis

此时的Redis数据取完已经为空了。

[root@002 ~]# redis-cli -h 127.0.0.1
127.0.0.1:6379> LLEN access-nginx
(integer) 0
127.0.0.1:6379>

Logstash的Grok转换

# logstash配置文件加入Grok转换，准备一个新的配置子文件
cat > /etc/logstash/conf.d/redis_grok.conf << 'EOF'
input {
  redis {
    host => "172.16.49.130"
    port =>"6379"
    db => "0"
    key => "nginx-access"
    data_type => "list"
  }
}

filter {
  grok {
     match => { "message" => "%{IP:client_ip} - - \[%{HTTPDATE:access_time}\] \"%{DATA:method} %{DATA:URL} %{DATA:http}\" %{NUMBER:status_code:long} %{NUMBER:response_bytes:long} \"(-|%{DATA:referrer})\" \"(-|%{DATA:user_agent})\" \"(-|%{IP:x_forwarded})\"" }
     remove_field => ["message"]
  }
}

output {
   stdout {}
   elasticsearch {
     hosts => "http://172.16.49.130:9200"
     manage_template => false
     index =>"nginx-access-%{+yyyy.MM}"
  }
}
EOF

# 使用新的子配置文件启动logstash
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/redis_grok.conf

# 使用ab压测工具生成一些日志信息
ab -c 1 -n 5 http://172.16.49.130/

此时，已经可以看到logstash通过Grok转换了普通的日子格式

补充

logstash配置文件里的Grok语法规则可以使用一个字段“HTTPD_COMBINEDLOG”来代替，这是为Apache准备的，Nginx也可以使用，但是匹配不是100%完整，会少最后一个日志字段，可选使用。

input {
  redis {
    host => "172.16.49.130"
    port =>"6379"
    db => "0"
    key => "nginx-access"
    data_type => "list"
  }
}

filter {
  grok {
     match => { "message" => "%{HTTPD_COMBINEDLOG}" }
     remove_field => ["message","log","ecs"]
  }
}

output {
   stdout {}
   elasticsearch {
     hosts => "http://172.16.49.130:9200"
     manage_template => false
     index =>"nginx-access-%{+yyyy.MM}"
  }
}

ES查看

此时查看ES的数据，也已经通过转换后的json格式存储

情况二：JSON格式日志

如果是JSON格式的日志，logstash从Redis中读取完日志后，就不需要再通过Grok规则来转换了。

目录CONTENT

logstash读取Redis缓存数据并转换格式