pkg/prom/discovery.go

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

package prom

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
)

// wellKnownConcurrency bounds the parallel Service.Get fan-out over the
// well-known locations so discovery doesn't open a burst of connections to the
// API server at once.
const wellKnownConcurrency = 8

// CandidateSource describes how a candidate was found.
type CandidateSource string

const (
	CandidateSourceWellKnown CandidateSource = "well_known"
	CandidateSourceDynamic   CandidateSource = "dynamic"
)

// Candidate is a Prometheus-compatible service the caller can attempt to
// reach. Discover populates the fields and orders candidates by priority,
// but does not probe — it leaves the transport choice (direct HTTP vs.
// port-forward vs. tunneled proxy) to the caller.
type Candidate struct {
	Namespace   string
	Name        string
	Port        int             // service port (for in-cluster addressing)
	TargetPort  int             // container port (for port-forwarding to the pod)
	ClusterAddr string          // http://{name}.{ns}.svc.cluster.local:{port}
	BasePath    string          // e.g. "/select/0/prometheus" for vmselect
	Score       int             // relative likelihood of being Prometheus
	Source      CandidateSource // well_known | dynamic
}

// DiscoverOptions tunes Discover's behavior.
type DiscoverOptions struct {
	// IncludeDynamic controls whether a cluster-wide service scan is performed.
	// The scan is an O(all services) List call plus a scoring pass; skip it
	// for callers that only need a quick well-known check.
	IncludeDynamic bool

	// MaxDynamic caps the number of dynamic candidates returned. Default 5.
	MaxDynamic int

	// Logger is optional; if set, Discover emits verbose progress messages.
	Logger func(format string, args ...interface{})

	// Services, when set, supplies every Service in the cluster from a
	// caller-held cache (an informer lister), replacing both the well-known
	// Gets and the cluster-wide List with in-memory lookups. It must return
	// ok=false unless the slice is the complete cluster-wide set: a partial
	// view (a namespace-scoped watch, an unsynced cache) would silently drop
	// candidates, so Discover falls back to the API instead. The returned
	// objects are read, never modified.
	Services func() (services []*corev1.Service, ok bool)
}

// wellKnownLoc is a namespace + service name where a Prometheus-compatible
// service is commonly installed.
type wellKnownLoc struct {
	Namespace string
	Name      string
	Port      int    // 0 = use service's first port
	BasePath  string // sub-path for Prometheus API
}

// WellKnownLocations is the ordered list of primary Prometheus-compatible
// service locations, ranked ahead of dynamically-discovered candidates.
var WellKnownLocations = []wellKnownLoc{
	// VictoriaMetrics — monitoring namespace first (workload metrics)
	{"monitoring", "victoria-metrics-victoria-metrics-single-server", 8428, ""},
	{"monitoring", "victoria-metrics-single-server", 8428, ""},
	{"monitoring", "vmsingle", 8428, ""},
	{"monitoring", "vmselect", 8481, "/select/0/prometheus"},
	{"victoria-metrics", "victoria-metrics-victoria-metrics-single-server", 8428, ""},
	{"victoria-metrics", "victoria-metrics-single-server", 8428, ""},
	{"victoria-metrics", "vmsingle", 8428, ""},
	{"victoria-metrics", "vmselect", 8481, "/select/0/prometheus"},
	// kube-prometheus-stack
	{"monitoring", "kube-prometheus-stack-prometheus", 9090, ""},
	{"monitoring", "prometheus-kube-prometheus-prometheus", 9090, ""},
	{"monitoring", "prometheus-operated", 9090, ""},
	// Standard Prometheus
	{"opencost", "prometheus-server", 0, ""},
	{"monitoring", "prometheus-server", 0, ""},
	{"prometheus", "prometheus-server", 0, ""},
	{"observability", "prometheus-server", 0, ""},
	{"metrics", "prometheus-server", 0, ""},
	{"kube-system", "prometheus", 0, ""},
	{"default", "prometheus", 0, ""},
}

// fallbackWellKnownLocations are last-resort well-known metrics services, ranked
// AFTER dynamically-discovered candidates. caretta-vm serves the traffic source's
// eBPF link metrics and commonly lacks cluster workload (node/pod) metrics, so it
// must never outrank a real Prometheus — whether that Prometheus is in the primary
// well-known list or is only turned up by the dynamic scan. It is still returned
// as a candidate so a cluster whose only metrics backend is caretta-vm keeps working.
var fallbackWellKnownLocations = []wellKnownLoc{
	{"caretta", "caretta-vm", 8428, ""},
}

