issue-514, added resize settings to the resources specs, added resize…

… operations to its statuses

Makefile

@@ -98,7 +98,7 @@ run: manifests generate fmt vet ## Run a controller from your host.
 	go run ./main.go
 
 .PHONY: docker-build
-docker-build: manifests generate test ## Build docker image with the manager.
+docker-build: manifests generate #test ## Build docker image with the manager.
 	docker build -t ${IMG} .
 
 .PHONY: docker-push

apis/clusters/v1beta1/cadence_types.go

@@ -73,6 +73,7 @@ type CadenceSpec struct {
 	SharedProvisioning   []*SharedProvisioning   `json:"sharedProvisioning,omitempty"`
 	PackagedProvisioning []*PackagedProvisioning `json:"packagedProvisioning,omitempty"`
 	TargetPrimaryCadence []*TargetCadence        `json:"targetPrimaryCadence,omitempty"`
+	ResizeSettings       []*ResizeSettings       `json:"resizeSettings,omitempty"`
 }
 
 type AWSArchival struct {
@@ -190,6 +191,7 @@ func (cs *CadenceSpec) ToInstAPI(ctx context.Context, k8sClient client.Client) (
 		SharedProvisioning:    sharedProvisioning,
 		StandardProvisioning:  standardProvisioning,
 		TargetPrimaryCadence:  cs.TargetCadenceToInstAPI(),
+		ResizeSettings:        resizeSettingsToInstAPI(cs.ResizeSettings),
 	}, nil
 }
 
@@ -352,6 +354,7 @@ func (c *Cadence) FromInstAPI(iData []byte) (*Cadence, error) {
 
 func (cs *CadenceSpec) FromInstAPI(iCad *models.CadenceCluster) (spec CadenceSpec) {
 	spec.DataCentres = cs.DCsFromInstAPI(iCad.DataCentres)
+	spec.ResizeSettings = resizeSettingsFromInstAPI(iCad.ResizeSettings)
 
 	return
 }

apis/clusters/v1beta1/cadence_webhook.go

@@ -184,6 +184,13 @@ func (cv *cadenceValidator) ValidateCreate(ctx context.Context, obj runtime.Obje
 		}
 	}
 
+	for _, rs := range c.Spec.ResizeSettings {
+		err := validateSingleConcurrentResize(rs.Concurrency)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -211,6 +218,13 @@ func (cv *cadenceValidator) ValidateUpdate(ctx context.Context, old runtime.Obje
 		return fmt.Errorf("cannot update immutable fields: %v", err)
 	}
 
+	for _, rs := range c.Spec.ResizeSettings {
+		err := validateSingleConcurrentResize(rs.Concurrency)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

apis/clusters/v1beta1/cassandra_types.go

@@ -70,6 +70,8 @@ type CassandraSpec struct {
 	Spark               []*Spark               `json:"spark,omitempty"`
 	BundledUseOnly      bool                   `json:"bundledUseOnly,omitempty"`
 	UserRefs            []*UserReference       `json:"userRefs,omitempty"`
+	//+kubebuilder:validate:MaxItems:=1
+	ResizeSettings []*ResizeSettings `json:"resizeSettings,omitempty"`
 }
 
 // CassandraStatus defines the observed state of Cassandra
@@ -181,9 +183,10 @@ func (cs *CassandraSpec) HasRestore() bool {
 	return false
 }
 
-func (cs *CassandraSpec) NewDCsUpdate() models.CassandraClusterAPIUpdate {
+func (cs *CassandraSpec) ToClusterUpdate() models.CassandraClusterAPIUpdate {
 	return models.CassandraClusterAPIUpdate{
-		DataCentres: cs.DCsToInstAPI(),
+		DataCentres:    cs.DCsToInstAPI(),
+		ResizeSettings: resizeSettingsToInstAPI(cs.ResizeSettings),
 	}
 }
 
@@ -218,6 +221,13 @@ func (cs *CassandraSpec) validateUpdate(oldSpec CassandraSpec) error {
 		return err
 	}
 
+	for _, dc := range cs.DataCentres {
+		err = cs.validateResizeSettings(dc.NodesNumber)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -302,6 +312,7 @@ func (cs *CassandraSpec) FromInstAPI(iCass *models.CassandraCluster) CassandraSp
 		PasswordAndUserAuth: iCass.PasswordAndUserAuth,
 		Spark:               cs.SparkFromInstAPI(iCass.Spark),
 		BundledUseOnly:      iCass.BundledUseOnly,
+		ResizeSettings:      resizeSettingsFromInstAPI(iCass.ResizeSettings),
 	}
 }
 
@@ -347,6 +358,7 @@ func (cs *CassandraSpec) ToInstAPI() *models.CassandraCluster {
 		PCIComplianceMode:     cs.PCICompliance,
 		TwoFactorDelete:       cs.TwoFactorDeletesToInstAPI(),
 		BundledUseOnly:        cs.BundledUseOnly,
+		ResizeSettings:        resizeSettingsToInstAPI(cs.ResizeSettings),
 	}
 }
 
@@ -452,6 +464,16 @@ func (cdc *CassandraDataCentre) newImmutableFields() *immutableCassandraDCFields
 	}
 }
 
