package k8score

import (
	"context"
	"errors"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"golang.org/x/sync/errgroup"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
)

const (
	// MetricsHistorySize is the number of data points to keep (1 hour at 30s intervals).
	MetricsHistorySize = 120
	// MetricsPollInterval is how often to poll the metrics API.
	MetricsPollInterval = 30 * time.Second
)

// MetricsDataPoint represents a single metrics sample.
type MetricsDataPoint struct {
	Timestamp time.Time `json:"timestamp"`
	CPU       int64     `json:"cpu"`    // nanocores
	Memory    int64     `json:"memory"` // bytes
}

// ContainerMetricsHistory holds historical metrics for a container.
type ContainerMetricsHistory struct {
	Name       string             `json:"name"`
	DataPoints []MetricsDataPoint `json:"dataPoints"`
}

// PodMetricsHistory holds historical metrics for a pod.
type PodMetricsHistory struct {
	Namespace                   string                    `json:"namespace"`
	Name                        string                    `json:"name"`
	Containers                  []ContainerMetricsHistory `json:"containers"`
	CollectionError             string                    `json:"collectionError,omitempty"`
	RawCollectionError          string                    `json:"rawCollectionError,omitempty"`
	MetricsUnavailableDiagnosis string                    `json:"metricsUnavailableDiagnosis,omitempty"`
	MetricsUnavailable          bool                      `json:"metricsUnavailable,omitempty"`
}

// NodeMetricsHistory holds historical metrics for a node.
type NodeMetricsHistory struct {
	Name                        string             `json:"name"`
	DataPoints                  []MetricsDataPoint `json:"dataPoints"`
	CollectionError             string             `json:"collectionError,omitempty"`
	RawCollectionError          string             `json:"rawCollectionError,omitempty"`
	MetricsUnavailableDiagnosis string             `json:"metricsUnavailableDiagnosis,omitempty"`
	MetricsUnavailable          bool               `json:"metricsUnavailable,omitempty"`
}

// ContainerResourceMetrics holds latest usage plus request/limit for a single
// container. Usage comes from the metrics API; request/limit from the pod spec.
type ContainerResourceMetrics struct {
	Name          string `json:"name"`
	CPU           int64  `json:"cpu"`
	CPURequest    int64  `json:"cpuRequest"`
	CPULimit      int64  `json:"cpuLimit"`
	Memory        int64  `json:"memory"`
	MemoryRequest int64  `json:"memoryRequest"`
	MemoryLimit   int64  `json:"memoryLimit"`
}

// TopPodMetrics holds the latest metrics snapshot for a single pod.
type TopPodMetrics struct {
	Namespace     string `json:"namespace"`
	Name          string `json:"name"`
	CPU           int64  `json:"cpu"`
	Memory        int64  `json:"memory"`
	CPURequest    int64  `json:"cpuRequest"`
	CPULimit      int64  `json:"cpuLimit"`
	MemoryRequest int64  `json:"memoryRequest"`
	MemoryLimit   int64  `json:"memoryLimit"`
	// Containers carries per-container usage and request/limit for pods with
	// more than one running container (regular + native sidecars). Omitted for
	// single-container pods, where the pod-level fields above already suffice.
	Containers []ContainerResourceMetrics `json:"containers,omitempty"`
}

// TopNodeMetrics holds the latest metrics snapshot for a single node.
type TopNodeMetrics struct {
	Name              string    `json:"name"`
	CPU               int64     `json:"cpu"`
	Memory            int64     `json:"memory"`
	ObservedAt        time.Time `json:"observedAt,omitzero"`
	PodCount          int       `json:"podCount"`
	CPUAllocatable    int64     `json:"cpuAllocatable"`
	MemoryAllocatable int64     `json:"memoryAllocatable"`
}

