Show router status in SHOW ROUTERS command response

coordinator/provider/coordinator.go

@@ -204,8 +204,20 @@ func (qc *qdbCoordinator) watchRouters(ctx context.Context) {
 					if _, err := rrClient.OpenRouter(ctx, &routerproto.OpenRouterRequest{}); err != nil {
 						return err
 					}
+
+					/* Mark router as opened in qdb */
+					err := qc.db.OpenRouter(ctx, internalR.ID)
+					if err != nil {
+						return err
+					}
 				case routerproto.RouterStatus_OPENED:
 					spqrlog.Zero.Debug().Msg("router is opened")
+
+					/* Mark router as opened in qdb */
+					err := qc.db.OpenRouter(ctx, internalR.ID)
+					if err != nil {
+						return err
+					}
 					// TODO: consistency checks
 				}
 				return nil
@@ -314,6 +326,7 @@ func (qc *qdbCoordinator) ListRouters(ctx context.Context) ([]*topology.Router,
 		retRouters = append(retRouters, &topology.Router{
 			ID:      v.ID,
 			Address: v.Address,
+			Status:  string(v.State),
 		})
 	}
 

pkg/clientinteractor/interactor.go

@@ -616,13 +616,13 @@ func (pi *PSQLInteractor) MoveKeyRange(_ context.Context, move *kr.MoveKeyRange)
 }
 
 func (pi *PSQLInteractor) Routers(resp []*topology.Router) error {
-	if err := pi.WriteHeader("show routers"); err != nil {
+	if err := pi.WriteHeader("show routers", "status"); err != nil {
 		spqrlog.Zero.Error().Err(err).Msg("")
 		return err
 	}
 
 	for _, msg := range resp {
-		if err := pi.WriteDataRow(fmt.Sprintf("router %s-%s", msg.ID, msg.Address)); err != nil {
+		if err := pi.WriteDataRow(fmt.Sprintf("router %s-%s", msg.ID, msg.Address), msg.Status); err != nil {
 			spqrlog.Zero.Error().Err(err).Msg("")
 			return err
 		}

pkg/models/topology/routers.go

@@ -10,6 +10,7 @@ import (
 type Router struct {
 	ID      string
 	Address string
+	Status  string /* open/close */
 }
 
 type RouterMgr interface {

qdb/etcdqdb.go

@@ -587,6 +587,59 @@ func (q *EtcdQDB) DeleteRouter(ctx context.Context, id string) error {
 	return nil
 }
 
+func (q *EtcdQDB) OpenRouter(ctx context.Context, id string) error {
+	spqrlog.Zero.Debug().
+		Str("id", id).
+		Msg("etcdqdb: open router")
+	getResp, err := q.cli.Get(ctx, routerNodePath(id))
+	if err != nil {
+		return err
+	}
+	if len(getResp.Kvs) == 0 {
+		return fmt.Errorf("router with id %s does not exists", id)
+	}
+
+	var routers []*Router
+	for _, e := range getResp.Kvs {
+		var st Router
+		if err := json.Unmarshal(e.Value, &st); err != nil {
+			return err
+		}
+		// TODO: create routers in qdb properly
+		routers = append(routers, &st)
+	}
+
+	/*  */
+
+	if len(routers) != 1 {
+		return fmt.Errorf("sync  failed: more taht one router with id %s", id)
+	}
+
+	if routers[0].State == OPENED {
+		spqrlog.Zero.Debug().
+			Msg("etcdqdb: router already opened in qdb")
+		/* notinh to do */
+		return nil
+	}
+
+	routers[0].State = OPENED
+
+	bts, err := json.Marshal(routers[0])
+	if err != nil {
+		return err
+	}
+	resp, err := q.cli.Put(ctx, routerNodePath(routers[0].ID), string(bts))
+	if err != nil {
+		return err
+	}
+
+	spqrlog.Zero.Debug().
+		Interface("response", resp).
+		Msg("etcdqdb: put router to qdb")
+
+	return nil
+}
+
 func (q *EtcdQDB) ListRouters(ctx context.Context) ([]*Router, error) {
 	spqrlog.Zero.Debug().Msg("etcdqdb: list routers")
 	resp, err := q.cli.Get(ctx, routersNamespace, clientv3.WithPrefix())

qdb/memqdb.go

@@ -378,6 +378,16 @@ func (q *MemQDB) DeleteRouter(ctx context.Context, id string) error {
 	return ExecuteCommands(q.DumpState, NewDeleteCommand(q.Routers, id))
 }
 
+func (q *MemQDB) OpenRouter(ctx context.Context, id string) error {
+	spqrlog.Zero.Debug().Str("router", id).Msg("memqdb: open router")
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	q.Routers[id].State = OPENED
+
+	return ExecuteCommands(q.DumpState, NewUpdateCommand(q.Routers, id, q.Routers[id]))
+}
+
 func (q *MemQDB) ListRouters(ctx context.Context) ([]*Router, error) {
 	spqrlog.Zero.Debug().Msg("memqdb: list routers")
 	q.mu.Lock()

qdb/qdb.go

@@ -26,6 +26,7 @@ type QDB interface {
 	ShareKeyRange(id string) error
 
 	AddRouter(ctx context.Context, r *Router) error
+	OpenRouter(ctx context.Context, rID string) error
 	DeleteRouter(ctx context.Context, rID string) error
 	ListRouters(ctx context.Context) ([]*Router, error)
 	LockRouter(ctx context.Context, id string) error

router/instance.go

@@ -195,6 +195,7 @@ func (r *InstanceImpl) Run(ctx context.Context, listener net.Listener) error {
 		select {
 		case conn := <-cChan:
 			if !r.Initialized() {
+				/* do not accept client connections on un-initialized router */
 				_ = conn.Close()
 			} else {
 				go func() {

test/regress/tests/console/expected/show_routers.out

@@ -6,8 +6,8 @@
 https://github.com/pg-sharding/spqr/tree/master/docs
 
 SHOW routers;
- show routers  
----------------
- router local-
+ show routers  | status 
+---------------+--------
+ router local- | 
 (1 row)