简介
Redis数据库是一个基于内存的 key-value存储系统，现在redis最常用的使用场景就是存储缓存用的数据，在需要高速读/写的场合使用它快速读/写，从而缓解应用数据库的压力，进而提升应用处理能力。许多数据库会提供慢查询日志帮助开发和运维人员定位系统存在的慢操作。所谓慢查询日志就是系统在命令执行前后计算每条命令的执行时间，当超过预设阀值，就将这条命令的相关信息（例如：发生时间，耗时，命令的详细信息）记录下来，Redis也提供了类似的功能。
本文将基于AWS托管的服务来介绍如何展示和分析大规模Redis 集群慢日志。
1.架构说明
通过Event Bridge 产生定时任务（1分钟）触发Lambda 执行，Lambda 通过aws 的sdk找到所有redis 的节点信息并一个个连接索取慢日志信息。 通过慢日志自带的时间戳过滤出最近1分钟产生的日志并上传到RDS数据库保存，最终通过托管的Grafana 展示并统计
2.创建RDS或者Aurora 数据库，并初始化表结构
在控制台创建RDS（过程略），要注意需要开公网访问，因为托管的Grafana需要通过公网访问mysql的数据源。这里会有安全的隐患，如果有自建的能力，可以考虑用VPC内自建的grafana 代替，或者开case 问AWS 后台托管的Grafana IP 地址范围，通过RDS 安全组做限制访问并配置复杂的数据库用户名密码。
创建好表结构 slowlog.detail
MySQL [slowlog]> show create table detail \G
*************************** 1. row ***************************
Table: detail
Create Table: CREATE TABLE `detail` (
`id` bigint(40) NOT NULL AUTO_INCREMENT,
`nodeid` varchar(100) NOT NULL,
`endpoint` varchar(100) NOT NULL,
`slowlog_time` int(20) NOT NULL,
`today` date NOT NULL,
`command` varchar(100) NOT NULL,
`redis_key` varchar(100) NOT NULL,
`duration` int(10) NOT NULL,
PRIMARY KEY (`id`),
KEY `idx_nodeid_time` (`nodeid`,`slowlog_time`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8
3.创建测试用的redis 实例
过程略
4.创建Lambda 函数
代码如下：
import redis
import boto3
import os,sys
import json
import time
from datetime import datetime,timedelta
import pymysql
import socket
import struct
# get system environment
AWS_REGION=os.environ['AWS_REGION']
DB_HOST=os.environ['db_host']
DB_DataBase=os.environ['db_database']
DB_PW=os.environ['db_pw']
DB_USER=os.environ['db_user']
PORT=int(os.environ['port'])
# get current time from internet
# time.time() 方式的时间不准
def RequestTimefromNtp(addr='0.de.pool.ntp.org'):
REF_TIME_1970 = 2208988800 # Reference time
client = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
data = b'\x1b' + 47 * b'\0'
client.sendto( data, (addr, 123))
data, address = client.recvfrom( 1024 )
if data:
t = struct.unpack( '!12I', data )[10]
t -= REF_TIME_1970
return t
# get redis client
def aws_client():
client = boto3.client(
'elasticache',
region_name=AWS_REGION,
)
return client
# get all redis cluster list
def get_redis_cluster_list(client):
clusters_metas = client.describe_cache_clusters()['CacheClusters']
clusters = []
for item in clusters_metas:
clusters.append(item['ReplicationGroupId'])
cluster_list = set(clusters)
return cluster_list
# get each redis cluster meta data
def get_redis_meta(client,groupid):
info={}
response=client.describe_replication_groups(
ReplicationGroupId=groupid)
NodeGroupMembers=response['ReplicationGroups'][0]['NodeGroups'][0]['NodeGroupMembers']
for item in NodeGroupMembers:
k = item['CacheClusterId']
v = item['ReadEndpoint']['Address']
info[k] = v
return info
# get slow log every 1 min
# 本次测试环境密码为空
def get_slow_log_info(endpoint,nodeid,date,port='',pw=''):
now=int(RequestTimefromNtp())
last=int(now - 60)
r=redis.Redis(host=endpoint)
slow_data = r.slowlog_get()
result=[]
for item in slow_data:
if (item['start_time']>= last and item['start_time']<= now ):
try:
command_type = str(item['command']).split('\'')[1].split()[0]
except Exception as err:
command_type = "null"
try:
command_key = str(item['command']).split('\'')[1].split()[1]
except Exception as err:
command_key = "null"
try:
duration = int(item['duration'])
except Exception as err:
duration = 0
data = {
'nodeid' : nodeid,
'endpoint' : endpoint,
'slowlog_time' : int(item['start_time']) ,
'today' : str(date),
'command' : command_type,
'key' : command_key,
'duration' : duration,
}
result.append(data)
return result
# upload data to RDS
def upload_logs(result,DB_HOST,DB_USER,DB_PW,DB_DataBase,PORT):
db = pymysql.connect(host=DB_HOST,
user=DB_USER,
password=DB_PW,
database=DB_DataBase,
port=PORT,
cursorclass=pymysql.cursors.DictCursor)
cur = db.cursor()
for item in result:
nodeid=item['nodeid']
endpoint=item['endpoint']
slowlog_time=item['slowlog_time']
today=item['today']
command=item['command']
redis_key=item['key']
duration=item['duration']
SQL = "insert into slowlog.detail (nodeid,endpoint,slowlog_time,today,command,redis_key,duration) values('%s','%s',%d,'%s','%s','%s',%d)" % \
(nodeid,endpoint,slowlog_time,today,command,redis_key,duration)
cur.execute(SQL)
db.commit()
db.close()
def lambda_handler(event, context):
date = datetime.now().strftime("%Y-%m-%d")
client=aws_client()
cluster_lists=get_redis_cluster_list(client)
for item in cluster_lists:
info = get_redis_meta(client,item)
for k in info.keys():
result=[]
result=get_slow_log_info(info[k],k,date)
print(result)
upload_logs(result,DB_HOST,DB_USER,DB_PW,DB_DataBase,PORT)
python 代码依赖redis 和pymysql 包，操作步骤如下：
1） 在本地Linux环境（x86）安装redis 和pymysql 两个包，建议通过virtual env 安装, 如果是pip安装则包的参考路径为 /usr/local/lib/python3.7/site-packages
2） 把redis 目录 和pymysql 目录拷贝到临时目录/tmp
3） 下载上面的代码到/tmp
4） 打包 zip -r lambda.zip pymysql redis lambda_function.py
5） 在控制台通过S3 或者本地上传
5.配置Lambda 函数
1）给Lambda 一个执行的role，policy 见下图
2） 配置环境变量
演示为了方便，使用lambda环境变量存数据库访问连接串，建议生产环境中使用secret manager 存取
3） 配置Lambda 使用VPC下的private 子网（确保子网配置了NAT-gw）
6.配置EventBridge
打开控制台，跳转到EventBridge ，新建一个rule
target 选中我们的lambda 函数
7.配置Grafana
1）首先配置一个新的workspace
2）使用托管的SSO 来配置用户密码
记录下URL
2）到SSO 创建用户
3）添加SSO 用户到Grafana
4）配置数据源
8.测试方案
1）修改redis 参数，确保能记录慢日志
修改 slowlog-log-slower-than = 1
2）使用redis-benchmark 模拟真实workload
过程可以观察lambda 的日志输出或者登陆数据库查看是否有数据插入
./redis-benchmark -h test-cluster-1-001.t5tzux.0001.apse1.cache.amazonaws.com -n 1000000 -c 20 -t hset,lpush,rpush,sadd,zadd,set,get -q > test1.log &
./redis-benchmark -h test-cluster-2-001.t5tzux.0001.apse1.cache.amazonaws.com -n 1000000 -c 20 -t hset,lpush,rpush,sadd,zadd,set,get -q > test2.log &
./redis-benchmark -h test-cluster-3-001.t5tzux.0001.apse1.cache.amazonaws.com -n 1000000 -c 20 -t hset,lpush,rpush,sadd,zadd,set,get -q > test3.log &
./redis-benchmark -h test-cluster-4-001.t5tzux.0001.apse1.cache.amazonaws.com -n 1000000 -c 20 -t hset,lpush,rpush,sadd,zadd,set,get -q > test4.log &
3） 在Grafana 作图观察分析
第一条SQL（Slow_log_details） 是看日志明细
SELECT
slowlog_time AS "time",
nodeid,
endpoint,
command,
redis_key,
duration as "duration(ms)"
FROM detail
WHERE
$__unixEpochFilter(slowlog_time)
ORDER BY slowlog_time
第二条SQL（slow_log_trend）查看总体趋势分析
SELECT
from_unixtime(slowlog_time) as time,
count(*) as cnt,
nodeid
FROM detail
group by time,nodeid
4） 查看效果图
参考文档：
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/elasticache.html
本篇作者
