admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-02-02 01:06:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix proxy panicking when QueryCoord offline (#23257)

Browse Source 
Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2023-04-08 09:52:28 +08:00 committed by GitHub
parent 8c55d1de1a
commit dfee10624a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 58 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
internal/proxy/impl.go
View File

@ -4041,15 +4041,18 @@ func (node *Proxy) GetReplicas(ctx context.Context, req *milvuspb.GetReplicasReq
req.CollectionID, _ = globalMetaCache.GetCollectionID(ctx, req.GetCollectionName())
}
resp, err := node.queryCoord.GetReplicas(ctx, req)
r, err := node.queryCoord.GetReplicas(ctx, req)
if err != nil {
log.Error("Failed to get replicas from Query Coordinator", zap.Error(err))
resp.Status.ErrorCode = commonpb.ErrorCode_UnexpectedError
resp.Status.Reason = err.Error()
log.Warn("Failed to get replicas from Query Coordinator",
zap.Error(err))
resp.Status = &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UnexpectedError,
Reason: err.Error(),
}
return resp, nil
}
log.Info("received get replicas response", zap.Any("resp", resp), zap.Error(err))
return resp, nil
log.Debug("received get replicas response", zap.String("resp", r.String()))
return r, nil
}
// GetCompactionState gets the compaction state of multiple segments

49
internal/proxy/impl_test.go
View File

@ -30,6 +30,7 @@ import (
"github.com/milvus-io/milvus/internal/mocks"
"github.com/milvus-io/milvus/internal/proto/datapb"
"github.com/milvus-io/milvus/internal/proto/proxypb"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/dependency"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/internal/util/typeutil"
@ -473,3 +474,51 @@ func TestProxy_GetFlushAllState(t *testing.T) {
assert.Equal(t, resp.GetStatus().GetErrorCode(), commonpb.ErrorCode_UnexpectedError)
})
}
func TestProxy_GetReplicas(t *testing.T) {
factory := dependency.NewDefaultFactory(true)
ctx := context.Background()
node, err := NewProxy(ctx, factory)
assert.NoError(t, err)
node.stateCode.Store(commonpb.StateCode_Healthy)
node.tsoAllocator = &timestampAllocator{
tso: newMockTimestampAllocatorInterface(),
}
mockQC := types.NewMockQueryCoord(t)
mockRC := mocks.NewRootCoord(t)
node.queryCoord = mockQC
node.rootCoord = mockRC
// set expectations
successStatus := &commonpb.Status{ErrorCode: commonpb.ErrorCode_Success}
t.Run("success", func(t *testing.T) {
mockQC.EXPECT().GetReplicas(mock.Anything, mock.AnythingOfType("*milvuspb.GetReplicasRequest")).Return(&milvuspb.GetReplicasResponse{Status: successStatus}, nil)
resp, err := node.GetReplicas(ctx, &milvuspb.GetReplicasRequest{
CollectionID: 1000,
})
assert.NoError(t, err)
assert.Equal(t, resp.GetStatus().GetErrorCode(), commonpb.ErrorCode_Success)
})
t.Run("proxy_not_healthy", func(t *testing.T) {
node.stateCode.Store(commonpb.StateCode_Abnormal)
resp, err := node.GetReplicas(ctx, &milvuspb.GetReplicasRequest{
CollectionID: 1000,
})
assert.NoError(t, err)
assert.Equal(t, resp.GetStatus().GetErrorCode(), commonpb.ErrorCode_UnexpectedError)
node.stateCode.Store(commonpb.StateCode_Healthy)
})
t.Run("QueryCoordClient_returnsError", func(t *testing.T) {
mockQC.ExpectedCalls = nil
mockQC.EXPECT().GetReplicas(mock.Anything, mock.AnythingOfType("*milvuspb.GetReplicasRequest")).Return(nil, errors.New("mocked"))
resp, err := node.GetReplicas(ctx, &milvuspb.GetReplicasRequest{
CollectionID: 1000,
})
assert.NoError(t, err)
assert.Equal(t, resp.GetStatus().GetErrorCode(), commonpb.ErrorCode_UnexpectedError)
})
}