From d87fe0221afc9c637d9c83b8caa84442c7f0c911 Mon Sep 17 00:00:00 2001 From: apCognixCh Date: Tue, 25 Jun 2024 14:13:19 +0300 Subject: [PATCH 1/2] patch milvus connection --- src/backend/core/storage/milvus.go | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/src/backend/core/storage/milvus.go b/src/backend/core/storage/milvus.go index 1281e914..5891823b 100644 --- a/src/backend/core/storage/milvus.go +++ b/src/backend/core/storage/milvus.go @@ -9,8 +9,10 @@ import ( milvus "github.com/milvus-io/milvus-sdk-go/v2/client" "github.com/milvus-io/milvus-sdk-go/v2/entity" "go.uber.org/fx" + "go.uber.org/zap" "strconv" "strings" + "time" ) const ( @@ -125,13 +127,26 @@ var MilvusModule = fx.Options( ) func NewMilvusClient(cfg *MilvusConfig) (MilvusClient, error) { - client, err := milvus.NewClient(context.Background(), milvus.Config{ + zap.S().Infof("_________________________ cfg %s ", cfg.Address) + + ctx := context.Background() + + ctx, cancel := context.WithTimeout(ctx, 2*time.Second) + + defer cancel() + + client, err := milvus.NewClient(ctx, milvus.Config{ Address: cfg.Address, Username: "root", Password: "sq5/6<$Y4aD`2;Gba'E#", + RetryRateLimit: &milvus.RetryRateLimitOption{ + MaxRetry: 2, + MaxBackoff: 2 * time.Second, + }, }) + zap.S().Debugf("________________________milvus client created") if err != nil { - return nil, err + zap.S().Errorf("connect to milvus error", zap.Error(err)) } return &milvusClient{ client: client, From 65c5f3252499e868ae6abb20787dd77d94d80c09 Mon Sep 17 00:00:00 2001 From: apCognixCh Date: Tue, 25 Jun 2024 14:16:52 +0300 Subject: [PATCH 2/2] patch milvus connection --- src/backend/core/storage/milvus.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/backend/core/storage/milvus.go b/src/backend/core/storage/milvus.go index 5891823b..7a988f23 100644 --- a/src/backend/core/storage/milvus.go +++ b/src/backend/core/storage/milvus.go @@ -57,6 +57,9 @@ type ( ) func (c *milvusClient) Delete(ctx context.Context, collection string, documentIDs ...int64) error { + if c.client == nil { + return fmt.Errorf("milvus is not initialized") + } var docsID []string for _, id := range documentIDs { docsID = append(docsID, strconv.FormatInt(id, 10)) @@ -92,6 +95,9 @@ func (v MilvusConfig) Validate() error { } func (c *milvusClient) Load(ctx context.Context, collection string, vector []float32) ([]*MilvusPayload, error) { + if c.client == nil { + return nil, fmt.Errorf("milvus is not initialized") + } vs := []entity.Vector{entity.FloatVector(vector)} sp, _ := entity.NewIndexFlatSearchParam() result, err := c.client.Search(ctx, collection, []string{}, "", responseColumns, vs, ColumnNameVector, c.MetricType, 10, sp) @@ -147,6 +153,7 @@ func NewMilvusClient(cfg *MilvusConfig) (MilvusClient, error) { zap.S().Debugf("________________________milvus client created") if err != nil { zap.S().Errorf("connect to milvus error", zap.Error(err)) + } return &milvusClient{ client: client, @@ -159,7 +166,9 @@ func (c *milvusClient) Save(ctx context.Context, collection string, payloads ... var ids, documentIDs, chunks []int64 var contents [][]byte var vectors [][]float32 - + if c.client == nil { + return fmt.Errorf("milvus is not initialized") + } for _, payload := range payloads { ids = append(ids, payload.ID) documentIDs = append(documentIDs, payload.DocumentID)