// metricsNamespaces are commonly used for metrics services; used as a scoring
// signal in dynamic discovery.
var metricsNamespaces = map[string]bool{
	"monitoring":       true,
	"prometheus":       true,
	"observability":    true,
	"metrics":          true,
	"victoria-metrics": true,
	"caretta":          true,
	"opencost":         true,
}

// skipNamespaces are excluded from dynamic discovery.
var skipNamespaces = map[string]bool{
	"kube-public":     true,
	"kube-node-lease": true,
}

// nonQueryComponents are metrics-adjacent services that are commonly named or
// namespaced like Prometheus but do NOT serve the Prometheus HTTP query API
// (/api/v1/query). Discovery must never treat them as candidates: each one
// costs a serial port-forward attempt that can only ever fail. Matched as a
// substring against the service name and the app/name/component labels, so
// e.g. "kube-prometheus-stack-prometheus-node-exporter" is excluded by
// "node-exporter" even though it also contains "prometheus".
var nonQueryComponents = []string{
	"node-exporter",
	"kube-state-metrics",
	"state-metrics",
	"alertmanager",
	"pushgateway",
	"prometheus-adapter",
	"grafana",
	"blackbox",
	"thanos-compact",
	"thanos-rule", // covers thanos-ruler
	"thanos-store",
	"thanos-sidecar",
	"thanos-receive",
	// Control-plane scrape-target Services created by kube-prometheus-stack:
	// they carry "prometheus" in the name but only proxy a component's /metrics,
	// they don't serve the query API.
	"kubelet",
	"kube-etcd",
	"kube-scheduler",
	"kube-controller-manager",
	"kube-proxy",
	"coredns",
	"kube-dns",
}

// queryEngineAppNames / queryEngineAppLabels are definitive identities of a
// Prometheus-family query engine. When one is present, the substring deny-list
// below must not fire — a real endpoint (e.g. a chart installed with release
// name "grafana", giving service name "grafana-prometheus-server", but
// app.kubernetes.io/name=prometheus) is a query API despite the generic token.
var queryEngineAppNames = map[string]bool{
	"prometheus": true, "victoria-metrics-single": true, "vmsingle": true,
	"vmselect": true, "thanos-query": true, "thanos-querier": true,
}
var queryEngineAppLabels = map[string]bool{
	"prometheus": true, "prometheus-server": true, "vmsingle": true, "vmselect": true,
}

// isNonQueryService reports whether a service is a known non-query metrics
// component (exporter, operator, alertmanager, …) that cannot answer PromQL.
func isNonQueryService(svc corev1.Service) bool {
	labels := svc.Labels
	name := strings.ToLower(svc.Name)
	appName := strings.ToLower(labels["app.kubernetes.io/name"])
	component := strings.ToLower(labels["app.kubernetes.io/component"])

	// A definitive query-engine identity wins over any name substring.
	if queryEngineAppNames[appName] || queryEngineAppLabels[strings.ToLower(labels["app"])] {
		return false
	}

	// The Prometheus operator is a control plane, not a query API. Match it by
	// name suffix or exact label rather than a substring, so a real Prometheus
	// whose name merely contains "operator" — e.g. "prometheus-operator-prometheus"
	// from a release named "prometheus-operator" — is not excluded.
	if strings.HasSuffix(name, "operator") || appName == "prometheus-operator" || component == "prometheus-operator" {
		return true
	}

	haystack := []string{name, appName, component, strings.ToLower(labels["app"])}
	for _, token := range nonQueryComponents {
		for _, h := range haystack {
			if h != "" && strings.Contains(h, token) {
				return true
			}
		}
	}
	return false
}

