Rollback to v1beta1 (#19)

Author: regadas

Makefile

@@ -38,7 +38,7 @@ run: generate fmt vet
 
 # Generate manifests e.g. CRD, RBAC etc.
 manifests: controller-gen
-	$(CONTROLLER_GEN) $(CRD_OPTIONS) rbac:roleName=manager-role webhook paths="./api/v1beta2/..." output:crd:artifacts:config=config/crd/bases
+	$(CONTROLLER_GEN) $(CRD_OPTIONS) rbac:roleName=manager-role webhook paths="./api/v1beta1/..." output:crd:artifacts:config=config/crd/bases
 	kubectl kustomize config/crd > config/crd/bases/patched_crd.yaml
 	mv config/crd/bases/patched_crd.yaml config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml
 	go mod tidy
@@ -53,7 +53,7 @@ vet:
 
 # Generate code
 generate: controller-gen
-	$(CONTROLLER_GEN) object:headerFile=./hack/boilerplate.go.txt paths=./api/v1beta2/...
+	$(CONTROLLER_GEN) object:headerFile=./hack/boilerplate.go.txt paths=./api/v1beta1/...
 
 # find or download controller-gen
 # download controller-gen if necessary

PROJECT

@@ -3,5 +3,5 @@ domain: k8s.io
 repo: github.com/spotify/flink-on-k8s-operator
 resources:
 - group: flinkoperator
-  version: v1beta2
+  version: v1beta1
   kind: FlinkCluster

README.md

@@ -72,7 +72,7 @@ The operator is still under active development, there is no Helm chart available
 
 ### API
 
-* [Custom Resource Definition (v1beta2)](docs/crd.md)
+* [Custom Resource Definition (v1beta1)](docs/crd.md)
 
 ### How to
 

api/v1beta1/flinkcluster_default.go

@@ -17,19 +17,25 @@ limitations under the License.
 package v1beta1
 
 import (
+	"github.com/hashicorp/go-version"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 )
 
+var (
+	v10, _ = version.NewVersion("1.10")
+)
+
 // 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
 	}
 	_SetImageDefault(&cluster.Spec.Image)
-	_SetJobManagerDefault(&cluster.Spec.JobManager)
-	_SetTaskManagerDefault(&cluster.Spec.TaskManager)
+	flinkVersion, _ := version.NewVersion(cluster.Spec.FlinkVersion)
+	_SetJobManagerDefault(&cluster.Spec.JobManager, flinkVersion)
+	_SetTaskManagerDefault(&cluster.Spec.TaskManager, flinkVersion)
 	_SetJobDefault(cluster.Spec.Job)
 	_SetHadoopConfigDefault(cluster.Spec.HadoopConfig)
 }
@@ -40,7 +46,7 @@ func _SetImageDefault(imageSpec *ImageSpec) {
 	}
 }
 