+func (c *CassandraSpec) validateResizeSettings(nodeNumber int) error {
+	for _, rs := range c.ResizeSettings {
+		if rs.Concurrency > nodeNumber {
+			return fmt.Errorf("resizeSettings.concurrency cannot be greater than number of nodes: %v", nodeNumber)
+		}
+	}
+
+	return nil
+}
+
 func init() {
 	SchemeBuilder.Register(&Cassandra{}, &CassandraList{})
 }

apis/clusters/v1beta1/cassandra_webhook.go

@@ -132,6 +132,10 @@ func (cv *cassandraValidator) ValidateCreate(ctx context.Context, obj runtime.Ob
 			return fmt.Errorf("number of nodes must be a multiple of replication factor: %v", dc.ReplicationFactor)
 		}
 
+		err = c.Spec.validateResizeSettings(dc.NodesNumber)
+		if err != nil {
+			return err
+		}
 	}
 
 	return nil

apis/clusters/v1beta1/kafka_types.go

@@ -78,6 +78,8 @@ type KafkaSpec struct {
 	//+kubebuilder:validation:MinItems:=1
 	//+kubebuilder:validation:MaxItems:=1
 	DataCentres []*KafkaDataCentre `json:"dataCentres"`
+	//+kubebuilder:validation:MaxItems:=1
+	ResizeSettings []*ResizeSettings `json:"resizeSettings,omitempty"`
 
 	// Provision additional dedicated nodes for Apache Zookeeper to run on.
 	// Zookeeper nodes will be co-located with Kafka if this is not provided
@@ -161,6 +163,7 @@ func (k *KafkaSpec) ToInstAPI() *models.KafkaCluster {
 		Kraft:                     k.kraftToInstAPI(),
 		KarapaceRestProxy:         k.karapaceRestProxyToInstAPI(),
 		KarapaceSchemaRegistry:    k.karapaceSchemaRegistryToInstAPI(),
+		ResizeSettings:            resizeSettingsToInstAPI(k.ResizeSettings),
 	}
 }
 
@@ -254,6 +257,7 @@ func (k *KafkaSpec) ToInstAPIUpdate() *models.KafkaInstAPIUpdateRequest {
 	return &models.KafkaInstAPIUpdateRequest{
 		DataCentre:         k.dcToInstAPI(),
 		DedicatedZookeeper: k.dedicatedZookeeperToInstAPIUpdate(),
+		ResizeSettings:     resizeSettingsToInstAPI(k.ResizeSettings),
 	}
 }
 
@@ -306,6 +310,7 @@ func (ks *KafkaSpec) FromInstAPI(iKafka *models.KafkaCluster) KafkaSpec {
 		Kraft:                     ks.kraftFromInstAPI(iKafka.Kraft),
 		KarapaceSchemaRegistry:    ks.KarapaceSchemaRegistryFromInstAPI(iKafka.KarapaceSchemaRegistry),
 		BundledUseOnly:            iKafka.BundledUseOnly,
+		ResizeSettings:            resizeSettingsFromInstAPI(iKafka.ResizeSettings),
 	}
 }
 

