[improve](dcr) support pvc template and be multi-data disk

pkg/common/utils/doris/storage_conf.go

@@ -0,0 +1,100 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package doris
+
+import (
+	"strings"
+)
+
+// ResolveStorageRootPath transforms a string of storage paths into a slice of StorageRootPathInfo.
+func ResolveStorageRootPath(configPath string) []string {
+	var res []string
+
+	if configPath == "" {
+		return res
+	}
+
+	// Separate multiple paths by ';'
+	configPathSplit := strings.Split(configPath, ";")
+
+	// Remove empty elements
+	for _, c := range configPathSplit {
+		if path := parseSinglePath(c); path != "" {
+			res = append(res, path)
+		}
+	}
+
+	return res
+}
+
+// Resolving a single storage path
+func parseSinglePath(pathConfig string) string {
+	if pathConfig == "" {
+		return ""
+	}
+	path := strings.Split(strings.Split(pathConfig, ".")[0], ",")[0]
+	path = strings.TrimSpace(path)
+	if strings.HasSuffix(path, "/") {
+		path = path[:len(path)-1]
+	}
+	return path
+}
+
+// GetNameOfEachPath is used to parse a set of paths to obtain unique and concise names for each path.
+// If the paths are repeated, the returned names may also be repeated.
+// And the order of each name in the array is consistent with the input paths.
+// For example:
+//
+//	["/path1"] >> ["path1"]
+//	["/opt/doris/path1"] >> ["path1"]
+//	["/path1", "/path2"] >> ["path1", "path2"]
+//	["/home/disk1/doris", "/home/disk2/doris"] >> ["doris", "disk2-doris"]
+//	["/home/doris/disk1", "/home/doris/disk2"] >> ["disk1", "disk2"]
+//	["/home/disk1/doris", "/home/disk1/doris", "/home/disk2/doris"] >> ["disk1-doris", "disk1-doris", "disk2-doris"]
+func GetNameOfEachPath(paths []string) []string {
+	namePath := map[string]string{}
+	pathName := map[string]string{}
+	for _, path := range paths {
+		//use unix path separator.
+		sp := strings.Split(path, "/")
+		name := ""
+		for i := 1; i <= len(sp); i++ {
+			if sp[len(sp)-i] == "" {
+				continue
+			}
+
+			if name == "" {
+				name = sp[len(sp)-i]
+			} else {
+				name = sp[len(sp)-i] + "-" + name
+			}
+
+			if _, ok := namePath[name]; !ok {
+				break
+			}
+		}
+
+		namePath[name] = path
+		pathName[path] = name
+	}
+	res := make([]string, len(paths))
+	for k := range paths {
+		res[k] = pathName[paths[k]]
+	}
+	return res
+}

pkg/common/utils/doris/storage_conf_test.go

