Fix secret reference error handling

generate.go

@@ -4,3 +4,4 @@ package main
 //go:generate mockgen -destination=./tests/mocks/unmanaged_reconciler.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers UnmanagedReconciler
 //go:generate mockgen -destination=./tests/mocks/controller.go -package=mocks sigs.k8s.io/controller-runtime/pkg/controller Controller
 //go:generate mockgen -destination=./tests/mocks/job_builder.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers/stream JobBuilder
+//go:generate mockgen -destination=./tests/mocks/job_mock/secret_reference_provider.go -package=mocks github.com/SneaksAndData/arcane-operator/services/job SecretReferenceProvider

services/controllers/stream/stream_metadata_service.go

@@ -1,7 +1,6 @@
 package stream
 
 import (
-	"fmt"
 	v1 "github.com/SneaksAndData/arcane-operator/pkg/apis/streaming/v1"
 	"github.com/SneaksAndData/arcane-operator/services/job"
 )
@@ -16,12 +15,8 @@ type streamClassMetadataService struct {
 func (s streamClassMetadataService) JobConfigurator() (job.Configurator, error) {
 	builder := job.NewConfiguratorChainBuilder()
 
-	for _, secretName := range s.streamClass.Spec.SecretRefs {
-		name, err := s.streamDefinition.GetReferenceForSecret(secretName)
-		if err != nil {
-			return nil, fmt.Errorf("error getting secret reference: %w", err)
-		}
-		builder = builder.WithConfigurator(job.NewSecretReferenceConfigurator(name))
+	for _, referenceFieldName := range s.streamClass.Spec.SecretRefs {
+		builder = builder.WithConfigurator(job.NewSecretReferenceConfigurator(referenceFieldName, s.streamDefinition))
 	}
 
 	return builder.Build(), nil

services/controllers/stream/stream_metadata_service_test.go

@@ -124,39 +124,6 @@ func Test_StreamMetadataService_JobConfigurator_SingleSecretRef(t *testing.T) {
 	require.Equal(t, "databaseCredentials", job.Spec.Template.Spec.Containers[0].EnvFrom[0].SecretRef.Name)
 }
 
-func Test_StreamMetadataService_JobConfigurator_MissingSecretField(t *testing.T) {
-	// Arrange
-	streamClass := &v1.StreamClass{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: "test-stream-class",
-		},
-		Spec: v1.StreamClassSpec{
-			APIGroupRef: "streaming.sneaksanddata.com",
-			APIVersion:  "v1",
-			KindRef:     "MockStreamDefinition",
-			PluralName:  "mockstreamdefinitions",
-			SecretRefs:  []string{"nonExistentSecret"},
-		},
-	}
-
-	fakeClient := setupFakeClient(nil)
-	unstructuredObj, err := getUnstructured(t, fakeClient)
-	require.NoError(t, err)
-
-	streamDefinition, err := fromUnstructured(&unstructuredObj)
-	require.NoError(t, err)
-
-	service := NewStreamMetadataService(streamClass, streamDefinition)
-
-	// Act
-	configurator, err := service.JobConfigurator()
-
-	// Assert
-	require.Error(t, err)
-	require.Nil(t, configurator)
-	require.ErrorContains(t, err, "error getting secret reference")
-}
-
 func Test_StreamMetadataService_JobConfigurator_NilSecretRefs(t *testing.T) {
 	// Arrange
 	streamClass := &v1.StreamClass{
@@ -343,40 +310,6 @@ func Test_StreamMetadataService_JobConfigurator_PreservesExistingEnvFrom(t *test
 	require.Equal(t, "my-secret", job.Spec.Template.Spec.Containers[0].EnvFrom[1].SecretRef.Name)
 }
 
-func Test_StreamMetadataService_JobConfigurator_PartialFailure(t *testing.T) {
-	// Arrange
-	streamClass := &v1.StreamClass{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: "test-stream-class",
-		},
-		Spec: v1.StreamClassSpec{
-			APIGroupRef: "streaming.sneaksanddata.com",
-			APIVersion:  "v1",
-			KindRef:     "MockStreamDefinition",
-			PluralName:  "mockstreamdefinitions",
-			SecretRefs:  []string{"secretRef", "nonExistentSecret"},
-		},
-	}
-
-	fakeClient := setupFakeClientWithSecrets("databaseCredentials")
-	unstructuredObj, err := getUnstructured(t, fakeClient)
-	require.NoError(t, err)
-
-	streamDefinition, err := fromUnstructured(&unstructuredObj)
-	require.NoError(t, err)
-
-	service := NewStreamMetadataService(streamClass, streamDefinition)
-
-	// Act
-	configurator, err := service.JobConfigurator()
-
-	// Assert - should fail on the first missing secret
-	require.Error(t, err)
-	require.Nil(t, configurator)
-	require.ErrorContains(t, err, "error getting secret reference")
-	require.ErrorContains(t, err, "nonExistentSecret")
-}
-
 // setupFakeClientWithSecrets creates a fake client with a MockStreamDefinition that has secret references
 func setupFakeClientWithSecrets(secrets string) client.WithWatch {
 	sd := testv1.MockStreamDefinition{

services/controllers/stream/stream_reconciler.go

@@ -7,6 +7,7 @@ import (
 	"github.com/SneaksAndData/arcane-operator/services/controllers"
 	"github.com/SneaksAndData/arcane-operator/services/job"
 	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -457,6 +458,10 @@ func (s *streamReconciler) startNewJob(ctx context.Context, definition Definitio
 	j, err := s.jobBuilder.BuildJob(ctx, templateReference, combinedConfigurator)
 	if err != nil { // coverage-ignore
 		logger.V(0).Error(err, "failed to build job")
+		s.eventRecorder.Eventf(definition.ToUnstructured(),
+			corev1.EventTypeWarning,
+			"FailedCreateJob",
+			"failed to create job: %v", err)
 		return err
 	}
 

services/controllers/stream/unstructured_wrapper.go

@@ -15,8 +15,9 @@ import (
 )
 
 var (
-	_ Definition               = (*unstructuredWrapper)(nil)
-	_ job.ConfiguratorProvider = (*unstructuredWrapper)(nil)
+	_ Definition                  = (*unstructuredWrapper)(nil)
+	_ job.ConfiguratorProvider    = (*unstructuredWrapper)(nil)
+	_ job.SecretReferenceProvider = (*unstructuredWrapper)(nil)
 )
 
 type unstructuredWrapper struct {

services/job/secret_reference_configurator.go

@@ -9,14 +9,16 @@ import (
 var _ Configurator = &secretReferenceConfigurator{}
 
 type secretReferenceConfigurator struct {
-	reference *corev1.LocalObjectReference
+	referenceFieldName string
+	streamDefinition   SecretReferenceProvider
 }
 
 func (s secretReferenceConfigurator) ConfigureJob(job *batchv1.Job) error {
-	if s.reference == nil {
-		return fmt.Errorf("secretReferenceConfigurator reference is nil")
+	reference, err := s.streamDefinition.GetReferenceForSecret(s.referenceFieldName)
+	if err != nil {
+		return fmt.Errorf("error getting secret reference: %w", err)
 	}
-	if s.reference.Name == "" {
+	if reference.Name == "" {
 		return fmt.Errorf("secretReferenceConfigurator reference name is empty")
 	}
 	if len(job.Spec.Template.Spec.Containers) == 0 {
@@ -31,16 +33,17 @@ func (s secretReferenceConfigurator) ConfigureJob(job *batchv1.Job) error {
 
 		job.Spec.Template.Spec.Containers[i].EnvFrom = append(job.Spec.Template.Spec.Containers[i].EnvFrom, corev1.EnvFromSource{
 			SecretRef: &corev1.SecretEnvSource{
-				LocalObjectReference: *s.reference,
+				LocalObjectReference: *reference,
 			},
 		})
 	}
 
 	return nil
 }
 
-func NewSecretReferenceConfigurator(reference *corev1.LocalObjectReference) Configurator {
+func NewSecretReferenceConfigurator(referenceFieldName string, sd SecretReferenceProvider) Configurator {
 	return &secretReferenceConfigurator{
-		reference: reference,
+		referenceFieldName: referenceFieldName,
+		streamDefinition:   sd,
 	}
 }

services/job/secret_reference_configurator_test.go

@@ -1,6 +1,8 @@
 package job
 
 import (
+	"github.com/SneaksAndData/arcane-operator/tests/mocks/job_mock"
+	"go.uber.org/mock/gomock"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -24,7 +26,12 @@ func Test_SecretReferenceConfigurator_Add_SecretRef_To_Empty_Job(t *testing.T) {
 		},
 	}
 
-	configurator := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: "my-secret"})
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	mockProvider := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider.EXPECT().GetReferenceForSecret("test-secret").Return(&corev1.LocalObjectReference{Name: "my-secret"}, nil)
+
+	configurator := NewSecretReferenceConfigurator("test-secret", mockProvider)
 	err := configurator.ConfigureJob(job)
 	require.NoError(t, err)
 	require.NotNil(t, job.Spec.Template.Spec.Containers[0].EnvFrom)
@@ -58,7 +65,12 @@ func Test_SecretReferenceConfigurator_Append_To_Existing_EnvFrom(t *testing.T) {
 		},
 	}
 
-	configurator := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: "new-secret"})
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	mockProvider := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider.EXPECT().GetReferenceForSecret("test-secret").Return(&corev1.LocalObjectReference{Name: "new-secret"}, nil)
+
+	configurator := NewSecretReferenceConfigurator("test-secret", mockProvider)
 	err := configurator.ConfigureJob(job)
 	require.NoError(t, err)
 	require.Len(t, job.Spec.Template.Spec.Containers[0].EnvFrom, 2)
@@ -79,7 +91,12 @@ func Test_SecretReferenceConfigurator_No_Containers(t *testing.T) {
 		},
 	}
 
-	configurator := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: "my-secret"})
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	mockProvider := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider.EXPECT().GetReferenceForSecret("test-secret").Return(&corev1.LocalObjectReference{Name: "my-secret"}, nil)
+
+	configurator := NewSecretReferenceConfigurator("test-secret", mockProvider)
 	err := configurator.ConfigureJob(job)
 	require.NoError(t, err)
 	require.Empty(t, job.Spec.Template.Spec.Containers)
@@ -96,7 +113,12 @@ func Test_SecretReferenceConfigurator_Nil_Containers(t *testing.T) {
 		},
 	}
 
-	configurator := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: "my-secret"})
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	mockProvider := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider.EXPECT().GetReferenceForSecret("test-secret").Return(&corev1.LocalObjectReference{Name: "my-secret"}, nil)
+
+	configurator := NewSecretReferenceConfigurator("test-secret", mockProvider)
 	err := configurator.ConfigureJob(job)
 	require.NoError(t, err)
 	require.Nil(t, job.Spec.Template.Spec.Containers)
@@ -118,11 +140,18 @@ func Test_SecretReferenceConfigurator_Multiple_Secrets(t *testing.T) {
 		},
 	}
 
-	configurator1 := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: "first-secret"})
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	mockProvider1 := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider1.EXPECT().GetReferenceForSecret("secret1").Return(&corev1.LocalObjectReference{Name: "first-secret"}, nil)
+	configurator1 := NewSecretReferenceConfigurator("secret1", mockProvider1)
 	err := configurator1.ConfigureJob(job)
 	require.NoError(t, err)
 
