Add pod-level metrics into heapster

This commit adds pod-level cpu, memory and ephemeralstorage metrics from
summary API to heaspter.

Before, some pod-level metrics are aggregated from containers. After
getting those metrics directly from pod summary API, it no longer needs
to aggregate from containers.

Author: jingxu97

Godeps/Godeps.json

@@ -29,6 +29,7 @@ const (
 var StandardMetrics = []Metric{
 	MetricUptime,
 	MetricCpuUsage,
+	MetricEphemeralStorageUsage,
 	MetricMemoryUsage,
 	MetricMemoryRSS,
 	MetricMemoryCache,
@@ -220,6 +221,15 @@ var MetricCpuUsage = Metric{
 	},
 }
 
+var MetricEphemeralStorageUsage = Metric{
+	MetricDescriptor: MetricDescriptor{
+		Name:        "ephemeralstorage/usage",
+		Description: "Ephemeral storage usage",
+		Type:        MetricGauge,
+		ValueType:   ValueInt64,
+		Units:       UnitsBytes,
+	},
+}
 var MetricMemoryUsage = Metric{
 	MetricDescriptor: MetricDescriptor{
 		Name:        "memory/usage",

metrics/core/metrics.go

@@ -42,6 +42,8 @@ func (this *PodAggregator) Name() string {
 func (this *PodAggregator) Process(batch *core.DataBatch) (*core.DataBatch, error) {
 	newPods := make(map[string]*core.MetricSet)
 
+	// If pod already has pod-level metrics, it no longer needs to aggregates its container's metrics.
+	requireAggregate := make(map[string]bool)
 	for key, metricSet := range batch.MetricSets {
 		if metricSetType, found := metricSet.Labels[core.LabelMetricSetType.Key]; !found || metricSetType != core.MetricSetTypePodContainer {
 			continue
@@ -72,23 +74,27 @@ func (this *PodAggregator) Process(batch *core.DataBatch) (*core.DataBatch, erro
 			}
 
 			aggregatedValue, found := pod.MetricValues[metricName]
-			if found {
-				if aggregatedValue.ValueType != metricValue.ValueType {
-					glog.Errorf("PodAggregator: inconsistent type in %s", metricName)
-					continue
-				}
-
-				switch aggregatedValue.ValueType {
-				case core.ValueInt64:
-					aggregatedValue.IntValue += metricValue.IntValue
-				case core.ValueFloat:
-					aggregatedValue.FloatValue += metricValue.FloatValue
-				default:
-					return nil, fmt.Errorf("PodAggregator: type not supported in %s", metricName)
-				}
-			} else {
+			if !found {
+				requireAggregate[podKey+metricName] = true
 				aggregatedValue = metricValue
+			} else {
+				if requireAggregate[podKey+metricName] {
+					if aggregatedValue.ValueType != metricValue.ValueType {
+						glog.Errorf("PodAggregator: inconsistent type in %s", metricName)
+						continue
+					}
+
+					switch aggregatedValue.ValueType {
+					case core.ValueInt64:
+						aggregatedValue.IntValue += metricValue.IntValue
+					case core.ValueFloat:
+						aggregatedValue.FloatValue += metricValue.FloatValue
+					default:
+						return nil, fmt.Errorf("PodAggregator: type not supported in %s", metricName)
+					}
+				}
 			}
+
 			pod.MetricValues[metricName] = aggregatedValue
 		}
 	}

metrics/processors/pod_aggregator.go

@@ -66,6 +66,41 @@ func TestPodAggregator(t *testing.T) {
 					},
 				},
 			},
+
+			core.PodKey("ns1", "pod2"): {
+				Labels: map[string]string{
+					core.LabelMetricSetType.Key: core.MetricSetTypePod,
+					core.LabelPodName.Key:       "pod2",
+					core.LabelNamespaceName.Key: "ns1",
+				},
+				MetricValues: map[string]core.MetricValue{
+					"m1": {
+						ValueType:  core.ValueInt64,
+						MetricType: core.MetricGauge,
+						IntValue:   100,
+					},
+				},
+			},
+
+			core.PodContainerKey("ns1", "pod2", "c1"): {
+				Labels: map[string]string{
+					core.LabelMetricSetType.Key: core.MetricSetTypePodContainer,
+					core.LabelPodName.Key:       "pod2",
+					core.LabelNamespaceName.Key: "ns1",
+				},
+				MetricValues: map[string]core.MetricValue{
+					"m1": {
+						ValueType:  core.ValueInt64,
+						MetricType: core.MetricGauge,
+						IntValue:   10,
+					},
+					"m2": {
+						ValueType:  core.ValueInt64,
+						MetricType: core.MetricGauge,
+						IntValue:   20,
+					},
+				},
+			},
 		},
 	}
 	processor := PodAggregator{}
@@ -93,4 +128,16 @@ func TestPodAggregator(t *testing.T) {
 	labelNsName, found := pod.Labels[core.LabelNamespaceName.Key]
 	assert.True(t, found)
 	assert.Equal(t, "ns1", labelNsName)
+
+	pod, found = result.MetricSets[core.PodKey("ns1", "pod2")]
+	assert.True(t, found)
+
+	m1, found = pod.MetricValues["m1"]
+	assert.True(t, found)
+	assert.Equal(t, int64(100), m1.IntValue)
+
+	m2, found = pod.MetricValues["m2"]
+	assert.True(t, found)
+	assert.Equal(t, int64(20), m2.IntValue)
+
 }

metrics/processors/pod_aggregator_test.go

@@ -188,6 +188,9 @@ func (this *summaryMetricsSource) decodePodStats(metrics map[string]*MetricSet,
 
 	this.decodeUptime(podMetrics, pod.StartTime.Time)
 	this.decodeNetworkStats(podMetrics, pod.Network)
+	this.decodeCPUStats(podMetrics, pod.CPU)
+	this.decodeMemoryStats(podMetrics, pod.Memory)
+	this.decodeEphemeralStorageStats(podMetrics, pod.EphemeralStorage)
 	for _, vol := range pod.VolumeStats {
 		this.decodeFsStats(podMetrics, VolumeResourcePrefix+vol.Name, &vol.FsStats)
 	}
@@ -255,6 +258,14 @@ func (this *summaryMetricsSource) decodeCPUStats(metrics *MetricSet, cpu *stats.
 	this.addIntMetric(metrics, &MetricCpuUsage, cpu.UsageCoreNanoSeconds)
 }
 
+func (this *summaryMetricsSource) decodeEphemeralStorageStats(metrics *MetricSet, storage *stats.FsStats) {
+	if storage == nil {
+		glog.V(9).Infof("missing storage usage metric!")
+		return
+	}
+	this.addIntMetric(metrics, &MetricEphemeralStorageUsage, storage.UsedBytes)
+}
+
 func (this *summaryMetricsSource) decodeMemoryStats(metrics *MetricSet, memory *stats.MemoryStats) {
 	if memory == nil {
 		glog.V(9).Infof("missing memory metrics!")

metrics/sources/summary/summary.go

@@ -153,8 +153,11 @@ func TestDecodeSummaryMetrics(t *testing.T) {
 				Name:      pName0,
 				Namespace: namespace0,
 			},
-			StartTime: metav1.NewTime(startTime),
-			Network:   genTestSummaryNetwork(seedPod0),
+			StartTime:        metav1.NewTime(startTime),
+			Network:          genTestSummaryNetwork(seedPod0),
+			EphemeralStorage: genTestSummaryFsStats(seedPod0),
+			CPU:              genTestSummaryCPU(seedPod0),
+			Memory:           genTestSummaryMemory(seedPod0),
 			Containers: []stats.ContainerStats{
 				genTestSummaryContainer(cName00, seedPod0Container0),
 				genTestSummaryContainer(cName01, seedPod0Container1),
@@ -235,14 +238,15 @@ func TestDecodeSummaryMetrics(t *testing.T) {
 
 	containerFs := []string{"/", "logs"}
 	expectations := []struct {
-		key          string
-		setType      string
-		seed         int64
-		cpu          bool
-		memory       bool
-		network      bool
-		accelerators bool
-		fs           []string
+		key              string
+		setType          string
+		seed             int64
+		cpu              bool
+		memory           bool
+		network          bool
+		accelerators     bool
+		ephemeralstorage bool
+		fs               []string
 	}{{
 		key:     core.NodeKey(nodeInfo.NodeName),
 		setType: core.MetricSetTypeNode,
@@ -270,10 +274,13 @@ func TestDecodeSummaryMetrics(t *testing.T) {
 		cpu:     true,
 		memory:  true,
 	}, {
-		key:     core.PodKey(namespace0, pName0),
-		setType: core.MetricSetTypePod,
-		seed:    seedPod0,
-		network: true,
+		key:              core.PodKey(namespace0, pName0),
+		setType:          core.MetricSetTypePod,
+		seed:             seedPod0,
+		network:          true,
+		cpu:              true,
+		memory:           true,
+		ephemeralstorage: true,
 	}, {
 		key:     core.PodKey(namespace0, pName1),
 		setType: core.MetricSetTypePod,
@@ -382,6 +389,9 @@ func TestDecodeSummaryMetrics(t *testing.T) {
 			checkAcceleratorMetric(t, m, e.key, core.MetricAcceleratorMemoryUsed, e.seed+offsetAcceleratorMemoryUsed)
 			checkAcceleratorMetric(t, m, e.key, core.MetricAcceleratorDutyCycle, e.seed+offsetAcceleratorDutyCycle)
 		}
+		if e.ephemeralstorage {
+			checkIntMetric(t, m, e.key, core.MetricEphemeralStorageUsage, e.seed+offsetFsUsed)
+		}
 		for _, label := range e.fs {
 			checkFsMetric(t, m, e.key, label, core.MetricFilesystemAvailable, e.seed+offsetFsAvailable)
 			checkFsMetric(t, m, e.key, label, core.MetricFilesystemLimit, e.seed+offsetFsCapacity)