pkg/ownership/cachesource/dependents.go

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

package cachesource

import (
	"cmp"
	"slices"

	apimeta "k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	apiruntime "k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"

	"github.com/skyhook-io/radar/pkg/k8score"
	"github.com/skyhook-io/radar/pkg/ownership"
	"github.com/skyhook-io/radar/pkg/readreason"
	"github.com/skyhook-io/radar/pkg/resourceid"
)

// candidate is a cached object an owner index returned, not yet copied.
type candidate struct {
	ref resourceid.Ref
	obj apiruntime.Object
}

// search collects one DependentCandidates call's candidates and gaps.
type search struct {
	uid        string
	scope      ownership.SearchScope
	namespaces []string
	found      []candidate
	gaps       []ownership.Gap
}

func (s *session) DependentCandidates(uid types.UID, scope ownership.SearchScope) ownership.DependentCandidatesRead {
	q := &search{uid: string(uid), scope: scope}
	if !scope.AllNamespaces {
		for _, ns := range slices.Compact(slices.Sorted(slices.Values(scope.Namespaces))) {
			if ns == "" {
				// A store read for "" would answer for every namespace.
				q.gaps = append(q.gaps, ownership.Gap{Reason: readreason.NamespaceRequired})
				continue
			}
			q.namespaces = append(q.namespaces, ns)
		}
	}
	searches := scope.Searches(true) || scope.Searches(false)
	if searches {
		s.prepareDependentKinds(scope)
	}
	if !s.catalog.Complete() && searches {
		q.gaps = append(q.gaps, ownership.Gap{Reason: readreason.ReadError})
	}
	var stable []ownership.Gap
	var gapMode uint8
	namespaced := scope.AllNamespaces || len(q.namespaces) > 0
	switch {
	case namespaced && scope.IncludeClusterScoped:
		stable = s.unwatchedAll
		gapMode = 3
	case namespaced:
		stable = s.unwatchedNamespaced
		gapMode = 1
	case scope.IncludeClusterScoped:
		stable = s.unwatchedCluster
		gapMode = 2
	}
	for _, kind := range s.dependentKinds {
		if !scope.Searches(kind.Namespaced) {
			continue
		}
		if b, typed := typedKinds[kind.GroupKind]; typed {
			s.typedCandidates(q, b)
		} else {
			s.dynamicCandidates(q, kind)
		}
	}
	for _, a := range s.catalog.AmbiguousDependentKinds() {
		if a.Namespaced && scope.Searches(true) || a.ClusterScoped && scope.Searches(false) {
			q.gap(a.GroupKind, "", readreason.ReadError)
		}
	}
	slices.SortFunc(q.found, func(a, b candidate) int { return compareRefs(a.ref, b.ref) })
	r := ownership.DependentCandidatesRead{Gaps: q.gaps}
	for _, c := range q.found {
		if read := observe(c.ref, c.obj); read.State == ownership.Present {
			r.Candidates = append(r.Candidates, read.Object)
		} else {
			r.Gaps = append(r.Gaps, read.Gap)
		}
	}
	// Reuse representation only after live gaps were read and compared; never
	// let later searches change a gap slice an earlier answer retained.
	if len(r.Gaps) == 0 {
		r.Gaps = stable[:len(stable):len(stable)]
	} else {
		slices.SortFunc(r.Gaps, compareGaps)
		r.Gaps = slices.Compact(r.Gaps)
		if s.lastGapMode != gapMode || !slices.Equal(s.lastExtraGaps, r.Gaps) {
			s.lastGapMode = gapMode
			s.lastExtraGaps = r.Gaps
			s.lastCombinedGaps = append(slices.Clone(stable), r.Gaps...)
			slices.SortFunc(s.lastCombinedGaps, compareGaps)
			s.lastCombinedGaps = slices.Compact(s.lastCombinedGaps)
		}
		r.Gaps = s.lastCombinedGaps[:len(s.lastCombinedGaps):len(s.lastCombinedGaps)]
	}
	return r
}

