backend: Update cache usage

src/backend/cmd/apiserver/apiserver.go

@@ -1,10 +1,11 @@
 package apiserver
 
 import (
-	"github.com/Qihoo360/wayne/src/backend/initial"
-	_ "github.com/Qihoo360/wayne/src/backend/routers"
 	"github.com/astaxie/beego"
 	"github.com/spf13/cobra"
+
+	"github.com/Qihoo360/wayne/src/backend/initial"
+	_ "github.com/Qihoo360/wayne/src/backend/routers"
 )
 
 var (

src/backend/controllers/kubernetes/daemonset/daemonset.go

@@ -161,7 +161,7 @@ func (c *KubeDaemonSetController) Get() {
 	name := c.Ctx.Input.Param(":daemonSet")
 	manager, err := client.Manager(cluster)
 	if err == nil {
-		result, err := daemonset.GetDaemonSetDetail(manager.Client, manager.Indexer, name, namespace)
+		result, err := daemonset.GetDaemonSetDetail(manager.Client, manager.CacheFactory, name, namespace)
 		if err != nil {
 			logs.Error("get kubernetes daemonSet detail error.", cluster, namespace, name, err)
 			c.HandleError(err)

src/backend/controllers/kubernetes/deployment/deployment.go

@@ -72,7 +72,7 @@ func (c *KubeDeploymentController) List() {
 
 	manager, err := client.Manager(cluster)
 	if err == nil {
-		result, err := deployment.GetDeploymentPage(manager.Indexer, namespace, param)
+		result, err := deployment.GetDeploymentPage(manager.CacheFactory, namespace, param)
 		if err != nil {
 			logs.Error("list kubernetes deployments error.", cluster, namespace, err)
 			c.HandleError(err)
@@ -295,7 +295,7 @@ func (c *KubeDeploymentController) GetDetail() {
 	name := c.Ctx.Input.Param(":deployment")
 	manager, err := client.Manager(cluster)
 	if err == nil {
-		result, err := deployment.GetDeploymentDetail(manager.Client, manager.Indexer, name, namespace)
+		result, err := deployment.GetDeploymentDetail(manager.Client, manager.CacheFactory, name, namespace)
 		if err != nil {
 			logs.Error("get kubernetes deployment detail error.", cluster, namespace, name, err)
 			c.HandleError(err)

src/backend/controllers/kubernetes/job/job.go

@@ -103,7 +103,7 @@ func (c *KubeJobController) GetPodsEvent() {
 	cronjobName := c.Ctx.Input.Param(":cronjob")
 	manager, err := client.Manager(cluster)
 	if err == nil {
-		result, err := job.GetPodsEvent(manager.Client, manager.Indexer, namespace, name, cronjobName)
+		result, err := job.GetPodsEvent(manager.Client, manager.CacheFactory, namespace, name, cronjobName)
 		if err != nil {
 			logs.Error("get kubernetes job pods event error.", cluster, namespace, name, err)
 			c.HandleError(err)

src/backend/controllers/kubernetes/namespace/namespace.go

@@ -4,6 +4,10 @@ import (
 	"encoding/json"
 	"sync"
 
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+
 	"github.com/Qihoo360/wayne/src/backend/client"
 	"github.com/Qihoo360/wayne/src/backend/controllers/base"
 	"github.com/Qihoo360/wayne/src/backend/models"
@@ -12,9 +16,7 @@ import (
 	"github.com/Qihoo360/wayne/src/backend/util"
 	"github.com/Qihoo360/wayne/src/backend/util/hack"
 	"github.com/Qihoo360/wayne/src/backend/util/logs"
-	"k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/labels"
-	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"github.com/Qihoo360/wayne/src/backend/util/maps"
 )
 
 type KubeNamespaceController struct {
@@ -180,8 +182,10 @@ func (c *KubeNamespaceController) Resources() {
 	syncResourceMap := sync.Map{}
 	var errs []error
 	wg := sync.WaitGroup{}
+
 	managers := client.Managers()
-	for _, manager := range managers {
+	managers.Range(func(key, value interface{}) bool {
+		manager := value.(*client.ClusterManager)
 		wg.Add(1)
 		go func(m *client.ClusterManager) {
 			defer wg.Done()
@@ -215,9 +219,11 @@ func (c *KubeNamespaceController) Resources() {
 				},
 			})
 		}(manager)
-	}
+		return true
+	})
 	wg.Wait()
-	if len(errs) == len(managers) && len(errs) > 0 {
+
+	if len(errs) == maps.SyncMapLen(managers) && len(errs) > 0 {
 		c.HandleError(utilerrors.NewAggregate(errs))
 		return
 	}
@@ -256,7 +262,8 @@ func (c *KubeNamespaceController) Statistics() {
 	var errs []error
 	wg := sync.WaitGroup{}
 	managers := client.Managers()
-	for _, manager := range managers {
+	managers.Range(func(key, value interface{}) bool {
+		manager := value.(*client.ClusterManager)
 		wg.Add(1)
 		go func(m *client.ClusterManager) {
 			defer wg.Done()
@@ -275,9 +282,11 @@ func (c *KubeNamespaceController) Statistics() {
 			}
 			syncResourceMap.Store(m.Cluster.Name, resourceUsage)
 		}(manager)
-	}
+		return true
+	})
+
 	wg.Wait()
-	if len(errs) == len(managers) && len(errs) > 0 {
+	if len(errs) == maps.SyncMapLen(managers) && len(errs) > 0 {
 		c.HandleError(utilerrors.NewAggregate(errs))
 		return
 	}

src/backend/controllers/kubernetes/node/node.go

@@ -46,20 +46,24 @@ func (c *KubeNodeController) NodeStatistics() {
 		managers := client.Managers()
 		var errs []error
 		wg := sync.WaitGroup{}
-		for clu, manager := range managers {
+
+		managers.Range(func(key, value interface{}) bool {
+			manager := value.(*client.ClusterManager)
+			clu := key.(string)
 			wg.Add(1)
 			go func(clu string, mang *client.ClusterManager) {
 				defer wg.Done()
-				count, err := node.GetNodeCounts(mang.Indexer)
+				count, err := node.GetNodeCounts(mang.CacheFactory)
 				if err != nil {
 					logs.Error("get k8s nodes count error. %v", err.Error())
 					errs = append(errs, err)
 				}
 				total += count
 				countSyncMap.Store(clu, count)
 			}(clu, manager)
+			return true
+		})
 
-		}
 		wg.Wait()
 		if len(errs) > 0 {
 			c.HandleError(utilerrors.NewAggregate(errs))
@@ -72,7 +76,7 @@ func (c *KubeNodeController) NodeStatistics() {
 	} else {
 		manager, err := client.Manager(cluster)
 		if err == nil {
-			count, err := node.GetNodeCounts(manager.Indexer)
+			count, err := node.GetNodeCounts(manager.CacheFactory)
 			if err != nil {
 				logs.Error("get k8s nodes count error. %v", err.Error())
 				c.HandleError(err)
@@ -95,7 +99,7 @@ func (c *KubeNodeController) List() {
 	cluster := c.Ctx.Input.Param(":cluster")
 	manager, err := client.Manager(cluster)
 	if err == nil {
-		result, err := node.ListNode(manager.Indexer)
+		result, err := node.ListNode(manager.CacheFactory)
 		if err != nil {
 			logs.Error("list node by cluster (%s) error.%v", cluster, err)
 			c.HandleError(err)

src/backend/controllers/kubernetes/pod/pod.go

@@ -50,15 +50,23 @@ func (c *KubePodController) PodStatistics() {
 	if cluster == "" {
 		managers := client.Managers()
 		wg := sync.WaitGroup{}
-		for _, manager := range managers {
+
+		managers.Range(func(key, value interface{}) bool {
+			manager := value.(*client.ClusterManager)
 			wg.Add(1)
 			go func(manager *client.ClusterManager) {
 				defer wg.Done()
-				count := pod.GetPodCounts(manager.Indexer)
+				count, err := pod.GetPodCounts(manager.CacheFactory)
+				if err != nil {
+					logs.Error("get pod counts error.", key, err)
+					return
+				}
 				total += count
 				countSyncMap.Store(manager.Cluster.Name, count)
 			}(manager)
-		}
+			return true
+		})
+
 		wg.Wait()
 		countSyncMap.Range(func(key, value interface{}) bool {
 			countMap[key.(string)] = value.(int)
@@ -68,7 +76,11 @@ func (c *KubePodController) PodStatistics() {
 	} else {
 		manager, err := client.Manager(cluster)
 		if err == nil {
-			count := pod.GetPodCounts(manager.Indexer)
+			count, err := pod.GetPodCounts(manager.CacheFactory)
+			if err != nil {
+				c.HandleError(err)
+				return
+			}
 			total += count
 		} else {
 			c.HandleError(err)
@@ -93,18 +105,18 @@ func (c *KubePodController) List() {
 	statefulset := c.Input().Get("statefulset")
 	daemonSet := c.Input().Get("daemonSet")
 	job := c.Input().Get("job")
-	cli, err := client.Client(cluster)
+	manager, err := client.Manager(cluster)
 	if err == nil {
 		var result interface{}
 		var err error
 		if deployment != "" {
-			result, err = pod.GetPodsByDeployment(cli, namespace, deployment)
+			result, err = pod.GetPodsByDeployment(manager.CacheFactory, namespace, deployment)
 		} else if statefulset != "" {
-			result, err = pod.GetPodsByStatefulset(cli, namespace, statefulset)
+			result, err = pod.GetPodsByStatefulset(manager.CacheFactory, namespace, statefulset)
 		} else if daemonSet != "" {
-			result, err = pod.GetPodsByDaemonSet(cli, namespace, daemonSet)
+			result, err = pod.GetPodsByDaemonSet(manager.CacheFactory, namespace, daemonSet)
 		} else if job != "" {
-			result, err = pod.GetPodsByJob(cli, namespace, job)
+			result, err = pod.GetPodsByJob(manager.CacheFactory, namespace, job)
 		} else {
 			err = fmt.Errorf("unknown resource type. ")
 		}

src/backend/controllers/kubernetes/service/service.go

@@ -58,7 +58,7 @@ func (c *KubeServiceController) GetDetail() {
 	if err != nil {
 		c.AbortBadRequestFormat("Cluster")
 	}
-	serviceDetail, err := service.GetServiceDetail(manager.Client, manager.Indexer, namespace, name)
+	serviceDetail, err := service.GetServiceDetail(manager.Client, manager.CacheFactory, namespace, name)
 	if err != nil {
 		logs.Error("get kubernetes(%s) namespace(%s) service(%s) detail error: %s", cluster, namespace, name, err.Error())
 		c.AbortInternalServerError("get kubernetes service detail error.")

src/backend/controllers/kubernetes/statefulset/statefulset.go

@@ -242,7 +242,7 @@ func (c *KubeStatefulsetController) Get() {
 	name := c.Ctx.Input.Param(":statefulset")
 	manager, err := client.Manager(cluster)
 	if err == nil {
-		result, err := statefulset.GetStatefulsetDetail(manager.Client, manager.Indexer, name, namespace)
+		result, err := statefulset.GetStatefulsetDetail(manager.Client, manager.CacheFactory, name, namespace)
 		if err != nil {
 			logs.Error("get kubernetes statefulset detail error.", cluster, namespace, name, err)
 			c.HandleError(err)

src/backend/controllers/openapi/deployment.go

@@ -159,7 +159,7 @@ func (c *OpenAPIController) GetDeploymentStatus() {
 	}
 	manager, err := client.Manager(param.Cluster)
 	if err == nil {
-		deployInfo, err := resdeployment.GetDeploymentDetail(manager.Client, manager.Indexer, param.Deployment, ns.MetaDataObj.Namespace)
+		deployInfo, err := resdeployment.GetDeploymentDetail(manager.Client, manager.CacheFactory, param.Deployment, ns.MetaDataObj.Namespace)
 		if err != nil {
 			logs.Error("Failed to get  k8s deployment state: %s", err.Error())
 			c.AddErrorAndResponse("", http.StatusInternalServerError)
@@ -182,7 +182,7 @@ func (c *OpenAPIController) GetDeploymentStatus() {
 		for _, e := range deployInfo.Pods.Warnings {
 			result.Body.DeploymentStatus.Deployment.PodsState.Warnings = append(result.Body.DeploymentStatus.Deployment.PodsState.Warnings, fmt.Sprint(e))
 		}
-		podInfo, err := pod.GetPodsByDeployment(manager.Client, ns.MetaDataObj.Namespace, param.Deployment)
+		podInfo, err := pod.GetPodsByDeployment(manager.CacheFactory, ns.MetaDataObj.Namespace, param.Deployment)
 		if err != nil {
 			logs.Error("Failed to get k8s pod state: %s", err.Error())
 			c.AddErrorAndResponse("", http.StatusInternalServerError)

src/backend/controllers/openapi/pod.go

@@ -5,6 +5,8 @@ import (
 	"net/http"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/labels"
+
 	"github.com/Qihoo360/wayne/src/backend/client"
 	"github.com/Qihoo360/wayne/src/backend/models"
 	"github.com/Qihoo360/wayne/src/backend/models/response"
@@ -64,21 +66,26 @@ func (c *OpenAPIController) GetPodInfo() {
 		c.AddErrorAndResponse("You can only use global APIKey in this action!", http.StatusUnauthorized)
 		return
 	}
-
-	clis := client.Clients()
 	podList := resppodlist{}
 	podList.Body.Code = http.StatusOK
 	params := PodInfoParam{c.GetString("labelSelector"), c.GetString("cluster")}
 	if params.Cluster == "" {
 		c.AddErrorAndResponse("Invalid cluster parameter:must required!", http.StatusBadRequest)
 		return
 	}
-	if clis[params.Cluster] == nil {
+	manager, err := client.Manager(params.Cluster)
+	if err != nil {
 		c.AddErrorAndResponse("Invalid cluster parameter:not exist!", http.StatusBadRequest)
 		return
 	}
-	cli := clis[params.Cluster]
-	pods, err := pod.GetAllPodByLabelSelector(cli, params.LabelSelector)
+
+	label, err := labels.ConvertSelectorToLabelsMap(params.LabelSelector)
+	if err != nil {
+		c.AddErrorAndResponse(fmt.Sprintf("Invalid LabelSelector parameter: %v!", err), http.StatusBadRequest)
+		return
+	}
+
+	pods, err := pod.ListPod(manager.CacheFactory, "", label)
 	if err != nil {
 		logs.Error(fmt.Sprintf("Failed to parse metadata: %s", err.Error()))
 		c.AddErrorAndResponse(fmt.Sprintf("Maybe a problematic k8s cluster(%s)!", params.Cluster), http.StatusInternalServerError)
@@ -124,7 +131,7 @@ func (c *OpenAPIController) GetPodInfoFromIP() {
 		c.AddErrorAndResponse("Invalid cluster parameter:not exist!", http.StatusBadRequest)
 		return
 	}
-	pods := pod.GetPodsBySelectorFromCache(manager.Indexer, "", nil)
+	pods, err := pod.ListKubePod(manager.CacheFactory, "", nil)
 	if err != nil {
 		logs.Error(fmt.Sprintf("Failed to parse metadata: %s", err.Error()))
 		c.AddErrorAndResponse(fmt.Sprintf("Maybe a problematic k8s cluster(%s)!", params.Cluster), http.StatusInternalServerError)

src/backend/initial/client.go

@@ -3,8 +3,9 @@ package initial
 import (
 	"time"
 
-	"github.com/Qihoo360/wayne/src/backend/client"
 	"k8s.io/apimachinery/pkg/util/wait"
+
+	"github.com/Qihoo360/wayne/src/backend/client"
 )
 
 func InitClient() {

src/backend/resources/daemonset/daemonset.go

@@ -1,16 +1,16 @@
 package daemonset
 
 import (
+	"k8s.io/api/extensions/v1beta1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
 	"github.com/Qihoo360/wayne/src/backend/client"
 	"github.com/Qihoo360/wayne/src/backend/resources/common"
 	"github.com/Qihoo360/wayne/src/backend/resources/event"
 	"github.com/Qihoo360/wayne/src/backend/resources/pod"
 	"github.com/Qihoo360/wayne/src/backend/util/maps"
-	"k8s.io/api/extensions/v1beta1"
-	"k8s.io/apimachinery/pkg/api/errors"
-	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/client-go/kubernetes"
 )
 
 type DaemonSet struct {
@@ -34,7 +34,7 @@ func CreateOrUpdateDaemonSet(cli *kubernetes.Clientset, daemonSet *v1beta1.Daemo
 	return cli.ExtensionsV1beta1().DaemonSets(daemonSet.Namespace).Update(old)
 }
 
-func GetDaemonSetDetail(cli *kubernetes.Clientset, indexer *client.CacheIndexer, name, namespace string) (*DaemonSet, error) {
+func GetDaemonSetDetail(cli *kubernetes.Clientset, indexer *client.CacheFactory, name, namespace string) (*DaemonSet, error) {
 	daemonSet, err := cli.ExtensionsV1beta1().DaemonSets(namespace).Get(name, metaV1.GetOptions{})
 	if err != nil {
 		return nil, err
@@ -48,13 +48,15 @@ func GetDaemonSetDetail(cli *kubernetes.Clientset, indexer *client.CacheIndexer,
 	podInfo.Current = daemonSet.Status.NumberAvailable
 	podInfo.Desired = daemonSet.Status.DesiredNumberScheduled
 
-	podSelector := labels.SelectorFromSet(daemonSet.Spec.Template.Labels).String()
-	pods, err := pod.GetPodsBySelector(cli, namespace, podSelector)
+	pods, err := pod.ListKubePod(indexer, namespace, daemonSet.Spec.Template.Labels)
 	if err != nil {
 		return nil, err
 	}
 
-	podInfo.Warnings = event.GetPodsWarningEvents(indexer, pods)
+	podInfo.Warnings, err = event.GetPodsWarningEvents(indexer, pods)
+	if err != nil {
+		return nil, err
+	}
 
 	result.Pods = podInfo