kubevela/pkg/cache/informer.go

/*
Copyright 2023 The KubeVela Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package cache

import (
	"context"
	"sync"
	"time"

	"github.com/kubevela/pkg/util/k8s"
	"github.com/kubevela/pkg/util/runtime"
	"github.com/kubevela/pkg/util/singleton"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/util/sets"
	utilfeature "k8s.io/apiserver/pkg/util/feature"
	kcache "k8s.io/client-go/tools/cache"
	"k8s.io/klog/v2"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
	"github.com/oam-dev/kubevela/pkg/features"
	"github.com/oam-dev/kubevela/pkg/utils/apply"
)

var (
	// ApplicationRevisionDefinitionCachePruneDuration the prune duration for application revision definition cache
	ApplicationRevisionDefinitionCachePruneDuration = time.Hour
)

// ObjectCacheEntry entry for object cache
type ObjectCacheEntry[T any] struct {
	ptr          *T
	refs         sets.Set[string]
	lastAccessed time.Time
}

// ObjectCache cache for objects
type ObjectCache[T any] struct {
	mu      sync.RWMutex
	objects map[string]*ObjectCacheEntry[T]
}

// NewObjectCache create an object cache
func NewObjectCache[T any]() *ObjectCache[T] {
	return &ObjectCache[T]{
		objects: map[string]*ObjectCacheEntry[T]{},
	}
}

// Get retrieve the cache entry
func (in *ObjectCache[T]) Get(hash string) *T {
	in.mu.RLock()
	defer in.mu.RUnlock()
	if entry, found := in.objects[hash]; found {
		return entry.ptr
	}
	return nil
}

// Add insert cache entry with ref, return the ptr of the entry
func (in *ObjectCache[T]) Add(hash string, obj *T, ref string) *T {
	in.mu.Lock()
	defer in.mu.Unlock()
	if entry, found := in.objects[hash]; found {
		entry.refs.Insert(ref)
		entry.lastAccessed = time.Now()
		return entry.ptr
	}
	in.objects[hash] = &ObjectCacheEntry[T]{
		ptr:          obj,
		refs:         sets.New[string](ref),
		lastAccessed: time.Now(),
	}
	return obj
}

// DeleteRef delete ref for an obj
func (in *ObjectCache[T]) DeleteRef(hash string, ref string) {
	in.mu.Lock()
	defer in.mu.Unlock()
	if entry, found := in.objects[hash]; found {
		entry.refs.Delete(ref)
		if entry.refs.Len() == 0 {
			delete(in.objects, hash)
		}
	}
}

// Remap relocate the object ptr with given ref
func (in *ObjectCache[T]) Remap(m map[string]*T, ref string) {
	for key, o := range m {
		if hash, err := apply.ComputeSpecHash(o); err == nil {
			m[key] = in.Add(hash, o, ref)
		}
	}
}

// Unmap drop all the hash object from the map
func (in *ObjectCache[T]) Unmap(m map[string]*T, ref string) {
	for _, o := range m {
		if hash, err := apply.ComputeSpecHash(o); err == nil {
			in.DeleteRef(hash, ref)
		}
	}
}

// Size get the size of the cache
func (in *ObjectCache[T]) Size() int {
	in.mu.RLock()
	defer in.mu.RUnlock()
	return len(in.objects)
}

// Prune remove outdated cache, return the pruned count
func (in *ObjectCache[T]) Prune(outdated time.Duration) int {
	in.mu.Lock()
	defer in.mu.Unlock()
	cnt := 0
	for key, entry := range in.objects {
		if time.Now().After(entry.lastAccessed.Add(outdated)) {
			delete(in.objects, key)
			cnt++
		}
	}
	return cnt
}

// DefinitionCache cache for definitions
type DefinitionCache struct {
	ComponentDefinitionCache    *ObjectCache[v1beta1.ComponentDefinition]
	TraitDefinitionCache        *ObjectCache[v1beta1.TraitDefinition]
	WorkflowStepDefinitionCache *ObjectCache[v1beta1.WorkflowStepDefinition]
}

// NewDefinitionCache create DefinitionCache
func NewDefinitionCache() *DefinitionCache {
	return &DefinitionCache{
		ComponentDefinitionCache:    NewObjectCache[v1beta1.ComponentDefinition](),
		TraitDefinitionCache:        NewObjectCache[v1beta1.TraitDefinition](),
		WorkflowStepDefinitionCache: NewObjectCache[v1beta1.WorkflowStepDefinition](),
	}
}

// RemapRevision remap all definitions in the given revision
func (in *DefinitionCache) RemapRevision(rev *v1beta1.ApplicationRevision) {
	ref := client.ObjectKeyFromObject(rev).String()
	in.ComponentDefinitionCache.Remap(rev.Spec.ComponentDefinitions, ref)
	in.TraitDefinitionCache.Remap(rev.Spec.TraitDefinitions, ref)
	in.WorkflowStepDefinitionCache.Remap(rev.Spec.WorkflowStepDefinitions, ref)
}

// UnmapRevision unmap definitions from the provided revision by the given ref
func (in *DefinitionCache) UnmapRevision(rev *v1beta1.ApplicationRevision) {
	ref := client.ObjectKeyFromObject(rev).String()
	in.ComponentDefinitionCache.Unmap(rev.Spec.ComponentDefinitions, ref)
	in.TraitDefinitionCache.Unmap(rev.Spec.TraitDefinitions, ref)
	in.WorkflowStepDefinitionCache.Unmap(rev.Spec.WorkflowStepDefinitions, ref)
}

// Start clear cache every duration
func (in *DefinitionCache) Start(ctx context.Context, store cache.Cache, duration time.Duration) {
	informer := runtime.Must(store.GetInformer(ctx, &v1beta1.ApplicationRevision{}))
	_, err := informer.AddEventHandler(kcache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			if rev, ok := obj.(*v1beta1.ApplicationRevision); ok {
				in.RemapRevision(rev)
			}
		},
		UpdateFunc: func(oldObj, newObj interface{}) {
			if rev, ok := oldObj.(*v1beta1.ApplicationRevision); ok {
				in.UnmapRevision(rev)
			}
			if rev, ok := newObj.(*v1beta1.ApplicationRevision); ok {
				in.RemapRevision(rev)
			}
		},
		DeleteFunc: func(obj interface{}) {
			if rev, ok := obj.(*v1beta1.ApplicationRevision); ok {
				in.UnmapRevision(rev)
			}
		},
	})
	if err != nil {
		klog.ErrorS(err, "failed to add event handler for definition cache")
	}
	for {
		select {
		case <-ctx.Done():
			return
		default:
			t0 := time.Now()
			compDefPruned := in.ComponentDefinitionCache.Prune(duration)
			traitDefPruned := in.TraitDefinitionCache.Prune(duration)
			wsDefPruned := in.WorkflowStepDefinitionCache.Prune(duration)
			klog.Infof("DefinitionCache prune finished. ComponentDefinition: %d(-%d), TraitDefinition: %d(-%d), WorkflowStepDefinition: %d(-%d). Time cost: %d ms.",
				in.ComponentDefinitionCache.Size(), compDefPruned,
				in.TraitDefinitionCache.Size(), traitDefPruned,
				in.WorkflowStepDefinitionCache.Size(), wsDefPruned,
				time.Since(t0).Microseconds())
			time.Sleep(duration)
		}
	}
}

// DefaultDefinitionCache the default definition cache
var DefaultDefinitionCache = singleton.NewSingleton(NewDefinitionCache)

func filterUnnecessaryField(o client.Object) {
	_ = k8s.DeleteAnnotation(o, "kubectl.kubernetes.io/last-applied-configuration")
	o.SetManagedFields(nil)
}

func wrapTransformFunc[T client.Object](fn func(T)) kcache.TransformFunc {
	return func(i interface{}) (interface{}, error) {
		if o, ok := i.(T); ok {
			filterUnnecessaryField(o)
			fn(o)
			return o, nil
		}
		return i, nil
	}
}

// AddInformerTransformFuncToCacheOption add informer transform func to cache option
// This will filter out the unnecessary fields for cached objects and use definition cache
// to reduce the duplicated storage of same definitions
func AddInformerTransformFuncToCacheOption(opts *cache.Options) {
	if utilfeature.DefaultMutableFeatureGate.Enabled(features.InformerCacheFilterUnnecessaryFields) {
		if opts.ByObject == nil {
			opts.ByObject = map[client.Object]cache.ByObject{}
		}
		opts.ByObject[&v1beta1.ApplicationRevision{}] = cache.ByObject{
			Transform: wrapTransformFunc(func(rev *v1beta1.ApplicationRevision) {
				if utilfeature.DefaultMutableFeatureGate.Enabled(features.SharedDefinitionStorageForApplicationRevision) {
					DefaultDefinitionCache.Get().RemapRevision(rev)
				}
			}),
		}
		opts.ByObject[&v1beta1.Application{}] = cache.ByObject{
			Transform: wrapTransformFunc(func(app *v1beta1.Application) {}),
		}
		opts.ByObject[&v1beta1.ResourceTracker{}] = cache.ByObject{
			Transform: wrapTransformFunc(func(rt *v1beta1.ResourceTracker) {}),
		}
	}
}

// NewResourcesToDisableCache get ClientDisableCacheFor for building controller
func NewResourcesToDisableCache() []client.Object {
	var objs []client.Object
	if utilfeature.DefaultMutableFeatureGate.Enabled(features.DisableWorkflowContextConfigMapCache) {
		objs = append(objs, &corev1.ConfigMap{})
	}
	return objs
}