move flags to FlagSet

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

@@ -0,0 +1,128 @@
+package collector
+
+import (
+	"time"
+
+	"k8s.io/metrics/pkg/apis/metrics/v1alpha1"
+	clientset "k8s.io/metrics/pkg/client/clientset/versioned"
+
+	"github.com/bcicen/ctop/config"
+	"github.com/bcicen/ctop/models"
+	"k8s.io/api/core/v1"
+
+	"k8s.io/client-go/kubernetes"
+)
+
+// Kubernetes collector
+type Kubernetes struct {
+	models.Metrics
+	name       string
+	client     clientset.Interface
+	clientset  *kubernetes.Clientset
+	running    bool
+	stream     chan models.Metrics
+	done       chan bool
+	lastCpu    float64
+	lastSysCpu float64
+	scaleCpu   bool
+}
+
+func NewKubernetes(client *kubernetes.Clientset, name string) *Kubernetes {
+	return &Kubernetes{
+		Metrics:   models.Metrics{},
+		name:      name,
+		client:    clientset.New(client.RESTClient()),
+		clientset: client,
+		scaleCpu:  config.GetSwitchVal("scaleCpu"),
+	}
+}
+
+func (k *Kubernetes) Start() {
+	k.done = make(chan bool)
+	k.stream = make(chan models.Metrics)
+
+	go func() {
+		k.running = false
+		for {
+
+			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("has error %s here %s", k.name, err.Error())
+				time.Sleep(1 * time.Second)
+				continue
+			}
+			k.ReadCPU(result)
+			k.ReadMem(result)
+			k.stream <- k.Metrics
+		}
+	}()
+
+	k.running = true
+	log.Infof("collector started for container: %s", k.name)
+}
+
+func (c *Kubernetes) Running() bool {
+	return c.running
+}
+
+func (c *Kubernetes) Stream() chan models.Metrics {
+	return c.stream
+}
+
+func (c *Kubernetes) Logs() LogCollector {
+	return NewKubernetesLogs(c.name, c.clientset)
+}
+
+// Stop collector
+func (c *Kubernetes) Stop() {
+	c.done <- true
+}
+
+func (k *Kubernetes) ReadCPU(metrics *v1alpha1.PodMetrics) {
+	all := int64(0)
+	for _, c := range metrics.Containers {
+		v := c.Usage[v1.ResourceCPU]
+		all += v.Value()
+	}
+	if all != 0 {
+		k.CPUUtil = round(float64(all))
+	}
+}
+
+func (k *Kubernetes) ReadMem(metrics *v1alpha1.PodMetrics) {
+	all := int64(0)
+	for _, c := range metrics.Containers {
+		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
+//	for _, network := range stats.Networks {
+//		rx += int64(network.RxBytes)
+//		tx += int64(network.TxBytes)
+//	}
+//	c.NetRx, c.NetTx = rx, tx
+//}
+//
+//func (c *Kubernetes) ReadIO(stats *api.Stats) {
+//	var read, write int64
+//	for _, blk := range stats.BlkioStats.IOServiceBytesRecursive {
+//		if blk.Op == "Read" {
+//			read = int64(blk.Value)
+//		}
+//		if blk.Op == "Write" {
+//			write = int64(blk.Value)
+//		}
+//	}
+//	c.IOBytesRead, c.IOBytesWrite = read, write
+//}

connector/collector/kubernetes_logs.go

@@ -0,0 +1,78 @@
+package collector
+
+import (
+	"time"
+
+	"github.com/bcicen/ctop/models"
+	"k8s.io/client-go/kubernetes"
+)
+
+type KubernetesLogs struct {
+	id     string
+	client *kubernetes.Clientset
+	done   chan bool
+}
+
+func NewKubernetesLogs(id string, client *kubernetes.Clientset) *KubernetesLogs {
+	return &KubernetesLogs{
+		id:     id,
+		client: client,
+		done:   make(chan bool),
+	}
+}
+
+func (l *KubernetesLogs) Stream() chan models.Log {
+	//r, w := io.Pipe()
+	logCh := make(chan models.Log)
+	//ctx, cancel := context.WithCancel(context.Background())
+
+	//opts := api.LogsOptions{
+	//	Context:      ctx,
+	//	Container:    l.id,
+	//	OutputStream: w,
+	//	ErrorStream:  w,
+	//	Stdout:       true,
+	//	Stderr:       true,
+	//	Tail:         "10",
+	//	Follow:       true,
+	//	Timestamps:   true,
+	//}
+
+	//// read io pipe into channel
+	//go func() {
+	//	scanner := bufio.NewScanner(r)
+	//	for scanner.Scan() {
+	//		parts := strings.Split(scanner.Text(), " ")
+	//		ts := l.parseTime(parts[0])
+	//		logCh <- models.Log{Timestamp: ts, Message: strings.Join(parts[1:], " ")}
+	//	}
+	//}()
+
+	//// connect to container log stream
+	//go func() {
+	//	err := l.client.Logs(opts)
+	//	if err != nil {
+	//		log.Errorf("error reading container logs: %s", err)
+	//	}
+	//	log.Infof("log reader stopped for container: %s", l.id)
+	//}()
+
+	//go func() {
+	//	<-l.done
+	//	cancel()
+	//}()
+
+	log.Infof("log reader started for container: %s", l.id)
+	return logCh
+}
+
+func (l *KubernetesLogs) Stop() { l.done <- true }
+
+func (l *KubernetesLogs) parseTime(s string) time.Time {
+	ts, err := time.Parse("2006-01-02T15:04:05.000000000Z", s)
+	if err != nil {
+		log.Errorf("failed to parse container log: %s", err)
+		ts = time.Now()
+	}
+	return ts
+}

connector/kubernetes.go

@@ -0,0 +1,218 @@
+package connector
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"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"
+	api "github.com/fsouza/go-dockerclient"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+func init() { enabled["kubernetes"] = NewKubernetes }
+
+type Kubernetes struct {
+	namespace    string
+	clientset    *kubernetes.Clientset
+	containers   map[string]*container.Container
+	needsRefresh chan string // container IDs requiring refresh
+	lock         sync.RWMutex
+}
+
+func NewKubernetes() Connector {
+	var kubeconfig string
+	//if home := homeDir(); home != "" {
+	//	kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
+	//} else {
+	//	kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
+	//}
+	//flag.Parse()
+	kubeconfig = filepath.Join(homeDir(), ".kube", "config")
+
+	// use the current context in kubeconfig
+	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
+	if err != nil {
+		log.Error(err.Error())
+		return nil
+	}
+
+	// create the clientset
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		log.Error(err.Error())
+		return nil
+	}
+
+	// init docker client
+	k := &Kubernetes{
+		clientset:    clientset,
+		containers:   make(map[string]*container.Container),
+		needsRefresh: make(chan string, 60),
+		lock:         sync.RWMutex{},
+		namespace:    bcfg.GetVal("namespace"),
+	}
+	go k.Loop()
+	k.refreshAll()
+	go k.watchEvents()
+	return k
+}
+
+func (k *Kubernetes) watchEvents() {
+	for {
+		log.Info("kubernetes event listener starting")
+		allEvents, err := k.clientset.CoreV1().Events(k.namespace).List(metav1.ListOptions{})
+		if err != nil {
+			log.Error(err.Error())
+			return
+		}
+
+		for _, e := range allEvents.Items {
+			if e.Kind != "pod" {
+				continue
+			}
+
+			actionName := strings.Split(e.Action, ":")[0]
+
+			switch actionName {
+			case "start", "die", "pause", "unpause", "health_status":
+				log.Debugf("handling docker event: action=%s id=%s", e.Action, e.UID)
+				k.needsRefresh <- e.Name
+			case "destroy":
+				log.Debugf("handling docker event: action=%s id=%s", e.Action, e.UID)
+				k.delByID(e.Name)
+			default:
+				log.Debugf("handling docker event: %v", e)
+				k.needsRefresh <- e.Name
+			}
+		}
+		time.Sleep(1 * time.Second)
+	}
+}
+func (k *Kubernetes) Loop() {
+	for id := range k.needsRefresh {
+		c := k.MustGet(id)
+		k.refresh(c)
+	}
+}
+
+// Get a single container, creating one anew if not existing
+func (k *Kubernetes) MustGet(name string) *container.Container {
+	c, ok := k.Get(name)
+	// append container struct for new containers
+	if !ok {
+		// create collector
+		collector := collector.NewKubernetes(k.clientset, name)
+		// create manager
+		manager := manager.NewKubernetes(k.clientset, name)
+		// create container
+		c = container.New(name, collector, manager)
+		k.lock.Lock()
+		k.containers[name] = c
+		k.lock.Unlock()
+	}
+	return c
+}
+
+func (k *Kubernetes) refresh(c *container.Container) {
+	insp := k.inspect(c.Id)
+	// remove container if no longer exists
+	if insp == nil {
+		k.delByID(c.Id)
+		return
+	}
+	c.SetMeta("name", insp.Name)
+	if len(insp.Spec.Containers) >= 1 {
+		c.SetMeta("image", insp.Spec.Containers[0].Image)
+		c.SetMeta("ports", k8sPort(insp.Spec.Containers[0].Ports))
+		for _, env := range insp.Spec.Containers[0].Env {
+			c.SetMeta("[ENV-VAR]", env.Name+"="+env.Value)
+		}
+	}
+	c.SetMeta("IPs", insp.Status.PodIP)
+	c.SetMeta("created", insp.CreationTimestamp.Format("Mon Jan 2 15:04:05 2006"))
+	c.SetMeta("health", string(insp.Status.Phase))
+	c.SetState("running")
+}
+
+func k8sPort(ports []v1.ContainerPort) string {
+	str := []string{}
+	for _, p := range ports {
+		str = append(str, fmt.Sprintf("%s:%d -> %d", p.HostIP, p.HostPort, p.ContainerPort))
+	}
+	return strings.Join(str, "\n")
+}
+
+func (k *Kubernetes) inspect(id string) *v1.Pod {
+	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())
+		}
+	}
+	return p
+}
+
+// Remove containers by ID
+func (k *Kubernetes) delByID(name string) {
+	k.lock.Lock()
+	delete(k.containers, name)
+	k.lock.Unlock()
+	log.Infof("removed dead container: %s", name)
+}
+
+func (k *Kubernetes) Get(name string) (c *container.Container, ok bool) {
+	k.lock.Lock()
+	c, ok = k.containers[name]
+	k.lock.Unlock()
+	return
+}
+
+// Mark all container IDs for refresh
+func (k *Kubernetes) refreshAll() {
+	allPods, err := k.clientset.CoreV1().Pods(k.namespace).List(metav1.ListOptions{})
+	if err != nil {
+		log.Error(err.Error())
+		return
+	}
+
+	for _, pod := range allPods.Items {
+		c := k.MustGet(pod.Name)
+		c.SetMeta("uid", string(pod.UID))
+		c.SetMeta("name", pod.Name)
+		if pod.Initializers != nil && pod.Initializers.Result != nil {
+			c.SetState(pod.Initializers.Result.Status)
+		} else {
+			c.SetState(string(pod.Status.Phase))
+		}
+		k.needsRefresh <- c.Id
+	}
+}
+
+func (k *Kubernetes) All() (containers container.Containers) {
+	k.lock.Lock()
+	for _, c := range k.containers {
+		containers = append(containers, c)
+	}
+
+	containers.Sort()
+	containers.Filter()
+	k.lock.Unlock()
+	return containers
+}
+
+func homeDir() string {
+	if h := os.Getenv("HOME"); h != "" {
+		return h
+	}
+	return os.Getenv("USERPROFILE") // windows
+}

