Make sure to skip fine-grained missing data

This skips missing pod or node data at the pod (if any container data is missing) or node level, so that things like autoscalers can make proper decisions.

Author: DirectXMan12

pkg/sources/summary/summary.go

@@ -109,10 +109,29 @@ func (src *summaryMetricsSource) Collect(ctx context.Context) (*sources.MetricsB
 
 var errs []error
 errs = append(errs, src.decodeNodeStats(&summary.Node, &res.Nodes[0])...)
-for i, pod := range summary.Pods {
-errs = append(errs, src.decodePodStats(&pod, &res.Pods[i])...)
+if len(errs) != 0 {
+// if we had errors providing node metrics, discard the data point
+// so that we don't incorrectly report metric values as zero.
+res.Nodes = res.Nodes[:1]
 }
 
+num := 0
+for _, pod := range summary.Pods {
+podErrs := src.decodePodStats(&pod, &res.Pods[num])
+errs = append(errs, podErrs...)
+if len(podErrs) != 0 {
+// NB: we explicitly want to discard pods with partial results, since
+// the horizontal pod autoscaler takes special action when a pod is missing
+// metrics (and zero CPU or memory does not count as "missing metrics")
+
+// we don't care if we reuse slots in the result array,
+// because they get completely overwritten in decodePodStats
+continue
+}
+num++
+}
+res.Pods = res.Pods[:num]
+
 return res, utilerrors.NewAggregate(errs)
 }
 
@@ -130,15 +149,16 @@ func (src *summaryMetricsSource) decodeNodeStats(nodeStats *stats.NodeStats, tar
 }
 var errs []error
 if err := decodeCPU(&target.CpuUsage, nodeStats.CPU); err != nil {
-errs = append(errs, fmt.Errorf("unable to get CPU for node %q: %v", src.node.ConnectAddress, err))
+errs = append(errs, fmt.Errorf("unable to get CPU for node %q, discarding data: %v", src.node.ConnectAddress, err))
 }
 if err := decodeMemory(&target.MemoryUsage, nodeStats.Memory); err != nil {
-errs = append(errs, fmt.Errorf("unable to get memory for node %q: %v", src.node.ConnectAddress, err))
+errs = append(errs, fmt.Errorf("unable to get memory for node %q, discarding data: %v", src.node.ConnectAddress, err))
 }
 return errs
 }
 
 func (src *summaryMetricsSource) decodePodStats(podStats *stats.PodStats, target *sources.PodMetricsPoint) []error {
+// completely overwrite data in the target
 *target = sources.PodMetricsPoint{
 Name: podStats.PodRef.Name,
 Namespace: podStats.PodRef.Namespace,
@@ -160,10 +180,10 @@ func (src *summaryMetricsSource) decodePodStats(podStats *stats.PodStats, target
 },
 }
 if err := decodeCPU(&point.CpuUsage, container.CPU); err != nil {
-errs = append(errs, fmt.Errorf("unable to get CPU for container %q in pod %s/%s on node %q: %v", container.Name, target.Namespace, target.Name, src.node.ConnectAddress, err))
+errs = append(errs, fmt.Errorf("unable to get CPU for container %q in pod %s/%s on node %q, discarding data: %v", container.Name, target.Namespace, target.Name, src.node.ConnectAddress, err))
 }
 if err := decodeMemory(&point.MemoryUsage, container.Memory); err != nil {
-errs = append(errs, fmt.Errorf("unable to get memory for container %q in pod %s/%s on node %q: %v", container.Name, target.Namespace, target.Name, src.node.ConnectAddress, err))
+errs = append(errs, fmt.Errorf("unable to get memory for container %q in pod %s/%s on node %q: %v, discarding data", container.Name, target.Namespace, target.Name, src.node.ConnectAddress, err))
 }
 
 target.Containers[i] = point

pkg/sources/summary/summary_test.go

@@ -124,22 +124,23 @@ func verifyPods(summary *stats.Summary, batch *sources.MetricsBatch) {
 var expectedPods []interface{}
 for _, pod := range summary.Pods {
 containers := make([]sources.ContainerMetricsPoint, len(pod.Containers))
+missingData := false
 for i, container := range pod.Containers {
 var cpuUsage, memoryUsage resource.Quantity
 var timestamp time.Time
-if container.CPU != nil {
-if container.CPU.UsageNanoCores != nil {
-cpuUsage = *resource.NewScaledQuantity(int64(*container.CPU.UsageNanoCores), -9)
-}
-timestamp = container.CPU.Time.Time
+if container.CPU == nil || container.CPU.UsageNanoCores == nil {
+missingData = true
+break
 }
-if container.Memory != nil {
-if container.Memory.WorkingSetBytes != nil {
-memoryUsage = *resource.NewQuantity(int64(*container.Memory.WorkingSetBytes), resource.BinarySI)
-}
-if timestamp.IsZero() {
-timestamp = container.Memory.Time.Time
-}
+cpuUsage = *resource.NewScaledQuantity(int64(*container.CPU.UsageNanoCores), -9)
+timestamp = container.CPU.Time.Time
+if container.Memory == nil || container.Memory.WorkingSetBytes == nil {
+missingData = true
+break
+}
+memoryUsage = *resource.NewQuantity(int64(*container.Memory.WorkingSetBytes), resource.BinarySI)
+if timestamp.IsZero() {
+timestamp = container.Memory.Time.Time
 }
 
 containers[i] = sources.ContainerMetricsPoint{
@@ -151,6 +152,9 @@ func verifyPods(summary *stats.Summary, batch *sources.MetricsBatch) {
 },
 }
 }
+if missingData {
+continue
+}
 expectedPods = append(expectedPods, sources.PodMetricsPoint{
 Name: pod.PodRef.Name,
 Namespace: pod.PodRef.Namespace,