@@ -0,0 +1,173 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package doris
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestResolveStorageRootPath(t *testing.T) {
+	var empty []string
+	tests := []struct {
+		name    string
+		input   string
+		want    []string
+		wantErr bool
+	}{
+		// Normal test
+		{
+			input: "",
+			want:  empty,
+		},
+		{
+			input: "/path1",
+			want:  []string{"/path1"},
+		},
+		{
+			input: "/path1;/path2",
+			want:  []string{"/path1", "/path2"},
+		},
+		{
+			input: "/home/disk1/doris.HDD,50",
+			want:  []string{"/home/disk1/doris"},
+		},
+		{
+			input: "/home/disk1/doris,medium:ssd,capacity:50",
+			want:  []string{"/home/disk1/doris"},
+		},
+		{
+			input: "/home/disk1/doris.SSD,100;/home/disk2/doris,medium:hdd,capacity:200",
+			want:  []string{"/home/disk1/doris", "/home/disk2/doris"},
+		},
+		{
+			input: "/home/disk1/doris/,capacity:50",
+			want:  []string{"/home/disk1/doris"},
+		},
+		{
+			input: "/home/disk1/doris.HDD,medium:ssd",
+			want:  []string{"/home/disk1/doris"},
+		},
+		{
+			input: "/home/disk1/doris,capacity:50;",
+			want:  []string{"/home/disk1/doris"},
+		},
+		{
+			input: " /home/disk1/doris , capacity : 50 ; /home/disk2/doris , medium : ssd ",
+			want:  []string{"/home/disk1/doris", "/home/disk2/doris"},
+		},
+
+		{
+			input: "/home/disk1/doris,capacity:50;/home/disk1/doris,medium:ssd",
+			want:  []string{"/home/disk1/doris", "/home/disk1/doris"},
+		},
+
+		{
+			input: ",capacity:50",
+			want:  empty,
+		},
+
+		{
+			input: "/home/disk1/doris/,unknown:value",
+			want:  []string{"/home/disk1/doris"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := ResolveStorageRootPath(tt.input)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("input: %s, got %v, Expectation %v", tt.input, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestGetNameOfEachPath(t *testing.T) {
+	tests := []struct {
+		name  string
+		input []string
+		want  []string
+	}{
+		{
+			input: []string{},
+			want:  []string{},
+		},
+		{
+			input: []string{"", "", "", ""},
+			want:  []string{"", "", "", ""},
+		},
+		{
+			input: []string{"", ""},
+			want:  []string{"", ""},
+		},
+		{
+			input: []string{"/path1"},
+			want:  []string{"path1"},
+		},
+		{
+			input: []string{"/opt/doris/path1"},
+			want:  []string{"path1"},
+		},
+		{
+			input: []string{"/path1", "/path2"},
+			want:  []string{"path1", "path2"},
+		},
+		{
+			input: []string{"/home/disk1/doris", "/home/disk2/doris"},
+			want:  []string{"doris", "disk2-doris"},
+		},
+		{
+			input: []string{"/home/disk1/doris", "/home/disk1/doris"},
+			want:  []string{"disk1-doris", "disk1-doris"},
+		},
+		{
+			input: []string{"/home/disk1/doris", "/home/disk1/doris", "/home/disk2/doris"},
+			want:  []string{"disk1-doris", "disk1-doris", "disk2-doris"},
+		},
+		{
+			input: []string{"/home/disk1/doris", "/home/disk2/doris", "/home/disk3/doris"},
+			want:  []string{"doris", "disk2-doris", "disk3-doris"},
+		},
+		{
+			input: []string{"/home/disk1/doris", "/home/disk1/doris/subdir"},
+			want:  []string{"doris", "subdir"},
+		},
+		{
+			input: []string{"/home/disk1/doris", "/home/disk1/doris/subdir", "/home/disk1/doris/subdir/subsubdir"},
+			want:  []string{"doris", "subdir", "subsubdir"},
+		},
+		{
+			input: []string{"/home/disk1/doris", "/home/disk1/doris/subdir", "/home/disk1/doris/subdir/subsubdir", "/home/disk1/doris/subdir/subsubdir"},
+			want:  []string{"doris", "subdir", "subdir-subsubdir", "subdir-subsubdir"},
+		},
+		{
+			input: []string{"/home/disk1/doris", "/home/disk1/doris", "/home/disk2/doris"},
+			want:  []string{"disk1-doris", "disk1-doris", "disk2-doris"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := GetNameOfEachPath(tt.input)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("input: %v, got %v, Expectation %v", tt.input, got, tt.want)
+			}
+		})
+	}
+}

pkg/common/utils/resource/configmap.go

@@ -24,6 +24,7 @@ import (
 	"github.com/spf13/viper"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/klog/v2"
+	"os"
 )
 
 // the fe ports key
@@ -124,7 +125,9 @@ func ResolveConfigMaps(configMaps []*corev1.ConfigMap, componentType dorisv1.Com
 			continue
 		}
 		if value, ok := configMap.Data[key]; ok {
+			os.Setenv("DORIS_HOME", getDefaultDorisHome(componentType))
 			viper.SetConfigType("properties")
+			viper.AutomaticEnv()
 			viper.ReadConfig(bytes.NewBuffer([]byte(value)))
 			return viper.AllSettings(), nil
 		}

pkg/common/utils/resource/persistent_volume_claim.go

@@ -19,9 +19,11 @@ package resource
 
 import (
 	dorisv1 "github.com/apache/doris-operator/api/doris/v1"
+	"github.com/apache/doris-operator/pkg/common/utils/doris"
 	"github.com/apache/doris-operator/pkg/common/utils/hash"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/klog/v2"
 )
 
 var (
@@ -66,3 +68,87 @@ func buildPVCAnnotations(volume dorisv1.PersistentVolume) Annotations {
 	}
 	return annotations
 }
+
+func getDefaultDorisHome(componentType dorisv1.ComponentType) string {
+	switch componentType {
+	case dorisv1.Component_FE:
+		return DEFAULT_ROOT_PATH + "/fe"
+	case dorisv1.Component_BE, dorisv1.Component_CN:
+		return DEFAULT_ROOT_PATH + "/be"
+	case dorisv1.Component_Broker:
+		return DEFAULT_ROOT_PATH + "/apache_hdfs_broker"
+	default:
+		klog.Infof("the componentType: %s have not default DORIS_HOME", componentType)
+	}
+	return ""
+}
+
+// GenerateEveryoneMountPathPersistentVolume is used to process the pvc template configuration in CRD.
+// The template is defined as follows:
+// - PersistentVolume.MountPath is "", it`s template configuration.
+// - PersistentVolume.MountPath is not "", it`s actual pvc configuration.
+// The Explain rules are as follows:
+// 1. Non-templated PersistentVolumes are returned directly in the result list.
+// 2. If there is a pvc template, return the actual list of pvcs after processing.
+// 3. The template needs to parse the configuration of the doris config file to create the pvc.
+// 4. If there are multiple templates, the last valid template will be used.
+func GenerateEveryoneMountPathPersistentVolume(spec *dorisv1.BaseSpec, config map[string]interface{}, componentType dorisv1.ComponentType) ([]dorisv1.PersistentVolume, error) {
+
+	// Only the last data pvc template configuration takes effect
+	var template *dorisv1.PersistentVolume
+	// pvs is the pvc that needs to be actually created, specified by the user
+	var pvs []dorisv1.PersistentVolume
+
+	for i := range spec.PersistentVolumes {
+		if spec.PersistentVolumes[i].MountPath != "" {
+			pvs = append(pvs, spec.PersistentVolumes[i])
+
+		} else {
+			template = (&spec.PersistentVolumes[i]).DeepCopy()
+		}
+	}
+
+	if template == nil {
+		return pvs, nil
+	}
+
+	// Processing pvc template
+	var dataPathKey, dataDefaultPath string
+	var dataPaths []string
+	dorisHome := getDefaultDorisHome(componentType)
+	switch componentType {
+	case dorisv1.Component_FE:
+		dataPathKey = "meta_dir"
+		dataDefaultPath = dorisHome + "/doris-meta"
+	case dorisv1.Component_BE, dorisv1.Component_CN:
+		dataPathKey = "storage_root_path"
+		dataDefaultPath = dorisHome + "/storage"
+	default:
+		klog.Infof("GenerateEveryoneMountPathPersistentVolume the componentType: %s is not supported, PersistentVolume template will not work ", componentType)
+		return pvs, nil
+	}
+
+	dataPathValue, dataExist := config[dataPathKey]
+	if !dataExist {
+		klog.Infof("GenerateEveryoneMountPathPersistentVolume: dataPathKey '%s' not found in config, default value will effect", dataPathKey)
+		dataPaths = append(dataPaths, dataDefaultPath)
+	} else {
+		dataPaths = doris.ResolveStorageRootPath(dataPathValue.(string))
+	}
+
+	if len(dataPaths) == 1 {
+		tmp := *template.DeepCopy()
+		tmp.MountPath = dataPaths[0]
+		pvs = append(pvs, tmp)
+	} else {
+		pathName := doris.GetNameOfEachPath(dataPaths)
+		for i := range dataPaths {
+			tmp := *template.DeepCopy()
+			tmp.Name = tmp.Name + "-" + pathName[i]
+			tmp.MountPath = dataPaths[i]
+			pvs = append(pvs, tmp)
+		}
+	}
+
+	return pvs, nil
+}