{cmd,}/k8s-operator: support IRSA for Recorder resources (#15913)

Adds Recorder fields to configure the name and annotations of the ServiceAccount
created for and used by its associated StatefulSet. This allows the created Pod
to authenticate with AWS without requiring a Secret with static credentials,
using AWS' IAM Roles for Service Accounts feature, documented here:
https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html

Fixes #15875

Change-Id: Ib0e15c0dbc357efa4be260e9ae5077bacdcb264f
Signed-off-by: Tom Proctor <tomhjp@users.noreply.github.com>

cmd/k8s-operator/deploy/crds/tailscale.com_recorders.yaml

@@ -1557,6 +1557,36 @@ spec:
                                     May also be set in PodSecurityContext. If set in both SecurityContext and
                                     PodSecurityContext, the value specified in SecurityContext takes precedence.
                                   type: string
+                        serviceAccount:
+                          description: |-
+                            Config for the ServiceAccount to create for the Recorder's StatefulSet.
+                            By default, the operator will create a ServiceAccount with the same
+                            name as the Recorder resource.
+                            https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#service-account
+                          type: object
+                          properties:
+                            annotations:
+                              description: |-
+                                Annotations to add to the ServiceAccount.
+                                https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/#syntax-and-character-set
+
+                                You can use this to add IAM roles to the ServiceAccount (IRSA) instead of
+                                providing static S3 credentials in a Secret.
+                                https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html
+
+                                For example:
+                                eks.amazonaws.com/role-arn: arn:aws:iam::<account-id>:role/<role-name>
+                              type: object
+                              additionalProperties:
+                                type: string
+                            name:
+                              description: |-
+                                Name of the ServiceAccount to create. Defaults to the name of the
+                                Recorder resource.
+                                https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#service-account
+                              type: string
+                              maxLength: 253
+                              pattern: ^[a-z0-9]([a-z0-9-.]{0,61}[a-z0-9])?$
                         tolerations:
                           description: |-
                             Tolerations for Recorder Pods. By default, the operator does not apply

cmd/k8s-operator/deploy/manifests/operator.yaml

@@ -4552,6 +4552,36 @@ spec:
                                                                 type: string
                                                         type: object
                                                 type: object
+                                            serviceAccount:
+                                                description: |-
+                                                    Config for the ServiceAccount to create for the Recorder's StatefulSet.
+                                                    By default, the operator will create a ServiceAccount with the same
+                                                    name as the Recorder resource.
+                                                    https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#service-account
+                                                properties:
+                                                    annotations:
+                                                        additionalProperties:
+                                                            type: string
+                                                        description: |-
+                                                            Annotations to add to the ServiceAccount.
+                                                            https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/#syntax-and-character-set
+
+                                                            You can use this to add IAM roles to the ServiceAccount (IRSA) instead of
+                                                            providing static S3 credentials in a Secret.
+                                                            https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html
+
+                                                            For example:
+                                                            eks.amazonaws.com/role-arn: arn:aws:iam::<account-id>:role/<role-name>
+                                                        type: object
+                                                    name:
+                                                        description: |-
+                                                            Name of the ServiceAccount to create. Defaults to the name of the
+                                                            Recorder resource.
+                                                            https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#service-account
+                                                        maxLength: 253
+                                                        pattern: ^[a-z0-9]([a-z0-9-.]{0,61}[a-z0-9])?$
+                                                        type: string
+                                                type: object
                                             tolerations:
                                                 description: |-
                                                     Tolerations for Recorder Pods. By default, the operator does not apply

cmd/k8s-operator/sts.go

@@ -1052,13 +1052,13 @@ func tailscaledConfigHash(c tailscaledConfigs) (string, error) {
 	return fmt.Sprintf("%x", h.Sum(nil)), nil
 }
 
-// createOrUpdate adds obj to the k8s cluster, unless the object already exists,
+// createOrMaybeUpdate adds obj to the k8s cluster, unless the object already exists,
-// in which case update is called to make changes to it. If update is nil, the
+// in which case update is called to make changes to it. If update is nil or returns
-// existing object is returned unmodified.
+// an error, the object is returned unmodified.
 //
 // obj is looked up by its Name and Namespace if Name is set, otherwise it's
 // looked up by labels.
-func createOrUpdate[T any, O ptrObject[T]](ctx context.Context, c client.Client, ns string, obj O, update func(O)) (O, error) {
+func createOrMaybeUpdate[T any, O ptrObject[T]](ctx context.Context, c client.Client, ns string, obj O, update func(O) error) (O, error) {
 	var (
 		existing O
 		err      error
@@ -1073,7 +1073,9 @@ func createOrUpdate[T any, O ptrObject[T]](ctx context.Context, c client.Client,
 	}
 	if err == nil && existing != nil {
 		if update != nil {
-			update(existing)
+			if err := update(existing); err != nil {
+				return nil, err
+			}
 			if err := c.Update(ctx, existing); err != nil {
 				return nil, err
 			}
@@ -1089,6 +1091,21 @@ func createOrUpdate[T any, O ptrObject[T]](ctx context.Context, c client.Client,
 	return obj, nil
 }
 
+// createOrUpdate adds obj to the k8s cluster, unless the object already exists,
+// in which case update is called to make changes to it. If update is nil, the
+// existing object is returned unmodified.
+//
+// obj is looked up by its Name and Namespace if Name is set, otherwise it's
+// looked up by labels.
+func createOrUpdate[T any, O ptrObject[T]](ctx context.Context, c client.Client, ns string, obj O, update func(O)) (O, error) {
+	return createOrMaybeUpdate(ctx, c, ns, obj, func(o O) error {
+		if update != nil {
+			update(o)
+		}
+		return nil
+	})
+}
+
 // getSingleObject searches for k8s objects of type T
 // (e.g. corev1.Service) with the given labels, and returns
 // it. Returns nil if no objects match the labels, and an error if

cmd/k8s-operator/tsrecorder.go

@@ -8,13 +8,13 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"slices"
 	"strings"
 	"sync"
 
-	"github.com/pkg/errors"
 	"go.uber.org/zap"
 	xslices "golang.org/x/exp/slices"
 	appsv1 "k8s.io/api/apps/v1"
@@ -22,8 +22,10 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	apivalidation "k8s.io/apimachinery/pkg/api/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -107,7 +109,7 @@ func (r *RecorderReconciler) Reconcile(ctx context.Context, req reconcile.Reques
 		if !apiequality.Semantic.DeepEqual(oldTSRStatus, &tsr.Status) {
 			// An error encountered here should get returned by the Reconcile function.
 			if updateErr := r.Client.Status().Update(ctx, tsr); updateErr != nil {
-				err = errors.Wrap(err, updateErr.Error())
+				err = errors.Join(err, updateErr)
 			}
 		}
 		return reconcile.Result{}, err
@@ -125,7 +127,7 @@ func (r *RecorderReconciler) Reconcile(ctx context.Context, req reconcile.Reques
 		}
 	}
 
-	if err := r.validate(tsr); err != nil {
+	if err := r.validate(ctx, tsr); err != nil {
 		message := fmt.Sprintf("Recorder is invalid: %s", err)
 		r.recorder.Eventf(tsr, corev1.EventTypeWarning, reasonRecorderInvalid, message)
 		return setStatusReady(tsr, metav1.ConditionFalse, reasonRecorderInvalid, message)
@@ -160,20 +162,26 @@ func (r *RecorderReconciler) maybeProvision(ctx context.Context, tsr *tsapi.Reco
 	if err := r.ensureAuthSecretCreated(ctx, tsr); err != nil {
 		return fmt.Errorf("error creating secrets: %w", err)
 	}
-	// State secret is precreated so we can use the Recorder CR as its owner ref.
+	// State Secret is precreated so we can use the Recorder CR as its owner ref.
 	sec := tsrStateSecret(tsr, r.tsNamespace)
 	if _, err := createOrUpdate(ctx, r.Client, r.tsNamespace, sec, func(s *corev1.Secret) {
 		s.ObjectMeta.Labels = sec.ObjectMeta.Labels
 		s.ObjectMeta.Annotations = sec.ObjectMeta.Annotations
-		s.ObjectMeta.OwnerReferences = sec.ObjectMeta.OwnerReferences
 	}); err != nil {
 		return fmt.Errorf("error creating state Secret: %w", err)
 	}
 	sa := tsrServiceAccount(tsr, r.tsNamespace)
-	if _, err := createOrUpdate(ctx, r.Client, r.tsNamespace, sa, func(s *corev1.ServiceAccount) {
+	if _, err := createOrMaybeUpdate(ctx, r.Client, r.tsNamespace, sa, func(s *corev1.ServiceAccount) error {
+		// Perform this check within the update function to make sure we don't
+		// have a race condition between the previous check and the update.
+		if err := saOwnedByRecorder(s, tsr); err != nil {
+			return err
+		}
+
 		s.ObjectMeta.Labels = sa.ObjectMeta.Labels
 		s.ObjectMeta.Annotations = sa.ObjectMeta.Annotations
-		s.ObjectMeta.OwnerReferences = sa.ObjectMeta.OwnerReferences
+
+		return nil
 	}); err != nil {
 		return fmt.Errorf("error creating ServiceAccount: %w", err)
 	}
@@ -181,7 +189,6 @@ func (r *RecorderReconciler) maybeProvision(ctx context.Context, tsr *tsapi.Reco
 	if _, err := createOrUpdate(ctx, r.Client, r.tsNamespace, role, func(r *rbacv1.Role) {
 		r.ObjectMeta.Labels = role.ObjectMeta.Labels
 		r.ObjectMeta.Annotations = role.ObjectMeta.Annotations
-		r.ObjectMeta.OwnerReferences = role.ObjectMeta.OwnerReferences
 		r.Rules = role.Rules
 	}); err != nil {
 		return fmt.Errorf("error creating Role: %w", err)
@@ -190,7 +197,6 @@ func (r *RecorderReconciler) maybeProvision(ctx context.Context, tsr *tsapi.Reco
 	if _, err := createOrUpdate(ctx, r.Client, r.tsNamespace, roleBinding, func(r *rbacv1.RoleBinding) {
 		r.ObjectMeta.Labels = roleBinding.ObjectMeta.Labels
 		r.ObjectMeta.Annotations = roleBinding.ObjectMeta.Annotations
-		r.ObjectMeta.OwnerReferences = roleBinding.ObjectMeta.OwnerReferences
 		r.RoleRef = roleBinding.RoleRef
 		r.Subjects = roleBinding.Subjects
 	}); err != nil {
@@ -200,12 +206,18 @@ func (r *RecorderReconciler) maybeProvision(ctx context.Context, tsr *tsapi.Reco
 	if _, err := createOrUpdate(ctx, r.Client, r.tsNamespace, ss, func(s *appsv1.StatefulSet) {
 		s.ObjectMeta.Labels = ss.ObjectMeta.Labels
 		s.ObjectMeta.Annotations = ss.ObjectMeta.Annotations
-		s.ObjectMeta.OwnerReferences = ss.ObjectMeta.OwnerReferences
 		s.Spec = ss.Spec
 	}); err != nil {
 		return fmt.Errorf("error creating StatefulSet: %w", err)
 	}
 
+	// ServiceAccount name may have changed, in which case we need to clean up
+	// the previous ServiceAccount. RoleBinding will already be updated to point
+	// to the new ServiceAccount.
+	if err := r.maybeCleanupServiceAccounts(ctx, tsr, sa.Name); err != nil {
+		return fmt.Errorf("error cleaning up ServiceAccounts: %w", err)
+	}
+
 	var devices []tsapi.RecorderTailnetDevice
 
 	device, ok, err := r.getDeviceInfo(ctx, tsr.Name)
@@ -224,6 +236,47 @@ func (r *RecorderReconciler) maybeProvision(ctx context.Context, tsr *tsapi.Reco
 	return nil
 }
 
+func saOwnedByRecorder(sa *corev1.ServiceAccount, tsr *tsapi.Recorder) error {
+	// If ServiceAccount name has been configured, check that we don't clobber
+	// a pre-existing SA not owned by this Recorder.
+	if sa.Name != tsr.Name && !apiequality.Semantic.DeepEqual(sa.OwnerReferences, tsrOwnerReference(tsr)) {
+		return fmt.Errorf("custom ServiceAccount name %q specified but conflicts with a pre-existing ServiceAccount in the %s namespace", sa.Name, sa.Namespace)
+	}
+
+	return nil
+}
+
+// maybeCleanupServiceAccounts deletes any dangling ServiceAccounts
+// owned by the Recorder if the ServiceAccount name has been changed.
+// They would eventually be cleaned up by owner reference deletion, but
+// this avoids a long-lived Recorder with many ServiceAccount name changes
+// accumulating a large amount of garbage.
+//
+// This is a no-op if the ServiceAccount name has not changed.
+func (r *RecorderReconciler) maybeCleanupServiceAccounts(ctx context.Context, tsr *tsapi.Recorder, currentName string) error {
+	logger := r.logger(tsr.Name)
+
+	// List all ServiceAccounts owned by this Recorder.
+	sas := &corev1.ServiceAccountList{}
+	if err := r.List(ctx, sas, client.InNamespace(r.tsNamespace), client.MatchingLabels(labels("recorder", tsr.Name, nil))); err != nil {
+		return fmt.Errorf("error listing ServiceAccounts for cleanup: %w", err)
+	}
+	for _, sa := range sas.Items {
+		if sa.Name == currentName {
+			continue
+		}
+		if err := r.Delete(ctx, &sa); err != nil {
+			if apierrors.IsNotFound(err) {
+				logger.Debugf("ServiceAccount %s not found, likely already deleted", sa.Name)
+			} else {
+				return fmt.Errorf("error deleting ServiceAccount %s: %w", sa.Name, err)
+			}
+		}
+	}
+
+	return nil
+}
+
 // maybeCleanup just deletes the device from the tailnet. All the kubernetes
 // resources linked to a Recorder will get cleaned up via owner references
 // (which we can use because they are all in the same namespace).
@@ -302,11 +355,41 @@ func (r *RecorderReconciler) ensureAuthSecretCreated(ctx context.Context, tsr *t
 	return nil
 }
 
-func (r *RecorderReconciler) validate(tsr *tsapi.Recorder) error {
+func (r *RecorderReconciler) validate(ctx context.Context, tsr *tsapi.Recorder) error {
 	if !tsr.Spec.EnableUI && tsr.Spec.Storage.S3 == nil {
 		return errors.New("must either enable UI or use S3 storage to ensure recordings are accessible")
 	}
 
+	// Check any custom ServiceAccount config doesn't conflict with pre-existing
+	// ServiceAccounts. This check is performed once during validation to ensure
+	// errors are raised early, but also again during any Updates to prevent a race.
+	specSA := tsr.Spec.StatefulSet.Pod.ServiceAccount
+	if specSA.Name != "" && specSA.Name != tsr.Name {
+		sa := &corev1.ServiceAccount{}
+		key := client.ObjectKey{
+			Name:      specSA.Name,
+			Namespace: r.tsNamespace,
+		}
+
+		err := r.Get(ctx, key, sa)
+		switch {
+		case apierrors.IsNotFound(err):
+			// ServiceAccount doesn't exist, so no conflict.
+		case err != nil:
+			return fmt.Errorf("error getting ServiceAccount %q for validation: %w", tsr.Spec.StatefulSet.Pod.ServiceAccount.Name, err)
+		default:
+			// ServiceAccount exists, check if it's owned by the Recorder.
+			if err := saOwnedByRecorder(sa, tsr); err != nil {
+				return err
+			}
+		}
+	}
+	if len(specSA.Annotations) > 0 {
+		if violations := apivalidation.ValidateAnnotations(specSA.Annotations, field.NewPath(".spec.statefulSet.pod.serviceAccount.annotations")); len(violations) > 0 {
+			return violations.ToAggregate()
+		}
+	}
+
 	return nil
 }
 

cmd/k8s-operator/tsrecorder_specs.go

@@ -39,7 +39,7 @@ func tsrStatefulSet(tsr *tsapi.Recorder, namespace string) *appsv1.StatefulSet {
 					Annotations: tsr.Spec.StatefulSet.Pod.Annotations,
 				},
 				Spec: corev1.PodSpec{
-					ServiceAccountName: tsr.Name,
+					ServiceAccountName: tsrServiceAccountName(tsr),
 					Affinity:           tsr.Spec.StatefulSet.Pod.Affinity,
 					SecurityContext:    tsr.Spec.StatefulSet.Pod.SecurityContext,
 					ImagePullSecrets:   tsr.Spec.StatefulSet.Pod.ImagePullSecrets,
@@ -100,14 +100,25 @@ func tsrStatefulSet(tsr *tsapi.Recorder, namespace string) *appsv1.StatefulSet {
 func tsrServiceAccount(tsr *tsapi.Recorder, namespace string) *corev1.ServiceAccount {
 	return &corev1.ServiceAccount{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:            tsr.Name,
+			Name:            tsrServiceAccountName(tsr),
 			Namespace:       namespace,
 			Labels:          labels("recorder", tsr.Name, nil),
 			OwnerReferences: tsrOwnerReference(tsr),
+			Annotations:     tsr.Spec.StatefulSet.Pod.ServiceAccount.Annotations,
 		},
 	}
 }
 
+func tsrServiceAccountName(tsr *tsapi.Recorder) string {
+	sa := tsr.Spec.StatefulSet.Pod.ServiceAccount
+	name := tsr.Name
+	if sa.Name != "" {
+		name = sa.Name
+	}
+
+	return name
+}
+
 func tsrRole(tsr *tsapi.Recorder, namespace string) *rbacv1.Role {
 	return &rbacv1.Role{
 		ObjectMeta: metav1.ObjectMeta{
@@ -154,7 +165,7 @@ func tsrRoleBinding(tsr *tsapi.Recorder, namespace string) *rbacv1.RoleBinding {
 		Subjects: []rbacv1.Subject{
 			{
 				Kind:      "ServiceAccount",
-				Name:      tsr.Name,
+				Name:      tsrServiceAccountName(tsr),
 				Namespace: namespace,
 			},
 		},

cmd/k8s-operator/tsrecorder_test.go

@@ -8,6 +8,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"strings"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -41,7 +42,7 @@ func TestRecorder(t *testing.T) {
 		Build()
 	tsClient := &fakeTSClient{}
 	zl, _ := zap.NewDevelopment()
-	fr := record.NewFakeRecorder(1)
+	fr := record.NewFakeRecorder(2)
 	cl := tstest.NewClock(tstest.ClockOpts{})
 	reconciler := &RecorderReconciler{
 		tsNamespace: tsNamespace,
@@ -52,7 +53,7 @@ func TestRecorder(t *testing.T) {
 		clock:       cl,
 	}
 
-	t.Run("invalid spec gives an error condition", func(t *testing.T) {
+	t.Run("invalid_spec_gives_an_error_condition", func(t *testing.T) {
 		expectReconciled(t, reconciler, "", tsr.Name)
 
 		msg := "Recorder is invalid: must either enable UI or use S3 storage to ensure recordings are accessible"
@@ -65,10 +66,66 @@ func TestRecorder(t *testing.T) {
 
 		expectedEvent := "Warning RecorderInvalid Recorder is invalid: must either enable UI or use S3 storage to ensure recordings are accessible"
 		expectEvents(t, fr, []string{expectedEvent})
+
+		tsr.Spec.EnableUI = true
+		tsr.Spec.StatefulSet.Pod.ServiceAccount.Annotations = map[string]string{
+			"invalid space characters": "test",
+		}
+		mustUpdate(t, fc, "", "test", func(t *tsapi.Recorder) {
+			t.Spec = tsr.Spec
+		})
+		expectReconciled(t, reconciler, "", tsr.Name)
+
+		// Only check part of this error message, because it's defined in an
+		// external package and may change.
+		if err := fc.Get(context.Background(), client.ObjectKey{
+			Name: tsr.Name,
+		}, tsr); err != nil {
+			t.Fatal(err)
+		}
+		if len(tsr.Status.Conditions) != 1 {
+			t.Fatalf("expected 1 condition, got %d", len(tsr.Status.Conditions))
+		}
+		cond := tsr.Status.Conditions[0]
+		if cond.Type != string(tsapi.RecorderReady) || cond.Status != metav1.ConditionFalse || cond.Reason != reasonRecorderInvalid {
+			t.Fatalf("expected condition RecorderReady false due to RecorderInvalid, got %v", cond)
+		}
+		for _, msg := range []string{cond.Message, <-fr.Events} {
+			if !strings.Contains(msg, `"invalid space characters"`) {
+				t.Fatalf("expected invalid annotation key in error message, got %q", cond.Message)
+			}
+		}
 	})
 
-	t.Run("observe Ready=true status condition for a valid spec", func(t *testing.T) {
+	t.Run("conflicting_service_account_config_marked_as_invalid", func(t *testing.T) {
-		tsr.Spec.EnableUI = true
+		mustCreate(t, fc, &corev1.ServiceAccount{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "pre-existing-sa",
+				Namespace: tsNamespace,
+			},
+		})
+
+		tsr.Spec.StatefulSet.Pod.ServiceAccount.Annotations = nil
+		tsr.Spec.StatefulSet.Pod.ServiceAccount.Name = "pre-existing-sa"
+		mustUpdate(t, fc, "", "test", func(t *tsapi.Recorder) {
+			t.Spec = tsr.Spec
+		})
+
+		expectReconciled(t, reconciler, "", tsr.Name)
+
+		msg := `Recorder is invalid: custom ServiceAccount name "pre-existing-sa" specified but conflicts with a pre-existing ServiceAccount in the tailscale namespace`
+		tsoperator.SetRecorderCondition(tsr, tsapi.RecorderReady, metav1.ConditionFalse, reasonRecorderInvalid, msg, 0, cl, zl.Sugar())
+		expectEqual(t, fc, tsr)
+		if expected := 0; reconciler.recorders.Len() != expected {
+			t.Fatalf("expected %d recorders, got %d", expected, reconciler.recorders.Len())
+		}
+
+		expectedEvent := "Warning RecorderInvalid " + msg
+		expectEvents(t, fr, []string{expectedEvent})
+	})
+
+	t.Run("observe_Ready_true_status_condition_for_a_valid_spec", func(t *testing.T) {
+		tsr.Spec.StatefulSet.Pod.ServiceAccount.Name = ""
 		mustUpdate(t, fc, "", "test", func(t *tsapi.Recorder) {
 			t.Spec = tsr.Spec
 		})
@@ -83,7 +140,42 @@ func TestRecorder(t *testing.T) {
 		expectRecorderResources(t, fc, tsr, true)
 	})
 
-	t.Run("populate node info in state secret, and see it appear in status", func(t *testing.T) {
+	t.Run("valid_service_account_config", func(t *testing.T) {
+		tsr.Spec.StatefulSet.Pod.ServiceAccount.Name = "test-sa"
+		tsr.Spec.StatefulSet.Pod.ServiceAccount.Annotations = map[string]string{
+			"test": "test",
+		}
+		mustUpdate(t, fc, "", "test", func(t *tsapi.Recorder) {
+			t.Spec = tsr.Spec
+		})
+
+		expectReconciled(t, reconciler, "", tsr.Name)
+
+		expectEqual(t, fc, tsr)
+		if expected := 1; reconciler.recorders.Len() != expected {
+			t.Fatalf("expected %d recorders, got %d", expected, reconciler.recorders.Len())
+		}
+		expectRecorderResources(t, fc, tsr, true)
+
+		// Get the service account and check the annotations.
+		sa := &corev1.ServiceAccount{}
+		if err := fc.Get(context.Background(), client.ObjectKey{
+			Name:      tsr.Spec.StatefulSet.Pod.ServiceAccount.Name,
+			Namespace: tsNamespace,
+		}, sa); err != nil {
+			t.Fatal(err)
+		}
+		if diff := cmp.Diff(sa.Annotations, tsr.Spec.StatefulSet.Pod.ServiceAccount.Annotations); diff != "" {
+			t.Fatalf("unexpected service account annotations (-got +want):\n%s", diff)
+		}
+		if sa.Name != tsr.Spec.StatefulSet.Pod.ServiceAccount.Name {
+			t.Fatalf("unexpected service account name: got %q, want %q", sa.Name, tsr.Spec.StatefulSet.Pod.ServiceAccount.Name)
+		}
+
+		expectMissing[corev1.ServiceAccount](t, fc, tsNamespace, tsr.Name)
+	})
+
+	t.Run("populate_node_info_in_state_secret_and_see_it_appear_in_status", func(t *testing.T) {
 		bytes, err := json.Marshal(map[string]any{
 			"Config": map[string]any{
 				"NodeID": "nodeid-123",
@@ -115,7 +207,7 @@ func TestRecorder(t *testing.T) {
 		expectEqual(t, fc, tsr)
 	})
 
-	t.Run("delete the Recorder and observe cleanup", func(t *testing.T) {
+	t.Run("delete_the_Recorder_and_observe_cleanup", func(t *testing.T) {
 		if err := fc.Delete(context.Background(), tsr); err != nil {
 			t.Fatal(err)
 		}

k8s-operator/api.md

@@ -726,6 +726,24 @@ _Appears in:_
 | `imagePullSecrets` _[LocalObjectReference](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.3/#localobjectreference-v1-core) array_ | Image pull Secrets for Recorder Pods.<br />https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#PodSpec |  |  |
 | `nodeSelector` _object (keys:string, values:string)_ | Node selector rules for Recorder Pods. By default, the operator does<br />not apply any node selector rules.<br />https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#scheduling |  |  |
 | `tolerations` _[Toleration](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.3/#toleration-v1-core) array_ | Tolerations for Recorder Pods. By default, the operator does not apply<br />any tolerations.<br />https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#scheduling |  |  |
+| `serviceAccount` _[RecorderServiceAccount](#recorderserviceaccount)_ | Config for the ServiceAccount to create for the Recorder's StatefulSet.<br />By default, the operator will create a ServiceAccount with the same<br />name as the Recorder resource.<br />https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#service-account |  |  |
+
+
+#### RecorderServiceAccount
+
+
+
+
+
+
+
+_Appears in:_
+- [RecorderPod](#recorderpod)
+
+| Field | Description | Default | Validation |
+| --- | --- | --- | --- |
+| `name` _string_ | Name of the ServiceAccount to create. Defaults to the name of the<br />Recorder resource.<br />https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#service-account |  | MaxLength: 253 <br />Pattern: `^[a-z0-9]([a-z0-9-.]{0,61}[a-z0-9])?$` <br />Type: string <br /> |
+| `annotations` _object (keys:string, values:string)_ | Annotations to add to the ServiceAccount.<br />https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/#syntax-and-character-set<br />You can use this to add IAM roles to the ServiceAccount (IRSA) instead of<br />providing static S3 credentials in a Secret.<br />https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html<br />For example:<br />eks.amazonaws.com/role-arn: arn:aws:iam::<account-id>:role/<role-name> |  |  |
 
 
 #### RecorderSpec

k8s-operator/apis/v1alpha1/types_recorder.go

@@ -142,6 +142,36 @@ type RecorderPod struct {
 	// https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#scheduling
 	// +optional
 	Tolerations []corev1.Toleration `json:"tolerations,omitempty"`
+
+	// Config for the ServiceAccount to create for the Recorder's StatefulSet.
+	// By default, the operator will create a ServiceAccount with the same
+	// name as the Recorder resource.
+	// https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#service-account
+	// +optional
+	ServiceAccount RecorderServiceAccount `json:"serviceAccount,omitempty"`
+}
+
+type RecorderServiceAccount struct {
+	// Name of the ServiceAccount to create. Defaults to the name of the
+	// Recorder resource.
+	// https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#service-account
+	// +kubebuilder:validation:Type=string
+	// +kubebuilder:validation:Pattern=`^[a-z0-9]([a-z0-9-.]{0,61}[a-z0-9])?$`
+	// +kubebuilder:validation:MaxLength=253
+	// +optional
+	Name string `json:"name,omitempty"`
+
+	// Annotations to add to the ServiceAccount.
+	// https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/#syntax-and-character-set
+	//
+	// You can use this to add IAM roles to the ServiceAccount (IRSA) instead of
+	// providing static S3 credentials in a Secret.
+	// https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html
+	//
+	// For example:
+	// eks.amazonaws.com/role-arn: arn:aws:iam::<account-id>:role/<role-name>
+	// +optional
+	Annotations map[string]string `json:"annotations,omitempty"`
 }
 
 type RecorderContainer struct {

k8s-operator/apis/v1alpha1/zz_generated.deepcopy.go

@@ -838,6 +838,7 @@ func (in *RecorderPod) DeepCopyInto(out *RecorderPod) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	in.ServiceAccount.DeepCopyInto(&out.ServiceAccount)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RecorderPod.
@@ -850,6 +851,28 @@ func (in *RecorderPod) DeepCopy() *RecorderPod {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *RecorderServiceAccount) DeepCopyInto(out *RecorderServiceAccount) {
+	*out = *in
+	if in.Annotations != nil {
+		in, out := &in.Annotations, &out.Annotations
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RecorderServiceAccount.
+func (in *RecorderServiceAccount) DeepCopy() *RecorderServiceAccount {
+	if in == nil {
+		return nil
+	}
+	out := new(RecorderServiceAccount)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *RecorderSpec) DeepCopyInto(out *RecorderSpec) {
 	*out = *in