// Unwatched kinds have no UID-specific work. Prepare only searched scopes so
// a typed-only search does not prematurely capture dynamic watches.
func (s *session) prepareDependentKinds(scope ownership.SearchScope) {
	namespaced := scope.Searches(true) && !s.preparedNamespaced
	cluster := scope.Searches(false) && !s.preparedCluster
	if !namespaced && !cluster {
		return
	}
	s.preparedNamespaced = s.preparedNamespaced || namespaced
	s.preparedCluster = s.preparedCluster || cluster
	for _, kind := range s.catalog.DependentKinds() {
		if kind.Namespaced && !namespaced || !kind.Namespaced && !cluster {
			continue
		}
		var reason readreason.Code
		if _, typed := typedKinds[kind.GroupKind]; !typed {
			if s.caches.Dynamic == nil {
				reason = readreason.RadarKindNotCollected
			} else {
				reason, _ = s.unwatchedReason(kind, s.servingVersions(kind))
			}
		}
		if reason == "" {
			s.dependentKinds = append(s.dependentKinds, kind)
			continue
		}
		gap := ownership.Gap{Kind: kind.GroupKind, Reason: reason}
		if kind.Namespaced {
			s.unwatchedNamespaced = append(s.unwatchedNamespaced, gap)
		} else {
			s.unwatchedCluster = append(s.unwatchedCluster, gap)
		}
	}
	if namespaced {
		slices.SortFunc(s.unwatchedNamespaced, compareGaps)
	}
	if cluster {
		slices.SortFunc(s.unwatchedCluster, compareGaps)
	}
	s.unwatchedAll = append(slices.Clone(s.unwatchedNamespaced), s.unwatchedCluster...)
	slices.SortFunc(s.unwatchedAll, compareGaps)
	s.lastExtraGaps = nil
	s.lastCombinedGaps = nil
}

// gap records that kind in namespace went unread; "" stands for the kind
// wherever the search looked, or the cluster for a cluster-scoped kind.
func (q *search) gap(gk resourceid.GroupKind, namespace string, reason readreason.Code) {
	q.gaps = append(q.gaps, ownership.Gap{Kind: gk, Namespace: namespace, Reason: reason})
}

// gapEverywhere records that a kind went unread wherever the search looked,
// for a reason of the whole kind, such as Radar not collecting it: one gap,
// however many namespaces the search covers.
func (q *search) gapEverywhere(gk resourceid.GroupKind, namespaced bool, reason readreason.Code) {
	if namespaced && !q.scope.AllNamespaces && len(q.namespaces) == 0 {
		// The scope named no namespace to look in.
		return
	}
	q.gap(gk, "", reason)
}

func (q *search) add(gk resourceid.GroupKind, obj apiruntime.Object, namespace, name string) {
	q.found = append(q.found, candidate{
		ref: resourceid.Ref{Group: gk.Group, Kind: gk.Kind, Namespace: namespace, Name: name},
		obj: obj,
	})
}

