Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
milvus
milvus
提交
764cf1a5
M
milvus
项目概览
milvus
/
milvus
11 个月 前同步成功
通知
261
Star
22476
Fork
2472
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
M
milvus
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
764cf1a5
编写于
12月 02, 2021
作者:
Y
yukun
提交者:
GitHub
12月 02, 2021
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Add rocksmq server close when standalone exit (#12255)
Signed-off-by:
N
fishpenguin
<
kun.yu@zilliz.com
>
上级
26405d13
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
48 addition
and
16 deletion
+48
-16
cmd/roles/roles.go
cmd/roles/roles.go
+16
-0
internal/util/rocksmq/server/rocksmq/global_rmq.go
internal/util/rocksmq/server/rocksmq/global_rmq.go
+21
-16
internal/util/rocksmq/server/rocksmq/global_rmq_test.go
internal/util/rocksmq/server/rocksmq/global_rmq_test.go
+11
-0
未找到文件。
cmd/roles/roles.go
浏览文件 @
764cf1a5
...
...
@@ -28,6 +28,7 @@ import (
"syscall"
"github.com/milvus-io/milvus/internal/util/healthz"
"github.com/milvus-io/milvus/internal/util/rocksmq/server/rocksmq"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"go.uber.org/zap"
...
...
@@ -56,6 +57,15 @@ func newMsgFactory(localMsg bool) msgstream.Factory {
return
msgstream
.
NewPmsFactory
()
}
func
initRocksmq
()
error
{
err
:=
rocksmq
.
InitRocksMQ
()
return
err
}
func
stopRocksmq
()
{
rocksmq
.
CloseRocksMQ
()
}
// MilvusRoles determines to run which components.
type
MilvusRoles
struct
{
EnableRootCoord
bool
`env:"ENABLE_ROOT_COORD"`
...
...
@@ -349,6 +359,12 @@ func (mr *MilvusRoles) Run(localMsg bool, alias string) {
cfg
:=
mr
.
setLogConfigFilename
(
"standalone.log"
)
logutil
.
SetupLogger
(
cfg
)
defer
log
.
Sync
()
err
:=
initRocksmq
()
if
err
!=
nil
{
panic
(
err
)
}
defer
stopRocksmq
()
}
else
{
err
:=
os
.
Setenv
(
metricsinfo
.
DeployModeEnvKey
,
metricsinfo
.
ClusterDeployMode
)
if
err
!=
nil
{
...
...
internal/util/rocksmq/server/rocksmq/global_rmq.go
浏览文件 @
764cf1a5
...
...
@@ -12,6 +12,7 @@
package
rocksmq
import
(
"errors"
"os"
"strconv"
"sync"
...
...
@@ -43,26 +44,33 @@ func InitRmq(rocksdbName string, idAllocator allocator.GIDAllocator) error {
// InitRocksMQ init global rocksmq single instance
func
InitRocksMQ
()
error
{
var
e
rr
error
var
finalE
rr
error
once
.
Do
(
func
()
{
params
.
Init
()
rocksdbName
,
_
:=
params
.
Load
(
"_RocksmqPath"
)
log
.
Debug
(
"RocksmqPath="
+
rocksdbName
)
_
,
err
=
os
.
Stat
(
rocksdbName
)
if
os
.
IsNotExist
(
err
)
{
err
=
os
.
MkdirAll
(
rocksdbName
,
os
.
ModePerm
)
if
err
!=
nil
{
errMsg
:=
"Create dir "
+
rocksdbName
+
" failed"
panic
(
errMsg
)
var
fi
os
.
FileInfo
fi
,
finalErr
=
os
.
Stat
(
rocksdbName
)
if
os
.
IsNotExist
(
finalErr
)
{
finalErr
=
os
.
MkdirAll
(
rocksdbName
,
os
.
ModePerm
)
if
finalErr
!=
nil
{
return
}
}
else
{
if
!
fi
.
IsDir
()
{
errMsg
:=
"can't create a directory because there exists a file with the same name"
finalErr
=
errors
.
New
(
errMsg
)
return
}
}
kvname
:=
rocksdbName
+
"_kv"
rocksdbKV
,
err
:=
rocksdbkv
.
NewRocksdbKV
(
kvname
)
if
err
!=
nil
{
panic
(
err
)
var
rkv
*
rocksdbkv
.
RocksdbKV
rkv
,
finalErr
=
rocksdbkv
.
NewRocksdbKV
(
kvname
)
if
finalErr
!=
nil
{
return
}
idAllocator
:=
allocator
.
NewGlobalIDAllocator
(
"rmq_id"
,
r
ocksdbKV
)
idAllocator
:=
allocator
.
NewGlobalIDAllocator
(
"rmq_id"
,
r
kv
)
_
=
idAllocator
.
Initialize
()
rawRmqPageSize
,
err
:=
params
.
Load
(
"rocksmq.rocksmqPageSize"
)
...
...
@@ -94,12 +102,9 @@ func InitRocksMQ() error {
}
log
.
Debug
(
""
,
zap
.
Any
(
"RocksmqRetentionTimeInMinutes"
,
RocksmqRetentionTimeInMinutes
),
zap
.
Any
(
"RocksmqRetentionSizeInMB"
,
RocksmqRetentionSizeInMB
),
zap
.
Any
(
"RocksmqPageSize"
,
RocksmqPageSize
))
Rmq
,
err
=
NewRocksMQ
(
rocksdbName
,
idAllocator
)
if
err
!=
nil
{
panic
(
err
)
}
Rmq
,
finalErr
=
NewRocksMQ
(
rocksdbName
,
idAllocator
)
})
return
e
rr
return
finalE
rr
}
// CloseRocksMQ is used to close global rocksmq
...
...
internal/util/rocksmq/server/rocksmq/global_rmq_test.go
浏览文件 @
764cf1a5
...
...
@@ -15,6 +15,7 @@ import (
"log"
"os"
"strings"
"sync"
"testing"
"github.com/milvus-io/milvus/internal/allocator"
...
...
@@ -69,3 +70,13 @@ func Test_InitRocksMQ(t *testing.T) {
}
Rmq
.
RegisterConsumer
(
consumer
)
}
func
Test_InitRocksMQError
(
t
*
testing
.
T
)
{
once
=
sync
.
Once
{}
dummyPath
:=
"/tmp/milvus/dummy"
os
.
Create
(
dummyPath
)
os
.
Setenv
(
"ROCKSMQ_PATH"
,
dummyPath
)
defer
os
.
RemoveAll
(
dummyPath
)
err
:=
InitRocksMQ
()
assert
.
Error
(
t
,
err
)
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录