connector/manager/kubernetes.go

@@ -0,0 +1,64 @@
+package manager
+
+import (
+	"k8s.io/client-go/kubernetes"
+)
+
+type Kubernetes struct {
+	id     string
+	client *kubernetes.Clientset
+}
+
+func NewKubernetes(client *kubernetes.Clientset, id string) *Kubernetes {
+	return &Kubernetes{
+		id:     id,
+		client: client,
+	}
+}
+
+func (dc *Kubernetes) Start() error {
+	//c, err := dc.client.InspectContainer(dc.id)
+	//if err != nil {
+	//	return fmt.Errorf("cannot inspect container: %v", err)
+	//}
+
+	//if err := dc.client.StartContainer(c.ID, c.HostConfig); err != nil {
+	//	return fmt.Errorf("cannot start container: %v", err)
+	//}
+	return nil
+}
+
+func (dc *Kubernetes) Stop() error {
+	//if err := dc.client.StopContainer(dc.id, 3); err != nil {
+	//	return fmt.Errorf("cannot stop container: %v", err)
+	//}
+	return nil
+}
+
+func (dc *Kubernetes) Remove() error {
+	//if err := dc.client.RemoveContainer(api.RemoveContainerOptions{ID: dc.id}); err != nil {
+	//	return fmt.Errorf("cannot remove container: %v", err)
+	//}
+	return nil
+}
+
+func (dc *Kubernetes) Pause() error {
+	//if err := dc.client.PauseContainer(dc.id); err != nil {
+	//	return fmt.Errorf("cannot pause container: %v", err)
+	//}
+	return nil
+}
+
+func (dc *Kubernetes) Unpause() error {
+	//if err := dc.client.UnpauseContainer(dc.id); err != nil {
+	//	return fmt.Errorf("cannot unpause container: %v", err)
+	//}
+	return nil
+}
+
+func (dc *Kubernetes) Restart() error {
+	//if err := dc.client.RestartContainer(dc.id, 3); err != nil {
+	//	return fmt.Errorf("cannot restart container: %v", err)
+	//}
+	return nil
+}

