feat: add Flink kind (#848)

Author: rriski

CHANGELOG.md

@@ -51,6 +51,7 @@
   minimum ~~`1`~~ → `0`
 - Change `Cassandra` field `userConfig.cassandra_version`: enum remove `4`
 - Change `PostgreSQL` field `userConfig.pg_version`: enum remove `12`
+- Add kind: `Flink`
 
 ## v0.25.0 - 2024-09-19
 

PROJECT

@@ -291,4 +291,17 @@ resources:
     defaulting: true
     validation: true
     webhookVersion: v1
+- api:
+    crdVersion: v1
+    namespaced: true
+  controller: true
+  domain: aiven.io
+  kind: Flink
+  path: github.com/aiven/aiven-operator/api/v1alpha1
+  version: v1alpha1
+  webhooks:
+    conversion: true
+    defaulting: true
+    validation: true
+    webhookVersion: v1
 version: "3"

api/v1alpha1/flink_types.go

@@ -0,0 +1,68 @@
+// Copyright (c) 2024 Aiven, Helsinki, Finland. https://aiven.io/
+
+package v1alpha1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	flinkuserconfig "github.com/aiven/aiven-operator/api/v1alpha1/userconfig/service/flink"
+)
+
+// FlinkSpec defines the desired state of Flink
+type FlinkSpec struct {
+	ServiceCommonSpec `json:",inline"`
+
+	// Cassandra specific user configuration options
+	UserConfig *flinkuserconfig.FlinkUserConfig `json:"userConfig,omitempty"`
+}
+
+// Flink is the Schema for the flinks API.
+// Info "Exposes secret keys": `FLINK_HOST`, `FLINK_PORT`, `FLINK_USER`, `FLINK_PASSWORD`, `FLINK_URI`, `FLINK_HOSTS`
+// +kubebuilder:object:root=true
+// +kubebuilder:subresource:status
+// +kubebuilder:printcolumn:name="Project",type="string",JSONPath=".spec.project"
+// +kubebuilder:printcolumn:name="Region",type="string",JSONPath=".spec.cloudName"
+// +kubebuilder:printcolumn:name="Plan",type="string",JSONPath=".spec.plan"
+// +kubebuilder:printcolumn:name="State",type="string",JSONPath=".status.state"
+type Flink struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   FlinkSpec     `json:"spec,omitempty"`
+	Status ServiceStatus `json:"status,omitempty"`
+}
+
+var _ AivenManagedObject = &Flink{}
+
+func (in *Flink) AuthSecretRef() *AuthSecretReference {
+	return in.Spec.AuthSecretRef
+}
+
+func (in *Flink) Conditions() *[]metav1.Condition {
+	return &in.Status.Conditions
+}
+
+func (in *Flink) NoSecret() bool {
+	return in.Spec.ConnInfoSecretTargetDisabled != nil && *in.Spec.ConnInfoSecretTargetDisabled
+}
+
+func (in *Flink) GetRefs() []*ResourceReferenceObject {
+	return in.Spec.GetRefs(in.GetNamespace())
+}
+
+func (in *Flink) GetConnInfoSecretTarget() ConnInfoSecretTarget {
+	return in.Spec.ConnInfoSecretTarget
+}
+
+//+kubebuilder:object:root=true
+
+// FlinkList contains a list of Flink
+type FlinkList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []Flink `json:"items"`
+}
+
+func init() {
+	SchemeBuilder.Register(&Flink{}, &FlinkList{})
+}

api/v1alpha1/flink_webhook.go

@@ -0,0 +1,61 @@
+// Copyright (c) 2024 Aiven, Helsinki, Finland. https://aiven.io/
+
+package v1alpha1
+
+import (
+	"errors"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+)
+
+// log is for logging in this package.
+var flinklog = logf.Log.WithName("flink-resource")
+
+func (in *Flink) SetupWebhookWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewWebhookManagedBy(mgr).
+		For(in).
+		Complete()
+}
+
+//+kubebuilder:webhook:path=/mutate-aiven-io-v1alpha1-flink,mutating=true,failurePolicy=fail,sideEffects=None,groups=aiven.io,resources=flinks,verbs=create;update,versions=v1alpha1,name=mflink.kb.io,admissionReviewVersions=v1
+
+var _ webhook.Defaulter = &Flink{}
+
+func (in *Flink) Default() {
+	flinklog.Info("default", "name", in.Name)
+}
+
+//+kubebuilder:webhook:verbs=create;update;delete,path=/validate-aiven-io-v1alpha1-flink,mutating=false,failurePolicy=fail,groups=aiven.io,resources=flinks,versions=v1alpha1,name=vflink.kb.io,sideEffects=none,admissionReviewVersions=v1
+
+var _ webhook.Validator = &Flink{}
+
+// ValidateCreate implements webhook.Validator so a webhook will be registered for the type
+func (in *Flink) ValidateCreate() error {
+	flinklog.Info("validate create", "name", in.Name)
+
+	return in.Spec.Validate()
+}
+
+// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
+func (in *Flink) ValidateUpdate(old runtime.Object) error {
+	flinklog.Info("validate update", "name", in.Name)
+	return in.Spec.Validate()
+}
+
+// ValidateDelete implements webhook.Validator so a webhook will be registered for the type
+func (in *Flink) ValidateDelete() error {
+	flinklog.Info("validate delete", "name", in.Name)
+
+	if in.Spec.TerminationProtection != nil && *in.Spec.TerminationProtection {
+		return errors.New("cannot delete Flink service, termination protection is on")
+	}
+
+	if in.Spec.ProjectVPCID != "" && in.Spec.ProjectVPCRef != nil {
+		return errors.New("cannot use both projectVpcId and projectVPCRef")
+	}
+
+	return nil
+}

api/v1alpha1/setup_webhooks.go

@@ -16,6 +16,9 @@ func SetupWebhooks(mgr ctrl.Manager) error {
 	if err := (&Database{}).SetupWebhookWithManager(mgr); err != nil {
 		return fmt.Errorf("webhook Database: %w", err)
 	}
+	if err := (&Flink{}).SetupWebhookWithManager(mgr); err != nil {
+		return fmt.Errorf("webhook Flink: %w", err)
+	}
 	if err := (&ConnectionPool{}).SetupWebhookWithManager(mgr); err != nil {
 		return fmt.Errorf("webhook ConnectionPool: %w", err)
 	}