pkg/k8score/metrics_history.go
Rama main · réplica de lectura del repositorio original.
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()
}