apis/clusters/v1beta1/kafka_webhook.go

@@ -140,6 +140,13 @@ func (kv *kafkaValidator) ValidateCreate(ctx context.Context, obj runtime.Object
 		}
 	}
 
+	for _, rs := range k.Spec.ResizeSettings {
+		err := validateSingleConcurrentResize(rs.Concurrency)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -171,6 +178,13 @@ func (kv *kafkaValidator) ValidateUpdate(ctx context.Context, old runtime.Object
 		return fmt.Errorf("cannot update, error: %v", err)
 	}
 
+	for _, rs := range k.Spec.ResizeSettings {
+		err := validateSingleConcurrentResize(rs.Concurrency)
+		if err != nil {
+			return err
+		}
+	}
+
 	if k.Status.ID == "" {
 		return kv.ValidateCreate(ctx, k)
 	}

apis/clusters/v1beta1/opensearch_types.go

@@ -18,6 +18,7 @@ package v1beta1
 
 import (
 	"encoding/json"
+	"fmt"
 	"strconv"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -53,6 +54,8 @@ type OpenSearchSpec struct {
 	AlertingPlugin           bool                    `json:"alertingPlugin,omitempty"`
 	BundledUseOnly           bool                    `json:"bundledUseOnly,omitempty"`
 	UserRefs                 []*UserReference        `json:"userRefs,omitempty"`
+	//+kubuilder:validation:MaxItems:=1
+	ResizeSettings []*ResizeSettings `json:"resizeSettings,omitempty"`
 }
 
 type OpenSearchDataCentre struct {
@@ -108,6 +111,7 @@ func (oss *OpenSearchSpec) ToInstAPI() *models.OpenSearchCluster {
 		IndexManagementPlugin:    oss.IndexManagementPlugin,
 		SLATier:                  oss.SLATier,
 		AlertingPlugin:           oss.AlertingPlugin,
+		ResizeSettings:           resizeSettingsToInstAPI(oss.ResizeSettings),
 	}
 }
 
@@ -246,6 +250,7 @@ func (oss *OpenSearchSpec) FromInstAPI(iOpenSearch *models.OpenSearchCluster) Op
 		IndexManagementPlugin:    oss.IndexManagementPlugin,
 		AlertingPlugin:           oss.AlertingPlugin,
 		BundledUseOnly:           oss.BundledUseOnly,
+		ResizeSettings:           resizeSettingsFromInstAPI(iOpenSearch.ResizeSettings),
 	}
 }
 
@@ -451,6 +456,7 @@ func (oss *OpenSearchSpec) ToInstAPIUpdate() models.OpenSearchInstAPIUpdateReque
 		DataNodes:            oss.dataNodesToInstAPI(),
 		OpenSearchDashboards: oss.dashboardsToInstAPI(),
 		ClusterManagerNodes:  oss.clusterManagerNodesToInstAPI(),
+		ResizeSettings:       resizeSettingsToInstAPI(oss.ResizeSettings),
 	}
 }
 
@@ -545,6 +551,16 @@ func (oss *OpenSearchSpec) HasRestore() bool {
 	return false
 }
 
+func (oss *OpenSearchSpec) validateResizeSettings(nodesNumber int) error {
+	for _, rs := range oss.ResizeSettings {
+		if rs.Concurrency > nodesNumber {
+			return fmt.Errorf("resizeSettings.concurrency cannot be greater than number of nodes: %v", nodesNumber)
+		}
+	}
+
+	return nil
+}
+
 func init() {
 	SchemeBuilder.Register(&OpenSearch{}, &OpenSearchList{})
 }

apis/clusters/v1beta1/opensearch_webhook.go

@@ -134,6 +134,18 @@ func (osv *openSearchValidator) ValidateCreate(ctx context.Context, obj runtime.
 		return err
 	}
 
