Fix metanode request idempotence problem

cli/cmd/const.go

@@ -112,7 +112,7 @@ const (
 	CliFlagForceInode          = "forceInode"
 	CliFlagEnableQuota         = "enableQuota"
 	CliFlagDeleteLockTime      = "delete-lock-time"
-
+	CliFlagEnableRemoveDupReq  = "enableRemoveDupReq"
 	//CliFlagSetDataPartitionCount	= "count" use dp-count instead
 
 	//Shorthand format of resource name

cli/cmd/fmt.go

@@ -141,6 +141,7 @@ func formatSimpleVolView(svv *proto.SimpleVolView) string {
 	if svv.Forbidden && svv.Status == 1 {
 		sb.WriteString(fmt.Sprintf("  DeleteDelayTime                 : %v\n", svv.DeleteExecTime.Sub(time.Now())))
 	}
+	sb.WriteString(fmt.Sprintf("  RemoveDupReq                    : %v\n", formatEnabledDisabled(svv.EnableRemoveDupReq)))
 	if svv.VolType == 1 {
 		sb.WriteString(fmt.Sprintf("  ObjBlockSize         : %v byte\n", svv.ObjBlockSize))
 		sb.WriteString(fmt.Sprintf("  CacheCapacity        : %v G\n", svv.CacheCapacity))

cli/cmd/vol.go

@@ -141,6 +141,7 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
 	var optTxConflictRetryInterval int64
 	var optDeleteLockTime int64
 	var optYes bool
+	var optEnableRemoveDupReq string
 	var cmd = &cobra.Command{
 		Use:   cmdVolCreateUse,
 		Short: cmdVolCreateShort,
@@ -173,6 +174,9 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
 			if optDeleteLockTime < 0 {
 				optDeleteLockTime = 0
 			}
+			if optEnableRemoveDupReq != "yes" {
+				optEnableRemoveDupReq = "false"
+			}
 
 			// ask user for confirm
 			if !optYes {
@@ -204,6 +208,8 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
 				stdout("  TransactionTimeout       : %v min\n", optTxTimeout)
 				stdout("  TxConflictRetryNum       : %v\n", optTxConflictRetryNum)
 				stdout("  TxConflictRetryInterval  : %v ms\n", optTxConflictRetryInterval)
+				stdout("  EnableQuota              : %v\n", optEnableQuota)
+				stdout("  EnableRemoveDupReq          : %v\n", optEnableRemoveDupReq)
 				stdout("\nConfirm (yes/no)[yes]: ")
 				var userConfirm string
 				_, _ = fmt.Scanln(&userConfirm)
@@ -219,7 +225,7 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
 				optZoneName, optCacheRuleKey, optEbsBlkSize, optCacheCap,
 				optCacheAction, optCacheThreshold, optCacheTTL, optCacheHighWater,
 				optCacheLowWater, optCacheLRUInterval, dpReadOnlyWhenVolFull,
-				optTxMask, optTxTimeout, optTxConflictRetryNum, optTxConflictRetryInterval, optEnableQuota)
+				optTxMask, optTxTimeout, optTxConflictRetryNum, optTxConflictRetryInterval, optEnableQuota, optEnableRemoveDupReq)
 			if err != nil {
 				err = fmt.Errorf("Create volume failed case:\n%v\n", err)
 				return
@@ -256,6 +262,7 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
 	cmd.Flags().Int64Var(&optTxConflictRetryInterval, CliTxConflictRetryInterval, 0, "Specify retry interval[Unit: ms] for transaction conflict [10-1000]")
 	cmd.Flags().StringVar(&optEnableQuota, CliFlagEnableQuota, "false", "Enable quota (default false)")
 	cmd.Flags().Int64Var(&optDeleteLockTime, CliFlagDeleteLockTime, 0, "Specify delete lock time[Unit: hour] for volume")
+	cmd.Flags().StringVar(&optEnableRemoveDupReq, CliFlagEnableRemoveDupReq, "false", "Enable remove duplicate (default false)")
 
 	return cmd
 }
@@ -292,6 +299,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
 	var optEnableQuota string
 	var confirmString = strings.Builder{}
 	var vv *proto.SimpleVolView
+	var optEnableRemoveDupReq string
 	var cmd = &cobra.Command{
 		Use:   CliOpUpdate + " [VOLUME NAME]",
 		Short: cmdVolUpdateShort,
@@ -577,6 +585,23 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
 					formatEnabledDisabled(vv.DpReadOnlyWhenVolFull)))
 			}
 
+			if optEnableRemoveDupReq != "" {
+				var enable bool
+				if enable, err = strconv.ParseBool(optEnableRemoveDupReq); err != nil {
+					return
+				}
+
+				if vv.EnableRemoveDupReq != enable {
+					isChange = true
+					confirmString.WriteString(fmt.Sprintf("  Remove dup req      : %v -> %v\n", vv.EnableRemoveDupReq, enable))
+					vv.EnableRemoveDupReq = enable
+				} else {
+					confirmString.WriteString(fmt.Sprintf("  Remove dup req      : %v\n", vv.EnableRemoveDupReq))
+				}
+			} else {
+				confirmString.WriteString(fmt.Sprintf("  Remove dup req      : %v\n", vv.EnableRemoveDupReq))
+			}
+
 			if err != nil {
 				return
 			}
@@ -634,7 +659,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
 	cmd.Flags().StringVar(&optReplicaNum, CliFlagReplicaNum, "", "Specify data partition replicas number(default 3 for normal volume,1 for low volume)")
 	cmd.Flags().StringVar(&optEnableQuota, CliFlagEnableQuota, "", "Enable quota")
 	cmd.Flags().Int64Var(&optDeleteLockTime, CliFlagDeleteLockTime, -1, "Specify delete lock time[Unit: hour] for volume")
-
+	cmd.Flags().StringVar(&optEnableRemoveDupReq, CliFlagEnableRemoveDupReq, "", "Enable remove dup")
 	return cmd
 
 }

