Merge pull request #54 from cybozu-go/implement-export

Implement export

charts/mantle-cluster-wide/templates/clusterrole.yaml

@@ -31,6 +31,7 @@ rules:
       - batch
     resources:
       - cronjobs
+      - jobs
     verbs:
       - create
       - delete

charts/mantle/templates/deployment.yaml

@@ -67,6 +67,12 @@ spec:
             {{- with .Values.controller.overwriteMBCSchedule }}
             - --overwrite-mbc-schedule={{ . }}
             {{- end }}
+            {{- with .Values.controller.objectStorageBucketName }}
+            - --object-storage-bucket-name={{ . }}
+            {{- end }}
+            {{- with .Values.controller.objectStorageEndpoint}}
+            - --object-storage-endpoint={{ . }}
+            {{- end }}
           env:
             - name: POD_NAME
               valueFrom:
@@ -76,6 +82,8 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
+            - name: POD_IMAGE
+              value: {{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}
           ports:
             {{- toYaml .Values.controller.ports | nindent 12 }}
         - command:

cmd/controller/main.go

@@ -42,13 +42,20 @@ var ControllerCmd = &cobra.Command{
 }
 
 var (
-	metricsAddr           string
-	enableLeaderElection  bool
-	probeAddr             string
-	zapOpts               zap.Options
-	overwriteMBCSchedule  string
-	role                  string
-	mantleServiceEndpoint string
+	metricsAddr             string
+	enableLeaderElection    bool
+	probeAddr               string
+	zapOpts                 zap.Options
+	overwriteMBCSchedule    string
+	role                    string
+	mantleServiceEndpoint   string
+	maxExportJobs           int
+	exportDataStorageClass  string
+	envSecret               string
+	objectStorageBucketName string
+	objectStorageEndpoint   string
+	caCertConfigMapSrc      string
+	caCertKeySrc            string
 
 	scheme   = runtime.NewScheme()
 	setupLog = ctrl.Log.WithName("setup")
@@ -73,6 +80,21 @@ func init() {
 			"(i) If --role is 'standalone', this option is ignored. (ii) If --role is 'primary', this option is required "+
 			"and is interpreted as the address that the primary mantle should connect to. (iii) If --role is 'secondary', "+
 			"this option is required and is interpreted as the address that the secondary mantle should listen to.")
+	flags.IntVar(&maxExportJobs, "max-export-jobs", 8,
+		"The maximum number of export jobs that can run simultaneously. If you set this to 0, there is no limit.")
+	flags.StringVar(&exportDataStorageClass, "export-data-storage-class", "",
+		"The storage class of PVCs used to store exported data temporarily.")
+	flags.StringVar(&envSecret, "env-secret", "",
+		"The name of the Secret resource that contains environment variables related to the controller and Jobs.")
+	flags.StringVar(&objectStorageBucketName, "object-storage-bucket-name", "",
+		"The bucket name of the object storage which should be used to store backups.")
+	flags.StringVar(&objectStorageEndpoint, "object-storage-endpoint", "",
+		"The endpoint URL to access the object storage.")
+	flags.StringVar(&caCertConfigMapSrc, "ca-cert-configmap", "",
+		"The name of the ConfigMap resource that contains the intermediate certificate used to access the object storage.")
+	flags.StringVar(&caCertKeySrc, "ca-cert-key", "ca.crt",
+		"The key of the ConfigMap specified by --ca-cert-config-map that contains the intermediate certificate. "+
+			"The default value is ca.crt. This option is just ignored if --ca-cert-configmap isn't specified.")
 
 	goflags := flag.NewFlagSet("goflags", flag.ExitOnError)
 	zapOpts.Development = true
@@ -90,12 +112,20 @@ func checkCommandlineArgs() error {
 	case controller.RoleStandalone:
 		// nothing to do
 	case controller.RolePrimary:
-		if mantleServiceEndpoint == "" {
-			return errors.New("--mantle-service-endpoint must be specified if --role is 'primary'")
-		}
+		fallthrough
 	case controller.RoleSecondary:
 		if mantleServiceEndpoint == "" {
-			return errors.New("--mantle-service-endpoint must be specified if --role is 'secondary'")
+			return errors.New("--mantle-service-endpoint must be specified if --role is 'primary' or 'secondary'")
+		}
+		if caCertConfigMapSrc != "" && caCertKeySrc == "" {
+			return errors.New("--ca-cert-key must be specified if --role is 'primary' or 'secondary', " +
+				"and --ca-cert-configmap is specified")
+		}
+		if objectStorageBucketName == "" {
+			return errors.New("--object-storage-bucket-name must be specified if --role is 'primary' or 'secondary'")
+		}
+		if objectStorageEndpoint == "" {
+			return errors.New("--object-storage-endpoint must be specified if --role is 'primary' or 'secondary'")
 		}
 	default:
 		return fmt.Errorf("role should be one of 'standalone', 'primary', or 'secondary': %s", role)
@@ -110,12 +140,31 @@ func setupReconcilers(mgr manager.Manager, primarySettings *controller.PrimarySe
 		return errors.New("POD_NAMESPACE is empty")
 	}
 
+	podImage := os.Getenv("POD_IMAGE")
+	if podImage == "" {
+		setupLog.Error(errors.New("POD_IMAGE must not be empty"), "POD_IMAGE must not be empty")
+		return errors.New("POD_IMAGE is empty")
+	}
+
+	var caCertConfigMap *string
+	if caCertConfigMapSrc != "" {
+		caCertConfigMap = &caCertConfigMapSrc
+	}
+
 	backupReconciler := controller.NewMantleBackupReconciler(
 		mgr.GetClient(),
 		mgr.GetScheme(),
 		managedCephClusterID,
 		role,
 		primarySettings,
+		podImage,
+		envSecret,
+		&controller.ObjectStorageSettings{
+			BucketName:      objectStorageBucketName,
+			Endpoint:        objectStorageEndpoint,
+			CACertConfigMap: caCertConfigMap,
+			CACertKey:       &caCertKeySrc,
+		},
 	)
 	if err := backupReconciler.SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "MantleBackup")
@@ -174,9 +223,11 @@ func setupPrimary(ctx context.Context, mgr manager.Manager, wg *sync.WaitGroup)
 	}()
 
 	primarySettings := &controller.PrimarySettings{
-		ServiceEndpoint: mantleServiceEndpoint,
-		Conn:            conn,
-		Client:          proto.NewMantleServiceClient(conn),
+		ServiceEndpoint:        mantleServiceEndpoint,
+		Conn:                   conn,
+		Client:                 proto.NewMantleServiceClient(conn),
+		MaxExportJobs:          maxExportJobs,
+		ExportDataStorageClass: exportDataStorageClass,
 	}
 
 	return setupReconcilers(mgr, primarySettings)

config/rbac/role.yaml

@@ -31,6 +31,7 @@ rules:
   - batch
   resources:
   - cronjobs
+  - jobs
   verbs:
   - create
   - delete

docs/controller-protocol.md

@@ -10,6 +10,8 @@
     - [CreateOrUpdatePVCResponse](#proto-CreateOrUpdatePVCResponse)
     - [ListMantleBackupRequest](#proto-ListMantleBackupRequest)
     - [ListMantleBackupResponse](#proto-ListMantleBackupResponse)
+    - [SetSynchronizingRequest](#proto-SetSynchronizingRequest)
+    - [SetSynchronizingResponse](#proto-SetSynchronizingResponse)
 
     - [MantleService](#proto-MantleService)
 
@@ -111,6 +113,33 @@ ListMantleBackupResponse is a response message for ListMantleBackup RPC.
 
 
 
+
+<a name="proto-SetSynchronizingRequest"></a>
+
+### SetSynchronizingRequest
+SetSynchronizingRequest is a request message for SetSynchronize RPC.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| name | [string](#string) |  |  |
+| namespace | [string](#string) |  |  |
+| diffFrom | [string](#string) | optional |  |
+
+
+
+
+
+
+<a name="proto-SetSynchronizingResponse"></a>
+
+### SetSynchronizingResponse
+SetSynchronizingResponse is a response message for SetSynchronize RPC.
+
+
+
+
+
 
 
 
@@ -128,6 +157,7 @@ ListMantleBackupResponse is a response message for ListMantleBackup RPC.
 | CreateOrUpdatePVC | [CreateOrUpdatePVCRequest](#proto-CreateOrUpdatePVCRequest) | [CreateOrUpdatePVCResponse](#proto-CreateOrUpdatePVCResponse) |  |
 | CreateOrUpdateMantleBackup | [CreateOrUpdateMantleBackupRequest](#proto-CreateOrUpdateMantleBackupRequest) | [CreateOrUpdateMantleBackupResponse](#proto-CreateOrUpdateMantleBackupResponse) |  |
 | ListMantleBackup | [ListMantleBackupRequest](#proto-ListMantleBackupRequest) | [ListMantleBackupResponse](#proto-ListMantleBackupResponse) |  |
+| SetSynchronizing | [SetSynchronizingRequest](#proto-SetSynchronizingRequest) | [SetSynchronizingResponse](#proto-SetSynchronizingResponse) |  |
 
 
 

internal/controller/internal/testutil/resources.go

@@ -7,6 +7,7 @@ import (
 	mantlev1 "github.com/cybozu-go/mantle/api/v1"
 	"github.com/cybozu-go/mantle/test/util"
 	. "github.com/onsi/gomega"
+	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -29,13 +30,26 @@ type ResourceManager struct {
 	PoolName         string
 }
 
-func NewResourceManager(client client.Client) *ResourceManager {
+func NewResourceManager(client client.Client) (*ResourceManager, error) {
+	clusterID := util.GetUniqueName("ceph-")
+
+	// Create a namespace of the same name as cluster ID
+	ns := corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: clusterID,
+		},
+	}
+	err := client.Create(context.Background(), &ns)
+	if err != nil {
+		return nil, err
+	}
+
 	return &ResourceManager{
 		client:           client,
 		StorageClassName: util.GetUniqueName("sc-"),
-		ClusterID:        util.GetUniqueName("ceph-"),
+		ClusterID:        clusterID,
 		PoolName:         util.GetUniqueName("pool-"),
-	}
+	}, nil
 }
 
 // EnvTest cannot delete namespace. So, we have to use another new namespace.
@@ -208,6 +222,27 @@ func (r *ResourceManager) WaitForBackupSyncedToRemote(ctx context.Context, backu
 	}).WithContext(ctx).Should(Succeed())
 }
 
+func (r *ResourceManager) ChangeJobCondition(ctx context.Context, job *batchv1.Job, condType batchv1.JobConditionType, condStatus corev1.ConditionStatus) error {
+	if job.Status.Conditions == nil {
+		job.Status.Conditions = []batchv1.JobCondition{}
+	}
+	updated := false
+	for i := range job.Status.Conditions {
+		if job.Status.Conditions[i].Type == condType {
+			job.Status.Conditions[i].Status = condStatus
+			updated = true
+			break
+		}
+	}
+	if !updated {
+		job.Status.Conditions = append(job.Status.Conditions, batchv1.JobCondition{
+			Type:   batchv1.JobComplete,
+			Status: corev1.ConditionTrue,
+		})
+	}
+	return r.client.Status().Update(ctx, job)
+}
+
 // cf. https://go.googlesource.com/proposal/+/refs/heads/master/design/43651-type-parameters.md#pointer-method-example
 type ObjectConstraint[T any] interface {
 	client.Object