func()

in pkg/controller/sub_controller/disaggregated_subcontroller.go [391:513]
98 lines of code
21 McCabe index (conditional complexity)

func (d *DisaggregatedSubDefaultController) PersistentVolumeArrayBuildVolumesVolumeMountsAndPVCs(commonSpec *v1.CommonSpec, confMap map[string]interface{}, componentType v1.DisaggregatedComponentType) ([]corev1.Volume, []corev1.VolumeMount, []corev1.PersistentVolumeClaim) {
	var requiredPaths []string

	//find storage mountPaths.
	switch componentType {
	case v1.DisaggregatedFE:
		metaPath := d.getFEMetaPath(confMap)
		requiredPaths = append(requiredPaths, metaPath)
	case v1.DisaggregatedBE:
		cachePaths, _ := d.getCacheMaxSizeAndPaths(confMap)
		requiredPaths = append(requiredPaths, cachePaths...)
	default:
	}

	//check logNotStore, if true should not generate log pvc.
	logNotStore := false
	for _, v1pv := range commonSpec.PersistentVolumes {
		if len(v1pv.MountPaths) != 0 {
			requiredPaths = append(requiredPaths, v1pv.MountPaths...)
		}

		logNotStore = logNotStore || v1pv.LogNotStore
	}

	//the last check logNotStore, fist check config in any one of persistentVolumes.
	if !logNotStore && !commonSpec.LogNotStore {
		logPath := d.getLogPath(confMap, componentType)
		requiredPaths = append(requiredPaths, logPath)
	}

	//generate name of persistentVolumeClaim use the mountPath
	namePath := map[string]string{}
	pathName := map[string]string{}
	for _, path := range requiredPaths {
		//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
	}

	pathPV := map[string]*v1.PersistentVolume{}
	//the template index.
	ti := -1
	for i, v1pv := range commonSpec.PersistentVolumes {
		if len(v1pv.MountPaths) == 0 {
			ti = i
			continue
		}
		for _, mp := range v1pv.MountPaths {
			pathPV[mp] = &commonSpec.PersistentVolumes[i]
		}
	}

	var vs []corev1.Volume
	var vms []corev1.VolumeMount
	var pvcs []corev1.PersistentVolumeClaim

	//generate pvc from the last path in requiredPaths, the mountPath that  configured by user is the highest wight, so first use the v1pv to generate pvc not template v1pv.
	ss := set.NewSetString()

	for i:= len(requiredPaths); i > 0; i-- {
		path := requiredPaths[i -1]
		//if the path have build volume, vm, pvc, skip it.
		if ss.Find(path) {
			continue
		}
		ss.Add(path)

		pv, ok := pathPV[path]
		name := pathName[path]
		metadataName := strings.ReplaceAll(name, "_", "-")
		//use specific PersistentVolume generate volume, vm, pvc
		if ok {
			vs = append(vs, corev1.Volume{Name: metadataName, VolumeSource: corev1.VolumeSource{
				PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
					ClaimName: metadataName,
				}}})
			vms = append(vms, corev1.VolumeMount{Name: metadataName, MountPath: path})
			pvcs = append(pvcs, corev1.PersistentVolumeClaim{
				ObjectMeta: metav1.ObjectMeta{
					Name:        metadataName,
					Annotations: pv.Annotations,
				},
				Spec: *pv.PersistentVolumeClaimSpec.DeepCopy(),
			})
		}

		//use template PersistentVolume generate volume, vm, pvc
		if !ok && ti != -1 {
			vs = append(vs, corev1.Volume{Name: metadataName, VolumeSource: corev1.VolumeSource{
				PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
					ClaimName: metadataName,
				}}})
			vms = append(vms, corev1.VolumeMount{Name: metadataName, MountPath: path})
			pvcs = append(pvcs, corev1.PersistentVolumeClaim{
				ObjectMeta: metav1.ObjectMeta{
					Name:        metadataName,
					Annotations: commonSpec.PersistentVolumes[ti].Annotations,
				},
				Spec: *commonSpec.PersistentVolumes[ti].PersistentVolumeClaimSpec.DeepCopy(),
			})
		}
	}

	return vs, vms, pvcs
}