update query

common/types.go

@@ -8,6 +8,7 @@ const (
 	SERVICE_BACKEND
 	SERVICE_NODE_MANAGER
 	SERVICE_SCHEDULER
+	SERVICE_WORKER
 )
 
 var ServiceTypeMap = map[ServiceType]string{
@@ -16,6 +17,7 @@ var ServiceTypeMap = map[ServiceType]string{
 	SERVICE_BACKEND:      "backend",
 	SERVICE_NODE_MANAGER: "node_manager",
 	SERVICE_SCHEDULER:    "scheduler",
+	SERVICE_WORKER:       "worker",
 }
 
 func (s ServiceType) String() string {

query/apigateway.go

 package query
 
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/odysseus/service-registry/common"
+	"github.com/redis/go-redis/v9"
+)
+
 type GatewayInfo struct {
 	Timestamp int64  `redis:"timestamp" json:"timestamp"`
 	Endpoint  string `redis:"endpoint" json:"endpoint"`
@@ -10,20 +18,70 @@ func (g GatewayInfo) TimeStamp() int64 {
 }
 
 type gatewayQuery struct {
+	rdb     *redis.Client
+	service common.ServiceType
 }
 
-func getGatewayQuery() ServiceQuery {
-	return gatewayQuery{}
+func getGatewayQuery(rdb *redis.Client) ServiceQuery {
+	return gatewayQuery{
+		rdb:     rdb,
+		service: common.SERVICE_API_GATEWAY,
+	}
 }
 
 func (g gatewayQuery) ModuleName() string {
-	return "gateway"
+	return g.service.String()
 }
 
 func (g gatewayQuery) List() ([]string, error) {
-	return []string{}, nil
+	all, err := getAllGw(g.rdb, g.service)
+	if err != nil {
+		return nil, err
+	}
+	var res []string
+	for _, v := range all {
+		d, _ := json.Marshal(v)
+		res = append(res, string(d))
+	}
+	return res, nil
 }
 
 func (g gatewayQuery) ServiceInfo(serviceid string) (string, error) {
-	return "", nil
+	info, err := getOneGw(g.rdb, g.service, serviceid)
+	if err != nil {
+		return "", err
+	}
+	d, err := json.Marshal(info)
+	if err != nil {
+		return "", err
+	}
+
+	return string(d), nil
+}
+
+func getAllGw(rdb *redis.Client, serviceType common.ServiceType) ([]GatewayInfo, error) {
+	keys, err := rdb.Keys(context.Background(), common.GetServiceKeyPrefix(serviceType)+"*").Result()
+	if err != nil {
+		return nil, err
+	}
+	var nmInfos []GatewayInfo
+	for _, key := range keys {
+		res := rdb.HGetAll(context.Background(), key)
+		var info GatewayInfo
+		if err := res.Scan(&info); err != nil {
+			continue
+		}
+		nmInfos = append(nmInfos, info)
+	}
+	return nmInfos, nil
+}
+
+func getOneGw(rdb *redis.Client, serviceType common.ServiceType, endpoint string) (GatewayInfo, error) {
+	k := fmt.Sprintf("%s%s", common.GetServiceKeyPrefix(serviceType), endpoint)
+	res := rdb.HGetAll(context.Background(), k)
+	var info GatewayInfo
+	if err := res.Scan(&info); err != nil {
+		return GatewayInfo{}, err
+	}
+	return info, nil
 }

query/backend.go

 package query
 
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/odysseus/service-registry/common"
+	"github.com/redis/go-redis/v9"
+)
+
 type BackendInfo struct {
 	Timestamp int64  `redis:"timestamp" json:"timestamp"`
 	Endpoint  string `redis:"endpoint" json:"endpoint"`
@@ -9,21 +17,72 @@ func (g BackendInfo) TimeStamp() int64 {
 	return g.Timestamp
 }
 
-type backendInfoQuery struct {
+type backendQuery struct {
+	rdb     *redis.Client
+	service common.ServiceType
+}
+
+func getBackendQuery(rdb *redis.Client) ServiceQuery {
+	return backendQuery{
+		rdb:     rdb,
+		service: common.SERVICE_BACKEND,
+	}
 }
 
-func getBackendInfoQuery() ServiceQuery {
-	return backendInfoQuery{}
+func (g backendQuery) ModuleName() string {
+	return g.service.String()
 }
 
-func (g backendInfoQuery) ModuleName() string {
-	return "backend"
+func (g backendQuery) List() ([]string, error) {
+	all, err := getAllBackend(g.rdb, g.service)
+	if err != nil {
+		return nil, err
+	}
+	var res []string
+	for _, v := range all {
+		d, _ := json.Marshal(v)
+		res = append(res, string(d))
+	}
+
+	return res, nil
+}
+
+func (g backendQuery) ServiceInfo(serviceid string) (string, error) {
+	info, err := getOneBackend(g.rdb, g.service, serviceid)
+	if err != nil {
+		return "", err
+	}
+	d, err := json.Marshal(info)
+	if err != nil {
+		return "", err
+	}
+
+	return string(d), nil
 }
 
-func (g backendInfoQuery) List() ([]string, error) {
-	return []string{}, nil
+func getAllBackend(rdb *redis.Client, serviceType common.ServiceType) ([]BackendInfo, error) {
+	keys, err := rdb.Keys(context.Background(), common.GetServiceKeyPrefix(serviceType)+"*").Result()
+	if err != nil {
+		return nil, err
+	}
+	var nmInfos []BackendInfo
+	for _, key := range keys {
+		res := rdb.HGetAll(context.Background(), key)
+		var info BackendInfo
+		if err := res.Scan(&info); err != nil {
+			continue
+		}
+		nmInfos = append(nmInfos, info)
+	}
+	return nmInfos, nil
 }
 
-func (g backendInfoQuery) ServiceInfo(serviceid string) (string, error) {
-	return "", nil
+func getOneBackend(rdb *redis.Client, serviceType common.ServiceType, endpoint string) (BackendInfo, error) {
+	k := fmt.Sprintf("%s%s", common.GetServiceKeyPrefix(serviceType), endpoint)
+	res := rdb.HGetAll(context.Background(), k)
+	var info BackendInfo
+	if err := res.Scan(&info); err != nil {
+		return BackendInfo{}, err
+	}
+	return info, nil
 }

query/nm.go

 package query
 
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/odysseus/service-registry/common"
+	"github.com/redis/go-redis/v9"
+)
+
 type NodeManagerInfo struct {
 	Timestamp int64  `redis:"timestamp" json:"timestamp"`
 	Endpoint  string `redis:"endpoint" json:"endpoint"`
@@ -10,20 +18,70 @@ func (g NodeManagerInfo) TimeStamp() int64 {
 }
 
 type nodeManagerQuery struct {
+	rdb     *redis.Client
+	service common.ServiceType
 }
 
-func getNodeManagerQuery() ServiceQuery {
-	return nodeManagerQuery{}
+func getNodeManagerQuery(rdb *redis.Client) ServiceQuery {
+	return nodeManagerQuery{
+		rdb:     rdb,
+		service: common.SERVICE_NODE_MANAGER,
+	}
 }
 
 func (g nodeManagerQuery) ModuleName() string {
-	return "nodemanager"
+	return g.service.String()
 }
 
 func (g nodeManagerQuery) List() ([]string, error) {
-	return []string{}, nil
+	all, err := getAllNm(g.rdb, g.service)
+	if err != nil {
+		return nil, err
+	}
+	var res []string
+	for _, v := range all {
+		d, _ := json.Marshal(v)
+		res = append(res, string(d))
+	}
+	return res, nil
 }
 
 func (g nodeManagerQuery) ServiceInfo(serviceid string) (string, error) {
-	return "", nil
+	info, err := getOneNm(g.rdb, g.service, serviceid)
+	if err != nil {
+		return "", err
+	}
+	d, err := json.Marshal(info)
+	if err != nil {
+		return "", err
+	}
+
+	return string(d), nil
+}
+
+func getAllNm(rdb *redis.Client, serviceType common.ServiceType) ([]NodeManagerInfo, error) {
+	keys, err := rdb.Keys(context.Background(), common.GetServiceKeyPrefix(serviceType)+"*").Result()
+	if err != nil {
+		return nil, err
+	}
+	var nmInfos []NodeManagerInfo
+	for _, key := range keys {
+		res := rdb.HGetAll(context.Background(), key)
+		var info NodeManagerInfo
+		if err := res.Scan(&info); err != nil {
+			continue
+		}
+		nmInfos = append(nmInfos, info)
+	}
+	return nmInfos, nil
+}
+
+func getOneNm(rdb *redis.Client, serviceType common.ServiceType, endpoint string) (NodeManagerInfo, error) {
+	k := fmt.Sprintf("%s%s", common.GetServiceKeyPrefix(serviceType), endpoint)
+	res := rdb.HGetAll(context.Background(), k)
+	var info NodeManagerInfo
+	if err := res.Scan(&info); err != nil {
+		return NodeManagerInfo{}, err
+	}
+	return info, nil
 }

query/query.go

 package query
 
-type ServiceInfo struct {
-	ServiceName string `json:"service_name"`
+import (
+	"github.com/odysseus/service-registry/common"
+	"github.com/odysseus/service-registry/registry"
+	"github.com/redis/go-redis/v9"
+)
+
+type ServiceQuerier struct {
+	rdb              *redis.Client
+	nmQuery          ServiceQuery
+	schedulerQuery   ServiceQuery
+	gatewayQuery     ServiceQuery
+	backendInfoQuery ServiceQuery
+	workerQuery      ServiceQuery
+}
+
+func NewQuery(redisParam registry.RedisConnParam) *ServiceQuerier {
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     redisParam.Addr,
+		Password: redisParam.Password,
+		DB:       redisParam.DbIndex,
+	})
+
+	return &ServiceQuerier{
+		rdb:              rdb,
+		nmQuery:          getNodeManagerQuery(rdb),
+		schedulerQuery:   getSchedulerQuery(rdb),
+		gatewayQuery:     getGatewayQuery(rdb),
+		backendInfoQuery: getBackendQuery(rdb),
+		workerQuery:      getWorkerQuery(rdb),
+	}
 }
 
-func init() {
+func (s *ServiceQuerier) Select(service common.ServiceType) ServiceQuery {
+	switch service {
+	case common.SERVICE_NODE_MANAGER:
+		return s.nmQuery
+	case common.SERVICE_SCHEDULER:
+		return s.schedulerQuery
+	case common.SERVICE_BACKEND:
+		return s.backendInfoQuery
+	case common.SERVICE_API_GATEWAY:
+		return s.gatewayQuery
+	default:
+		return nil
+	}
 
 }

query/schedule.go

 package query
 
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/odysseus/service-registry/common"
+	"github.com/redis/go-redis/v9"
+)
+
 type SchedulerInfo struct {
 	Timestamp int64  `redis:"timestamp" json:"timestamp"`
 	Endpoint  string `redis:"endpoint" json:"endpoint"`
@@ -10,20 +18,70 @@ func (g SchedulerInfo) TimeStamp() int64 {
 }
 
 type schedulerQuery struct {
+	rdb     *redis.Client
+	service common.ServiceType
 }
 
-func getSchedulerQuery() ServiceQuery {
-	return schedulerQuery{}
+func getSchedulerQuery(rdb *redis.Client) ServiceQuery {
+	return schedulerQuery{
+		rdb:     rdb,
+		service: common.SERVICE_SCHEDULER,
+	}
 }
 
 func (g schedulerQuery) ModuleName() string {
-	return "scheduler"
+	return g.service.String()
 }
 
 func (g schedulerQuery) List() ([]string, error) {
-	return []string{}, nil
+	all, err := getAllSc(g.rdb, g.service)
+	if err != nil {
+		return nil, err
+	}
+	var res []string
+	for _, v := range all {
+		d, _ := json.Marshal(v)
+		res = append(res, string(d))
+	}
+	return res, nil
 }
 
 func (g schedulerQuery) ServiceInfo(serviceid string) (string, error) {
-	return "", nil
+	info, err := getOneSc(g.rdb, g.service, serviceid)
+	if err != nil {
+		return "", err
+	}
+	d, err := json.Marshal(info)
+	if err != nil {
+		return "", err
+	}
+
+	return string(d), nil
+}
+
+func getAllSc(rdb *redis.Client, serviceType common.ServiceType) ([]SchedulerInfo, error) {
+	keys, err := rdb.Keys(context.Background(), common.GetServiceKeyPrefix(serviceType)+"*").Result()
+	if err != nil {
+		return nil, err
+	}
+	var nmInfos []SchedulerInfo
+	for _, key := range keys {
+		res := rdb.HGetAll(context.Background(), key)
+		var info SchedulerInfo
+		if err := res.Scan(&info); err != nil {
+			continue
+		}
+		nmInfos = append(nmInfos, info)
+	}
+	return nmInfos, nil
+}
+
+func getOneSc(rdb *redis.Client, serviceType common.ServiceType, endpoint string) (SchedulerInfo, error) {
+	k := fmt.Sprintf("%s%s", common.GetServiceKeyPrefix(serviceType), endpoint)
+	res := rdb.HGetAll(context.Background(), k)
+	var info SchedulerInfo
+	if err := res.Scan(&info); err != nil {
+		return SchedulerInfo{}, err
+	}
+	return info, nil
 }

query/worker.go

 package query
 
+import (
+	"github.com/odysseus/service-registry/common"
+	"github.com/redis/go-redis/v9"
+)
+
 type WorkerInfo struct {
 	Timestamp int64  `redis:"timestamp" json:"timestamp"`
 	Endpoint  string `redis:"endpoint" json:"endpoint"`
@@ -10,14 +15,19 @@ func (g WorkerInfo) TimeStamp() int64 {
 }
 
 type workerQuery struct {
+	rdb     *redis.Client
+	service common.ServiceType
 }
 
-func getWorkerQuery() ServiceQuery {
-	return workerQuery{}
+func getWorkerQuery(rdb *redis.Client) ServiceQuery {
+	return workerQuery{
+		rdb:     rdb,
+		service: common.SERVICE_WORKER,
+	}
 }
 
 func (g workerQuery) ModuleName() string {
-	return "worker"
+	return g.service.String()
 }
 
 func (g workerQuery) List() ([]string, error) {

registry/registry.go

-package nmregistry
+package registry
 
 import (
 	"context"

registry/registry_test.go

-package nmregistry
+package registry
 
 import (
 	"encoding/json"