Repository: akuity/kargo
Pull request #1859 (2024-04-17 16:59:11)

fix: create a new Event on every correlated event
Part of https://github.com/akuity/kargo/issues/1231

The default Kubernetes event recorder aggregates (patches) Event instead of creating them for correlated events. Since we heavily depend on Kubernetes events to track Kargo events and metrics, it has been difficult to track exact events.

This commit fixes this behavior by creating a custom event recorder that creates a new Event for every emitted event.

---

#### References

- [client-go/tools/record/event.go](https://github.com/kubernetes/client-go/blob/aa7909e7d7c0661792ba21b9e882f3cd6ad0ce53/tools/record/event.go#L288-L304)

netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *docs-kargo-akuity-io* processing.


|  Name | Link |
|:-:|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 172ca1994ed6ee300ff7599e9630f4c4252149c6 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/docs-kargo-akuity-io/deploys/661fffe184604c0007d084d7 |
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/akuity/kargo/pull/1859?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) Report
Attention: Patch coverage is `39.34426%` with `37 lines` in your changes are missing coverage. Please review.
> Project coverage is 46.14%. Comparing base [(`fcf8948`)](https://app.codecov.io/gh/akuity/kargo/commit/fcf894808be0aa345b2bcd43f1a4333d820fbcb9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) to head [(`172ca19`)](https://app.codecov.io/gh/akuity/kargo/pull/1859?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity).

| [Files](https://app.codecov.io/gh/akuity/kargo/pull/1859?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) | Patch % | Lines |
|---|---|---|
| [internal/kubeclient/event.go](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&filepath=internal%2Fkubeclient%2Fevent.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity#diff-aW50ZXJuYWwva3ViZWNsaWVudC9ldmVudC5nbw==) | 43.63% | [31 Missing :warning: ](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) |
| [cmd/controlplane/webhooks.go](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&filepath=cmd%2Fcontrolplane%2Fwebhooks.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity#diff-Y21kL2NvbnRyb2xwbGFuZS93ZWJob29rcy5nbw==) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) |
| [cmd/controlplane/api.go](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&filepath=cmd%2Fcontrolplane%2Fapi.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity#diff-Y21kL2NvbnRyb2xwbGFuZS9hcGkuZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) |
| [internal/controller/promotions/promotions.go](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&filepath=internal%2Fcontroller%2Fpromotions%2Fpromotions.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity#diff-aW50ZXJuYWwvY29udHJvbGxlci9wcm9tb3Rpb25zL3Byb21vdGlvbnMuZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) |
| [internal/webhook/freight/webhook.go](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&filepath=internal%2Fwebhook%2Ffreight%2Fwebhook.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity#diff-aW50ZXJuYWwvd2ViaG9vay9mcmVpZ2h0L3dlYmhvb2suZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) |
| [internal/webhook/promotion/webhook.go](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&filepath=internal%2Fwebhook%2Fpromotion%2Fwebhook.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity#diff-aW50ZXJuYWwvd2ViaG9vay9wcm9tb3Rpb24vd2ViaG9vay5nbw==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/akuity/kargo/pull/1859?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #1859      +/-   ##
==========================================
- Coverage   46.15%   46.14%   -0.01%     
==========================================
  Files         213      214       +1     
  Lines       13865    13920      +55     
==========================================
+ Hits         6400     6424      +24     
- Misses       7181     7212      +31     
  Partials      284      284              
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/akuity/kargo/pull/1859?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity).

krancour: Noted a few minor things, but high level, this LGTM.

Patch:
diff --git a/cmd/controlplane/api.go b/cmd/controlplane/api.go
index ee7d8552..de0becd2 100644
--- a/cmd/controlplane/api.go
+++ b/cmd/controlplane/api.go
@@ -22,6 +22,7 @@ import (
 	"github.com/akuity/kargo/internal/api/kubernetes"
 	rollouts "github.com/akuity/kargo/internal/controller/rollouts/api/v1alpha1"
 	"github.com/akuity/kargo/internal/kubeclient"
+	libEvent "github.com/akuity/kargo/internal/kubernetes/event"
 	"github.com/akuity/kargo/internal/os"
 	versionpkg "github.com/akuity/kargo/internal/version"
 )
@@ -163,7 +164,7 @@ func (o *apiOptions) setupAPIClient(ctx context.Context) (*rest.Config, client.C
 		}
 	}()
 
-	return restCfg, mgr.GetClient(), mgr.GetEventRecorderFor("api"), nil
+	return restCfg, mgr.GetClient(), libEvent.NewRecorder(ctx, scheme, mgr.GetClient(), "api"), nil
 }
 
 func registerKargoIndexers(ctx context.Context, mgr ctrl.Manager) error {
diff --git a/cmd/controlplane/webhooks.go b/cmd/controlplane/webhooks.go
index 50b6be4e..a54ec97e 100644
--- a/cmd/controlplane/webhooks.go
+++ b/cmd/controlplane/webhooks.go
@@ -108,7 +108,7 @@ func (o *webhooksServerOptions) run(ctx context.Context) error {
 		return fmt.Errorf("index Stages by Freight: %w", err)
 	}
 
-	if err = freight.SetupWebhookWithManager(webhookCfg, mgr); err != nil {
+	if err = freight.SetupWebhookWithManager(ctx, webhookCfg, mgr); err != nil {
 		return fmt.Errorf("setup Freight webhook: %w", err)
 	}
 	if err = project.SetupWebhookWithManager(
@@ -117,7 +117,7 @@ func (o *webhooksServerOptions) run(ctx context.Context) error {
 	); err != nil {
 		return fmt.Errorf("setup Project webhook: %w", err)
 	}
-	if err = promotion.SetupWebhookWithManager(webhookCfg, mgr); err != nil {
+	if err = promotion.SetupWebhookWithManager(ctx, webhookCfg, mgr); err != nil {
 		return fmt.Errorf("setup Promotion webhook: %w", err)
 	}
 	if err = stage.SetupWebhookWithManager(webhookCfg, mgr); err != nil {
diff --git a/internal/cli/cmd/server/server.go b/internal/cli/cmd/server/server.go
index 8f90884e..9c926d52 100644
--- a/internal/cli/cmd/server/server.go
+++ b/internal/cli/cmd/server/server.go
@@ -14,7 +14,7 @@ import (
 	"github.com/akuity/kargo/internal/api/kubernetes"
 	"github.com/akuity/kargo/internal/cli/option"
 	"github.com/akuity/kargo/internal/cli/templates"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 )
 
 type serverOptions struct {
@@ -100,7 +100,7 @@ func (o *serverOptions) run(ctx context.Context) error {
 		},
 		client,
 		client,
-		&fakekubeclient.EventRecorder{},
+		&fakeevent.EventRecorder{},
 	)
 	if err := srv.Serve(ctx, l); err != nil {
 		return fmt.Errorf("serve error: %w", err)
diff --git a/internal/controller/promotions/promotions.go b/internal/controller/promotions/promotions.go
index 109638f3..b646750d 100644
--- a/internal/controller/promotions/promotions.go
+++ b/internal/controller/promotions/promotions.go
@@ -25,6 +25,7 @@ import (
 	"github.com/akuity/kargo/internal/credentials"
 	"github.com/akuity/kargo/internal/kargo"
 	"github.com/akuity/kargo/internal/kubeclient"
+	libEvent "github.com/akuity/kargo/internal/kubernetes/event"
 	"github.com/akuity/kargo/internal/logging"
 )
 
@@ -93,7 +94,7 @@ func SetupReconcilerWithManager(
 	reconciler := newReconciler(
 		kargoMgr.GetClient(),
 		argocdClient,
-		kargoMgr.GetEventRecorderFor(cfg.Name()),
+		libEvent.NewRecorder(ctx, kargoMgr.GetScheme(), kargoMgr.GetClient(), cfg.Name()),
 		credentialsDB,
 		cfg,
 	)
diff --git a/internal/controller/stages/stages.go b/internal/controller/stages/stages.go
index e9c54342..6ada3b0d 100644
--- a/internal/controller/stages/stages.go
+++ b/internal/controller/stages/stages.go
@@ -32,6 +32,7 @@ import (
 	rollouts "github.com/akuity/kargo/internal/controller/rollouts/api/v1alpha1"
 	"github.com/akuity/kargo/internal/kargo"
 	"github.com/akuity/kargo/internal/kubeclient"
+	libEvent "github.com/akuity/kargo/internal/kubernetes/event"
 	"github.com/akuity/kargo/internal/logging"
 )
 
@@ -332,7 +333,7 @@ func SetupReconcilerWithManager(
 			newReconciler(
 				kargoMgr.GetClient(),
 				argocdClient,
-				kargoMgr.GetEventRecorderFor(cfg.Name()),
+				libEvent.NewRecorder(ctx, kargoMgr.GetScheme(), kargoMgr.GetClient(), cfg.Name()),
 				cfg,
 				shardRequirement,
 			),
diff --git a/internal/kubernetes/event/event.go b/internal/kubernetes/event/event.go
new file mode 100644
index 00000000..0dd6362b
--- /dev/null
+++ b/internal/kubernetes/event/event.go
@@ -0,0 +1,139 @@
+package event
+
+import (
+	"context"
+	"errors"
+
+	log "github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/retry"
+	libClient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/akuity/kargo/internal/logging"
+)
+
+type recorder struct {
+	backoff wait.Backoff
+	sink    record.EventSink
+	logger  *log.Entry
+
+	newEventHandlerFn func(event *corev1.Event) func() error
+}
+
+// NewRecorder returns a new record.EventRecorder that records all events
+// without aggregation, even the given event is correlated.
+//
+// NOTE: This recorder must be used with caution as it creates a new Event
+// on every event without throttling / spam filtering features - which are
+// included in the event recorder from k8s.io/client-go. This could lead
+// to performance issues with the Kubernetes API server.
+func NewRecorder(
+	ctx context.Context,
+	scheme *runtime.Scheme,
+	client libClient.Client,
+	name string,
+) record.EventRecorder {
+	logger := logging.LoggerFromContext(ctx)
+	internalRecorder := newRecorder(ctx, client, logger)
+	b := record.NewBroadcaster()
+	b.StartEventWatcher(internalRecorder.handleEvent)
+	return b.NewRecorder(
+		scheme,
+		corev1.EventSource{
+			Component: name,
+		},
+	)
+}
+
+func newRecorder(
+	ctx context.Context,
+	client libClient.Client,
+	logger *log.Entry,
+) *recorder {
+	r := &recorder{
+		backoff: retry.DefaultRetry, // TODO: Make it configurable
+		sink:    newSink(ctx, client),
+		logger:  logger,
+	}
+	r.newEventHandlerFn = r.createEvent
+	return r
+}
+
+func (r *recorder) handleEvent(event *corev1.Event) {
+	if err := retry.OnError(
+		r.backoff,
+		r.newRetryDecider(event),
+		r.newEventHandlerFn(event),
+	); err != nil {
+		r.logger.WithError(err).Error("Unable to handle event", "event", event)
+	}
+}
+
+func (r *recorder) createEvent(event *corev1.Event) func() error {
+	return func() error {
+		// Always create event instead of patching correlated events
+		_, err := r.sink.Create(event)
+		return err
+	}
+}
+
+// newRetryDecider returns a function that decides whether
+// to re-record event or not on given error.
+func (r *recorder) newRetryDecider(event *corev1.Event) func(error) bool {
+	return func(err error) bool {
+		logger := r.logger.
+			WithField("event", event).
+			WithError(err)
+
+		var statusErr *apierrors.StatusError
+		if errors.As(err, &statusErr) {
+			if apierrors.IsAlreadyExists(err) ||
+				apierrors.HasStatusCause(err, corev1.NamespaceTerminatingCause) {
+				logger.Info("Server rejected event (will not retry!)")
+				return false
+			}
+			// Retry on other status errors
+		}
+		logger.Error("Unable to write event (may retry after backoff)")
+		return true
+	}
+}
+
+var (
+	_ record.EventSink = &sink{}
+)
+
+type sink struct {
+	ctx    context.Context
+	client libClient.Client
+}
+
+func newSink(
+	ctx context.Context,
+	client libClient.Client,
+) *sink {
+	return &sink{
+		ctx:    ctx,
+		client: client,
+	}
+}
+
+func (e *sink) Create(event *corev1.Event) (*corev1.Event, error) {
+	err := e.client.Create(e.ctx, event)
+	return event, err
+}
+
+func (e *sink) Update(event *corev1.Event) (*corev1.Event, error) {
+	err := e.client.Update(e.ctx, event)
+	return event, err
+}
+
+func (e *sink) Patch(event *corev1.Event, data []byte) (*corev1.Event, error) {
+	err := e.client.Patch(e.ctx, event, libClient.RawPatch(types.StrategicMergePatchType, data))
+	return event, err
+}
diff --git a/internal/kubeclient/fake/event.go b/internal/kubernetes/event/fake/event.go
similarity index 100%
rename from internal/kubeclient/fake/event.go
rename to internal/kubernetes/event/fake/event.go
diff --git a/internal/webhook/freight/webhook.go b/internal/webhook/freight/webhook.go
index e4409026..c94f1600 100644
--- a/internal/webhook/freight/webhook.go
+++ b/internal/webhook/freight/webhook.go
@@ -19,6 +19,7 @@ import (
 
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
 	"github.com/akuity/kargo/internal/kubeclient"
+	libEvent "github.com/akuity/kargo/internal/kubernetes/event"
 	libWebhook "github.com/akuity/kargo/internal/webhook"
 )
 
@@ -68,13 +69,14 @@ type webhook struct {
 }
 
 func SetupWebhookWithManager(
+	ctx context.Context,
 	cfg libWebhook.Config,
 	mgr ctrl.Manager,
 ) error {
 	w := newWebhook(
 		cfg,
 		mgr.GetClient(),
-		mgr.GetEventRecorderFor("freight-webhook"),
+		libEvent.NewRecorder(ctx, mgr.GetScheme(), mgr.GetClient(), "freight-webhook"),
 	)
 	return ctrl.NewWebhookManagedBy(mgr).
 		For(&kargoapi.Freight{}).
diff --git a/internal/webhook/promotion/webhook.go b/internal/webhook/promotion/webhook.go
index 62c25f93..579f390a 100644
--- a/internal/webhook/promotion/webhook.go
+++ b/internal/webhook/promotion/webhook.go
@@ -18,6 +18,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
+	libEvent "github.com/akuity/kargo/internal/kubernetes/event"
 	"github.com/akuity/kargo/internal/logging"
 	libWebhook "github.com/akuity/kargo/internal/webhook"
 )
@@ -77,13 +78,14 @@ type webhook struct {
 }
 
 func SetupWebhookWithManager(
+	ctx context.Context,
 	cfg libWebhook.Config,
 	mgr ctrl.Manager,
 ) error {
 	w := newWebhook(
 		cfg,
 		mgr.GetClient(),
-		mgr.GetEventRecorderFor("promotion-webhook"),
+		libEvent.NewRecorder(ctx, mgr.GetScheme(), mgr.GetClient(), "promotion-webhook"),
 	)
 	return ctrl.NewWebhookManagedBy(mgr).
 		For(&kargoapi.Promotion{}).


Test patch:
diff --git a/internal/api/approve_freight_v1alpha1_test.go b/internal/api/approve_freight_v1alpha1_test.go
index 45477e8f..ac9b9516 100644
--- a/internal/api/approve_freight_v1alpha1_test.go
+++ b/internal/api/approve_freight_v1alpha1_test.go
@@ -13,7 +13,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 	svcv1alpha1 "github.com/akuity/kargo/pkg/api/service/v1alpha1"
 )
 
@@ -24,7 +24,7 @@ func TestApproveFreight(t *testing.T) {
 		server     *server
 		assertions func(
 			*testing.T,
-			*fakekubeclient.EventRecorder,
+			*fakeevent.EventRecorder,
 			*connect.Response[svcv1alpha1.ApproveFreightResponse],
 			error,
 		)
@@ -35,7 +35,7 @@ func TestApproveFreight(t *testing.T) {
 			server: &server{},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -59,7 +59,7 @@ func TestApproveFreight(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -90,7 +90,7 @@ func TestApproveFreight(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -121,7 +121,7 @@ func TestApproveFreight(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -163,7 +163,7 @@ func TestApproveFreight(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -201,7 +201,7 @@ func TestApproveFreight(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -252,7 +252,7 @@ func TestApproveFreight(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -306,7 +306,7 @@ func TestApproveFreight(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -360,7 +360,7 @@ func TestApproveFreight(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.ApproveFreightResponse],
 				err error,
 			) {
@@ -374,7 +374,7 @@ func TestApproveFreight(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			recorder := fakekubeclient.NewEventRecorder(1)
+			recorder := fakeevent.NewEventRecorder(1)
 			testCase.server.recorder = recorder
 			resp, err := testCase.server.ApproveFreight(
 				context.Background(),
diff --git a/internal/api/promote_to_stage_subscribers_v1alpha1_test.go b/internal/api/promote_to_stage_subscribers_v1alpha1_test.go
index 4a78c287..bad6d64c 100644
--- a/internal/api/promote_to_stage_subscribers_v1alpha1_test.go
+++ b/internal/api/promote_to_stage_subscribers_v1alpha1_test.go
@@ -13,7 +13,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 	svcv1alpha1 "github.com/akuity/kargo/pkg/api/service/v1alpha1"
 )
 
@@ -24,7 +24,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 		server     *server
 		assertions func(
 			*testing.T,
-			*fakekubeclient.EventRecorder,
+			*fakeevent.EventRecorder,
 			*connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 			error,
 		)
@@ -34,7 +34,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			server: &server{},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -58,7 +58,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -87,7 +87,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -116,7 +116,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -166,7 +166,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -212,7 +212,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -265,7 +265,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -324,7 +324,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -379,7 +379,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -449,7 +449,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -526,7 +526,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -606,7 +606,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				res *connect.Response[svcv1alpha1.PromoteToStageSubscribersResponse],
 				err error,
 			) {
@@ -622,7 +622,7 @@ func TestPromoteToStageSubscribers(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			recorder := fakekubeclient.NewEventRecorder(1)
+			recorder := fakeevent.NewEventRecorder(1)
 			testCase.server.recorder = recorder
 			resp, err := testCase.server.PromoteToStageSubscribers(
 				context.Background(),
diff --git a/internal/api/promote_to_stage_v1alpha1_test.go b/internal/api/promote_to_stage_v1alpha1_test.go
index 47f9eb2f..771a4d22 100644
--- a/internal/api/promote_to_stage_v1alpha1_test.go
+++ b/internal/api/promote_to_stage_v1alpha1_test.go
@@ -13,7 +13,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 	svcv1alpha1 "github.com/akuity/kargo/pkg/api/service/v1alpha1"
 )
 
@@ -24,7 +24,7 @@ func TestPromoteToStage(t *testing.T) {
 		server     *server
 		assertions func(
 			*testing.T,
-			*fakekubeclient.EventRecorder,
+			*fakeevent.EventRecorder,
 			*connect.Response[svcv1alpha1.PromoteToStageResponse],
 			error,
 		)
@@ -35,7 +35,7 @@ func TestPromoteToStage(t *testing.T) {
 			server: &server{},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -59,7 +59,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -88,7 +88,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -117,7 +117,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -167,7 +167,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -213,7 +213,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -266,7 +266,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -330,7 +330,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -395,7 +395,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				_ *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -460,7 +460,7 @@ func TestPromoteToStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				res *connect.Response[svcv1alpha1.PromoteToStageResponse],
 				err error,
 			) {
@@ -476,7 +476,7 @@ func TestPromoteToStage(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			recorder := fakekubeclient.NewEventRecorder(1)
+			recorder := fakeevent.NewEventRecorder(1)
 			testCase.server.recorder = recorder
 			res, err := testCase.server.PromoteToStage(
 				context.Background(),
diff --git a/internal/api/server_test.go b/internal/api/server_test.go
index 7953ed8b..5155800a 100644
--- a/internal/api/server_test.go
+++ b/internal/api/server_test.go
@@ -12,7 +12,7 @@ import (
 
 	"github.com/akuity/kargo/internal/api/config"
 	"github.com/akuity/kargo/internal/api/kubernetes"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 )
 
 func TestNewServer(t *testing.T) {
@@ -31,7 +31,7 @@ func TestNewServer(t *testing.T) {
 		},
 	)
 	require.NoError(t, err)
-	testRecorder := fakekubeclient.NewEventRecorder(0)
+	testRecorder := fakeevent.NewEventRecorder(0)
 
 	s, ok := NewServer(
 		testServerConfig,
diff --git a/internal/controller/promotions/promotions_test.go b/internal/controller/promotions/promotions_test.go
index e47d2bc5..463d39ff 100644
--- a/internal/controller/promotions/promotions_test.go
+++ b/internal/controller/promotions/promotions_test.go
@@ -15,7 +15,7 @@ import (
 	"github.com/akuity/kargo/api/v1alpha1"
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
 	"github.com/akuity/kargo/internal/credentials"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 )
 
 func TestNewPromotionReconciler(t *testing.T) {
@@ -23,7 +23,7 @@ func TestNewPromotionReconciler(t *testing.T) {
 	r := newReconciler(
 		kubeClient,
 		kubeClient,
-		&fakekubeclient.EventRecorder{},
+		&fakeevent.EventRecorder{},
 		&credentials.FakeDB{},
 		ReconcilerConfig{},
 	)
@@ -35,7 +35,7 @@ func TestNewPromotionReconciler(t *testing.T) {
 
 func newFakeReconciler(
 	t *testing.T,
-	recorder *fakekubeclient.EventRecorder,
+	recorder *fakeevent.EventRecorder,
 	objects ...client.Object,
 ) *reconciler {
 	scheme := k8sruntime.NewScheme()
@@ -151,7 +151,7 @@ func TestReconcile(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx := context.TODO()
-			recorder := fakekubeclient.NewEventRecorder(1)
+			recorder := fakeevent.NewEventRecorder(1)
 			r := newFakeReconciler(t, recorder, tc.promos...)
 			promoteWasCalled := false
 			r.getStageFn = func(context.Context, client.Client, types.NamespacedName) (*kargoapi.Stage, error) {
@@ -204,7 +204,7 @@ func TestReconcileInitializeQueues(t *testing.T) {
 		newPromo("fake-namespace", "fake-promo1", "fake-stage", kargoapi.PromotionPhasePending, before),
 		newPromo("fake-namespace", "fake-promo2", "fake-stage", kargoapi.PromotionPhasePending, now),
 	}
-	recorder := &fakekubeclient.EventRecorder{}
+	recorder := &fakeevent.EventRecorder{}
 	r := newFakeReconciler(t, recorder, promos...)
 
 	// reconcile a non-existent promo to trigger initializeQueues
diff --git a/internal/controller/stages/stages_test.go b/internal/controller/stages/stages_test.go
index 11a81ec5..ec2cbc80 100644
--- a/internal/controller/stages/stages_test.go
+++ b/internal/controller/stages/stages_test.go
@@ -16,7 +16,7 @@ import (
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
 	"github.com/akuity/kargo/internal/controller"
 	rollouts "github.com/akuity/kargo/internal/controller/rollouts/api/v1alpha1"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 )
 
 var (
@@ -30,7 +30,7 @@ func TestNewReconciler(t *testing.T) {
 	kubeClient := fake.NewClientBuilder().Build()
 	requirement, err := controller.GetShardRequirement(testCfg.ShardName)
 	require.NoError(t, err)
-	recorder := &fakekubeclient.EventRecorder{Events: nil}
+	recorder := &fakeevent.EventRecorder{Events: nil}
 	r := newReconciler(
 		kubeClient,
 		kubeClient,
@@ -83,7 +83,7 @@ func TestSyncControlFlowStage(t *testing.T) {
 		reconciler *reconciler
 		assertions func(
 			t *testing.T,
-			recorder *fakekubeclient.EventRecorder,
+			recorder *fakeevent.EventRecorder,
 			initialStatus kargoapi.StageStatus,
 			newStatus kargoapi.StageStatus,
 			err error,
@@ -112,7 +112,7 @@ func TestSyncControlFlowStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -152,7 +152,7 @@ func TestSyncControlFlowStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -204,7 +204,7 @@ func TestSyncControlFlowStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -257,7 +257,7 @@ func TestSyncControlFlowStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				_ kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -279,7 +279,7 @@ func TestSyncControlFlowStage(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			recorder := fakekubeclient.NewEventRecorder(1)
+			recorder := fakeevent.NewEventRecorder(1)
 			testCase.reconciler.nowFn = fakeNow
 			testCase.reconciler.recorder = recorder
 			newStatus, err := testCase.reconciler.syncControlFlowStage(
@@ -306,7 +306,7 @@ func TestSyncNormalStage(t *testing.T) {
 		reconciler *reconciler
 		assertions func(
 			t *testing.T,
-			recorder *fakekubeclient.EventRecorder,
+			recorder *fakeevent.EventRecorder,
 			initialStatus kargoapi.StageStatus,
 			newStatus kargoapi.StageStatus,
 			err error,
@@ -326,7 +326,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -355,7 +355,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -413,7 +413,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				_ kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -484,7 +484,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -533,7 +533,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -606,7 +606,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -680,7 +680,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -754,7 +754,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -846,7 +846,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				_ kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -951,7 +951,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				_ kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -994,7 +994,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1050,7 +1050,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1116,7 +1116,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1183,7 +1183,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1256,7 +1256,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1319,7 +1319,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1391,7 +1391,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				_ *fakekubeclient.EventRecorder,
+				_ *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1467,7 +1467,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				initialStatus kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1597,7 +1597,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				_ kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1736,7 +1736,7 @@ func TestSyncNormalStage(t *testing.T) {
 			},
 			assertions: func(
 				t *testing.T,
-				recorder *fakekubeclient.EventRecorder,
+				recorder *fakeevent.EventRecorder,
 				_ kargoapi.StageStatus,
 				newStatus kargoapi.StageStatus,
 				err error,
@@ -1782,7 +1782,7 @@ func TestSyncNormalStage(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			recorder := fakekubeclient.NewEventRecorder(2)
+			recorder := fakeevent.NewEventRecorder(2)
 			testCase.reconciler.nowFn = fakeNow
 			testCase.reconciler.recorder = recorder
 			newStatus, err := testCase.reconciler.syncNormalStage(
diff --git a/internal/kubernetes/event/event_test.go b/internal/kubernetes/event/event_test.go
new file mode 100644
index 00000000..365e0e62
--- /dev/null
+++ b/internal/kubernetes/event/event_test.go
@@ -0,0 +1,86 @@
+package event
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"testing"
+
+	testlog "github.com/sirupsen/logrus/hooks/test"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"github.com/akuity/kargo/internal/logging"
+)
+
+func Test_newRecorder(t *testing.T) {
+	ctx := context.TODO()
+	client := fake.NewClientBuilder().Build()
+	logger := logging.LoggerFromContext(ctx)
+	r := newRecorder(ctx, client, logger)
+
+	require.NotNil(t, r.backoff)
+	require.NotNil(t, r.sink)
+	require.NotNil(t, r.logger)
+	require.NotNil(t, r.newEventHandlerFn)
+}
+
+func Test_retryDecider(t *testing.T) {
+	eventGR := schema.GroupResource{
+		Group:    corev1.GroupName,
+		Resource: "Event",
+	}
+	testCases := map[string]struct {
+		input       error
+		shouldRetry bool
+	}{
+		"event already exists": {
+			input:       apierrors.NewAlreadyExists(eventGR, "fake-event"),
+			shouldRetry: false,
+		},
+		"namespace is terminating": {
+			input: &apierrors.StatusError{
+				ErrStatus: metav1.Status{
+					Code: http.StatusForbidden,
+					Details: &metav1.StatusDetails{
+						Causes: []metav1.StatusCause{
+							{
+								Type: corev1.NamespaceTerminatingCause,
+							},
+						},
+					},
+				},
+			},
+			shouldRetry: false,
+		},
+		"unknown error": {
+			input:       errors.New("unknown error"),
+			shouldRetry: true,
+		},
+	}
+	for name, tc := range testCases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			logger, _ := testlog.NewNullLogger()
+			r := &recorder{
+				logger: logger.WithFields(nil),
+			}
+			require.Equal(t, tc.shouldRetry, r.newRetryDecider(&corev1.Event{})(tc.input))
+		})
+	}
+}
+
+func Test_newSink(t *testing.T) {
+	s := newSink(
+		context.TODO(),
+		fake.NewClientBuilder().Build(),
+	)
+	require.NotNil(t, s.client)
+	require.NotNil(t, s.ctx)
+}
diff --git a/internal/webhook/freight/webhook_test.go b/internal/webhook/freight/webhook_test.go
index e652ebfb..80a16a35 100644
--- a/internal/webhook/freight/webhook_test.go
+++ b/internal/webhook/freight/webhook_test.go
@@ -20,7 +20,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 	libWebhook "github.com/akuity/kargo/internal/webhook"
 )
 
@@ -29,7 +29,7 @@ func TestNewWebhook(t *testing.T) {
 	w := newWebhook(
 		libWebhook.Config{},
 		kubeClient,
-		&fakekubeclient.EventRecorder{},
+		&fakeevent.EventRecorder{},
 	)
 	require.NotNil(t, w.freightAliasGenerator)
 	// Assert that all overridable behaviors were initialized to a default:
@@ -311,7 +311,7 @@ func TestValidateUpdate(t *testing.T) {
 		webhook    *webhook
 		userInfo   *authnv1.UserInfo
 		setup      func() (*kargoapi.Freight, *kargoapi.Freight)
-		assertions func(*testing.T, *fakekubeclient.EventRecorder, error)
+		assertions func(*testing.T, *fakeevent.EventRecorder, error)
 	}{
 		{
 			name: "error listing freight",
@@ -341,7 +341,7 @@ func TestValidateUpdate(t *testing.T) {
 					return errors.New("something went wrong")
 				},
 			},
-			assertions: func(t *testing.T, _ *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, _ *fakeevent.EventRecorder, err error) {
 				statusErr, ok := err.(*apierrors.StatusError)
 				require.True(t, ok)
 				require.Equal(
@@ -382,7 +382,7 @@ func TestValidateUpdate(t *testing.T) {
 					return nil
 				},
 			},
-			assertions: func(t *testing.T, _ *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, _ *fakeevent.EventRecorder, err error) {
 				statusErr, ok := err.(*apierrors.StatusError)
 				require.True(t, ok)
 				require.Equal(t, int32(http.StatusConflict), statusErr.Status().Code)
@@ -417,7 +417,7 @@ func TestValidateUpdate(t *testing.T) {
 					return nil
 				},
 			},
-			assertions: func(t *testing.T, _ *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, _ *fakeevent.EventRecorder, err error) {
 				require.Error(t, err)
 				require.Contains(t, err.Error(), "is invalid")
 				require.Contains(t, err.Error(), "freight is immutable")
@@ -447,7 +447,7 @@ func TestValidateUpdate(t *testing.T) {
 					return nil
 				},
 			},
-			assertions: func(t *testing.T, _ *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, _ *fakeevent.EventRecorder, err error) {
 				require.Error(t, err)
 				require.Contains(t, err.Error(), "is invalid")
 				require.Contains(t, err.Error(), "freight is immutable")
@@ -488,7 +488,7 @@ func TestValidateUpdate(t *testing.T) {
 			userInfo: &authnv1.UserInfo{
 				Username: "fake-user",
 			},
-			assertions: func(t *testing.T, r *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, r *fakeevent.EventRecorder, err error) {
 				require.NoError(t, err)
 				// Recorder should not record non-freight approval events
 				require.Empty(t, r.Events)
@@ -533,7 +533,7 @@ func TestValidateUpdate(t *testing.T) {
 			userInfo: &authnv1.UserInfo{
 				Username: "fake-user",
 			},
-			assertions: func(t *testing.T, r *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, r *fakeevent.EventRecorder, err error) {
 				require.NoError(t, err)
 				require.Len(t, r.Events, 1)
 				event := <-r.Events
@@ -579,7 +579,7 @@ func TestValidateUpdate(t *testing.T) {
 			userInfo: &authnv1.UserInfo{
 				Username: serviceaccount.ServiceAccountUsernamePrefix + "kargo:kargo-api",
 			},
-			assertions: func(t *testing.T, r *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, r *fakeevent.EventRecorder, err error) {
 				require.NoError(t, err)
 				require.Empty(t, r.Events)
 			},
@@ -589,7 +589,7 @@ func TestValidateUpdate(t *testing.T) {
 		t.Run(testCase.name, func(t *testing.T) {
 			oldFreight, newFreight := testCase.setup()
 
-			recorder := fakekubeclient.NewEventRecorder(1)
+			recorder := fakeevent.NewEventRecorder(1)
 			testCase.webhook.recorder = recorder
 
 			var req admission.Request
diff --git a/internal/webhook/promotion/webhook_test.go b/internal/webhook/promotion/webhook_test.go
index 02118c8e..33479792 100644
--- a/internal/webhook/promotion/webhook_test.go
+++ b/internal/webhook/promotion/webhook_test.go
@@ -18,7 +18,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
-	fakekubeclient "github.com/akuity/kargo/internal/kubeclient/fake"
+	fakeevent "github.com/akuity/kargo/internal/kubernetes/event/fake"
 	libWebhook "github.com/akuity/kargo/internal/webhook"
 )
 
@@ -27,7 +27,7 @@ func TestNewWebhook(t *testing.T) {
 	w := newWebhook(
 		libWebhook.Config{},
 		kubeClient,
-		&fakekubeclient.EventRecorder{},
+		&fakeevent.EventRecorder{},
 	)
 	// Assert that all overridable behaviors were initialized to a default:
 	require.NotNil(t, w.getFreightFn)
@@ -136,7 +136,7 @@ func TestValidateCreate(t *testing.T) {
 		name       string
 		webhook    *webhook
 		userInfo   *authnv1.UserInfo
-		assertions func(*testing.T, *fakekubeclient.EventRecorder, error)
+		assertions func(*testing.T, *fakeevent.EventRecorder, error)
 	}{
 		{
 			name: "error validating project",
@@ -150,7 +150,7 @@ func TestValidateCreate(t *testing.T) {
 					return errors.New("something went wrong")
 				},
 			},
-			assertions: func(t *testing.T, _ *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, _ *fakeevent.EventRecorder, err error) {
 				require.Error(t, err)
 				require.Equal(t, "something went wrong", err.Error())
 			},
@@ -170,7 +170,7 @@ func TestValidateCreate(t *testing.T) {
 					return errors.New("something went wrong")
 				},
 			},
-			assertions: func(t *testing.T, _ *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, _ *fakeevent.EventRecorder, err error) {
 				require.Error(t, err)
 				require.Equal(t, "something went wrong", err.Error())
 			},
@@ -204,7 +204,7 @@ func TestValidateCreate(t *testing.T) {
 			userInfo: &authnv1.UserInfo{
 				Username: "fake-user",
 			},
-			assertions: func(t *testing.T, r *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, r *fakeevent.EventRecorder, err error) {
 				require.NoError(t, err)
 				require.Len(t, r.Events, 1)
 				event := <-r.Events
@@ -233,7 +233,7 @@ func TestValidateCreate(t *testing.T) {
 			userInfo: &authnv1.UserInfo{
 				Username: serviceaccount.ServiceAccountUsernamePrefix + "kargo:kargo-api",
 			},
-			assertions: func(t *testing.T, r *fakekubeclient.EventRecorder, err error) {
+			assertions: func(t *testing.T, r *fakeevent.EventRecorder, err error) {
 				require.NoError(t, err)
 				require.Empty(t, r.Events)
 			},
@@ -241,7 +241,7 @@ func TestValidateCreate(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			recorder := fakekubeclient.NewEventRecorder(1)
+			recorder := fakeevent.NewEventRecorder(1)
 			testCase.webhook.recorder = recorder
 
 			var req admission.Request