func (s *session) typedCandidates(q *search, b resourceid.Builtin) {
	gk := b.GroupKind()
	var target []string
	if b.Namespaced && !q.scope.AllNamespaces {
		target = append([]string{}, q.namespaces...)
	}
	a := s.caches.Typed.AssessKindRead(b.Resource, target)
	if a.Readiness == k8score.KindUnavailable {
		q.gapEverywhere(gk, b.Namespaced, readreason.RadarKindNotCollected)
		return
	}
	read := a.Read
	if a.AllNamespaces {
		// One informer holds the whole kind, so its state is the kind's in
		// every namespace the search names: one gap says so.
		switch a.Readiness {
		case k8score.KindReady:
		case k8score.KindFailed:
			q.gapEverywhere(gk, b.Namespaced, readreason.KindSyncTimedOut)
			return
		default:
			q.gapEverywhere(gk, b.Namespaced, readreason.KindSyncPending)
			return
		}
	} else {
		for _, ns := range a.Loading {
			q.gap(gk, ns, readreason.KindSyncPending)
		}
		for _, ns := range a.Failed {
			q.gap(gk, ns, readreason.KindSyncTimedOut)
		}
		for _, ns := range a.Uncovered {
			q.gap(gk, ns, readreason.RadarNamespaceNotCollected)
		}
		if target == nil {
			// The collector holds only some namespaces; the rest are unread.
			q.gap(gk, "", readreason.RadarNamespaceNotCollected)
		}
		if len(read) == 0 {
			return
		}
	}
	objs, ok := s.caches.Typed.ReadableDependents(b.Resource, read, q.uid)
	if !ok {
		q.gapEverywhere(gk, b.Namespaced, readreason.ReadError)
		return
	}
	for _, obj := range objs {
		if m, err := apimeta.Accessor(obj); err == nil {
			q.add(gk, obj, m.GetNamespace(), m.GetName())
		}
	}
}

func (s *session) dynamicCandidates(q *search, kind ownership.CatalogKind) {
	s.dynamicIndexed(q, kind, k8score.OwnerUIDIndex, []string{q.uid}, func(u *unstructured.Unstructured) {
		q.add(kind.GroupKind, u, u.GetNamespace(), u.GetName())
	})
}

// dynamicIndexed reads the objects index files under any of keys for a
// dynamic kind across q's scope, each store once for all the keys, handing
// each object found to found and recording a gap for each part of the scope
// it can't read. A resource watched at several versions answers from one:
// the most stable version whose cluster-wide store has loaded, or else,
// namespace by namespace, the most stable version that answers there. Every
// version receives the same objects, so the one read can be behind another,
// for as long as its watch lags or fails, but never contradicts it.
func (s *session) dynamicIndexed(q *search, kind ownership.CatalogKind, index string, keys []string, found func(*unstructured.Unstructured)) {
	gk := kind.GroupKind
	if s.caches.Dynamic == nil {
		q.gapEverywhere(gk, kind.Namespaced, readreason.RadarKindNotCollected)
		return
	}
	versions := s.servingVersions(kind)
	if reason, unwatched := s.unwatchedReason(kind, versions); unwatched {
		q.gapEverywhere(gk, kind.Namespaced, reason)
		return
	}
	// A cluster-wide store serves every namespace, so it is read once and its
	// matches kept by namespace: reading it per namespace would gather all of
	// them for each.
	var inScope map[string]bool
	if kind.Namespaced && !q.scope.AllNamespaces {
		inScope = make(map[string]bool, len(q.namespaces))
		for _, ns := range q.namespaces {
			inScope[ns] = true
		}
	}
	for _, gvr := range versions {
		objs, ok := s.caches.Dynamic.ReadableIndexed(gvr, "", index, keys...)
		if !ok {
			continue
		}
		for _, u := range objs {
			if inScope == nil || inScope[u.GetNamespace()] {
				found(u)
			}
		}
		return
	}
	if !kind.Namespaced {
		q.gap(gk, "", unansweredReason(s.caches.Dynamic, versions, ""))
		return
	}
	var watchedByNamespace []string
	for _, gvr := range versions {
		if watched, byNamespace := s.namespaceWatches(gvr); byNamespace {
			watchedByNamespace = append(watchedByNamespace, watched...)
		}
	}
	if len(watchedByNamespace) == 0 {
		// Each version's one informer serves every namespace and none can
		// answer, so the kind's state is the same in each: one gap.
		q.gapEverywhere(gk, true, unansweredReason(s.caches.Dynamic, versions, ""))
		return
	}
	namespaces := q.namespaces
	if q.scope.AllNamespaces {
		slices.Sort(watchedByNamespace)
		namespaces = slices.Compact(watchedByNamespace)
	}
	// Reuse scope bookkeeping across searches. The view refreshes when
	// informers change and still reads readiness live.
	obs := s.observationViews[gk]
	if obs == nil {
		obs = s.caches.Dynamic.NewObservationView(versions)
		if s.observationViews == nil {
			s.observationViews = map[resourceid.GroupKind]*k8score.DynamicObservationView{}
		}
		s.observationViews[gk] = obs
	}
	for _, ns := range namespaces {
		s.readNamespace(q, gk, versions, ns, index, keys, obs, found)
	}
	if q.scope.AllNamespaces {
		// The namespaces not watched one by one are unread: still loading if
		// a version is watched cluster-wide, otherwise uncollected.
		rest := readreason.RadarNamespaceNotCollected
		for _, gvr := range versions {
			if _, byNamespace := s.namespaceWatches(gvr); !byNamespace {
				rest = dynamicReason(s.caches.Dynamic.NamespaceObservation(gvr, ""))
			}
		}
		q.gap(gk, "", rest)
	}
}

