internal/waste/meters.go

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

package waste

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

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

const (
	liveSource    = "metrics-server"
	liveWindow    = "right now"
	promSource    = "prometheus"
	sampledSource = "spinoza's own samples"

	windowPodBudget = 200
	windowReaders   = 8
	milliPerCore    = 1000
)

type Sampler interface {
	Metrics(ctx context.Context) api.Metrics
}

type Historian interface {
	MetricHistory(ctx context.Context, namespace, pod string, span time.Duration) (api.MetricHistory, error)
}

type live struct {
	from Sampler
}

func RightNow(from Sampler) Meter {
	return &live{from: from}
}

func (lv *live) Usage(ctx context.Context, pods []Ref) (Reading, error) {
	held := lv.from.Metrics(ctx)
	if held.Error != "" {
		return Reading{}, errors.New(held.Error)
	}
	out := make(map[string]api.ResourceUsage, len(pods))
	for _, ref := range pods {
		key := ref.Namespace + "/" + ref.Name
		use, ok := held.Pods[key]
		if !ok {
			continue
		}
		out[key] = use
	}
	if len(out) == 0 {
		return Reading{}, errors.New("metrics-server named none of the pods this asked about")
	}
	return Reading{Pods: out, Source: liveSource, Window: liveWindow}, nil
}

type window struct {
	from   Historian
	span   time.Duration
	budget int
}

func OverAWindow(from Historian, span time.Duration) Meter {
	return &window{from: from, span: span, budget: windowPodBudget}
}

type sampled struct {
	key    string
	source string
	use    api.ResourceUsage
	ok     bool
	err    error
}

func (wd *window) Usage(ctx context.Context, pods []Ref) (Reading, error) {
	if len(pods) > wd.budget {
		return Reading{}, fmt.Errorf(
			"a window is read one pod at a time, and this asks about %d pods against the %d it will read; "+
				"ask about one namespace",
			len(pods), wd.budget,
		)
	}
	out := Reading{Pods: make(map[string]api.ResourceUsage, len(pods)), Window: windowWords(wd.span)}
	var last error
	for _, one := range wd.readAll(ctx, pods) {
		if !one.ok {
			last = one.err
			continue
		}
		if out.Source == "" {
			out.Source = one.source
		}
		out.Pods[one.key] = one.use
	}
	if len(out.Pods) > 0 {
		return out, nil
	}
	if last == nil {
		last = errors.New("no pod had a history to read")
	}
	return Reading{}, last
}

func (wd *window) readAll(ctx context.Context, pods []Ref) []sampled {
	out := make([]sampled, len(pods))
	gate := make(chan struct{}, windowReaders)
	var group sync.WaitGroup
	group.Add(len(pods))
	for at, ref := range pods {
		gate <- struct{}{}
		safe.Go("reading a pod's usage over a window", func() {
			defer group.Done()
			defer func() { <-gate }()
			out[at] = wd.one(ctx, ref)
		})
	}
	group.Wait()
	return out
}

func (wd *window) one(ctx context.Context, ref Ref) sampled {
	key := ref.Namespace + "/" + ref.Name
	history, err := wd.from.MetricHistory(ctx, ref.Namespace, ref.Name, wd.span)
	if err != nil {
		return sampled{err: err}
	}
	if len(history.CPU) == 0 && len(history.Memory) == 0 {
		return sampled{err: errors.New(key + " has no measured history")}
	}
	return sampled{key: key, source: sourceOf(history), use: meanOf(history), ok: true}
}

func sourceOf(history api.MetricHistory) string {
	if history.Sampled {
		return sampledSource
	}
	if history.Source != "" {
		return history.Source
	}
	return promSource
}

func meanOf(history api.MetricHistory) api.ResourceUsage {
	return api.ResourceUsage{
		CPUMilli: int64(mean(history.CPU) * milliPerCore),
		MemoryMi: int64(mean(history.Memory)) / bytesPerMi,
	}
}

func mean(points []api.MetricPoint) float64 {
	if len(points) == 0 {
		return 0
	}
	var total float64
	for _, point := range points {
		total += point.Value
	}
	return total / float64(len(points))
}

func windowWords(span time.Duration) string {
	if span <= 0 {
		return liveWindow
	}
	seconds := int64(span / time.Second)
	if seconds < 60 {
		return "the last " + counted(seconds, "second")
	}
	minutes := seconds / 60
	if minutes < 60 {
		return "the last " + counted(minutes, "minute")
	}
	hours := minutes / 60
	if hours < 24 {
		return "the last " + counted(hours, "hour")
	}
	return "the last " + counted(hours/24, "day")
}

func counted(count int64, noun string) string {
	if count == 1 {
		return noun
	}
	return strconv.FormatInt(count, 10) + " " + noun + "s"
}