pkg/prom/workload_history.go

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

package prom

import (
	"fmt"
	"strconv"
	"strings"
	"time"
)

type WorkloadHistory struct {
	Kind, Namespace, Name string
	Scope                 WorkloadMetricsScope
	Recorded              bool
}

func (h WorkloadHistory) OwnerQuery() (string, error) {
	cluster, err := h.Scope.Matchers()
	if err != nil {
		return "", err
	}
	if h.Namespace == "" || h.Name == "" {
		return "", fmt.Errorf("workload history requires namespace and name")
	}
	kind := ""
	switch strings.ToLower(h.Kind) {
	case "deployment":
		kind = "Deployment"
	case "statefulset":
		kind = "StatefulSet"
	case "daemonset":
		kind = "DaemonSet"
	default:
		return "", fmt.Errorf("unsupported historical workload kind %q", h.Kind)
	}
	ns := "namespace=" + strconv.Quote(h.Namespace)
	metric := func(name, target string) string { return workloadMetricSelector(name, cluster, target) }
	if h.Recorded {
		return "max by (namespace,pod) (" + metric("namespace_workload_pod:kube_pod_owner:relabel", ns+",workload="+strconv.Quote(h.Name)+",workload_type=~"+strconv.Quote(strings.ToLower(kind)+"|"+kind)) + ")", nil
	}
	if kind == "Deployment" {
		return `label_replace(max by (namespace,pod,owner_name) (` + metric("kube_pod_owner", ns+`,owner_kind="ReplicaSet",owner_is_controller="true"`) + `), "replicaset", "$1", "owner_name", "(.*)") * on (namespace,replicaset) group_left() max by (namespace,replicaset) (` + metric("kube_replicaset_owner", ns+`,owner_kind="Deployment",owner_name=`+strconv.Quote(h.Name)+`,owner_is_controller="true"`) + `)`, nil
	}
	return "max by (namespace,pod) (" + metric("kube_pod_owner", ns+",owner_kind="+strconv.Quote(kind)+",owner_name="+strconv.Quote(h.Name)+`,owner_is_controller="true"`) + ")", nil
}

func (h WorkloadHistory) expression(metric, target, window string, beyla bool) (string, error) {
	owner, err := h.OwnerQuery()
	if err != nil {
		return "", err
	}
	cluster, _ := h.Scope.Matchers()
	expr := workloadMetricExpression(metric, cluster, target, nil, window)
	if beyla {
		expr = `label_replace(label_replace(` + expr + `,"namespace","$1","k8s_namespace_name","(.*)"),"pod","$1","k8s_pod_name","(.*)")`
	}
	return "(" + expr + ") and on (namespace,pod) (" + owner + ")", nil
}

func BuildHistoryResourceQuery(step time.Duration, h WorkloadHistory, category MetricCategory) (string, error) {
	target := "namespace=" + strconv.Quote(h.Namespace) + `,container!="",container!="POD"`
	metric, window := "container_memory_working_set_bytes", ""
	if category == CategoryCPU {
		metric, window = "container_cpu_usage_seconds_total", WorkloadRateWindow(step).String()
	} else if category != CategoryMemory {
		return "", fmt.Errorf("unsupported workload history category %q", category)
	}
	expr, err := h.expression(metric, target, window, false)
	if err != nil {
		return "", err
	}
	pods := "sum by (pod) (max by (pod,container) (" + expr + "))"
	return workloadAggregateLines("sum("+pods+")", "max("+pods+")"), nil
}

func BuildHistoryThrottleQuery(step time.Duration, h WorkloadHistory) (string, error) {
	target := "namespace=" + strconv.Quote(h.Namespace) + `,container!="",container!="POD"`
	numerator, err := h.expression("container_cpu_cfs_throttled_periods_total", target, WorkloadRateWindow(step).String(), false)
	if err != nil {
		return "", err
	}
	denominator, _ := h.expression("container_cpu_cfs_periods_total", target, WorkloadRateWindow(step).String(), false)
	numerator = "sum by (pod) (max by (pod,container) (" + numerator + "))"
	denominator = "sum by (pod) (max by (pod,container) (" + denominator + "))"
	return workloadAggregateLines("100 * sum("+numerator+") / sum("+denominator+")", "max(100 * ("+numerator+") / ("+denominator+"))"), nil
}

func workloadAggregateLines(total, maximum string) string {
	return `label_replace((` + total + `),"aggregation","Workload","",".*") or label_replace((` + maximum + `),"aggregation","Maximum Pod","",".*")`
}

func BuildHistoryResourcePopulationQuery(step time.Duration, h WorkloadHistory, family string) (string, error) {
	metrics := map[string][]string{
		"cpu":        {"container_cpu_usage_seconds_total"},
		"memory":     {"container_memory_working_set_bytes"},
		"throttling": {"container_cpu_cfs_periods_total", "container_cpu_cfs_throttled_periods_total"},
	}[family]
	if len(metrics) == 0 {
		return "", fmt.Errorf("unsupported resource family %q", family)
	}
	window := WorkloadRateWindow(step).String()
	if family == "memory" {
		window = ""
	}
	parts := make([]string, 0, len(metrics))
	for _, metric := range metrics {
		expr, err := h.expression(metric, "namespace="+strconv.Quote(h.Namespace)+`,container!="",container!="POD"`, window, false)
		if err != nil {
			return "", err
		}
		parts = append(parts, expr)
	}
	return "count(count by (job,replica,prometheus_replica,__replica__) (" + strings.Join(parts, " or ") + "))", nil
}