diff --git a/server/internal/biz/node.go b/server/internal/biz/node.go index 32027722..536279d6 100644 --- a/server/internal/biz/node.go +++ b/server/internal/biz/node.go @@ -51,6 +51,7 @@ type NodeRepo interface { GetNode(context.Context, string) (*Node, error) ListAllDevices(context.Context) ([]*DeviceInfo, error) FindDeviceByAliasId(string) (*DeviceInfo, error) + Ready() bool } type NodeUsecase struct { @@ -77,3 +78,7 @@ func (uc *NodeUsecase) ListAllDevices(ctx context.Context) ([]*DeviceInfo, error func (uc *NodeUsecase) FindDeviceByAliasId(aliasId string) (*DeviceInfo, error) { return uc.repo.FindDeviceByAliasId(aliasId) } + +func (uc *NodeUsecase) Ready() bool { + return uc.repo.Ready() +} diff --git a/server/internal/biz/pod.go b/server/internal/biz/pod.go index 4162db75..98da6d4c 100644 --- a/server/internal/biz/pod.go +++ b/server/internal/biz/pod.go @@ -37,6 +37,7 @@ type PodInfo struct { type PodRepo interface { ListAll(context.Context) ([]*Container, error) FindOne(context.Context, string, string) (*Container, error) + Ready() bool } type PodUseCase struct { @@ -72,6 +73,10 @@ func (uc *PodUseCase) ListAll(ctx context.Context) ([]*Container, error) { return uc.repo.ListAll(ctx) } +func (uc *PodUseCase) Ready() bool { + return uc.repo.Ready() +} + func ContainersStatisticsInfo(containers []*Container, deviceId string) (int32, int32, int32) { var vGPU int32 = 0 var core int32 = 0 diff --git a/server/internal/data/node.go b/server/internal/data/node.go index 4f59ab6e..a9537176 100644 --- a/server/internal/data/node.go +++ b/server/internal/data/node.go @@ -30,6 +30,7 @@ type nodeRepo struct { log *log.Helper mutex sync.RWMutex providers []provider.Provider + ready bool } // NewNodeRepo . @@ -100,6 +101,16 @@ func (r *nodeRepo) updateLocalNodes() { } r.mutex.Lock() r.nodes = n + if !r.ready { + r.ready = true + r.log.Infof("node data initialized: %d nodes, %d devices", len(n), func() int { + count := 0 + for _, node := range n { + count += len(node.Devices) + } + return count + }()) + } r.mutex.Unlock() } } @@ -116,7 +127,26 @@ func (r *nodeRepo) init() { }) stopCh := make(chan struct{}) informerFactory.Start(stopCh) - informerFactory.WaitForCacheSync(stopCh) + synced := informerFactory.WaitForCacheSync(stopCh) + syncedOK := true + for _, ok := range synced { + if !ok { + syncedOK = false + break + } + } + if syncedOK { + r.log.Info("node informer synced successfully") + } else { + r.log.Warn("node informer failed to sync, data may be incomplete") + } +} + +// Ready returns true if the node informer has successfully synced and populated data. +func (r *nodeRepo) Ready() bool { + r.mutex.RLock() + defer r.mutex.RUnlock() + return r.ready } func (r *nodeRepo) onAddNode(obj interface{}) { diff --git a/server/internal/data/pod.go b/server/internal/data/pod.go index e17857ec..6fa7f30b 100644 --- a/server/internal/data/pod.go +++ b/server/internal/data/pod.go @@ -24,6 +24,7 @@ type podRepo struct { pods map[k8stypes.UID]*biz.PodInfo mutex sync.RWMutex log *log.Helper + ready bool } func NewPodRepo(data *Data, logger log.Logger) biz.PodRepo { @@ -47,7 +48,29 @@ func (r *podRepo) init() { }) stopCh := make(chan struct{}) informerFactory.Start(stopCh) - informerFactory.WaitForCacheSync(stopCh) + synced := informerFactory.WaitForCacheSync(stopCh) + syncedOK := true + for _, ok := range synced { + if !ok { + syncedOK = false + break + } + } + if syncedOK { + r.mutex.Lock() + r.ready = true + r.mutex.Unlock() + r.log.Info("pod informer synced successfully") + } else { + r.log.Warn("pod informer failed to sync, data may be incomplete") + } +} + +// Ready returns true if the pod informer has successfully synced. +func (r *podRepo) Ready() bool { + r.mutex.RLock() + defer r.mutex.RUnlock() + return r.ready } func (r *podRepo) onAddPod(obj interface{}) { diff --git a/server/internal/server/http.go b/server/internal/server/http.go index 73906944..213417b0 100644 --- a/server/internal/server/http.go +++ b/server/internal/server/http.go @@ -57,10 +57,13 @@ func NewHTTPServer(c *conf.Bootstrap, srv.HandlePrefix("/q/", openapiv2.NewHandler()) srv.Handle("/metrics", promhttp.Handler()) srv.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { - // Reaching this handler means the server is listening and the informer - // caches have already synced (the backend only starts serving after that). - w.WriteHeader(nethttp.StatusOK) - _, _ = w.Write([]byte("ok")) + if node.Ready() { + w.WriteHeader(nethttp.StatusOK) + _, _ = w.Write([]byte("ok")) + } else { + w.WriteHeader(nethttp.StatusServiceUnavailable) + _, _ = w.Write([]byte("not ready")) + } }) return srv } diff --git a/server/internal/service/card.go b/server/internal/service/card.go index 4d3ec2b4..044e327c 100644 --- a/server/internal/service/card.go +++ b/server/internal/service/card.go @@ -21,7 +21,13 @@ func NewCardService(node *biz.NodeUsecase, pod *biz.PodUseCase, ms *MonitorServi } func (s *CardService) GetAllGPUs(ctx context.Context, req *pb.GetAllGpusReq) (*pb.GPUsReply, error) { - filters := req.Filters + var filters *pb.GetAllGpusReq_Filters + if req != nil { + filters = req.Filters + } + if filters == nil { + filters = &pb.GetAllGpusReq_Filters{} + } deviceInfos, err := s.node.ListAllDevices(ctx) if err != nil { return nil, err @@ -115,7 +121,13 @@ func (s *CardService) GetAllGPUTypes(ctx context.Context, req *pb.GetAllGpusReq) var res = &pb.GPUsReply{List: []*pb.GPUReply{}} seenTypes := make(map[string]struct{}) - filters := req.Filters + var filters *pb.GetAllGpusReq_Filters + if req != nil { + filters = req.Filters + } + if filters == nil { + filters = &pb.GetAllGpusReq_Filters{} + } provider := strings.Trim(filters.Provider, " ") for _, device := range deviceInfos { if provider != "" && provider != device.Provider { @@ -134,6 +146,9 @@ func (s *CardService) GetAllGPUTypes(ctx context.Context, req *pb.GetAllGpusReq) } func (s *CardService) GetGPU(ctx context.Context, req *pb.GetGpuReq) (*pb.GPUReply, error) { + if req == nil { + return &pb.GPUReply{}, nil + } devices, err := s.node.ListAllDevices(ctx) if err != nil { return nil, err diff --git a/server/internal/service/container.go b/server/internal/service/container.go index c108586d..91872b0d 100644 --- a/server/internal/service/container.go +++ b/server/internal/service/container.go @@ -45,7 +45,13 @@ func uniqueNonEmpty(values []string) []string { } func (s *ContainerService) GetAllContainers(ctx context.Context, req *pb.GetAllContainersReq) (*pb.ContainersReply, error) { - filters := req.Filters + var filters *pb.GetAllContainersReq_Filters + if req != nil { + filters = req.Filters + } + if filters == nil { + filters = &pb.GetAllContainersReq_Filters{} + } containers, err := s.pod.ListAllContainers(ctx) if err != nil { return nil, err diff --git a/server/internal/service/node.go b/server/internal/service/node.go index 37b43d1c..5f1e954a 100644 --- a/server/internal/service/node.go +++ b/server/internal/service/node.go @@ -26,16 +26,33 @@ func NewNodeService(uc *biz.NodeUsecase, pod *biz.PodUseCase, summary *biz.Summa return &NodeService{uc: uc, pod: pod, summary: summary, ms: ms} } +// Ready returns true if both node and pod informers have synced. +func (s *NodeService) Ready() bool { + return s.uc.Ready() && s.pod.Ready() +} + func (s *NodeService) GetSummary(ctx context.Context, req *pb.GetSummaryReq) (*pb.DeviceSummaryReply, error) { - filters := req.Filters + var filters *pb.GetSummaryReq_Filters + if req != nil { + filters = req.Filters + } var res = &pb.DeviceSummaryReply{} + if filters == nil { + filters = &pb.GetSummaryReq_Filters{} + } t, err := s.summary.GetGPUSummary(ctx, filters.DeviceId, filters.NodeUid, filters.Type) copier.Copy(&res, &t) return res, err } func (s *NodeService) GetAllNodes(ctx context.Context, req *pb.GetAllNodesReq) (*pb.NodesReply, error) { - filters := req.Filters + var filters *pb.GetAllNodesReq_Filters + if req != nil { + filters = req.Filters + } + if filters == nil { + filters = &pb.GetAllNodesReq_Filters{} + } nodes, err := s.uc.ListAllNodes(ctx) if err != nil { return nil, err @@ -88,6 +105,9 @@ func (s *NodeService) GetAllNodes(ctx context.Context, req *pb.GetAllNodesReq) ( } func (s *NodeService) GetNode(ctx context.Context, req *pb.GetNodeReq) (*pb.NodeReply, error) { + if req == nil || req.Uid == "" { + return &pb.NodeReply{}, nil + } node, err := s.uc.GetNode(ctx, req.Uid) if err != nil { return nil, err