// Discover enumerates candidate Prometheus-compatible services reachable to
// the given k8sClient. Well-known locations are returned first in declared
// priority order, optionally followed by dynamically-discovered services
// ranked by ScoreService.
//
// Discover does NOT probe any candidate — callers decide how to reach each
// (direct HTTP, port-forward, tunneled proxy) and then use
// pkg/prom.Client.Probe to validate.
func Discover(ctx context.Context, k8sClient kubernetes.Interface, opts DiscoverOptions) ([]Candidate, error) {
	if k8sClient == nil {
		return nil, fmt.Errorf("prom.Discover: k8sClient is nil")
	}
	if opts.MaxDynamic <= 0 {
		opts.MaxDynamic = 5
	}
	logf := opts.Logger
	if logf == nil {
		logf = func(string, ...interface{}) {}
	}

	// cached is the caller's complete cluster-wide Service set, when it has
	// one; every lookup below then reads it instead of the API.
	var cached []*corev1.Service
	fromCache := false
	if opts.Services != nil {
		cached, fromCache = opts.Services()
	}
	var byName map[string]*corev1.Service
	if fromCache {
		byName = make(map[string]*corev1.Service, len(cached))
		for _, svc := range cached {
			byName[svc.Namespace+"/"+svc.Name] = svc
		}
	}
	collect := func(locs []wellKnownLoc, seen map[string]bool) []Candidate {
		if fromCache {
			found := make([]*corev1.Service, len(locs))
			for i, loc := range locs {
				found[i] = byName[loc.Namespace+"/"+loc.Name]
			}
			return assembleWellKnownLocations(locs, found, nil, seen, logf)
		}
		return collectWellKnownLocations(ctx, k8sClient, locs, seen, logf)
	}

	// seen de-duplicates candidates by namespace/name/port/basePath so a
	// service reachable via more than one layer is probed and port-forwarded only
	// once. Earlier layers win ties because they populate seen first.
	seen := map[string]bool{}

	// Layer 1: primary well-known locations, ranked ahead of everything else.
	out := collect(WellKnownLocations, seen)

	// Fallback locations (caretta-vm) are fetched now — so their keys enter `seen`
	// and the dynamic scan below can't re-surface them as ordinary scored
	// candidates — but they are held and appended LAST. caretta-vm serves only the
	// traffic source's link metrics and commonly lacks cluster workload metrics, so
	// a real Prometheus (well-known or dynamically discovered) must always outrank
	// it; it stays a candidate only so a caretta-vm-only cluster keeps working.
	fallback := collect(fallbackWellKnownLocations, seen)

	if !opts.IncludeDynamic {
		return append(out, fallback...), nil
	}

	// Layer 2: dynamic cluster-wide scan. The score only *ranks* candidates; it
	// is not an admission gate. Anything we're certain can't answer PromQL is
	// already excluded by ScoreService (isNonQueryService), and the caller's
	// probe is the authoritative check — so a weak but positive score is kept
	// and ranked low rather than dropped, which would silently miss a real but
	// lightly-labeled Prometheus. MaxDynamic still bounds how many are returned.
	svcs := cached
	if !fromCache {
		list, err := k8sClient.CoreV1().Services("").List(ctx, metav1.ListOptions{})
		if err != nil {
			// The cluster-wide list failed, but the well-known + fallback Gets above
			// still succeeded — return them (including the caretta-vm fallback) rather
			// than dropping the fallback tier. This is the get-but-not-list RBAC case.
			logf("prom.Discover: failed to list services: %v", err)
			return append(out, fallback...), nil
		}
		svcs = make([]*corev1.Service, len(list.Items))
		for i := range list.Items {
			svcs[i] = &list.Items[i]
		}
	}

	// Services already claimed by the fallback tier must never resurface among the
	// dynamic candidates — that would rank caretta-vm by score instead of forcing
	// it last. Match by namespace/name (not the port-keyed `seen`), so a caretta-vm
	// Service fronted by a non-8428 service port is still suppressed here.
	fallbackNames := make(map[string]bool, len(fallback))
	for _, c := range fallback {
		fallbackNames[c.Namespace+"/"+c.Name] = true
	}

	var scored []Candidate
	for _, svc := range svcs {
		if fallbackNames[svc.Namespace+"/"+svc.Name] {
			continue
		}
		score, bp, identity := ScoreService(*svc)
		// Admit on a Prometheus-family identity signal, not on score alone: a
		// namespace-only match (score without identity) would send the operator's
		// configured Prometheus headers to an unrelated neighbor when probed.
		if !identity {
			continue
		}
		port := resolvePort(*svc, 0)
		if seen[candidateKey(svc.Namespace, svc.Name, port, bp)] {
			continue
		}
		scored = append(scored, Candidate{
			Namespace:   svc.Namespace,
			Name:        svc.Name,
			Port:        port,
			TargetPort:  resolveTargetPort(*svc, port),
			ClusterAddr: buildClusterAddr(svc.Name, svc.Namespace, port),
			BasePath:    bp,
			Score:       score,
			Source:      CandidateSourceDynamic,
		})
	}

	// Equal scores order by namespace/name so the MaxDynamic cut keeps the same
	// candidates whatever order the Services arrived in (an API list is sorted,
	// an informer lister is not).
	sort.Slice(scored, func(i, j int) bool {
		if scored[i].Score != scored[j].Score {
			return scored[i].Score > scored[j].Score
		}
		if scored[i].Namespace != scored[j].Namespace {
			return scored[i].Namespace < scored[j].Namespace
		}
		return scored[i].Name < scored[j].Name
	})

	if len(scored) > opts.MaxDynamic {
		scored = scored[:opts.MaxDynamic]
	}
	out = append(out, scored...)

	// Fallback tier last: after both the primary well-known list and the dynamic
	// scan, so a real Prometheus always wins over caretta-vm when one exists.
	out = append(out, fallback...)
	return out, nil
}

