add api to get cpu/memory realtime data

Makefile

@@ -9,7 +9,7 @@ TRAG.Version:=$(TRAG.Gopkg)/pkg/version
 DOCKER_TAGS=latest
 RUN_IN_DOCKER:=docker run -it -v `pwd`:/go/src/$(TRAG.Gopkg) -v `pwd`/tmp/cache:/root/.cache/go-build  -w /go/src/$(TRAG.Gopkg) -e GOBIN=/go/src/$(TRAG.Gopkg)/tmp/bin -e USER_ID=`id -u` -e GROUP_ID=`id -g` kubesphere/kubesphere-builder
 GO_FMT:=goimports -l -w -e -local=kubesphere -srcdir=/go/src/$(TRAG.Gopkg)
-GO_FILES:=./cmd ./test ./pkg
+GO_FILES:=./cmd ./pkg
 
 define get_diff_files
     $(eval DIFF_FILES=$(shell git diff --name-only --diff-filter=ad | grep -E "^(test|cmd|pkg)/.+\.go"))

pkg/apis/v1alpha/install.go

@@ -21,6 +21,7 @@ import (
 	"kubesphere.io/kubesphere/pkg/apis/v1alpha/kubeconfig"
 	"kubesphere.io/kubesphere/pkg/apis/v1alpha/kubectl"
 	"kubesphere.io/kubesphere/pkg/apis/v1alpha/nodes"
+	"kubesphere.io/kubesphere/pkg/apis/v1alpha/pods"
 	"kubesphere.io/kubesphere/pkg/apis/v1alpha/registries"
 	"kubesphere.io/kubesphere/pkg/apis/v1alpha/storage"
 	"kubesphere.io/kubesphere/pkg/apis/v1alpha/volumes"
@@ -29,7 +30,7 @@ import (
 func init() {
 
 	ws := new(restful.WebService)
-	ws.Path("/api/v1alpha")
+	ws.Path("/api/v1alpha1")
 
 	nodes.Register(ws, "/nodes")
 	kubeconfig.Register(ws, "/namespaces/{namespace}/kubeconfig")
@@ -37,7 +38,8 @@ func init() {
 	registries.Register(ws, "/registries")
 	storage.Register(ws, "/storage")
 	volumes.Register(ws, "/volumes")
-
+	nodes.Register(ws, "/nodes")
+	pods.Register(ws)
 	// add webservice to default container
 	restful.Add(ws)
 

pkg/apis/v1alpha/nodes/nodes_handler.go

@@ -18,16 +18,49 @@ package nodes
 
 import (
 	"github.com/emicklei/go-restful"
-	"kubesphere.io/kubesphere/pkg/models"
+
+	"kubesphere.io/kubesphere/pkg/constants"
 	"kubesphere.io/kubesphere/pkg/filter/route"
+	"kubesphere.io/kubesphere/pkg/models"
 )
 
 func Register(ws *restful.WebService, subPath string) {
 
-	ws.Route(ws.GET(subPath).To(models.HandleNodes).Filter(route.RouteLogging)).
+	ws.Route(ws.GET(subPath).To(handleNodes).Filter(route.RouteLogging)).
+		Consumes(restful.MIME_JSON, restful.MIME_XML).
+		Produces(restful.MIME_JSON)
+	ws.Route(ws.GET(subPath+"/{nodename}").To(handleSingleNode).Filter(route.RouteLogging)).
 		Consumes(restful.MIME_JSON, restful.MIME_XML).
 		Produces(restful.MIME_JSON)
-
 }
 
+func handleNodes(request *restful.Request, response *restful.Response) {
+	var result constants.ResultMessage
+	var resultNodes []models.ResultNode
+	var resultNode models.ResultNode
 
+	nodes := models.GetNodes()
+
+	for _, node := range nodes {
+
+		resultNode = models.FormatNodeMetrics(node)
+		resultNodes = append(resultNodes, resultNode)
+
+	}
+
+	result.Data = resultNodes
+	response.WriteAsJson(result)
+}
+
+func handleSingleNode(request *restful.Request, response *restful.Response) {
+	nodeName := request.PathParameter("nodename")
+	var result constants.ResultMessage
+	var resultNodes []models.ResultNode
+	var resultNode models.ResultNode
+
+	resultNode = models.FormatNodeMetrics(nodeName)
+	resultNodes = append(resultNodes, resultNode)
+
+	result.Data = resultNodes
+	response.WriteAsJson(result)
+}

pkg/apis/v1alpha/pods/pods_handler.go

@@ -0,0 +1,69 @@
+/*
+Copyright 2018 The KubeSphere Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package pods
+
+import (
+	"github.com/emicklei/go-restful"
+
+	"kubesphere.io/kubesphere/pkg/filter/route"
+
+	"kubesphere.io/kubesphere/pkg/constants"
+	"kubesphere.io/kubesphere/pkg/models"
+
+)
+
+func Register(ws *restful.WebService) {
+
+	ws.Route(ws.GET("/pods").To(handleAllPods).Filter(route.RouteLogging)).
+		Consumes(restful.MIME_JSON, restful.MIME_XML).
+		Produces(restful.MIME_JSON)
+
+	ws.Route(ws.GET("/namespaces/{namespace}/pods").To(handlePodsUnderNameSpace).Filter(route.RouteLogging)).
+		Consumes(restful.MIME_JSON, restful.MIME_XML).
+		Produces(restful.MIME_JSON)
+}
+
+func handleAllPods(request *restful.Request, response *restful.Response) {
+	var result constants.ResultMessage
+	var resultNameSpaces []models.ResultNameSpace
+	var resultNameSpace models.ResultNameSpace
+
+	namespaces := models.GetNameSpaces()
+
+	for _, namespace := range namespaces {
+
+		resultNameSpace = models.FormatNameSpaceMetrics(namespace)
+		resultNameSpaces = append(resultNameSpaces, resultNameSpace)
+
+	}
+
+	result.Data = resultNameSpaces
+	response.WriteAsJson(result)
+}
+
+func handlePodsUnderNameSpace(request *restful.Request, response *restful.Response) {
+	var result constants.ResultMessage
+	var resultNameSpaces []models.ResultNameSpace
+	var resultNameSpace models.ResultNameSpace
+
+	resultNameSpace = models.FormatNameSpaceMetrics(request.PathParameter("namespace"))
+
+	resultNameSpaces = append(resultNameSpaces, resultNameSpace)
+
+	result.Data = resultNameSpaces
+	response.WriteAsJson(result)
+}

pkg/client/heapsterclient.go

@@ -17,26 +17,39 @@ limitations under the License.
 package client
 
 import (
-	"fmt"
 	"net/http"
 
-	"os"
+	"github.com/golang/glog"
 	"io/ioutil"
+	"os"
 )
 
+const (
+	DefaultHeapsterScheme  = "http"
+	DefaultHeapsterService = "heapster" //"heapster"
+	DefaultHeapsterPort    = "80"       // use the first exposed port on the service
+)
+
+var (
+	prefix = "/api/v1/model"
+)
 
 func GetHeapsterMetrics(url string) string {
-	response, err := http.Get(url)
+	//glog.Info("Querying data from " + DefaultHeapsterScheme + "://" + DefaultHeapsterService + ":" + DefaultHeapsterPort + prefix + url)
+	response, err := http.Get(DefaultHeapsterScheme + "://" + DefaultHeapsterService + ":" + DefaultHeapsterPort + prefix + url)
 	if err != nil {
-		fmt.Printf("%s", err)
+		glog.Error(err)
 		os.Exit(1)
 	} else {
 		defer response.Body.Close()
+
 		contents, err := ioutil.ReadAll(response.Body)
+
 		if err != nil {
-			fmt.Printf("%s", err)
+			glog.Error(err)
 			os.Exit(1)
 		}
+
 		return string(contents)
 	}
 	return ""

pkg/models/nodes.go

@@ -17,33 +17,128 @@ limitations under the License.
 package models
 
 import (
-	"github.com/emicklei/go-restful"
+	"encoding/json"
 
-	"net/http"
+	"strings"
 
 	"github.com/golang/glog"
+
 	"kubesphere.io/kubesphere/pkg/client"
+
+	ksutil "kubesphere.io/kubesphere/pkg/util"
+
+	"fmt"
+	"strconv"
 )
 
-type ResultMessage struct {
+type ResultNodes struct {
-	Ret  int         `json:"ret"`
+	Nodes []ResultNode `json:"nodes"`
-	Msg  string      `json:"msg"`
+}
-	Data interface{} `json:"data"`
+type ResultNode struct {
+	NodeName string       `json:"node_name"`
+	CPU      []CPUNode    `json:"cpu"`
+	Memory   []MemoryNode `json:"memory"`
 }
 
-func HandleNodes(request *restful.Request, response *restful.Response) {
+type CPUNode struct {
-
+	TimeStamp      string `json:"timestamp"`
-	var result ResultMessage
+	UsedCPU        string `json:"used_cpu"`
-
+	TotalCPU       string `json:"total_cpu"`
-
+	CPUUtilization string `json:"cpu_utilization"`
-	data := make(map[string]string)
+}
-
+
-
+type MemoryNode struct {
-	data["output"] = client.GetHeapsterMetrics("http://139.198.0.79/api/monitor/v1/model/namespaces/kube-system/pods/qingcloud-volume-provisioner-i-o5pmakm7/metrics/cpu/request")
+	TimeStamp         string `json:"timestamp"`
-
+	UsedMemory        string `json:"used_mem"`
-	result.Data = data
+	TotalMemory       string `json:"total_mem"`
-	result.Ret = http.StatusOK
+	MemoryUtilization string `json:"mem_utilization"`
-	result.Msg = "success"
+}
-	glog.Infoln(result)
+
-	response.WriteAsJson(result)
+/*
+Get all nodes in default cluster
+*/
+func GetNodes() []string {
+	nodesList := client.GetHeapsterMetrics("/nodes")
+	var nodes []string
+	dec := json.NewDecoder(strings.NewReader(nodesList))
+	err := dec.Decode(&nodes)
+	if err != nil {
+		glog.Error(err)
+	}
+	return nodes
+}
+
+/*
+Format cpu/memory data for specified node
+*/
+func FormatNodeMetrics(nodeName string) ResultNode {
+	var resultNode ResultNode
+	var nodeCPUMetrics []CPUNode
+	var nodeMemMetrics []MemoryNode
+	var cpuMetrics CPUNode
+	var memMetrics MemoryNode
+	var total_cpu float64
+	var total_mem float64
+
+	cpuNodeAllocated := client.GetHeapsterMetrics("/nodes/" + nodeName + "/metrics/cpu/node_allocatable")
+	if cpuNodeAllocated != "" {
+		var err error
+		total_cpu, err = strconv.ParseFloat(ksutil.JsonRawMessage(cpuNodeAllocated).Find("metrics").ToList()[0].Find("value").ToString(), 64)
+		if err == nil {
+			total_cpu = total_cpu / 1000
+		}
+	}
+
+	cpuUsageRate := client.GetHeapsterMetrics("/nodes/" + nodeName + "/metrics/cpu/usage_rate")
+	if cpuUsageRate != "" {
+		metrics := ksutil.JsonRawMessage(cpuUsageRate).Find("metrics").ToList()
+
+		for _, metric := range metrics {
+			timestamp := metric.Find("timestamp")
+			cpu_utilization, _ := strconv.ParseFloat(metric.Find("value").ToString(), 64)
+			cpuMetrics.TimeStamp = timestamp.ToString()
+			cpuMetrics.TotalCPU = fmt.Sprintf("%.1f", total_cpu)
+			cpuMetrics.CPUUtilization = fmt.Sprintf("%.3f", cpu_utilization/1000)
+			cpuMetrics.UsedCPU = fmt.Sprintf("%.1f", total_cpu*cpu_utilization/1000)
+
+			glog.Info("node " + nodeName + " has total cpu " + fmt.Sprintf("%.1f", total_cpu) + " CPU utilization " + fmt.Sprintf("%.3f", cpu_utilization/1000) + " at time" + timestamp.ToString())
+			nodeCPUMetrics = append(nodeCPUMetrics, cpuMetrics)
+		}
+
+	}
+
+	memNodeAllocated := client.GetHeapsterMetrics("/nodes/" + nodeName + "/metrics/memory/node_allocatable")
+	var total_mem_bytes, used_mem_bytes float64
+	if memNodeAllocated != "" {
+		var err error
+		total_mem_bytes, err = strconv.ParseFloat(ksutil.JsonRawMessage(memNodeAllocated).Find("metrics").ToList()[0].Find("value").ToString(), 64)
+		if err == nil {
+			total_mem = total_mem_bytes / 1024 / 1024 / 1024
+		}
+	}
+
+	memUsage := client.GetHeapsterMetrics("/nodes/" + nodeName + "/metrics/memory/usage")
+	if memUsage != "" {
+		metrics := ksutil.JsonRawMessage(memUsage).Find("metrics").ToList()
+
+		for _, metric := range metrics {
+			timestamp := metric.Find("timestamp")
+			used_mem_bytes, _ = strconv.ParseFloat(metric.Find("value").ToString(), 64)
+			used_mem := used_mem_bytes / 1024 / 1024 / 1024
+
+			memMetrics.TimeStamp = timestamp.ToString()
+			memMetrics.TotalMemory = fmt.Sprintf("%.1f", total_mem)
+			memMetrics.UsedMemory = fmt.Sprintf("%.1f", used_mem)
+			memMetrics.MemoryUtilization = fmt.Sprintf("%.3f", used_mem_bytes/total_mem_bytes)
+			glog.Info("node " + nodeName + " has total mem " + fmt.Sprintf("%.1f", total_mem) + " mem utilization " + fmt.Sprintf("%.3f", used_mem_bytes/total_mem_bytes) + " at time" + timestamp.ToString())
+			nodeMemMetrics = append(nodeMemMetrics, memMetrics)
+		}
+	}
+
+	resultNode.NodeName = nodeName
+
+	resultNode.CPU = nodeCPUMetrics
+	resultNode.Memory = nodeMemMetrics
+
+	return resultNode
 }

pkg/models/pods.go

@@ -0,0 +1,196 @@
+package models
+
+import (
+	"encoding/json"
+
+	"strings"
+
+	"github.com/golang/glog"
+
+	"kubesphere.io/kubesphere/pkg/client"
+
+	ksutil "kubesphere.io/kubesphere/pkg/util"
+
+	"fmt"
+	"strconv"
+)
+
+type ResultNameSpaces struct {
+	NameSpaces []ResultNameSpace `json:"namespaces"`
+}
+type ResultNameSpace struct {
+	NameSpace string      `json:"namespace"`
+	PodsCount string      `json:"pods_count"`
+	Pods      []ResultPod `json:"pods"`
+}
+type ResultPod struct {
+	PodName       string      `json:"pod_name"`
+	CPURequest    string      `json:"cpu_request"`
+	CPULimit      string      `json:"cpu_limit"`
+	MemoryRequest string      `json:"mem_request"`
+	MemoryLimit   string      `json:"mem_limit"`
+	CPU           []CPUPod    `json:"cpu"`
+	Memory        []MemoryPod `json:"memory"`
+}
+type CPUPod struct {
+	TimeStamp      string `json:"timestamp"`
+	UsedCPU        string `json:"used_cpu"`
+	CPUUtilization string `json:"cpu_utilization"`
+}
+
+type MemoryPod struct {
+	TimeStamp         string `json:"timestamp"`
+	UsedMemory        string `json:"used_mem"`
+	MemoryUtilization string `json:"mem_utilization"`
+}
+
+/*
+Get all namespaces in default cluster
+*/
+func GetNameSpaces() []string {
+	namespacesList := client.GetHeapsterMetrics("/namespaces")
+	var namespaces []string
+	dec := json.NewDecoder(strings.NewReader(namespacesList))
+	err := dec.Decode(&namespaces)
+	if err != nil {
+		glog.Error(err)
+	}
+	return namespaces
+}
+
+/*
+Get all pods under specified namespace in default cluster
+*/
+func GetPods(namespace string) []string {
+	fmt.Println(namespace)
+	podsList := client.GetHeapsterMetrics("/namespaces/" + namespace + "/pods")
+	var pods []string
+	dec := json.NewDecoder(strings.NewReader(podsList))
+	err := dec.Decode(&pods)
+	if err != nil {
+		glog.Error(err)
+	}
+	return pods
+}
+
+func FormatNameSpaceMetrics(namespace string) ResultNameSpace {
+	var resultNameSpace ResultNameSpace
+	var resultPods []ResultPod
+	var resultPod ResultPod
+
+	pods := GetPods(namespace)
+
+	resultNameSpace.NameSpace = namespace
+	resultNameSpace.PodsCount = strconv.Itoa(len(pods))
+
+	for _, pod := range pods {
+		resultPod = FormatPodMetrics(namespace, pod)
+		resultPods = append(resultPods, resultPod)
+	}
+	resultNameSpace.Pods = resultPods
+	return resultNameSpace
+}
+
+func FormatPodMetrics(namespace, pod string) ResultPod {
+	var resultPod ResultPod
+	var podCPUMetrics []CPUPod
+	var podMemMetrics []MemoryPod
+	var cpuMetrics CPUPod
+	var memMetrics MemoryPod
+
+	resultPod.PodName = pod
+	cpuRequest := client.GetHeapsterMetrics("/namespaces/" + namespace + "/pods/" + pod + "/metrics/cpu/request")
+	cpuRequest = ksutil.JsonRawMessage(cpuRequest).Find("metrics").ToList()[0].Find("value").ToString()
+	if cpuRequest != "" && cpuRequest != "0" {
+		resultPod.CPURequest = cpuRequest
+	} else {
+		resultPod.CPURequest = "inf"
+	}
+
+	cpuLimit := client.GetHeapsterMetrics("/namespaces/" + namespace + "/pods/" + pod + "/metrics/cpu/limit")
+	cpuLimit = ksutil.JsonRawMessage(cpuLimit).Find("metrics").ToList()[0].Find("value").ToString()
+	if cpuLimit != "" && cpuLimit != "0" {
+		resultPod.CPULimit = cpuLimit
+	} else {
+		resultPod.CPULimit = "inf"
+	}
+	memoryRequest := client.GetHeapsterMetrics("/namespaces/" + namespace + "/pods/" + pod + "/metrics/memory/request")
+	resultPod.MemoryRequest = convertMemory(memoryRequest)
+
+	memoryLimit := client.GetHeapsterMetrics("/namespaces/" + namespace + "/pods/" + pod + "/metrics/memory/limit")
+	resultPod.MemoryLimit = convertMemory(memoryLimit)
+
+	cpuUsageRate := client.GetHeapsterMetrics("/namespaces/" + namespace + "/pods/" + pod + "/metrics/cpu/usage_rate")
+	if cpuUsageRate != "" {
+		metrics := ksutil.JsonRawMessage(cpuUsageRate).Find("metrics").ToList()
+
+		for _, metric := range metrics {
+			timestamp := metric.Find("timestamp")
+			cpu_utilization, _ := strconv.ParseFloat(metric.Find("value").ToString(), 64)
+			cpuMetrics.TimeStamp = timestamp.ToString()
+			cpuMetrics.CPUUtilization = fmt.Sprintf("%.3f", cpu_utilization/1000)
+			if resultPod.CPULimit != "inf" {
+				cpu_limit, _ := strconv.ParseFloat(resultPod.CPULimit, 64)
+				cpuMetrics.UsedCPU = fmt.Sprintf("%.1f", cpu_limit*cpu_utilization/1000)
+			} else {
+				cpuMetrics.UsedCPU = "inf"
+			}
+			glog.Info("pod " + pod + " has limit cpu " + resultPod.CPULimit + " CPU utilization " + fmt.Sprintf("%.3f", cpu_utilization/1000) + " at time" + timestamp.ToString())
+			podCPUMetrics = append(podCPUMetrics, cpuMetrics)
+		}
+
+	}
+
+	resultPod.CPU = podCPUMetrics
+
+	var used_mem_bytes float64
+
+	memUsage := client.GetHeapsterMetrics("/namespaces/" + namespace + "/pods/" + pod + "/metrics/memory/usage")
+	if memUsage != "" {
+		metrics := ksutil.JsonRawMessage(memUsage).Find("metrics").ToList()
+
+		for _, metric := range metrics {
+			timestamp := metric.Find("timestamp")
+			used_mem_bytes, _ = strconv.ParseFloat(metric.Find("value").ToString(), 64)
+			used_mem := used_mem_bytes / 1024 / 1024
+			memMetrics.TimeStamp = timestamp.ToString()
+			memMetrics.UsedMemory = fmt.Sprintf("%.1f", used_mem)
+			if resultPod.MemoryLimit != "inf" {
+				mem_limit, _ := strconv.ParseFloat(resultPod.MemoryLimit, 64)
+				memMetrics.MemoryUtilization = fmt.Sprintf("%.3f", used_mem/mem_limit)
+			} else {
+				memMetrics.MemoryUtilization = "inf"
+			}
+
+			glog.Info("pod " + pod + " has limit mem " + resultPod.MemoryLimit + " mem utilization " + memMetrics.MemoryUtilization + " at time" + timestamp.ToString())
+			podMemMetrics = append(podMemMetrics, memMetrics)
+		}
+	}
+
+	resultPod.Memory = podMemMetrics
+	return resultPod
+}
+
+func convertMemory(memBytes string) string {
+	var mem string
+
+	if memBytes != "" {
+		memMetric := ksutil.JsonRawMessage(memBytes).Find("metrics").ToList()[0].Find("value").ToString()
+
+		if memMetric != "" && memMetric != "0" {
+
+			memBytes, error := strconv.ParseFloat(memMetric, 64)
+			if error == nil {
+				mem = fmt.Sprintf("%.3f", memBytes/1024/1024)
+			} else {
+				mem = "inf"
+			}
+		} else {
+			mem = "inf"
+		}
+
+	} else {
+		mem = "inf"
+	}
+	return mem
+}

pkg/util/util.go

@@ -0,0 +1,56 @@
+/*
+Copyright 2018 The KubeSphere Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import (
+	"encoding/json"
+	"github.com/golang/glog"
+	"strings"
+)
+
+type JsonRawMessage []byte
+
+func (m JsonRawMessage) Find(key string) JsonRawMessage {
+	var objmap map[string]json.RawMessage
+	err := json.Unmarshal(m, &objmap)
+	if err != nil {
+		glog.Errorf("Resolve JSON Key failed, find key =%s, err=%s",
+			key, err)
+		return nil
+	}
+	return JsonRawMessage(objmap[key])
+}
+
+func (m JsonRawMessage) ToList() []JsonRawMessage {
+	var lists []json.RawMessage
+	err := json.Unmarshal(m, &lists)
+	if err != nil {
+		glog.Errorf("Resolve JSON List failed, err=%s",
+			err)
+		return nil
+	}
+	var res []JsonRawMessage
+	for _, v := range lists {
+		res = append(res, JsonRawMessage(v))
+	}
+	return res
+}
+
+func (m JsonRawMessage) ToString() string {
+	res := strings.Replace(string(m[:]), "\"", "", -1)
+	return res
+}