Repository: kubernetes/kubernetes (language: go)

Issue:
HPA with container metrics fails when any pod is not in a ready state
### What happened?

We've noticed that our HPA will stay at it's current scale and not make decisions when a pod is in an un-Ready state. IE a single pod is in CrashLoopBackoff and the HPA shows:

```
...
resource cpu of container "mycontainer" on pods  (as a percentage of request):     <unknown> / 50%
...
Warning  FailedGetResourceMetric  78s (x7 over 4m7s)  horizontal-pod-autoscaler  unable to get metric cpu: failed to get container metrics: container guides3 not present in metrics for pod namesapce/mycontainer-20240904t1338-433ca6a1-db998df74-s5rfh
```

That particular pod was in CrashLoopBackoff at the time and it wasn't until i removed it was it able to make scaling decisions

We're also running istio so there is another container in these pods and it's running ok; this may be why it's showing up in the metrics but unable to get info on the main container

### What did you expect to happen?

I would expect the HPA to ignore pods in this state and use only the Ready pods for its decisions

### How can we reproduce it (as minimally and precisely as possible)?

Setup an HPA with container based metrics
Put one of the pods in bad state, could just have a single pod consume a lot of memory and have it OOM a few times
See the HPA make no decisions while it's in this state

### Anything else we need to know?

I've looked through the code and it seems to be steming from:
`metrics, err := c.client.PodMetricses(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector.String()})`
This client just uses the selector and doesn't care about the state any of the pods might be in. It does end up filtering out pods later in the reconcile cycle but this returns an error here if it's not in the list of metrics because a single container may not be running 

### Kubernetes version

<details>

```console
$ kubectl version
Client Version: v1.29.6
Kustomize Version: v5.0.4-0.20230601165947-6ce0bf390ce3
Server Version: v1.29.6-gke.1326000
```

</details>


### Cloud provider

<details>
GCP
</details>


### OS version

_No response_

### Install tools

_No response_

### Container runtime (CRI) and version (if applicable)

_No response_

### Related plugins (CNI, CSI, ...) and versions (if applicable)

_No response_


Discussion / hints:
The container metric should come from cadvisor. When an OOM occurs, the container will be deleted and recreated, so HPA may not be able to obtain the container's metrics at this time.
Yeah I understand _why_ it's unable to get metrics but this feels really bad to break the hpa behavior because of a single container; especially one that is in a CrashLoopBackOff state and could get exponentially longer in that state. 


I've seen other instances as well with similar behavior. Another container is on a node that is has memory pressure, gets soft evicted but continues to consume ram, node goes unresponsive and the hpa breaks in a similar way. This single pod not having metrics and in a Terminating state seems line something that should be ignored in favor of the Ready pods, which is what is considered by the end of the reconcile loop
Because this is horizontal expansion, if do not consider these CrashLoopBack pods, the so-called ready pods that are expanded are likely to become CrashLoopBack. Will this lead to the creation of more and more pods, and maybe CrashLoopBack still become ready after the container is rebuilt

If the container's metrics cannot be obtained and the container's status is OOM, is it feasible to treat its memory usage as 100%?
Sure that's half of the equation; in my situation it was scaled to max and unable to scale down even though all metrics of the running pods showed it well under the targets. But even in the Expansion, i'm not sure how adding more pods in a situation like that would cause them to become Crash Loop Back off if you're expanding the pool of pods to take on the workload?

> If the container's metrics cannot be obtained and the container's status is OOM, is it feasible to treat its memory usage as 100%?

Sure if that's what folks agree on. 

I would advocate for treating the returned value much like it does for the Pod metric:
After it gets the metrics in `GetMetricReplicas` we then pass it to `calcPlainMetricReplicas` which takes into consideration the state of the pods and removes them from the calculations 

