Introduce kubebuilder markers for type validation/defaulting

apis/flinkcluster/v1beta1/assets/test/flinkcluster_type_fetched.yaml

@@ -0,0 +1,53 @@
+metadata:
+  name: state-machine
+  namespace: default
+spec:
+  flinkVersion: "1.15"
+  image:
+    name: arm64v8/flink:1.15
+    pullPolicy: Always
+  jobManager:
+    replicas: 1
+    accessScope: Cluster
+    ports:
+      rpc: 6123
+      blob: 6124
+      query: 6125
+      ui: 8081
+    resources:
+      limits:
+        cpu: "2"
+        memory: 2Gi
+      requests:
+        cpu: 200m
+        memory: 512Mi
+    memoryOffHeapMin: "0"
+  taskManager:
+    deploymentType: StatefulSet
+    replicas: 3
+    ports:
+      data: 6121
+      rpc: 6122
+      query: 6125
+    resources:
+      limits:
+        cpu: "2"
+        memory: 2Gi
+      requests:
+        cpu: 200m
+        memory: 512Mi
+    memoryOffHeapMin: "0"
+  job:
+    jarFile: ./examples/streaming/StateMachineExample.jar
+    allowNonRestoredState: false
+    parallelism: 1
+    noLoggingToStdout: false
+    restartPolicy: Never
+    cleanupPolicy:
+      afterJobSucceeds: DeleteCluster
+      afterJobFails: KeepCluster
+      afterJobCancelled: DeleteCluster
+    mode: Detached
+  flinkProperties:
+    taskmanager.numberOfTaskSlots: "1"
+  recreateOnUpdate: true

apis/flinkcluster/v1beta1/assets/test/flinkcluster_type_test.yaml

@@ -0,0 +1,13 @@
+apiVersion: flinkoperator.k8s.io/v1beta1
+kind: FlinkCluster
+metadata:
+  namespace: default
+  name: state-machine
+spec:
+  flinkVersion: "1.15"
+  image:
+    name: arm64v8/flink:1.15
+  job:
+    jarFile: ./examples/streaming/StateMachineExample.jar
+  flinkProperties:
+    taskmanager.numberOfTaskSlots: "1"

apis/flinkcluster/v1beta1/flinkcluster_default.go

