Adding Per core cpu utilization from to heasper MetrciSet

metrics/sinks/poseidon/poseidon_sink.go

@@ -8,6 +8,7 @@ import (
 	"k8s.io/heapster/metrics/core"
 	"k8s.io/heapster/metrics/sinks/poseidon/stats"
 	"net/url"
+	"strconv"
 	"sync"
 	"time"
 )
@@ -37,16 +38,17 @@ func (sink *PoseidonSink) Stop() {
 func (sink *PoseidonSink) ProcessNodeStats(nodeMetrics *core.MetricSet) {
 	glog.V(2).Infoln("ProcessNodeStats called...")
 	nodeStats := &stats.NodeStats{
-		Hostname:       nodeMetrics.Labels[core.LabelHostname.Key],
-		Timestamp:      uint64(nodeMetrics.ScrapeTime.UnixNano()) / uint64(time.Microsecond),
-		CpuAllocatable: int64(getMetricValue(nodeMetrics, core.MetricNodeCpuAllocatable).FloatValue),
-		CpuCapacity:    int64(getMetricValue(nodeMetrics, core.MetricNodeCpuCapacity).FloatValue),
-		CpuReservation: float64(getMetricValue(nodeMetrics, core.MetricNodeCpuReservation).FloatValue),
-		CpuUtilization: float64(getMetricValue(nodeMetrics, core.MetricNodeCpuUtilization).FloatValue),
-		MemAllocatable: int64(getMetricValue(nodeMetrics, core.MetricNodeMemoryAllocatable).FloatValue / KILOBYTE),
-		MemCapacity:    int64(getMetricValue(nodeMetrics, core.MetricNodeMemoryCapacity).FloatValue / KILOBYTE),
-		MemReservation: float64(getMetricValue(nodeMetrics, core.MetricNodeMemoryReservation).FloatValue),
-		MemUtilization: float64(getMetricValue(nodeMetrics, core.MetricNodeMemoryUtilization).FloatValue),
+		Hostname:           nodeMetrics.Labels[core.LabelHostname.Key],
+		Timestamp:          uint64(nodeMetrics.ScrapeTime.UnixNano()) / uint64(time.Microsecond),
+		CpuAllocatable:     int64(getMetricValue(nodeMetrics, core.MetricNodeCpuAllocatable).FloatValue),
+		CpuCapacity:        int64(getMetricValue(nodeMetrics, core.MetricNodeCpuCapacity).FloatValue),
+		CpuReservation:     float64(getMetricValue(nodeMetrics, core.MetricNodeCpuReservation).FloatValue),
+		CpuUtilization:     float64(getMetricValue(nodeMetrics, core.MetricNodeCpuUtilization).FloatValue),
+		MemAllocatable:     int64(getMetricValue(nodeMetrics, core.MetricNodeMemoryAllocatable).FloatValue / KILOBYTE),
+		MemCapacity:        int64(getMetricValue(nodeMetrics, core.MetricNodeMemoryCapacity).FloatValue / KILOBYTE),
+		MemReservation:     float64(getMetricValue(nodeMetrics, core.MetricNodeMemoryReservation).FloatValue),
+		MemUtilization:     float64(getMetricValue(nodeMetrics, core.MetricNodeMemoryUtilization).FloatValue),
+		CpuCoreUtilization: getCoreValues(nodeMetrics),
 	}
 	glog.V(2).Infoln("Sending nodeStat to nodeStatChan", nodeStats)
 	sink.nodeStatChan <- nodeStats
@@ -171,6 +173,19 @@ func (sink *PoseidonSink) ExportData(dataBatch *core.DataBatch) {
 	}
 }
 
+func getCoreValues(metricSet *core.MetricSet) []int64 {
+	//TODO:(shiv) Assuming 128 core to be the max core per node
+	var coreUsage []int64
+	for i := 0; i < 128; i++ {
+		if value, ok := metricSet.MetricValues["cpucore/core_"+strconv.Itoa(i)]; ok {
+			coreUsage = append(coreUsage, value.IntValue)
+		} else {
+			break
+		}
+	}
+	return coreUsage
+}
+
 func getMetricValue(metricSet *core.MetricSet, metric core.Metric) core.MetricValue {
 	return metricSet.MetricValues[metric.MetricDescriptor.Name]
 }

metrics/sinks/poseidon/stats/poseidonstats.proto

@@ -40,6 +40,8 @@ message NodeStats {
   // Memory stats (fraction of total).
   double mem_reservation = 9;
   double mem_utilization = 10;
+  // per core utilization
+  repeated int64 cpu_core_utilization = 11;
 }
 
 enum NodeStatsResponseType {
@@ -91,4 +93,4 @@ message PodStatsResponse {
   PodStatsResponseType type = 1;
   string name = 2;
   string namespace = 3;
-}
+}

metrics/sources/kubelet/kubelet_client.go

@@ -145,7 +145,6 @@ func (self *KubeletClient) GetSummary(host Host) (*stats.Summary, error) {
 	if self.config != nil && self.config.EnableHttps {
 		url.Scheme = "https"
 	}
-
 	req, err := http.NewRequest("GET", url.String(), nil)
 	if err != nil {
 		return nil, err
@@ -156,6 +155,30 @@ func (self *KubeletClient) GetSummary(host Host) (*stats.Summary, error) {
 		client = http.DefaultClient
 	}
 	err = self.postRequestAndGetValue(client, req, summary)
+	self.GetCPUCoreStats(host, summary)
+	return summary, err
+}
+
+func (self *KubeletClient) GetCPUCoreStats(host Host, summary *stats.Summary) (*stats.Summary, error) {
+	url := url.URL{
+		Scheme: "http",
+		Host:   fmt.Sprintf("%s:%d", host.IP, host.Port),
+		Path:   "/stats",
+	}
+	if self.config != nil && self.config.EnableHttps {
+		url.Scheme = "https"
+	}
+	req, err := http.NewRequest("GET", url.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+	cpuCoreStats := &cadvisor.ContainerInfo{}
+	client := self.client
+	if client == nil {
+		client = http.DefaultClient
+	}
+	err = self.postRequestAndGetValue(client, req, cpuCoreStats)
+	summary.Node.CPUPerCore = &cpuCoreStats.Stats[len(cpuCoreStats.Stats)-1].Cpu
 	return summary, err
 }
 

metrics/sources/summary/summary.go

@@ -17,12 +17,14 @@ package summary
 import (
 	"fmt"
 	"net/url"
+	"strconv"
 	"time"
 
 	. "k8s.io/heapster/metrics/core"
 	"k8s.io/heapster/metrics/sources/kubelet"
 
 	"github.com/golang/glog"
+	cadvisor "github.com/google/cadvisor/info/v1"
 	"github.com/prometheus/client_golang/prometheus"
 	"k8s.io/apimachinery/pkg/labels"
 	kube_client "k8s.io/client-go/kubernetes"
@@ -157,6 +159,7 @@ func (this *summaryMetricsSource) decodeNodeStats(metrics map[string]*MetricSet,
 
 	this.decodeUptime(nodeMetrics, node.StartTime.Time)
 	this.decodeCPUStats(nodeMetrics, node.CPU)
+	this.decodeCpuCoreStats(nodeMetrics, node.CPUPerCore)
 	this.decodeMemoryStats(nodeMetrics, node.Memory)
 	this.decodeNetworkStats(nodeMetrics, node.Network)
 	this.decodeFsStats(nodeMetrics, RootFsKey, node.Fs)
@@ -241,6 +244,25 @@ func (this *summaryMetricsSource) decodeCPUStats(metrics *MetricSet, cpu *stats.
 	this.addIntMetric(metrics, &MetricCpuUsage, cpu.UsageCoreNanoSeconds)
 }
 
+func (this *summaryMetricsSource) decodeCpuCoreStats(metrics *MetricSet, coreStats *cadvisor.CpuStats) {
+	if coreStats == nil {
+		glog.V(9).Infof("missing cpu core usage metric!")
+		return
+	}
+	for index, value := range coreStats.Usage.PerCpu {
+		MetricCpuCoreUsage := Metric{
+			MetricDescriptor: MetricDescriptor{
+				Name:        "cpucore/core_" + strconv.Itoa(index),
+				Description: "CPU core usage in nanocores.",
+				Type:        MetricGauge,
+				ValueType:   ValueInt64,
+				Units:       UnitsCount,
+			},
+		}
+		this.addIntMetric(metrics, &MetricCpuCoreUsage, &value)
+	}
+}
+
 func (this *summaryMetricsSource) decodeMemoryStats(metrics *MetricSet, memory *stats.MemoryStats) {
 	if memory == nil {
 		glog.V(9).Infof("missing memory metrics!")