add volume server disks

api/v1/seaweed_types.go

@@ -36,12 +36,15 @@ type SeaweedSpec struct {
 
 	// VolumeServerCount is the number of volume servers, default to 1
 	VolumeServerCount int32 `json:"volumeServerCount,omitempty"`
+	VolumeServerDiskCount int32 `json:"volumeServerDiskCount,omitempty"`
+	VolumeServerDiskSizeInGiB int32 `json:"volumeServerDiskSizeInGiB,omitempty"`
 
 	// FilerCount is the number of filers, default to 1
 	FilerCount int32 `json:"filerCount,omitempty"`
 
 	// ingress
 	Hosts []string `json:"hosts"`
+
 }
 
 // SeaweedStatus defines the observed state of Seaweed

config/crd/bases/seaweed.seaweedfs.com_seaweeds.yaml

@@ -56,6 +56,12 @@ spec:
                 to 1
               format: int32
               type: integer
+            volumeServerDiskCount:
+              format: int32
+              type: integer
+            volumeServerDiskSizeInGiB:
+              format: int32
+              type: integer
           required:
           - hosts
           type: object

controllers/controller_filer_statefulset.go

@@ -110,7 +110,7 @@ func (r *SeaweedReconciler) createFilerStatefulSet(m *seaweedv1.Seaweed) *appsv1
 						LivenessProbe: &corev1.Probe{
 							Handler: corev1.Handler{
 								HTTPGet: &corev1.HTTPGetAction{
-									Path: "/",
+									Path:   "/",
 									Port:   intstr.FromInt(8888),
 									Scheme: corev1.URISchemeHTTP,
 								},

controllers/controller_volume_statefulset.go

@@ -2,9 +2,12 @@ package controllers
 
 import (
 	"fmt"
+	"log"
+	"strings"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 
@@ -17,6 +20,51 @@ func (r *SeaweedReconciler) createVolumeServerStatefulSet(m *seaweedv1.Seaweed)
 	rollingUpdatePartition := int32(0)
 	enableServiceLinks := false
 
+	volumeQuantity := fmt.Sprintf("%dGi", m.Spec.VolumeServerDiskSizeInGiB)
+	volumeCount := int(m.Spec.VolumeServerDiskCount)
+	quantity, err := resource.ParseQuantity(volumeQuantity)
+	if err != nil {
+		log.Fatalf("can not parse quantity %s", volumeQuantity)
+	}
+	volumeRequests := make(corev1.ResourceList)
+	volumeRequests[corev1.ResourceStorage] = quantity
+
+	// connect all the disks
+	var volumeMounts []corev1.VolumeMount
+	var volumes []corev1.Volume
+	var persistentVolumeClaims []corev1.PersistentVolumeClaim
+	var dirs []string
+	for i := 0; i < volumeCount; i++ {
+		volumeMounts = append(volumeMounts, corev1.VolumeMount{
+			Name:      fmt.Sprintf("mount%d", i),
+			ReadOnly:  false,
+			MountPath: fmt.Sprintf("/data%d/", i),
+		})
+		volumes = append(volumes, corev1.Volume{
+			Name: fmt.Sprintf("mount%d", i),
+			VolumeSource: corev1.VolumeSource{
+				PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
+					ClaimName: fmt.Sprintf("mount%d", i),
+					ReadOnly:  false,
+				},
+			},
+		})
+		persistentVolumeClaims = append(persistentVolumeClaims, corev1.PersistentVolumeClaim{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("mount%d", i),
+			},
+			Spec: corev1.PersistentVolumeClaimSpec{
+				AccessModes: []corev1.PersistentVolumeAccessMode{
+					corev1.ReadWriteOnce,
+				},
+				Resources: corev1.ResourceRequirements{
+					Requests: volumeRequests,
+				},
+			},
+		})
+		dirs = append(dirs, fmt.Sprintf("/data%d", i))
+	}
+
 	dep := &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      m.Name + "-volume",
@@ -74,8 +122,9 @@ func (r *SeaweedReconciler) createVolumeServerStatefulSet(m *seaweedv1.Seaweed)
 						Command: []string{
 							"/bin/sh",
 							"-ec",
-							fmt.Sprintf("weed volume -port=8444 -max=0 %s %s",
+							fmt.Sprintf("weed volume -port=8444 -max=0 %s %s %s",
 								fmt.Sprintf("-ip=$(POD_NAME).%s-volume", m.Name),
+								fmt.Sprintf("-dir=%s", strings.Join(dirs, ",")),
 								fmt.Sprintf("-mserver=%s-master-0.%s-master:9333,%s-master-1.%s-master:9333,%s-master-2.%s-master:9333",
 									m.Name, m.Name, m.Name, m.Name, m.Name, m.Name),
 							),
@@ -106,7 +155,7 @@ func (r *SeaweedReconciler) createVolumeServerStatefulSet(m *seaweedv1.Seaweed)
 						LivenessProbe: &corev1.Probe{
 							Handler: corev1.Handler{
 								HTTPGet: &corev1.HTTPGetAction{
-									Path: "/status",
+									Path:   "/status",
 									Port:   intstr.FromInt(8444),
 									Scheme: corev1.URISchemeHTTP,
 								},
@@ -117,9 +166,13 @@ func (r *SeaweedReconciler) createVolumeServerStatefulSet(m *seaweedv1.Seaweed)
 							SuccessThreshold:    1,
 							FailureThreshold:    6,
 						},
+						VolumeMounts: volumeMounts,
 					}},
+
+					Volumes: volumes,
 				},
 			},
+			VolumeClaimTemplates: persistentVolumeClaims,
 		},
 	}
 	return dep