Say when rightsizing could not read a workload's pods

internal/prometheus/rightsizing.go

@@ -93,8 +93,13 @@ type RightsizingRow struct {
 	CurrentPodOOM           bool                  `json:"currentPodOOM,omitempty"`
 	WindowOOMEvidence       bool                  `json:"windowOomEvidence,omitempty"`
 	OOMEvidenceAvailable    bool                  `json:"oomEvidenceAvailable"`
-	LimitConflict           bool                  `json:"limitConflict,omitempty"`
-	QueryError              string                `json:"queryError,omitempty"`
+	// The cluster cache could not answer for this workload's pods — denied, or
+	// still warming when the wait ran out — so CurrentPodOOM is false for want
+	// of an answer. Unset when the pods were read, including when the workload
+	// genuinely has none.
+	LiveInventoryUnavailable bool   `json:"liveInventoryUnavailable,omitempty"`
+	LimitConflict            bool   `json:"limitConflict,omitempty"`
+	QueryError               string `json:"queryError,omitempty"`
 }
 
 type RightsizingResponse struct {
@@ -149,7 +154,7 @@ func handleRightsizing(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	workload, err := loadRightsizingWorkload(kind, namespace, name)
+	workload, err := loadRightsizingWorkload(r.Context(), kind, namespace, name)
 	if err != nil {
 		switch {
 		case errors.Is(err, errCacheNotReady):
@@ -158,6 +163,9 @@ func handleRightsizing(w http.ResponseWriter, r *http.Request) {
 			writeError(w, http.StatusForbidden, err.Error())
 		case errors.Is(err, errWorkloadMissing):
 			writeError(w, http.StatusNotFound, err.Error())
+		case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
+			// The client is gone; nothing to report and nothing worth logging.
+			return
 		default:
 			errorlog.Record("prometheus", "error", "rightsizing: failed to load containers for %s %s/%s: %v", kind, namespace, name, err)
 			writeError(w, http.StatusInternalServerError, "failed to load workload containers")
@@ -204,9 +212,37 @@ type rightsizingWorkload struct {
 	hpaManaged    map[string]bool
 	hpaAvailable  bool
 	scaledToZero  bool
+	// The live pod list could not be read, so podNames and currentPodOOM are
+	// empty because nothing answered — not because the workload has no pods.
+	liveInventoryUnavailable bool
+}
+
+// warmingRetryBudget bounds how long a rightsizing read waits for an informer
+// that is still filling. Initial sync takes well under this; the budget exists
+// so a cache that never syncs cannot hold the request open.
+const warmingRetryBudget = 3 * time.Second
+
+// workloadPodsOnceWarm reads the workload's pods, waiting out a cache that is
+// still loading. Warming and denial arrive at this call as the same empty pod
+// list, but they are not the same thing: warming clears on its own within
+// seconds, so surfacing it would put a caveat on the recommendation that the
+// next refresh silently removes. A denial never clears, and is returned.
+func workloadPodsOnceWarm(ctx context.Context, cache *k8s.ResourceCache, kind, namespace, name string) ([]*corev1.Pod, error) {
+	deadline := time.Now().Add(warmingRetryBudget)
+	for {
+		pods, err := k8s.WorkloadPods(cache, kind, namespace, name)
+		if !errors.Is(err, k8s.ErrWorkloadCacheWarming) || time.Now().After(deadline) {
+			return pods, err
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
 }
 
-func loadRightsizingWorkload(kind, namespace, name string) (rightsizingWorkload, error) {
+func loadRightsizingWorkload(ctx context.Context, kind, namespace, name string) (rightsizingWorkload, error) {
 	cache := k8s.GetResourceCache()
 	if cache == nil {
 		return rightsizingWorkload{}, errCacheNotReady
@@ -259,14 +295,20 @@ func loadRightsizingWorkload(kind, namespace, name string) (rightsizingWorkload,
 		hpaAvailable:  hpaAvailable,
 		scaledToZero:  scaledToZero,
 	}
-	pods, err := k8s.WorkloadPods(cache, kind, namespace, name)
+	pods, err := workloadPodsOnceWarm(ctx, cache, kind, namespace, name)
+	if ctx.Err() != nil {
+		// The caller went away mid-wait. Reporting that as an unreadable
+		// inventory would answer a question nobody is listening for, and would
+		// send the rest of this request into Prometheus on a dead context.
+		return rightsizingWorkload{}, ctx.Err()
+	}
 	if err != nil {
-		// Rightsizing deliberately ignores the denial: its recommendation
-		// comes from seven days of kube-state-metrics ownership, which does
-		// not need the pod list, and the current-pod fallback simply finds
-		// nothing. Nothing in the response distinguishes that from a workload
-		// with no pods, which is a gap worth closing when the response gains
-		// a field for it.
+		// The recommendation itself survives: it comes from seven days of
+		// kube-state-metrics ownership, which never needed the pod list. What
+		// does not survive is everything read from the live pods — the pod
+		// names, and whether those pods have been OOM-killed — so the row is
+		// marked rather than served as if the workload simply had no pods.
+		workload.liveInventoryUnavailable = true
 		return workload, nil
 	}
 	for _, pod := range pods {
@@ -394,6 +436,12 @@ func computeRightsizing(ctx context.Context, client rightsizingQuerier, kind, na
 	if !resp.SampleAvailable {
 		if results["cpu_stat"].err != nil || results["cpu_coverage"].err != nil || results["memory_stat"].err != nil || results["memory_coverage"].err != nil {
 			resp.Reason = "Prometheus rightsizing queries failed."
+		} else if workload.liveInventoryUnavailable && coverage == OwnerCoverageCurrentPods {
+			// Without retained ownership the selection falls back to the current
+			// pods, and those could not be listed — so the query matched nothing
+			// by construction. "No samples" would send the reader after missing
+			// metrics instead of the unreadable inventory.
+			resp.Reason = "Radar could not read this workload's pods, and no retained ownership history was available, so there was nothing to measure."
 		} else if len(workload.podNames) == 0 && coverage == OwnerCoverageCurrentPods {
 			resp.Reason = "No current or retained workload ownership samples are available."
 		} else {
@@ -532,7 +580,7 @@ func terminationEvidenceByContainer(result *prom.QueryResult) map[string]termina
 }
 
 func buildRightsizingRow(container containerSpec, resourceName string, expected int, ownerCoverage OwnerCoverage, workload rightsizingWorkload, results map[string]queryOutcome) RightsizingRow {
-	row := RightsizingRow{Container: container.name, Resource: resourceName, Fit: FitInsufficientHistory, Confidence: ConfidenceLow, ExpectedSamples: expected, HPAManaged: workload.hpaManaged[resourceName], HPAEvidenceAvailable: workload.hpaAvailable}
+	row := RightsizingRow{Container: container.name, Resource: resourceName, Fit: FitInsufficientHistory, Confidence: ConfidenceLow, ExpectedSamples: expected, HPAManaged: workload.hpaManaged[resourceName], HPAEvidenceAvailable: workload.hpaAvailable, LiveInventoryUnavailable: workload.liveInventoryUnavailable}
 	var req, lim *resource.Quantity
 	statistic := "P95"
 	if resourceName == "cpu" {

internal/prometheus/rightsizing_scan.go

@@ -110,7 +110,7 @@ func ScanRightsizing(ctx context.Context, scope RightsizingScanScope) Rightsizin
 		resp.Reason = "resource_cache_unavailable"
 		return resp
 	}
-	workloads, unavailable := snapshotScanWorkloads(cache, scope.NamespacesByKind)
+	workloads, unavailable := snapshotScanWorkloads(ctx, cache, scope.NamespacesByKind)
 	resp.Coverage.UnavailableKinds = unavailable
 	return computeRightsizingScan(ctx, client, workloads, resp)
 }
@@ -443,6 +443,7 @@ func buildScanRow(container containerSpec, resourceName string, key scanKey, exp
 	row := RightsizingRow{
 		Container: container.name, Resource: resourceName, Fit: FitInsufficientHistory, Confidence: ConfidenceLow,
 		ExpectedSamples: expected, HPAManaged: workload.hpaManaged[resourceName], HPAEvidenceAvailable: workload.hpaAvailable,
+		LiveInventoryUnavailable: workload.liveInventoryUnavailable,
 	}
 	var request, limit = container.cpuReq, container.cpuLim
 	statistic := "P95"
@@ -596,7 +597,7 @@ func sortScanWorkloads(workloads []scanWorkload) {
 	})
 }
 
-func snapshotScanWorkloads(cache *k8s.ResourceCache, scopes map[string][]string) ([]scanWorkload, []string) {
+func snapshotScanWorkloads(ctx context.Context, cache *k8s.ResourceCache, scopes map[string][]string) ([]scanWorkload, []string) {
 	workloads := map[string]*scanWorkload{}
 	var unavailable []string
 	add := func(kind, namespace, name string, replicas int, podSpec *corev1.PodSpec, scaledToZero bool) {
@@ -652,7 +653,7 @@ func snapshotScanWorkloads(cache *k8s.ResourceCache, scopes map[string][]string)
 	}
 
 	enrichScanHPA(cache, scopes, workloads)
-	enrichScanCurrentOOM(cache, scopes, workloads)
+	enrichScanCurrentOOM(ctx, cache, scopes, workloads)
 	out := make([]scanWorkload, 0, len(workloads))
 	for _, workload := range workloads {
 		out = append(out, *workload)
@@ -758,36 +759,94 @@ func markScanHPAAvailable(workloads map[string]*scanWorkload, namespace string)
 	}
 }
 
-func enrichScanCurrentOOM(cache *k8s.ResourceCache, scopes map[string][]string, workloads map[string]*scanWorkload) {
-	if cache.Pods() == nil {
-		return
+// waitForInformerSynced reports whether an informer finished its initial sync,
+// waiting out a warming cache within the same budget a single-workload read
+// uses. A kind this cache never watched is not warming, and is left to the
+// nil-lister check.
+func waitForInformerSynced(ctx context.Context, cache *k8s.ResourceCache, key string) bool {
+	deadline := time.Now().Add(warmingRetryBudget)
+	for {
+		synced, known := cache.InformerSynced(key)
+		if synced || !known {
+			return true
+		}
+		if time.Now().After(deadline) {
+			return false
+		}
+		select {
+		case <-ctx.Done():
+			return false
+		case <-time.After(50 * time.Millisecond):
+		}
 	}
-	namespaces := scanScopeNamespaces(scopes)
-	replicaSetOwners := map[string]string{}
-	if cache.ReplicaSets() != nil {
-		var replicaSets []*appsv1.ReplicaSet
-		if namespaces == nil {
-			replicaSets, _ = cache.ReplicaSets().List(labels.Everything())
-		} else {
-			for _, namespace := range namespaces {
-				items, _ := cache.ReplicaSets().ReplicaSets(namespace).List(labels.Everything())
-				replicaSets = append(replicaSets, items...)
-			}
+}
+
+// markScanLiveInventoryUnavailable records that a live pod read did not happen
+// for these workloads, so an absent OOM signal means nothing was asked rather
+// than nothing was found. An empty kind marks every workload.
+func markScanLiveInventoryUnavailable(workloads map[string]*scanWorkload, kind string) {
+	for _, workload := range workloads {
+		if kind == "" || strings.EqualFold(workload.kind, kind) {
+			workload.workload.liveInventoryUnavailable = true
 		}
-		for _, replicaSet := range replicaSets {
-			if owner := metav1.GetControllerOf(replicaSet); owner != nil && owner.Kind == "Deployment" {
-				replicaSetOwners[replicaSet.Namespace+"\x00"+replicaSet.Name] = owner.Name
-			}
+	}
+}
+
+// scanNeedsReplicaSets reports whether any workload in the scan resolves its
+// pods through a ReplicaSet. A StatefulSet or DaemonSet owns its pods directly,
+// so a scan of only those must not wait on a deferred informer it never reads.
+func scanNeedsReplicaSets(workloads map[string]*scanWorkload) bool {
+	for _, workload := range workloads {
+		if strings.EqualFold(workload.kind, "Deployment") {
+			return true
 		}
 	}
-	var pods []*corev1.Pod
-	if namespaces == nil {
-		pods, _ = cache.Pods().List(labels.Everything())
-	} else {
-		for _, namespace := range namespaces {
-			items, _ := cache.Pods().Pods(namespace).List(labels.Everything())
-			pods = append(pods, items...)
+	return false
+}
+
+func enrichScanCurrentOOM(ctx context.Context, cache *k8s.ResourceCache, scopes map[string][]string, workloads map[string]*scanWorkload) {
+	// A warming informer hands out a non-nil lister that answers nothing, which
+	// would read as a cluster with no OOM-killed pods anywhere. The scan pays
+	// this wait once rather than per workload.
+	if !waitForInformerSynced(ctx, cache, "pods") || cache.Pods() == nil {
+		markScanLiveInventoryUnavailable(workloads, "")
+		return
+	}
+	// Default false: a scan that needs no ReplicaSets never establishes them,
+	// and must not then be treated as having read a lister it never touched.
+	ownersReadable := false
+	if scanNeedsReplicaSets(workloads) {
+		ownersReadable = cache.ReplicaSets() != nil && waitForInformerSynced(ctx, cache, "replicasets")
+	}
+	if ctx.Err() != nil {
+		// Nobody is waiting for this scan; do not walk the cluster for it.
+		markScanLiveInventoryUnavailable(workloads, "")
+		return
+	}
+	// A lister that is synced and non-nil still only answers for the namespaces
+	// its informer watches. An empty result for a namespace outside that scope
+	// is not "no OOM-killed pods" — it is no answer, and it has to be reported
+	// per workload because the scan can span both kinds of namespace at once.
+	for _, workload := range workloads {
+		if !cache.KindCoversNamespace("pods", workload.namespace) {
+			workload.workload.liveInventoryUnavailable = true
+			continue
 		}
+		if strings.EqualFold(workload.kind, "Deployment") &&
+			(!ownersReadable || !cache.KindCoversNamespace("replicasets", workload.namespace)) {
+			workload.workload.liveInventoryUnavailable = true
+		}
+	}
+
+	namespaces := scanScopeNamespaces(scopes)
+	replicaSetOwners, ownersListed := scanReplicaSetOwners(ctx, cache, namespaces, ownersReadable)
+	pods, podsListed := scanPods(cache, namespaces)
+	if !podsListed {
+		markScanLiveInventoryUnavailable(workloads, "")
+		return
+	}
+	if !ownersListed {
+		markScanLiveInventoryUnavailable(workloads, "Deployment")
 	}
 	for _, pod := range pods {
 		owner := metav1.GetControllerOf(pod)
@@ -807,6 +866,59 @@ func enrichScanCurrentOOM(cache *k8s.ResourceCache, scopes map[string][]string,
 	}
 }
 
+// scanReplicaSetOwners maps each ReplicaSet to its Deployment. It reports false
+// if any list failed, because a partial map silently drops the pods it could
+// not resolve rather than failing loudly.
+func scanReplicaSetOwners(ctx context.Context, cache *k8s.ResourceCache, namespaces []string, readable bool) (map[string]string, bool) {
+	owners := map[string]string{}
+	if !readable || cache.ReplicaSets() == nil {
+		return owners, false
+	}
+	var replicaSets []*appsv1.ReplicaSet
+	if namespaces == nil {
+		list, err := cache.ReplicaSets().List(labels.Everything())
+		if err != nil {
+			return owners, false
+		}
+		replicaSets = list
+	} else {
+		for _, namespace := range namespaces {
+			items, err := cache.ReplicaSets().ReplicaSets(namespace).List(labels.Everything())
+			if err != nil {
+				return owners, false
+			}
+			replicaSets = append(replicaSets, items...)
+		}
+	}
+	for _, replicaSet := range replicaSets {
+		if owner := metav1.GetControllerOf(replicaSet); owner != nil && owner.Kind == "Deployment" {
+			owners[replicaSet.Namespace+"\x00"+replicaSet.Name] = owner.Name
+		}
+	}
+	return owners, true
+}
+
+// scanPods reports false if any list failed: a short pod list reads as pods
+// without OOM history rather than pods that were never examined.
+func scanPods(cache *k8s.ResourceCache, namespaces []string) ([]*corev1.Pod, bool) {
+	if namespaces == nil {
+		pods, err := cache.Pods().List(labels.Everything())
+		if err != nil {
+			return nil, false
+		}
+		return pods, true
+	}
+	var pods []*corev1.Pod
+	for _, namespace := range namespaces {
+		items, err := cache.Pods().Pods(namespace).List(labels.Everything())
+		if err != nil {
+			return nil, false
+		}
+		pods = append(pods, items...)
+	}
+	return pods, true
+}
+
 func scanScopeNamespaces(scopes map[string][]string) []string {
 	seen := map[string]bool{}
 	for _, kind := range []string{"Deployment", "StatefulSet", "DaemonSet"} {