// MetricsCollectionHealth reports the health of the metrics collection loop.
type MetricsCollectionHealth struct {
	PodMetrics       MetricsSourceHealth `json:"podMetrics"`
	NodeMetrics      MetricsSourceHealth `json:"nodeMetrics"`
	LastAttempt      string              `json:"lastAttempt,omitempty"`
	TotalCollections int64               `json:"totalCollections"`
	BufferSize       int                 `json:"bufferSize"`
	PollIntervalSec  int                 `json:"pollIntervalSec"`
}

// MetricsSourceHealth reports health for a single metrics source.
type MetricsSourceHealth struct {
	Collecting        bool   `json:"collecting"`
	LastSuccess       string `json:"lastSuccess,omitempty"`
	ConsecutiveErrors int    `json:"consecutiveErrors"`
	LastError         string `json:"lastError,omitempty"`
	TrackedCount      int    `json:"trackedCount"`
	TotalDataPoints   int    `json:"totalDataPoints"`
}

// MetricsHistoryStore stores historical metrics data polled from the metrics.k8s.io API.
type MetricsHistoryStore struct {
	mu         sync.RWMutex
	dynClient  dynamic.Interface
	resolveGVR MetricsGVRResolver

	// Namespaces, when it returns a non-empty list, limits pod metrics
	// collection to those namespaces, one LIST each. An install whose RBAC is
	// granted per namespace cannot list pod metrics cluster-wide.
	Namespaces func() []string

	podMetrics  map[string]*podMetricsBuffer
	nodeMetrics map[string]*nodeMetricsBuffer

	lastSuccessfulPodCollection  time.Time
	lastSuccessfulNodeCollection time.Time
	lastCollectionAttempt        time.Time
	totalCollections             int64
	consecutivePodErrors         int
	partialPodErrors             int // collections where some, not all, namespaces failed
	consecutiveNodeErrors        int
	lastPodError                 string
	lastNodeError                string

	// OnError is called when a metrics collection error is logged.
	// It allows callers to record errors in an external error log.
	OnError func(subsystem, level, format string, args ...any)

	stopCh    chan struct{}
	startOnce sync.Once
	stopOnce  sync.Once
	wg        sync.WaitGroup
}

type podMetricsBuffer struct {
	namespace  string
	name       string
	containers map[string]*ringBuffer
	lastSeen   time.Time
}

type nodeMetricsBuffer struct {
	name     string
	buffer   *ringBuffer
	lastSeen time.Time
}

type ringBuffer struct {
	data  []MetricsDataPoint
	head  int
	count int
	size  int
}

func newRingBuffer(size int) *ringBuffer {
	return &ringBuffer{data: make([]MetricsDataPoint, size), size: size}
}

func (rb *ringBuffer) Add(point MetricsDataPoint) {
	rb.data[rb.head] = point
	rb.head = (rb.head + 1) % rb.size
	if rb.count < rb.size {
		rb.count++
	}
}

func (rb *ringBuffer) GetAll() []MetricsDataPoint {
	if rb.count == 0 {
		return nil
	}
	result := make([]MetricsDataPoint, rb.count)
	if rb.count < rb.size {
		copy(result, rb.data[:rb.count])
	} else {
		start := rb.head
		for i := 0; i < rb.count; i++ {
			result[i] = rb.data[(start+i)%rb.size]
		}
	}
	return result
}

func metricsCollectionErrorLevel(err error) string {
	if MetricsAPIUnavailable(err) {
		return "warning"
	}
	return "error"
}

func MetricsAPIUnavailable(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, ErrMetricsAPINotDiscovered) {
		return true
	}
	if nf, ok := err.(namespacesFailedError); ok {
		return nf.metricsUnavailable()
	}
	if apierrors.IsNotFound(err) {
		return true
	}
	msg := strings.ToLower(err.Error())
	if !strings.Contains(msg, "metrics") {
		return false
	}
	if apierrors.IsServiceUnavailable(err) {
		return true
	}
	return strings.Contains(msg, "not found") ||
		strings.Contains(msg, "could not find the requested resource") ||
		strings.Contains(msg, "no matches for kind") ||
		strings.Contains(msg, "no resource matches") ||
		strings.Contains(msg, "no metrics known") ||
		strings.Contains(msg, "not available") ||
		strings.Contains(msg, "unable to fetch metrics") ||
		strings.Contains(msg, "currently unable to handle the request")
}

