internal/metrics/metrics.go

Rama main · réplica de lectura del repositorio original.

package metrics

import (
	"context"
	"sync"
	"sync/atomic"
	"time"

	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"

	"github.com/sophotechlabs/spinoza/internal/api"
	"github.com/sophotechlabs/spinoza/internal/listerr"
	"github.com/sophotechlabs/spinoza/internal/safe"
	"github.com/sophotechlabs/spinoza/internal/unstr"
)

var (
	podMetricsGVR  = schema.GroupVersionResource{Group: "metrics.k8s.io", Version: "v1beta1", Resource: "pods"}
	nodeMetricsGVR = schema.GroupVersionResource{Group: "metrics.k8s.io", Version: "v1beta1", Resource: "nodes"}
	nodeGVR        = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "nodes"}
)

var buildTimeout = 20 * time.Second

type Nodes interface {
	List(ctx context.Context) ([]*unstructured.Unstructured, error)
}

type dynamicNodes struct {
	dyn dynamic.Interface
}

func FromCluster(dyn dynamic.Interface) Nodes {
	return dynamicNodes{dyn: dyn}
}

func (d dynamicNodes) List(ctx context.Context) ([]*unstructured.Unstructured, error) {
	list, err := d.dyn.Resource(nodeGVR).List(ctx, metav1.ListOptions{})
	if err != nil {
		return nil, err
	}
	out := make([]*unstructured.Unstructured, 0, len(list.Items))
	for i := range list.Items {
		out = append(out, &list.Items[i])
	}
	return out, nil
}

func Build(ctx context.Context, dyn dynamic.Interface, nodes Nodes) api.Metrics {
	bounded, cancel := context.WithTimeout(ctx, buildTimeout)
	defer cancel()
	ctx = bounded
	failures := listerr.New()
	var podsMissing atomic.Bool
	var nodesMissing atomic.Bool
	var pods map[string]api.ResourceUsage
	var used map[string]api.ResourceUsage
	var group sync.WaitGroup
	group.Add(2)
	safe.Go("reading pod metrics", func() {
		defer group.Done()
		defer func() {
			failures.RecordPanic("pods.metrics.k8s.io", "reading pod metrics", recover())
		}()
		pods = podUsage(ctx, dyn, "", failures, &podsMissing)
	})
	safe.Go("reading node metrics", func() {
		defer group.Done()
		defer func() {
			failures.RecordPanic("nodes.metrics.k8s.io", "reading node metrics", recover())
		}()
		used = nodeUsage(ctx, dyn, nodes, failures, &nodesMissing)
	})
	group.Wait()
	return api.Metrics{
		Pods:   pods,
		Nodes:  used,
		Absent: podsMissing.Load() && nodesMissing.Load(),
		Error:  failures.Message(),
	}
}

func BuildIn(ctx context.Context, dyn dynamic.Interface, namespace string) api.Metrics {
	bounded, cancel := context.WithTimeout(ctx, buildTimeout)
	defer cancel()
	failures := listerr.New()
	var missing atomic.Bool
	pods := podUsage(bounded, dyn, namespace, failures, &missing)
	return api.Metrics{
		Pods:   pods,
		Nodes:  map[string]api.ResourceUsage{},
		Absent: missing.Load(),
		Error:  failures.Message(),
	}
}

func NotServed(err error) bool {
	return apierrors.IsNotFound(err)
}

func podUsage(
	ctx context.Context,
	dyn dynamic.Interface,
	namespace string,
	failures *listerr.Collector,
	missing *atomic.Bool,
) map[string]api.ResourceUsage {
	out := map[string]api.ResourceUsage{}
	list, err := dyn.Resource(podMetricsGVR).Namespace(namespace).List(ctx, metav1.ListOptions{})
	if NotServed(err) {
		missing.Store(true)
		return out
	}
	failures.Record("pods.metrics.k8s.io", err)
	if err != nil {
		return out
	}
	for i := range list.Items {
		u := &list.Items[i]
		cpu, mem := containerTotals(u)
		out[u.GetNamespace()+"/"+u.GetName()] = api.ResourceUsage{CPUMilli: cpu, MemoryMi: mem}
	}
	return out
}

func containerTotals(u *unstructured.Unstructured) (cpuMilli, memMi int64) {
	for _, c := range unstr.Slice(u, "containers") {
		m, ok := c.(map[string]any)
		if !ok {
			continue
		}
		usage, ok := m["usage"].(map[string]any)
		if !ok {
			continue
		}
		cpuMilli += milli(usage)
		memMi += mebi(usage)
	}
	return cpuMilli, memMi
}

func NodeUsage(ctx context.Context, dyn dynamic.Interface) (map[string]api.ResourceUsage, error) {
	list, err := dyn.Resource(nodeMetricsGVR).List(ctx, metav1.ListOptions{})
	if err != nil {
		return nil, err
	}
	out := map[string]api.ResourceUsage{}
	for i := range list.Items {
		obj := &list.Items[i]
		usage, ok := unstr.Map(obj, "usage")
		if !ok {
			continue
		}
		out[obj.GetName()] = api.ResourceUsage{CPUMilli: milli(usage), MemoryMi: mebi(usage)}
	}
	return out, nil
}

func nodeUsage(
	ctx context.Context,
	dyn dynamic.Interface,
	nodes Nodes,
	failures *listerr.Collector,
	missing *atomic.Bool,
) map[string]api.ResourceUsage {
	usage, err := NodeUsage(ctx, dyn)
	if NotServed(err) {
		missing.Store(true)
		return map[string]api.ResourceUsage{}
	}
	failures.Record("nodes.metrics.k8s.io", err)
	if err != nil {
		return map[string]api.ResourceUsage{}
	}
	allocatable := nodeAllocatable(ctx, nodes, failures)
	out := map[string]api.ResourceUsage{}
	for name, use := range usage {
		alloc, ok := allocatable[name]
		if ok {
			use.CPUPercent = percent(use.CPUMilli, alloc.CPUMilli)
			use.MemPercent = percent(use.MemoryMi, alloc.MemoryMi)
			use.CPUAllocatableMilli = alloc.CPUMilli
			use.MemAllocatableMi = alloc.MemoryMi
		}
		out[name] = use
	}
	return out
}

func nodeAllocatable(
	ctx context.Context,
	nodes Nodes,
	failures *listerr.Collector,
) map[string]api.ResourceUsage {
	out := map[string]api.ResourceUsage{}
	found, err := nodes.List(ctx)
	failures.Record("nodes", err)
	if err != nil {
		return out
	}
	for _, u := range found {
		alloc, ok := unstr.Map(u, "status", "allocatable")
		if !ok {
			continue
		}
		out[u.GetName()] = api.ResourceUsage{CPUMilli: milli(alloc), MemoryMi: mebi(alloc)}
	}
	return out
}

func percent(used, total int64) int64 {
	if total <= 0 {
		return 0
	}
	return used * 100 / total
}

func milli(m map[string]any) int64 {
	q, ok := quantity(m, "cpu")
	if !ok {
		return 0
	}
	return q.MilliValue()
}

func mebi(m map[string]any) int64 {
	q, ok := quantity(m, "memory")
	if !ok {
		return 0
	}
	return q.Value() / (1024 * 1024)
}

func quantity(m map[string]any, key string) (resource.Quantity, bool) {
	s, ok := m[key].(string)
	if !ok {
		return resource.Quantity{}, false
	}
	q, err := resource.ParseQuantity(s)
	if err != nil {
		return resource.Quantity{}, false
	}
	return q, true
}