Read node port from config

Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com>

configs/milvus.yaml

@@ -13,7 +13,6 @@
 nodeID: # will be deprecated after v0.2
   proxyIDList: [0]
   queryNodeIDList: [1, 2]
-  writeNodeIDList: [3] # GOOSE TODO: remove this
   dataNodeIDList: [3] 
 
 etcd:
@@ -49,7 +48,7 @@ master:
 
 proxyNode:
   address: localhost
-  port: 19530
+  port: 21122
 
 queryService:
   address: localhost
@@ -61,15 +60,15 @@ proxyService:
 
 queryNode:
   gracefulTime: 5000 #ms
-
-indexBuilder:
-    address: localhost
-    port: 31000
+  port: 21123
 
 indexServer:
   address: localhost
   port: 31000
 
+indexNode:
+  port: 21121
+
 dataService:
   address: localhost
   port: 13333

internal/distributed/datanode/param_table.go

@@ -26,7 +26,7 @@ func (pt *ParamTable) Init() {
 		pt.BaseTable.Init()
 		pt.initMasterAddress()
 		pt.initDataServiceAddress()
-		pt.initPort() // todo random generate
+		pt.initPort()
 	})
 }
 

internal/distributed/datanode/service.go

@@ -117,7 +117,9 @@ func (s *Server) Stop() error {
 
 func (s *Server) init() error {
 	Params.Init()
-	Params.Port = funcutil.GetAvailablePort()
+	if !funcutil.CheckPortAvailable(Params.Port) {
+		Params.Port = funcutil.GetAvailablePort()
+	}
 	Params.LoadFromEnv()
 	Params.LoadFromArgs()
 

internal/distributed/indexnode/paramtable.go

@@ -48,7 +48,7 @@ func (pt *ParamTable) LoadFromEnv() {
 }
 
 func (pt *ParamTable) initParams() {
-
+	pt.initPort()
 	pt.initIndexServerAddress()
 }
 
@@ -77,3 +77,8 @@ func (pt *ParamTable) initIndexServerAddress() {
 
 	pt.IndexServerAddress = addr + ":" + port
 }
+
+func (pt *ParamTable) initPort() {
+	port := pt.ParseInt("indexNode.port")
+	pt.Port = port
+}

internal/distributed/indexnode/service.go

@@ -68,9 +68,12 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 func (s *Server) init() error {
 	var err error
 	Params.Init()
+	if !funcutil.CheckPortAvailable(Params.Port) {
+		Params.Port = funcutil.GetAvailablePort()
+	}
 	Params.LoadFromEnv()
 	Params.LoadFromArgs()
-	Params.Port = funcutil.GetAvailablePort()
+
 	Params.Address = Params.IP + ":" + strconv.FormatInt(int64(Params.Port), 10)
 
 	defer func() {

internal/distributed/proxynode/paramtable.go

@@ -78,7 +78,7 @@ func (pt *ParamTable) LoadFromEnv() {
 
 func (pt *ParamTable) initParams() {
 	pt.initPoxyServicePort()
-
+	pt.initPort()
 	pt.initProxyServiceAddress()
 	pt.initMasterAddress()
 	pt.initIndexServerAddress()
@@ -182,3 +182,8 @@ func (pt *ParamTable) initQueryServiceAddress() {
 	}
 	pt.QueryServiceAddress = addr + ":" + port
 }
+
+func (pt *ParamTable) initPort() {
+	port := pt.ParseInt("proxyNode.port")
+	pt.Port = port
+}

internal/distributed/proxynode/service.go

@@ -118,11 +118,12 @@ func (s *Server) Run() error {
 func (s *Server) init() error {
 	var err error
 	Params.Init()
-
+	if !funcutil.CheckPortAvailable(Params.Port) {
+		Params.Port = funcutil.GetAvailablePort()
+	}
 	Params.LoadFromEnv()
 	Params.LoadFromArgs()
 
-	Params.Port = funcutil.GetAvailablePort()
 	Params.Address = Params.IP + ":" + strconv.FormatInt(int64(Params.Port), 10)
 
 	log.Println("proxy host: ", Params.IP)

internal/distributed/querynode/param_table.go

@@ -28,6 +28,7 @@ type ParamTable struct {
 func (pt *ParamTable) Init() {
 	once.Do(func() {
 		pt.BaseTable.Init()
+		pt.initPort()
 		pt.initMasterAddress()
 		pt.initIndexServiceAddress()
 		pt.initDataServiceAddress()
@@ -93,3 +94,8 @@ func (pt *ParamTable) initQueryServiceAddress() {
 	}
 	pt.QueryServiceAddress = ret
 }
+
+func (pt *ParamTable) initPort() {
+	port := pt.ParseInt("queryNode.port")
+	pt.QueryNodePort = port
+}

internal/distributed/querynode/service.go

@@ -57,7 +57,9 @@ func NewServer(ctx context.Context, factory msgstream.Factory) (*Server, error)
 func (s *Server) init() error {
 
 	Params.Init()
-	Params.QueryNodePort = funcutil.GetAvailablePort()
+	if !funcutil.CheckPortAvailable(Params.QueryNodePort) {
+		Params.QueryNodePort = funcutil.GetAvailablePort()
+	}
 	Params.LoadFromEnv()
 	Params.LoadFromArgs()
 

internal/indexservice/paramtable.go

@@ -56,7 +56,7 @@ func (pt *ParamTable) initAddress() {
 	hostName, _ := net.LookupHost(addr)
 	if len(hostName) <= 0 {
 		if ip := net.ParseIP(addr); ip == nil {
-			panic("invalid ip indexBuilder.address")
+			panic("invalid ip indexServer.address")
 		}
 	}
 

internal/masterservice/master_service.go

@@ -108,7 +108,7 @@ type Core struct {
 		Segment Flush Watcher, monitor if segment has flushed into disk
 
 		IndexService Interface:
-		indexBuilder Sch, tell index service to build index
+		IndexService Sch, tell index service to build index
 	*/
 
 	MetaTable *metaTable

internal/util/funcutil/func.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"strconv"
 	"time"
 
 	"github.com/go-basic/ipv4"
@@ -23,6 +24,15 @@ func CheckGrpcReady(ctx context.Context, targetCh chan error) {
 	}
 }
 
+func CheckPortAvailable(port int) bool {
+	addr := ":" + strconv.Itoa(port)
+	listener, err := net.Listen("tcp", addr)
+	if listener != nil {
+		listener.Close()
+	}
+	return err == nil
+}
+
 func GetAvailablePort() int {
 	listener, err := net.Listen("tcp", ":0")
 	if err != nil {

internal/util/paramtable/paramtable.go

@@ -164,11 +164,11 @@ func (gp *BaseTable) tryloadFromEnv() {
 
 	indexBuilderAddress := os.Getenv("INDEX_SERVICE_ADDRESS")
 	if indexBuilderAddress == "" {
-		indexBuilderHost, err := gp.Load("indexBuilder.address")
+		indexBuilderHost, err := gp.Load("indexServer.address")
 		if err != nil {
 			panic(err)
 		}
-		port, err := gp.Load("indexBuilder.port")
+		port, err := gp.Load("indexServer.port")
 		if err != nil {
 			panic(err)
 		}
@@ -331,7 +331,7 @@ func (gp *BaseTable) ParseInt(key string) int {
 
 // GOOSE TODO: remove writenode
 func (gp *BaseTable) WriteNodeIDList() []UniqueID {
-	proxyIDStr, err := gp.Load("nodeID.writeNodeIDList")
+	proxyIDStr, err := gp.Load("nodeID.dataNodeIDList")
 	if err != nil {
 		panic(err)
 	}