// NewMetricsHistoryStore creates a MetricsHistoryStore. Call Start() to begin polling.
func NewMetricsHistoryStore(client dynamic.Interface) *MetricsHistoryStore {
	return NewMetricsHistoryStoreWithResolver(client, func(resource string) (schema.GroupVersionResource, bool) {
		switch resource {
		case "pods":
			return PodMetricsGVR, true
		case "nodes":
			return NodeMetricsGVR, true
		default:
			return schema.GroupVersionResource{}, false
		}
	})
}

// NewMetricsHistoryStoreWithResolver creates a store that follows discovered metrics API versions.
func NewMetricsHistoryStoreWithResolver(client dynamic.Interface, resolveGVR MetricsGVRResolver) *MetricsHistoryStore {
	return &MetricsHistoryStore{
		dynClient:   client,
		resolveGVR:  resolveGVR,
		podMetrics:  make(map[string]*podMetricsBuffer),
		nodeMetrics: make(map[string]*nodeMetricsBuffer),
		stopCh:      make(chan struct{}),
	}
}

// Start begins background polling. Idempotent: subsequent calls are no-ops.
func (s *MetricsHistoryStore) Start() {
	s.startOnce.Do(func() {
		s.wg.Add(1)
		go s.pollLoop()
		log.Println("Metrics history collection started")
	})
}

// Stop halts background polling and waits for the goroutine to exit.
func (s *MetricsHistoryStore) Stop() {
	s.stopOnce.Do(func() { close(s.stopCh) })
	s.wg.Wait()
	log.Println("Metrics history collection stopped")
}

func (s *MetricsHistoryStore) pollLoop() {
	defer s.wg.Done()
	s.collectMetrics()
	ticker := time.NewTicker(MetricsPollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-s.stopCh:
			return
		case <-ticker.C:
			s.collectMetrics()
		}
	}
}

func (s *MetricsHistoryStore) collectMetrics() {
	now := time.Now()
	s.mu.Lock()
	s.lastCollectionAttempt = now
	s.totalCollections++
	s.mu.Unlock()
	podCtx, cancelPods := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancelPods()
	s.collectPodMetrics(podCtx, now)
	nodeCtx, cancelNodes := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancelNodes()
	s.collectNodeMetrics(nodeCtx, now)
}

func (s *MetricsHistoryStore) metricsGVR(resource string) (schema.GroupVersionResource, bool) {
	if s.resolveGVR == nil {
		return schema.GroupVersionResource{}, false
	}
	gvr, ok := s.resolveGVR(resource)
	if !ok || gvr.Group != MetricsAPIGroup || gvr.Version == "" || gvr.Resource != resource {
		return schema.GroupVersionResource{}, false
	}
	return gvr, true
}

func (s *MetricsHistoryStore) recordPodCollectionError(err error) {
	errMsg := err.Error()
	s.mu.Lock()
	s.consecutivePodErrors++
	s.lastPodError = errMsg
	shouldReport := s.consecutivePodErrors == 1 || s.consecutivePodErrors%20 == 0
	count := s.consecutivePodErrors
	s.mu.Unlock()
	if shouldReport {
		log.Printf("[metrics] Pod metrics collection failed (count=%d): %v", count, err)
		if s.OnError != nil {
			s.OnError("metrics", metricsCollectionErrorLevel(err), "pod metrics collection failed (count=%d): %v", count, err)
		}
	}
}

func (s *MetricsHistoryStore) recordPartialPodCollectionError(err error) {
	s.mu.Lock()
	s.partialPodErrors++
	count := s.partialPodErrors
	s.mu.Unlock()
	if count == 1 || count%20 == 0 {
		log.Printf("[metrics] Pod metrics collection failed in some namespaces (count=%d): %v", count, err)
		// Diagnostics are readable by every signed-in user, so they get no
		// namespace names; the server log above has the detail.
		if s.OnError != nil {
			s.OnError("metrics", metricsCollectionErrorLevel(err), "pod metrics collection failed in some namespaces (count=%d); see the Radar log", count)
		}
	}
}