client/fs/file.go

@@ -643,7 +643,7 @@ func (f *File) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *fu
 		return ParseError(err)
 	}
 	value := info.Get(name)
-	if pos > 0 {
+	if pos > 0 && pos < uint32(len(value)) {
 		value = value[pos:]
 	}
 	if size > 0 && size < uint32(len(value)) {

master/api_args_parse.go

@@ -364,6 +364,7 @@ type updateVolReq struct {
 	coldArgs                *coldVolArgs
 	dpReadOnlyWhenVolFull   bool
 	enableQuota             bool
+	enableRemoveDupReq      bool
 }
 
 func parseColdVolUpdateArgs(r *http.Request, vol *Vol) (args *coldVolArgs, err error) {
@@ -516,6 +517,10 @@ func parseVolUpdateReq(r *http.Request, vol *Vol, req *updateVolReq) (err error)
 		}
 	}
 
+	if req.enableRemoveDupReq, err = extractBoolWithDefault(r, proto.EnableRemoveDupReq, vol.enableRemoveDupReq); err != nil {
+		return
+	}
+
 	return
 }
 
@@ -628,6 +633,7 @@ type createVolReq struct {
 	DpReadOnlyWhenVolFull                bool
 	enableTransaction                    proto.TxOpMask
 	enableQuota                          bool
+	enableRemoveDupReq                   bool
 	txTimeout                            int64
 	txConflictRetryNum                   int64
 	txConflictRetryInterval              int64
@@ -801,6 +807,9 @@ func parseRequestToCreateVol(r *http.Request, req *createVolReq) (err error) {
 		return
 	}
 
+	if req.enableRemoveDupReq, err = extractBoolWithDefault(r, proto.EnableRemoveDupReq, false); err != nil {
+		return
+	}
 	return
 }
 

master/api_service.go