-	configurator2 := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: "second-secret"})
+	mockProvider2 := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider2.EXPECT().GetReferenceForSecret("secret2").Return(&corev1.LocalObjectReference{Name: "second-secret"}, nil)
+	configurator2 := NewSecretReferenceConfigurator("secret2", mockProvider2)
 	err = configurator2.ConfigureJob(job)
 	require.NoError(t, err)
 
@@ -147,7 +176,12 @@ func Test_SecretReferenceConfigurator_Empty_Secret_Name(t *testing.T) {
 		},
 	}
 
-	configurator := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: ""})
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	mockProvider := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider.EXPECT().GetReferenceForSecret("test-secret").Return(&corev1.LocalObjectReference{Name: ""}, nil)
+
+	configurator := NewSecretReferenceConfigurator("test-secret", mockProvider)
 	err := configurator.ConfigureJob(job)
 	require.EqualError(t, err, "secretReferenceConfigurator reference name is empty")
 }
@@ -176,7 +210,12 @@ func Test_SecretReferenceConfigurator_Affects_All_Containers(t *testing.T) {
 		},
 	}
 
-	configurator := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: "my-secret"})
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	mockProvider := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider.EXPECT().GetReferenceForSecret("test-secret").Return(&corev1.LocalObjectReference{Name: "my-secret"}, nil)
+
+	configurator := NewSecretReferenceConfigurator("test-secret", mockProvider)
 	err := configurator.ConfigureJob(job)
 	require.NoError(t, err)
 
@@ -216,7 +255,12 @@ func Test_SecretReferenceConfigurator_With_Existing_SecretRef(t *testing.T) {
 		},
 	}
 
-	configurator := NewSecretReferenceConfigurator(&corev1.LocalObjectReference{Name: "new-secret"})
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	mockProvider := job_mock.NewMockSecretReferenceProvider(mockCtrl)
+	mockProvider.EXPECT().GetReferenceForSecret("test-secret").Return(&corev1.LocalObjectReference{Name: "new-secret"}, nil)
+
+	configurator := NewSecretReferenceConfigurator("test-secret", mockProvider)
 	err := configurator.ConfigureJob(job)
 	require.NoError(t, err)
 

services/job/secret_reference_provider.go

@@ -0,0 +1,10 @@
+package job
+
+import corev1 "k8s.io/api/core/v1"
+
+// SecretReferenceProvider defines an interface for types that can provide a reference to a Kubernetes Secret.
+type SecretReferenceProvider interface {
+
+	// GetReferenceForSecret retrieves a LocalObjectReference for the specified secret name.
+	GetReferenceForSecret(name string) (*corev1.LocalObjectReference, error)
+}