feat: introduce new pitr subcommand delete

pitr/cli/internal/cmd/common.go

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cmd
+
+import (
+	"fmt"
+
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg"
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg/model"
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg/xerr"
+)
+
+func validate(ls pkg.ILocalStorage, csn, recordID string) (*model.LsBackup, error) {
+	var (
+		bak *model.LsBackup
+		err error
+	)
+	if CSN != "" {
+		bak, err = ls.ReadByCSN(csn)
+		if err != nil {
+			return bak, xerr.NewCliErr(fmt.Sprintf("read backup record by csn failed. err: %s", err))
+		}
+	}
+
+	if RecordID != "" {
+		bak, err = ls.ReadByID(recordID)
+		if err != nil {
+			return bak, xerr.NewCliErr(fmt.Sprintf("read backup record by id failed. err: %s", err))
+		}
+	}
+	return bak, nil
+}

pitr/cli/internal/cmd/delete.go

@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cmd
+
+import (
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg"
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg/model"
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg/xerr"
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/pkg/logging"
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/pkg/prettyoutput"
+	"github.com/jedib0t/go-pretty/v6/table"
+
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+)
+
+//nolint:dupl
+var DeleteCmd = &cobra.Command{
+	Use:   "delete",
+	Short: "Delete a backup record",
+	Run: func(cmd *cobra.Command, args []string) {
+		cmd.Flags().VisitAll(func(flag *pflag.Flag) {
+			fmt.Printf("Flag: %s Value: %s\n", flag.Name, flag.Value)
+		})
+
+		if CSN == "" && RecordID == "" {
+			logging.Error("Please specify csn or record id")
+			return
+		}
+
+		if CSN != "" && RecordID != "" {
+			logging.Error("Please specify only one of csn and record id")
+			return
+		}
+
+		if err := deleteRecord(); err != nil {
+			logging.Error(err.Error())
+		}
+	},
+}
+
+//nolint:dupl
+func init() {
+	RootCmd.AddCommand(DeleteCmd)
+
+	DeleteCmd.Flags().StringVarP(&Host, "host", "H", "", "ss-proxy hostname or ip")
+	_ = DeleteCmd.MarkFlagRequired("host")
+	DeleteCmd.Flags().Uint16VarP(&Port, "port", "P", 0, "ss-proxy port")
+	_ = DeleteCmd.MarkFlagRequired("port")
+	DeleteCmd.Flags().StringVarP(&Username, "username", "u", "", "ss-proxy username")
+	_ = DeleteCmd.MarkFlagRequired("username")
+	DeleteCmd.Flags().StringVarP(&Password, "password", "p", "", "ss-proxy password")
+	_ = DeleteCmd.MarkFlagRequired("password")
+	DeleteCmd.Flags().StringVarP(&BackupPath, "dn-backup-path", "B", "", "openGauss data backup path")
+	_ = DeleteCmd.MarkFlagRequired("dn-backup-path")
+	DeleteCmd.Flags().Uint16VarP(&AgentPort, "agent-port", "a", 443, "agent server port")
+	_ = DeleteCmd.MarkFlagRequired("agent-port")
+
+	DeleteCmd.Flags().StringVarP(&CSN, "csn", "", "", "commit sequence number")
+	DeleteCmd.Flags().StringVarP(&RecordID, "id", "", "", "backup record id")
+}
+
+func deleteRecord() error {
+	// init local storage
+	ls, err := pkg.NewLocalStorage(pkg.DefaultRootDir())
+	if err != nil {
+		return xerr.NewCliErr(fmt.Sprintf("new local storage failed. err: %s", err.Error()))
+	}
+
+	// get backup record
+	var bak *model.LsBackup
+	bak, err = validate(ls, CSN, RecordID)
+	if err != nil {
+		return err
+	}
+
+	if bak == nil {
+		return xerr.NewCliErr(fmt.Sprintf("backup record not found. err: %s", err))
+	}
+
+	// check agent server status
+	logging.Info("Checking agent server status...")
+	if available := checkAgentServerStatus(bak); !available {
+		return xerr.NewCliErr("one or more agent server are not available.")
+	}
+
+	// mark the target backup record to be deleted
+	// meanwhile this record cannot be restored
+	if err := ls.HideByName(bak.Info.FileName); err != nil {
+		return xerr.NewCliErr("cannot mark backup record.")
+	}
+
+	// exec delete
+	logging.Info("Start delete backup data to openGauss...")
+	if err := _execDelete(bak); err != nil {
+		return xerr.NewCliErr(fmt.Sprintf("exec delete failed. err: %s", err))
+	}
+	logging.Info("Delete backup data success!")
+
+	// delete the backup record
+	if err := ls.DeleteByHidedName(bak.Info.FileName); err != nil {
+		return xerr.NewCliErr(fmt.Sprintf("exec delete backup record failed. err: %s", err))
+	}
+
+	logging.Info("Delete success!")
+	return nil
+}
+
+func _execDelete(lsBackup *model.LsBackup) error {
+	var (
+		dataNodeMap       = make(map[string]*model.DataNode)
+		totalNum          = len(lsBackup.SsBackup.StorageNodes)
+		resultCh          = make(chan *model.DeleteBackupResult, totalNum)
+		dnResult          = make([]*model.DeleteBackupResult, 0)
+		deleteFinalStatus = "Completed"
+	)
+	for _, dn := range lsBackup.DnList {
+		dataNodeMap[dn.IP] = dn
+	}
+
+	if totalNum == 0 {
+		logging.Info("No data node need to delete backup files")
+		return nil
+	}
+
+	pw := prettyoutput.NewPW(totalNum)
+	go pw.Render()
+
+	for _, storagenode := range lsBackup.SsBackup.StorageNodes {
+		sn := storagenode
+		if dn, ok := dataNodeMap[sn.IP]; !ok {
+			logging.Warn(fmt.Sprintf("SKIPPED! data node %s:%d not found in backup info.", sn.IP, sn.Port))
+			continue
+		} else {
+			as := pkg.NewAgentServer(fmt.Sprintf("%s:%d", convertLocalhost(sn.IP), AgentPort))
+			go doDelete(as, sn, dn, resultCh, pw)
+		}
+	}
+
+	time.Sleep(time.Millisecond * 100)
+
+	for pw.IsRenderInProgress() {
+		time.Sleep(time.Millisecond * 100)
+	}
+
+	close(resultCh)
+
+	for result := range resultCh {
+		dnResult = append(dnResult, result)
+		if result.Status != "Completed" {
+			deleteFinalStatus = "Failed"
+		}
+	}
+
+	t := table.NewWriter()
+	t.SetOutputMirror(os.Stdout)
+	t.SetTitle("Delete Backup Files Result")
+	t.AppendHeader(table.Row{"#", "Node IP", "Node Port", "Result", "Message"})
+	t.SetColumnConfigs([]table.ColumnConfig{{Number: 5, WidthMax: 50}})
+
+	for i, result := range dnResult {
+		t.AppendRow([]interface{}{i + 1, result.IP, result.Port, result.Status, result.Msg})
+		t.AppendSeparator()
+	}
+
+	t.Render()
+
+	if deleteFinalStatus == "Failed" {
+		return xerr.NewCliErr("delete failed, please check the log for more details.")
+	}
+
+	return nil
+}