```
	readyPodCount, unreadyPods, missingPods, ignoredPods := groupPods(podList, metrics, resource, c.cpuInitializationPeriod, c.delayOfInitialReadinessStatus)
	removeMetricsForPods(metrics, ignoredPods)
	removeMetricsForPods(metrics, unreadyPods)
```

instead it just errors out if it's not in the list of metrics found, bubbles up and ends the hpa decision cycle
```
		if !containerFound {
			return nil, fmt.Errorf("container %s not present in metrics for pod %s/%s", container, m.Namespace, m.Name)
		}
		
```


If nothing else there should be some doc update to let people know this is the behaviour and how it differs from the Pod metric based HPA calculations
/sig autoscaling


Fix patch:
diff --git a/pkg/controller/podautoscaler/metrics/client.go b/pkg/controller/podautoscaler/metrics/client.go
index 03ecc22c443bd..71f2cc616bb41 100644
--- a/pkg/controller/podautoscaler/metrics/client.go
+++ b/pkg/controller/podautoscaler/metrics/client.go
@@ -75,10 +75,7 @@ func (c *resourceMetricsClient) GetResourceMetric(ctx context.Context, resource
 	}
 	var res PodMetricsInfo
 	if container != "" {
-		res, err = getContainerMetrics(metrics.Items, resource, container)
-		if err != nil {
-			return nil, time.Time{}, fmt.Errorf("failed to get container metrics: %v", err)
-		}
+		res = getContainerMetrics(ctx, metrics.Items, resource, container)
 	} else {
 		res = getPodMetrics(ctx, metrics.Items, resource)
 	}
@@ -86,7 +83,7 @@ func (c *resourceMetricsClient) GetResourceMetric(ctx context.Context, resource
 	return res, timestamp, nil
 }
 
