Add replication controller stats in CLI (#794)

* Add replication controller stats in CLI
* Fix pod status in stat summary tests

Signed-off-by: Kevin Lingerfelt <kl@buoyant.io>

cli/cmd/stat.go

@@ -34,22 +34,28 @@ Valid resource types include:
 	* deployments
 	* namespaces
 	* pods
+	* replicationcontrollers
 
 This command will hide resources that have completed, such as pods that are in the Succeeded or Failed phases.
 If no resource name is specified, displays stats about all resources of the specified RESOURCETYPE`,
 	Example: `  # Get all deployments in the test namespace.
   conduit stat deployments -n test
 
-  # Get the hello1 deployment in the test namespace.
-  conduit stat deployments hello1 -n test
+  # Get the hello1 replication controller in the test namespace.
+  conduit stat replicationcontrollers hello1 -n test
 
   # Get all namespaces.
   conduit stat namespaces
 
   # Get all pods in all namespaces that call the hello1 deployment in the test namesapce.
   conduit stat pods --to hello1 --to-resource deployment --to-namespace test --all-namespaces`,
-	Args:      cobra.RangeArgs(1, 2),
-	ValidArgs: []string{k8s.KubernetesDeployments, k8s.KubernetesNamespaces, k8s.KubernetesPods},
+	Args: cobra.RangeArgs(1, 2),
+	ValidArgs: []string{
+		k8s.KubernetesDeployments,
+		k8s.KubernetesNamespaces,
+		k8s.KubernetesPods,
+		k8s.KubernetesReplicationControllers,
+	},
 	RunE: func(cmd *cobra.Command, args []string) error {
 		switch len(args) {
 		case 1:

cli/cmd/testdata/install_default.golden

@@ -23,7 +23,7 @@ rules:
   resources: ["deployments", "replicasets"]
   verbs: ["list", "get", "watch"]
 - apiGroups: [""]
-  resources: ["pods", "endpoints", "services", "namespaces"]
+  resources: ["pods", "endpoints", "services", "namespaces", "replicationcontrollers"]
   verbs: ["list", "get", "watch"]
 
 ---

cli/cmd/testdata/install_output.golden

@@ -23,7 +23,7 @@ rules:
   resources: ["deployments", "replicasets"]
   verbs: ["list", "get", "watch"]
 - apiGroups: [""]
-  resources: ["pods", "endpoints", "services", "namespaces"]
+  resources: ["pods", "endpoints", "services", "namespaces", "replicationcontrollers"]
   verbs: ["list", "get", "watch"]
 
 ---

cli/install/template.go

@@ -26,7 +26,7 @@ rules:
   resources: ["deployments", "replicasets"]
   verbs: ["list", "get", "watch"]
 - apiGroups: [""]
-  resources: ["pods", "endpoints", "services", "namespaces"]
+  resources: ["pods", "endpoints", "services", "namespaces", "replicationcontrollers"]
   verbs: ["list", "get", "watch"]
 
 ---

controller/api/public/grpc_server.go

@@ -22,19 +22,20 @@ import (
 
 type (
 	grpcServer struct {
-		prometheusAPI       promv1.API
-		tapClient           tapPb.TapClient
-		namespaceLister     corelisters.NamespaceLister
-		deployLister        applisters.DeploymentLister
-		replicaSetLister    applisters.ReplicaSetLister
-		podLister           corelisters.PodLister
-		controllerNamespace string
-		ignoredNamespaces   []string
+		prometheusAPI               promv1.API
+		tapClient                   tapPb.TapClient
+		namespaceLister             corelisters.NamespaceLister
+		deployLister                applisters.DeploymentLister
+		replicaSetLister            applisters.ReplicaSetLister
+		podLister                   corelisters.PodLister
+		replicationControllerLister corelisters.ReplicationControllerLister
+		controllerNamespace         string
+		ignoredNamespaces           []string
 	}
 )
 
 const (
-	podQuery                   = "sum(request_total) by (pod)"
+	podQuery                   = "count(process_start_time_seconds) by (pod)"
 	K8sClientSubsystemName     = "kubernetes"
 	K8sClientCheckDescription  = "control plane can talk to Kubernetes"
 	PromClientSubsystemName    = "prometheus"
@@ -48,18 +49,20 @@ func newGrpcServer(
 	deployLister applisters.DeploymentLister,
 	replicaSetLister applisters.ReplicaSetLister,
 	podLister corelisters.PodLister,
+	replicationControllerLister corelisters.ReplicationControllerLister,
 	controllerNamespace string,
 	ignoredNamespaces []string,
 ) *grpcServer {
 	return &grpcServer{
-		prometheusAPI:       promAPI,
-		tapClient:           tapClient,
-		namespaceLister:     namespaceLister,
-		deployLister:        deployLister,
-		replicaSetLister:    replicaSetLister,
-		podLister:           podLister,
-		controllerNamespace: controllerNamespace,
-		ignoredNamespaces:   ignoredNamespaces,
+		prometheusAPI:               promAPI,
+		tapClient:                   tapClient,
+		namespaceLister:             namespaceLister,
+		deployLister:                deployLister,
+		replicaSetLister:            replicaSetLister,
+		podLister:                   podLister,
+		replicationControllerLister: replicationControllerLister,
+		controllerNamespace:         controllerNamespace,
+		ignoredNamespaces:           ignoredNamespaces,
 	}
 }
 

controller/api/public/grpc_server_test.go

@@ -170,6 +170,7 @@ spec:
 			deployInformer := sharedInformers.Apps().V1beta2().Deployments()
 			replicaSetInformer := sharedInformers.Apps().V1beta2().ReplicaSets()
 			podInformer := sharedInformers.Core().V1().Pods()
+			replicationControllerInformer := sharedInformers.Core().V1().ReplicationControllers()
 
 			fakeGrpcServer := newGrpcServer(
 				&MockProm{Res: exp.promRes},
@@ -178,6 +179,7 @@ spec:
 				deployInformer.Lister(),
 				replicaSetInformer.Lister(),
 				podInformer.Lister(),
+				replicationControllerInformer.Lister(),
 				"conduit",
 				[]string{},
 			)
@@ -189,6 +191,7 @@ spec:
 				deployInformer.Informer().HasSynced,
 				replicaSetInformer.Informer().HasSynced,
 				podInformer.Informer().HasSynced,
+				replicationControllerInformer.Informer().HasSynced,
 			) {
 				t.Fatalf("timed out wait for caches to sync")
 			}

controller/api/public/http_server.go

@@ -200,6 +200,7 @@ func NewServer(
 	deployLister applisters.DeploymentLister,
 	replicaSetLister applisters.ReplicaSetLister,
 	podLister corelisters.PodLister,
+	replicationControllerLister corelisters.ReplicationControllerLister,
 	controllerNamespace string,
 	ignoredNamespaces []string,
 ) *http.Server {
@@ -211,6 +212,7 @@ func NewServer(
 			deployLister,
 			replicaSetLister,
 			podLister,
+			replicationControllerLister,
 			controllerNamespace,
 			ignoredNamespaces,
 		),

controller/api/public/stat_summary.go

@@ -35,7 +35,11 @@ const (
 	promLatencyP95 = promType("0.95")
 	promLatencyP99 = promType("0.99")
 
-	namespaceLabel    = model.LabelName("namespace")
+	deploymentLabel            = model.LabelName("deployment")
+	namespaceLabel             = model.LabelName("namespace")
+	podLabel                   = model.LabelName("pod")
+	replicationControllerLabel = model.LabelName("replication_controller")
+
 	dstNamespaceLabel = model.LabelName("dst_namespace")
 )
 
@@ -43,9 +47,10 @@ var (
 	promTypes = []promType{promRequests, promLatencyP50, promLatencyP95, promLatencyP99}
 
 	k8sResourceTypesToPromLabels = map[string]model.LabelName{
-		k8s.KubernetesDeployments: "deployment",
-		k8s.KubernetesNamespaces:  namespaceLabel,
-		k8s.KubernetesPods:        "pod",
+		k8s.KubernetesDeployments:            deploymentLabel,
+		k8s.KubernetesNamespaces:             namespaceLabel,
+		k8s.KubernetesPods:                   podLabel,
+		k8s.KubernetesReplicationControllers: replicationControllerLabel,
 	}
 )
 
@@ -66,6 +71,8 @@ func (s *grpcServer) StatSummary(ctx context.Context, req *pb.StatSummaryRequest
 		objectMap, meshCount, err = s.getNamespaces(req.Selector.Resource)
 	case k8s.KubernetesPods:
 		objectMap, meshCount, err = s.getPods(req.Selector.Resource)
+	case k8s.KubernetesReplicationControllers:
+		objectMap, meshCount, err = s.getReplicationControllers(req.Selector.Resource)
 	default:
 		err = fmt.Errorf("Unimplemented resource type: %v", req.Selector.Resource.Type)
 	}
@@ -405,7 +412,7 @@ func (s *grpcServer) getPods(res *pb.Resource) (map[string]metav1.ObjectMeta, ma
 	meshedPodCount := make(map[string]*meshedCount)
 	podMap := make(map[string]metav1.ObjectMeta)
 	for _, pod := range pods {
-		if pod.Status.Phase != apiv1.PodRunning {
+		if !isPendingOrRunning(pod) {
 			continue
 		}
 
@@ -425,6 +432,43 @@ func (s *grpcServer) getPods(res *pb.Resource) (map[string]metav1.ObjectMeta, ma
 	return podMap, meshedPodCount, nil
 }
 
+func (s *grpcServer) getReplicationControllers(res *pb.Resource) (map[string]metav1.ObjectMeta, map[string]*meshedCount, error) {
+	var err error
+	var rcs []*apiv1.ReplicationController
+
+	if res.Namespace == "" {
+		rcs, err = s.replicationControllerLister.List(labels.Everything())
+	} else if res.Name == "" {
+		rcs, err = s.replicationControllerLister.ReplicationControllers(res.Namespace).List(labels.Everything())
+	} else {
+		var rc *apiv1.ReplicationController
+		rc, err = s.replicationControllerLister.ReplicationControllers(res.Namespace).Get(res.Name)
+		rcs = []*apiv1.ReplicationController{rc}
+	}
+
+	if err != nil {
+		return nil, nil, err
+	}
+
+	meshedPodCount := make(map[string]*meshedCount)
+	rcMap := make(map[string]metav1.ObjectMeta)
+	for _, rc := range rcs {
+		key, err := cache.MetaNamespaceKeyFunc(rc)
+		if err != nil {
+			return nil, nil, err
+		}
+		rcMap[key] = rc.ObjectMeta
+
+		meshCount, err := s.getMeshedPodCount(rc.Namespace, rc)
+		if err != nil {
+			return nil, nil, err
+		}
+		meshedPodCount[key] = meshCount
+	}
+
+	return rcMap, meshedPodCount, nil
+}
+
 func (s *grpcServer) getMeshedPodCount(namespace string, obj runtime.Object) (*meshedCount, error) {
 	selector, err := getSelectorFromObject(obj)
 	if err != nil {
@@ -438,6 +482,10 @@ func (s *grpcServer) getMeshedPodCount(namespace string, obj runtime.Object) (*m
 
 	meshCount := &meshedCount{}
 	for _, pod := range pods {
+		if !isPendingOrRunning(pod) {
+			continue
+		}
+
 		meshCount.total++
 		if isInMesh(pod) {
 			meshCount.inMesh++
@@ -452,6 +500,13 @@ func isInMesh(pod *apiv1.Pod) bool {
 	return ok
 }
 
+func isPendingOrRunning(pod *apiv1.Pod) bool {
+	pending := pod.Status.Phase == apiv1.PodPending
+	running := pod.Status.Phase == apiv1.PodRunning
+	terminating := pod.DeletionTimestamp != nil
+	return (pending || running) && !terminating
+}
+
 func getSelectorFromObject(obj runtime.Object) (labels.Selector, error) {
 	switch typed := obj.(type) {
 	case *apiv1.Namespace:
@@ -460,6 +515,9 @@ func getSelectorFromObject(obj runtime.Object) (labels.Selector, error) {
 	case *appsv1beta2.Deployment:
 		return labels.Set(typed.Spec.Selector.MatchLabels).AsSelector(), nil
 
+	case *apiv1.ReplicationController:
+		return labels.Set(typed.Spec.Selector).AsSelector(), nil
+
 	default:
 		return nil, fmt.Errorf("Cannot get object selector: %v", obj)
 	}

controller/api/public/stat_summary_test.go

@@ -56,6 +56,8 @@ metadata:
     app: emoji-svc
   annotations:
     conduit.io/proxy-version: testinjectversion
+status:
+  phase: Running
 `, `
 apiVersion: v1
 kind: Pod
@@ -64,6 +66,20 @@ metadata:
   namespace: emojivoto
   labels:
     app: emoji-svc
+status:
+  phase: Running
+`, `
+apiVersion: v1
+kind: Pod
+metadata:
+  name: emojivoto-meshed-not-running
+  namespace: emojivoto
+  labels:
+    app: emoji-svc
+  annotations:
+    conduit.io/proxy-version: testinjectversion
+status:
+  phase: Completed
 `,
 				},
 				promRes: model.Vector{
@@ -140,6 +156,7 @@ metadata:
 			deployInformer := sharedInformers.Apps().V1beta2().Deployments()
 			replicaSetInformer := sharedInformers.Apps().V1beta2().ReplicaSets()
 			podInformer := sharedInformers.Core().V1().Pods()
+			replicationControllerInformer := sharedInformers.Core().V1().ReplicationControllers()
 
 			fakeGrpcServer := newGrpcServer(
 				&MockProm{Res: exp.promRes},
@@ -148,6 +165,7 @@ metadata:
 				deployInformer.Lister(),
 				replicaSetInformer.Lister(),
 				podInformer.Lister(),
+				replicationControllerInformer.Lister(),
 				"conduit",
 				[]string{},
 			)
@@ -159,6 +177,7 @@ metadata:
 				deployInformer.Informer().HasSynced,
 				replicaSetInformer.Informer().HasSynced,
 				podInformer.Informer().HasSynced,
+				replicationControllerInformer.Informer().HasSynced,
 			) {
 				t.Fatalf("timed out wait for caches to sync")
 			}
