Added loading CPU and Mem for containers in pod

config/param.go

@@ -12,6 +12,11 @@ var params = []*Param{
 		Val:   "state",
 		Label: "Container Sort Field",
 	},
+	&Param{
+		Key:   "namespace",
+		Val:   "state",
+		Label: "Kubernetes namespace for monitoring",
+	},
 }
 
 type Param struct {

connector/collector/kubernetes.go

@@ -1,13 +1,16 @@
 package collector
 
 import (
+	"time"
+
+	"k8s.io/metrics/pkg/apis/metrics/v1alpha1"
 	"k8s.io/metrics/pkg/client/clientset_generated/clientset"
 
 	"github.com/bcicen/ctop/config"
 	"github.com/bcicen/ctop/models"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/api/core/v1"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 )
 
 // Kubernetes collector
@@ -42,30 +45,20 @@ func (k *Kubernetes) Start() {
 		k.running = false
 		for {
 
-			cm, err := k.client.Metrics().PodMetricses("akozlenkov").List(metav1.ListOptions{})
+			result := &v1alpha1.PodMetrics{}
+			err := k.clientset.RESTClient().Get().AbsPath("/api/v1/namespaces/kube-system/services/http:heapster:/proxy/apis/metrics/v1alpha1/namespaces/" + config.GetVal("namespace") + "/pods/" + k.name).Do().Into(result)
+
 			if err != nil {
-				log.Errorf(">>>>>> %s here %s", k.name, err.Error())
+				log.Errorf("has error %s here %s", k.name, err.Error())
+				time.Sleep(1 * time.Second)
 				continue
 			}
-			log.Debugf(">>>> %+v", cm)
+			k.ReadCPU(result)
-			//for _, m := range cm.Containers {
+			k.ReadMem(result)
-			//	log.Debugf(">>>> %+v", m)
+			k.stream <- k.Metrics
-			//}
 		}
 	}()
 
-	//go func() {
-	//	defer close(c.stream)
-	//	for s := range stats {
-	//		c.ReadCPU(s)
-	//		c.ReadMem(s)
-	//		c.ReadNet(s)
-	//		c.ReadIO(s)
-	//		c.stream <- c.Metrics
-	//	}
-	//	log.Infof("collector stopped for container: %s", c.id)
-	//}()
-
 	k.running = true
 	log.Infof("collector started for container: %s", k.name)
 }
@@ -87,30 +80,30 @@ func (c *Kubernetes) Stop() {
 	c.done <- true
 }
 
-//
+func (k *Kubernetes) ReadCPU(metrics *v1alpha1.PodMetrics) {
-//func (c *Kubernetes) ReadCPU(stats *api.Stats) {
+	all := int64(0)
-//	ncpus := float64(len(stats.CPUStats.CPUUsage.PercpuUsage))
+	for _, c := range metrics.Containers {
-//	total := float64(stats.CPUStats.CPUUsage.TotalUsage)
+		v := c.Usage[v1.ResourceCPU]
-//	system := float64(stats.CPUStats.SystemCPUUsage)
+		all += v.Value()
-//
+	}
-//	cpudiff := total - c.lastCpu
+	if all != 0 {
-//	syscpudiff := system - c.lastSysCpu
+		k.CPUUtil = round(float64(all))
-//
+	}
-//	if c.scaleCpu {
+}
-//		c.CPUUtil = round((cpudiff / syscpudiff * 100))
-//	} else {
-//		c.CPUUtil = round((cpudiff / syscpudiff * 100) * ncpus)
-//	}
-//	c.lastCpu = total
-//	c.lastSysCpu = system
-//	c.Pids = int(stats.PidsStats.Current)
-//}
 
-//func (c *Kubernetes) ReadMem(stats *api.Stats) {
+func (k *Kubernetes) ReadMem(metrics *v1alpha1.PodMetrics) {
-//	c.MemUsage = int64(stats.MemoryStats.Usage - stats.MemoryStats.Stats.Cache)
+	all := int64(0)
-//	c.MemLimit = int64(stats.MemoryStats.Limit)
+	for _, c := range metrics.Containers {
-//	c.MemPercent = percent(float64(c.MemUsage), float64(c.MemLimit))
+		v := c.Usage[v1.ResourceMemory]
-//}
+		a, ok := v.AsInt64()
+		if ok {
+			all += a
+		}
+	}
+	k.MemUsage = all
+	k.MemLimit = int64(0)
+	//k.MemPercent = percent(float64(k.MemUsage), float64(k.MemLimit))
+}
 
 //func (c *Kubernetes) ReadNet(stats *api.Stats) {
 //	var rx, tx int64

connector/kubernetes.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	bcfg "github.com/bcicen/ctop/config"
 	"github.com/bcicen/ctop/connector/collector"
 	"github.com/bcicen/ctop/connector/manager"
 	"github.com/bcicen/ctop/container"
@@ -20,9 +21,8 @@ import (
 
 func init() { enabled["kubernetes"] = NewKubernetes }
 
-var namespace = "akozlenkov"
-
 type Kubernetes struct {
+	namespace    string
 	clientset    *kubernetes.Clientset
 	containers   map[string]*container.Container
 	needsRefresh chan string // container IDs requiring refresh
@@ -59,6 +59,7 @@ func NewKubernetes() Connector {
 		containers:   make(map[string]*container.Container),
 		needsRefresh: make(chan string, 60),
 		lock:         sync.RWMutex{},
+		namespace:    bcfg.GetVal("namespace"),
 	}
 	go k.Loop()
 	k.refreshAll()
@@ -69,7 +70,7 @@ func NewKubernetes() Connector {
 func (k *Kubernetes) watchEvents() {
 	for {
 		log.Info("kubernetes event listener starting")
-		allEvents, err := k.clientset.CoreV1().Events(namespace).List(metav1.ListOptions{})
+		allEvents, err := k.clientset.CoreV1().Events(k.namespace).List(metav1.ListOptions{})
 		if err != nil {
 			log.Error(err.Error())
 			return
@@ -152,7 +153,7 @@ func k8sPort(ports []v1.ContainerPort) string {
 }
 
 func (k *Kubernetes) inspect(id string) *v1.Pod {
-	p, err := k.clientset.CoreV1().Pods(namespace).Get(id, metav1.GetOptions{})
+	p, err := k.clientset.CoreV1().Pods(k.namespace).Get(id, metav1.GetOptions{})
 	if err != nil {
 		if _, ok := err.(*api.NoSuchContainer); !ok {
 			log.Errorf(err.Error())
@@ -178,7 +179,7 @@ func (k *Kubernetes) Get(name string) (c *container.Container, ok bool) {
 
 // Mark all container IDs for refresh
 func (k *Kubernetes) refreshAll() {
-	allPods, err := k.clientset.CoreV1().Pods(namespace).List(metav1.ListOptions{})
+	allPods, err := k.clientset.CoreV1().Pods(k.namespace).List(metav1.ListOptions{})
 	if err != nil {
 		log.Error(err.Error())
 		return

main.go

@@ -45,6 +45,7 @@ func main() {
 		invertFlag      = flag.Bool("i", false, "invert default colors")
 		scaleCpu        = flag.Bool("scale-cpu", false, "show cpu as % of system total")
 		connectorFlag   = flag.String("connector", "docker", "container connector to use")
+		namespace       = flag.String("n", "default", "container connector to use")
 	)
 	flag.Parse()
 
@@ -91,6 +92,7 @@ func main() {
 	if *invertFlag {
 		InvertColorMap()
 	}
+	config.Update("namespace", *namespace)
 	ui.ColorMap = ColorMap // override default colormap
 	if err := ui.Init(); err != nil {
 		panic(err)