wip

backup/restore.go

@@ -31,22 +31,23 @@ import (
 )
 
 type RestoreManager struct {
- log logr.Logger
- client client.Client
- scheme *runtime.Scheme
- namespace string
- name string
- password string
- threads int
- bucket bucket.Bucket
- keyPrefix string
- restorePoint time.Time
- workDir string
+ log logr.Logger
+ client client.Client
+ scheme *runtime.Scheme
+ namespace string
+ name string
+ password string
+ threads int
+ bucket bucket.Bucket
+ keyPrefix string
+ restorePoint time.Time
+ workDir string
+ includeSchemas []string
 }
 
 var ErrBadConnection = errors.New("the connection hasn't reflected the latest user's privileges")
 
-func NewRestoreManager(cfg *rest.Config, bc bucket.Bucket, dir, srcNS, srcName, ns, name, password string, threads int, restorePoint time.Time) (*RestoreManager, error) {
+func NewRestoreManager(cfg *rest.Config, bc bucket.Bucket, dir, srcNS, srcName, ns, name, password string, threads int, restorePoint time.Time, includeSchemas []string) (*RestoreManager, error) {
  log := zap.New(zap.WriteTo(os.Stderr), zap.StacktraceLevel(zapcore.DPanicLevel))
  scheme := runtime.NewScheme()
  if err := clientgoscheme.AddToScheme(scheme); err != nil {
@@ -63,17 +64,18 @@ func NewRestoreManager(cfg *rest.Config, bc bucket.Bucket, dir, srcNS, srcName,
 
  prefix := calcPrefix(srcNS, srcName)
  return &RestoreManager{
- log: log,
- client: k8sClient,
- scheme: scheme,
- namespace: ns,
- name: name,
- password: password,
- threads: threads,
- bucket: bc,
- keyPrefix: prefix,
- restorePoint: restorePoint,
- workDir: dir,
+ log: log,
+ client: k8sClient,
+ scheme: scheme,
+ namespace: ns,
+ name: name,
+ password: password,
+ threads: threads,
+ bucket: bc,
+ keyPrefix: prefix,
+ restorePoint: restorePoint,
+ workDir: dir,
+ includeSchemas: includeSchemas,
  }, nil
 }
 
@@ -254,7 +256,7 @@ func (rm *RestoreManager) loadDump(ctx context.Context, op bkop.Operator, key st
  return fmt.Errorf("failed to untar dump file: %w", err)
  }
 
- return op.LoadDump(ctx, dumpDir)
+ return op.LoadDump(ctx, dumpDir, rm.includeSchemas)
 }
 
 func (rm *RestoreManager) applyBinlog(ctx context.Context, op bkop.Operator, key string) error {

cmd/moco-backup/cmd/restore.go

@@ -12,7 +12,7 @@ import (
 )
 
 var restoreCmd = &cobra.Command{
- Use: "restore BUCKET SOURCE_NAMESPACE SOURCE_NAME NAMESPACE NAME YYYYMMDD-hhmmss",
+ Use: "restore BUCKET SOURCE_NAMESPACE SOURCE_NAME NAMESPACE NAME YYYYMMDD-hhmmss SCHEMAS",
  Short: "restore MySQL data from a backup",
  Long: `Restore MySQL data from a backup.
 
@@ -21,8 +21,10 @@ SOURCE_NAMESPACE: The source MySQLCluster's namespace.
 SOURCE_NAME: The source MySQLCluster's name.
 NAMESPACE: The target MySQLCluster's namespace.
 NAME: The target MySQLCluster's name.
-YYYYMMDD-hhmmss: The point-in-time to restore data. e.g. 20210523-150423`,
- Args: cobra.ExactArgs(6),
+YYYYMMDD-hhmmss: The point-in-time to restore data. e.g. 20210523-150423
+SCHEMAS: The list of schemas to restore. e.g. "db1 db2 db3"`,
+
+ Args: cobra.ExactArgs(7),
  RunE: func(cmd *cobra.Command, args []string) error {
  maxRetry := 3
  for i := 0; i < maxRetry; i++ {
@@ -54,6 +56,7 @@ func runRestore(cmd *cobra.Command, args []string) (e error) {
  srcName := args[2]
  namespace := args[3]
  name := args[4]
+ includeSchemas := strings.Split(args[6], " ")
 
  restorePoint, err := time.Parse(constants.BackupTimeFormat, args[5])
  if err != nil {
@@ -75,7 +78,8 @@ func runRestore(cmd *cobra.Command, args []string) (e error) {
  namespace, name,
  mysqlPassword,
  commonArgs.threads,
- restorePoint)
+ restorePoint,
+ includeSchemas)
  if err != nil {
  return fmt.Errorf("failed to create a restore manager: %w", err)
  }

controllers/mysqlcluster_controller.go

@@ -1448,6 +1448,7 @@ func (r *MySQLClusterReconciler) reconcileV1RestoreJob(ctx context.Context, req
  args = append(args, cluster.Spec.Restore.SourceNamespace, cluster.Spec.Restore.SourceName)
  args = append(args, cluster.Namespace, cluster.Name)
  args = append(args, cluster.Spec.Restore.RestorePoint.UTC().Format(constants.BackupTimeFormat))
+ args = append(args, strings.Join(cluster.Spec.Restore.IncludeSchemas, " "))
 
  resources := corev1ac.ResourceRequirements()
  if !noJobResource {

pkg/bkop/operator.go

@@ -37,7 +37,7 @@ type Operator interface {
  PrepareRestore(context.Context) error
 
  // LoadDump loads data dumped by `DumpFull`.
- LoadDump(ctx context.Context, dir string) error
+ LoadDump(ctx context.Context, dir string, schemas []string) error
 
  // LoadBinLog applies binary logs up to `restorePoint`.
  LoadBinlog(ctx context.Context, binlogDir, tmpDir string, restorePoint time.Time) error

pkg/bkop/restore.go

@@ -21,7 +21,7 @@ func (o operator) PrepareRestore(ctx context.Context) error {
  return nil
 }
 
-func (o operator) LoadDump(ctx context.Context, dir string) error {
+func (o operator) LoadDump(ctx context.Context, dir string, schemas []string) error {
  args := []string{
  fmt.Sprintf("mysql://%s@%s", o.user, net.JoinHostPort(o.host, fmt.Sprint(o.port))),
  "-p" + o.password,
@@ -38,6 +38,9 @@ func (o operator) LoadDump(ctx context.Context, dir string) error {
  "--deferTableIndexes=all",
  "--updateGtidSet=replace",
  }
+ if len(schemas) > 0 {
+ args = append(args, "--includeSchemas="+strings.Join(schemas, ","))
+ }
 
  cmd := exec.CommandContext(ctx, "mysqlsh", args...)
  cmd.Stdout = os.Stdout