chore: add route http interface & fix some bugs

server/cluster/error.go

@@ -14,4 +14,5 @@ var (
 	ErrShardNotFound        = coderr.NewCodeError(coderr.NotFound, "shard not found")
 	ErrNodeNotFound         = coderr.NewCodeError(coderr.NotFound, "NodeName not found")
 	ErrTableAlreadyExists   = coderr.NewCodeError(coderr.Internal, "table already exists")
+	ErrRouteTable           = coderr.NewCodeError(coderr.Internal, "route table")
 )

server/cluster/table_manager.go

@@ -265,8 +265,10 @@ func (m *TableManagerImpl) loadTable(ctx context.Context) error {
 		for _, table := range tablesResult.Tables {
 			tables, ok := m.schemaTables[table.SchemaID]
 			if !ok {
-				tables.tables = make(map[string]storage.Table, 0)
-				tables.tablesByID = make(map[storage.TableID]storage.Table, 0)
+				tables = &Tables{
+					tables:     make(map[string]storage.Table, 0),
+					tablesByID: make(map[storage.TableID]storage.Table, 0),
+				}
 			}
 
 			tables.tables[table.Name] = table

server/cluster/types.go

@@ -3,7 +3,6 @@
 package cluster
 
 import (
-	"github.com/CeresDB/ceresdbproto/golang/pkg/clusterpb"
 	"github.com/CeresDB/ceresdbproto/golang/pkg/metaservicepb"
 	"github.com/CeresDB/ceresmeta/server/storage"
 )
@@ -25,8 +24,9 @@ type ShardTables struct {
 }
 
 type ShardInfo struct {
-	ID      storage.ShardID
-	Role    storage.ShardRole
+	ID   storage.ShardID
+	Role storage.ShardRole
+	// ShardViewVersion
 	Version uint64
 }
 
@@ -88,7 +88,7 @@ func (n RegisteredNode) IsExpired(now uint64, aliveThreshold uint64) bool {
 func ConvertShardsInfoToPB(shard ShardInfo) *metaservicepb.ShardInfo {
 	return &metaservicepb.ShardInfo{
 		Id:      uint32(shard.ID),
-		Role:    clusterpb.ShardRole(shard.Role),
+		Role:    storage.ConvertShardRoleToPB(shard.Role),
 		Version: shard.Version,
 	}
 }

server/coordinator/scheduler.go

@@ -93,7 +93,7 @@ func (s *Scheduler) checkNode(ctx context.Context, ticker *time.Ticker) {
 				nodeShardsMapping[nodeShard.ShardNode.NodeName] = append(nodeShardsMapping[nodeShard.ShardNode.NodeName], cluster.ShardInfo{
 					ID:      nodeShard.ShardNode.ID,
 					Role:    nodeShard.ShardNode.ShardRole,
-					Version: nodeShards.ClusterTopologyVersion,
+					Version: nodeShard.ShardInfo.Version,
 				})
 			}
 			s.processNodes(ctx, nodes, t, nodeShardsMapping)

server/service/grpc/service.go

@@ -75,6 +75,8 @@ func (s *Service) NodeHeartbeat(ctx context.Context, req *metaservicepb.NodeHear
 		}, ShardInfos: shardInfos,
 	}
 
+	log.Info("registerNode", zap.String("name", req.Info.Endpoint), zap.String("info", req.Info.String()))
+
 	err = s.h.GetClusterManager().RegisterNode(ctx, req.GetHeader().GetClusterName(), registeredNode)
 	if err != nil {
 		return &metaservicepb.NodeHeartbeatResponse{Header: responseHeader(err, "grpc heartbeat")}, nil
@@ -332,7 +334,7 @@ func convertRouteTableResult(routeTablesResult cluster.RouteTablesResult) *metas
 				Endpoint: nodeShard.ShardNode.NodeName,
 				ShardInfo: &metaservicepb.ShardInfo{
 					Id:   uint32(nodeShard.ShardNode.ID),
-					Role: clusterpb.ShardRole(nodeShard.ShardNode.ShardRole),
+					Role: storage.ConvertShardRoleToPB(nodeShard.ShardNode.ShardRole),
 				},
 			})
 		}
