reconciler.go

/*


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 core

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/go-logr/logr"
	mutationsv1 "github.com/open-policy-agent/gatekeeper/v3/apis/mutations/v1"
	statusv1beta1 "github.com/open-policy-agent/gatekeeper/v3/apis/status/v1beta1"
	ctrlmutators "github.com/open-policy-agent/gatekeeper/v3/pkg/controller/mutators"
	"github.com/open-policy-agent/gatekeeper/v3/pkg/logging"
	"github.com/open-policy-agent/gatekeeper/v3/pkg/mutation"
	mutationschema "github.com/open-policy-agent/gatekeeper/v3/pkg/mutation/schema"
	"github.com/open-policy-agent/gatekeeper/v3/pkg/mutation/types"
	"github.com/open-policy-agent/gatekeeper/v3/pkg/readiness"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	apiTypes "k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/event"
	logf "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
)

// newReconciler returns a new reconcile.Reconciler.
func newReconciler(
	mgr manager.Manager,
	mutationSystem *mutation.System,
	tracker *readiness.Tracker,
	getPod func(context.Context) (*corev1.Pod, error),
	kind string,
	newMutationObj func() client.Object,
	mutatorFor func(client.Object) (types.Mutator, error),
	events chan event.GenericEvent,
) *Reconciler {
	r := &Reconciler{
		system:         mutationSystem,
		Client:         mgr.GetClient(),
		tracker:        tracker,
		getPod:         getPod,
		scheme:         mgr.GetScheme(),
		reporter:       ctrlmutators.NewStatsReporter(),
		cache:          ctrlmutators.NewMutationCache(),
		gvk:            mutationsv1.GroupVersion.WithKind(kind),
		newMutationObj: newMutationObj,
		mutatorFor:     mutatorFor,
		log:            logf.Log.WithName("controller").WithValues(logging.Process, fmt.Sprintf("%s_controller", strings.ToLower(kind))),
		events:         events,
	}
	if getPod == nil {
		r.getPod = r.defaultGetPod
	}
	return r
}

// Reconciler reconciles mutator objects.
type Reconciler struct {
	client.Client
	gvk            schema.GroupVersionKind
	newMutationObj func() client.Object
	mutatorFor     func(client.Object) (types.Mutator, error)

	system   *mutation.System
	tracker  *readiness.Tracker
	getPod   func(context.Context) (*corev1.Pod, error)
	scheme   *runtime.Scheme
	reporter ctrlmutators.StatsReporter
	cache    *ctrlmutators.Cache
	log      logr.Logger

	events chan event.GenericEvent
}

// +kubebuilder:rbac:groups=mutations.gatekeeper.sh,resources=*,verbs=get;list;watch;create;update;patch;delete

// Reconcile reads that state of the cluster for a mutator object and syncs it with the mutation system.
func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
	r.log.Info("Reconcile", "request", request)
	startTime := time.Now()

	mutationObj, deleted, err := r.getOrDefault(ctx, request.NamespacedName)
	if err != nil {
		return reconcile.Result{}, err
	}

	// default ingestion status to error, only change it if we successfully
	// reconcile without conflicts
	ingestionStatus := ctrlmutators.MutatorStatusError

	// default conflict to false, only set to true if we find a conflict
	conflict := false

	// Encasing this call in a function prevents the arguments from being evaluated early.
	id := types.MakeID(mutationObj)
	defer func() {
		if !deleted {
			r.cache.Upsert(id, ingestionStatus, conflict)
		}
		r.reportMutator(id, ingestionStatus, startTime, deleted)
	}()

	// previousConflicts records the conflicts this Mutator has with other mutators
	// before making any changes.
	previousConflicts := r.system.GetConflicts(id)

	if deleted {
		// Either the mutator was deleted before we were able to process this request, or it has been marked for
		// deletion.
		r.getTracker().CancelExpect(mutationObj)
		err = r.reconcileDeleted(ctx, id)
	} else {
		err = r.reconcileUpsert(ctx, id, mutationObj)
	}

	if err != nil {
		return reconcile.Result{}, err
	}

	newConflicts := r.system.GetConflicts(id)

	// diff is the set of mutators which either:
	// 1) previously conflicted with mutationObj but do not after this change, or
	// 2) now conflict with mutationObj but did not before this change.
	diff := symmetricDifference(previousConflicts, newConflicts)
	delete(diff, id)

	// Now that we've made changes to the recorded Mutator schemas, we can re-check
	// for conflicts.
	r.queueConflicts(diff)

	// Any mutator that's in conflict with another should be in the "error" state.
	if len(newConflicts) == 0 {
		ingestionStatus = ctrlmutators.MutatorStatusActive
	} else {
		conflict = true
	}

	return reconcile.Result{}, nil
}

func (r *Reconciler) reconcileUpsert(ctx context.Context, id types.ID, obj client.Object) error {
	mutator, err := r.mutatorFor(obj)
	if err != nil {
		r.log.Error(err, "Creating mutator for resource failed", "resource",
			client.ObjectKeyFromObject(obj))
		r.getTracker().TryCancelExpect(obj)

		return r.updateStatusWithError(ctx, obj, err)
	}

	if errToUpsert := r.system.Upsert(mutator); errToUpsert != nil {
		r.log.Error(err, "Insert failed", "resource",
			client.ObjectKeyFromObject(obj))
		r.getTracker().TryCancelExpect(obj)

		// Since we got an error upserting obj, update its PodStatus first.
		return r.updateStatusWithError(ctx, obj, errToUpsert)
	}

	r.getTracker().Observe(obj)

	return r.updateStatus(ctx, id,
		setID(obj.GetUID()), setGeneration(obj.GetGeneration()),
		setEnforced(true), setErrors(nil))
}

func (r *Reconciler) getOrCreatePodStatus(ctx context.Context, mutatorID types.ID) (*statusv1beta1.MutatorPodStatus, error) {
	pod, err := r.getPod(ctx)
	if err != nil {
		return nil, err
	}

	statusObj := &statusv1beta1.MutatorPodStatus{}
	sName, err := statusv1beta1.KeyForMutatorID(pod.Name, mutatorID)
	if err != nil {
		return nil, err
	}

	key := apiTypes.NamespacedName{Name: sName, Namespace: pod.Namespace}
	if err := r.Get(ctx, key, statusObj); err != nil {
		if !apierrors.IsNotFound(err) {
			return nil, err
		}
	} else {
		return statusObj, nil
	}

	statusObj, err = statusv1beta1.NewMutatorStatusForPod(pod, mutatorID, r.scheme)
	if err != nil {
		return nil, err
	}
	if err := r.Create(ctx, statusObj); err != nil {
		return nil, err
	}
	return statusObj, nil
}

func (r *Reconciler) defaultGetPod(_ context.Context) (*corev1.Pod, error) {
	// require injection of GetPod in order to control what client we use to
	// guarantee we don't inadvertently create a watch
	panic("GetPod must be injected to Reconciler")
}

func (r *Reconciler) reportMutator(_ types.ID, ingestionStatus ctrlmutators.MutatorIngestionStatus, startTime time.Time, deleted bool) {
	if r.reporter == nil {
		return
	}

	if !deleted {
		if err := r.reporter.ReportMutatorIngestionRequest(ingestionStatus, time.Since(startTime)); err != nil {
			r.log.Error(err, "failed to report mutator ingestion request")
		}
	}

	for status, count := range r.cache.TallyStatus() {
		if err := r.reporter.ReportMutatorsStatus(status, count); err != nil {
			r.log.Error(err, "failed to report mutator status request")
		}
	}

	if err := r.reporter.ReportMutatorsInConflict(r.cache.TallyConflict()); err != nil {
		r.log.Error(err, "failed to report mutators in conflict request")
	}
}

// getOrDefault attempts to get the Mutator from the cluster, or returns a default-instantiated Mutator if one does not
// exist.
func (r *Reconciler) getOrDefault(ctx context.Context, namespacedName apiTypes.NamespacedName) (client.Object, bool, error) {
	obj := r.newMutationObj()
	err := r.Get(ctx, namespacedName, obj)
	switch {
	case err == nil:
		// Treat objects with a DeletionTimestamp as if they are deleted.
		deleted := !obj.GetDeletionTimestamp().IsZero()
		return obj, deleted, nil
	case apierrors.IsNotFound(err):
		obj = r.newMutationObj()
		obj.SetName(namespacedName.Name)
		obj.SetNamespace(namespacedName.Namespace)
		obj.GetObjectKind().SetGroupVersionKind(r.gvk)
		return obj, true, nil
	default:
		return nil, false, err
	}
}

func (r *Reconciler) getTracker() readiness.Expectations {
	return r.tracker.For(r.gvk)
}

// reconcileDeleted removes the Mutator from the controller and deletes the corresponding PodStatus.
func (r *Reconciler) reconcileDeleted(ctx context.Context, id types.ID) error {
	r.cache.Remove(id)

	if err := r.system.Remove(id); err != nil {
		r.log.Error(err, "Remove failed", "resource",
			apiTypes.NamespacedName{Name: id.Name, Namespace: id.Namespace})
		return err
	}

	pod, err := r.getPod(ctx)
	if err != nil {
		return err
	}

	sName, err := statusv1beta1.KeyForMutatorID(pod.Name, id)
	if err != nil {
		return err
	}

	status := &statusv1beta1.MutatorPodStatus{}
	status.SetName(sName)
	status.SetNamespace(pod.Namespace)
	if err = r.Delete(ctx, status); err != nil && !apierrors.IsNotFound(err) {
		return err
	}

	return nil
}

// queueConflicts queues updates for Mutators in ids.
// We send events to the handler's event queue rather than attempting the update
// ourselves to delegate handling failures to the existing controller logic.
func (r *Reconciler) queueConflicts(ids mutationschema.IDSet) {
	if r.events == nil {
		return
	}

	for id := range ids {
		u := &unstructured.Unstructured{}
		u.SetGroupVersionKind(schema.GroupVersionKind{Group: r.gvk.Group, Kind: id.Kind})
		u.SetNamespace(id.Namespace)
		u.SetName(id.Name)

		r.events <- event.GenericEvent{Object: u}
	}
}

// updateStatus updates the PodStatus corresponding to the passed Mutator with whether the Mutator is enforced, and
// whether there is an error instantiating the Mutator within the controller.
func (r *Reconciler) updateStatus(ctx context.Context, id types.ID, updates ...statusUpdate) error {
	status, err := r.getOrCreatePodStatus(ctx, id)
	if err != nil {
		r.log.Info("could not get/create pod status object", "error", err)
		return err
	}

	for _, update := range updates {
		update(status)
	}

	err = r.Update(ctx, status)
	if err != nil {
		r.log.Error(err, "could not update mutator status")
	}

	return err
}

// updateStatusWithError unconditionally updates the PodStatus corresponding
// to obj with error.
func (r *Reconciler) updateStatusWithError(ctx context.Context, obj client.Object, err error) error {
	id := types.MakeID(obj)

	return r.updateStatus(ctx, id,
		setID(obj.GetUID()), setGeneration(obj.GetGeneration()),
		setEnforced(false), setErrors(err))
}

func symmetricDifference(left, right mutationschema.IDSet) mutationschema.IDSet {
	result := make(mutationschema.IDSet)

	for id := range left {
		if !right[id] {
			result[id] = true
		}
	}
	for id := range right {
		if !left[id] {
			result[id] = true
		}
	}

	return result
}