func (s *MetricsHistoryStore) recordNodeCollectionError(err error) {
	errMsg := err.Error()
	s.mu.Lock()
	s.consecutiveNodeErrors++
	s.lastNodeError = errMsg
	shouldReport := s.consecutiveNodeErrors == 1 || s.consecutiveNodeErrors%20 == 0
	count := s.consecutiveNodeErrors
	s.mu.Unlock()
	if shouldReport {
		log.Printf("[metrics] Node metrics collection failed (count=%d): %v", count, err)
		if s.OnError != nil {
			s.OnError("metrics", metricsCollectionErrorLevel(err), "node metrics collection failed (count=%d): %v", count, err)
		}
	}
}

func (s *MetricsHistoryStore) collectPodMetrics(ctx context.Context, now time.Time) {
	if s.dynClient == nil {
		var shouldReport bool
		var count int
		s.mu.Lock()
		s.consecutivePodErrors++
		s.lastPodError = "dynamic client not initialized"
		shouldReport = s.consecutivePodErrors == 1 || s.consecutivePodErrors%20 == 0
		count = s.consecutivePodErrors
		s.mu.Unlock()
		if shouldReport {
			log.Printf("[metrics] Pod metrics collection failed (count=%d): %s", count, "dynamic client not initialized")
			if s.OnError != nil {
				s.OnError("metrics", "error", "pod metrics collection failed (count=%d): %s", count, "dynamic client not initialized")
			}
		}
		return
	}
	podMetricsGVR, ok := s.metricsGVR("pods")
	if !ok {
		s.recordPodCollectionError(ErrMetricsAPINotDiscovered)
		return
	}

	items, listed, err := s.listPodMetrics(ctx, podMetricsGVR)
	if !listed {
		s.recordPodCollectionError(err)
		return
	}
	if err != nil {
		// Some namespaces failed. The ones that answered are still recorded,
		// and the failure goes to the Radar log only: the collection error is
		// shown with any pod's history, and must not name another namespace
		// to someone who can read only theirs.
		s.recordPartialPodCollectionError(err)
	}

	s.mu.Lock()
	defer s.mu.Unlock()

	if s.consecutivePodErrors > 0 {
		log.Printf("[metrics] Pod metrics collection recovered after %d failures", s.consecutivePodErrors)
	}
	s.consecutivePodErrors = 0
	s.lastPodError = ""
	s.lastSuccessfulPodCollection = now

	for _, item := range items {
		namespace := item.GetNamespace()
		name := item.GetName()
		key := namespace + "/" + name

		podBuf, exists := s.podMetrics[key]
		if !exists {
			podBuf = &podMetricsBuffer{
				namespace:  namespace,
				name:       name,
				containers: make(map[string]*ringBuffer),
			}
			s.podMetrics[key] = podBuf
		}
		podBuf.lastSeen = now

		containers, ok := item.Object["containers"].([]any)
		if !ok {
			continue
		}

		for _, c := range containers {
			container, ok := c.(map[string]any)
			if !ok {
				continue
			}
			containerName, _ := container["name"].(string)
			if containerName == "" {
				continue
			}
			usage, ok := container["usage"].(map[string]any)
			if !ok {
				continue
			}

			containerBuf, exists := podBuf.containers[containerName]
			if !exists {
				containerBuf = newRingBuffer(MetricsHistorySize)
				podBuf.containers[containerName] = containerBuf
			}
			cpuStr, _ := usage["cpu"].(string)
			memStr, _ := usage["memory"].(string)
			containerBuf.Add(MetricsDataPoint{
				Timestamp: now,
				CPU:       parseCPU(cpuStr),
				Memory:    parseMemory(memStr),
			})
		}
	}

	staleThreshold := now.Add(-5 * time.Minute)
	for key, podBuf := range s.podMetrics {
		if podBuf.lastSeen.Before(staleThreshold) {
			delete(s.podMetrics, key)
		}
	}
}

