Init blobstorage Job instead of exec InPod (#182)

Author: kobzonega

api/v1alpha1/database_types.go

@@ -264,3 +264,9 @@ type DatabaseServices struct {
 func init() {
 	SchemeBuilder.Register(&Database{}, &DatabaseList{})
 }
+
+func (r *Database) AnyCertificatesAdded() bool {
+	return len(r.Spec.CABundle) > 0 ||
+		r.Spec.Service.GRPC.TLSConfiguration.Enabled ||
+		r.Spec.Service.Interconnect.TLSConfiguration.Enabled
+}

api/v1alpha1/storage_types.go

@@ -13,6 +13,16 @@ type StorageSpec struct {
 
 	StorageNodeSpec `json:",inline"`
 
+	// (Optional) Operator connection settings
+	// Default: (not specified)
+	// +optional
+	OperatorConnection *ConnectionOptions `json:"operatorConnection,omitempty"`
+
+	// (Optional) Init blobstorage Job settings
+	// Default: (not specified)
+	// +optional
+	InitJob *StorageInitJobSpec `json:"initJob,omitempty"`
+
 	// (Optional) NodeSet inline configuration to split into multiple StatefulSets
 	// Default: (not specified)
 	// +optional
@@ -28,11 +38,6 @@ type StorageClusterSpec struct {
 	// +optional
 	Domain string `json:"domain"`
 
-	// (Optional) Operator connection settings
-	// Default: (not specified)
-	// +optional
-	OperatorConnection *ConnectionOptions `json:"operatorConnection,omitempty"`
-
 	// Data storage topology mode
 	// For details, see https://ydb.tech/docs/en/cluster/topology
 	// FIXME mirror-3-dc is only supported with external configuration
@@ -162,6 +167,36 @@ type StorageNodeSpec struct {
 	AdditionalAnnotations map[string]string `json:"additionalAnnotations,omitempty"`
 }
 
+type StorageInitJobSpec struct {
+	// (Optional) Container resource limits. Any container limits
+	// can be specified.
+	// Default: (not specified)
+	// +optional
+	Resources *corev1.ResourceRequirements `json:"resources,omitempty"`
+
+	// (Optional) NodeSelector is a selector which must be true for the pod to fit on a node.
+	// Selector which must match a node's labels for the pod to be scheduled on that node.
+	// More info: https://kubernetes.io/docs/concepts/configuration/assign-pod-node/
+	// +optional
+	NodeSelector map[string]string `json:"nodeSelector,omitempty"`
+
+	// (Optional) If specified, the pod's scheduling constraints
+	// +optional
+	Affinity *corev1.Affinity `json:"affinity,omitempty"`
+
+	// (Optional) If specified, the pod's tolerations.
+	// +optional
+	Tolerations []corev1.Toleration `json:"tolerations,omitempty"`
+
+	// (Optional) Additional custom resource labels that are added to all resources
+	// +optional
+	AdditionalLabels map[string]string `json:"additionalLabels,omitempty"`
+
+	// (Optional) Additional custom resource annotations that are added to all resources
+	// +optional
+	AdditionalAnnotations map[string]string `json:"additionalAnnotations,omitempty"`
+}
+
 // StorageStatus defines the observed state of Storage
 type StorageStatus struct {
 	State      constants.ClusterState `json:"state"`
@@ -203,3 +238,9 @@ type StorageServices struct {
 func init() {
 	SchemeBuilder.Register(&Storage{}, &StorageList{})
 }
+
+func (r *Storage) AnyCertificatesAdded() bool {
+	return len(r.Spec.CABundle) > 0 ||
+		r.Spec.Service.GRPC.TLSConfiguration.Enabled ||
+		r.Spec.Service.Interconnect.TLSConfiguration.Enabled
+}

api/v1alpha1/storage_webhook.go

@@ -3,18 +3,20 @@ package v1alpha1
 import (
 	"context"
 	"fmt"
+	"math/rand"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"gopkg.in/yaml.v3"
-	v1 "k8s.io/api/core/v1"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/utils/strings/slices"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
+	"github.com/ydb-platform/ydb-kubernetes-operator/internal/configuration/schema"
 	. "github.com/ydb-platform/ydb-kubernetes-operator/internal/controllers/constants" //nolint:revive,stylecheck
 )
 
@@ -29,15 +31,28 @@ func (r *Storage) SetupWebhookWithManager(mgr ctrl.Manager) error {
 }
 
 func (r *Storage) GetStorageEndpointWithProto() string {
+	return fmt.Sprintf("%s%s", r.GetStorageProto(), r.GetStorageEndpoint())
+}
+
+func (r *Storage) GetStorageProto() string {
 	proto := GRPCProto
 	if r.IsStorageEndpointSecure() {
 		proto = GRPCSProto
 	}
 
-	return fmt.Sprintf("%s%s", proto, r.GetStorageEndpoint())
+	return proto
 }
 
 func (r *Storage) GetStorageEndpoint() string {
+	endpoint := r.GetGRPCServiceEndpoint()
+	if r.IsRemoteNodeSetsOnly() {
+		endpoint = r.GetHostFromConfigEndpoint()
+	}
+
+	return endpoint
+}
+
+func (r *Storage) GetGRPCServiceEndpoint() string {
 	host := fmt.Sprintf(GRPCServiceFQDNFormat, r.Name, r.Namespace)
 	if r.Spec.Service.GRPC.ExternalHost != "" {
 		host = r.Spec.Service.GRPC.ExternalHost
@@ -46,15 +61,44 @@ func (r *Storage) GetStorageEndpoint() string {
 	return fmt.Sprintf("%s:%d", host, GRPCPort)
 }
 
+// +k8s:deepcopy-gen=false
+type PartialHostsConfig struct {
+	Hosts []schema.Host `yaml:"hosts,flow"`
+}
+
+func (r *Storage) GetHostFromConfigEndpoint() string {
+	// skip handle error because we already checked in webhook
+	hostsConfig := PartialHostsConfig{}
+	_ = yaml.Unmarshal([]byte(r.Spec.Configuration), &hostsConfig)
+
+	randNum := rand.Int31n(r.Spec.Nodes) // #nosec G404
+	host := hostsConfig.Hosts[randNum].Host
+	return fmt.Sprintf("%s:%d", host, GRPCPort)
+}
+
 func (r *Storage) IsStorageEndpointSecure() bool {
 	if r.Spec.Service.GRPC.TLSConfiguration != nil {
 		return r.Spec.Service.GRPC.TLSConfiguration.Enabled
 	}
 	return false
 }
 
+func (r *Storage) IsRemoteNodeSetsOnly() bool {
+	if len(r.Spec.NodeSets) == 0 {
+		return false
+	}
+
+	for _, nodeSet := range r.Spec.NodeSets {
+		if nodeSet.Remote == nil {
+			return false
+		}
+	}
+
+	return true
+}
+
 // +k8s:deepcopy-gen=false
-type PartialYamlConfig struct {
+type PartialDomainsConfig struct {
 	DomainsConfig struct {
 		SecurityConfig struct {
 			EnforceUserTokenRequirement bool `yaml:"enforce_user_token_requirement"`
@@ -88,12 +132,12 @@ func (r *StorageDefaulter) Default(ctx context.Context, obj runtime.Object) erro
 	}
 
 	if storage.Spec.Image.PullPolicyName == nil {
-		policy := v1.PullIfNotPresent
+		policy := corev1.PullIfNotPresent
 		storage.Spec.Image.PullPolicyName = &policy
 	}
 
 	if storage.Spec.Resources == nil {
-		storage.Spec.Resources = &v1.ResourceRequirements{}
+		storage.Spec.Resources = &corev1.ResourceRequirements{}
 	}
 
 	if storage.Spec.Service == nil {
@@ -142,23 +186,35 @@ func (r *Storage) ValidateCreate() error {
 	configuration := make(map[string]interface{})
 	err := yaml.Unmarshal([]byte(r.Spec.Configuration), &configuration)
 	if err != nil {
-		return fmt.Errorf("failed to parse Storage.spec.configuration, error: %w", err)
+		return fmt.Errorf("failed to parse .spec.configuration, error: %w", err)
+	}
+
+	hostsConfig := PartialHostsConfig{}
+	err = yaml.Unmarshal([]byte(r.Spec.Configuration), &hostsConfig)
+	if err != nil {
+		return fmt.Errorf("failed to parse YAML to determine `hosts`, error: %w", err)
 	}
+
 	var nodesNumber int32
-	if configuration["hosts"] == nil {
+	if len(hostsConfig.Hosts) == 0 {
 		nodesNumber = r.Spec.Nodes
 	} else {
-		hosts, ok := configuration["hosts"].([]interface{})
-		if !ok {
-			return fmt.Errorf("failed to parse Storage.spec.configuration, error: invalid hosts section")
-		}
-		nodesNumber = int32(len(hosts))
+		nodesNumber = int32(len(hostsConfig.Hosts))
+	}
+
+	minNodesPerErasure := map[ErasureType]int32{
+		ErasureMirror3DC: 9,
+		ErasureBlock42:   8,
+		None:             1,
+	}
+	if nodesNumber < minNodesPerErasure[r.Spec.Erasure] {
+		return fmt.Errorf("erasure type %v requires at least %v storage nodes", r.Spec.Erasure, minNodesPerErasure[r.Spec.Erasure])
 	}
 
-	yamlConfig := PartialYamlConfig{}
+	yamlConfig := PartialDomainsConfig{}
 	err = yaml.Unmarshal([]byte(r.Spec.Configuration), &yamlConfig)
 	if err != nil {
-		return fmt.Errorf("failed to parse YAML to determine `enforce_user_token_requirement`")
+		return fmt.Errorf("failed to parse YAML to determine `enforce_user_token_requirement`, error: %w", err)
 	}
 
 	var authEnabled bool
@@ -170,15 +226,6 @@ func (r *Storage) ValidateCreate() error {
 		return fmt.Errorf("field 'spec.operatorConnection' does not satisfy with config option `enforce_user_token_requirement: %t`", authEnabled)
 	}
 
-	minNodesPerErasure := map[ErasureType]int32{
-		ErasureMirror3DC: 9,
-		ErasureBlock42:   8,
-		None:             1,
-	}
-	if nodesNumber < minNodesPerErasure[r.Spec.Erasure] {
-		return fmt.Errorf("erasure type %v requires at least %v storage nodes", r.Spec.Erasure, minNodesPerErasure[r.Spec.Erasure])
-	}
-
 	if r.Spec.NodeSets != nil {
 		var nodesInSetsCount int32
 		for _, nodeSetInline := range r.Spec.NodeSets {
@@ -237,7 +284,29 @@ func (r *Storage) ValidateUpdate(old runtime.Object) error {
 	configuration := make(map[string]interface{})
 	err := yaml.Unmarshal([]byte(r.Spec.Configuration), &configuration)
 	if err != nil {
-		return fmt.Errorf("failed to parse Storage.spec.configuration, error: %w", err)
+		return fmt.Errorf("failed to parse .spec.configuration, error: %w", err)
+	}
+
+	hostsConfig := PartialHostsConfig{}
+	err = yaml.Unmarshal([]byte(r.Spec.Configuration), &hostsConfig)
+	if err != nil {
+		return fmt.Errorf("failed to parse YAML to determine `hosts`, error: %w", err)
+	}
+
+	var nodesNumber int32
+	if len(hostsConfig.Hosts) == 0 {
+		nodesNumber = r.Spec.Nodes
+	} else {
+		nodesNumber = int32(len(hostsConfig.Hosts))
+	}
+
+	minNodesPerErasure := map[ErasureType]int32{
+		ErasureMirror3DC: 9,
+		ErasureBlock42:   8,
+		None:             1,
+	}
+	if nodesNumber < minNodesPerErasure[r.Spec.Erasure] {
+		return fmt.Errorf("erasure type %v requires at least %v storage nodes", r.Spec.Erasure, minNodesPerErasure[r.Spec.Erasure])
 	}
 
 	if !r.Spec.OperatorSync {
@@ -258,10 +327,10 @@ func (r *Storage) ValidateUpdate(old runtime.Object) error {
 		}
 	}
 
-	yamlConfig := PartialYamlConfig{}
+	yamlConfig := PartialDomainsConfig{}
 	err = yaml.Unmarshal([]byte(r.Spec.Configuration), &yamlConfig)
 	if err != nil {
-		return fmt.Errorf("failed to parse YAML to determine `enforce_user_token_requirement`")
+		return fmt.Errorf("failed to parse YAML to determine `enforce_user_token_requirement`, error: %w", err)
 	}
 
 	var authEnabled bool