shell脚本如何获取nginx访问日志的最近一个小时数据？

Question

最近主机被人刷了，我写了个脚本，对nginx访问日志分析，发现访问量大于3000的ip就加入到nginx的黑名单中去，但是这个我发现虽然能防住刷主机的垃圾，但是有可能误杀一些正常的访问ip，于是我想只仅处理最近1小时的访问记录，对其ip进行分析，如果访问量太大就封杀，但是搜了半天无果，请大神帮忙看看，谢谢。

nginx_access 日志格式如下：

66.249.79.234 - - [20/Aug/2022:02:18:51 +0800] "GET /healthy-eating/ HTTP/1.1" 200 8448 "-" "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)" "-"

nginx_error 日志格式如下：

2022/08/16 17:48:11 [error] 36492#0: *10499 FastCGI sent in stderr: "Primary script unknown" while reading response header from upstream, client: 54.36.148.246, server: new.taobao.com, request: "GET /link.php?url=http://pinkberrylicious.blogspot.com/2016/07/clone-sony-x-bo-v11-lollipop-51.html HTTP/1.1", upstream: "fastcgi://unix:/dev/shm/php-fpm.sock:", host: "www.taobao.com"

Answer 1

nginx_access

时间格式不好判断，我是按照 nginx_error 的时间格式来的，
(nginx日志的输出格式可以指定的)
日志格式是：66.249.79.234 - - [2022/08/16 17:48:11] "GET /healthy-eating/

lastTime=$(date -d "-1 hour" "+%Y/%m/%d %H:%M:%S")
echo "获取截止时间以后的ip: "$lastTime

awk -vlastTime="$lastTime" '{            # 参数传入指定截止时间
    accessTime=$4" "$5                   # 合并第4、5列，合并后格式为 [2022/08/16 17:48:11]
    accessTime=substr(accessTime, 2, 19) # 去掉两端的中括号
    if (accessTime >= lastTime) {        # 根据时间过滤，只打印截止时间后的ip
        print $1
    }
}' nginx_access.log | # 指定日志文件
    sort |            # uniq对相同但不相邻的ip不会去重，所以需要先sort对ip进行排序
    uniq -c |         # 去重及计数，每行的输出格式为：num ip
    awk '{
        if ($1 > 3000) { # $1为ip出现次数,大于指定次数则输出
            print $2
        }
    }'

按照最初时间格式的改进写法：66.249.79.234 - - [20/Aug/2022:02:18:51 +0800] "GET /healthy-eating

lastTime=$(date -d "-1 hour" "+%Y/%m/%d %H:%M:%S")
echo "获取截止时间以后的ip: "$lastTime

awk -vlastTime="$lastTime" '
BEGIN {
    # 月份转换用
    monthToNum["Jan"] = "01"
    monthToNum["Feb"] = "02"
    monthToNum["Mar"] = "03"
    monthToNum["Apr"] = "04"
    monthToNum["May"] = "05"
    monthToNum["Jun"] = "06"
    monthToNum["Jul"] = "07"
    monthToNum["Aug"] = "08"
    monthToNum["Sep"] = "09"
    monthToNum["Oct"] = "10"
    monthToNum["Nov"] = "11"
    monthToNum["Dec"] = "12"
}
{
    # 匹配出每行 ip 和 时间
    match($0, /^(([0-9]{1,3}\.){3}[0-9]{1,3}).*\[([0-9]{2})\/([a-zA-Z]{3})\/([0-9]{4}):([0-9]{2}:[0-9]{2}:[0-9]{2}) \+0800\]/, arr)
    accessTime = arr[5]"/"monthToNum[arr[4]]"/"arr[3]" "arr[6]; # 拼接成时间格式: Y/m/d H:M:S
    if (accessTime >= lastTime) {        # 根据时间过滤，只打印截止时间后的ip
        print arr[1]
    }
}' nginx_access.log | # 指定日志文件
    sort |            # uniq对相同但不相邻的ip不会去重，所以需要先sort对ip进行排序
    uniq -c |         # 去重及计数，每行的输出格式为：num ip
    awk '{
        if ($1 > 3000) {                    # $1为ip出现次数,大于指定次数则输出
            print $2
        }
    }'

nginx_error

按照楼主提供的日志格式：2022/08/16 17:48:11 xxxxxx client: 54.36.148.246

lastTime=$(date -d "-1 hour" "+%Y/%m/%d %H:%M:%S")
echo "获取截止时间以后的ip: "$lastTime

