Add support for customizing pod and containers

api/v1alpha/conduit_types.go

@@ -75,6 +75,8 @@ type ConduitSpec struct {
 	Registry   *SchemaRegistry     `json:"schemaRegistry,omitempty"`
 	Connectors []*ConduitConnector `json:"connectors,omitempty"`
 	Processors []*ConduitProcessor `json:"processors,omitempty"`
+
+	PodTemplate *ConduitPodTemplate `json:"podTemplate,omitempty"`
 }
 
 type ConduitConnector struct {
@@ -120,6 +122,14 @@ type SettingsVar struct {
 	ConfigMapRef *GlobalConfigMapRef       `json:"configMapRef,omitempty"`
 }
 
+// Settings will be applied to the conduit instance pod and containers.
+type ConduitPodTemplate struct {
+	ServiceAccountName       string                       `json:"serviceAccountName,omitempty"`
+	PodSecurityContext       *corev1.PodSecurityContext   `json:"podSecurityContext,omitempty"`
+	ContainerSecurityContext *corev1.SecurityContext      `json:"containerSecurityContext,omitempty"`
+	Resources                *corev1.ResourceRequirements `json:"resources,omitempty"`
+}
+
 // ConduitStatus defines the observed state of Conduit
 type ConduitStatus struct {
 	ObservedGeneration int64        `json:"observedGeneration,omitempty"`

config/samples/conduit-generator-with-pod-template.yaml

@@ -0,0 +1,39 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: test-account
+---
+apiVersion: operator.conduit.io/v1alpha
+kind: Conduit
+metadata:
+  name: conduit-generator
+spec:
+  podTemplate:
+    serviceAccountName: test-account
+#    podSecurityContext:
+#      runAsNonRoot: true
+    containerSecurityContext:
+      privileged: false
+  running: true
+  name: generator.log
+  description: generator pipeline
+  connectors:
+    - name: source-connector
+      type: source
+      plugin: builtin:generator
+      settings:
+        - name: format.type
+          value: structured
+        - name: format.options.id
+          value: "int"
+        - name: format.options.name
+          value: "string"
+        - name: format.options.company
+          value: "string"
+        - name: format.options.trial
+          value: "bool"
+        - name: recordCount
+          value: "3"
+    - name: destination-connector
+      type: destination
+      plugin: builtin:log

internal/controller/conduit_controller.go

@@ -412,6 +412,21 @@ func (r *ConduitReconciler) CreateOrUpdateDeployment(ctx context.Context, c *v1.
 			return err
 		}
 
+		// Assign template specs to the pod template
+		if c.Spec.PodTemplate != nil {
+			deployment.Spec.Template.Spec.ServiceAccountName = c.Spec.PodTemplate.ServiceAccountName
+			deployment.Spec.Template.Spec.SecurityContext = c.Spec.PodTemplate.PodSecurityContext
+			deployment.Spec.Template.Spec.Resources = c.Spec.PodTemplate.Resources
+
+			for i := range spec.Template.Spec.InitContainers {
+				deployment.Spec.Template.Spec.InitContainers[i].SecurityContext = c.Spec.PodTemplate.ContainerSecurityContext
+			}
+
+			for i := range spec.Template.Spec.Containers {
+				deployment.Spec.Template.Spec.Containers[i].SecurityContext = c.Spec.PodTemplate.ContainerSecurityContext
+			}
+		}
+
 		// Ensure labels and annotations are always exact.
 		deployment.Spec.Selector.MatchLabels = matchLabels
 		deployment.Spec.Template.ObjectMeta.Labels = labels

internal/webhook/v1alpha/conduit_webhook.go

@@ -27,8 +27,10 @@ import (
 	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/validation/field"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
@@ -54,7 +56,8 @@ func init() {
 func SetupConduitWebhookWithManager(ctx context.Context, mgr ctrl.Manager) error {
 	return ctrl.NewWebhookManagedBy(mgr).For(&v1alpha.Conduit{}).
 		WithValidator(&ConduitCustomValidator{
-			validation.NewValidator(ctx, mgr.GetClient(), log.Log.WithName("webhook-validation")),
+			c:                mgr.GetClient(),
+			ValidatorService: validation.NewValidator(ctx, mgr.GetClient(), log.Log.WithName("webhook-validation")),
 		}).
 		WithDefaulter(&ConduitCustomDefaulter{}).
 		Complete()
@@ -154,14 +157,11 @@ func (*ConduitCustomDefaulter) proccessorDefaulter(pp []*v1alpha.ConduitProcesso
 // when it is created, updated, or deleted.
 type ConduitCustomValidator struct {
 	validation.ValidatorService
+	c client.Client
 }
 
 var _ webhook.CustomValidator = &ConduitCustomValidator{}
 
-func NewConduitCustomValidator(validator validation.ValidatorService) *ConduitCustomValidator {
-	return &ConduitCustomValidator{validator}
-}
-
 // ValidateCreate implements webhook.CustomValidator so a webhook will be registered for the type Conduit.
 func (v *ConduitCustomValidator) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
 	conduit, ok := obj.(*v1alpha.Conduit)
@@ -190,6 +190,10 @@ func (v *ConduitCustomValidator) ValidateCreate(ctx context.Context, obj runtime
 		errs = append(errs, err)
 	}
 
+	if verrs := v.validatePodTemplate(ctx, conduit); len(verrs) > 0 {
+		errs = append(errs, verrs...)
+	}
+
 	if len(errs) > 0 {
 		return nil, apierrors.NewInvalid(v1alpha.GroupKind, conduit.Name, errs)
 	}
@@ -292,3 +296,45 @@ func (*ConduitCustomValidator) validateRegistry(sr *v1alpha.SchemaRegistry) *fie
 
 	return nil
 }
+
+func (v *ConduitCustomValidator) validatePodTemplate(ctx context.Context, conduit *v1alpha.Conduit) field.ErrorList {
+	if conduit.Spec.PodTemplate == nil {
+		return nil
+	}
+
+	errs := field.ErrorList{}
+	fp := field.NewPath("spec").Child("podTemplate")
+
+	if err := v.validateServiceAccount(
+		ctx,
+		conduit.Spec.PodTemplate.ServiceAccountName,
+		conduit.Namespace, fp,
+	); err != nil {
+		errs = append(errs, err)
+	}
+
+	if len(errs) > 0 {
+		return errs
+	}
+
+	return nil
+}
+
+func (v *ConduitCustomValidator) validateServiceAccount(ctx context.Context, sa, ns string, fp *field.Path) *field.Error {
+	if sa == "" {
+		return nil
+	}
+
+	serviceAccount := corev1.ServiceAccount{}
+	if err := v.c.Get(ctx, types.NamespacedName{
+		Name:      sa,
+		Namespace: ns,
+	}, &serviceAccount); err != nil {
+		if apierrors.IsNotFound(err) {
+			return field.Invalid(fp.Child("serviceAccount"), sa, "service account not found")
+		}
+		return field.Invalid(fp.Child("serviceAccount"), sa, err.Error())
+	}
+
+	return nil
+}

internal/webhook/v1alpha/conduit_webhook_test.go

@@ -40,7 +40,10 @@ func TestWebhookValidate_ConduitVersion(t *testing.T) {
 			is := is.New(t)
 			ctx := context.Background()
 			cl := fake.NewClientBuilder().Build()
-			v := &ConduitCustomValidator{conduit.NewValidator(ctx, cl, log.Log.WithName("webhook-validation"))}
+			v := &ConduitCustomValidator{
+				c:                cl,
+				ValidatorService: conduit.NewValidator(ctx, cl, log.Log.WithName("webhook-validation")),
+			}
 
 			fieldErr := v.validateConduitVersion(tc.ver)
 			if tc.expectedErr != nil {
@@ -113,7 +116,8 @@ func TestWebhook_ValidateCreate(t *testing.T) {
 			c := tc.setup()
 			cl := fake.NewClientBuilder().Build()
 			v := ConduitCustomValidator{
-				conduit.NewValidator(ctx, cl, log.Log.WithName("webhook-validation")),
+				c:                cl,
+				ValidatorService: conduit.NewValidator(ctx, cl, log.Log.WithName("webhook-validation")),
 			}
 
 			_, err := v.ValidateCreate(context.Background(), runtime.Object(c))
@@ -188,7 +192,8 @@ func TestWebhook_ValidateUpdate(t *testing.T) {
 			c := tc.setup()
 			cl := fake.NewClientBuilder().Build()
 			v := ConduitCustomValidator{
-				conduit.NewValidator(ctx, cl, log.Log.WithName("webhook-validation")),
+				c:                cl,
+				ValidatorService: conduit.NewValidator(ctx, cl, log.Log.WithName("webhook-validation")),
 			}
 
 			_, err := v.ValidateUpdate(context.Background(), runtime.Object(nil), runtime.Object(c))