-func getContainerMetrics(rawMetrics []metricsapi.PodMetrics, resource v1.ResourceName, container string) (PodMetricsInfo, error) {
+func getContainerMetrics(ctx context.Context, rawMetrics []metricsapi.PodMetrics, resource v1.ResourceName, container string) PodMetricsInfo {
 	res := make(PodMetricsInfo, len(rawMetrics))
 	for _, m := range rawMetrics {
 		containerFound := false
@@ -104,10 +101,10 @@ func getContainerMetrics(rawMetrics []metricsapi.PodMetrics, resource v1.Resourc
 			}
 		}
 		if !containerFound {
-			return nil, fmt.Errorf("container %s not present in metrics for pod %s/%s", container, m.Namespace, m.Name)
+			klog.FromContext(ctx).V(2).Info("Missing container metric", "container", container, "pod", klog.KRef(m.Namespace, m.Name))
 		}
 	}
-	return res, nil
+	return res
 }
 
 func getPodMetrics(ctx context.Context, rawMetrics []metricsapi.PodMetrics, resource v1.ResourceName) PodMetricsInfo {
diff --git a/pkg/controller/podautoscaler/metrics/interfaces.go b/pkg/controller/podautoscaler/metrics/interfaces.go
index 6b9b9f1aebfb9..bd1b6d7057849 100644
--- a/pkg/controller/podautoscaler/metrics/interfaces.go
+++ b/pkg/controller/podautoscaler/metrics/interfaces.go
@@ -41,6 +41,7 @@ type MetricsClient interface {
 	// GetResourceMetric gets the given resource metric (and an associated oldest timestamp)
 	// for the specified named container in all pods matching the specified selector in the given namespace and when
 	// the container is an empty string it returns the sum of all the container metrics.
+	// Missing metrics will not error and callers should rely on Pod status to filter metrics info returned from this interface.
 	GetResourceMetric(ctx context.Context, resource v1.ResourceName, namespace string, selector labels.Selector, container string) (PodMetricsInfo, time.Time, error)
 
 	// GetRawMetric gets the given metric (and an associated oldest timestamp)


Test patch:
diff --git a/pkg/controller/podautoscaler/metrics/client_test.go b/pkg/controller/podautoscaler/metrics/client_test.go
index 04418c6c0f7d7..59fbcc804aad3 100644
--- a/pkg/controller/podautoscaler/metrics/client_test.go
+++ b/pkg/controller/podautoscaler/metrics/client_test.go
@@ -429,7 +429,22 @@ func TestRESTClientContainerCPUEmptyMetricsForOnePod(t *testing.T) {
 		container:          "test-1",
 		targetTimestamp:    targetTimestamp,
 		window:             window,
-		desiredError:       fmt.Errorf("failed to get container metrics"),
+		reportedPodMetrics: []map[string]int64{{"test-1": 100}, {"test-1": 300, "test-2": 400}, {}},
+	}
+	tc.runTest(t)
+}
+
+func TestRESTClientContainerCPUEmptyMetricsForOnePodReturnsOnlyFoundContainer(t *testing.T) {
+	targetTimestamp := 1
+	window := 30 * time.Second
+	tc := restClientTestCase{
+		resourceName: v1.ResourceCPU,
+		desiredMetricValues: PodMetricsInfo{
+			"test-pod-1": {Value: 400, Timestamp: offsetTimestampBy(targetTimestamp), Window: window},
+		},
+		container:          "test-2",
+		targetTimestamp:    targetTimestamp,
+		window:             window,
 		reportedPodMetrics: []map[string]int64{{"test-1": 100}, {"test-1": 300, "test-2": 400}, {}},
 	}
 	tc.runTest(t)
diff --git a/pkg/controller/podautoscaler/replica_calculator_test.go b/pkg/controller/podautoscaler/replica_calculator_test.go
index 6a8653378b5f4..c90871bd4e85a 100644
--- a/pkg/controller/podautoscaler/replica_calculator_test.go
+++ b/pkg/controller/podautoscaler/replica_calculator_test.go
@@ -446,20 +446,6 @@ func TestReplicaCalcDisjointResourcesMetrics(t *testing.T) {
 	tc.runTest(t)
 }
 
-func TestReplicaCalcMissingContainerMetricError(t *testing.T) {
-	tc := replicaCalcTestCase{
-		currentReplicas: 1,
-		expectedError:   fmt.Errorf("container container2 not present in metrics for pod test-namespace/test-pod-0"),
-		resource: &resourceInfo{
-			name:     v1.ResourceCPU,
-			requests: []resource.Quantity{resource.MustParse("1.0")},
-			levels:   [][]int64{{0}},
-		},
-		container: "container2",
-	}
-	tc.runTest(t)
-}
-
 func TestReplicaCalcScaleUp(t *testing.T) {
 	tc := replicaCalcTestCase{
 		currentReplicas:  3,
@@ -606,6 +592,26 @@ func TestReplicaCalcScaleUpIgnoresFailedPods(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestReplicaCalcMissingContainerMetricScaleUpIgnoresFailedPods(t *testing.T) {
+	tc := replicaCalcTestCase{
+		currentReplicas:  2,
+		expectedReplicas: 4,
+		podReadiness:     []v1.ConditionStatus{v1.ConditionTrue, v1.ConditionTrue, v1.ConditionFalse, v1.ConditionFalse},
+		podPhase:         []v1.PodPhase{v1.PodRunning, v1.PodRunning, v1.PodFailed, v1.PodFailed},
+		resource: &resourceInfo{
+			name:     v1.ResourceCPU,
+			requests: []resource.Quantity{resource.MustParse("1.0"), resource.MustParse("1.0"), resource.MustParse("1.0"), resource.MustParse("1.0")},
+			levels:   [][]int64{{1000, 500}, {9000, 700}, {1000}, {9000}},
+
+			targetUtilization:   30,
+			expectedUtilization: 60,
+			expectedValue:       600,
+		},
+		container: "container2",
+	}
+	tc.runTest(t)
+}
+
 func TestReplicaCalcScaleUpContainerIgnoresFailedPods(t *testing.T) {
 	tc := replicaCalcTestCase{
 		currentReplicas:  2,