// listPodMetrics lists pod metrics cluster-wide, or per namespace when
// Namespaces is set. listed reports whether any LIST succeeded; err joins the
// failures.
func (s *MetricsHistoryStore) listPodMetrics(ctx context.Context, gvr schema.GroupVersionResource) (items []unstructured.Unstructured, listed bool, err error) {
	var namespaces []string
	if s.Namespaces != nil {
		namespaces = s.Namespaces()
	}
	if len(namespaces) == 0 {
		result, err := s.dynClient.Resource(gvr).List(ctx, metav1.ListOptions{})
		if err != nil {
			return nil, false, err
		}
		return result.Items, true, nil
	}
	results := make([]*unstructured.UnstructuredList, len(namespaces))
	errs := make([]error, len(namespaces))
	var lists errgroup.Group
	lists.SetLimit(8)
	for i, namespace := range namespaces {
		lists.Go(func() error {
			results[i], errs[i] = s.dynClient.Resource(gvr).Namespace(namespace).List(ctx, metav1.ListOptions{})
			if errs[i] != nil {
				errs[i] = fmt.Errorf("namespace %s: %w", namespace, errs[i])
			}
			return nil
		})
	}
	_ = lists.Wait()
	for _, result := range results {
		if result != nil {
			listed = true
			items = append(items, result.Items...)
		}
	}
	err = errors.Join(errs...)
	if !listed && err != nil {
		// The collection error is shown with any pod's history, so it must
		// not name namespaces the reader may not see. The detail goes to the
		// Radar log; the cause is kept for classification.
		log.Printf("[metrics] Pod metrics could not be listed in any namespace: %v", err)
		return nil, false, namespacesFailedError{cause: firstError(errs)}
	}
	return items, listed, err
}

// namespacesFailedError reports that pod metrics failed in every listed
// namespace without naming them.
type namespacesFailedError struct{ cause error }

func (e namespacesFailedError) Error() string {
	// Callers classify the stored text, so a metrics API that is down keeps
	// words they recognise as unavailable.
	if e.metricsUnavailable() {
		return "pod metrics API not available in any namespace; see the Radar log"
	}
	return "pod metrics could not be listed in any namespace; see the Radar log"
}

// metricsUnavailable classifies the cause as a pod metrics error, the way it
// was classified when the joined per-namespace errors were the message.
func (e namespacesFailedError) metricsUnavailable() bool {
	return e.cause != nil && MetricsAPIUnavailable(fmt.Errorf("pod metrics: %w", e.cause))
}

func (e namespacesFailedError) Unwrap() error { return e.cause }

func firstError(errs []error) error {
	for _, err := range errs {
		if err != nil {
			return err
		}
	}
	return nil
}