// IsCandidateService reports whether Discover would return svc as a
// candidate: it sits at a well-known location, or carries a Prometheus-family
// identity signal. Callers watching Services use it to tell when a newly
// created one is worth a discovery run.
func IsCandidateService(svc *corev1.Service) bool {
	if svc == nil {
		return false
	}
	for _, locs := range [][]wellKnownLoc{WellKnownLocations, fallbackWellKnownLocations} {
		for _, loc := range locs {
			if loc.Namespace == svc.Namespace && loc.Name == svc.Name {
				return true
			}
		}
	}
	_, _, identity := ScoreService(*svc)
	return identity
}

// collectWellKnownLocations fetches the given well-known locations concurrently
// (targeted Gets, not a cluster-wide List, so it works for callers with
// get-but-not-list on Services) and returns the ones that exist as candidates in
// declared order. Locations whose key is already in seen are skipped; newly
// added keys are recorded so later layers de-duplicate against them.
func collectWellKnownLocations(ctx context.Context, k8sClient kubernetes.Interface, locs []wellKnownLoc, seen map[string]bool, logf func(string, ...interface{})) []Candidate {
	found := make([]*corev1.Service, len(locs))
	getErrs := make([]error, len(locs))
	sem := make(chan struct{}, wellKnownConcurrency)
	var wg sync.WaitGroup
	for i, loc := range locs {
		wg.Go(func() {
			sem <- struct{}{}
			defer func() { <-sem }()
			svc, err := k8sClient.CoreV1().Services(loc.Namespace).Get(ctx, loc.Name, metav1.GetOptions{})
			if err != nil {
				getErrs[i] = err
				return
			}
			found[i] = svc
		})
	}
	wg.Wait()
	return assembleWellKnownLocations(locs, found, getErrs, seen, logf)
}

// assembleWellKnownLocations turns the Services found at each location (nil
// when absent) into candidates in declared order. Logging happens here,
// serially — DiscoverOptions.Logger has no concurrency contract. getErrs may
// be nil when the lookups could not fail.
func assembleWellKnownLocations(locs []wellKnownLoc, found []*corev1.Service, getErrs []error, seen map[string]bool, logf func(string, ...interface{})) []Candidate {
	var out []Candidate
	for i, loc := range locs {
		svc := found[i]
		if svc == nil {
			if getErrs != nil {
				if err := getErrs[i]; err != nil && !apierrors.IsNotFound(err) {
					logf("prom.Discover: error checking %s/%s: %v", loc.Namespace, loc.Name, err)
				}
			}
			continue
		}
		port := resolvePort(*svc, loc.Port)
		key := candidateKey(svc.Namespace, svc.Name, port, loc.BasePath)
		if seen[key] {
			continue
		}
		seen[key] = true
		out = append(out, Candidate{
			Namespace:   svc.Namespace,
			Name:        svc.Name,
			Port:        port,
			TargetPort:  resolveTargetPort(*svc, port),
			ClusterAddr: buildClusterAddr(svc.Name, svc.Namespace, port),
			BasePath:    loc.BasePath,
			Source:      CandidateSourceWellKnown,
		})
	}
	return out
}

// promPortScore returns the heuristic weight for a port number matching a
// well-known Prometheus-family default (0 for anything else).
func promPortScore(port int32) int {
	switch port {
	case 9090: // Prometheus default
		return 30
	case 8428: // VictoriaMetrics single-node default
		return 30
	case 8481: // VictoriaMetrics vmselect default
		return 25
	case 9009, 10902: // Thanos Query (gRPC-era default; common HTTP port)
		return 25
	}
	return 0
}

