chore: Remove deprovisioning.Reporter in favor of native eventing

pkg/controllers/deprovisioning/consolidation.go

@@ -29,8 +29,10 @@ import (
 
 	"github.com/aws/karpenter-core/pkg/apis/v1alpha5"
 	"github.com/aws/karpenter-core/pkg/cloudprovider"
+	deprovisioningevents "github.com/aws/karpenter-core/pkg/controllers/deprovisioning/events"
 	"github.com/aws/karpenter-core/pkg/controllers/provisioning"
 	"github.com/aws/karpenter-core/pkg/controllers/state"
+	"github.com/aws/karpenter-core/pkg/events"
 	"github.com/aws/karpenter-core/pkg/metrics"
 	"github.com/aws/karpenter-core/pkg/scheduling"
 )
@@ -43,19 +45,19 @@ type consolidation struct {
 	kubeClient             client.Client
 	provisioner            *provisioning.Provisioner
 	cloudProvider          cloudprovider.CloudProvider
-	reporter               *Reporter
+	recorder               events.Recorder
 	lastConsolidationState int64
 }
 
 func makeConsolidation(clock clock.Clock, cluster *state.Cluster, kubeClient client.Client, provisioner *provisioning.Provisioner,
-	cloudProvider cloudprovider.CloudProvider, reporter *Reporter) consolidation {
+	cloudProvider cloudprovider.CloudProvider, recorder events.Recorder) consolidation {
 	return consolidation{
 		clock:                  clock,
 		cluster:                cluster,
 		kubeClient:             kubeClient,
 		provisioner:            provisioner,
 		cloudProvider:          cloudProvider,
-		reporter:               reporter,
+		recorder:               recorder,
 		lastConsolidationState: 0,
 	}
 }