// readNamespace reads namespace from the most stable version that answers
// there, or records why none did.
func (s *session) readNamespace(q *search, gk resourceid.GroupKind, versions []schema.GroupVersionResource, namespace, index string, keys []string, obs observer, found func(*unstructured.Unstructured)) {
	for _, gvr := range versions {
		if objs, ok := s.caches.Dynamic.ReadableIndexed(gvr, namespace, index, keys...); ok {
			for _, u := range objs {
				found(u)
			}
			return
		}
	}
	q.gap(gk, namespace, unansweredReason(obs, versions, namespace))
}

// unwatchedReason reports a kind none of whose served versions is watched,
// and why, as of the session's first look: most kinds a cluster serves aren't
// watched, and every search would otherwise ask again.
func (s *session) unwatchedReason(kind ownership.CatalogKind, versions []schema.GroupVersionResource) (readreason.Code, bool) {
	gr := schema.GroupResource{Group: kind.Group, Resource: kind.Resource}
	if reason, seen := s.unwatched[gr]; seen {
		return reason, reason != ""
	}
	var reason readreason.Code
	if !slices.ContainsFunc(s.watched[gr], func(gvr schema.GroupVersionResource) bool { return gvr == versions[0] }) {
		reason = dynamicReason(s.caches.Dynamic.NamespaceObservation(versions[0], ""))
	}
	if s.unwatched == nil {
		s.unwatched = map[schema.GroupResource]readreason.Code{}
	}
	s.unwatched[gr] = reason
	return reason, reason != ""
}

// namespaceWatches returns the namespaces gvr is watched in when it is
// watched namespace by namespace, as of the session's first look, so a watch
// started mid-session can't widen one answer and not another.
func (s *session) namespaceWatches(gvr schema.GroupVersionResource) ([]string, bool) {
	if w, seen := s.byNamespace[gvr]; seen {
		return w, w != nil
	}
	o := s.caches.Dynamic.NamespaceObservation(gvr, "")
	var namespaces []string
	if o.Scope == k8score.DynamicObservationScopeExplicitNamespaces {
		namespaces = append([]string{}, o.Namespaces...)
	}
	if s.byNamespace == nil {
		s.byNamespace = map[schema.GroupVersionResource][]string{}
	}
	s.byNamespace[gvr] = namespaces
	return namespaces, namespaces != nil
}

func compareRefs(a, b resourceid.Ref) int {
	return cmp.Or(cmp.Compare(a.Group, b.Group), cmp.Compare(a.Kind, b.Kind), cmp.Compare(a.Namespace, b.Namespace), cmp.Compare(a.Name, b.Name))
}

func compareGaps(a, b ownership.Gap) int {
	return cmp.Or(cmp.Compare(a.Kind.Group, b.Kind.Group), cmp.Compare(a.Kind.Kind, b.Kind.Kind), cmp.Compare(a.Namespace, b.Namespace), cmp.Compare(a.Reason, b.Reason))
}