+	err = os.Spec.validateResizeSettings(len(os.Spec.ClusterManagerNodes))
+	if err != nil {
+		return err
+	}
+
+	for _, node := range os.Spec.DataNodes {
+		err = os.Spec.validateResizeSettings(node.NodesNumber)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

apis/clusters/v1beta1/postgresql_types.go

@@ -89,6 +89,8 @@ type PgSpec struct {
 	ClusterConfigurations map[string]string `json:"clusterConfigurations,omitempty"`
 	Description           string            `json:"description,omitempty"`
 	SynchronousModeStrict bool              `json:"synchronousModeStrict,omitempty"`
+	//+kubebuilder:validate:MaxItems:=1
+	ResizeSettings []*ResizeSettings `json:"resizeSettings,omitempty"`
 }
 
 // PgStatus defines the observed state of PostgreSQL
@@ -196,6 +198,13 @@ func (pgs *PgSpec) DCsToInstAPI() (iDCs []*models.PGDataCentre) {
 	return
 }
 
+func (pgs *PgSpec) ToClusterUpdate() *models.PGClusterUpdate {
+	return &models.PGClusterUpdate{
+		DataCentres:    pgs.DCsToInstAPI(),
+		ResizeSettings: resizeSettingsToInstAPI(pgs.ResizeSettings),
+	}
+}
+
 func (pdc *PgDataCentre) ToInstAPI() *models.PGDataCentre {
 	return &models.PGDataCentre{
 		DataCentre:                 pdc.DataCentre.ToInstAPI(),

apis/clusters/v1beta1/postgresql_webhook.go

@@ -128,6 +128,13 @@ func (pgv *pgValidator) ValidateCreate(ctx context.Context, obj runtime.Object)
 		}
 	}
 
+	for _, rs := range pg.Spec.ResizeSettings {
+		err := validateSingleConcurrentResize(rs.Concurrency)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -163,6 +170,13 @@ func (pgv *pgValidator) ValidateUpdate(ctx context.Context, old runtime.Object,
 		return fmt.Errorf("immutable fields validation error: %v", err)
 	}
 
+	for _, rs := range pg.Spec.ResizeSettings {
+		err := validateSingleConcurrentResize(rs.Concurrency)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

apis/clusters/v1beta1/redis_types.go

@@ -80,6 +80,8 @@ type RedisSpec struct {
 	DataCentres []*RedisDataCentre `json:"dataCentres,omitempty"`
 	Description string             `json:"description,omitempty"`
 	UserRefs    []*UserReference   `json:"userRefs,omitempty"`
+	//+kubebuilder:validation:MaxItems:=1
+	ResizeSettings []*ResizeSettings `json:"resizeSettings,omitempty"`
 }
 
 // RedisStatus defines the observed state of Redis
@@ -192,9 +194,10 @@ func (rs *RedisSpec) DCsToInstAPI() (iDCs []*models.RedisDataCentre) {
 	return
 }
 
-func (rs *RedisSpec) DCsToInstAPIUpdate() *models.RedisDataCentreUpdate {
-	return &models.RedisDataCentreUpdate{
-		DataCentres: rs.DCsToInstAPI(),
+func (rs *RedisSpec) ToClusterUpdate() *models.RedisUpdate {
+	return &models.RedisUpdate{
+		DataCentres:    rs.DCsToInstAPI(),
+		ResizeSettings: resizeSettingsToInstAPI(rs.ResizeSettings),
 	}
 }
 

apis/clusters/v1beta1/redis_webhook.go

@@ -123,6 +123,13 @@ func (rv *redisValidator) ValidateCreate(ctx context.Context, obj runtime.Object
 		}
 	}
 
+	for _, rs := range r.Spec.ResizeSettings {
+		err := validateSingleConcurrentResize(rs.Concurrency)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -157,6 +164,13 @@ func (rv *redisValidator) ValidateUpdate(ctx context.Context, old runtime.Object
 		return fmt.Errorf("update validation error: %v", err)
 	}
 
+	for _, rs := range r.Spec.ResizeSettings {
+		err := validateSingleConcurrentResize(rs.Concurrency)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }