kubevela/pkg/resourcekeeper/statekeep.go

/*
Copyright 2021 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 resourcekeeper

import (
	"context"
	"sync"

	"github.com/crossplane/crossplane-runtime/pkg/fieldpath"
	"github.com/kubevela/pkg/util/maps"
	"github.com/kubevela/pkg/util/slices"
	"github.com/pkg/errors"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/types"

	kerrors "k8s.io/apimachinery/pkg/api/errors"

	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1alpha1"
	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
	"github.com/oam-dev/kubevela/pkg/auth"
	"github.com/oam-dev/kubevela/pkg/multicluster"
	"github.com/oam-dev/kubevela/pkg/utils/apply"
	velaerrors "github.com/oam-dev/kubevela/pkg/utils/errors"
)

// StateKeep run this function to keep resources up-to-date
func (h *resourceKeeper) StateKeep(ctx context.Context) error {
	if h.applyOncePolicy != nil && h.applyOncePolicy.Enable && h.applyOncePolicy.Rules == nil {
		return nil
	}
	ctx = auth.ContextWithUserInfo(ctx, h.app)
	mrs := make(map[string]v1beta1.ManagedResource)
	belongs := make(map[string]*v1beta1.ResourceTracker)
	for _, rt := range []*v1beta1.ResourceTracker{h._currentRT, h._rootRT} {
		if rt != nil && rt.GetDeletionTimestamp() == nil {
			for _, mr := range rt.Spec.ManagedResources {
				key := mr.ResourceKey()
				mrs[key] = mr
				belongs[key] = rt
			}
		}
	}
	var stalesMu sync.Mutex
	var staleEntries []staleEntry
	errs := slices.ParMap(maps.Values(mrs), func(mr v1beta1.ManagedResource) error {
		rt := belongs[mr.ResourceKey()]
		entry := h.cache.get(ctx, mr)
		if entry.err != nil {
			return entry.err
		}
		if mr.Deleted {
			if entry.exists && entry.obj != nil && entry.obj.GetDeletionTimestamp() == nil {
				deleteCtx := multicluster.ContextWithClusterName(ctx, mr.Cluster)
				if err := h.Client.Delete(deleteCtx, entry.obj); err != nil {
					return errors.Wrapf(err, "failed to delete outdated resource %s in resourcetracker %s", mr.ResourceKey(), rt.Name)
				}
			}
		} else {
			if mr.Data == nil || mr.Data.Raw == nil {
				// no state-keep
				return nil
			}
			manifest, err := mr.ToUnstructuredWithData()
			if err != nil {
				return errors.Wrapf(err, "failed to decode resource %s from resourcetracker", mr.ResourceKey())
			}
			applyCtx := multicluster.ContextWithClusterName(ctx, mr.Cluster)
			manifest, err = ApplyStrategies(applyCtx, h, manifest, v1alpha1.ApplyOnceStrategyOnAppStateKeep)
			if err != nil {
				return errors.Wrapf(err, "failed to apply once resource %s from resourcetracker %s", mr.ResourceKey(), rt.Name)
			}
			if manifest == nil {
				stalesMu.Lock()
				staleEntries = append(staleEntries, staleEntry{mr: mr, rt: rt})
				stalesMu.Unlock()
				return nil
			}
			ao := []apply.ApplyOption{apply.MustBeControlledByApp(h.app)}
			if h.isShared(manifest) {
				ao = append([]apply.ApplyOption{apply.SharedByApp(h.app)}, ao...)
			}
			if h.isReadOnly(manifest) {
				ao = append([]apply.ApplyOption{apply.ReadOnly()}, ao...)
			}
			if h.canTakeOver(manifest) {
				ao = append([]apply.ApplyOption{apply.TakeOver()}, ao...)
			}
			if strategy := h.getUpdateStrategy(manifest); strategy != nil {
				ao = append([]apply.ApplyOption{apply.WithUpdateStrategy(*strategy)}, ao...)
			}
			if err = h.applicator.Apply(applyCtx, manifest, ao...); err != nil {
				return errors.Wrapf(err, "failed to re-apply resource %s from resourcetracker %s", mr.ResourceKey(), rt.Name)
			}
		}
		return nil
	}, slices.Parallelism(MaxDispatchConcurrent))
	if err := h.cleanupStaleEntries(ctx, staleEntries); err != nil {
		errs = append(errs, err)
	}
	return velaerrors.AggregateErrors(errs)
}

// staleEntry records a managed resource whose backing object no longer exists
// and should be removed from its ResourceTracker.
type staleEntry struct {
	mr v1beta1.ManagedResource
	rt *v1beta1.ResourceTracker
}

// cleanupStaleEntries removes managed resources from their ResourceTrackers
// when the underlying resource no longer exists (e.g. externally deleted
// apply-once resources). This avoids unnecessary API server GETs on future
// StateKeep cycles.
func (h *resourceKeeper) cleanupStaleEntries(ctx context.Context, entries []staleEntry) error {
	if len(entries) == 0 {
		return nil
	}
	// Group by ResourceTracker to batch updates
	rtUpdates := make(map[string]*v1beta1.ResourceTracker)
	rtRemovals := make(map[string][]v1beta1.ManagedResource)
	for _, e := range entries {
		key := string(e.rt.UID)
		rtUpdates[key] = e.rt
		rtRemovals[key] = append(rtRemovals[key], e.mr)
	}
	var errs []error
	for key, rt := range rtUpdates {
		for _, mr := range rtRemovals[key] {
			obj := mr.ToUnstructured()
			rt.DeleteManagedResource(obj, true)
		}
		if err := h.Client.Update(multicluster.ContextInLocalCluster(ctx), rt); err != nil {
			errs = append(errs, errors.Wrapf(err, "failed to remove stale entries from resourcetracker %s", rt.Name))
		}
	}
	return velaerrors.AggregateErrors(errs)
}

// ApplyStrategies will generate manifest with applyOnceStrategy
func ApplyStrategies(ctx context.Context, h *resourceKeeper, manifest *unstructured.Unstructured, matchedAffectStage v1alpha1.ApplyOnceAffectStrategy) (*unstructured.Unstructured, error) {
	if h.applyOncePolicy == nil {
		return manifest, nil
	}
	strategy := h.applyOncePolicy.FindStrategy(manifest)
	if strategy != nil {
		affectStage := strategy.ApplyOnceAffectStrategy
		if shouldMerge(affectStage, matchedAffectStage) {
			un := new(unstructured.Unstructured)
			un.SetAPIVersion(manifest.GetAPIVersion())
			un.SetKind(manifest.GetKind())
			err := h.Get(ctx, types.NamespacedName{Name: manifest.GetName(), Namespace: manifest.GetNamespace()}, un)
			if err != nil {
				if kerrors.IsNotFound(err) {
					if matchedAffectStage == v1alpha1.ApplyOnceStrategyOnAppStateKeep {
						return nil, nil
					}
					return manifest, nil
				}
				return nil, err
			}
			return mergeValue(strategy.Path, manifest, un)
		}

	}
	return manifest, nil
}

func shouldMerge(affectStage v1alpha1.ApplyOnceAffectStrategy, matchedAffectType v1alpha1.ApplyOnceAffectStrategy) bool {
	return affectStage == "" || affectStage == v1alpha1.ApplyOnceStrategyAlways || affectStage == matchedAffectType
}

func mergeValue(paths []string, manifest *unstructured.Unstructured, un *unstructured.Unstructured) (*unstructured.Unstructured, error) {
	for _, path := range paths {
		if path == "*" {
			manifest = un.DeepCopy()
			break
		}
		value, err := fieldpath.Pave(un.UnstructuredContent()).GetValue(path)
		if err != nil {
			return nil, err
		}
		err = fieldpath.Pave(manifest.UnstructuredContent()).SetValue(path, value)
		if err != nil {
			return nil, err
		}
	}
	return manifest, nil
}