pkg/ownership/cachesource/source.go
Rama main · réplica de lectura del repositorio original.
// Package cachesource serves ownership reads from Radar's informer caches.
package cachesource
import (
"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"
"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"
)
// Caches are the caches one cluster connection reads through. Typed is
// required; without Dynamic or Discovery the kinds they serve are unreadable.
type Caches struct {
Typed *k8score.ResourceCache
Dynamic *k8score.DynamicResourceCache
Discovery *k8score.ResourceDiscovery
}
// Source reads from the caches current returns. It calls current when a
// session begins and again when it validates. A host replaces all of its
// caches on a switch, the typed cache with them, so a new typed cache marks a
// new connection.
type Source struct {
current func() Caches
}
// New returns a source reading the caches current returns.
func New(current func() Caches) *Source {
return &Source{current: current}
}
// Begin opens a session over the current caches. Open one per request, not
// one per item: a session's first read of a custom kind lists every dynamic
// watch, which costs milliseconds with thousands of watched namespaces.
func (s *Source) Begin() (ownership.ReadSession, error) {
caches := s.current()
if caches.Typed == nil {
return nil, ownership.ErrNotConnected
}
return &session{caches: caches, current: s.current, catalog: catalogOf(caches.Discovery)}, nil
}
func catalogOf(discovery *k8score.ResourceDiscovery) *ownership.Catalog {
snapshot := discovery.Catalog()
served := make([]ownership.ServedResource, 0, len(snapshot.Resources))
for _, r := range snapshot.Resources {
served = append(served, ownership.ServedResource{
Group: r.Group, Version: r.Version, Kind: r.Kind, Resource: r.Name, Namespaced: r.Namespaced, Verbs: r.Verbs,
})
}
return ownership.NewCatalog(served, snapshot.Complete)
}
// typedKinds are the kinds the typed cache holds, keyed to the built-in that
// names their informer and the version it reads them at.
var typedKinds = func() map[resourceid.GroupKind]resourceid.Builtin {
out := map[resourceid.GroupKind]resourceid.Builtin{}
for _, kl := range k8score.AllKindListers() {
if b, ok := resourceid.BuiltinForKind(kl.Kind()); ok && b.Group == kl.Group() {
out[b.GroupKind()] = b
}
}
return out
}()
type session struct {
caches Caches
current func() Caches
catalog *ownership.Catalog
// watched holds the versions each dynamically watched resource is watched
// at, as of the session's first dynamic read.
watched map[schema.GroupResource][]schema.GroupVersionResource
serving map[resourceid.GroupKind][]schema.GroupVersionResource
// byNamespace holds the namespaces each resource version is watched in
// when it is watched namespace by namespace (nil otherwise).
byNamespace map[schema.GroupVersionResource][]string
// unwatched holds why each resource with no watched version is unread
// ("" for a watched one).
unwatched map[schema.GroupResource]readreason.Code
dependentKinds []ownership.CatalogKind
preparedNamespaced bool
preparedCluster bool
unwatchedNamespaced []ownership.Gap
unwatchedCluster []ownership.Gap
unwatchedAll []ownership.Gap
lastGapMode uint8
lastExtraGaps []ownership.Gap
lastCombinedGaps []ownership.Gap
membershipStates map[resourceid.Ref]membershipStateObservation
// recordingManagersRead, set only by tests, runs after RecordingManagers reads a manager
// kind, with the keys it read together.
recordingManagersRead func(kind resourceid.GroupKind, keys []string)
observationViews map[resourceid.GroupKind]*k8score.DynamicObservationView
}
// TypedCache returns the typed cache s reads from, when this package began s.
// What a caller reads through it comes from the same cluster connection as s's
// own reads, and s's ValidateCaches refuses it with them once the caches are
// replaced.
func TypedCache(s ownership.ReadSession) (*k8score.ResourceCache, bool) {
cs, ok := s.(*session)
if !ok || cs.caches.Typed == nil {
return nil, false
}
return cs.caches.Typed, true
}
func (s *session) Catalog() *ownership.Catalog {
return s.catalog
}
func (s *session) Object(ref resourceid.Ref) ownership.ObjectRead {
return s.readObject(ref, nil)
}
// readObject is Object, lending the cached object it answered from, when
// present, to borrow, which must not modify or keep it.
func (s *session) readObject(ref resourceid.Ref, borrow func(*unstructured.Unstructured)) ownership.ObjectRead {
if ref.Kind == "" || ref.Name == "" {
return unavailable(ref, readreason.ReadError)
}
if ownership.IsEvent(ref.GroupKind()) {
return ownership.ObjectRead{State: ownership.Excluded}
}
if b, ok := typedKinds[ref.GroupKind()]; ok {
return s.typedObject(ref, b)
}
return s.dynamicObject(ref, borrow)
}
func (s *session) typedObject(ref resourceid.Ref, b resourceid.Builtin) ownership.ObjectRead {
if reason, wrong := scopeMismatch(ref, b.Namespaced); wrong {
return unavailable(ref, reason)
}
obj, found, ok := s.caches.Typed.ReadableObject(b.Resource, ref.Namespace, ref.Name)
switch {
case !ok:
return unavailable(ref, typedReason(s.caches.Typed, b.Resource, ref.Namespace))
case !found:
return ownership.ObjectRead{State: ownership.Absent}
}
return observe(ref, obj)
}
func (s *session) dynamicObject(ref resourceid.Ref, borrow func(*unstructured.Unstructured)) ownership.ObjectRead {
if s.catalog.Ambiguous(ref.GroupKind()) {
return unavailable(ref, readreason.ReadError)
}
kind, known := s.catalog.Kind(ref.GroupKind())
if !known {
if !s.catalog.Complete() {
// The group discovery missed may serve it.
return unavailable(ref, readreason.ReadError)
}
return unavailable(ref, readreason.KindUnknown)
}
if reason, wrong := scopeMismatch(ref, kind.Namespaced); wrong {
return unavailable(ref, reason)
}
if s.caches.Dynamic == nil {
return unavailable(ref, readreason.RadarKindNotCollected)
}
// A resource watched at two versions answers from the most stable one
// that has loaded: every version receives the same objects with the same
// owner references, so the one read can be behind another, for as long as
// its watch lags or fails, but never contradicts it.
versions := s.servingVersions(kind)
for _, gvr := range versions {
obj, found, ok := s.caches.Dynamic.ReadableObject(gvr, ref.Namespace, ref.Name)
switch {
case !ok:
continue
case !found:
return ownership.ObjectRead{State: ownership.Absent}
}
read := observe(ref, obj)
if borrow != nil && read.State == ownership.Present {
borrow(obj)
}
return read
}
return unavailable(ref, unansweredReason(s.caches.Dynamic, versions, ref.Namespace))
}
// observer reports how Radar collects a resource version in a namespace: the
// dynamic cache, or a view of it that builds each version's scope once.
type observer interface {
NamespaceObservation(gvr schema.GroupVersionResource, namespace string) k8score.DynamicResourceObservation
}
// unansweredReason says why none of versions answered in namespace: the first
// version watched there, which answers once it loads, or else the first.
func unansweredReason(obs observer, versions []schema.GroupVersionResource, namespace string) readreason.Code {
var first readreason.Code
for i, gvr := range versions {
o := obs.NamespaceObservation(gvr, namespace)
if o.State != k8score.DynamicObservationUnwatched {
return dynamicReason(o)
}
if i == 0 {
first = dynamicReason(o)
}
}
return first
}
// servingVersions returns the versions the session reads kind's resource at,
// most stable first: the watched versions the catalog serves. With none
// watched it is the most stable served version, whose observation says why.
// The result is shared within the session and must not be modified.
func (s *session) servingVersions(kind ownership.CatalogKind) []schema.GroupVersionResource {
if versions, ok := s.serving[kind.GroupKind]; ok {
return versions
}
gr := schema.GroupResource{Group: kind.Group, Resource: kind.Resource}
if s.watched == nil {
s.watched = map[schema.GroupResource][]schema.GroupVersionResource{}
for _, gvr := range s.caches.Dynamic.WatchedGVRs() {
s.watched[gvr.GroupResource()] = append(s.watched[gvr.GroupResource()], gvr)
}
}
var versions []schema.GroupVersionResource
for _, gvr := range s.watched[gr] {
if slices.Contains(kind.Versions, gvr.Version) {
versions = append(versions, gvr)
}
}
slices.SortFunc(versions, func(a, b schema.GroupVersionResource) int { return stabilityOrder(a.Version, b.Version) })
if len(versions) == 0 {
versions = []schema.GroupVersionResource{{Group: kind.Group, Version: mostStable(kind.Versions), Resource: kind.Resource}}
}
if s.serving == nil {
s.serving = map[resourceid.GroupKind][]schema.GroupVersionResource{}
}
s.serving[kind.GroupKind] = versions
return versions
}
func (s *session) ValidateCaches() error {
if s.caches.Typed != s.current().Typed {
return ownership.ErrCachesReplaced
}
return nil
}
// scopeMismatch reports a ref whose namespace can't hold its kind. Neither is
// absent: a namespaced kind read without a namespace has no single store to
// answer, and a cluster-scoped kind read in a namespace means the caller
// resolved scope wrongly, which an absence would hide as a dangling owner.
func scopeMismatch(ref resourceid.Ref, namespaced bool) (readreason.Code, bool) {
switch {
case namespaced && ref.Namespace == "":
return readreason.NamespaceRequired, true
case !namespaced && ref.Namespace != "":
return readreason.ReadError, true
}
return "", false
}
func typedReason(rc *k8score.ResourceCache, resource, namespace string) readreason.Code {
a := rc.AssessKindRead(resource, []string{namespace})
switch {
case a.Readiness == k8score.KindUnavailable:
// Radar's access probe leaves a kind off when its identity is
// refused, when the probe can't tell, and when the kind isn't
// installed; the cache knows only that it is off, and the host can
// say which.
return readreason.RadarKindNotCollected
case len(a.Uncovered) > 0:
return readreason.RadarNamespaceNotCollected
case len(a.Failed) > 0:
return readreason.KindSyncTimedOut
default:
// Loading, or it finished loading after the read.
return readreason.KindSyncPending
}
}
func dynamicReason(o k8score.DynamicResourceObservation) readreason.Code {
switch o.State {
case k8score.DynamicObservationDenied:
if o.ReasonCode == "credentials_rejected" {
return readreason.RadarCredentialsRejected
}
return readreason.RadarForbidden
case k8score.DynamicObservationUnsupported:
return readreason.KindNotWatchable
case k8score.DynamicObservationDeferred:
switch o.ReasonCode {
case "eager_watch_pending":
return readreason.KindSyncPending
case "scope_probe_incomplete":
// The probe for where Radar may watch never finished.
return readreason.ReadError
}
return readreason.RadarKindNotCollected
case k8score.DynamicObservationSyncing:
if o.ReasonCode == "sync_stalled" {
return readreason.KindSyncTimedOut
}
return readreason.KindSyncPending
case k8score.DynamicObservationSynced:
// It finished loading after the read.
return readreason.KindSyncPending
case k8score.DynamicObservationUnwatched:
if o.ReasonCode == "cache_stopped" {
// The connection was switched or closed after the session began.
return readreason.ConnectionChanged
}
return readreason.RadarKindNotCollected
default:
return readreason.RadarKindNotCollected
}
}
func unavailable(ref resourceid.Ref, reason readreason.Code) ownership.ObjectRead {
return ownership.ObjectRead{Gap: ownership.Gap{Kind: ref.GroupKind(), Namespace: ref.Namespace, Reason: reason}}
}
// observe copies what ownership needs out of a cached object, which must not
// be modified or kept.
func observe(ref resourceid.Ref, obj apiruntime.Object) ownership.ObjectRead {
m, err := apimeta.Accessor(obj)
if err != nil {
return unavailable(ref, readreason.ReadError)
}
return ownership.ObjectRead{State: ownership.Present, Object: ownership.Observe(ref, m)}
}
func mostStable(versions []string) string {
if len(versions) == 0 {
return ""
}
return slices.MinFunc(versions, stabilityOrder)
}
// stabilityOrder sorts more stable versions first.
func stabilityOrder(a, b string) int {
switch {
case k8score.IsMoreStableVersion(a, b):
return -1
case k8score.IsMoreStableVersion(b, a):
return 1
}
return 0
}