admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-06 17:18:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: dynamic log level for streaming node (#42964)

Browse Source 
issue: #42963

Signed-off-by: chyezh <chyezh@outlook.com>
This commit is contained in:
Zhen Ye 2025-06-26 19:12:50 +08:00 committed by GitHub
parent 5dd1f841d2
commit 3602817c53
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 31 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
internal/streamingnode/server/builder.go
View File

@ -67,13 +67,12 @@ func (b *ServerBuilder) WithMetaKV(kv kv.MetaKv) *ServerBuilder {
// Build builds a streaming node server.
func (b *ServerBuilder) Build() *Server {
resource.Apply(
resource.Init(
resource.OptETCD(b.etcdClient),
resource.OptChunkManager(b.chunkManager),
resource.OptMixCoordClient(b.mixc),
resource.OptStreamingNodeCatalog(streamingnode.NewCataLog(b.kv)),
)
resource.Done()
s := &Server{
session: b.session,
grpcServer: b.grpcServer,

39
internal/streamingnode/server/resource/resource.go
View File

@ -19,9 +19,7 @@ import (
"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
)
var r = &resourceImpl{
logger: log.With(log.FieldModule(typeutil.StreamingNodeRole)),
} // singleton resource instance
var r *resourceImpl // singleton resource instance
// optResourceInit is the option to initialize the resource.
type optResourceInit func(r *resourceImpl)
@ -66,20 +64,27 @@ func Apply(opts ...optResourceInit) {
}
// Done finish all initialization of resources.
func Done() {
r.segmentStatsManager = stats.NewStatsManager()
r.timeTickInspector = tinspector.NewTimeTickSyncInspector()
r.syncMgr = syncmgr.NewSyncManager(r.chunkManager)
r.wbMgr = writebuffer.NewManager(r.syncMgr)
r.wbMgr.Start()
assertNotNil(r.ChunkManager())
assertNotNil(r.TSOAllocator())
assertNotNil(r.MixCoordClient())
assertNotNil(r.StreamingNodeCatalog())
assertNotNil(r.SegmentStatsManager())
assertNotNil(r.TimeTickInspector())
assertNotNil(r.SyncManager())
assertNotNil(r.WriteBufferManager())
func Init(opts ...optResourceInit) {
newR := &resourceImpl{}
for _, opt := range opts {
opt(newR)
}
newR.logger = log.With(log.FieldModule(typeutil.StreamingNodeRole))
newR.segmentStatsManager = stats.NewStatsManager()
newR.timeTickInspector = tinspector.NewTimeTickSyncInspector()
newR.syncMgr = syncmgr.NewSyncManager(newR.chunkManager)
newR.wbMgr = writebuffer.NewManager(newR.syncMgr)
newR.wbMgr.Start()
assertNotNil(newR.ChunkManager())
assertNotNil(newR.TSOAllocator())
assertNotNil(newR.MixCoordClient())
assertNotNil(newR.StreamingNodeCatalog())
assertNotNil(newR.SegmentStatsManager())
assertNotNil(newR.TimeTickInspector())
assertNotNil(newR.SyncManager())
assertNotNil(newR.WriteBufferManager())
r = newR
}
// Release releases the singleton of resources.

13
internal/streamingnode/server/resource/resource_test.go
View File

@ -19,23 +19,18 @@ func TestMain(m *testing.M) {
os.Exit(m.Run())
}
func TestApply(t *testing.T) {
Apply()
Apply(OptETCD(&clientv3.Client{}))
Apply(OptMixCoordClient(syncutil.NewFuture[types.MixCoordClient]()))
func TestInit(t *testing.T) {
assert.Panics(t, func() {
Done()
Init(OptETCD(&clientv3.Client{}),
OptMixCoordClient(syncutil.NewFuture[types.MixCoordClient]()))
})
Apply(
Init(
OptChunkManager(mock_storage.NewMockChunkManager(t)),
OptETCD(&clientv3.Client{}),
OptMixCoordClient(syncutil.NewFuture[types.MixCoordClient]()),
OptStreamingNodeCatalog(mock_metastore.NewMockStreamingNodeCataLog(t)),
)
Done()
assert.NotNil(t, Resource().TSOAllocator())
assert.NotNil(t, Resource().ETCD())
assert.NotNil(t, Resource().MixCoordClient())

2
internal/streamingnode/server/service/handler/consumer/consume_server_test.go
View File

@ -13,6 +13,7 @@ import (
"github.com/milvus-io/milvus/internal/mocks/streamingnode/server/mock_wal"
"github.com/milvus-io/milvus/internal/mocks/streamingnode/server/mock_walmanager"
"github.com/milvus-io/milvus/internal/streamingnode/server/resource"
"github.com/milvus-io/milvus/internal/streamingnode/server/walmanager"
"github.com/milvus-io/milvus/internal/util/streamingutil/service/contextutil"
"github.com/milvus-io/milvus/internal/util/streamingutil/status"
@ -32,6 +33,7 @@ func TestMain(m *testing.M) {
}
func TestCreateConsumeServer(t *testing.T) {
resource.InitForTest(t)
manager := mock_walmanager.NewMockManager(t)
grpcConsumeServer := mock_streamingpb.NewMockStreamingNodeHandlerService_ConsumeServer(t)

2
internal/streamingnode/server/service/handler/producer/produce_server_test.go
View File

@ -16,6 +16,7 @@ import (
"github.com/milvus-io/milvus/internal/mocks/streamingnode/server/mock_wal"
"github.com/milvus-io/milvus/internal/mocks/streamingnode/server/mock_walmanager"
"github.com/milvus-io/milvus/internal/streamingnode/server/resource"
"github.com/milvus-io/milvus/internal/streamingnode/server/wal"
"github.com/milvus-io/milvus/internal/streamingnode/server/walmanager"
"github.com/milvus-io/milvus/internal/util/streamingutil/service/contextutil"
@ -35,6 +36,7 @@ func TestMain(m *testing.M) {
}
func TestCreateProduceServer(t *testing.T) {
resource.InitForTest(t)
manager := mock_walmanager.NewMockManager(t)
grpcProduceServer := mock_streamingpb.NewMockStreamingNodeHandlerService_ProduceServer(t)