awk -vlastTime="$lastTime" '{
    # 匹配出每行的时间: arr[1]、ip: arr[2]
    match($0, /^([[:digit:]]{4}\/[[:digit:]]{2}\/[[:digit:]]{2} [[:digit:]]{2}:[[:digit:]]{2}:[[:digit:]]{2}).*client: (([0-9]{1,3}\.){3}[0-9]{1,3})/, arr)
    if (arr[1] >= lastTime) { # 根据时间过滤，只打印截止时间后的ip
        print arr[2];
    }
};'  nginx_error.log |   # 指定日志文件
    sort |               # uniq对相同但不相邻的ip不会去重，所以需要先sort对ip进行排序
    uniq -c |            # 去重及计数，每行的输出格式为：num ip
    awk '{
        if ($1 > 3000) { # $1为ip出现次数,大于指定次数则输出
            print $2
        }
    }'

Answer 2

提供两个方案：

1. 预先按小时分隔日志，然后脚本就取最近一次分隔的日志即可（推荐）

分隔可以通过自带的日志分隔工具logrotate实现，步骤如下：
先创建分隔配置

vim cut-nginx-log

内容为：

/home/wwwlogs/xxx.log {
 su root root
 copytruncate
 rotate 2
 missingok
 dateext
 dateformat .%Y%m%d-%H
 olddir /home/wwwlogs/cut-logs
}

执行测试分隔命令
logrotate -vf cut-nginx-log

最终分隔出来的日志会以xx.log.20201010-10的形式存储在/home/wwwlogs/cut-logs里面

配置一个定时任务，每分钟执行一次分隔命令即可，配合脚本每小时执行一次分析

配置参数参考文档：https://linux.die.net/man/8/l...

2. 每次读nginx日志，然后递归所有行日志，具体看你的日志，假设以空格分隔，那分隔成数组后，一般取第一列的时间，判断是否一小时内，然后再走你的分析逻辑

Answer 3

经过大神的指点，我最终形成了这样一个脚本：

#!/bin/bash

lastTime=$(date -d "-1 hour" "+%Y/%m/%d %H:%M:%S")
#echo "获取截止时间以后的ip: "$lastTime

var1=$(awk -vlastTime="$lastTime" '
BEGIN {
    # 月份转换用
    monthToNum["Jan"] = "01"
    monthToNum["Feb"] = "02"
    monthToNum["Mar"] = "03"
    monthToNum["Apr"] = "04"
    monthToNum["May"] = "05"
    monthToNum["Jun"] = "06"
    monthToNum["Jul"] = "07"
    monthToNum["Aug"] = "08"
    monthToNum["Sep"] = "09"
    monthToNum["Oct"] = "10"
    monthToNum["Nov"] = "11"
    monthToNum["Dec"] = "12"
}
{
    # 匹配出每行 ip 和 时间
    match($0, /^(([0-9]{1,3}\.){3}[0-9]{1,3}).*\[([0-9]{2})\/([a-zA-Z]{3})\/([0-9]{4}):([0-9]{2}:[0-9]{2}:[0-9]{2}) \+0800\]/, arr)
    accessTime = arr[5]"/"monthToNum[arr[4]]"/"arr[3]" "arr[6]; # 拼接成时间格式: Y/m/d H:M:S
    if (accessTime >= lastTime) {        # 根据时间过滤，只打印截止时间后的ip
        print arr[1]
    }
}' /home/www/nginx_access.log | # 指定日志文件
    sort |            # uniq对相同但不相邻的ip不会去重，所以需要先sort对ip进行排序
    uniq -c |         # 去重及计数，每行的输出格式为：num ip
    awk '{
        if ($1 > 100) {                    # $1为ip出现次数,大于指定次数则输出
            print "deny "$2";"
        }
}')

var2=$(awk -vlastTime="$lastTime" '{
    # 匹配出每行的时间: arr[1]、ip: arr[2]
    match($0, /^([[:digit:]]{4}\/[[:digit:]]{2}\/[[:digit:]]{2} [[:digit:]]{2}:[[:digit:]]{2}:[[:digit:]]{2}).*client: (([0-9]{1,3}\.){3}[0-9]{1,3})/, arr)
    if (arr[1] >= lastTime) { # 根据时间过滤，只打印截止时间后的ip
        print arr[2];
    }
};' /home/www/nginx_error.log |   # 指定日志文件
    sort |               # uniq对相同但不相邻的ip不会去重，所以需要先sort对ip进行排序
    uniq -c |            # 去重及计数，每行的输出格式为：num ip
    awk '{
        if ($1 > 50) { # $1为ip出现次数,大于指定次数则输出
            print "deny "$2";"
        }
    }')

if [ ! -z "$var1" ]; then
    var3="$var1"
fi

if [ ! -z "$var3" ]; then
    if [ ! -z "$var2" ]; then
        var3="$var2\n$var3"
    fi
else
    var3="$var2"
fi

if [ ! -z "$var3" ]; then
    echo -e "${var3}"|sort|uniq > /home/www/nginx_ip_black.conf
    #重载nginx服务
    service nginx reload
fi

请大神帮忙看看我这样写的有什么问题没有，谢谢，需要说明的是目前还报我上面说的那个错误，但是不影响程序的运行。