Add Distributed Archiver Support for MariaDB

apis/kubedb/constants.go

@@ -62,6 +62,7 @@ const (
 	DistributedDBReplicaENV                    = "DB_REPLICAS"
 	DistributedMaxVolumeUsed                   = "max_used"
 	DistributedVolumeCapacity                  = "capacity"
+	KubesliceContainerExcludeLabel             = "kubeslice.io/exclude"
 
 	KubeSliceNSMIPKey         = "kubeslice.io/nsmIP"
 	KubeSlicePodIPVolumeName  = "podip"
@@ -398,11 +399,37 @@ const (
 	DatabasePodMasterComponent  = "Master"
 	DatabasePodSlaveComponent   = "Slave"
 
-	MariaDBDistributedUpgradeCommand           = "mariadb-upgrade"
-	MariaDBDistributedPodMetricGetCommand      = "get-pod-metrics"
-	MariaDBDistributedPodGetCommand            = "get-pod"
-	MariaDBDistributedVolumeUsageGetCommand    = "get-volume-usage"
-	MariaDBDistributedVolumeCapacityGetCommand = "get-volume-capacity"
+	MariaDBDistributedUpgradeCommand            = "mariadb-upgrade"
+	MariaDBDistributedPodMetricGetCommand       = "get-pod-metrics"
+	MariaDBDistributedPodGetCommand             = "get-pod"
+	MariaDBDistributedVolumeUsageGetCommand     = "get-volume-usage"
+	MariaDBDistributedVolumeCapacityGetCommand  = "get-volume-capacity"
+	MariaDBDistributedBackupCommand             = "take-backup"
+	MariaDBDistributedRestoreCommand            = "restore-backup"
+	MariaDBDistributedRecoveryFileCreateCommand = "create-recovery-done-file"
+	MariaDBArchiverPVCRestorerSuffix            = "pvc-restorer"
+	MariaDBBinlogRestoreSidekickSuffix          = "binlog-restorer"
+	MariaDBBinlogRestoreServiceSuffix           = "binlog-restore"
+	MariaDBXtraBackupInfoFile                   = "/var/lib/mysql/mariadb_backup_binlog_info"
+	MariaDBBackupInfoFile                       = "/var/lib/mysql/xtrabackup_binlog_info"
+	MariaDBArchiverRestoreRecoveryFileName      = "/tmp/recovery.done"
+	MariaDBArchiverBackupJobSelector            = GroupName + "/archiver-job-name"
+	MariaDBSidekickNameLabelKey                 = GroupName + "/sidekick-name"
+	MariaDBArchiverBaseBackupRestic             = "Restic"
+	MariaDBArchiverBaseBackupVolumeSnapshooter  = "VolumeSnapshotter"
+
+	// Distributed Archiver
+	BackupsessionAnnotation                 = "kubestash.com/backupsession"
+	RestoresessionAnnotation                = "kubestash.com/restoresession"
+	BackupconfigurationAnnotation           = "kubestash.com/backupconfiguration"
+	DistributedSnapshotinfoAnnotation       = "kubestash.com/distributedsnapshotinfo"
+	SnapshotsKey                            = "snapshots"
+	RestoreSessionKey                       = "restoresession"
+	DistributedArchiverBackupCMNameSuffix   = "backup"
+	DistributedArchiverRestoreCMNameSuffix  = "restore"
+	DistributedArchiverSnapshotCMNameSuffix = "snapshots"
+	MariaDBKubestashBackupContainerName     = "physical-backup-1"
+	MariaDBKubestashRestoreContainerName    = "distributed-physical-backup-restore-0"
 
 	// Maxscale
 	MaxscaleCommonName            = "mx"

apis/kubedb/v1/mariadb_helpers.go

@@ -114,6 +114,18 @@ func (m MariaDB) OffshootMaxscaleLabels() map[string]string {
 	return m.offshootLabels(m.OffshootMaxscaleSelectors(), nil)
 }
 
+func (m MariaDB) GetPVCRestoreSessionName(ordinal int) string {
+	return fmt.Sprintf("%s-%s-%v-%s", kubedb.DefaultVolumeClaimTemplateName, m.OffshootName(), ordinal, kubedb.MariaDBArchiverPVCRestorerSuffix)
+}
+
+func (m MariaDB) GetBinlogRestoreSidekickName(ordinal int) string {
+	return fmt.Sprintf("%s-%s-%d", m.OffshootName(), kubedb.MariaDBBinlogRestoreSidekickSuffix, ordinal)
+}
+
+func (m MariaDB) GetBinlogRestoreServiceName(ordinal int) string {
+	return fmt.Sprintf("%s-%s-%d", m.OffshootName(), kubedb.MariaDBBinlogRestoreServiceSuffix, ordinal)
+}
+
 func (m MariaDB) PodLabels() map[string]string {
 	return m.offshootLabels(m.OffshootSelectors(), m.Spec.PodTemplate.Labels)
 }

pkg/utils/grpc/mariadb/client.go

@@ -104,6 +104,25 @@ func RunCommand(grpcClient pb.CommandServiceClient, cmd string) ([]byte, error)
 	return resp.Output, nil
 }
 
+func RunCommandWithPayload(grpcClient pb.CommandServiceClient, cmd string, data []byte) ([]byte, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second)
+	defer cancel()
+
+	req := &pb.CommandRequest{
+		Command: cmd,
+		Data:    data,
+	}
+	resp, err := grpcClient.ExecuteCommand(ctx, req)
+	if err != nil {
+		klog.Infof("failed to execute command: %v", err)
+		return nil, err
+	}
+	if resp.Status != "success" {
+		return nil, fmt.Errorf("failed to execute command: %s, Output: %s, err: %v", cmd, string(resp.Output), resp.Error)
+	}
+	return resp.Output, nil
+}
+
 func serverAddress(db *dbapi.MariaDB, podName string) string {
 	return fmt.Sprintf("%s.%s.%s.svc:%v", podName, db.GoverningServiceName(), db.Namespace, port)
 }

pkg/utils/grpc/mariadb/proto/api.proto

@@ -1,7 +1,7 @@
 syntax="proto3";
 
 package proto;
-option go_package = "grpc/protogen";
+option go_package = "pkg/utils/grpc/mariadb/protogen";
 
 // CommandService provides a method to execute commands in the pod.
 service CommandService {
@@ -12,6 +12,7 @@ service CommandService {
 // CommandRequest defines the command to execute and authentication key.
 message CommandRequest {
   string command = 1; // The command to run (e.g., "cat /scripts/seqno").
+  bytes data = 2;  // backup job.
 }
 
 // CommandResponse contains the command output or error.