func (s *MetricsHistoryStore) collectNodeMetrics(ctx context.Context, now time.Time) {
	if s.dynClient == nil {
		var shouldReport bool
		var count int
		s.mu.Lock()
		s.consecutiveNodeErrors++
		s.lastNodeError = "dynamic client not initialized"
		shouldReport = s.consecutiveNodeErrors == 1 || s.consecutiveNodeErrors%20 == 0
		count = s.consecutiveNodeErrors
		s.mu.Unlock()
		if shouldReport {
			log.Printf("[metrics] Node metrics collection failed (count=%d): %s", count, "dynamic client not initialized")
			if s.OnError != nil {
				s.OnError("metrics", "error", "node metrics collection failed (count=%d): %s", count, "dynamic client not initialized")
			}
		}
		return
	}
	nodeMetricsGVR, ok := s.metricsGVR("nodes")
	if !ok {
		s.recordNodeCollectionError(ErrMetricsAPINotDiscovered)
		return
	}

	result, err := s.dynClient.Resource(nodeMetricsGVR).List(ctx, metav1.ListOptions{})
	if err != nil {
		s.recordNodeCollectionError(err)
		return
	}

	s.mu.Lock()
	defer s.mu.Unlock()

	if s.consecutiveNodeErrors > 0 {
		log.Printf("[metrics] Node metrics collection recovered after %d failures", s.consecutiveNodeErrors)
	}
	s.consecutiveNodeErrors = 0
	s.lastNodeError = ""
	s.lastSuccessfulNodeCollection = now

	for _, item := range result.Items {
		name := item.GetName()
		nodeBuf, exists := s.nodeMetrics[name]
		if !exists {
			nodeBuf = &nodeMetricsBuffer{name: name, buffer: newRingBuffer(MetricsHistorySize)}
			s.nodeMetrics[name] = nodeBuf
		}
		nodeBuf.lastSeen = now

		usage, ok := item.Object["usage"].(map[string]any)
		if !ok {
			continue
		}
		cpuStr, _ := usage["cpu"].(string)
		memStr, _ := usage["memory"].(string)
		nodeBuf.buffer.Add(MetricsDataPoint{
			Timestamp: now,
			CPU:       parseCPU(cpuStr),
			Memory:    parseMemory(memStr),
		})
	}

	staleThreshold := now.Add(-5 * time.Minute)
	for name, nodeBuf := range s.nodeMetrics {
		if nodeBuf.lastSeen.Before(staleThreshold) {
			delete(s.nodeMetrics, name)
		}
	}
}

// GetPodMetricsHistory returns historical metrics for a specific pod.
func (s *MetricsHistoryStore) GetPodMetricsHistory(namespace, name string) *PodMetricsHistory {
	if s == nil {
		return nil
	}
	s.mu.RLock()
	defer s.mu.RUnlock()

	podBuf, exists := s.podMetrics[namespace+"/"+name]
	if !exists {
		return nil
	}

	history := &PodMetricsHistory{
		Namespace:  namespace,
		Name:       name,
		Containers: make([]ContainerMetricsHistory, 0, len(podBuf.containers)),
	}
	for containerName, buf := range podBuf.containers {
		history.Containers = append(history.Containers, ContainerMetricsHistory{
			Name:       containerName,
			DataPoints: buf.GetAll(),
		})
	}
	return history
}

// GetNodeMetricsHistory returns historical metrics for a specific node.
func (s *MetricsHistoryStore) GetNodeMetricsHistory(name string) *NodeMetricsHistory {
	if s == nil {
		return nil
	}
	s.mu.RLock()
	defer s.mu.RUnlock()

	nodeBuf, exists := s.nodeMetrics[name]
	if !exists {
		return nil
	}
	return &NodeMetricsHistory{Name: name, DataPoints: nodeBuf.buffer.GetAll()}
}

// GetAllPodMetricsLatest returns the latest metrics for all tracked pods.
func (s *MetricsHistoryStore) GetAllPodMetricsLatest() []TopPodMetrics {
	if s == nil {
		return nil
	}
	s.mu.RLock()
	defer s.mu.RUnlock()

	result := make([]TopPodMetrics, 0, len(s.podMetrics))
	for _, podBuf := range s.podMetrics {
		var totalCPU, totalMem int64
		for _, containerBuf := range podBuf.containers {
			if points := containerBuf.GetAll(); len(points) > 0 {
				last := points[len(points)-1]
				totalCPU += last.CPU
				totalMem += last.Memory
			}
		}
		result = append(result, TopPodMetrics{
			Namespace: podBuf.namespace,
			Name:      podBuf.name,
			CPU:       totalCPU,
			Memory:    totalMem,
		})
	}
	return result
}