cwidgets/compact/status.go

@@ -67,11 +67,11 @@ func (s *Status) SetHealth(val string) {
 	color := ui.ColorDefault
 
 	switch val {
-	case "healthy":
+	case "healthy", "Succeeded":
 		color = ui.ThemeAttr("status.ok")
-	case "unhealthy":
+	case "unhealthy", "Failed", "Unknown":
 		color = ui.ThemeAttr("status.danger")
-	case "starting":
+	case "starting", "Pending", "Running":
 		color = ui.ThemeAttr("status.warn")
 	}
 

go.mod

@@ -13,29 +13,49 @@ require (
 	github.com/docker/go-connections v0.0.0-20170301234100-a2afab980204 // indirect
 	github.com/docker/go-units v0.3.2 // indirect
 	github.com/fsouza/go-dockerclient v0.0.0-20170307141636-318513eb1ab2
+	github.com/ghodss/yaml v1.0.0 // indirect
 	github.com/gizak/termui v2.3.0+incompatible
 	github.com/godbus/dbus v0.0.0-20151105175453-c7fdd8b5cd55 // indirect
-	github.com/golang/protobuf v0.0.0-20170712042213-0a4f71a498b7 // indirect
+	github.com/gogo/protobuf v1.1.1 // indirect
+	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b // indirect
+	github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c // indirect
+	github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf // indirect
+	github.com/googleapis/gnostic v0.2.0 // indirect
+	github.com/gregjones/httpcache v0.0.0-20181110185634-c63ab54fda8f // indirect
 	github.com/hashicorp/go-cleanhttp v0.0.0-20170211013415-3573b8b52aa7 // indirect
+	github.com/imdario/mergo v0.3.6 // indirect
 	github.com/jgautheron/codename-generator v0.0.0-20150829203204-16d037c7cc3c
-	github.com/kr/pretty v0.1.0 // indirect
-	github.com/maruel/panicparse v0.0.0-20170227222818-25bcac0d793c // indirect
-	github.com/maruel/ut v1.0.0 // indirect
+	github.com/json-iterator/go v1.1.5 // indirect
 	github.com/mattn/go-runewidth v0.0.0-20170201023540-14207d285c6c // indirect
 	github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/nsf/termbox-go v0.0.0-20180303152453-e2050e41c884
 	github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d
 	github.com/op/go-logging v0.0.0-20160211212156-b2cb9fa56473
 	github.com/opencontainers/runc v0.1.1
+	github.com/peterbourgon/diskv v2.0.1+incompatible // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/seccomp/libseccomp-golang v0.0.0-20150813023252-1b506fc7c24e // indirect
+	github.com/spf13/pflag v1.0.3 // indirect
 	github.com/stretchr/testify v1.2.2 // indirect
 	github.com/syndtr/gocapability v0.0.0-20150716010906-2c00daeb6c3b // indirect
 	github.com/vishvananda/netlink v0.0.0-20150820014904-1e2e08e8a2dc // indirect
 	github.com/vishvananda/netns v0.0.0-20180720170159-13995c7128cc // indirect
-	golang.org/x/net v0.0.0-20170308210134-a6577fac2d73 // indirect
+	golang.org/x/crypto v0.0.0-20181127143415-eb0de9b17e85 // indirect
+	golang.org/x/oauth2 v0.0.0-20181128211412-28207608b838 // indirect
 	golang.org/x/sync v0.0.0-20181108010431-42b317875d0f // indirect
-	golang.org/x/sys v0.0.0-20170308153327-99f16d856c98 // indirect
+	golang.org/x/sys v0.0.0-20181128092732-4ed8d59d0b35 // indirect
+	golang.org/x/time v0.0.0-20181108054448-85acf8d2951c // indirect
+	google.golang.org/appengine v1.3.0 // indirect
+	gopkg.in/inf.v0 v0.9.1 // indirect
+	gopkg.in/yaml.v2 v2.2.2 // indirect
+	k8s.io/api v0.0.0-20181130031204-d04500c8c3dd
+	k8s.io/apimachinery v0.0.0-20181201231028-18a5ff3097b4
+	k8s.io/client-go v9.0.0+incompatible
+	k8s.io/klog v0.1.0 // indirect
+	k8s.io/metrics v0.0.0-20181121073115-d8618695b08f
+	sigs.k8s.io/yaml v1.1.0 // indirect
 )
 
 replace github.com/gizak/termui => github.com/bcicen/termui v0.0.0-20180326052246-4eb80249d3f5

main.go

@@ -29,24 +29,24 @@ var (
 	status *widgets.StatusLine
 
 	versionStr = fmt.Sprintf("ctop version %v, build %v %v", version, build, goVersion)
+
+	fs              = flag.NewFlagSet("ctop", flag.ExitOnError)
+	versionFlag     = fs.Bool("version", false, "output version information and exit")
+	helpFlag        = fs.Bool("h", false, "display this help dialog")
+	filterFlag      = fs.String("f", "", "filter containers")
+	activeOnlyFlag  = fs.Bool("a", false, "show active containers only")
+	sortFieldFlag   = fs.String("s", "", "select container sort field")
+	reverseSortFlag = fs.Bool("r", false, "reverse container sort order")
+	invertFlag      = fs.Bool("i", false, "invert default colors")
+	scaleCpu        = fs.Bool("scale-cpu", false, "show cpu as % of system total")
+	connectorFlag   = fs.String("connector", "docker", "container connector to use")
+	namespaceFlag   = fs.String("n", "default", "Kubernetes namespace for monitoring")
 )
 
 func main() {
 	defer panicExit()
 
-	// parse command line arguments
-	var (
-		versionFlag     = flag.Bool("v", false, "output version information and exit")
-		helpFlag        = flag.Bool("h", false, "display this help dialog")
-		filterFlag      = flag.String("f", "", "filter containers")
-		activeOnlyFlag  = flag.Bool("a", false, "show active containers only")
-		sortFieldFlag   = flag.String("s", "", "select container sort field")
-		reverseSortFlag = flag.Bool("r", false, "reverse container sort order")
-		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")
-	)
-	flag.Parse()
+	fs.Parse(os.Args[1:])
 
 	if *versionFlag {
 		fmt.Println(versionStr)
@@ -91,6 +91,7 @@ func main() {
 	if *invertFlag {
 		InvertColorMap()
 	}
+	config.Update("namespace", *namespaceFlag)
 	ui.ColorMap = ColorMap // override default colormap
 	if err := ui.Init(); err != nil {
 		panic(err)
@@ -149,7 +150,7 @@ options:
 
 func printHelp() {
 	fmt.Println(helpMsg)
-	flag.PrintDefaults()
+	fs.PrintDefaults()
 	fmt.Printf("\navailable connectors: ")
 	fmt.Println(strings.Join(connector.Enabled(), ", "))
 }