@@ -218,6 +237,7 @@ metadata:
 				sharedInformers.Apps().V1beta2().Deployments().Lister(),
 				sharedInformers.Apps().V1beta2().ReplicaSets().Lister(),
 				sharedInformers.Core().V1().Pods().Lister(),
+				sharedInformers.Core().V1().ReplicationControllers().Lister(),
 				"conduit",
 				[]string{},
 			)

controller/cmd/public-api/main.go

@@ -69,6 +69,9 @@ func main() {
 	podInformer := sharedInformers.Core().V1().Pods()
 	podInformerSynced := podInformer.Informer().HasSynced
 
+	replicationControllerInformer := sharedInformers.Core().V1().ReplicationControllers()
+	replicationControllerInformerSynced := replicationControllerInformer.Informer().HasSynced
+
 	sharedInformers.Start(nil)
 
 	prometheusClient, err := promApi.NewClient(promApi.Config{Address: *prometheusUrl})
@@ -84,6 +87,7 @@ func main() {
 		deployInformer.Lister(),
 		replicaSetInformer.Lister(),
 		podInformer.Lister(),
+		replicationControllerInformer.Lister(),
 		*controllerNamespace,
 		strings.Split(*ignoredNamespaces, ","),
 	)
@@ -99,6 +103,7 @@ func main() {
 			deployInformerSynced,
 			replicaSetInformerSynced,
 			podInformerSynced,
+			replicationControllerInformerSynced,
 		) {
 			log.Fatalf("timed out wait for caches to sync")
 		}

pkg/k8s/k8s.go

@@ -9,9 +9,10 @@ import (
 )
 
 const (
-	KubernetesDeployments = "deployments"
-	KubernetesNamespaces  = "namespaces"
-	KubernetesPods        = "pods"
+	KubernetesDeployments            = "deployments"
+	KubernetesNamespaces             = "namespaces"
+	KubernetesPods                   = "pods"
+	KubernetesReplicationControllers = "replicationcontrollers"
 )
 
 func generateKubernetesApiBaseUrlFor(schemeHostAndPort string, namespace string, extraPathStartingWithSlash string) (*url.URL, error) {
@@ -63,6 +64,8 @@ func CanonicalKubernetesNameFromFriendlyName(friendlyName string) (string, error
 		return KubernetesNamespaces, nil
 	case "po", "pod", "pods":
 		return KubernetesPods, nil
+	case "rc", "replicationcontroller", "replicationcontrollers":
+		return KubernetesReplicationControllers, nil
 	}
 
 	return "", fmt.Errorf("cannot find Kubernetes canonical name from friendly name [%s]", friendlyName)