@@ -79,7 +81,7 @@ func (c *consolidation) sortAndFilterCandidates(ctx context.Context, nodes []Can
 	// filter out nodes that can't be terminated
 	nodes = lo.Filter(nodes, func(cn CandidateNode, _ int) bool {
 		if reason, canTerminate := canBeTerminated(cn, pdbs); !canTerminate {
-			c.reporter.RecordUnconsolidatableReason(ctx, cn.Node, reason)
+			c.recorder.Publish(deprovisioningevents.UnconsolidatableReason(cn.Node, reason))
 			return false
 		}
 		return true
@@ -94,15 +96,15 @@ func (c *consolidation) sortAndFilterCandidates(ctx context.Context, nodes []Can
 // ShouldDeprovision is a predicate used to filter deprovisionable nodes
 func (c *consolidation) ShouldDeprovision(ctx context.Context, n *state.Node, provisioner *v1alpha5.Provisioner, _ []*v1.Pod) bool {
 	if val, ok := n.Annotations()[v1alpha5.DoNotConsolidateNodeAnnotationKey]; ok {
-		c.reporter.RecordUnconsolidatableReason(ctx, n.Node, fmt.Sprintf("%s annotation exists", v1alpha5.DoNotConsolidateNodeAnnotationKey))
+		c.recorder.Publish(deprovisioningevents.UnconsolidatableReason(n.Node, fmt.Sprintf("%s annotation exists", v1alpha5.DoNotConsolidateNodeAnnotationKey)))
 		return val != "true"
 	}
 	if provisioner == nil {
-		c.reporter.RecordUnconsolidatableReason(ctx, n.Node, "provisioner is unknown")
+		c.recorder.Publish(deprovisioningevents.UnconsolidatableReason(n.Node, "provisioner is unknown"))
 		return false
 	}
 	if provisioner.Spec.Consolidation == nil || !ptr.BoolValue(provisioner.Spec.Consolidation.Enabled) {
-		c.reporter.RecordUnconsolidatableReason(ctx, n.Node, fmt.Sprintf("provisioner %s has consolidation disabled", provisioner.Name))
+		c.recorder.Publish(deprovisioningevents.UnconsolidatableReason(n.Node, fmt.Sprintf("provisioner %s has consolidation disabled", provisioner.Name)))
 		return false
 	}
 	return true
@@ -191,7 +193,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, nodes ...Candi
 	if !allPodsScheduled {
 		// This method is used by multi-node consolidation as well, so we'll only report in the single node case
 		if len(nodes) == 1 {
-			c.reporter.RecordUnconsolidatableReason(ctx, nodes[0].Node, "not all pods would schedule")
+			c.recorder.Publish(deprovisioningevents.UnconsolidatableReason(nodes[0].Node, "not all pods would schedule"))
 		}
 		return Command{action: actionDoNothing}, nil
 	}
@@ -207,7 +209,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, nodes ...Candi
 	// we're not going to turn a single node into multiple nodes
 	if len(newNodes) != 1 {
 		if len(nodes) == 1 {
-			c.reporter.RecordUnconsolidatableReason(ctx, nodes[0].Node, fmt.Sprintf("can't remove without creating %d nodes", len(newNodes)))
+			c.recorder.Publish(deprovisioningevents.UnconsolidatableReason(nodes[0].Node, fmt.Sprintf("can't remove without creating %d nodes", len(newNodes))))
 		}
 		return Command{action: actionDoNothing}, nil
 	}
@@ -221,7 +223,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, nodes ...Candi
 	newNodes[0].InstanceTypeOptions = filterByPrice(newNodes[0].InstanceTypeOptions, newNodes[0].Requirements, nodesPrice)
 	if len(newNodes[0].InstanceTypeOptions) == 0 {
 		if len(nodes) == 1 {
-			c.reporter.RecordUnconsolidatableReason(ctx, nodes[0].Node, "can't replace with a cheaper node")
+			c.recorder.Publish(deprovisioningevents.UnconsolidatableReason(nodes[0].Node, "can't replace with a cheaper node"))
 		}
 		// no instance types remain after filtering by price
 		return Command{action: actionDoNothing}, nil
@@ -240,7 +242,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, nodes ...Candi
 	if allExistingAreSpot &&
 		newNodes[0].Requirements.Get(v1alpha5.LabelCapacityType).Has(v1alpha5.CapacityTypeSpot) {
 		if len(nodes) == 1 {
-			c.reporter.RecordUnconsolidatableReason(ctx, nodes[0].Node, "can't replace a spot node with a spot node")
+			c.recorder.Publish(deprovisioningevents.UnconsolidatableReason(nodes[0].Node, "can't replace a spot node with a spot node"))
 		}
 		return Command{action: actionDoNothing}, nil
 	}

pkg/controllers/deprovisioning/controller.go

@@ -52,7 +52,6 @@ type Controller struct {
 	clock          clock.Clock
 	cloudProvider  cloudprovider.CloudProvider
 	deprovisioners []Deprovisioner
-	reporter       *Reporter
 }
 
 // pollingPeriod that we inspect cluster to look for opportunities to deprovision
@@ -73,14 +72,12 @@ var waitRetryOptions = []retry.Option{
 func NewController(clk clock.Clock, kubeClient client.Client, provisioner *provisioning.Provisioner,
 	cp cloudprovider.CloudProvider, recorder events.Recorder, cluster *state.Cluster) *Controller {
 
-	reporter := NewReporter(recorder)
 	return &Controller{
 		clock:         clk,
 		kubeClient:    kubeClient,
 		cluster:       cluster,
 		provisioner:   provisioner,
 		recorder:      recorder,
-		reporter:      reporter,
 		cloudProvider: cp,
 		deprovisioners: []Deprovisioner{
 			// Expire any nodes that must be deleted, allowing their pods to potentially land on currently
@@ -90,11 +87,11 @@ func NewController(clk clock.Clock, kubeClient client.Client, provisioner *provi
 			// Delete any remaining empty nodes as there is zero cost in terms of dirsuption.  Emptiness and
 			// emptyNodeConsolidation are mutually exclusive, only one of these will operate
 			NewEmptiness(clk),
-			NewEmptyNodeConsolidation(clk, cluster, kubeClient, provisioner, cp, reporter),
+			NewEmptyNodeConsolidation(clk, cluster, kubeClient, provisioner, cp, recorder),
 			// Attempt to identify multiple nodes that we can consolidate simultaneously to reduce pod churn
-			NewMultiNodeConsolidation(clk, cluster, kubeClient, provisioner, cp, reporter),
+			NewMultiNodeConsolidation(clk, cluster, kubeClient, provisioner, cp, recorder),
 			// And finally fall back our single node consolidation to further reduce cluster cost.
-			NewSingleNodeConsolidation(clk, cluster, kubeClient, provisioner, cp, reporter),
+			NewSingleNodeConsolidation(clk, cluster, kubeClient, provisioner, cp, recorder),
 		},
 	}
 }

pkg/controllers/deprovisioning/emptynodeconsolidation.go

@@ -28,6 +28,7 @@ import (
 	"github.com/aws/karpenter-core/pkg/cloudprovider"
 	"github.com/aws/karpenter-core/pkg/controllers/provisioning"
 	"github.com/aws/karpenter-core/pkg/controllers/state"
+	"github.com/aws/karpenter-core/pkg/events"
 )
 
 // EmptyNodeConsolidation is the consolidation controller that performs multi-node consolidation of entirely empty nodes
@@ -36,8 +37,8 @@ type EmptyNodeConsolidation struct {
 }
 
 func NewEmptyNodeConsolidation(clk clock.Clock, cluster *state.Cluster, kubeClient client.Client,
-	provisioner *provisioning.Provisioner, cp cloudprovider.CloudProvider, reporter *Reporter) *EmptyNodeConsolidation {
-	return &EmptyNodeConsolidation{consolidation: makeConsolidation(clk, cluster, kubeClient, provisioner, cp, reporter)}
+	provisioner *provisioning.Provisioner, cp cloudprovider.CloudProvider, recorder events.Recorder) *EmptyNodeConsolidation {
+	return &EmptyNodeConsolidation{consolidation: makeConsolidation(clk, cluster, kubeClient, provisioner, cp, recorder)}
 }
 
 // ComputeCommand generates a deprovisioning command given deprovisionable nodes

pkg/controllers/deprovisioning/events/events.go

@@ -16,6 +16,7 @@ package events
 
 import (
 	"fmt"
+	"time"
 
 	v1 "k8s.io/api/core/v1"
 
@@ -69,5 +70,6 @@ func UnconsolidatableReason(node *v1.Node, reason string) events.Event {
 		Reason:         "Unconsolidatable",
 		Message:        reason,
 		DedupeValues:   []string{node.Name},
+		DedupeTimeout:  time.Minute * 15,
 	}
 }

pkg/controllers/deprovisioning/multinodeconsolidation.go

@@ -27,15 +27,16 @@ import (
 	"github.com/aws/karpenter-core/pkg/controllers/provisioning"
 	"github.com/aws/karpenter-core/pkg/controllers/provisioning/scheduling"
 	"github.com/aws/karpenter-core/pkg/controllers/state"
+	"github.com/aws/karpenter-core/pkg/events"
 )
 
 type MultiNodeConsolidation struct {
 	consolidation
 }
 
 func NewMultiNodeConsolidation(clk clock.Clock, cluster *state.Cluster, kubeClient client.Client,
-	provisioner *provisioning.Provisioner, cp cloudprovider.CloudProvider, reporter *Reporter) *MultiNodeConsolidation {
-	return &MultiNodeConsolidation{makeConsolidation(clk, cluster, kubeClient, provisioner, cp, reporter)}
+	provisioner *provisioning.Provisioner, cp cloudprovider.CloudProvider, recorder events.Recorder) *MultiNodeConsolidation {
+	return &MultiNodeConsolidation{makeConsolidation(clk, cluster, kubeClient, provisioner, cp, recorder)}
 }
 
 func (m *MultiNodeConsolidation) ComputeCommand(ctx context.Context, candidates ...CandidateNode) (Command, error) {

pkg/controllers/deprovisioning/singlenodeconsolidation.go

@@ -25,6 +25,7 @@ import (
 	"github.com/aws/karpenter-core/pkg/cloudprovider"
 	"github.com/aws/karpenter-core/pkg/controllers/provisioning"
 	"github.com/aws/karpenter-core/pkg/controllers/state"
+	"github.com/aws/karpenter-core/pkg/events"
 )
 
 // SingleNodeConsolidation is the consolidation controller that performs single node consolidation.
@@ -33,8 +34,8 @@ type SingleNodeConsolidation struct {
 }
 
 func NewSingleNodeConsolidation(clk clock.Clock, cluster *state.Cluster, kubeClient client.Client, provisioner *provisioning.Provisioner,
-	cp cloudprovider.CloudProvider, reporter *Reporter) *SingleNodeConsolidation {
-	return &SingleNodeConsolidation{consolidation: makeConsolidation(clk, cluster, kubeClient, provisioner, cp, reporter)}
+	cp cloudprovider.CloudProvider, recorder events.Recorder) *SingleNodeConsolidation {
+	return &SingleNodeConsolidation{consolidation: makeConsolidation(clk, cluster, kubeClient, provisioner, cp, recorder)}
 }
 
 // ComputeCommand generates a deprovisioning command given deprovisionable nodes

pkg/events/recorder.go

@@ -31,6 +31,7 @@ type Event struct {
 	Reason         string
 	Message        string
 	DedupeValues   []string
+	DedupeTimeout  time.Duration
 	RateLimiter    flowcontrol.RateLimiter
 }
 
@@ -50,17 +51,24 @@ type recorder struct {
 	cache *cache.Cache
 }
 
+const defaultDedupeTimeout = 2 * time.Minute
+
 func NewRecorder(r record.EventRecorder) Recorder {
 	return &recorder{
 		rec:   r,
-		cache: cache.New(120*time.Second, 10*time.Second),
+		cache: cache.New(defaultDedupeTimeout, 10*time.Second),
 	}
 }
 
 // Publish creates a Kubernetes event using the passed event struct
 func (r *recorder) Publish(evt Event) {
+	// Override the timeout if one is set for an event
+	timeout := defaultDedupeTimeout
+	if evt.DedupeTimeout != 0 {
+		timeout = evt.DedupeTimeout
+	}
 	// Dedupe same events that involve the same object and are close together
-	if len(evt.DedupeValues) > 0 && !r.shouldCreateEvent(evt.dedupeKey()) {
+	if len(evt.DedupeValues) > 0 && !r.shouldCreateEvent(evt.dedupeKey(), timeout) {
 		return
 	}
 	// If the event is rate-limited, then validate we should create the event
@@ -70,10 +78,10 @@ func (r *recorder) Publish(evt Event) {
 	r.rec.Event(evt.InvolvedObject, evt.Type, evt.Reason, evt.Message)
 }
 
-func (r *recorder) shouldCreateEvent(key string) bool {
+func (r *recorder) shouldCreateEvent(key string, timeout time.Duration) bool {
 	if _, exists := r.cache.Get(key); exists {
 		return false
 	}
-	r.cache.SetDefault(key, nil)
+	r.cache.Set(key, nil, timeout)
 	return true
 }

pkg/events/suite_test.go

@@ -103,6 +103,22 @@ var _ = Describe("Dedupe", func() {
 		}
 		Expect(internalRecorder.Calls(events.EvictPod(PodWithUID()).Reason)).To(Equal(1))
 	})
+	It("should allow the dedupe timeout to be overridden", func() {
+		pod := PodWithUID()
+		evt := events.EvictPod(pod)
+		evt.DedupeTimeout = time.Second * 2
+
+		// Generate a set of events within the dedupe timeout
+		for i := 0; i < 10; i++ {
+			eventRecorder.Publish(evt)
+		}
+		Expect(internalRecorder.Calls(events.EvictPod(PodWithUID()).Reason)).To(Equal(1))
+
+		// Wait until after the overridden dedupe timeout
+		time.Sleep(time.Second * 3)
+		eventRecorder.Publish(evt)
+		Expect(internalRecorder.Calls(events.EvictPod(PodWithUID()).Reason)).To(Equal(2))
+	})
 	It("should allow events with different entities to be created", func() {
 		for i := 0; i < 100; i++ {
 			eventRecorder.Publish(events.EvictPod(PodWithUID()))