add node to faulty list on rpc error

jd-opensource · Oct 26, 2023 · a65a868 · a65a868
1 parent 5e7c421
commit a65a868
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 19 deletions.
diff --git a/README.md b/README.md
@@ -7,6 +7,10 @@
 
 Vearch is a scalable distributed system for efficient similarity search of deep learning vectors. 
 
+## Document
+
+* [English](https://vearch.readthedocs.io/en/latest) | [中文](https://vearch.readthedocs.io/zh_CN/latest)
+
 ## Quick start
 
 ### Install Vearch
@@ -47,10 +51,6 @@ Vearch Python SDK enables vearch to use locally. Vearch python sdk can be instal
 ### PythonSDKAPI
 * [APIPythonSDK.md](docs/APIPythonSDK.md)
 
-## Document
-
-* [English](https://vearch.readthedocs.io/en/latest) | [中文](https://vearch.readthedocs.io/zh_CN/latest)
-
 ## Components
 
 <details>

diff --git a/client/client.go b/client/client.go
@@ -482,7 +482,7 @@ func (r *routerRequest) searchFromPartition(ctx context.Context, partitionID ent
 	}
 
 	clientType := pd.SearchRequest.Head.ClientType
-	//ensure node is alive
+	// ensure node is alive
 	servers := r.client.Master().Cache().serverCache
 
 	rpcEnd, rpcStart := time.Now(), time.Now()
@@ -590,10 +590,11 @@ func (r *routerRequest) searchFromPartition(ctx context.Context, partitionID ent
 		}
 
 		if strings.Contains(err.Error(), "connect: connection refused") {
-			r.client.PS().AddFaulty(nodeID)
+			r.client.PS().AddFaulty(nodeID, time.Second*30)
 			nodeID = GetNodeIdsByClientType(clientType, partition, servers)
 		} else {
 			log.Error("rpc err [%v], nodeID %v", err, nodeID)
+			r.client.PS().AddFaulty(nodeID, time.Second*5)
 			break
 		}
 	}

diff --git a/client/ps.go b/client/ps.go
@@ -58,15 +58,15 @@ const (
 	MSearchNewHandler    = "MSearchNewHandler"
 	StreamSearchHandler  = "StreamSearchHandler"
 
-	GetDocHandler               = "GetDocHandler"
-	GetDocsHandler              = "GetDocsHandler"
-	GetDocsByPartitionHandler   = "GetDocsByPartitionHandler"
-	CreateDocHandler            = "CreateDocHandler"
-	DeleteDocsHandler           = "DeleteDocsHandler"
-	ReplaceDocHandler           = "ReplaceDocHandler"
-	BatchHandler                = "BatchHandler"
-	ForceMergeHandler           = "ForceMergeHandler"
-	FlushHandler                = "FlushHandler"
+	GetDocHandler             = "GetDocHandler"
+	GetDocsHandler            = "GetDocsHandler"
+	GetDocsByPartitionHandler = "GetDocsByPartitionHandler"
+	CreateDocHandler          = "CreateDocHandler"
+	DeleteDocsHandler         = "DeleteDocsHandler"
+	ReplaceDocHandler         = "ReplaceDocHandler"
+	BatchHandler              = "BatchHandler"
+	ForceMergeHandler         = "ForceMergeHandler"
+	FlushHandler              = "FlushHandler"
 
 	CreatePartitionHandler = "CreatePartitionHandler"
 	DeletePartitionHandler = "DeletePartitionHandler"
@@ -134,11 +134,11 @@ func (ps *psClient) GetOrCreateRPCClient(ctx context.Context, nodeID entity.Node
 }
 
 func (ps *psClient) initFaultylist() {
-	ps.faultyList = cache.New(time.Second*30, time.Second*30)
+	ps.faultyList = cache.New(time.Second*30, time.Second*5)
 }
 
-func (ps *psClient) AddFaulty(nodeID uint64) {
-	ps.faultyList.Set(fmt.Sprint(nodeID), nodeID, 0)
+func (ps *psClient) AddFaulty(nodeID uint64, d time.Duration) {
+	ps.faultyList.Set(fmt.Sprint(nodeID), nodeID, d)
 }
 
 func (ps *psClient) TestFaulty(nodeID uint64) bool {

diff --git a/docs/APILowLevel.md b/docs/APILowLevel.md
@@ -825,7 +825,7 @@ curl -H "content-type: application/json" -XPOST -d'
 * `online_log_level` : "debug", is print debug info 
 * `quick` : default is false, if quick=true it not use precision sorting
 * `vector_value` : default is false, is return vector value
-* `client_type` : search partition type, default is leader, `random` or `no_leader`
+* `client_type` : search partition type, include `random`, `no_leader`, `leader`, default is `random`
 * `l2_sqrt` : default FALSE, don't do sqrt; TRUE, do sqrt
 
 ### delete Document