feat: recursive expansion (open-policy-agent#2679)

Implements recursive expansion

Signed-off-by: Anlan Du <[email protected]>

go.mod

@@ -8,6 +8,7 @@ require (
 	contrib.go.opencensus.io/exporter/prometheus v0.4.2
 	contrib.go.opencensus.io/exporter/stackdriver v0.13.14
 	github.com/davecgh/go-spew v1.1.1
+	github.com/dominikbraun/graph v0.16.2
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-logr/logr v1.2.4
 	github.com/go-logr/zapr v1.2.3

go.sum

@@ -406,6 +406,8 @@ github.com/docker/libtrust v0.0.0-20150114040149-fa567046d9b1 h1:ZClxb8laGDf5arX
 github.com/docker/libtrust v0.0.0-20150114040149-fa567046d9b1/go.mod h1:cyGadeNEkKy96OOhEzfZl+yxihPEzKnqJwvfuSUqbZE=
 github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
+github.com/dominikbraun/graph v0.16.2 h1:EUndsCgHNQDHBdT4Q4M9GBePH3Tt0sV7DDPVWzfbEh4=
+github.com/dominikbraun/graph v0.16.2/go.mod h1:yOjYyogZLY1LSG9E33JWZJiq5k83Qy2C6POAuiViluc=
 github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=

pkg/controller/expansion/expansion_controller.go

@@ -8,6 +8,7 @@ import (
 	"github.com/open-policy-agent/frameworks/constraint/pkg/externaldata"
 	"github.com/open-policy-agent/gatekeeper/apis/expansion/unversioned"
 	"github.com/open-policy-agent/gatekeeper/apis/expansion/v1alpha1"
+	expansionv1alpha1 "github.com/open-policy-agent/gatekeeper/apis/expansion/v1alpha1"
 	"github.com/open-policy-agent/gatekeeper/apis/status/v1beta1"
 	statusv1beta1 "github.com/open-policy-agent/gatekeeper/apis/status/v1beta1"
 	"github.com/open-policy-agent/gatekeeper/pkg/expansion"
@@ -20,10 +21,12 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -33,6 +36,9 @@ import (
 
 var log = logf.Log.WithName("controller").WithValues("kind", "ExpansionTemplate", logging.Process, "template_expansion_controller")
 
+// eventQueueSize is how many events to queue before blocking.
+const eventQueueSize = 1024
+
 type Adder struct {
 	WatchManager    *watch.Manager
 	ExpansionSystem *expansion.System
@@ -78,11 +84,18 @@ type Reconciler struct {
 	registry     *etRegistry
 	statusClient client.StatusClient
 	tracker      *readiness.Tracker
+	events       chan event.GenericEvent
+	eventSource  source.Source
 
 	getPod func(context.Context) (*corev1.Pod, error)
 }
 
-func newReconciler(mgr manager.Manager, system *expansion.System, getPod func(ctx context.Context) (*corev1.Pod, error), tracker *readiness.Tracker) *Reconciler {
+func newReconciler(mgr manager.Manager,
+	system *expansion.System,
+	getPod func(ctx context.Context) (*corev1.Pod, error),
+	tracker *readiness.Tracker,
+) *Reconciler {
+	ev := make(chan event.GenericEvent, eventQueueSize)
 	return &Reconciler{
 		Client:       mgr.GetClient(),
 		system:       system,
@@ -91,23 +104,36 @@ func newReconciler(mgr manager.Manager, system *expansion.System, getPod func(ct
 		statusClient: mgr.GetClient(),
 		getPod:       getPod,
 		tracker:      tracker,
+		events:       ev,
+		eventSource:  &source.Channel{Source: ev, DestBufferSize: 1024},
 	}
 }
 
-func add(mgr manager.Manager, r reconcile.Reconciler) error {
+func add(mgr manager.Manager, r *Reconciler) error {
 	c, err := controller.New("expansion-template-controller", mgr, controller.Options{Reconciler: r})
 	if err != nil {
 		return err
 	}
 
+	// Watch for enqueued events
+	if r.eventSource != nil {
+		err = c.Watch(
+			r.eventSource,
+			&handler.EnqueueRequestForObject{})
+		if err != nil {
+			return err
+		}
+	}
+
+	// Watch for changes to ExpansionTemplates
 	return c.Watch(
 		&source.Kind{Type: &v1alpha1.ExpansionTemplate{}},
 		&handler.EnqueueRequestForObject{})
 }
 
 func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
 	defer r.registry.report(ctx)
-	log.Info("Reconcile", "request", request, "namespace", request.Namespace, "name", request.Name)
+	log.V(logging.DebugLevel).Info("Reconcile", "request", request, "namespace", request.Namespace, "name", request.Name)
 
 	deleted := false
 	versionedET := &v1alpha1.ExpansionTemplate{}
@@ -123,6 +149,11 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) (
 	if err := r.scheme.Convert(versionedET, et, nil); err != nil {
 		return reconcile.Result{}, err
 	}
+	oldConflicts := r.system.GetConflicts()
+
+	if !et.GetDeletionTimestamp().IsZero() {
+		deleted = true
+	}
 
 	if deleted {
 		// et will be an empty struct. We set the metadata name, which is
@@ -132,9 +163,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) (
 			r.getTracker().TryCancelExpect(versionedET)
 			return reconcile.Result{}, err
 		}
-		log.Info("removed expansion template", "template name", et.GetName())
+		log.V(logging.DebugLevel).Info("removed expansion template", "template name", et.GetName())
 		r.registry.remove(request.NamespacedName)
 		r.getTracker().CancelExpect(versionedET)
+		r.queueConflicts(oldConflicts)
 		return reconcile.Result{}, r.deleteStatus(ctx, request.NamespacedName.Name)
 	}
 
@@ -149,9 +181,38 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) (
 		log.Error(upsertErr, "upserting template", "template_name", et.GetName())
 	}
 
+	r.queueConflicts(oldConflicts)
 	return reconcile.Result{}, r.updateOrCreatePodStatus(ctx, et, upsertErr)
 }
 
+func (r *Reconciler) queueConflicts(old expansion.IDSet) {
+	for tID := range symmetricDiff(old, r.system.GetConflicts()) {
+		u := &unstructured.Unstructured{}
+		u.SetGroupVersionKind(expansionv1alpha1.GroupVersion.WithKind("ExpansionTemplate"))
+		// ExpansionTemplate is cluster-scoped, so we do not set namespace
+		u.SetName(string(tID))
+
+		r.events <- event.GenericEvent{Object: u}
+	}
+}
+
+func symmetricDiff(x, y expansion.IDSet) expansion.IDSet {
+	sDiff := make(expansion.IDSet)
+
+	for id := range x {
+		if _, exists := y[id]; !exists {
+			sDiff[id] = true
+		}
+	}
+	for id := range y {
+		if _, exists := x[id]; !exists {
+			sDiff[id] = true
+		}
+	}
+
+	return sDiff
+}
+
 func (r *Reconciler) deleteStatus(ctx context.Context, etName string) error {
 	status := &v1beta1.ExpansionTemplatePodStatus{}
 	pod, err := r.getPod(ctx)
@@ -227,5 +288,5 @@ func setStatusError(status *v1beta1.ExpansionTemplatePodStatus, etErr error) {
 	}
 
 	e := &v1beta1.ExpansionTemplateError{Message: etErr.Error()}
-	status.Status.Errors = append(status.Status.Errors, e)
+	status.Status.Errors = []*statusv1beta1.ExpansionTemplateError{e}
 }