// ScoreService computes a heuristic score for a service being
// Prometheus-compatible. It returns the ranking score, an inferred BasePath for
// vmselect-style services, and whether the service carries a Prometheus-family
// *identity* signal (a name, label, or port-NAME match — NOT a bare port number).
// identity gates admission: a matched port number and namespace membership
// contribute to the score for ranking but must not, on their own, make an
// unrelated Service a candidate — discovery probes candidates with the operator's
// configured Prometheus headers, so a neighbor that merely shares a popular port
// (9090 is a common gRPC/plugin default too) or a namespace must not be admitted.
func ScoreService(svc corev1.Service) (score int, basePath string, identity bool) {
	if svc.Spec.Type == corev1.ServiceTypeExternalName {
		return 0, "", false
	}
	if skipNamespaces[svc.Namespace] {
		return 0, "", false
	}
	if isNonQueryService(svc) {
		return 0, "", false
	}

	labels := svc.Labels
	appName := labels["app.kubernetes.io/name"]
	appLabel := labels["app"]
	component := labels["app.kubernetes.io/component"]

	switch appName {
	case "prometheus":
		score += 100
		identity = true
	case "victoria-metrics-single", "vmsingle":
		score += 100
		identity = true
	case "vmselect":
		score += 90
		basePath = "/select/0/prometheus"
		identity = true
	case "thanos-query", "thanos-querier":
		score += 80
		identity = true
	}

	switch appLabel {
	case "prometheus", "prometheus-server":
		score += 80
		identity = true
	case "vmsingle":
		score += 80
		identity = true
	case "vmselect":
		score += 80
		basePath = "/select/0/prometheus"
		identity = true
	}

	if score > 0 && component == "server" {
		score += 20
	}

	for _, p := range svc.Spec.Ports {
		// Credit a recognized port on either the service port or the container
		// targetPort — a real Prometheus is often fronted by a service port of
		// 80/http with targetPort 9090 — but only once per port entry. A bare port
		// NUMBER is a ranking signal only: it must NOT set identity, because
		// popular defaults like 9090 are shared with unrelated components (e.g. a
		// gRPC plugin), and admitting on the number alone sends the operator's
		// Prometheus headers to — and wastes a doomed port-forward on — that
		// neighbor. A port explicitly NAMED "prometheus" is a genuine family signal.
		if s := promPortScore(p.Port); s > 0 {
			score += s
		} else {
			score += promPortScore(p.TargetPort.IntVal)
		}
		if strings.Contains(strings.ToLower(p.Name), "prometheus") {
			score += 10
			identity = true
		}
	}

	nameLower := strings.ToLower(svc.Name)
	if strings.Contains(nameLower, "prometheus") {
		score += 20
		identity = true
	}
	if strings.Contains(nameLower, "victoria") || strings.Contains(nameLower, "vmsingle") || strings.Contains(nameLower, "vmselect") {
		score += 20
		identity = true
		if strings.Contains(nameLower, "vmselect") && basePath == "" {
			basePath = "/select/0/prometheus"
		}
	}
	if strings.Contains(nameLower, "thanos") {
		score += 15
		identity = true
	}

	// identity is set above only by a name, label, or port-name signal — genuine
	// Prometheus-family membership. A bare port number and the metrics-namespace
	// bonus below rank a candidate but must never admit one on their own, so the
	// caller never probes (and never sends the operator's configured Prometheus
	// headers to) an unrelated neighbor that merely shares port 9090 or a namespace.
	if metricsNamespaces[svc.Namespace] {
		score += 10
	}

	return score, basePath, identity
}

// Key identifies the backend independently of its direct or forwarded address.
func (c Candidate) Key() string {
	return candidateKey(c.Namespace, c.Name, c.Port, c.BasePath)
}

func candidateKey(namespace, name string, port int, basePath string) string {
	return fmt.Sprintf("%s/%s:%d%s", namespace, name, port, basePath)
}

func resolvePort(svc corev1.Service, defaultPort int) int {
	if defaultPort != 0 {
		return defaultPort
	}
	if len(svc.Spec.Ports) > 0 {
		return int(svc.Spec.Ports[0].Port)
	}
	return 80
}

// resolveTargetPort returns the container port, for port-forwarding which
// bypasses the Service. When the service port differs from the container's
// targetPort (e.g., service:80 → container:9090), port-forward needs the
// container port.
func resolveTargetPort(svc corev1.Service, servicePort int) int {
	for _, p := range svc.Spec.Ports {
		if int(p.Port) == servicePort {
			if p.TargetPort.IntVal > 0 {
				return int(p.TargetPort.IntVal)
			}
			return servicePort
		}
	}
	return servicePort
}

// buildClusterAddr returns the in-cluster HTTP URL for a service.
// A headless Service publishes A records for its ready pods under its own name,
// so the plain service address works for both kinds. Addressing a headless one as
// {service}-0.{service} assumes the StatefulSet is named after the Service: true
// for caretta-vm, false for kube-prometheus-stack's prometheus-operated, whose
// pod is prometheus-{release}-kube-prometheus-stack-prometheus-0. That guess made
// a reachable backend look unreachable in-cluster.
func buildClusterAddr(name, namespace string, port int) string {
	return fmt.Sprintf("http://%s.%s.svc.cluster.local:%d", name, namespace, port)
}