@@ -357,7 +359,7 @@ func convertToGetNodesResponse(nodesResult cluster.GetNodeShardsResult) *metaser
 			Endpoint: shardNodeWithVersion.ShardNode.NodeName,
 			ShardInfo: &metaservicepb.ShardInfo{
 				Id:   uint32(shardNodeWithVersion.ShardNode.ID),
-				Role: clusterpb.ShardRole(shardNodeWithVersion.ShardNode.ShardRole),
+				Role: storage.ConvertShardRoleToPB(shardNodeWithVersion.ShardNode.ShardRole),
 			},
 		})
 	}

server/service/http/api.go

@@ -4,6 +4,7 @@ package http
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"io"
 	"net/http"
@@ -43,6 +44,7 @@ func (a *API) NewAPIRouter() *Router {
 	router := New().WithPrefix(apiPrefix).WithInstrumentation(printRequestInsmt)
 
 	router.Post("/transferLeader", a.transferLeader)
+	router.Post("/route", a.route)
 
 	return router
 }
@@ -120,7 +122,7 @@ func (a *API) transferLeader(writer http.ResponseWriter, req *http.Request) {
 	err := json.NewDecoder(req.Body).Decode(&transferLeaderRequest)
 	if err != nil {
 		log.Error("decode request body failed", zap.Error(err))
-		a.respondError(writer, ErrParseRequest, nil)
+		a.respondError(writer, ErrParseRequest, "decode request body failed")
 		return
 	}
 
@@ -132,15 +134,47 @@ func (a *API) transferLeader(writer http.ResponseWriter, req *http.Request) {
 	})
 	if err != nil {
 		log.Error("create transfer leader procedure", zap.Error(err))
-		a.respondError(writer, procedure.ErrCreateProcedure, nil)
+		a.respondError(writer, procedure.ErrCreateProcedure, "create transfer leader procedure")
 		return
 	}
 	err = a.procedureManager.Submit(req.Context(), transferLeaderProcedure)
 	if err != nil {
 		log.Error("submit transfer leader procedure", zap.Error(err))
-		a.respondError(writer, procedure.ErrSubmitProcedure, nil)
+		a.respondError(writer, procedure.ErrSubmitProcedure, "submit transfer leader procedure")
 		return
 	}
 
 	a.respond(writer, "ok")
 }
+
+type RouteRequest struct {
+	ClusterName string   `json:"clusterName"`
+	SchemaName  string   `json:"schemaName"`
+	Tables      []string `json:"table"`
+}
+
+func (a *API) route(writer http.ResponseWriter, req *http.Request) {
+	var routeRequest RouteRequest
+	err := json.NewDecoder(req.Body).Decode(&routeRequest)
+	if err != nil {
+		log.Error("decode request body failed", zap.Error(err))
+		a.respondError(writer, ErrParseRequest, "decode request body failed")
+		return
+	}
+
+	result, err := a.clusterManager.RouteTables(context.Background(), routeRequest.ClusterName, routeRequest.SchemaName, routeRequest.Tables)
+	if err != nil {
+		log.Error("route tables failed", zap.Error(err))
+		a.respondError(writer, cluster.ErrRouteTable, "route tables failed")
+		return
+	}
+
+	resultByte, err := json.Marshal(result)
+	if err != nil {
+		log.Error("route tables result marshal failed", zap.Error(err))
+		a.respondError(writer, ErrParseResponse, "route tables result marshal failed")
+		return
+	}
+
+	a.respond(writer, string(resultByte))
+}

server/storage/types.go

@@ -254,7 +254,7 @@ func convertClusterStatePB(state clusterpb.ClusterTopology_ClusterState) Cluster
 	return ClusterStateEmpty
 }
 
-func convertShardRoleToPB(role ShardRole) clusterpb.ShardRole {
+func ConvertShardRoleToPB(role ShardRole) clusterpb.ShardRole {
 	switch role {
 	case ShardRoleLeader:
 		return clusterpb.ShardRole_LEADER
@@ -267,7 +267,7 @@ func convertShardRoleToPB(role ShardRole) clusterpb.ShardRole {
 func convertShardNodeToPB(shardNode ShardNode) clusterpb.Shard {
 	return clusterpb.Shard{
 		Id:        uint32(shardNode.ID),
-		ShardRole: convertShardRoleToPB(shardNode.ShardRole),
+		ShardRole: ConvertShardRoleToPB(shardNode.ShardRole),
 		Node:      shardNode.NodeName,
 	}
 }