// GetAllPodContainerMetricsLatest returns the latest per-container usage for all
// tracked pods, keyed by "namespace/name" then container name. Unlike
// GetAllPodMetricsLatest it does not sum across containers. Only the usage
// fields (CPU, Memory) are populated; request/limit are filled from the pod
// spec by callers.
func (s *MetricsHistoryStore) GetAllPodContainerMetricsLatest() map[string]map[string]ContainerResourceMetrics {
	if s == nil {
		return nil
	}
	s.mu.RLock()
	defer s.mu.RUnlock()

	result := make(map[string]map[string]ContainerResourceMetrics, len(s.podMetrics))
	for key, podBuf := range s.podMetrics {
		containers := make(map[string]ContainerResourceMetrics, len(podBuf.containers))
		for name, buf := range podBuf.containers {
			if points := buf.GetAll(); len(points) > 0 {
				last := points[len(points)-1]
				containers[name] = ContainerResourceMetrics{
					Name:   name,
					CPU:    last.CPU,
					Memory: last.Memory,
				}
			}
		}
		result[key] = containers
	}
	return result
}

// GetAllNodeMetricsLatest returns the latest metrics for all tracked nodes.
func (s *MetricsHistoryStore) GetAllNodeMetricsLatest() []TopNodeMetrics {
	if s == nil {
		return nil
	}
	s.mu.RLock()
	defer s.mu.RUnlock()

	result := make([]TopNodeMetrics, 0, len(s.nodeMetrics))
	for _, nodeBuf := range s.nodeMetrics {
		if points := nodeBuf.buffer.GetAll(); len(points) > 0 {
			last := points[len(points)-1]
			result = append(result, TopNodeMetrics{Name: nodeBuf.name, CPU: last.CPU, Memory: last.Memory, ObservedAt: last.Timestamp})
		}
	}
	return result
}

// CollectionHealth returns the current health of metrics collection.
func (s *MetricsHistoryStore) CollectionHealth() MetricsCollectionHealth {
	if s == nil {
		return MetricsCollectionHealth{}
	}
	s.mu.RLock()
	defer s.mu.RUnlock()

	podDataPoints := 0
	for _, podBuf := range s.podMetrics {
		for _, containerBuf := range podBuf.containers {
			podDataPoints += containerBuf.count
		}
	}
	nodeDataPoints := 0
	for _, nodeBuf := range s.nodeMetrics {
		nodeDataPoints += nodeBuf.buffer.count
	}

	health := MetricsCollectionHealth{
		PodMetrics: MetricsSourceHealth{
			Collecting:        s.consecutivePodErrors == 0 && !s.lastSuccessfulPodCollection.IsZero(),
			ConsecutiveErrors: s.consecutivePodErrors,
			LastError:         s.lastPodError,
			TrackedCount:      len(s.podMetrics),
			TotalDataPoints:   podDataPoints,
		},
		NodeMetrics: MetricsSourceHealth{
			Collecting:        s.consecutiveNodeErrors == 0 && !s.lastSuccessfulNodeCollection.IsZero(),
			ConsecutiveErrors: s.consecutiveNodeErrors,
			LastError:         s.lastNodeError,
			TrackedCount:      len(s.nodeMetrics),
			TotalDataPoints:   nodeDataPoints,
		},
		TotalCollections: s.totalCollections,
		BufferSize:       MetricsHistorySize,
		PollIntervalSec:  int(MetricsPollInterval.Seconds()),
	}

	if !s.lastSuccessfulPodCollection.IsZero() {
		health.PodMetrics.LastSuccess = s.lastSuccessfulPodCollection.Format(time.RFC3339)
	}
	if !s.lastSuccessfulNodeCollection.IsZero() {
		health.NodeMetrics.LastSuccess = s.lastSuccessfulNodeCollection.Format(time.RFC3339)
	}
	if !s.lastCollectionAttempt.IsZero() {
		health.LastAttempt = s.lastCollectionAttempt.Format(time.RFC3339)
	}
	return health
}

// parseCPU converts a Kubernetes CPU quantity to nanocores, rounding up.
// Unparseable input reads as 0.
func parseCPU(s string) int64 {
	q, err := resource.ParseQuantity(s)
	if err != nil {
		return 0
	}
	return q.ScaledValue(resource.Nano)
}

// parseMemory converts a Kubernetes memory quantity to bytes, rounding up.
// Unparseable input reads as 0.
func parseMemory(s string) int64 {
	q, err := resource.ParseQuantity(s)
	if err != nil {
		return 0
	}
	return q.Value()
}