1
0
mirror of https://github.com/go-kratos/kratos.git synced 2025-01-10 00:29:01 +02:00

Merge branch 'master' of github.com:bilibili/kratos into fix/cpu_nums

This commit is contained in:
longXboy 2019-10-15 16:15:23 +08:00
commit 3ded57a9d7
36 changed files with 3887 additions and 103 deletions

View File

@ -2,6 +2,10 @@ language: go
go:
- 1.12.x
- 1.13.x
services:
- docker
# Only clone the most recent commit.
git:
@ -16,6 +20,13 @@ env:
- DEPLOY_ENV=dev
- DISCOVERY_NODES=127.0.0.1:7171
- HTTP_PERF=tcp://0.0.0.0:0
- DOCKER_COMPOSE_VERSION=1.24.1
before_install:
- sudo rm /usr/local/bin/docker-compose
- curl -L https://github.com/docker/compose/releases/download/${DOCKER_COMPOSE_VERSION}/docker-compose-`uname -s`-`uname -m` > docker-compose
- chmod +x docker-compose
- sudo mv docker-compose /usr/local/bin
# Skip the install step. Don't `go get` dependencies. Only build with the code
# in vendor/

View File

@ -35,3 +35,7 @@
* [genbts](kratos-genbts.md)
* [限流bbr](ratelimit.md)
* [熔断breaker](breaker.md)
* [UT单元测试](ut.md)
* [testcli UT运行环境构建工具](ut-testcli.md)
* [testgen UT代码自动生成器](ut-testgen.md)
* [support UT周边辅助工具](ut-support.md)

497
doc/wiki-cn/ut-support.md Normal file
View File

