Add remove compaction command

Signed-off-by: wayblink <[email protected]>

states/etcd/commands.go

@@ -97,6 +97,8 @@ func RemoveCommand(cli clientv3.KV, instanceName, basePath string) *cobra.Comman
 		remove.EtcdConfigCommand(cli, instanceName),
 		// remove collection has been dropped
 		remove.CollectionCleanCommand(cli, basePath),
+		// remove compaction task
+		remove.CompactionTaskCleanCommand(cli, basePath),
 	)
 
 	return removeCmd

states/etcd/remove/compaction_task.go

@@ -0,0 +1,66 @@
+package remove
+
+import (
+	"context"
+	"fmt"
+	"path"
+
+	"github.com/spf13/cobra"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
+	"github.com/milvus-io/birdwatcher/models"
+	"github.com/milvus-io/birdwatcher/states/etcd/common"
+)
+
+// CompactionTaskCleanCommand returns command to remove
+func CompactionTaskCleanCommand(cli clientv3.KV, basePath string) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "compaction",
+		Short: "Remove compaction task",
+		Run: func(cmd *cobra.Command, args []string) {
+			compactionType, err := cmd.Flags().GetString("type")
+			if err != nil {
+				fmt.Println(err.Error())
+				return
+			}
+			triggerID, err := cmd.Flags().GetString("jobID")
+			if err != nil {
+				fmt.Println(err.Error())
+				return
+			}
+			planID, err := cmd.Flags().GetString("taskID")
+			if err != nil {
+				fmt.Println(err.Error())
+				return
+			}
+
+			compactionTasks, err := common.ListCompactionTask(context.TODO(), cli, basePath, func(task *models.CompactionTask) bool {
+				if compactionType != task.GetType().String() {
+					return false
+				}
+				if triggerID != "" && fmt.Sprint(task.GetTriggerID()) != triggerID {
+					return false
+				}
+				if planID != "" && fmt.Sprint(task.GetPlanID()) != planID {
+					return false
+				}
+				return true
+			})
+
+			for _, task := range compactionTasks {
+				key := path.Join(basePath, common.CompactionTaskPrefix, task.GetType().String(), fmt.Sprint(task.GetTriggerID()), fmt.Sprint(task.GetPlanID()))
+				_, err = cli.Delete(context.TODO(), key, clientv3.WithPrefix())
+				if err != nil {
+					fmt.Println(err.Error())
+					return
+				}
+				fmt.Printf("clean compaction task done, prefix: %s\n", key)
+			}
+		},
+	}
+
+	cmd.Flags().String("type", "ClusteringCompaction", "compaction type")
+	cmd.Flags().String("jobID", "", "jobID also known as triggerID")
+	cmd.Flags().String("taskID", "", "taskID also known as planID")
+	return cmd
+}