@@ -48,18 +48,12 @@ var (
 
 // Sets default values for unspecified FlinkCluster properties.
 func _SetDefault(cluster *FlinkCluster) {
-	if cluster.Spec.RecreateOnUpdate == nil {
-		cluster.Spec.RecreateOnUpdate = new(bool)
-		*cluster.Spec.RecreateOnUpdate = true
-	}
-
 	if cluster.Spec.BatchSchedulerName != nil {
 		cluster.Spec.BatchScheduler = &BatchSchedulerSpec{
 			Name: *cluster.Spec.BatchSchedulerName,
 		}
 	}
 
-	_SetImageDefault(&cluster.Spec.Image)
 	flinkVersion, _ := version.NewVersion(cluster.Spec.FlinkVersion)
 	if cluster.Spec.JobManager == nil {
 		cluster.Spec.JobManager = &JobManagerSpec{}
@@ -69,52 +63,13 @@ func _SetDefault(cluster *FlinkCluster) {
 		cluster.Spec.TaskManager = &TaskManagerSpec{}
 	}
 	_SetTaskManagerDefault(cluster.Spec.TaskManager, flinkVersion)
-	_SetJobDefault(cluster.Spec.Job)
-	_SetHadoopConfigDefault(cluster.Spec.HadoopConfig)
-
-}
-
-func _SetImageDefault(imageSpec *ImageSpec) {
-	if len(imageSpec.PullPolicy) == 0 {
-		imageSpec.PullPolicy = corev1.PullAlways
-	}
 }
 
 func _SetJobManagerDefault(jmSpec *JobManagerSpec, flinkVersion *version.Version) {
 	if jmSpec == nil {
 		return
 	}
 
-	if jmSpec.Replicas == nil {
-		jmSpec.Replicas = new(int32)
-		*jmSpec.Replicas = DefaultJobManagerReplicas
-	}
-	if len(jmSpec.AccessScope) == 0 {
-		jmSpec.AccessScope = AccessScopeCluster
-	}
-	if jmSpec.Ingress != nil {
-		if jmSpec.Ingress.UseTLS == nil {
-			jmSpec.Ingress.UseTLS = new(bool)
-			*jmSpec.Ingress.UseTLS = false
-		}
-	}
-	if jmSpec.Ports.RPC == nil {
-		jmSpec.Ports.RPC = new(int32)
-		*jmSpec.Ports.RPC = 6123
-	}
-	if jmSpec.Ports.Blob == nil {
-		jmSpec.Ports.Blob = new(int32)
-		*jmSpec.Ports.Blob = 6124
-	}
-	if jmSpec.Ports.Query == nil {
-		jmSpec.Ports.Query = new(int32)
-		*jmSpec.Ports.Query = 6125
-	}
-	if jmSpec.Ports.UI == nil {
-		jmSpec.Ports.UI = new(int32)
-		*jmSpec.Ports.UI = 8081
-	}
-
 	if flinkVersion == nil || flinkVersion.LessThan(v10) {
 		if jmSpec.MemoryOffHeapMin.Format == "" {
 			jmSpec.MemoryOffHeapMin = *resource.NewScaledQuantity(600, 6) // 600MB
@@ -129,9 +84,6 @@ func _SetJobManagerDefault(jmSpec *JobManagerSpec, flinkVersion *version.Version
 			*jmSpec.MemoryProcessRatio = 80
 		}
 	}
-	if jmSpec.Resources.Size() == 0 {
-		jmSpec.Resources = DefaultResources
-	}
 
 	var livenessProbe = corev1.Probe{
 		Handler: corev1.Handler{
@@ -170,22 +122,6 @@ func _SetTaskManagerDefault(tmSpec *TaskManagerSpec, flinkVersion *version.Versi
 	if tmSpec == nil {
 		return
 	}
-	if tmSpec.Replicas == nil {
-		tmSpec.Replicas = new(int32)
-		*tmSpec.Replicas = DefaultTaskManagerReplicas
-	}
-	if tmSpec.Ports.Data == nil {
-		tmSpec.Ports.Data = new(int32)
-		*tmSpec.Ports.Data = 6121
-	}
-	if tmSpec.Ports.RPC == nil {
-		tmSpec.Ports.RPC = new(int32)
-		*tmSpec.Ports.RPC = 6122
-	}
-	if tmSpec.Ports.Query == nil {
-		tmSpec.Ports.Query = new(int32)
-		*tmSpec.Ports.Query = 6125
-	}
 	if flinkVersion == nil || flinkVersion.LessThan(v10) {
 		if tmSpec.MemoryOffHeapMin.Format == "" {
 			tmSpec.MemoryOffHeapMin = *resource.NewScaledQuantity(600, 6) // 600MB
@@ -200,9 +136,6 @@ func _SetTaskManagerDefault(tmSpec *TaskManagerSpec, flinkVersion *version.Versi
 			*tmSpec.MemoryProcessRatio = 80
 		}
 	}
-	if tmSpec.Resources.Size() == 0 {
-		tmSpec.Resources = DefaultResources
-	}
 
 	var livenessProbe = corev1.Probe{
 		Handler: corev1.Handler{
@@ -235,49 +168,4 @@ func _SetTaskManagerDefault(tmSpec *TaskManagerSpec, flinkVersion *version.Versi
 		mergo.Merge(&readinessProbe, tmSpec.ReadinessProbe, mergo.WithOverride)
 	}
 	tmSpec.ReadinessProbe = &readinessProbe
-
-	if tmSpec.DeploymentType == "" {
-		tmSpec.DeploymentType = DeploymentTypeStatefulSet
-	}
-}
-
-func _SetJobDefault(jobSpec *JobSpec) {
-	if jobSpec == nil {
-		return
-	}
-	if jobSpec.AllowNonRestoredState == nil {
-		jobSpec.AllowNonRestoredState = new(bool)
-		*jobSpec.AllowNonRestoredState = false
-	}
-	if jobSpec.NoLoggingToStdout == nil {
-		jobSpec.NoLoggingToStdout = new(bool)
-		*jobSpec.NoLoggingToStdout = false
-	}
-	if jobSpec.RestartPolicy == nil {
-		jobSpec.RestartPolicy = new(JobRestartPolicy)
-		*jobSpec.RestartPolicy = JobRestartPolicyNever
-	}
-	if jobSpec.CleanupPolicy == nil {
-		jobSpec.CleanupPolicy = &CleanupPolicy{
-			AfterJobSucceeds:  CleanupActionDeleteCluster,
-			AfterJobFails:     CleanupActionKeepCluster,
-			AfterJobCancelled: CleanupActionDeleteCluster,
-		}
-	}
-	if jobSpec.Mode == nil {
-		jobSpec.Mode = new(JobMode)
-		*jobSpec.Mode = JobModeDetached
-	}
-	if jobSpec.Resources.Size() == 0 {
-		jobSpec.Resources = DefaultResources
-	}
-}
-
-func _SetHadoopConfigDefault(hadoopConfig *HadoopConfig) {
-	if hadoopConfig == nil {
-		return
-	}
-	if len(hadoopConfig.MountPath) == 0 {
-		hadoopConfig.MountPath = "/etc/hadoop/conf"
-	}
 }