-func _SetJobManagerDefault(jmSpec *JobManagerSpec) {
+func _SetJobManagerDefault(jmSpec *JobManagerSpec, flinkVersion *version.Version) {
 	if jmSpec.Replicas == nil {
 		jmSpec.Replicas = new(int32)
 		*jmSpec.Replicas = 1
@@ -70,16 +76,24 @@ func _SetJobManagerDefault(jmSpec *JobManagerSpec) {
 		jmSpec.Ports.UI = new(int32)
 		*jmSpec.Ports.UI = 8081
 	}
-	if jmSpec.MemoryOffHeapMin.Format == "" {
-		jmSpec.MemoryOffHeapMin = *resource.NewScaledQuantity(600, 6) // 600MB
-	}
-	if jmSpec.MemoryOffHeapRatio == nil {
-		jmSpec.MemoryOffHeapRatio = new(int32)
-		*jmSpec.MemoryOffHeapRatio = 25
+
+	if flinkVersion == nil || flinkVersion.LessThan(v10) {
+		if jmSpec.MemoryOffHeapMin.Format == "" {
+			jmSpec.MemoryOffHeapMin = *resource.NewScaledQuantity(600, 6) // 600MB
+		}
+		if jmSpec.MemoryOffHeapRatio == nil {
+			jmSpec.MemoryOffHeapRatio = new(int32)
+			*jmSpec.MemoryOffHeapRatio = 25
+		}
+	} else {
+		if jmSpec.MemoryProcessRatio == nil {
+			jmSpec.MemoryProcessRatio = new(int32)
+			*jmSpec.MemoryProcessRatio = 80
+		}
 	}
 }
 
-func _SetTaskManagerDefault(tmSpec *TaskManagerSpec) {
+func _SetTaskManagerDefault(tmSpec *TaskManagerSpec, flinkVersion *version.Version) {
 	if tmSpec.Ports.Data == nil {
 		tmSpec.Ports.Data = new(int32)
 		*tmSpec.Ports.Data = 6121
@@ -92,12 +106,19 @@ func _SetTaskManagerDefault(tmSpec *TaskManagerSpec) {
 		tmSpec.Ports.Query = new(int32)
 		*tmSpec.Ports.Query = 6125
 	}
-	if tmSpec.MemoryOffHeapMin.Format == "" {
-		tmSpec.MemoryOffHeapMin = *resource.NewScaledQuantity(600, 6) // 600MB
-	}
-	if tmSpec.MemoryOffHeapRatio == nil {
-		tmSpec.MemoryOffHeapRatio = new(int32)
-		*tmSpec.MemoryOffHeapRatio = 25
+	if flinkVersion == nil || flinkVersion.LessThan(v10) {
+		if tmSpec.MemoryOffHeapMin.Format == "" {
+			tmSpec.MemoryOffHeapMin = *resource.NewScaledQuantity(600, 6) // 600MB
+		}
+		if tmSpec.MemoryOffHeapRatio == nil {
+			tmSpec.MemoryOffHeapRatio = new(int32)
+			*tmSpec.MemoryOffHeapRatio = 25
+		}
+	} else {
+		if tmSpec.MemoryProcessRatio == nil {
+			tmSpec.MemoryProcessRatio = new(int32)
+			*tmSpec.MemoryProcessRatio = 80
+		}
 	}
 }
 

api/v1beta2/flinkcluster_default_test.go renamed to api/v1beta1/flinkcluster_default_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package v1beta2
+package v1beta1
 
 import (
 	"testing"

api/v1beta1/flinkcluster_types.go

@@ -190,8 +190,6 @@ type JobManagerSpec struct {
 	// More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/
 	Resources corev1.ResourceRequirements `json:"resources,omitempty"`
 
-	// TODO: Memory calculation would be change. Let's watch the issue FLINK-13980.
-
 	// Percentage of off-heap memory in containers, as a safety margin to avoid OOM kill, default: 25
 	MemoryOffHeapRatio *int32 `json:"memoryOffHeapRatio,omitempty"`
 
@@ -200,6 +198,9 @@ type JobManagerSpec struct {
 	// More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#meaning-of-memory
 	MemoryOffHeapMin resource.Quantity `json:"memoryOffHeapMin,omitempty"`
 
+	// For Flink 1.10+. Percentage of memory process, as a safety margin to avoid OOM kill, default: 20
+	MemoryProcessRatio *int32 `json:"memoryProcessRatio,omitempty"`
+
 	// Volumes in the JobManager pod.
 	Volumes []corev1.Volume `json:"volumes,omitempty"`
 
@@ -273,6 +274,9 @@ type TaskManagerSpec struct {
 	// More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#meaning-of-memory
 	MemoryOffHeapMin resource.Quantity `json:"memoryOffHeapMin,omitempty"`
 
+	// For Flink 1.10+. Percentage of process memory, as a safety margin to avoid OOM kill, default: 20
+	MemoryProcessRatio *int32 `json:"memoryProcessRatio,omitempty"`
+
 	// Volumes in the TaskManager pods.
 	// More info: https://kubernetes.io/docs/concepts/storage/volumes/
 	Volumes []corev1.Volume `json:"volumes,omitempty"`
@@ -418,6 +422,9 @@ type JobSpec struct {
 
 // FlinkClusterSpec defines the desired state of FlinkCluster
 type FlinkClusterSpec struct {
+	// The version of Flink to be managed. This version must match the version in the image.
+	FlinkVersion string `json:"flinkVersion"`
+
 	// Flink image spec for the cluster's components.
 	Image ImageSpec `json:"image"`
 

api/v1beta2/flinkcluster_types_test.go renamed to api/v1beta1/flinkcluster_types_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package v1beta2
+package v1beta1
 
 import (
 	. "github.com/onsi/ginkgo"

api/v1beta1/flinkcluster_validate.go

@@ -22,6 +22,7 @@ import (
 	"reflect"
 	"strings"
 
+	"github.com/hashicorp/go-version"
 	"k8s.io/apimachinery/pkg/api/resource"
 
 	corev1 "k8s.io/api/core/v1"
@@ -47,6 +48,15 @@ func (v *Validator) ValidateCreate(cluster *FlinkCluster) error {
 	if err != nil {
 		return err
 	}
+
+	var flinkVersion *version.Version
+	if len(cluster.Spec.FlinkVersion) != 0 {
+		flinkVersion, err = version.NewVersion(cluster.Spec.FlinkVersion)
+		if err != nil {
+			return err
+		}
+	}
+
 	err = v.validateHadoopConfig(cluster.Spec.HadoopConfig)
 	if err != nil {
 		return err
@@ -59,11 +69,11 @@ func (v *Validator) ValidateCreate(cluster *FlinkCluster) error {
 	if err != nil {
 		return err
 	}
-	err = v.validateJobManager(&cluster.Spec.JobManager)
+	err = v.validateJobManager(flinkVersion, &cluster.Spec.JobManager)
 	if err != nil {
 		return err
 	}
-	err = v.validateTaskManager(&cluster.Spec.TaskManager)
+	err = v.validateTaskManager(flinkVersion, &cluster.Spec.TaskManager)
 	if err != nil {
 		return err
 	}
@@ -282,7 +292,7 @@ func (v *Validator) validateImage(imageSpec *ImageSpec) error {
 	return nil
 }
 
-func (v *Validator) validateJobManager(jmSpec *JobManagerSpec) error {
+func (v *Validator) validateJobManager(flinkVersion *version.Version, jmSpec *JobManagerSpec) error {
 	var err error
 
 	// Replicas.
@@ -330,22 +340,39 @@ func (v *Validator) validateJobManager(jmSpec *JobManagerSpec) error {
 		return err
 	}
 
-	// MemoryOffHeapRatio
-	err = v.validateMemoryOffHeapRatio(jmSpec.MemoryOffHeapRatio, "jobmanager")
-	if err != nil {
-		return err
-	}
+	if flinkVersion == nil || flinkVersion.LessThan(v10) {
+		if jmSpec.MemoryProcessRatio != nil {
+			return fmt.Errorf("MemoryProcessRatio config cannot be used with flinkVersion < 1.11', use " +
+				"memoryOffHeapRatio instead")
+		}
 
-	// MemoryOffHeapMin
-	err = v.validateMemoryOffHeapMin(&jmSpec.MemoryOffHeapMin, jmSpec.Resources.Limits.Memory(), "jobmanager")
-	if err != nil {
-		return err
+		// MemoryOffHeapRatio
+		err = v.validateRatio(jmSpec.MemoryOffHeapRatio, "jobmanager", "memoryOffHeapRatio")
+		if err != nil {
+			return err
+		}
+
+		// MemoryOffHeapMin
+		err = v.validateMemoryOffHeapMin(&jmSpec.MemoryOffHeapMin, jmSpec.Resources.Limits.Memory(), "jobmanager")
+		if err != nil {
+			return err
+		}
+	} else {
+		if jmSpec.MemoryOffHeapRatio != nil || !jmSpec.MemoryOffHeapMin.IsZero() {
+			return fmt.Errorf("MemoryOffHeapRatio or MemoryOffHeapMin config cannot be used with flinkVersion >= 1.11'; " +
+				"use memoryProcessRatio istead")
+		}
+		// MemoryProcessRatio
+		err = v.validateRatio(jmSpec.MemoryProcessRatio, "jobmanager", "memoryProcessRatio")
+		if err != nil {
+			return err
+		}
 	}
 
 	return nil
 }
 
-func (v *Validator) validateTaskManager(tmSpec *TaskManagerSpec) error {
+func (v *Validator) validateTaskManager(flinkVersion *version.Version, tmSpec *TaskManagerSpec) error {
 	// Replicas.
 	if tmSpec.Replicas < 1 {
 		return fmt.Errorf("invalid TaskManager replicas, it must >= 1")
@@ -376,16 +403,33 @@ func (v *Validator) validateTaskManager(tmSpec *TaskManagerSpec) error {
 		return err
 	}
 
-	// MemoryOffHeapRatio
-	err = v.validateMemoryOffHeapRatio(tmSpec.MemoryOffHeapRatio, "taskmanager")
-	if err != nil {
-		return err
-	}
+	if flinkVersion == nil || flinkVersion.LessThan(v10) {
+		if tmSpec.MemoryProcessRatio != nil {
+			return fmt.Errorf("MemoryProcessRatio config cannot be used with flinkVersion < 1.11', use " +
+				"memoryOffHeapRatio instead")
+		}
 
-	// MemoryOffHeapMin
-	err = v.validateMemoryOffHeapMin(&tmSpec.MemoryOffHeapMin, tmSpec.Resources.Limits.Memory(), "taskmanager")
-	if err != nil {
-		return err
+		// MemoryOffHeapRatio
+		err = v.validateRatio(tmSpec.MemoryOffHeapRatio, "taskmanager", "memoryOffHeapRatio")
+		if err != nil {
+			return err
+		}
+
+		// MemoryOffHeapMin
+		err = v.validateMemoryOffHeapMin(&tmSpec.MemoryOffHeapMin, tmSpec.Resources.Limits.Memory(), "taskmanager")
+		if err != nil {
+			return err
+		}
+	} else {
+		if tmSpec.MemoryOffHeapRatio != nil || !tmSpec.MemoryOffHeapMin.IsZero() {
+			return fmt.Errorf("MemoryOffHeapRatio or MemoryOffHeapMin config cannot be used with flinkVersion >= 1.11'; " +
+				"use memoryProcessRatio istead")
+		}
+		// MemoryProcessRatio
+		err = v.validateRatio(tmSpec.MemoryProcessRatio, "taskmanager", "memoryProcessRatio")
+		if err != nil {
+			return err
+		}
 	}
 
 	return nil
@@ -487,10 +531,9 @@ func (v *Validator) validateCleanupAction(
 	return nil
 }
 
-func (v *Validator) validateMemoryOffHeapRatio(
-	offHeapRatio *int32, component string) error {
-	if offHeapRatio == nil || *offHeapRatio > 100 || *offHeapRatio < 0 {
-		return fmt.Errorf("invalid %v memoryOffHeapRatio, it must be between 0 and 100", component)
+func (v *Validator) validateRatio(ratio *int32, component, property string) error {
+	if ratio == nil || *ratio > 100 || *ratio < 0 {
+		return fmt.Errorf("invalid %v %v, it must be between 0 and 100", component, property)
 	}
 	return nil
 }

api/v1beta2/flinkcluster_validate_test.go renamed to api/v1beta1/flinkcluster_validate_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package v1beta2
+package v1beta1
 
 import (
 	"encoding/json"
@@ -104,9 +104,7 @@ func TestInvalidImageSpec(t *testing.T) {
 			Name:      "mycluster",
 			Namespace: "default",
 		},
-		Spec: FlinkClusterSpec{
-			FlinkVersion: "1.8",
-		},
+		Spec: FlinkClusterSpec{},
 	}
 	var err = validator.ValidateCreate(&cluster)
 	var expectedErr = "image name is unspecified"
@@ -118,7 +116,6 @@ func TestInvalidImageSpec(t *testing.T) {
 			Namespace: "default",
 		},
 		Spec: FlinkClusterSpec{
-			FlinkVersion: "1.8",
 			Image: ImageSpec{
 				Name:       "flink:1.8.1",
 				PullPolicy: corev1.PullPolicy("XXX"),

api/v1beta2/suite_test.go renamed to api/v1beta1/suite_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package v1beta2
+package v1beta1
 
 import (
 	"path/filepath"
@@ -43,7 +43,7 @@ func TestAPIs(t *testing.T) {
 	RegisterFailHandler(Fail)
 
 	RunSpecsWithDefaultAndCustomReporters(t,
-		"v1beta2 Suite",
+		"v1beta1 Suite",
 		[]Reporter{printer.NewlineReporter{}})
 }