mosuka · mosuka · Jul 2, 2019 · Jun 27, 2019 · Jun 30, 2019 · Jun 30, 2019
diff --git a/cmd/blastd/indexer.go b/cmd/blastd/indexer.go
@@ -132,7 +132,7 @@ func startIndexer(c *cli.Context) error {
 		"index_storage_type": indexStorageType,
 	}
 
-	svr, err := indexer.NewServer(managerAddr, clusterId, nodeId, metadata, peerAddr, indexConfig, logger, httpAccessLogger)
+	svr, err := indexer.NewServer(managerAddr, clusterId, nodeId, metadata, peerAddr, indexConfig, logger.Named(nodeId), httpAccessLogger)
 	if err != nil {
 		return err
 	}

diff --git a/cmd/blastd/manager.go b/cmd/blastd/manager.go
@@ -129,7 +129,7 @@ func startManager(c *cli.Context) error {
 		"index_storage_type": indexStorageType,
 	}
 
-	svr, err := manager.NewServer(nodeId, metadata, peerAddr, indexConfig, logger, httpAccessLogger)
+	svr, err := manager.NewServer(nodeId, metadata, peerAddr, indexConfig, logger.Named(nodeId), httpAccessLogger)
 	if err != nil {
 		return err
 	}

diff --git a/grpc/client.go b/grpc/client.go
@@ -18,10 +18,12 @@ import (
 	"context"
 	"errors"
 	"math"
+	"time"
 
 	"github.com/blevesearch/bleve"
 	"github.com/golang/protobuf/ptypes/any"
 	"github.com/golang/protobuf/ptypes/empty"
+	grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
 	blasterrors "github.com/mosuka/blast/errors"
 	"github.com/mosuka/blast/protobuf"
 	"google.golang.org/grpc"
@@ -38,18 +40,30 @@ type Client struct {
 
 func NewContext() (context.Context, context.CancelFunc) {
 	baseCtx := context.TODO()
-	return context.WithCancel(baseCtx)
+	return context.WithTimeout(baseCtx, 60*time.Second)
 }
 
 func NewClient(address string) (*Client, error) {
 	ctx, cancel := NewContext()
 
+	streamRetryOpts := []grpc_retry.CallOption{
+		grpc_retry.Disable(),
+	}
+
+	unaryRetryOpts := []grpc_retry.CallOption{
+		grpc_retry.WithBackoff(grpc_retry.BackoffLinear(100 * time.Millisecond)),
+		grpc_retry.WithCodes(codes.Unavailable),
+		grpc_retry.WithMax(100),
+	}
+
 	dialOpts := []grpc.DialOption{
 		grpc.WithInsecure(),
 		grpc.WithDefaultCallOptions(
 			grpc.MaxCallSendMsgSize(math.MaxInt32),
 			grpc.MaxCallRecvMsgSize(math.MaxInt32),
 		),
+		grpc.WithStreamInterceptor(grpc_retry.StreamClientInterceptor(streamRetryOpts...)),
+		grpc.WithUnaryInterceptor(grpc_retry.UnaryClientInterceptor(unaryRetryOpts...)),
 	}
 
 	conn, err := grpc.DialContext(ctx, address, dialOpts...)
@@ -401,24 +415,40 @@ func (c *Client) DeleteDocument(ids []string, opts ...grpc.CallOption) (int, err
 	return int(resp.Count), nil
 }
 
-func (c *Client) GetIndexConfig(opts ...grpc.CallOption) (*protobuf.GetIndexConfigResponse, error) {
-	conf, err := c.client.GetIndexConfig(c.ctx, &empty.Empty{}, opts...)
+func (c *Client) GetIndexConfig(opts ...grpc.CallOption) (map[string]interface{}, error) {
+	resp, err := c.client.GetIndexConfig(c.ctx, &empty.Empty{}, opts...)
+	if err != nil {
+		st, _ := status.FromError(err)
+
+		return nil, errors.New(st.Message())
+	}
+
+	indexConfigIntr, err := protobuf.MarshalAny(resp.IndexConfig)
 	if err != nil {
 		st, _ := status.FromError(err)
 
 		return nil, errors.New(st.Message())
 	}
+	indexConfig := *indexConfigIntr.(*map[string]interface{})
 
-	return conf, nil
+	return indexConfig, nil
 }
 
-func (c *Client) GetIndexStats(opts ...grpc.CallOption) (*protobuf.GetIndexStatsResponse, error) {
-	stats, err := c.client.GetIndexStats(c.ctx, &empty.Empty{}, opts...)
+func (c *Client) GetIndexStats(opts ...grpc.CallOption) (map[string]interface{}, error) {
+	resp, err := c.client.GetIndexStats(c.ctx, &empty.Empty{}, opts...)
+	if err != nil {
+		st, _ := status.FromError(err)
+
+		return nil, errors.New(st.Message())
+	}
+
+	indexStatsIntr, err := protobuf.MarshalAny(resp.IndexStats)
 	if err != nil {
 		st, _ := status.FromError(err)
 
 		return nil, errors.New(st.Message())
 	}
+	indexStats := *indexStatsIntr.(*map[string]interface{})
 
-	return stats, nil
+	return indexStats, nil
 }
diff --git a/indexer/grpc_service.go b/indexer/grpc_service.go
@@ -509,6 +509,18 @@ func (s *GRPCService) startUpdateCluster(checkInterval time.Duration) {
 					c <- *resp
 				}
 
+				// update the cluster config to manager if it is a leader
+				if s.managerAddr != "" && s.raftServer.IsLeader() {
+					client, err := s.getManagerClient()
+					if err != nil {
+						s.logger.Error(err.Error())
+					}
+					err = client.SetState(fmt.Sprintf("cluster_config/clusters/%s/nodes", s.clusterId), cluster)
+					if err != nil {
+						s.logger.Error(err.Error())
+					}
+				}
+
 				// keep current cluster
 				s.cluster = cluster
 				s.logger.Debug("cluster", zap.Any("cluster", cluster))
@@ -654,6 +666,7 @@ func (s *GRPCService) SetNode(ctx context.Context, req *protobuf.SetNodeRequest)
 			s.logger.Error(err.Error())
 			return resp, status.Error(codes.Internal, err.Error())
 		}
+
 	}
 
 	return resp, nil

diff --git a/indexer/server.go b/indexer/server.go
@@ -21,7 +21,6 @@ import (
 	"github.com/mosuka/blast/errors"
 	"github.com/mosuka/blast/grpc"
 	"github.com/mosuka/blast/http"
-	"github.com/mosuka/blast/protobuf"
 	"go.uber.org/zap"
 )
 
@@ -164,19 +163,19 @@ func (s *Server) Start() {
 		}
 
 		s.logger.Debug("pull index config from cluster peer", zap.String("address", pc.GetAddress()))
-		resp, err := pc.GetIndexConfig()
+		s.indexConfig, err = pc.GetIndexConfig()
 		if err != nil {
 			s.logger.Fatal(err.Error())
 			return
 		}
 
-		ins, err := protobuf.MarshalAny(resp.IndexConfig)
-		if err != nil {
-			s.logger.Fatal(err.Error())
-			return
-		}
-
-		s.indexConfig = *ins.(*map[string]interface{})
+		//ins, err := protobuf.MarshalAny(resp.IndexConfig)
+		//if err != nil {
+		//	s.logger.Fatal(err.Error())
+		//	return
+		//}
+		//
+		//s.indexConfig = *ins.(*map[string]interface{})
 	}
 
 	var err error