@@ -2035,6 +2035,7 @@ func (m *Server) updateVol(w http.ResponseWriter, r *http.Request) {
 	newArgs.txConflictRetryInterval = req.txConflictRetryInterval
 	newArgs.txOpLimit = req.txOpLimit
 	newArgs.enableQuota = req.enableQuota
+	newArgs.enableRemoveDupReq = req.enableRemoveDupReq
 	if req.coldArgs != nil {
 		newArgs.coldArgs = req.coldArgs
 	}
@@ -2420,6 +2421,7 @@ func newSimpleView(vol *Vol) (view *proto.SimpleVolView) {
 		TrashInterval:           vol.TrashInterval,
 		Forbidden:               vol.Forbidden,
 		DisableAuditLog:         vol.DisableAuditLog,
+		EnableRemoveDupReq:      vol.enableRemoveDupReq,
 		DeleteExecTime:          vol.DeleteExecTime,
 	}
 
@@ -4609,7 +4611,7 @@ func (m *Server) listVols(w http.ResponseWriter, r *http.Request) {
 			}
 			stat := volStat(vol, false)
 			volInfo := proto.NewVolInfo(vol.Name, vol.Owner, vol.createTime, vol.status(), stat.TotalSize,
-				stat.UsedSize, stat.DpReadOnlyWhenVolFull)
+				stat.UsedSize, stat.DpReadOnlyWhenVolFull, vol.enableRemoveDupReq)
 			volsInfo = append(volsInfo, volInfo)
 		}
 	}

master/api_service_test.go

@@ -124,7 +124,6 @@ func createDefaultMasterServerForTest() *Server {
 	}`
 
 	testServer, err := createMasterServer(cfgJSON)
-	testServer.cluster.cfg.volForceDeletion = true
 
 	if err != nil {
 		panic(err)

master/cluster.go

@@ -721,6 +721,11 @@ func (c *Cluster) checkMetaNodeHeartbeat() {
 				Mask:       vol.enableTransaction,
 				OpLimitVal: vol.txOpLimit,
 			})
+
+			hbReq.RemoveDupReqInfo = append(hbReq.RemoveDupReqInfo, &proto.RemoveDupReqInfo{
+				Volume: vol.Name,
+				Enable: vol.enableRemoveDupReq,
+			})
 		}
 		log.LogDebugf("checkMetaNodeHeartbeat start")
 		for _, info := range hbReq.QuotaHbInfos {
@@ -3111,6 +3116,7 @@ func (c *Cluster) doCreateVol(req *createVolReq) (vol *Vol, err error) {
 		FlowWlimit:   req.qosLimitArgs.flowWVal,
 
 		DpReadOnlyWhenVolFull: req.DpReadOnlyWhenVolFull,
+		EnableRemoveDupReq:    req.enableRemoveDupReq,
 	}
 
 	log.LogInfof("[doCreateVol] volView, %v", vv)
@@ -3894,7 +3900,7 @@ func (c *Cluster) listQuotaAll() (volsInfo []*proto.VolInfo) {
 		if vol.quotaManager.HasQuota() {
 			stat := volStat(vol, false)
 			volInfo := proto.NewVolInfo(vol.Name, vol.Owner, vol.createTime, vol.status(), stat.TotalSize,
-				stat.UsedSize, stat.DpReadOnlyWhenVolFull)
+				stat.UsedSize, stat.DpReadOnlyWhenVolFull, vol.enableRemoveDupReq)
 			volsInfo = append(volsInfo, volInfo)
 		}
 	}

master/gapi_volume.go

@@ -322,6 +322,7 @@ func (s *VolumeService) updateVolume(ctx context.Context, args struct {
 	if args.Description != nil {
 		newArgs.description = *args.Description
 	}
+	newArgs.enableRemoveDupReq = false
 
 	if err = s.cluster.updateVol(args.Name, args.AuthKey, newArgs); err != nil {
 		return nil, err

master/metadata_fsm_op.go

@@ -274,6 +274,7 @@ type volValue struct {
 	TrashInterval                                          int64
 	Forbidden                                              bool
 	DisableAuditLog                                        bool
+	EnableRemoveDupReq                                     bool
 }
 
 func (v *volValue) Bytes() (raw []byte, err error) {
@@ -341,6 +342,7 @@ func newVolValue(vol *Vol) (vv *volValue) {
 		AuthKey:               vol.authKey,
 		DeleteExecTime:        vol.DeleteExecTime,
 		User:                  vol.user,
+		EnableRemoveDupReq:    vol.enableRemoveDupReq,
 	}
 
 	return