pitr/cli/internal/cmd/delete_test.go

@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cmd
+
+import (
+	"reflect"
+	"time"
+
+	"bou.ke/monkey"
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg"
+	mock_pkg "github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg/mocks"
+	"github.com/apache/shardingsphere-on-cloud/pitr/cli/internal/pkg/model"
+	"github.com/golang/mock/gomock"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("test delete", func() {
+	var (
+		proxy *mock_pkg.MockIShardingSphereProxy
+		ls    *mock_pkg.MockILocalStorage
+		as    *mock_pkg.MockIAgentServer
+		bak   = &model.LsBackup{
+			Info: &model.BackupMetaInfo{
+				ID:       "backup-id-1",
+				FileName: "backup",
+			},
+			DnList: []*model.DataNode{
+				{
+					IP: "127.0.0.1",
+				},
+			},
+			SsBackup: &model.SsBackup{
+				Status: "",
+				ClusterInfo: &model.ClusterInfo{
+					MetaData: model.MetaData{
+						Databases: map[string]string{},
+						Props:     "",
+						Rules:     "",
+					},
+					SnapshotInfo: nil,
+				},
+				StorageNodes: []*model.StorageNode{
+					{
+						IP: "127.0.0.1",
+					},
+				},
+			},
+		}
+	)
+
+	BeforeEach(func() {
+		ctrl = gomock.NewController(GinkgoT())
+		proxy = mock_pkg.NewMockIShardingSphereProxy(ctrl)
+		as = mock_pkg.NewMockIAgentServer(ctrl)
+		ls = mock_pkg.NewMockILocalStorage(ctrl)
+		monkey.Patch(pkg.NewShardingSphereProxy, func(user, password, database, host string, port uint16) (pkg.IShardingSphereProxy, error) {
+			return proxy, nil
+		})
+		monkey.Patch(pkg.NewLocalStorage, func(rootDir string) (pkg.ILocalStorage, error) {
+			return ls, nil
+		})
+	})
+
+	AfterEach(func() {
+		ctrl.Finish()
+		monkey.UnpatchAll()
+	})
+
+	It("test exec delete main func", func() {
+		// patch ReadByID of mock ls
+		monkey.PatchInstanceMethod(reflect.TypeOf(ls), "ReadByID", func(_ *mock_pkg.MockILocalStorage, _ string) (*model.LsBackup, error) { return bak, nil })
+		monkey.Patch(pkg.NewAgentServer, func(_ string) pkg.IAgentServer { return as })
+
+		RecordID = "backup-id"
+		as.EXPECT().CheckStatus(gomock.Any()).Return(nil)
+		ls.EXPECT().HideByName(bak.Info.FileName).Return(nil)
+		as.EXPECT().DeleteBackup(gomock.Any()).Return(nil)
+		ls.EXPECT().DeleteByHidedName(bak.Info.FileName).Return(nil)
+
+		Expect(deleteRecord()).To(BeNil())
+	})
+
+	Context("test exec delete", func() {
+		It("should be success", func() {
+			ctrl := gomock.NewController(GinkgoT())
+			as := mock_pkg.NewMockIAgentServer(ctrl)
+			monkey.Patch(pkg.NewAgentServer, func(_ string) pkg.IAgentServer {
+				return as
+			})
+			defer func() {
+				ctrl.Finish()
+				monkey.UnpatchAll()
+			}()
+			as.EXPECT().DeleteBackup(gomock.Any()).Do(func(_ *model.DeleteBackupIn) {
+				time.Sleep(3 * time.Second)
+			}).Return(nil)
+			Expect(_execDelete(bak)).To(BeNil())
+		})
+	})
+})

pitr/cli/internal/cmd/restore.go

@@ -40,6 +40,7 @@ var (
 	databaseNamesExist []string
 )
 
+//nolint:dupl
 var RestoreCmd = &cobra.Command{
 	Use:   "restore",
 	Short: "Restore a database cluster ",
@@ -64,6 +65,7 @@ var RestoreCmd = &cobra.Command{
 	},
 }
 
+//nolint:dupl
 func init() {
 	RootCmd.AddCommand(RestoreCmd)
 
@@ -97,18 +99,9 @@ func restore() error {
 
 	// get backup record
 	var bak *model.LsBackup
-	if CSN != "" {
-		bak, err = ls.ReadByCSN(CSN)
-		if err != nil {
-			return xerr.NewCliErr(fmt.Sprintf("read backup record by csn failed. err: %s", err))
-		}
-	}
-
-	if RecordID != "" {
-		bak, err = ls.ReadByID(RecordID)
-		if err != nil {
-			return xerr.NewCliErr(fmt.Sprintf("read backup record by id failed. err: %s", err))
-		}
+	bak, err = validate(ls, CSN, RecordID)
+	if err != nil {
+		return err
 	}
 	if bak == nil {
 		return xerr.NewCliErr(fmt.Sprintf("backup record not found. err: %s", err))