@ -0,0 +1,497 @@
## 单元测试辅助工具
在单元测试中,我们希望每个测试用例都是独立的。这时候就需要Stub, Mock, Fakes等工具来帮助我们进行用例和依赖之间的隔离。
同时通过对错误情况的 Mock 也可以帮我们检查代码多个分支结果,从而提高覆盖率。
以下工具已加入到 Kratos 框架 go modules,可以借助 testgen 代码生成器自动生成部分工具代码,请放心食用。更多使用方法还欢迎大家多多探索。
### GoConvey
GoConvey是一套针对golang语言的BDD类型的测试框架。提供了良好的管理和执行测试用例的方式,包含丰富的断言函数,而且同时有测试执行和报告Web界面的支持。
#### 使用特性
为了更好的使用 GoConvey 来编写和组织测试用例,需要注意以下几点特性:
1. Convey方法和So方法的使用
> - Convey方法声明了一种规格的组织,每个组织内包含一句描述和一个方法。在方法内也可以嵌套其他Convey语句和So语句。
```Go
// 顶层Convey方法,需引入*testing.T对象
Convey(description string, t *testing.T, action func())
// 其他嵌套Convey方法,无需引入*testing.T对象
Convey(description string, action func())
```
注:同一Scope下的Convey语句描述不可以相同!
> - So方法是断言方法,用于对执行结果进行比对。GoConvey官方提供了大量断言,同时也可以自定义自己的断言([戳这里了解官方文档](https://github.com/smartystreets/goconvey/wiki/Assertions))
```Go
// A=B断言
So(A, ShouldEqual, B)
// A不为空断言
So(A, ShouldNotBeNil)
```
2. 执行次序
> 假设有以下Convey伪代码,执行次序将为A1B2A1C3。将Convey方法类比树的结点的话,整体执行类似树的遍历操作。
> 所以Convey A部分可在组织测试用例时,充当“Setup”的方法。用于初始化等一些操作。
```Go
Convey伪代码
Convey A
So 1
Convey B
So 2
Convey C
So 3
```
3. Reset方法
> GoConvey提供了Reset方法来进行“Teardown”的操作。用于执行完测试用例后一些状态的回收,连接关闭等操作。Reset方法不可与顶层Convey语句在同层。
```Go
// Reset
Reset func(action func())
```
假设有以下带有Reset方法的伪代码,同层Convey语句执行完后均会执行同层的Reset方法。执行次序为A1B2C3EA1D4E。
```Go
Convey A
So 1
Convey B
So 2
Convey C
So 3
Convey D
So 4
Reset E
```
4. 自然语言逻辑到测试用例的转换
> 在了解了Convey方法的特性和执行次序后,我们可以通过这些性质把对一个方法的测试用例按照日常逻辑组织起来。尤其建议使用Given-When-Then的形式来组织
> - 比较直观的组织示例
```Go
Convey("Top-level", t, func() {
// Setup 工作,在本层内每个Convey方法执行前都会执行的部分:
db.Open()
db.Initialize()
Convey("Test a query", func() {
db.Query()
// TODO: assertions here
})
Convey("Test inserts", func() {
db.Insert()
// TODO: assertions here
})
Reset(func() {
// Teardown工作,在本层内每个Convey方法执行完后都会执行的部分:
db.Close()
})
})
```
> - 定义单独的包含Setup和Teardown的帮助方法
```Go
package main
import (
"database/sql"
"testing"
_ "github.com/lib/pq"
. "github.com/smartystreets/goconvey/convey"
)
// 帮助方法,将原先所需的处理方法以参数(f)形式传入
func WithTransaction(db *sql.DB, f func(tx *sql.Tx)) func() {
return func() {
// Setup工作
tx, err := db.Begin()
So(err, ShouldBeNil)
Reset(func() {
// Teardown工作
/* Verify that the transaction is alive by executing a command */
_, err := tx.Exec("SELECT 1")
So(err, ShouldBeNil)
tx.Rollback()
})
// 调用传入的闭包做实际的事务处理
f(tx)
}
}
func TestUsers(t *testing.T) {
db, err := sql.Open("postgres", "postgres://localhost?sslmode=disable")
if err != nil {
panic(err)
}
Convey("Given a user in the database", t, WithTransaction(db, func(tx *sql.Tx) {
_, err := tx.Exec(`INSERT INTO "Users" ("id", "name") VALUES (1, 'Test User')`)
So(err, ShouldBeNil)
Convey("Attempting to retrieve the user should return the user", func() {
var name string
data := tx.QueryRow(`SELECT "name" FROM "Users" WHERE "id" = 1`)
err = data.Scan(&name)
So(err, ShouldBeNil)
So(name, ShouldEqual, "Test User")
})
}))
}
```
#### 使用建议
强烈建议使用 [testgen](https://github.com/bilibili/kratos/blob/master/doc/wiki-cn/ut-testgen.md) 进行测试用例的生成,生成后每个方法将包含一个符合以下规范的正向用例。
用例规范:
1. 每个方法至少包含一个测试方法(命名为Test[PackageName][FunctionName])
2. 每个测试方法包含一个顶层Convey语句,仅在此引入admin *testing.T类型的对象,在该层进行变量声明。
3. 每个测试方法不同的用例用Convey方法组织
4. 每个测试用例的一组断言用一个Convey方法组织
5. 使用convey.C保持上下文一致
### MonkeyPatching
#### 特性和使用条件
1. Patch()对任何无接收者的方法均有效
2. PatchInstanceMethod()对有接收者的包内/私有方法无法工作(因使用到了反射机制)。可以采用给私有方法的下一级打补丁,或改为无接收者的方法,或将方法转为公有
#### 适用场景(建议)
项目代码中上层对下层包依赖时,下层包方法Mock(例如service层对dao层方法依赖时)
基础库(MySql, Memcache, Redis)错误Mock
其他标准库,基础库以及第三方包方法Mock
#### 使用示例
1. 上层包对下层包依赖示例
Service层对Dao层依赖:
```GO
// 原方法
func (s *Service) realnameAlipayApply(c context.Context, mid int64) (info *model.RealnameAlipayApply, err error) {
if info, err = s.mbDao.RealnameAlipayApply(c, mid); err != nil {
return
}
...
return
}
// 测试方法
func TestServicerealnameAlipayApply(t *testing.T) {
convey.Convey("realnameAlipayApply", t, func(ctx convey.C) {
...
ctx.Convey("When everything goes positive", func(ctx convey.C) {
guard := monkey.PatchInstanceMethod(reflect.TypeOf(s.mbDao), "RealnameAlipayApply", func(_ *dao.Dao, _ context.Context, _ int64) (*model.RealnameAlipayApply, error) {
return nil, nil
})
defer guard.Unpatch()
info, err := s.realnameAlipayApply(c, mid)
ctx.Convey("Then err should be nil,info should not be nil", func(ctx convey.C) {
ctx.So(info, convey.ShouldNotBeNil)
ctx.So(err, convey.ShouldBeNil)
})
})
})
}
```
2. 基础库错误Mock示例
```Go
// 原方法(部分)
func (d *Dao) BaseInfoCache(c context.Context, mid int64) (info *model.BaseInfo, err error) {
...
conn := d.mc.Get(c)
defer conn.Close()
item, err := conn.Get(key)
if err != nil {
log.Error("conn.Get(%s) error(%v)", key, err)
return
}
...
return
}
// 测试方法(错误Mock部分)
func TestDaoBaseInfoCache(t *testing.T) {
convey.Convey("BaseInfoCache", t, func(ctx convey.C) {
...
Convey("When conn.Get gets error", func(ctx convey.C) {
guard := monkey.PatchInstanceMethod(reflect.TypeOf(d.mc), "Get", func(_ *memcache.Pool, _ context.Context) memcache.Conn {
return memcache.MockWith(memcache.ErrItemObject)
})
defer guard.Unpatch()
_, err := d.BaseInfoCache(c, mid)
ctx.Convey("Error should be equal to memcache.ErrItemObject", func(ctx convey.C) {
ctx.So(err, convey.ShouldEqual, memcache.ErrItemObject)
})
})
})
}
```
#### 注意事项
- Monkey非线程安全
- Monkey无法针对Inline方法打补丁,在测试时可以使用go test -gcflags=-l来关闭inline编译的模式(一些简单的go inline介绍戳这里)
- Monkey在一些面向安全不允许内存页写和执行同时进行的操作系统上无法工作
- 更多详情请戳:https://github.com/bouk/monkey
### Gock——HTTP请求Mock工具
#### 特性和使用条件
#### 工作原理
1. 截获任意通过 http.DefaultTransport或者自定义http.Transport对外的http.Client请求
2. 以“先进先出”原则将对外需求和预定义好的HTTP Mock池中进行匹配
3. 如果至少一个Mock被匹配,将按照2中顺序原则组成Mock的HTTP返回
4. 如果没有Mock被匹配,若实际的网络可用,将进行实际的HTTP请求。否则将返回错误
#### 特性
- 内建帮助工具实现JSON/XML简单Mock
- 支持持久的、易失的和TTL限制的Mock
- 支持HTTP Mock请求完整的正则表达式匹配
- 可通过HTTP方法,URL参数,请求头和请求体匹配
- 可扩展和可插件化的HTTP匹配规则
- 具备在Mock和实际网络模式之间切换的能力
- 具备过滤和映射HTTP请求到正确的Mock匹配的能力
- 支持映射和过滤可以更简单的掌控Mock
- 通过使用http.RoundTripper接口广泛兼容HTTP拦截器
- 可以在任意net/http兼容的Client上工作
- 网络延迟模拟(beta版本)
- 无其他依赖
#### 适用场景(建议)
任何需要进行HTTP请求的操作,建议全部用Gock进行Mock,以减少对环境的依赖。
使用示例:
1. net/http 标准库 HTTP 请求Mock
```Go
import gock "gopkg.in/h2non/gock.v1"
// 原方法
func (d *Dao) Upload(c context.Context, fileName, fileType string, expire int64, body io.Reader) (location string, err error) {
...
resp, err = d.bfsClient.Do(req) //d.bfsClient类型为*http.client
...
if resp.StatusCode != http.StatusOK {
...
}
header = resp.Header
code = header.Get("Code")
if code != strconv.Itoa(http.StatusOK) {
...
}
...
return
}
// 测试方法
func TestDaoUpload(t *testing.T) {
convey.Convey("Upload", t, func(ctx convey.C) {
...
// d.client 类型为 *http.client 根据Gock包描述需要设置http.Client的Transport情况。也可在TestMain中全局设置,则所有的HTTP请求均通过Gock来解决
d.client.Transport = gock.DefaultTransport // !注意:进行httpMock前需要对http 请求进行拦截,否则Mock失败
// HTTP请求状态和Header都正确的Mock
ctx.Convey("When everything is correct", func(ctx convey.C) {
httpMock("PUT", url).Reply(200).SetHeaders(map[string]string{
"Code": "200",
"Location": "SomePlace",
})
location, err := d.Upload(c, fileName, fileType, expire, body)
ctx.Convey("Then err should be nil.location should not be nil.", func(ctx convey.C) {
ctx.So(err, convey.ShouldBeNil)
ctx.So(location, convey.ShouldNotBeNil)
})
})
...
// HTTP请求状态错误Mock
ctx.Convey("When http request status != 200", func(ctx convey.C) {
d.client.Transport = gock.DefaultTransport
httpMock("PUT", url).Reply(404)
_, err := d.Upload(c, fileName, fileType, expire, body)
ctx.Convey("Then err should not be nil", func(ctx convey.C) {
ctx.So(err, convey.ShouldNotBeNil)
})
})
// HTTP请求Header中Code值错误Mock
ctx.Convey("When http request Code in header != 200", func(ctx convey.C) {
d.client.Transport = gock.DefaultTransport
httpMock("PUT", url).Reply(404).SetHeaders(map[string]string{
"Code": "404",
"Location": "SomePlace",
})
_, err := d.Upload(c, fileName, fileType, expire, body)
ctx.Convey("Then err should not be nil", func(ctx convey.C) {
ctx.So(err, convey.ShouldNotBeNil)
})
})
// 由于同包内有其他进行实际HTTP请求的测试。所以再每次用例结束后,进行现场恢复(关闭Gock设置默认的Transport)
ctx.Reset(func() {
gock.OffAll()
d.client.Transport = http.DefaultClient.Transport
})
})
}
func httpMock(method, url string) *gock.Request {
r := gock.New(url)
r.Method = strings.ToUpper(method)
return r
}
```
2. blademaster库HTTP请求Mock
```Go
// 原方法
func (d *Dao) SendWechatToGroup(c context.Context, chatid, msg string) (err error) {
...
if err = d.client.Do(c, req, &res); err != nil {
...
}
if res.Code != 0 {
...
}
return
}
// 测试方法
func TestDaoSendWechatToGroup(t *testing.T) {
convey.Convey("SendWechatToGroup", t, func(ctx convey.C) {
...
// 根据Gock包描述需要设置bm.Client的Transport情况。也可在TestMain中全局设置,则所有的HTTP请求均通过Gock来解决。
// d.client 类型为 *bm.client
d.client.SetTransport(gock.DefaultTransport) // !注意:进行httpMock前需要对http 请求进行拦截,否则Mock失败
// HTTP请求状态和返回内容正常Mock
ctx.Convey("When everything gose postive", func(ctx convey.C) {
httpMock("POST", _sagaWechatURL+"/appchat/send").Reply(200).JSON(`{"code":0,"message":"0"}`)
err := d.SendWechatToGroup(c, d.c.WeChat.ChatID, msg)
...
})
// HTTP请求状态错误Mock
ctx.Convey("When http status != 200", func(ctx convey.C) {
httpMock("POST", _sagaWechatURL+"/appchat/send").Reply(404)
err := d.SendWechatToGroup(c, d.c.WeChat.ChatID, msg)
...
})
// HTTP请求返回值错误Mock
ctx.Convey("When http response code != 0", func(ctx convey.C) {
httpMock("POST", _sagaWechatURL+"/appchat/send").Reply(200).JSON(`{"code":-401,"message":"0"}`)
err := d.SendWechatToGroup(c, d.c.WeChat.ChatID, msg)
...
})
// 由于同包内有其他进行实际HTTP请求的测试。所以再每次用例结束后,进行现场恢复(关闭Gock设置默认的Transport)。
ctx.Reset(func() {
gock.OffAll()
d.client.SetTransport(http.DefaultClient.Transport)
})
})
}
func httpMock(method, url string) *gock.Request {
r := gock.New(url)
r.Method = strings.ToUpper(method)
return r
}
```
#### 注意事项
- Gock不是完全线程安全的
- 如果执行并发代码,在配置Gock和解释定制的HTTP clients时,要确保Mock已经事先声明好了来避免不需要的竞争机制
- 更多详情请戳:https://github.com/h2non/gock
### GoMock
#### 使用条件
只能对公有接口(interface)定义的代码进行Mock,并仅能在测试过程中进行
#### 使用方法
- 官方安装使用步骤
```shell
## 获取GoMock包和自动生成Mock代码工具mockgen
go get github.com/golang/mock/gomock
go install github.com/golang/mock/mockgen
## 生成mock文件
## 方法1:生成对应文件下所有interface
mockgen -source=path/to/your/interface/file.go
## 方法2:生成对应包内指定多个interface,并用逗号隔开
mockgen database/sql/driver Conn,Driver
## 示例:
mockgen -destination=$GOPATH/kratos/app/xxx/dao/dao_mock.go -package=dao kratos/app/xxx/dao DaoInterface
```
- testgen 使用步骤(GoMock生成功能已集成在Creater工具中,无需额外安装步骤即可直接使用)
```shell
## 直接给出含有接口类型定义的包路径,生成Mock文件将放在包目录下一级mock/pkgName_mock.go中
./creater --m mock absolute/path/to/your/pkg
```
- 测试代码内使用方法
```Go
// 测试用例内直接使用
// 需引入的包
import (
...
"github.com/otokaze/mock/gomock"
...
)
func TestPkgFoo(t *testing.T) {
convey.Convey("Foo", t, func(ctx convey.C) {
...
ctx.Convey("Mock Interface to test", func(ctx convey.C) {
// 1. 使用gomock.NewController新增一个控制器
mockCtrl := gomock.NewController(t)
// 2. 测试完成后关闭控制器
defer mockCtrl.Finish()
// 3. 以控制器为参数生成Mock对象
yourMock := mock.NewMockYourClient(mockCtrl)
// 4. 使用Mock对象替代原代码中的对象
yourClient = yourMock
// 5. 使用EXPECT().方法名(方法参数).Return(返回值)来构造所需输入/输出
yourMock.EXPECT().YourMethod(gomock.Any()).Return(nil)
res:= Foo(params)
...
})
...
})
}
// 可以利用Convey执行顺序方式适当调整以简化代码
func TestPkgFoo(t *testing.T) {
convey.Convey("Foo", t, func(ctx convey.C) {
...
mockCtrl := gomock.NewController(t)
yourMock := mock.NewMockYourClient(mockCtrl)
ctx.Convey("Mock Interface to test1", func(ctx convey.C) {
yourMock.EXPECT().YourMethod(gomock.Any()).Return(nil)
...
})
ctx.Convey("Mock Interface to test2", func(ctx convey.C) {
yourMock.EXPECT().YourMethod(args).Return(res)
...
})
...
ctx.Reset(func(){
mockCtrl.Finish()
})
})
}
```
#### 适用场景(建议)
1. gRPC中的Client接口
2. 也可改造现有代码构造Interface后使用(具体可配合Creater的功能进行Interface和Mock的生成)
3. 任何对接口中定义方法依赖的场景
#### 注意事项
- 如有Mock文件在包内,在执行单元测试时Mock代码会被识别进行测试。请注意Mock文件的放置。
- 更多详情请戳:https://github.com/golang/mock

154
doc/wiki-cn/ut-testcli.md Normal file
View File

@ -0,0 +1,154 @@
## testcli UT运行环境构建工具
基于 docker-compose 实现跨平台跨语言环境的容器依赖管理方案,以解决运行ut场景下的 (mysql, redis, mc)容器依赖问题。
*这个是testing/lich的二进制工具版本(Go请直接使用库版本:github.com/bilibili/kratos/pkg/testing/lich)*
### 功能和特性
- 自动读取 test 目录下的 yaml 并启动依赖
- 自动导入 test 目录下的 DB 初始化 SQL
- 提供特定容器内的 healthcheck (mysql, mc, redis)
- 提供一站式解决 UT 服务依赖的工具版本 (testcli)
### 编译安装
*使用本工具/库需要前置安装好 docker & docker-compose@v1.24.1^*
#### Method 1. With go get
```shell
go get -u github.com/bilibili/kratos/tool/testcli
$GOPATH/bin/testcli -h
```
#### Method 2. Build with Go
```shell
cd github.com/bilibili/kratos/tool/testcli
go build -o $GOPATH/bin/testcli
$GOPATH/bin/testcli -h
```
#### Method 3. Import with Kratos pkg
```Go
import "github.com/bilibili/kratos/pkg/testing/lich"
```
### 构建数据
#### Step 1. create docker-compose.yml
创建依赖服务的 docker-compose.yml,并把它放在项目路径下的 test 文件夹下面。例如:
```shell
mkdir -p $YOUR_PROJECT/test
```
```yaml
version: "3.7"
services:
db:
image: mysql:5.6
ports:
- 3306:3306
environment:
- MYSQL_ROOT_PASSWORD=root
volumes:
- .:/docker-entrypoint-initdb.d
command: [
'--character-set-server=utf8',
'--collation-server=utf8_unicode_ci'
]
redis:
image: redis
ports:
- 6379:6379
```
一般来讲,我们推荐在项目根目录创建 test 目录,里面存放描述服务的yml,以及需要初始化的数据(database.sql等)。
同时也需要注意,正确的对容器内服务进行健康检测,testcli会在容器的health状态执行UT,其实我们也内置了针对几个较为通用镜像(mysql mariadb mc redis)的健康检测,也就是不写也没事(^^;;
#### Step 2. export database.sql
构造初始化的数据(database.sql等),当然也把它也在 test 文件夹里。
```sql
CREATE DATABASE IF NOT EXISTS `YOUR_DATABASE_NAME`;
SET NAMES 'utf8';
USE `YOUR_DATABASE_NAME`;
CREATE TABLE IF NOT EXISTS `YOUR_TABLE_NAME` (
`id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
PRIMARY KEY (`id`),
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='YOUR_TABLE_NAME';
```
这里需要注意,在创建库/表的时候尽量加上 IF NOT EXISTS,以给予一定程度的容错,以及 SET NAMES 'utf8'; 用于解决客户端连接乱码问题。
#### Step 3. change your project mysql config
```toml
[mysql]
addr = "127.0.0.1:3306"
dsn = "root:root@tcp(127.0.0.1:3306)/YOUR_DATABASE?timeout=1s&readTimeout=1s&writeTimeout=1s&parseTime=true&loc=Local&charset=utf8mb4,utf8"
active = 20
idle = 10
idleTimeout ="1s"
queryTimeout = "1s"
execTimeout = "1s"
tranTimeout = "1s"
```
*Step 1* 我们已经指定了服务对外暴露的端口为3306(这当然也可以是你指定的任何值),那理所应当的我们也要修改项目连接数据库的配置~
Great! 至此你已经完成了运行所需要用到的数据配置,接下来就来运行它。
### 运行
开头也说过本工具支持两种运行方式:testcli 二进制工具版本和 go package 源码包,业务方可以根据需求场景进行选择。
#### Method 1. With testcli tool
*已支持的 flag: -f,--nodown,down,run*
- -f,指定 docker-compose.yaml 文件路径,默认为当前目录下。
- --nodown,指定是否在UT执行完成后保留容器,以供下次复用。
- down,teardown 销毁当前项目下这个 compose 文件产生的容器。
- run,运行你当前语言的单测执行命令(如:golang为 go test -v ./)
example:
```shell
testcli -f ../../test/docker-compose.yaml run go test -v ./
```
#### Method 2. Import with Kratos pkg
- Step1. 在 Dao|Service 层中的 TestMain 单测主入口中,import "github.com/bilibili/kratos/pkg/testing/lich" 引入testcli工具的go库版本。
- Step2. 使用 flag.Set("f", "../../test/docker-compose.yaml") 指定 docker-compose.yaml 文件的路径。
- Step3. 在 flag.Parse() 后即可使用 lich.Setup() 安装依赖&初始化数据(注意测试用例执行结束后 lich.Teardown() 回收下~)
- Step4. 运行 `go test -v ./ `看看效果吧~
example:
```Go
package dao
import (
"flag"
"os"
"strings"
"testing"
"github.com/bilibili/kratos/pkg/conf/paladin"
"github.com/bilibili/kratos/pkg/testing/lich"
)
var (
d *Dao
)
func TestMain(m *testing.M) {
flag.Set("conf", "../../configs")
flag.Set("f", "../../test/docker-compose.yaml")
flag.Parse()
if err := paladin.Init(); err != nil {
panic(err)
}
if err := lich.Setup(); err != nil {
panic(err)
}
defer lich.Teardown()
d = New()
if code := m.Run(); code != 0 {
panic(code)
}
}
```
## 注意
因为启动mysql容器较为缓慢,健康检测的机制会重试3次,每次暂留5秒钟,基本在10s内mysql就能从creating到服务正常启动!
当然你也可以在使用 testcli 时加上 --nodown,使其不用每次跑都新建容器,只在第一次跑的时候会初始化容器,后面都进行复用,这样速度会快很多。
成功启动后就欢乐奔放的玩耍吧~ Good Lucky!

52
doc/wiki-cn/ut-testgen.md Normal file
View File

@ -0,0 +1,52 @@
## testgen UT代码自动生成器
解放你的双手,让你的UT一步到位!
### 功能和特性
- 支持生成 Dao|Service 层UT代码功能(每个方法包含一个正向用例)
- 支持生成 Dao|Service 层测试入口文件dao_test.go, service_test.go(用于控制初始化,控制测试流程等)
- 支持生成Mock代码(使用GoMock框架)
- 支持选择不同模式生成不同代码(使用"–m mode"指定)
- 生成单元测试代码时,同时支持传入目录或文件
- 支持指定方法追加生成测试用例(使用"–func funcName"指定)
### 编译安装
#### Method 1. With go get
```shell
go get -u github.com/bilibili/kratos/tool/testgen
$GOPATH/bin/testgen -h
```
#### Method 2. Build with Go
```shell
cd github.com/bilibili/kratos/tool/testgen
go build -o $GOPATH/bin/testgen
$GOPATH/bin/testgen -h
```
### 运行
#### 生成Dao/Service层单元UT
```shell
$GOPATH/bin/testgen YOUR_PROJECT/dao # default mode
$GOPATH/bin/testgen --m test path/to/your/pkg
$GOPATH/bin/testgen --func functionName path/to/your/pkg
```
#### 生成接口类型
```shell
$GOPATH/bin/testgen --m interface YOUR_PROJECT/dao #当前仅支持传目录,如目录包含子目录也会做处理
```
#### 生成Mock代码
```shell
$GOPATH/bin/testgen --m mock YOUR_PROJECT/dao #仅传入包路径即可
```
#### 生成Monkey代码
```shell
$GOPATH/bin/testgen --m monkey yourCodeDirPath #仅传入包路径即可
```
### 赋诗一首
```
莫生气 莫生气
代码辣鸡非我意
自己动手分田地
谈笑风生活长命
```

38
doc/wiki-cn/ut.md Normal file
View File

@ -0,0 +1,38 @@
# 背景
单元测试即对最小可测试单元进行检查和验证,它可以很好的让你的代码在上测试环境之前自己就能前置的发现问题,解决问题。当然每个语言都有原生支持的 UT 框架,不过在 Kratos 里面我们需要有一些配套设施以及周边工具来辅助我们构筑整个 UT 生态。
# 工具链
- testgen UT代码自动生成器(README: tool/testgen/README.md)
- testcli UT运行环境构建工具(README: tool/testcli/README.md)
# 测试框架选型
golang 的单元测试,既可以用官方自带的 testing 包,也有开源的如 testify、goconvey 业内知名,使用非常多也很好用的框架。
根据一番调研和内部使用经验,我们确定:
> - testing 作为基础库测试框架(非常精简不过够用)
> - goconvey 作为业务程序的单元测试框架(因为涉及比较多的业务场景和流程控制判断,比如更丰富的res值判断、上下文嵌套支持、还有webUI等)
# 单元测试标准
1. 覆盖率,当前标准:60%(所有包均需达到)
尽量达到70%以上。当然覆盖率并不能完全说明单元测试的质量,开发者需要考虑关键的条件判断和预期的结果。复杂的代码是需要好好设计测试用例的。
2. 通过率,当前标准:100%(所有用例中的断言必须通过)
# 书写建议
1. 结果验证
> - 校验err是否为nil. err是go函数的标配了,也是最基础的判断,如果err不为nil,基本上函数返回值或者处理肯定是有问题了。
> - 检验res值是否正确。res值的校验是非常重要的,也是很容易忽略的地方。比如返回结构体对象,要对结构体的成员进行判断,而有可能里面是0值。goconvey对res值的判断支持是非常友好的。
2. 逻辑验证
> 业务代码经常是流程比较复杂的,而函数的执行结果也是有上下文的,比如有不同条件分支。goconvey就非常优雅的支持了这种情况,可以嵌套执行。单元测试要结合业务代码逻辑,才能尽量的减少线上bug。
3. 如何mock
主要分以下3块:
> - 基础组件,如mc、redis、mysql等,由 testcli(testing/lich) 起基础镜像支持(需要提供建表、INSERT语句)与本地开发环境一致,也保证了结果的一致性。
> - rpc server,如 xxxx-service 需要定义 interface 供业务依赖方使用。所有rpc server 都必须要提供一个interface+mock代码(gomock)。
> - http server则直接写mock代码gock。
# 注意
单元测试极其重要,良好的单元测试习惯能很大程度上避免代码变更引起的bug!
单元测试极其重要,良好的单元测试习惯能很大程度上避免代码变更引起的bug!
单元测试极其重要,良好的单元测试习惯能很大程度上避免代码变更引起的bug!
以为很重要所以重复 3 遍~

16
go.mod
View File

@ -5,6 +5,8 @@ go 1.12
require (
github.com/BurntSushi/toml v0.3.1
github.com/aristanetworks/goarista v0.0.0-20190912214011-b54698eaaca6 // indirect
github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f // indirect
github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f // indirect
github.com/cznic/b v0.0.0-20181122101859-a26611c4d92d // indirect
github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548 // indirect
github.com/cznic/strutil v0.0.0-20181122101858-275e90344537 // indirect
@ -15,10 +17,11 @@ require (
github.com/go-playground/locales v0.12.1 // indirect
github.com/go-playground/universal-translator v0.16.0 // indirect
github.com/go-sql-driver/mysql v1.4.1
github.com/gogo/protobuf v1.2.1
github.com/gogo/protobuf v1.3.0
github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 // indirect
github.com/golang/mock v1.3.1 // indirect
github.com/golang/protobuf v1.3.2
github.com/google/uuid v1.1.1 // indirect
github.com/gorilla/websocket v1.4.0 // indirect
github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
github.com/leodido/go-urn v1.1.0 // indirect
@ -38,15 +41,16 @@ require (
github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5 // indirect
github.com/tsuna/gohbase v0.0.0-20190502052937-24ffed0537aa
github.com/urfave/cli v1.20.0
go.etcd.io/etcd v3.4.1+incompatible
go.etcd.io/etcd v0.0.0-20190917205325-a14579fbfb1a
go.uber.org/atomic v1.4.0 // indirect
go.uber.org/multierr v1.2.0 // indirect
golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 // indirect
golang.org/x/net v0.0.0-20190912160710-24e19bdeb0f2
golang.org/x/time v0.0.0-20190513212739-9d24e82272b4 // indirect
golang.org/x/net v0.0.0-20191011234655-491137f69257
golang.org/x/sys v0.0.0-20191010194322-b09406accb47 // indirect
golang.org/x/tools v0.0.0-20190912185636-87d9f09c5d89
google.golang.org/appengine v1.6.1 // indirect
google.golang.org/genproto v0.0.0-20190716160619-c506a9f90610
google.golang.org/grpc v1.23.1
google.golang.org/genproto v0.0.0-20191009194640-548a555dbc03
google.golang.org/grpc v1.24.0
gopkg.in/go-playground/assert.v1 v1.2.1 // indirect
gopkg.in/go-playground/validator.v9 v9.29.1
gopkg.in/yaml.v2 v2.2.2

29
go.sum
View File

@ -26,8 +26,12 @@ github.com/coreos/go-semver v0.2.0 h1:3Jm3tLmsgAYcjC+4Up7hJrFBPr+n7rAqYeSw/SZazu
github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
github.com/coreos/go-systemd v0.0.0-20180511133405-39ca1b05acc7 h1:u9SHYsPQNyt5tgDm3YN7+9dYrpK96E5wFilTFWIDZOM=
github.com/coreos/go-systemd v0.0.0-20180511133405-39ca1b05acc7/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f h1:JOrtw2xFKzlg+cbHpyrpLDmnN1HqhBfnX7WDiW7eG2c=
github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf h1:CAKfRE2YtTUIjjh1bkBtyYFaUT/WmOqsJjgtihT0vMI=
github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f h1:lBNOc5arjvs8E5mO2tbpBpLoyyu8B6e44T7hJy6potg=
github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
github.com/cznic/b v0.0.0-20181122101859-a26611c4d92d h1:SwD98825d6bdB+pEuTxWOXiSjBrHdOl/UVp75eI7JT8=
github.com/cznic/b v0.0.0-20181122101859-a26611c4d92d/go.mod h1:URriBxXwVq5ijiJ12C7iIZqlA69nTlI+LgI6/pwftG8=
@ -72,6 +76,8 @@ github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7a
github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/gogo/protobuf v1.2.1 h1:/s5zKNz0uPFCZ5hddgPdo2TK2TVrUNMn0OOX8/aZMTE=
github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
github.com/gogo/protobuf v1.3.0 h1:G8O7TerXerS4F6sx9OV7/nRfJdnXgHZu/S/7F2SN+UE=
github.com/gogo/protobuf v1.3.0/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@ -93,6 +99,8 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
github.com/google/uuid v1.0.0 h1:b4Gk+7WdP/d3HZH8EJsZpvV7EtDOgaZLtnaNGIu1adA=
github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/context v1.1.1/go.mod h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg=
github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
@ -116,6 +124,7 @@ github.com/json-iterator/go v1.1.7 h1:KfgG9LzI+pYjr4xvmz/5H4FXjokeP+rlHLhv3iH62F
github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/klauspost/cpuid v1.2.1/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
github.com/klauspost/reedsolomon v1.9.2/go.mod h1:CwCi+NUr9pqSVktrkN+Ondf06rkhYZ/pcNv7fu+8Un4=
@ -228,13 +237,15 @@ github.com/xtaci/kcp-go v5.4.5+incompatible/go.mod h1:bN6vIwHQbfHaHtFpEssmWsN45a
github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae/go.mod h1:gXtu8J62kEgmN++bm9BVICuT/e8yiLI2KFobd/TRFsE=
go.etcd.io/bbolt v1.3.3 h1:MUGmc65QhB3pIlaQ5bB4LwqSj6GIonVJXpZiaKNyaKk=
go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
go.etcd.io/etcd v3.4.1+incompatible h1:zROK0lqjLEsSf65FZrEWm6Jn7HMje74rWTSsKv9IZr0=
go.etcd.io/etcd v3.4.1+incompatible/go.mod h1:dnLIgRNXwCJa5e+c6mIZCrds/GIG4ncV9HhK5PX7jPg=
go.etcd.io/etcd v0.0.0-20190917205325-a14579fbfb1a h1:OpCyFK9+wUB3g4o1guENLYOUZhG4hswKiFbE+jC12Cc=
go.etcd.io/etcd v0.0.0-20190917205325-a14579fbfb1a/go.mod h1:dnLIgRNXwCJa5e+c6mIZCrds/GIG4ncV9HhK5PX7jPg=
go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
go.uber.org/multierr v1.2.0 h1:6I+W7f5VwC5SV9dNrZ3qXrDB9mD0dyGOi/ZJmYw03T4=
go.uber.org/multierr v1.2.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
go.uber.org/zap v1.10.0 h1:ORx85nbTijNz8ljznvCMR1ZBIPKFn3jQrag10X2AsuM=
go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@ -261,6 +272,8 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190912160710-24e19bdeb0f2 h1:4dVFTC832rPn4pomLSz1vA+are2+dU19w1H8OngV7nc=
golang.org/x/net v0.0.0-20190912160710-24e19bdeb0f2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20191011234655-491137f69257 h1:ry8e2D+cwaV6hk7lb3aRTjjZo24shrbK0e11QEOkTIg=
golang.org/x/net v0.0.0-20191011234655-491137f69257/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@ -281,15 +294,17 @@ golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190912141932-bc967efca4b8 h1:41hwlulw1prEMBxLQSlMSux1zxJf07B3WPsdjJlKZxE=
golang.org/x/sys v0.0.0-20190912141932-bc967efca4b8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191010194322-b09406accb47 h1:/XfQ9z7ib8eEJX2hdgFTZJ/ntt0swNk5oYBziWeTCvY=
golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/time v0.0.0-20190513212739-9d24e82272b4 h1:RMGusaKverhgGR5KBERIKiTyWoWHRd84GCtsNlvLvIo=
golang.org/x/time v0.0.0-20190513212739-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
@ -305,12 +320,14 @@ google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7
google.golang.org/appengine v1.6.1 h1:QzqyMA1tlu6CgqCDUtU9V+ZKhLFT2dkJuANu5QaxI3I=
google.golang.org/appengine v1.6.1/go.mod h1:i06prIuMbXzDqacNJfV5OdTW448YApPu5ww/cMBSeb0=
google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
google.golang.org/genproto v0.0.0-20190716160619-c506a9f90610 h1:Ygq9/SRJX9+dU0WCIICM8RkWvDw03lvB77hrhJnpxfU=
google.golang.org/genproto v0.0.0-20190716160619-c506a9f90610/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
google.golang.org/genproto v0.0.0-20191009194640-548a555dbc03 h1:4HYDjxeNXAOTv3o1N2tjo8UUSlhQgAD52FVkwxnWgM8=
google.golang.org/genproto v0.0.0-20191009194640-548a555dbc03/go.mod h1:n3cpQtvxv34hfy77yVDNjmbRyujviMdxYliBSkLhpCc=
google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
google.golang.org/grpc v1.20.0/go.mod h1:chYK+tFQF0nDUGJgXMSgLCQk3phJEuONr2DCgLDdAQM=
google.golang.org/grpc v1.23.1 h1:q4XQuHFC6I28BKZpo6IYyb3mNO+l7lSOxRuYTCiDfXk=
google.golang.org/grpc v1.23.1/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
google.golang.org/grpc v1.24.0 h1:vb/1TCsVn3DcJlQ0Gs1yB1pKI6Do2/QNwxdKqmc/b0s=
google.golang.org/grpc v1.24.0/go.mod h1:XDChyiUovWa60DnaeDeZmSW86xtLtjtZbwvSiRnRtcA=
gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
gopkg.in/bsm/ratelimit.v1 v1.0.0-20160220154919-db14e161995a/go.mod h1:KF9sEfUPAXdG8Oev9e99iLGnl2uJMjc5B+4y3O7x610=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

27
pkg/cache/redis/commandinfo_test.go vendored Normal file
View File

@ -0,0 +1,27 @@
package redis
import "testing"
func TestLookupCommandInfo(t *testing.T) {
for _, n := range []string{"watch", "WATCH", "wAtch"} {
if LookupCommandInfo(n) == (CommandInfo{}) {
t.Errorf("LookupCommandInfo(%q) = CommandInfo{}, expected non-zero value", n)
}
}
}
func benchmarkLookupCommandInfo(b *testing.B, names ...string) {
for i := 0; i < b.N; i++ {
for _, c := range names {
LookupCommandInfo(c)
}
}
}
func BenchmarkLookupCommandInfoCorrectCase(b *testing.B) {
benchmarkLookupCommandInfo(b, "watch", "WATCH", "monitor", "MONITOR")
}
func BenchmarkLookupCommandInfoMixedCase(b *testing.B) {
benchmarkLookupCommandInfo(b, "wAtch", "WeTCH", "monItor", "MONiTOR")
}

View File

@ -30,6 +30,33 @@ import (
"github.com/pkg/errors"
)
// Conn represents a connection to a Redis server.
type Conn interface {
// Close closes the connection.
Close() error
// Err returns a non-nil value if the connection is broken. The returned
// value is either the first non-nil value returned from the underlying
// network connection or a protocol parsing error. Applications should
// close broken connections.
Err() error
// Do sends a command to the server and returns the received reply.
Do(commandName string, args ...interface{}) (reply interface{}, err error)
// Send writes the command to the client's output buffer.
Send(commandName string, args ...interface{}) error
// Flush flushes the output buffer to the Redis server.
Flush() error
// Receive receives a single reply from the Redis server
Receive() (reply interface{}, err error)
// WithContext returns Conn with the input ctx.
WithContext(ctx context.Context) Conn
}
// conn is the low-level implementation of Conn
type conn struct {
// Shared
@ -38,6 +65,8 @@ type conn struct {
err error
conn net.Conn
ctx context.Context
// Read
readTimeout time.Duration
br *bufio.Reader
@ -226,6 +255,7 @@ func NewConn(c *Config) (cn Conn, err error) {
func (c *conn) Close() error {
c.mu.Lock()
c.ctx = nil
err := c.err
if c.err == nil {
c.err = errors.New("redigo: closed")
@ -295,7 +325,7 @@ func (c *conn) writeFloat64(n float64) error {
func (c *conn) writeCommand(cmd string, args []interface{}) (err error) {
if c.writeTimeout != 0 {
c.conn.SetWriteDeadline(time.Now().Add(c.writeTimeout))
c.conn.SetWriteDeadline(shrinkDeadline(c.ctx, c.writeTimeout))
}
c.writeLen('*', 1+len(args))
err = c.writeString(cmd)
@ -478,7 +508,7 @@ func (c *conn) Send(cmd string, args ...interface{}) (err error) {
func (c *conn) Flush() (err error) {
if c.writeTimeout != 0 {
c.conn.SetWriteDeadline(time.Now().Add(c.writeTimeout))
c.conn.SetWriteDeadline(shrinkDeadline(c.ctx, c.writeTimeout))
}
if err = c.bw.Flush(); err != nil {
c.fatal(err)
@ -488,7 +518,7 @@ func (c *conn) Flush() (err error) {
func (c *conn) Receive() (reply interface{}, err error) {
if c.readTimeout != 0 {
c.conn.SetReadDeadline(time.Now().Add(c.readTimeout))
c.conn.SetReadDeadline(shrinkDeadline(c.ctx, c.readTimeout))
}
if reply, err = c.readReply(); err != nil {
return nil, c.fatal(err)
@ -511,7 +541,7 @@ func (c *conn) Receive() (reply interface{}, err error) {
return
}
func (c *conn) Do(cmd string, args ...interface{}) (interface{}, error) {
func (c *conn) Do(cmd string, args ...interface{}) (reply interface{}, err error) {
c.mu.Lock()
pending := c.pending
c.pending = 0
@ -519,7 +549,7 @@ func (c *conn) Do(cmd string, args ...interface{}) (interface{}, error) {
if cmd == "" && pending == 0 {
return nil, nil
}
var err error
if cmd != "" {
err = c.writeCommand(cmd, args)
}
@ -530,7 +560,7 @@ func (c *conn) Do(cmd string, args ...interface{}) (interface{}, error) {
return nil, c.fatal(err)
}
if c.readTimeout != 0 {
c.conn.SetReadDeadline(time.Now().Add(c.readTimeout))
c.conn.SetReadDeadline(shrinkDeadline(c.ctx, c.readTimeout))
}
if cmd == "" {
reply := make([]interface{}, pending)
@ -548,7 +578,6 @@ func (c *conn) Do(cmd string, args ...interface{}) (interface{}, error) {
return reply, nil
}
var reply interface{}
for i := 0; i <= pending; i++ {
var e error
if reply, e = c.readReply(); e != nil {
@ -561,5 +590,20 @@ func (c *conn) Do(cmd string, args ...interface{}) (interface{}, error) {
return reply, err
}
// WithContext FIXME: implement WithContext
func (c *conn) WithContext(ctx context.Context) Conn { return c }
func (c *conn) copy() *conn {
return &conn{
pending: c.pending,
err: c.err,
conn: c.conn,
bw: c.bw,
br: c.br,
readTimeout: c.readTimeout,
writeTimeout: c.writeTimeout,
}
}
func (c *conn) WithContext(ctx context.Context) Conn {
c2 := c.copy()
c2.ctx = ctx
return c2
}

670
pkg/cache/redis/conn_test.go vendored Normal file
View File

@ -0,0 +1,670 @@
// Copyright 2012 Gary Burd
//
// Licensed under the Apache License, Version 2.0 (the "License"): you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations
// under the License.
package redis
import (
"bytes"
"context"
"io"
"math"
"net"
"os"
"reflect"
"strings"
"testing"
"time"
)
type tConn struct {
io.Reader
io.Writer
}
func (*tConn) Close() error { return nil }
func (*tConn) LocalAddr() net.Addr { return nil }
func (*tConn) RemoteAddr() net.Addr { return nil }
func (*tConn) SetDeadline(t time.Time) error { return nil }
func (*tConn) SetReadDeadline(t time.Time) error { return nil }
func (*tConn) SetWriteDeadline(t time.Time) error { return nil }
func dialTestConn(r io.Reader, w io.Writer) DialOption {
return DialNetDial(func(net, addr string) (net.Conn, error) {
return &tConn{Reader: r, Writer: w}, nil
})
}
var writeTests = []struct {
args []interface{}
expected string
}{
{
[]interface{}{"SET", "key", "value"},
"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n",
},
{
[]interface{}{"SET", "key", "value"},
"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n",
},
{
[]interface{}{"SET", "key", byte(100)},
"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$3\r\n100\r\n",
},
{
[]interface{}{"SET", "key", 100},
"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$3\r\n100\r\n",
},
{
[]interface{}{"SET", "key", int64(math.MinInt64)},
"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$20\r\n-9223372036854775808\r\n",
},
{
[]interface{}{"SET", "key", float64(1349673917.939762)},
"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$21\r\n1.349673917939762e+09\r\n",
},
{
[]interface{}{"SET", "key", ""},
"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$0\r\n\r\n",
},
{
[]interface{}{"SET", "key", nil},
"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$0\r\n\r\n",
},
{
[]interface{}{"ECHO", true, false},
"*3\r\n$4\r\nECHO\r\n$1\r\n1\r\n$1\r\n0\r\n",
},
}
func TestWrite(t *testing.T) {
for _, tt := range writeTests {
var buf bytes.Buffer
c, _ := Dial("", "", dialTestConn(nil, &buf))
err := c.Send(tt.args[0].(string), tt.args[1:]...)
if err != nil {
t.Errorf("Send(%v) returned error %v", tt.args, err)
continue
}
c.Flush()
actual := buf.String()
if actual != tt.expected {
t.Errorf("Send(%v) = %q, want %q", tt.args, actual, tt.expected)
}
}
}
var errorSentinel = &struct{}{}
var readTests = []struct {
reply string
expected interface{}
}{
{
"+OK\r\n",
"OK",
},
{
"+PONG\r\n",
"PONG",
},
{
"@OK\r\n",
errorSentinel,
},
{
"$6\r\nfoobar\r\n",
[]byte("foobar"),
},
{
"$-1\r\n",
nil,
},
{
":1\r\n",
int64(1),
},
{
":-2\r\n",
int64(-2),
},
{
"*0\r\n",
[]interface{}{},
},
{
"*-1\r\n",
nil,
},
{
"*4\r\n$3\r\nfoo\r\n$3\r\nbar\r\n$5\r\nHello\r\n$5\r\nWorld\r\n",
[]interface{}{[]byte("foo"), []byte("bar"), []byte("Hello"), []byte("World")},
},
{
"*3\r\n$3\r\nfoo\r\n$-1\r\n$3\r\nbar\r\n",
[]interface{}{[]byte("foo"), nil, []byte("bar")},
},
{
// "x" is not a valid length
"$x\r\nfoobar\r\n",
errorSentinel,
},
{
// -2 is not a valid length
"$-2\r\n",
errorSentinel,
},
{
// "x" is not a valid integer
":x\r\n",
errorSentinel,
},
{
// missing \r\n following value
"$6\r\nfoobar",
errorSentinel,
},
{
// short value
"$6\r\nxx",
errorSentinel,
},
{
// long value
"$6\r\nfoobarx\r\n",
errorSentinel,
},
}
func TestRead(t *testing.T) {
for _, tt := range readTests {
c, _ := Dial("", "", dialTestConn(strings.NewReader(tt.reply), nil))
actual, err := c.Receive()
if tt.expected == errorSentinel {
if err == nil {
t.Errorf("Receive(%q) did not return expected error", tt.reply)
}
} else {
if err != nil {
t.Errorf("Receive(%q) returned error %v", tt.reply, err)
continue
}
if !reflect.DeepEqual(actual, tt.expected) {
t.Errorf("Receive(%q) = %v, want %v", tt.reply, actual, tt.expected)
}
}
}
}
var testCommands = []struct {
args []interface{}
expected interface{}
}{
{
[]interface{}{"PING"},
"PONG",
},
{
[]interface{}{"SET", "foo", "bar"},
"OK",
},
{
[]interface{}{"GET", "foo"},
[]byte("bar"),
},
{
[]interface{}{"GET", "nokey"},
nil,
},
{
[]interface{}{"MGET", "nokey", "foo"},
[]interface{}{nil, []byte("bar")},
},
{
[]interface{}{"INCR", "mycounter"},
int64(1),
},
{
[]interface{}{"LPUSH", "mylist", "foo"},
int64(1),
},
{
[]interface{}{"LPUSH", "mylist", "bar"},
int64(2),
},
{
[]interface{}{"LRANGE", "mylist", 0, -1},
[]interface{}{[]byte("bar"), []byte("foo")},
},
{
[]interface{}{"MULTI"},
"OK",
},
{
[]interface{}{"LRANGE", "mylist", 0, -1},
"QUEUED",
},
{
[]interface{}{"PING"},
"QUEUED",
},
{
[]interface{}{"EXEC"},
[]interface{}{
[]interface{}{[]byte("bar"), []byte("foo")},
"PONG",
},
},
}
func TestDoCommands(t *testing.T) {
c, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer c.Close()
for _, cmd := range testCommands {
actual, err := c.Do(cmd.args[0].(string), cmd.args[1:]...)
if err != nil {
t.Errorf("Do(%v) returned error %v", cmd.args, err)
continue
}
if !reflect.DeepEqual(actual, cmd.expected) {
t.Errorf("Do(%v) = %v, want %v", cmd.args, actual, cmd.expected)
}
}
}
func TestPipelineCommands(t *testing.T) {
c, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer c.Close()
for _, cmd := range testCommands {
if err := c.Send(cmd.args[0].(string), cmd.args[1:]...); err != nil {
t.Fatalf("Send(%v) returned error %v", cmd.args, err)
}
}
if err := c.Flush(); err != nil {
t.Errorf("Flush() returned error %v", err)
}
for _, cmd := range testCommands {
actual, err := c.Receive()
if err != nil {
t.Fatalf("Receive(%v) returned error %v", cmd.args, err)
}
if !reflect.DeepEqual(actual, cmd.expected) {
t.Errorf("Receive(%v) = %v, want %v", cmd.args, actual, cmd.expected)
}
}
}
func TestBlankCommmand(t *testing.T) {
c, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer c.Close()
for _, cmd := range testCommands {
if err = c.Send(cmd.args[0].(string), cmd.args[1:]...); err != nil {
t.Fatalf("Send(%v) returned error %v", cmd.args, err)
}
}
reply, err := Values(c.Do(""))
if err != nil {
t.Fatalf("Do() returned error %v", err)
}
if len(reply) != len(testCommands) {
t.Fatalf("len(reply)=%d, want %d", len(reply), len(testCommands))
}
for i, cmd := range testCommands {
actual := reply[i]
if !reflect.DeepEqual(actual, cmd.expected) {
t.Errorf("Receive(%v) = %v, want %v", cmd.args, actual, cmd.expected)
}
}
}
func TestRecvBeforeSend(t *testing.T) {
c, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer c.Close()
done := make(chan struct{})
go func() {
c.Receive()
close(done)
}()
time.Sleep(time.Millisecond)
c.Send("PING")
c.Flush()
<-done
_, err = c.Do("")
if err != nil {
t.Fatalf("error=%v", err)
}
}
func TestError(t *testing.T) {
c, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer c.Close()
c.Do("SET", "key", "val")
_, err = c.Do("HSET", "key", "fld", "val")
if err == nil {
t.Errorf("Expected err for HSET on string key.")
}
if c.Err() != nil {
t.Errorf("Conn has Err()=%v, expect nil", c.Err())
}
_, err = c.Do("SET", "key", "val")
if err != nil {
t.Errorf("Do(SET, key, val) returned error %v, expected nil.", err)
}
}
func TestReadTimeout(t *testing.T) {
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("net.Listen returned %v", err)
}
defer l.Close()
go func() {
for {
c, err1 := l.Accept()
if err1 != nil {
return
}
go func() {
time.Sleep(time.Second)
c.Write([]byte("+OK\r\n"))
c.Close()
}()
}
}()
// Do
c1, err := Dial(l.Addr().Network(), l.Addr().String(), DialReadTimeout(time.Millisecond))
if err != nil {
t.Fatalf("Dial returned %v", err)
}
defer c1.Close()
_, err = c1.Do("PING")
if err == nil {
t.Fatalf("c1.Do() returned nil, expect error")
}
if c1.Err() == nil {
t.Fatalf("c1.Err() = nil, expect error")
}
// Send/Flush/Receive
c2, err := Dial(l.Addr().Network(), l.Addr().String(), DialReadTimeout(time.Millisecond))
if err != nil {
t.Fatalf("Dial returned %v", err)
}
defer c2.Close()
c2.Send("PING")
c2.Flush()
_, err = c2.Receive()
if err == nil {
t.Fatalf("c2.Receive() returned nil, expect error")
}
if c2.Err() == nil {
t.Fatalf("c2.Err() = nil, expect error")
}
}
var dialErrors = []struct {
rawurl string
expectedError string
}{
{
"localhost",
"invalid redis URL scheme",
},
// The error message for invalid hosts is diffferent in different
// versions of Go, so just check that there is an error message.
{
"redis://weird url",
"",
},
{
"redis://foo:bar:baz",
"",
},
{
"http://www.google.com",
"invalid redis URL scheme: http",
},
{
"redis://localhost:6379/abc123",
"invalid database: abc123",
},
}
func TestDialURLErrors(t *testing.T) {
for _, d := range dialErrors {
_, err := DialURL(d.rawurl)
if err == nil || !strings.Contains(err.Error(), d.expectedError) {
t.Errorf("DialURL did not return expected error (expected %v to contain %s)", err, d.expectedError)
}
}
}
func TestDialURLPort(t *testing.T) {
checkPort := func(network, address string) (net.Conn, error) {
if address != "localhost:6379" {
t.Errorf("DialURL did not set port to 6379 by default (got %v)", address)
}
return nil, nil
}
_, err := DialURL("redis://localhost", DialNetDial(checkPort))
if err != nil {
t.Error("dial error:", err)
}
}
func TestDialURLHost(t *testing.T) {
checkHost := func(network, address string) (net.Conn, error) {
if address != "localhost:6379" {
t.Errorf("DialURL did not set host to localhost by default (got %v)", address)
}
return nil, nil
}
_, err := DialURL("redis://:6379", DialNetDial(checkHost))
if err != nil {
t.Error("dial error:", err)
}
}
func TestDialURLPassword(t *testing.T) {
var buf bytes.Buffer
_, err := DialURL("redis://x:abc123@localhost", dialTestConn(strings.NewReader("+OK\r\n"), &buf))
if err != nil {
t.Error("dial error:", err)
}
expected := "*2\r\n$4\r\nAUTH\r\n$6\r\nabc123\r\n"
actual := buf.String()
if actual != expected {
t.Errorf("commands = %q, want %q", actual, expected)
}
}
func TestDialURLDatabase(t *testing.T) {
var buf bytes.Buffer
_, err := DialURL("redis://localhost/3", dialTestConn(strings.NewReader("+OK\r\n"), &buf))
if err != nil {
t.Error("dial error:", err)
}
expected := "*2\r\n$6\r\nSELECT\r\n$1\r\n3\r\n"
actual := buf.String()
if actual != expected {
t.Errorf("commands = %q, want %q", actual, expected)
}
}
// Connect to local instance of Redis running on the default port.
func ExampleDial() {
c, err := Dial("tcp", ":6379")
if err != nil {
// handle error
}
defer c.Close()
}
// Connect to remote instance of Redis using a URL.
func ExampleDialURL() {
c, err := DialURL(os.Getenv("REDIS_URL"))
if err != nil {
// handle connection error
}
defer c.Close()
}
// TextExecError tests handling of errors in a transaction. See
// http://io/topics/transactions for information on how Redis handles
// errors in a transaction.
func TestExecError(t *testing.T) {
c, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer c.Close()
// Execute commands that fail before EXEC is called.
c.Do("DEL", "k0")
c.Do("ZADD", "k0", 0, 0)
c.Send("MULTI")
c.Send("NOTACOMMAND", "k0", 0, 0)
c.Send("ZINCRBY", "k0", 0, 0)
v, err := c.Do("EXEC")
if err == nil {
t.Fatalf("EXEC returned values %v, expected error", v)
}
// Execute commands that fail after EXEC is called. The first command
// returns an error.
c.Do("DEL", "k1")
c.Do("ZADD", "k1", 0, 0)
c.Send("MULTI")
c.Send("HSET", "k1", 0, 0)
c.Send("ZINCRBY", "k1", 0, 0)
v, err = c.Do("EXEC")
if err != nil {
t.Fatalf("EXEC returned error %v", err)
}
vs, err := Values(v, nil)
if err != nil {
t.Fatalf("Values(v) returned error %v", err)
}
if len(vs) != 2 {
t.Fatalf("len(vs) == %d, want 2", len(vs))
}
if _, ok := vs[0].(error); !ok {
t.Fatalf("first result is type %T, expected error", vs[0])
}
if _, ok := vs[1].([]byte); !ok {
t.Fatalf("second result is type %T, expected []byte", vs[1])
}
// Execute commands that fail after EXEC is called. The second command
// returns an error.
c.Do("ZADD", "k2", 0, 0)
c.Send("MULTI")
c.Send("ZINCRBY", "k2", 0, 0)
c.Send("HSET", "k2", 0, 0)
v, err = c.Do("EXEC")
if err != nil {
t.Fatalf("EXEC returned error %v", err)
}
vs, err = Values(v, nil)
if err != nil {
t.Fatalf("Values(v) returned error %v", err)
}
if len(vs) != 2 {
t.Fatalf("len(vs) == %d, want 2", len(vs))
}
if _, ok := vs[0].([]byte); !ok {
t.Fatalf("first result is type %T, expected []byte", vs[0])
}
if _, ok := vs[1].(error); !ok {
t.Fatalf("second result is type %T, expected error", vs[2])
}
}
func BenchmarkDoEmpty(b *testing.B) {
c, err := DialDefaultServer()
if err != nil {
b.Fatal(err)
}
defer c.Close()
b.ResetTimer()
for i := 0; i < b.N; i++ {
if _, err := c.Do(""); err != nil {
b.Fatal(err)
}
}
}
func BenchmarkDoPing(b *testing.B) {
c, err := DialDefaultServer()
if err != nil {
b.Fatal(err)
}
defer c.Close()
b.ResetTimer()
for i := 0; i < b.N; i++ {
if _, err := c.Do("PING"); err != nil {
b.Fatal(err)
}
}
}
func BenchmarkConn(b *testing.B) {
for i := 0; i < b.N; i++ {
c, err := DialDefaultServer()
if err != nil {
b.Fatal(err)
}
c2 := c.WithContext(context.TODO())
if _, err := c2.Do("PING"); err != nil {
b.Fatal(err)
}
c2.Close()
}
}

View File

@ -16,16 +16,18 @@ package redis
import (
"bytes"
"context"
"fmt"
"log"
)
// NewLoggingConn returns a logging wrapper around a connection.
// ATTENTION: ONLY use loggingConn in developing, DO NOT use this in production.
func NewLoggingConn(conn Conn, logger *log.Logger, prefix string) Conn {
if prefix != "" {
prefix = prefix + "."
}
return &loggingConn{conn, logger, prefix}
return &loggingConn{Conn: conn, logger: logger, prefix: prefix}
}
type loggingConn struct {
@ -98,16 +100,16 @@ func (c *loggingConn) print(method, commandName string, args []interface{}, repl
c.logger.Output(3, buf.String())
}
func (c *loggingConn) Do(commandName string, args ...interface{}) (interface{}, error) {
reply, err := c.Conn.Do(commandName, args...)
func (c *loggingConn) Do(commandName string, args ...interface{}) (reply interface{}, err error) {
reply, err = c.Conn.Do(commandName, args...)
c.print("Do", commandName, args, reply, err)
return reply, err
}
func (c *loggingConn) Send(commandName string, args ...interface{}) error {
err := c.Conn.Send(commandName, args...)
func (c *loggingConn) Send(commandName string, args ...interface{}) (err error) {
err = c.Conn.Send(commandName, args...)
c.print("Send", commandName, args, nil, err)
return err
return
}
func (c *loggingConn) Receive() (interface{}, error) {
@ -115,3 +117,7 @@ func (c *loggingConn) Receive() (interface{}, error) {
c.print("Receive", "", nil, reply, err)
return reply, err
}
func (c *loggingConn) WithContext(ctx context.Context) Conn {
return c
}

67
pkg/cache/redis/main_test.go vendored Normal file
View File

@ -0,0 +1,67 @@
package redis
import (
"flag"
"os"
"testing"
"time"
"github.com/bilibili/kratos/pkg/container/pool"
"github.com/bilibili/kratos/pkg/testing/lich"
xtime "github.com/bilibili/kratos/pkg/time"
)
var (
testRedisAddr string
testPool *Pool
testConfig *Config
)
func setupTestConfig(addr string) {
c := getTestConfig(addr)
c.Config = &pool.Config{
Active: 20,
Idle: 2,
IdleTimeout: xtime.Duration(90 * time.Second),
}
testConfig = c
}
func getTestConfig(addr string) *Config {
return &Config{
Name: "test",
Proto: "tcp",
Addr: addr,
DialTimeout: xtime.Duration(time.Second),
ReadTimeout: xtime.Duration(time.Second),
WriteTimeout: xtime.Duration(time.Second),
}
}
func setupTestPool() {
testPool = NewPool(testConfig)
}
// DialDefaultServer starts the test server if not already started and dials a
// connection to the server.
func DialDefaultServer() (Conn, error) {
c, err := Dial("tcp", testRedisAddr, DialReadTimeout(1*time.Second), DialWriteTimeout(1*time.Second))
if err != nil {
return nil, err
}
c.Do("FLUSHDB")
return c, nil
}
func TestMain(m *testing.M) {
flag.Set("f", "./test/docker-compose.yaml")
if err := lich.Setup(); err != nil {
panic(err)
}
defer lich.Teardown()
testRedisAddr = "localhost:6379"
setupTestConfig(testRedisAddr)
setupTestPool()
ret := m.Run()
os.Exit(ret)
}

View File

@ -1,6 +1,8 @@
package redis
import "github.com/bilibili/kratos/pkg/stat/metric"
import (
"github.com/bilibili/kratos/pkg/stat/metric"
)
const namespace = "redis_client"

View File

@ -1,8 +1,6 @@
package redis
import (
"context"
)
import "context"
// MockErr for unit test.
type MockErr struct {

85
pkg/cache/redis/pipeline.go vendored Normal file
View File

@ -0,0 +1,85 @@
package redis
import (
"context"
"errors"
)
type Pipeliner interface {
// Send writes the command to the client's output buffer.
Send(commandName string, args ...interface{})
// Exec executes all commands and get replies.
Exec(ctx context.Context) (rs *Replies, err error)
}
var (
ErrNoReply = errors.New("redis: no reply in result set")
)
type pipeliner struct {
pool *Pool
cmds []*cmd
}
type Replies struct {
replies []*reply
}
type reply struct {
reply interface{}
err error
}
func (rs *Replies) Next() bool {
return len(rs.replies) > 0
}
func (rs *Replies) Scan() (reply interface{}, err error) {
if !rs.Next() {
return nil, ErrNoReply
}
reply, err = rs.replies[0].reply, rs.replies[0].err
rs.replies = rs.replies[1:]
return
}
type cmd struct {
commandName string
args []interface{}
}
func (p *pipeliner) Send(commandName string, args ...interface{}) {
p.cmds = append(p.cmds, &cmd{commandName: commandName, args: args})
return
}
func (p *pipeliner) Exec(ctx context.Context) (rs *Replies, err error) {
n := len(p.cmds)
if n == 0 {
return &Replies{}, nil
}
c := p.pool.Get(ctx)
defer c.Close()
for len(p.cmds) > 0 {
cmd := p.cmds[0]
p.cmds = p.cmds[1:]
if err := c.Send(cmd.commandName, cmd.args...); err != nil {
p.cmds = p.cmds[:0]
return nil, err
}
}
if err = c.Flush(); err != nil {
p.cmds = p.cmds[:0]
return nil, err
}
rps := make([]*reply, 0, n)
for i := 0; i < n; i++ {
rp, err := c.Receive()
rps = append(rps, &reply{reply: rp, err: err})
}
rs = &Replies{
replies: rps,
}
return
}

96
pkg/cache/redis/pipeline_test.go vendored Normal file
View File

@ -0,0 +1,96 @@
package redis
import (
"context"
"fmt"
"reflect"
"testing"
"time"
"github.com/bilibili/kratos/pkg/container/pool"
xtime "github.com/bilibili/kratos/pkg/time"
)
func TestRedis_Pipeline(t *testing.T) {
conf := &Config{
Name: "test",
Proto: "tcp",
Addr: testRedisAddr,
DialTimeout: xtime.Duration(1 * time.Second),
ReadTimeout: xtime.Duration(1 * time.Second),
WriteTimeout: xtime.Duration(1 * time.Second),
}
conf.Config = &pool.Config{
Active: 10,
Idle: 2,
IdleTimeout: xtime.Duration(90 * time.Second),
}
r := NewRedis(conf)
r.Do(context.TODO(), "FLUSHDB")
p := r.Pipeline()
for _, cmd := range testCommands {
p.Send(cmd.args[0].(string), cmd.args[1:]...)
}
replies, err := p.Exec(context.TODO())
i := 0
for replies.Next() {
cmd := testCommands[i]
actual, err := replies.Scan()
if err != nil {
t.Fatalf("Receive(%v) returned error %v", cmd.args, err)
}
if !reflect.DeepEqual(actual, cmd.expected) {
t.Errorf("Receive(%v) = %v, want %v", cmd.args, actual, cmd.expected)
}
i++
}
err = r.Close()
if err != nil {
t.Errorf("Close() error %v", err)
}
}
func ExamplePipeliner() {
r := NewRedis(testConfig)
defer r.Close()
pip := r.Pipeline()
pip.Send("SET", "hello", "world")
pip.Send("GET", "hello")
replies, err := pip.Exec(context.TODO())
if err != nil {
fmt.Printf("%#v\n", err)
}
for replies.Next() {
s, err := String(replies.Scan())
if err != nil {
fmt.Printf("err %#v\n", err)
}
fmt.Printf("%#v\n", s)
}
// Output:
// "OK"
// "world"
}
func BenchmarkRedisPipelineExec(b *testing.B) {
r := NewRedis(testConfig)
defer r.Close()
r.Do(context.TODO(), "SET", "abcde", "fghiasdfasdf")
b.ResetTimer()
for i := 0; i < b.N; i++ {
p := r.Pipeline()
p.Send("GET", "abcde")
_, err := p.Exec(context.TODO())
if err != nil {
b.Fatal(err)
}
}
}

View File

@ -45,24 +45,14 @@ type Pool struct {
statfunc func(name, addr, cmd string, t time.Time, err error) func()
}
// Config client settings.
type Config struct {
*pool.Config
Name string // redis name, for trace
Proto string
Addr string
Auth string
DialTimeout xtime.Duration
ReadTimeout xtime.Duration
WriteTimeout xtime.Duration
}
// NewPool creates a new pool.
func NewPool(c *Config, options ...DialOption) (p *Pool) {
if c.DialTimeout <= 0 || c.ReadTimeout <= 0 || c.WriteTimeout <= 0 {
panic("must config redis timeout")
}
if c.SlowLog <= 0 {
c.SlowLog = xtime.Duration(250 * time.Millisecond)
}
ops := []DialOption{
DialConnectTimeout(time.Duration(c.DialTimeout)),
DialReadTimeout(time.Duration(c.ReadTimeout)),
@ -71,12 +61,18 @@ func NewPool(c *Config, options ...DialOption) (p *Pool) {
}
ops = append(ops, options...)
p1 := pool.NewSlice(c.Config)
// new pool
p1.New = func(ctx context.Context) (io.Closer, error) {
conn, err := Dial(c.Proto, c.Addr, ops...)
if err != nil {
return nil, err
}
return &traceConn{Conn: conn, connTags: []trace.Tag{trace.TagString(trace.TagPeerAddress, c.Addr)}}, nil
return &traceConn{
Conn: conn,
connTags: []trace.Tag{trace.TagString(trace.TagPeerAddress, c.Addr)},
slowLogThreshold: time.Duration(c.SlowLog),
}, nil
}
p = &Pool{Slice: p1, c: c, statfunc: pstat}
return
@ -93,7 +89,7 @@ func (p *Pool) Get(ctx context.Context) Conn {
return errorConnection{err}
}
c1, _ := c.(Conn)
return &pooledConnection{p: p, c: c1.WithContext(ctx), ctx: ctx, now: beginTime}
return &pooledConnection{p: p, c: c1.WithContext(ctx), rc: c1, now: beginTime}
}
// Close releases the resources used by the pool.
@ -103,12 +99,12 @@ func (p *Pool) Close() error {
type pooledConnection struct {
p *Pool
rc Conn
c Conn
state int
now time.Time
cmds []string
ctx context.Context
}
var (
@ -180,7 +176,7 @@ func (pc *pooledConnection) Close() error {
}
}
_, err := c.Do("")
pc.p.Slice.Put(context.Background(), c, pc.state != 0 || c.Err() != nil)
pc.p.Slice.Put(context.Background(), pc.rc, pc.state != 0 || c.Err() != nil)
return err
}
@ -227,7 +223,6 @@ func (pc *pooledConnection) Receive() (reply interface{}, err error) {
}
func (pc *pooledConnection) WithContext(ctx context.Context) Conn {
pc.ctx = ctx
return pc
}

540
pkg/cache/redis/pool_test.go vendored Normal file
View File

@ -0,0 +1,540 @@
// Copyright 2011 Gary Burd
//
// Licensed under the Apache License, Version 2.0 (the "License"): you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations
// under the License.
package redis
import (
"context"
"errors"
"io"
"reflect"
"sync"
"testing"
"time"
"github.com/bilibili/kratos/pkg/container/pool"
)
type poolTestConn struct {
d *poolDialer
err error
c Conn
ctx context.Context
}
func (c *poolTestConn) Flush() error {
return c.c.Flush()
}
func (c *poolTestConn) Receive() (reply interface{}, err error) {
return c.c.Receive()
}
func (c *poolTestConn) WithContext(ctx context.Context) Conn {
c.c.WithContext(ctx)
c.ctx = ctx
return c
}
func (c *poolTestConn) Close() error {
c.d.mu.Lock()
c.d.open--
c.d.mu.Unlock()
return c.c.Close()
}
func (c *poolTestConn) Err() error { return c.err }
func (c *poolTestConn) Do(commandName string, args ...interface{}) (reply interface{}, err error) {
if commandName == "ERR" {
c.err = args[0].(error)
commandName = "PING"
}
if commandName != "" {
c.d.commands = append(c.d.commands, commandName)
}
return c.c.Do(commandName, args...)
}
func (c *poolTestConn) Send(commandName string, args ...interface{}) error {
c.d.commands = append(c.d.commands, commandName)
return c.c.Send(commandName, args...)
}
type poolDialer struct {
mu sync.Mutex
t *testing.T
dialed int
open int
commands []string
dialErr error
}
func (d *poolDialer) dial() (Conn, error) {
d.mu.Lock()
d.dialed += 1
dialErr := d.dialErr
d.mu.Unlock()
if dialErr != nil {
return nil, d.dialErr
}
c, err := DialDefaultServer()
if err != nil {
return nil, err
}
d.mu.Lock()
d.open += 1
d.mu.Unlock()
return &poolTestConn{d: d, c: c}, nil
}
func (d *poolDialer) check(message string, p *Pool, dialed, open int) {
d.mu.Lock()
if d.dialed != dialed {
d.t.Errorf("%s: dialed=%d, want %d", message, d.dialed, dialed)
}
if d.open != open {
d.t.Errorf("%s: open=%d, want %d", message, d.open, open)
}
// if active := p.ActiveCount(); active != open {
// d.t.Errorf("%s: active=%d, want %d", message, active, open)
// }
d.mu.Unlock()
}
func TestPoolReuse(t *testing.T) {
d := poolDialer{t: t}
p := NewPool(testConfig)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
var err error
for i := 0; i < 10; i++ {
c1 := p.Get(context.TODO())
c1.Do("PING")
c2 := p.Get(context.TODO())
c2.Do("PING")
c1.Close()
c2.Close()
}
d.check("before close", p, 2, 2)
err = p.Close()
if err != nil {
t.Fatal(err)
}
d.check("after close", p, 2, 0)
}
func TestPoolMaxIdle(t *testing.T) {
d := poolDialer{t: t}
p := NewPool(testConfig)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
defer p.Close()
for i := 0; i < 10; i++ {
c1 := p.Get(context.TODO())
c1.Do("PING")
c2 := p.Get(context.TODO())
c2.Do("PING")
c3 := p.Get(context.TODO())
c3.Do("PING")
c1.Close()
c2.Close()
c3.Close()
}
d.check("before close", p, 12, 2)
p.Close()
d.check("after close", p, 12, 0)
}
func TestPoolError(t *testing.T) {
d := poolDialer{t: t}
p := NewPool(testConfig)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
defer p.Close()
c := p.Get(context.TODO())
c.Do("ERR", io.EOF)
if c.Err() == nil {
t.Errorf("expected c.Err() != nil")
}
c.Close()
c = p.Get(context.TODO())
c.Do("ERR", io.EOF)
c.Close()
d.check(".", p, 2, 0)
}
func TestPoolClose(t *testing.T) {
d := poolDialer{t: t}
p := NewPool(testConfig)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
defer p.Close()
c1 := p.Get(context.TODO())
c1.Do("PING")
c2 := p.Get(context.TODO())
c2.Do("PING")
c3 := p.Get(context.TODO())
c3.Do("PING")
c1.Close()
if _, err := c1.Do("PING"); err == nil {
t.Errorf("expected error after connection closed")
}
c2.Close()
c2.Close()
p.Close()
d.check("after pool close", p, 3, 1)
if _, err := c1.Do("PING"); err == nil {
t.Errorf("expected error after connection and pool closed")
}
c3.Close()
d.check("after conn close", p, 3, 0)
c1 = p.Get(context.TODO())
if _, err := c1.Do("PING"); err == nil {
t.Errorf("expected error after pool closed")
}
}
func TestPoolConcurrenSendReceive(t *testing.T) {
p := NewPool(testConfig)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return DialDefaultServer()
}
defer p.Close()
c := p.Get(context.TODO())
done := make(chan error, 1)
go func() {
_, err := c.Receive()
done <- err
}()
c.Send("PING")
c.Flush()
err := <-done
if err != nil {
t.Fatalf("Receive() returned error %v", err)
}
_, err = c.Do("")
if err != nil {
t.Fatalf("Do() returned error %v", err)
}
c.Close()
}
func TestPoolMaxActive(t *testing.T) {
d := poolDialer{t: t}
conf := getTestConfig(testRedisAddr)
conf.Config = &pool.Config{
Active: 2,
Idle: 2,
}
p := NewPool(conf)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
defer p.Close()
c1 := p.Get(context.TODO())
c1.Do("PING")
c2 := p.Get(context.TODO())
c2.Do("PING")
d.check("1", p, 2, 2)
c3 := p.Get(context.TODO())
if _, err := c3.Do("PING"); err != pool.ErrPoolExhausted {
t.Errorf("expected pool exhausted")
}
c3.Close()
d.check("2", p, 2, 2)
c2.Close()
d.check("3", p, 2, 2)
c3 = p.Get(context.TODO())
if _, err := c3.Do("PING"); err != nil {
t.Errorf("expected good channel, err=%v", err)
}
c3.Close()
d.check("4", p, 2, 2)
}
func TestPoolMonitorCleanup(t *testing.T) {
d := poolDialer{t: t}
p := NewPool(testConfig)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
defer p.Close()
c := p.Get(context.TODO())
c.Send("MONITOR")
c.Close()
d.check("", p, 1, 0)
}
func TestPoolPubSubCleanup(t *testing.T) {
d := poolDialer{t: t}
p := NewPool(testConfig)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
defer p.Close()
c := p.Get(context.TODO())
c.Send("SUBSCRIBE", "x")
c.Close()
want := []string{"SUBSCRIBE", "UNSUBSCRIBE", "PUNSUBSCRIBE", "ECHO"}
if !reflect.DeepEqual(d.commands, want) {
t.Errorf("got commands %v, want %v", d.commands, want)
}
d.commands = nil
c = p.Get(context.TODO())
c.Send("PSUBSCRIBE", "x*")
c.Close()
want = []string{"PSUBSCRIBE", "UNSUBSCRIBE", "PUNSUBSCRIBE", "ECHO"}
if !reflect.DeepEqual(d.commands, want) {
t.Errorf("got commands %v, want %v", d.commands, want)
}
d.commands = nil
}
func TestPoolTransactionCleanup(t *testing.T) {
d := poolDialer{t: t}
p := NewPool(testConfig)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
defer p.Close()
c := p.Get(context.TODO())
c.Do("WATCH", "key")
c.Do("PING")
c.Close()
want := []string{"WATCH", "PING", "UNWATCH"}
if !reflect.DeepEqual(d.commands, want) {
t.Errorf("got commands %v, want %v", d.commands, want)
}
d.commands = nil
c = p.Get(context.TODO())
c.Do("WATCH", "key")
c.Do("UNWATCH")
c.Do("PING")
c.Close()
want = []string{"WATCH", "UNWATCH", "PING"}
if !reflect.DeepEqual(d.commands, want) {
t.Errorf("got commands %v, want %v", d.commands, want)
}
d.commands = nil
c = p.Get(context.TODO())
c.Do("WATCH", "key")
c.Do("MULTI")
c.Do("PING")
c.Close()
want = []string{"WATCH", "MULTI", "PING", "DISCARD"}
if !reflect.DeepEqual(d.commands, want) {
t.Errorf("got commands %v, want %v", d.commands, want)
}
d.commands = nil
c = p.Get(context.TODO())
c.Do("WATCH", "key")
c.Do("MULTI")
c.Do("DISCARD")
c.Do("PING")
c.Close()
want = []string{"WATCH", "MULTI", "DISCARD", "PING"}
if !reflect.DeepEqual(d.commands, want) {
t.Errorf("got commands %v, want %v", d.commands, want)
}
d.commands = nil
c = p.Get(context.TODO())
c.Do("WATCH", "key")
c.Do("MULTI")
c.Do("EXEC")
c.Do("PING")
c.Close()
want = []string{"WATCH", "MULTI", "EXEC", "PING"}
if !reflect.DeepEqual(d.commands, want) {
t.Errorf("got commands %v, want %v", d.commands, want)
}
d.commands = nil
}
func startGoroutines(p *Pool, cmd string, args ...interface{}) chan error {
errs := make(chan error, 10)
for i := 0; i < cap(errs); i++ {
go func() {
c := p.Get(context.TODO())
_, err := c.Do(cmd, args...)
errs <- err
c.Close()
}()
}
// Wait for goroutines to block.
time.Sleep(time.Second / 4)
return errs
}
func TestWaitPoolDialError(t *testing.T) {
testErr := errors.New("test")
d := poolDialer{t: t}
config1 := testConfig
config1.Config = &pool.Config{
Active: 1,
Idle: 1,
Wait: true,
}
p := NewPool(config1)
p.Slice.New = func(ctx context.Context) (io.Closer, error) {
return d.dial()
}
defer p.Close()
c := p.Get(context.TODO())
errs := startGoroutines(p, "ERR", testErr)
d.check("before close", p, 1, 1)
d.dialErr = errors.New("dial")
c.Close()
nilCount := 0
errCount := 0
timeout := time.After(2 * time.Second)
for i := 0; i < cap(errs); i++ {
select {
case err := <-errs:
switch err {
case nil:
nilCount++
case d.dialErr:
errCount++
default:
t.Fatalf("expected dial error or nil, got %v", err)
}
case <-timeout:
t.Logf("Wait all the time and timeout %d", i)
return
}
}
if nilCount != 1 {
t.Errorf("expected one nil error, got %d", nilCount)
}
if errCount != cap(errs)-1 {
t.Errorf("expected %d dial erors, got %d", cap(errs)-1, errCount)
}
d.check("done", p, cap(errs), 0)
}
func BenchmarkPoolGet(b *testing.B) {
b.StopTimer()
p := NewPool(testConfig)
c := p.Get(context.Background())
if err := c.Err(); err != nil {
b.Fatal(err)
}
c.Close()
defer p.Close()
b.StartTimer()
for i := 0; i < b.N; i++ {
c := p.Get(context.Background())
c.Close()
}
}
func BenchmarkPoolGetErr(b *testing.B) {
b.StopTimer()
p := NewPool(testConfig)
c := p.Get(context.Background())
if err := c.Err(); err != nil {
b.Fatal(err)
}
c.Close()
defer p.Close()
b.StartTimer()
for i := 0; i < b.N; i++ {
c = p.Get(context.Background())
if err := c.Err(); err != nil {
b.Fatal(err)
}
c.Close()
}
}
func BenchmarkPoolGetPing(b *testing.B) {
b.StopTimer()
p := NewPool(testConfig)
c := p.Get(context.Background())
if err := c.Err(); err != nil {
b.Fatal(err)
}
c.Close()
defer p.Close()
b.StartTimer()
for i := 0; i < b.N; i++ {
c := p.Get(context.Background())
if _, err := c.Do("PING"); err != nil {
b.Fatal(err)
}
c.Close()
}
}
func BenchmarkPooledConn(b *testing.B) {
p := NewPool(testConfig)
defer p.Close()
for i := 0; i < b.N; i++ {
ctx := context.TODO()
c := p.Get(ctx)
c2 := c.WithContext(context.TODO())
if _, err := c2.Do("PING"); err != nil {
b.Fatal(err)
}
c2.Close()
}
}

146
pkg/cache/redis/pubsub_test.go vendored Normal file
View File

@ -0,0 +1,146 @@
// Copyright 2012 Gary Burd
//
// Licensed under the Apache License, Version 2.0 (the "License"): you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations
// under the License.
package redis
import (
"fmt"
"reflect"
"sync"
"testing"
)
func publish(channel, value interface{}) {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
c.Do("PUBLISH", channel, value)
}
// Applications can receive pushed messages from one goroutine and manage subscriptions from another goroutine.
func ExamplePubSubConn() {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
var wg sync.WaitGroup
wg.Add(2)
psc := PubSubConn{Conn: c}
// This goroutine receives and prints pushed notifications from the server.
// The goroutine exits when the connection is unsubscribed from all
// channels or there is an error.
go func() {
defer wg.Done()
for {
switch n := psc.Receive().(type) {
case Message:
fmt.Printf("Message: %s %s\n", n.Channel, n.Data)
case PMessage:
fmt.Printf("PMessage: %s %s %s\n", n.Pattern, n.Channel, n.Data)
case Subscription:
fmt.Printf("Subscription: %s %s %d\n", n.Kind, n.Channel, n.Count)
if n.Count == 0 {
return
}
case error:
fmt.Printf("error: %v\n", n)
return
}
}
}()
// This goroutine manages subscriptions for the connection.
go func() {
defer wg.Done()
psc.Subscribe("example")
psc.PSubscribe("p*")
// The following function calls publish a message using another
// connection to the Redis server.
publish("example", "hello")
publish("example", "world")
publish("pexample", "foo")
publish("pexample", "bar")
// Unsubscribe from all connections. This will cause the receiving
// goroutine to exit.
psc.Unsubscribe()
psc.PUnsubscribe()
}()
wg.Wait()
// Output:
// Subscription: subscribe example 1
// Subscription: psubscribe p* 2
// Message: example hello
// Message: example world
// PMessage: p* pexample foo
// PMessage: p* pexample bar
// Subscription: unsubscribe example 1
// Subscription: punsubscribe p* 0
}
func expectPushed(t *testing.T, c PubSubConn, message string, expected interface{}) {
actual := c.Receive()
if !reflect.DeepEqual(actual, expected) {
t.Errorf("%s = %v, want %v", message, actual, expected)
}
}
func TestPushed(t *testing.T) {
pc, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer pc.Close()
sc, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer sc.Close()
c := PubSubConn{Conn: sc}
c.Subscribe("c1")
expectPushed(t, c, "Subscribe(c1)", Subscription{Kind: "subscribe", Channel: "c1", Count: 1})
c.Subscribe("c2")
expectPushed(t, c, "Subscribe(c2)", Subscription{Kind: "subscribe", Channel: "c2", Count: 2})
c.PSubscribe("p1")
expectPushed(t, c, "PSubscribe(p1)", Subscription{Kind: "psubscribe", Channel: "p1", Count: 3})
c.PSubscribe("p2")
expectPushed(t, c, "PSubscribe(p2)", Subscription{Kind: "psubscribe", Channel: "p2", Count: 4})
c.PUnsubscribe()
expectPushed(t, c, "Punsubscribe(p1)", Subscription{Kind: "punsubscribe", Channel: "p1", Count: 3})
expectPushed(t, c, "Punsubscribe()", Subscription{Kind: "punsubscribe", Channel: "p2", Count: 2})
pc.Do("PUBLISH", "c1", "hello")
expectPushed(t, c, "PUBLISH c1 hello", Message{Channel: "c1", Data: []byte("hello")})
c.Ping("hello")
expectPushed(t, c, `Ping("hello")`, Pong{"hello"})
c.Conn.Send("PING")
c.Conn.Flush()
expectPushed(t, c, `Send("PING")`, Pong{})
}

View File

@ -16,6 +16,9 @@ package redis
import (
"context"
"github.com/bilibili/kratos/pkg/container/pool"
xtime "github.com/bilibili/kratos/pkg/time"
)
// Error represents an error returned in a command reply.
@ -23,29 +26,53 @@ type Error string
func (err Error) Error() string { return string(err) }
// Conn represents a connection to a Redis server.
type Conn interface {
// Close closes the connection.
Close() error
// Config client settings.
type Config struct {
*pool.Config
// Err returns a non-nil value if the connection is broken. The returned
// value is either the first non-nil value returned from the underlying
// network connection or a protocol parsing error. Applications should
// close broken connections.
Err() error
// Do sends a command to the server and returns the received reply.
Do(commandName string, args ...interface{}) (reply interface{}, err error)
// Send writes the command to the client's output buffer.
Send(commandName string, args ...interface{}) error
// Flush flushes the output buffer to the Redis server.
Flush() error
// Receive receives a single reply from the Redis server
Receive() (reply interface{}, err error)
// WithContext
WithContext(ctx context.Context) Conn
Name string // redis name, for trace
Proto string
Addr string
Auth string
DialTimeout xtime.Duration
ReadTimeout xtime.Duration
WriteTimeout xtime.Duration
SlowLog xtime.Duration
}
type Redis struct {
pool *Pool
conf *Config
}
func NewRedis(c *Config, options ...DialOption) *Redis {
return &Redis{
pool: NewPool(c, options...),
conf: c,
}
}
// Do gets a new conn from pool, then execute Do with this conn, finally close this conn.
// ATTENTION: Don't use this method with transaction command like MULTI etc. Because every Do will close conn automatically, use r.Conn to get a raw conn for this situation.
func (r *Redis) Do(ctx context.Context, commandName string, args ...interface{}) (reply interface{}, err error) {
conn := r.pool.Get(ctx)
defer conn.Close()
reply, err = conn.Do(commandName, args...)
return
}
// Close closes connection pool
func (r *Redis) Close() error {
return r.pool.Close()
}
// Conn direct gets a connection
func (r *Redis) Conn(ctx context.Context) Conn {
return r.pool.Get(ctx)
}
func (r *Redis) Pipeline() (p Pipeliner) {
return &pipeliner{
pool: r.pool,
}
}

324
pkg/cache/redis/redis_test.go vendored Normal file
View File

@ -0,0 +1,324 @@
package redis
import (
"context"
"reflect"
"testing"
"time"
"github.com/bilibili/kratos/pkg/container/pool"
xtime "github.com/bilibili/kratos/pkg/time"
)
func TestRedis(t *testing.T) {
testSet(t, testPool)
testSend(t, testPool)
testGet(t, testPool)
testErr(t, testPool)
if err := testPool.Close(); err != nil {
t.Errorf("redis: close error(%v)", err)
}
conn, err := NewConn(testConfig)
if err != nil {
t.Errorf("redis: new conn error(%v)", err)
}
if err := conn.Close(); err != nil {
t.Errorf("redis: close error(%v)", err)
}
}
func testSet(t *testing.T, p *Pool) {
var (
key = "test"
value = "test"
conn = p.Get(context.TODO())
)
defer conn.Close()
if reply, err := conn.Do("set", key, value); err != nil {
t.Errorf("redis: conn.Do(SET, %s, %s) error(%v)", key, value, err)
} else {
t.Logf("redis: set status: %s", reply)
}
}
func testSend(t *testing.T, p *Pool) {
var (
key = "test"
value = "test"
expire = 1000
conn = p.Get(context.TODO())
)
defer conn.Close()
if err := conn.Send("SET", key, value); err != nil {
t.Errorf("redis: conn.Send(SET, %s, %s) error(%v)", key, value, err)
}
if err := conn.Send("EXPIRE", key, expire); err != nil {
t.Errorf("redis: conn.Send(EXPIRE key(%s) expire(%d)) error(%v)", key, expire, err)
}
if err := conn.Flush(); err != nil {
t.Errorf("redis: conn.Flush error(%v)", err)
}
for i := 0; i < 2; i++ {
if _, err := conn.Receive(); err != nil {
t.Errorf("redis: conn.Receive error(%v)", err)
return
}
}
t.Logf("redis: set value: %s", value)
}
func testGet(t *testing.T, p *Pool) {
var (
key = "test"
conn = p.Get(context.TODO())
)
defer conn.Close()
if reply, err := conn.Do("GET", key); err != nil {
t.Errorf("redis: conn.Do(GET, %s) error(%v)", key, err)
} else {
t.Logf("redis: get value: %s", reply)
}
}
func testErr(t *testing.T, p *Pool) {
conn := p.Get(context.TODO())
if err := conn.Close(); err != nil {
t.Errorf("redis: close error(%v)", err)
}
if err := conn.Err(); err == nil {
t.Errorf("redis: err not nil")
} else {
t.Logf("redis: err: %v", err)
}
}
func BenchmarkRedis(b *testing.B) {
conf := &Config{
Name: "test",
Proto: "tcp",
Addr: testRedisAddr,
DialTimeout: xtime.Duration(time.Second),
ReadTimeout: xtime.Duration(time.Second),
WriteTimeout: xtime.Duration(time.Second),
}
conf.Config = &pool.Config{
Active: 10,
Idle: 5,
IdleTimeout: xtime.Duration(90 * time.Second),
}
benchmarkPool := NewPool(conf)
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
conn := benchmarkPool.Get(context.TODO())
if err := conn.Close(); err != nil {
b.Errorf("redis: close error(%v)", err)
}
}
})
if err := benchmarkPool.Close(); err != nil {
b.Errorf("redis: close error(%v)", err)
}
}
var testRedisCommands = []struct {
args []interface{}
expected interface{}
}{
{
[]interface{}{"PING"},
"PONG",
},
{
[]interface{}{"SET", "foo", "bar"},
"OK",
},
{
[]interface{}{"GET", "foo"},
[]byte("bar"),
},
{
[]interface{}{"GET", "nokey"},
nil,
},
{
[]interface{}{"MGET", "nokey", "foo"},
[]interface{}{nil, []byte("bar")},
},
{
[]interface{}{"INCR", "mycounter"},
int64(1),
},
{
[]interface{}{"LPUSH", "mylist", "foo"},
int64(1),
},
{
[]interface{}{"LPUSH", "mylist", "bar"},
int64(2),
},
{
[]interface{}{"LRANGE", "mylist", 0, -1},
[]interface{}{[]byte("bar"), []byte("foo")},
},
}
func TestNewRedis(t *testing.T) {
type args struct {
c *Config
options []DialOption
}
tests := []struct {
name string
args args
wantErr bool
}{
{
"new_redis",
args{
testConfig,
make([]DialOption, 0),
},
false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
r := NewRedis(tt.args.c, tt.args.options...)
if r == nil {
t.Errorf("NewRedis() error, got nil")
return
}
err := r.Close()
if err != nil {
t.Errorf("Close() error %v", err)
}
})
}
}
func TestRedis_Do(t *testing.T) {
r := NewRedis(testConfig)
r.Do(context.TODO(), "FLUSHDB")
for _, cmd := range testRedisCommands {
actual, err := r.Do(context.TODO(), cmd.args[0].(string), cmd.args[1:]...)
if err != nil {
t.Errorf("Do(%v) returned error %v", cmd.args, err)
continue
}
if !reflect.DeepEqual(actual, cmd.expected) {
t.Errorf("Do(%v) = %v, want %v", cmd.args, actual, cmd.expected)
}
}
err := r.Close()
if err != nil {
t.Errorf("Close() error %v", err)
}
}
func TestRedis_Conn(t *testing.T) {
type args struct {
ctx context.Context
}
tests := []struct {
name string
p *Redis
args args
wantErr bool
g int
c int
}{
{
"Close",
NewRedis(&Config{
Config: &pool.Config{
Active: 1,
Idle: 1,
},
Name: "test_get",
Proto: "tcp",
Addr: testRedisAddr,
DialTimeout: xtime.Duration(time.Second),
ReadTimeout: xtime.Duration(time.Second),
WriteTimeout: xtime.Duration(time.Second),
}),
args{context.TODO()},
false,
3,
3,
},
{
"CloseExceededPoolSize",
NewRedis(&Config{
Config: &pool.Config{
Active: 1,
Idle: 1,
},
Name: "test_get_out",
Proto: "tcp",
Addr: testRedisAddr,
DialTimeout: xtime.Duration(time.Second),
ReadTimeout: xtime.Duration(time.Second),
WriteTimeout: xtime.Duration(time.Second),
}),
args{context.TODO()},
true,
5,
3,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
for i := 1; i <= tt.g; i++ {
got := tt.p.Conn(tt.args.ctx)
if err := got.Close(); err != nil {
if !tt.wantErr {
t.Error(err)
}
}
if i <= tt.c {
if err := got.Close(); err != nil {
t.Error(err)
}
}
}
})
}
}
func BenchmarkRedisDoPing(b *testing.B) {
r := NewRedis(testConfig)
defer r.Close()
b.ResetTimer()
for i := 0; i < b.N; i++ {
if _, err := r.Do(context.Background(), "PING"); err != nil {
b.Fatal(err)
}
}
}
func BenchmarkRedisDoSET(b *testing.B) {
r := NewRedis(testConfig)
defer r.Close()
b.ResetTimer()
for i := 0; i < b.N; i++ {
if _, err := r.Do(context.Background(), "SET", "a", "b"); err != nil {
b.Fatal(err)
}
}
}
func BenchmarkRedisDoGET(b *testing.B) {
r := NewRedis(testConfig)
defer r.Close()
r.Do(context.Background(), "SET", "a", "b")
b.ResetTimer()
for i := 0; i < b.N; i++ {
if _, err := r.Do(context.Background(), "GET", "b"); err != nil {
b.Fatal(err)
}
}
}

179
pkg/cache/redis/reply_test.go vendored Normal file
View File

@ -0,0 +1,179 @@
// Copyright 2012 Gary Burd
//
// Licensed under the Apache License, Version 2.0 (the "License"): you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations
// under the License.
package redis
import (
"fmt"
"reflect"
"testing"
"github.com/pkg/errors"
)
type valueError struct {
v interface{}
err error
}
func ve(v interface{}, err error) valueError {
return valueError{v, err}
}
var replyTests = []struct {
name interface{}
actual valueError
expected valueError
}{
{
"ints([v1, v2])",
ve(Ints([]interface{}{[]byte("4"), []byte("5")}, nil)),
ve([]int{4, 5}, nil),
},
{
"ints(nil)",
ve(Ints(nil, nil)),
ve([]int(nil), ErrNil),
},
{
"strings([v1, v2])",
ve(Strings([]interface{}{[]byte("v1"), []byte("v2")}, nil)),
ve([]string{"v1", "v2"}, nil),
},
{
"strings(nil)",
ve(Strings(nil, nil)),
ve([]string(nil), ErrNil),
},
{
"byteslices([v1, v2])",
ve(ByteSlices([]interface{}{[]byte("v1"), []byte("v2")}, nil)),
ve([][]byte{[]byte("v1"), []byte("v2")}, nil),
},
{
"byteslices(nil)",
ve(ByteSlices(nil, nil)),
ve([][]byte(nil), ErrNil),
},
{
"values([v1, v2])",
ve(Values([]interface{}{[]byte("v1"), []byte("v2")}, nil)),
ve([]interface{}{[]byte("v1"), []byte("v2")}, nil),
},
{
"values(nil)",
ve(Values(nil, nil)),
ve([]interface{}(nil), ErrNil),
},
{
"float64(1.0)",
ve(Float64([]byte("1.0"), nil)),
ve(float64(1.0), nil),
},
{
"float64(nil)",
ve(Float64(nil, nil)),
ve(float64(0.0), ErrNil),
},
{
"uint64(1)",
ve(Uint64(int64(1), nil)),
ve(uint64(1), nil),
},
{
"uint64(-1)",
ve(Uint64(int64(-1), nil)),
ve(uint64(0), errNegativeInt),
},
}
func TestReply(t *testing.T) {
for _, rt := range replyTests {
if errors.Cause(rt.actual.err) != rt.expected.err {
t.Errorf("%s returned err %v, want %v", rt.name, rt.actual.err, rt.expected.err)
continue
}
if !reflect.DeepEqual(rt.actual.v, rt.expected.v) {
t.Errorf("%s=%+v, want %+v", rt.name, rt.actual.v, rt.expected.v)
}
}
}
// dial wraps DialDefaultServer() with a more suitable function name for examples.
func dial() (Conn, error) {
return DialDefaultServer()
}
func ExampleBool() {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
c.Do("SET", "foo", 1)
exists, _ := Bool(c.Do("EXISTS", "foo"))
fmt.Printf("%#v\n", exists)
// Output:
// true
}
func ExampleInt() {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
c.Do("SET", "k1", 1)
n, _ := Int(c.Do("GET", "k1"))
fmt.Printf("%#v\n", n)
n, _ = Int(c.Do("INCR", "k1"))
fmt.Printf("%#v\n", n)
// Output:
// 1
// 2
}
func ExampleInts() {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
c.Do("SADD", "set_with_integers", 4, 5, 6)
ints, _ := Ints(c.Do("SMEMBERS", "set_with_integers"))
fmt.Printf("%#v\n", ints)
// Output:
// []int{4, 5, 6}
}
func ExampleString() {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
c.Do("SET", "hello", "world")
s, _ := String(c.Do("GET", "hello"))
fmt.Printf("%#v", s)
// Output:
// "world"
}

438
pkg/cache/redis/scan_test.go vendored Normal file
View File

@ -0,0 +1,438 @@
// Copyright 2012 Gary Burd
//
// Licensed under the Apache License, Version 2.0 (the "License"): you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations
// under the License.
package redis
import (
"fmt"
"math"
"reflect"
"testing"
)
var scanConversionTests = []struct {
src interface{}
dest interface{}
}{
{[]byte("-inf"), math.Inf(-1)},
{[]byte("+inf"), math.Inf(1)},
{[]byte("0"), float64(0)},
{[]byte("3.14159"), float64(3.14159)},
{[]byte("3.14"), float32(3.14)},
{[]byte("-100"), int(-100)},
{[]byte("101"), int(101)},
{int64(102), int(102)},
{[]byte("103"), uint(103)},
{int64(104), uint(104)},
{[]byte("105"), int8(105)},
{int64(106), int8(106)},
{[]byte("107"), uint8(107)},
{int64(108), uint8(108)},
{[]byte("0"), false},
{int64(0), false},
{[]byte("f"), false},
{[]byte("1"), true},
{int64(1), true},
{[]byte("t"), true},
{"hello", "hello"},
{[]byte("hello"), "hello"},
{[]byte("world"), []byte("world")},
{[]interface{}{[]byte("foo")}, []interface{}{[]byte("foo")}},
{[]interface{}{[]byte("foo")}, []string{"foo"}},
{[]interface{}{[]byte("hello"), []byte("world")}, []string{"hello", "world"}},
{[]interface{}{[]byte("bar")}, [][]byte{[]byte("bar")}},
{[]interface{}{[]byte("1")}, []int{1}},
{[]interface{}{[]byte("1"), []byte("2")}, []int{1, 2}},
{[]interface{}{[]byte("1"), []byte("2")}, []float64{1, 2}},
{[]interface{}{[]byte("1")}, []byte{1}},
{[]interface{}{[]byte("1")}, []bool{true}},
}
func TestScanConversion(t *testing.T) {
for _, tt := range scanConversionTests {
values := []interface{}{tt.src}
dest := reflect.New(reflect.TypeOf(tt.dest))
values, err := Scan(values, dest.Interface())
if err != nil {
t.Errorf("Scan(%v) returned error %v", tt, err)
continue
}
if !reflect.DeepEqual(tt.dest, dest.Elem().Interface()) {
t.Errorf("Scan(%v) returned %v values: %v, want %v", tt, dest.Elem().Interface(), values, tt.dest)
}
}
}
var scanConversionErrorTests = []struct {
src interface{}
dest interface{}
}{
{[]byte("1234"), byte(0)},
{int64(1234), byte(0)},
{[]byte("-1"), byte(0)},
{int64(-1), byte(0)},
{[]byte("junk"), false},
{Error("blah"), false},
}
func TestScanConversionError(t *testing.T) {
for _, tt := range scanConversionErrorTests {
values := []interface{}{tt.src}
dest := reflect.New(reflect.TypeOf(tt.dest))
values, err := Scan(values, dest.Interface())
if err == nil {
t.Errorf("Scan(%v) did not return error values: %v", tt, values)
}
}
}
func ExampleScan() {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
c.Send("HMSET", "album:1", "title", "Red", "rating", 5)
c.Send("HMSET", "album:2", "title", "Earthbound", "rating", 1)
c.Send("HMSET", "album:3", "title", "Beat")
c.Send("LPUSH", "albums", "1")
c.Send("LPUSH", "albums", "2")
c.Send("LPUSH", "albums", "3")
values, err := Values(c.Do("SORT", "albums",
"BY", "album:*->rating",
"GET", "album:*->title",
"GET", "album:*->rating"))
if err != nil {
fmt.Println(err)
return
}
for len(values) > 0 {
var title string
rating := -1 // initialize to illegal value to detect nil.
values, err = Scan(values, &title, &rating)
if err != nil {
fmt.Println(err)
return
}
if rating == -1 {
fmt.Println(title, "not-rated")
} else {
fmt.Println(title, rating)
}
}
// Output:
// Beat not-rated
// Earthbound 1
// Red 5
}
type s0 struct {
X int
Y int `redis:"y"`
Bt bool
}
type s1 struct {
X int `redis:"-"`
I int `redis:"i"`
U uint `redis:"u"`
S string `redis:"s"`
P []byte `redis:"p"`
B bool `redis:"b"`
Bt bool
Bf bool
s0
}
var scanStructTests = []struct {
title string
reply []string
value interface{}
}{
{"basic",
[]string{"i", "-1234", "u", "5678", "s", "hello", "p", "world", "b", "t", "Bt", "1", "Bf", "0", "X", "123", "y", "456"},
&s1{I: -1234, U: 5678, S: "hello", P: []byte("world"), B: true, Bt: true, Bf: false, s0: s0{X: 123, Y: 456}},
},
}
func TestScanStruct(t *testing.T) {
for _, tt := range scanStructTests {
var reply []interface{}
for _, v := range tt.reply {
reply = append(reply, []byte(v))
}
value := reflect.New(reflect.ValueOf(tt.value).Type().Elem())
if err := ScanStruct(reply, value.Interface()); err != nil {
t.Fatalf("ScanStruct(%s) returned error %v", tt.title, err)
}
if !reflect.DeepEqual(value.Interface(), tt.value) {
t.Fatalf("ScanStruct(%s) returned %v, want %v", tt.title, value.Interface(), tt.value)
}
}
}
func TestBadScanStructArgs(t *testing.T) {
x := []interface{}{"A", "b"}
test := func(v interface{}) {
if err := ScanStruct(x, v); err == nil {
t.Errorf("Expect error for ScanStruct(%T, %T)", x, v)
}
}
test(nil)
var v0 *struct{}
test(v0)
var v1 int
test(&v1)
x = x[:1]
v2 := struct{ A string }{}
test(&v2)
}
var scanSliceTests = []struct {
src []interface{}
fieldNames []string
ok bool
dest interface{}
}{
{
[]interface{}{[]byte("1"), nil, []byte("-1")},
nil,
true,
[]int{1, 0, -1},
},
{
[]interface{}{[]byte("1"), nil, []byte("2")},
nil,
true,
[]uint{1, 0, 2},
},
{
[]interface{}{[]byte("-1")},
nil,
false,
[]uint{1},
},
{
[]interface{}{[]byte("hello"), nil, []byte("world")},
nil,
true,
[][]byte{[]byte("hello"), nil, []byte("world")},
},
{
[]interface{}{[]byte("hello"), nil, []byte("world")},
nil,
true,
[]string{"hello", "", "world"},
},
{
[]interface{}{[]byte("a1"), []byte("b1"), []byte("a2"), []byte("b2")},
nil,
true,
[]struct{ A, B string }{{"a1", "b1"}, {"a2", "b2"}},
},
{
[]interface{}{[]byte("a1"), []byte("b1")},
nil,
false,
[]struct{ A, B, C string }{{"a1", "b1", ""}},
},
{
[]interface{}{[]byte("a1"), []byte("b1"), []byte("a2"), []byte("b2")},
nil,
true,
[]*struct{ A, B string }{{"a1", "b1"}, {"a2", "b2"}},
},
{
[]interface{}{[]byte("a1"), []byte("b1"), []byte("a2"), []byte("b2")},
[]string{"A", "B"},
true,
[]struct{ A, C, B string }{{"a1", "", "b1"}, {"a2", "", "b2"}},
},
{
[]interface{}{[]byte("a1"), []byte("b1"), []byte("a2"), []byte("b2")},
nil,
false,
[]struct{}{},
},
}
func TestScanSlice(t *testing.T) {
for _, tt := range scanSliceTests {
typ := reflect.ValueOf(tt.dest).Type()
dest := reflect.New(typ)
err := ScanSlice(tt.src, dest.Interface(), tt.fieldNames...)
if tt.ok != (err == nil) {
t.Errorf("ScanSlice(%v, []%s, %v) returned error %v", tt.src, typ, tt.fieldNames, err)
continue
}
if tt.ok && !reflect.DeepEqual(dest.Elem().Interface(), tt.dest) {
t.Errorf("ScanSlice(src, []%s) returned %#v, want %#v", typ, dest.Elem().Interface(), tt.dest)
}
}
}
func ExampleScanSlice() {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
c.Send("HMSET", "album:1", "title", "Red", "rating", 5)
c.Send("HMSET", "album:2", "title", "Earthbound", "rating", 1)
c.Send("HMSET", "album:3", "title", "Beat", "rating", 4)
c.Send("LPUSH", "albums", "1")
c.Send("LPUSH", "albums", "2")
c.Send("LPUSH", "albums", "3")
values, err := Values(c.Do("SORT", "albums",
"BY", "album:*->rating",
"GET", "album:*->title",
"GET", "album:*->rating"))
if err != nil {
fmt.Println(err)
return
}
var albums []struct {
Title string
Rating int
}
if err := ScanSlice(values, &albums); err != nil {
fmt.Println(err)
return
}
fmt.Printf("%v\n", albums)
// Output:
// [{Earthbound 1} {Beat 4} {Red 5}]
}
var argsTests = []struct {
title string
actual Args
expected Args
}{
{"struct ptr",
Args{}.AddFlat(&struct {
I int `redis:"i"`
U uint `redis:"u"`
S string `redis:"s"`
P []byte `redis:"p"`
M map[string]string `redis:"m"`
Bt bool
Bf bool
}{
-1234, 5678, "hello", []byte("world"), map[string]string{"hello": "world"}, true, false,
}),
Args{"i", int(-1234), "u", uint(5678), "s", "hello", "p", []byte("world"), "m", map[string]string{"hello": "world"}, "Bt", true, "Bf", false},
},
{"struct",
Args{}.AddFlat(struct{ I int }{123}),
Args{"I", 123},
},
{"slice",
Args{}.Add(1).AddFlat([]string{"a", "b", "c"}).Add(2),
Args{1, "a", "b", "c", 2},
},
{"struct omitempty",
Args{}.AddFlat(&struct {
I int `redis:"i,omitempty"`
U uint `redis:"u,omitempty"`
S string `redis:"s,omitempty"`
P []byte `redis:"p,omitempty"`
M map[string]string `redis:"m,omitempty"`
Bt bool `redis:"Bt,omitempty"`
Bf bool `redis:"Bf,omitempty"`
}{
0, 0, "", []byte{}, map[string]string{}, true, false,
}),
Args{"Bt", true},
},
}
func TestArgs(t *testing.T) {
for _, tt := range argsTests {
if !reflect.DeepEqual(tt.actual, tt.expected) {
t.Fatalf("%s is %v, want %v", tt.title, tt.actual, tt.expected)
}
}
}
func ExampleArgs() {
c, err := dial()
if err != nil {
fmt.Println(err)
return
}
defer c.Close()
var p1, p2 struct {
Title string `redis:"title"`
Author string `redis:"author"`
Body string `redis:"body"`
}
p1.Title = "Example"
p1.Author = "Gary"
p1.Body = "Hello"
if _, err := c.Do("HMSET", Args{}.Add("id1").AddFlat(&p1)...); err != nil {
fmt.Println(err)
return
}
m := map[string]string{
"title": "Example2",
"author": "Steve",
"body": "Map",
}
if _, err := c.Do("HMSET", Args{}.Add("id2").AddFlat(m)...); err != nil {
fmt.Println(err)
return
}
for _, id := range []string{"id1", "id2"} {
v, err := Values(c.Do("HGETALL", id))
if err != nil {
fmt.Println(err)
return
}
if err := ScanStruct(v, &p2); err != nil {
fmt.Println(err)
return
}
fmt.Printf("%+v\n", p2)
}
// Output:
// {Title:Example Author:Gary Body:Hello}
// {Title:Example2 Author:Steve Body:Map}
}

103
pkg/cache/redis/script_test.go vendored Normal file
View File

@ -0,0 +1,103 @@
// Copyright 2012 Gary Burd
//
// Licensed under the Apache License, Version 2.0 (the "License"): you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations
// under the License.
package redis
import (
"fmt"
"reflect"
"testing"
"time"
)
func ExampleScript() {
c, err := Dial("tcp", ":6379")
if err != nil {
// handle error
}
defer c.Close()
// Initialize a package-level variable with a script.
var getScript = NewScript(1, `return call('get', KEYS[1])`)
// In a function, use the script Do method to evaluate the script. The Do
// method optimistically uses the EVALSHA command. If the script is not
// loaded, then the Do method falls back to the EVAL command.
if _, err = getScript.Do(c, "foo"); err != nil {
// handle error
}
}
func TestScript(t *testing.T) {
c, err := DialDefaultServer()
if err != nil {
t.Fatalf("error connection to database, %v", err)
}
defer c.Close()
// To test fall back in Do, we make script unique by adding comment with current time.
script := fmt.Sprintf("--%d\nreturn {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}", time.Now().UnixNano())
s := NewScript(2, script)
reply := []interface{}{[]byte("key1"), []byte("key2"), []byte("arg1"), []byte("arg2")}
v, err := s.Do(c, "key1", "key2", "arg1", "arg2")
if err != nil {
t.Errorf("s.Do(c, ...) returned %v", err)
}
if !reflect.DeepEqual(v, reply) {
t.Errorf("s.Do(c, ..); = %v, want %v", v, reply)
}
err = s.Load(c)
if err != nil {
t.Errorf("s.Load(c) returned %v", err)
}
err = s.SendHash(c, "key1", "key2", "arg1", "arg2")
if err != nil {
t.Errorf("s.SendHash(c, ...) returned %v", err)
}
err = c.Flush()
if err != nil {
t.Errorf("c.Flush() returned %v", err)
}
v, err = c.Receive()
if err != nil {
t.Errorf("c.Receive() returned %v", err)
}
if !reflect.DeepEqual(v, reply) {
t.Errorf("s.SendHash(c, ..); c.Receive() = %v, want %v", v, reply)
}
err = s.Send(c, "key1", "key2", "arg1", "arg2")
if err != nil {
t.Errorf("s.Send(c, ...) returned %v", err)
}
err = c.Flush()
if err != nil {
t.Errorf("c.Flush() returned %v", err)
}
v, err = c.Receive()
if err != nil {
t.Errorf("c.Receive() returned %v", err)
}
if !reflect.DeepEqual(v, reply) {
t.Errorf("s.Send(c, ..); c.Receive() = %v, want %v", v, reply)
}
}

View File

@ -0,0 +1,12 @@
version: "3.7"
services:
redis:
image: redis
ports:
- 6379:6379
healthcheck:
test: ["CMD", "redis-cli","ping"]
interval: 20s
timeout: 1s
retries: 20

View File

@ -10,10 +10,9 @@ import (
)
const (
_traceComponentName = "pkg/cache/redis"
_traceComponentName = "library/cache/redis"
_tracePeerService = "redis"
_traceSpanKind = "client"
_slowLogDuration = time.Millisecond * 250
)
var _internalTags = []trace.Tag{
@ -24,26 +23,28 @@ var _internalTags = []trace.Tag{
type traceConn struct {
// tr for pipeline, if tr != nil meaning on pipeline
tr trace.Trace
ctx context.Context
tr trace.Trace
// connTag include e.g. ip,port
connTags []trace.Tag
ctx context.Context
// origin redis conn
Conn
pending int
// TODO: split slow log from trace.
slowLogThreshold time.Duration
}
func (t *traceConn) Do(commandName string, args ...interface{}) (reply interface{}, err error) {
statement := getStatement(commandName, args...)
defer slowLog(statement, time.Now())
root, ok := trace.FromContext(t.ctx)
defer t.slowLog(statement, time.Now())
// NOTE: ignored empty commandName
// current sdk will Do empty command after pipeline finished
if !ok || commandName == "" {
if t.tr == nil || commandName == "" {
return t.Conn.Do(commandName, args...)
}
tr := root.Fork("", "Redis:"+commandName)
tr := t.tr.Fork("", "Redis:"+commandName)
tr.SetTag(_internalTags...)
tr.SetTag(t.connTags...)
tr.SetTag(trace.TagString(trace.TagDBStatement, statement))
@ -52,16 +53,15 @@ func (t *traceConn) Do(commandName string, args ...interface{}) (reply interface
return
}
func (t *traceConn) Send(commandName string, args ...interface{}) error {
func (t *traceConn) Send(commandName string, args ...interface{}) (err error) {
statement := getStatement(commandName, args...)
defer slowLog(statement, time.Now())
defer t.slowLog(statement, time.Now())
t.pending++
root, ok := trace.FromContext(t.ctx)
if !ok {
if t.tr == nil {
return t.Conn.Send(commandName, args...)
}
if t.tr == nil {
t.tr = root.Fork("", "Redis:Pipeline")
if t.pending == 1 {
t.tr = t.tr.Fork("", "Redis:Pipeline")
t.tr.SetTag(_internalTags...)
t.tr.SetTag(t.connTags...)
}
@ -69,8 +69,7 @@ func (t *traceConn) Send(commandName string, args ...interface{}) error {
trace.Log(trace.LogEvent, "Send"),
trace.Log("db.statement", statement),
)
err := t.Conn.Send(commandName, args...)
if err != nil {
if err = t.Conn.Send(commandName, args...); err != nil {
t.tr.SetTag(trace.TagBool(trace.TagError, true))
t.tr.SetLog(
trace.Log(trace.LogEvent, "Send Fail"),
@ -81,7 +80,7 @@ func (t *traceConn) Send(commandName string, args ...interface{}) error {
}
func (t *traceConn) Flush() error {
defer slowLog("Flush", time.Now())
defer t.slowLog("Flush", time.Now())
if t.tr == nil {
return t.Conn.Flush()
}
@ -98,7 +97,7 @@ func (t *traceConn) Flush() error {
}
func (t *traceConn) Receive() (reply interface{}, err error) {
defer slowLog("Receive", time.Now())
defer t.slowLog("Receive", time.Now())
if t.tr == nil {
return t.Conn.Receive()
}
@ -122,13 +121,16 @@ func (t *traceConn) Receive() (reply interface{}, err error) {
}
func (t *traceConn) WithContext(ctx context.Context) Conn {
t.ctx = ctx
t.Conn = t.Conn.WithContext(ctx)
if root, ok := trace.FromContext(ctx); ok {
t.tr = root
}
return t
}
func slowLog(statement string, now time.Time) {
func (t *traceConn) slowLog(statement string, now time.Time) {
du := time.Since(now)
if du > _slowLogDuration {
if du > t.slowLogThreshold {
log.Warn("%s slow log statement: %s time: %v", _tracePeerService, statement, du)
}
}

192
pkg/cache/redis/trace_test.go vendored Normal file
View File

@ -0,0 +1,192 @@
package redis
import (
"context"
"fmt"
"testing"
"time"
"github.com/bilibili/kratos/pkg/net/trace"
"github.com/stretchr/testify/assert"
)
const testTraceSlowLogThreshold = time.Duration(250 * time.Millisecond)
type mockTrace struct {
tags []trace.Tag
logs []trace.LogField
perr *error
operationName string
finished bool
}
func (m *mockTrace) Fork(serviceName string, operationName string) trace.Trace {
m.operationName = operationName
return m
}
func (m *mockTrace) Follow(serviceName string, operationName string) trace.Trace {
panic("not implemented")
}
func (m *mockTrace) Finish(err *error) {
m.perr = err
m.finished = true
}
func (m *mockTrace) SetTag(tags ...trace.Tag) trace.Trace {
m.tags = append(m.tags, tags...)
return m
}
func (m *mockTrace) SetLog(logs ...trace.LogField) trace.Trace {
m.logs = append(m.logs, logs...)
return m
}
func (m *mockTrace) Visit(fn func(k, v string)) {}
func (m *mockTrace) SetTitle(title string) {}
func (m *mockTrace) TraceID() string { return "" }
type mockConn struct{}
func (c *mockConn) Close() error { return nil }
func (c *mockConn) Err() error { return nil }
func (c *mockConn) Do(commandName string, args ...interface{}) (reply interface{}, err error) {
return nil, nil
}
func (c *mockConn) Send(commandName string, args ...interface{}) error { return nil }
func (c *mockConn) Flush() error { return nil }
func (c *mockConn) Receive() (reply interface{}, err error) { return nil, nil }
func (c *mockConn) WithContext(context.Context) Conn { return c }
func TestTraceDo(t *testing.T) {
tr := &mockTrace{}
ctx := trace.NewContext(context.Background(), tr)
tc := &traceConn{Conn: &mockConn{}, slowLogThreshold: testTraceSlowLogThreshold}
conn := tc.WithContext(ctx)
conn.Do("GET", "test")
assert.Equal(t, "Redis:GET", tr.operationName)
assert.NotEmpty(t, tr.tags)
assert.True(t, tr.finished)
}
func TestTraceDoErr(t *testing.T) {
tr := &mockTrace{}
ctx := trace.NewContext(context.Background(), tr)
tc := &traceConn{Conn: MockErr{Error: fmt.Errorf("hhhhhhh")},
slowLogThreshold: testTraceSlowLogThreshold}
conn := tc.WithContext(ctx)
conn.Do("GET", "test")
assert.Equal(t, "Redis:GET", tr.operationName)
assert.True(t, tr.finished)
assert.NotNil(t, *tr.perr)
}
func TestTracePipeline(t *testing.T) {
tr := &mockTrace{}
ctx := trace.NewContext(context.Background(), tr)
tc := &traceConn{Conn: &mockConn{}, slowLogThreshold: testTraceSlowLogThreshold}
conn := tc.WithContext(ctx)
N := 2
for i := 0; i < N; i++ {
conn.Send("GET", "hello, world")
}
conn.Flush()
for i := 0; i < N; i++ {
conn.Receive()
}
assert.Equal(t, "Redis:Pipeline", tr.operationName)
assert.NotEmpty(t, tr.tags)
assert.NotEmpty(t, tr.logs)
assert.True(t, tr.finished)
}
func TestTracePipelineErr(t *testing.T) {
tr := &mockTrace{}
ctx := trace.NewContext(context.Background(), tr)
tc := &traceConn{Conn: MockErr{Error: fmt.Errorf("hahah")},
slowLogThreshold: testTraceSlowLogThreshold}
conn := tc.WithContext(ctx)
N := 2
for i := 0; i < N; i++ {
conn.Send("GET", "hello, world")
}
conn.Flush()
for i := 0; i < N; i++ {
conn.Receive()
}
assert.Equal(t, "Redis:Pipeline", tr.operationName)
assert.NotEmpty(t, tr.tags)
assert.NotEmpty(t, tr.logs)
assert.True(t, tr.finished)
var isError bool
for _, tag := range tr.tags {
if tag.Key == "error" {
isError = true
}
}
assert.True(t, isError)
}
func TestSendStatement(t *testing.T) {
tr := &mockTrace{}
ctx := trace.NewContext(context.Background(), tr)
tc := &traceConn{Conn: MockErr{Error: fmt.Errorf("hahah")},
slowLogThreshold: testTraceSlowLogThreshold}
conn := tc.WithContext(ctx)
conn.Send("SET", "hello", "test")
conn.Flush()
conn.Receive()
assert.Equal(t, "Redis:Pipeline", tr.operationName)
assert.NotEmpty(t, tr.tags)
assert.NotEmpty(t, tr.logs)
assert.Equal(t, "event", tr.logs[0].Key)
assert.Equal(t, "Send", tr.logs[0].Value)
assert.Equal(t, "db.statement", tr.logs[1].Key)
assert.Equal(t, "SET hello", tr.logs[1].Value)
assert.True(t, tr.finished)
var isError bool
for _, tag := range tr.tags {
if tag.Key == "error" {
isError = true
}
}
assert.True(t, isError)
}
func TestDoStatement(t *testing.T) {
tr := &mockTrace{}
ctx := trace.NewContext(context.Background(), tr)
tc := &traceConn{Conn: MockErr{Error: fmt.Errorf("hahah")},
slowLogThreshold: testTraceSlowLogThreshold}
conn := tc.WithContext(ctx)
conn.Do("SET", "hello", "test")
assert.Equal(t, "Redis:SET", tr.operationName)
assert.Equal(t, "SET hello", tr.tags[len(tr.tags)-1].Value)
assert.True(t, tr.finished)
}
func BenchmarkTraceConn(b *testing.B) {
for i := 0; i < b.N; i++ {
c, err := DialDefaultServer()
if err != nil {
b.Fatal(err)
}
t := &traceConn{
Conn: c,
connTags: []trace.Tag{trace.TagString(trace.TagPeerAddress, "abc")},
slowLogThreshold: time.Duration(1 * time.Second),
}
c2 := t.WithContext(context.TODO())
if _, err := c2.Do("PING"); err != nil {
b.Fatal(err)
}
c2.Close()
}
}

17
pkg/cache/redis/util.go vendored Normal file
View File

@ -0,0 +1,17 @@
package redis
import (
"context"
"time"
)
func shrinkDeadline(ctx context.Context, timeout time.Duration) time.Time {
var timeoutTime = time.Now().Add(timeout)
if ctx == nil {
return timeoutTime
}
if deadline, ok := ctx.Deadline(); ok && timeoutTime.After(deadline) {
return deadline
}
return timeoutTime
}

37
pkg/cache/redis/util_test.go vendored Normal file
View File

@ -0,0 +1,37 @@
package redis
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
)
func TestShrinkDeadline(t *testing.T) {
t.Run("test not deadline", func(t *testing.T) {
timeout := time.Second
timeoutTime := time.Now().Add(timeout)
tm := shrinkDeadline(context.Background(), timeout)
assert.True(t, tm.After(timeoutTime))
})
t.Run("test big deadline", func(t *testing.T) {
timeout := time.Second
timeoutTime := time.Now().Add(timeout)
deadlineTime := time.Now().Add(2 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
tm := shrinkDeadline(ctx, timeout)
assert.True(t, tm.After(timeoutTime) && tm.Before(deadlineTime))
})
t.Run("test small deadline", func(t *testing.T) {
timeout := time.Second
deadlineTime := time.Now().Add(500 * time.Millisecond)
ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
defer cancel()
tm := shrinkDeadline(ctx, timeout)
assert.True(t, tm.After(deadlineTime) && tm.Before(time.Now().Add(timeout)))
})
}

View File

@ -13,7 +13,7 @@ type Mock struct {
}
// NewMock new a config mock client.
func NewMock(vs map[string]string) *Mock {
func NewMock(vs map[string]string) Client {
values := make(map[string]*Value, len(vs))
for k, v := range vs {
values[k] = &Value{val: v, raw: v}

View File

@ -137,8 +137,8 @@ func TestBalancerDone(t *testing.T) {
latency, count := picker.(*wrrPicker).subConns[0].latencySummary()
expectLatency := float64(100*time.Millisecond) / 1e5
if !(expectLatency < latency && latency < (expectLatency+100)) {
t.Fatalf("latency is less than 100ms or greter than 100ms, %f", latency)
if latency < expectLatency || latency > (expectLatency+500) {
t.Fatalf("latency is less than 100ms or greater than 150ms, %f", latency)
}
assert.Equal(t, int64(1), count)

View File

@ -293,7 +293,7 @@ func testBreaker(t *testing.T) {
}
defer conn.Close()
c := pb.NewGreeterClient(conn)
for i := 0; i < 50; i++ {
for i := 0; i < 1000; i++ {
_, err := c.SayHello(context.Background(), &pb.HelloRequest{Name: "breaker_test"})
if err != nil {
if ecode.EqualError(ecode.ServiceUnavailable, err) {

View File

@ -21,8 +21,8 @@ func TestProbabilitySampling(t *testing.T) {
count++
}
}
if count < 60 || count > 120 {
t.Errorf("expect count between 60~120 get %d", count)
if count < 60 || count > 150 {
t.Errorf("expect count between 60~150 get %d", count)
}
})
}

View File

@ -51,8 +51,8 @@ func Setup() (err error) {
return
}
defer func() {
if err != err {
go Teardown()
if err != nil {
Teardown()
}
}()
if _, err = getServices(); err != nil {

View File

@ -105,7 +105,7 @@ example:
testcli -f ../../test/docker-compose.yaml run go test -v ./
```
#### Method 2. Import with Kratos pkg
- Step1. 在 Dao|Service 层中的 TestMain 单测主入口中,import "go-common/library/testing/lich" 引入testcli工具的go库版本。
- Step1. 在 Dao|Service 层中的 TestMain 单测主入口中,import "github.com/bilibili/kratos/pkg/testing/lich" 引入testcli工具的go库版本。
- Step2. 使用 flag.Set("f", "../../test/docker-compose.yaml") 指定 docker-compose.yaml 文件的路径。
- Step3. 在 flag.Parse() 后即可使用 lich.Setup() 安装依赖&初始化数据(注意测试用例执行结束后 lich.Teardown() 回收下~)
- Step4. 运行 `go test -v ./ `看看效果吧~