[#1350] node: Add ability to evacuate objects from REP 1 only

Signed-off-by: Anton Nikiforov <an.nikiforov@yadro.com>

cmd/frostfs-cli/modules/control/evacuate_shard.go

@@ -24,6 +24,7 @@ func evacuateShard(cmd *cobra.Command, _ []string) {
 	req := &control.EvacuateShardRequest{Body: new(control.EvacuateShardRequest_Body)}
 	req.Body.Shard_ID = getShardIDList(cmd)
 	req.Body.IgnoreErrors, _ = cmd.Flags().GetBool(ignoreErrorsFlag)
+	req.Body.RepOneOnly, _ = cmd.Flags().GetBool(repOneOnlyFlag)
 
 	signRequest(cmd, pk, req)
 

cmd/frostfs-cli/modules/control/evacuation.go

@@ -20,6 +20,7 @@ const (
 	awaitFlag      = "await"
 	noProgressFlag = "no-progress"
 	scopeFlag      = "scope"
+	repOneOnlyFlag = "rep-one-only"
 
 	scopeAll     = "all"
 	scopeObjects = "objects"
@@ -64,12 +65,14 @@ func startEvacuateShard(cmd *cobra.Command, _ []string) {
 	pk := key.Get(cmd)
 
 	ignoreErrors, _ := cmd.Flags().GetBool(ignoreErrorsFlag)
+	repOneOnly, _ := cmd.Flags().GetBool(repOneOnlyFlag)
 
 	req := &control.StartShardEvacuationRequest{
 		Body: &control.StartShardEvacuationRequest_Body{
 			Shard_ID:     getShardIDList(cmd),
 			IgnoreErrors: ignoreErrors,
 			Scope:        getEvacuationScope(cmd),
+			RepOneOnly:   repOneOnly,
 		},
 	}
 
@@ -371,6 +374,7 @@ func initControlStartEvacuationShardCmd() {
 	flags.String(scopeFlag, scopeAll, fmt.Sprintf("Evacuation scope; possible values: %s, %s, %s", scopeTrees, scopeObjects, scopeAll))
 	flags.Bool(awaitFlag, false, "Block execution until evacuation is completed")
 	flags.Bool(noProgressFlag, false, fmt.Sprintf("Print progress if %s provided", awaitFlag))
+	flags.Bool(repOneOnlyFlag, false, "Evacuate objects only from containers with policy `REP 1 ...`")
 
 	startEvacuationShardCmd.MarkFlagsMutuallyExclusive(shardIDFlag, shardAllFlag)
 }

pkg/local_object_storage/engine/evacuate.go

@@ -79,6 +79,7 @@ type EvacuateShardPrm struct {
 	IgnoreErrors   bool
 	Async          bool
 	Scope          EvacuateScope
+	RepOneOnly     bool
 }
 
 // EvacuateShardRes represents result of the EvacuateShard operation.
@@ -270,6 +271,7 @@ func (e *StorageEngine) evacuateShards(ctx context.Context, shardIDs []string, p
 			attribute.Bool("async", prm.Async),
 			attribute.Bool("ignoreErrors", prm.IgnoreErrors),
 			attribute.Stringer("scope", prm.Scope),
+			attribute.Bool("repOneOnly", prm.RepOneOnly),
 		))
 
 	defer func() {
@@ -653,7 +655,16 @@ func (e *StorageEngine) evacuateObjects(ctx context.Context, sh *shard.Shard, to
 		default:
 		}
 		addr := toEvacuate[i].Address
-
+		if prm.RepOneOnly {
+			repOne, err := e.isRepOne(addr.Container())
+			if err != nil {
+				return err
+			}
+			if !repOne {
+				res.objSkipped.Add(1)
+				continue
+			}
+		}
 		var getPrm shard.GetPrm
 		getPrm.SetAddress(addr)
 		getPrm.SkipEvacCheck(true)
@@ -703,6 +714,20 @@ func (e *StorageEngine) evacuateObjects(ctx context.Context, sh *shard.Shard, to
 	return nil
 }
 
+func (e *StorageEngine) isRepOne(cid cid.ID) (bool, error) {
+	c, err := e.containerSource.Load().cs.Get(cid)
+	if err != nil {
+		return false, err
+	}
+	p := c.Value.PlacementPolicy()
+	for i := 0; i < p.NumberOfReplicas(); i++ {
+		if p.ReplicaDescriptor(i).NumberOfObjects() == 1 {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 func (e *StorageEngine) tryEvacuateObjectLocal(ctx context.Context, addr oid.Address, object *objectSDK.Object, sh *shard.Shard,
 	shards []pooledShard, shardsToEvacuate map[string]*shard.Shard, res *EvacuateShardRes,
 ) (bool, error) {

pkg/local_object_storage/engine/evacuate_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	coreContainer "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
 	objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/fstree"
@@ -18,14 +19,31 @@ import (
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger/test"
+	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
 	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
 	cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test"
+	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
 	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
 	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
 )
 
+type containerStorage struct {
+	cntmap map[cid.ID]*container.Container
+}
+
+func (cs *containerStorage) Get(id cid.ID) (*coreContainer.Container, error) {
+	coreCnt := coreContainer.Container{
+		Value: *cs.cntmap[id],
+	}
+	return &coreCnt, nil
+}
+
+func (cs *containerStorage) DeletionInfo(cid.ID) (*coreContainer.DelInfo, error) {
+	return nil, nil
+}
+
 func newEngineEvacuate(t *testing.T, shardNum int, objPerShard int) (*StorageEngine, []*shard.ID, []*objectSDK.Object) {
 	dir := t.TempDir()
 
@@ -605,3 +623,67 @@ func TestEvacuateTreesRemote(t *testing.T) {
 
 	require.Equal(t, expectedTreeOps, evacuatedTreeOps)
 }
+
+func TestEvacuateShardObjectsRepOneOnly(t *testing.T) {
+	e, ids, _ := newEngineEvacuate(t, 2, 0)
+	defer func() {
+		require.NoError(t, e.Close(context.Background()))
+	}()
+
+	// Create container with policy REP 2
+	cnr1 := container.Container{}
+	p1 := netmap.PlacementPolicy{}
+	p1.SetContainerBackupFactor(1)
+	x1 := netmap.ReplicaDescriptor{}
+	x1.SetNumberOfObjects(2)
+	p1.AddReplicas(x1)
+	cnr1.SetPlacementPolicy(p1)
+
+	var idCnr1 cid.ID
+	container.CalculateID(&idCnr1, cnr1)
+
+	cnrmap := make(map[cid.ID]*container.Container)
+	var cids []cid.ID
+	cnrmap[idCnr1] = &cnr1
+	cids = append(cids, idCnr1)
+
+	// Create container with policy REP 1
+	cnr2 := container.Container{}
+	p2 := netmap.PlacementPolicy{}
+	p2.SetContainerBackupFactor(1)
+	x2 := netmap.ReplicaDescriptor{}
+	x2.SetNumberOfObjects(1)
+	p2.AddReplicas(x2)
+	cnr2.SetPlacementPolicy(p2)
+
+	var idCnr2 cid.ID
+	container.CalculateID(&idCnr2, cnr2)
+	cnrmap[idCnr2] = &cnr2
+	cids = append(cids, idCnr2)
+
+	e.SetContainerSource(&containerStorage{cntmap: cnrmap})
+
+	for _, sh := range ids {
+		for i := 0; i < 4; i++ {
+			obj := testutil.GenerateObjectWithCID(cids[i%2])
+
+			var putPrm shard.PutPrm
+			putPrm.SetObject(obj)
+			_, err := e.shards[sh.String()].Put(context.Background(), putPrm)
+			require.NoError(t, err)
+		}
+	}
+
+	var prm EvacuateShardPrm
+	prm.ShardID = ids[0:1]
+	prm.Scope = EvacuateScopeObjects
+	prm.RepOneOnly = true
+
+	require.NoError(t, e.shards[ids[0].String()].SetMode(mode.ReadOnly))
+
+	res, err := e.Evacuate(context.Background(), prm)
+	require.NoError(t, err)
+	require.Equal(t, uint64(2), res.ObjectsEvacuated())
+	require.Equal(t, uint64(2), res.ObjectsSkipped())
+	require.Equal(t, uint64(0), res.ObjectsFailed())
+}

pkg/services/control/server/evacuate.go

@@ -37,6 +37,7 @@ func (s *Server) EvacuateShard(ctx context.Context, req *control.EvacuateShardRe
 		IgnoreErrors:   req.GetBody().GetIgnoreErrors(),
 		ObjectsHandler: s.replicateObject,
 		Scope:          engine.EvacuateScopeObjects,
+		RepOneOnly:     req.GetBody().GetRepOneOnly(),
 	}
 
 	res, err := s.s.Evacuate(ctx, prm)

pkg/services/control/server/evacuate_async.go

@@ -29,6 +29,7 @@ func (s *Server) StartShardEvacuation(ctx context.Context, req *control.StartSha
 		TreeHandler:    s.replicateTree,
 		Async:          true,
 		Scope:          engine.EvacuateScope(req.GetBody().GetScope()),
+		RepOneOnly:     req.GetBody().GetRepOneOnly(),
 	}
 
 	_, err = s.s.Evacuate(ctx, prm)

pkg/services/control/service.proto

@@ -313,6 +313,9 @@ message EvacuateShardRequest {
 
     // Flag indicating whether object read errors should be ignored.
     bool ignore_errors = 2;
+
+    // Choose for evacuation objects in `REP 1` containers only.
+    bool rep_one_only = 3;
   }
 
   Body body = 1;
@@ -391,6 +394,8 @@ message StartShardEvacuationRequest {
     bool ignore_errors = 2;
     // Evacuation scope.
     uint32 scope = 3;
+    // Choose for evacuation objects in `REP 1` containers only.
+    bool rep_one_only = 4;
   }
 
   Body body = 1;

pkg/services/control/service_frostfs.pb.go

@@ -4535,6 +4535,7 @@ func (x *SynchronizeTreeResponse) UnmarshalEasyJSON(in *jlexer.Lexer) {
 type EvacuateShardRequest_Body struct {
 	Shard_ID     [][]byte `json:"shardID"`
 	IgnoreErrors bool     `json:"ignoreErrors"`
+	RepOneOnly   bool     `json:"repOneOnly"`
 }
 
 var (
@@ -4553,6 +4554,7 @@ func (x *EvacuateShardRequest_Body) StableSize() (size int) {
 	}
 	size += proto.RepeatedBytesSize(1, x.Shard_ID)
 	size += proto.BoolSize(2, x.IgnoreErrors)
+	size += proto.BoolSize(3, x.RepOneOnly)
 	return size
 }
 
@@ -4575,6 +4577,9 @@ func (x *EvacuateShardRequest_Body) EmitProtobuf(mm *easyproto.MessageMarshaler)
 	if x.IgnoreErrors {
 		mm.AppendBool(2, x.IgnoreErrors)
 	}
+	if x.RepOneOnly {
+		mm.AppendBool(3, x.RepOneOnly)
+	}
 }
 
 // UnmarshalProtobuf implements the encoding.ProtoUnmarshaler interface.
@@ -4598,6 +4603,12 @@ func (x *EvacuateShardRequest_Body) UnmarshalProtobuf(src []byte) (err error) {
 				return fmt.Errorf("cannot unmarshal field %s", "IgnoreErrors")
 			}
 			x.IgnoreErrors = data
+		case 3: // RepOneOnly
+			data, ok := fc.Bool()
+			if !ok {
+				return fmt.Errorf("cannot unmarshal field %s", "RepOneOnly")
+			}
+			x.RepOneOnly = data
 		}
 	}
 	return nil
@@ -4620,6 +4631,15 @@ func (x *EvacuateShardRequest_Body) GetIgnoreErrors() bool {
 func (x *EvacuateShardRequest_Body) SetIgnoreErrors(v bool) {
 	x.IgnoreErrors = v
 }
+func (x *EvacuateShardRequest_Body) GetRepOneOnly() bool {
+	if x != nil {
+		return x.RepOneOnly
+	}
+	return false
+}
+func (x *EvacuateShardRequest_Body) SetRepOneOnly(v bool) {
+	x.RepOneOnly = v
+}
 
 // MarshalJSON implements the json.Marshaler interface.
 func (x *EvacuateShardRequest_Body) MarshalJSON() ([]byte, error) {
@@ -4650,6 +4670,11 @@ func (x *EvacuateShardRequest_Body) MarshalEasyJSON(out *jwriter.Writer) {
 		out.RawString(prefix)
 		out.Bool(x.IgnoreErrors)
 	}
+	{
+		const prefix string = ",\"repOneOnly\":"
+		out.RawString(prefix)
+		out.Bool(x.RepOneOnly)
+	}
 	out.RawByte('}')
 }
 
@@ -4697,6 +4722,12 @@ func (x *EvacuateShardRequest_Body) UnmarshalEasyJSON(in *jlexer.Lexer) {
 				f = in.Bool()
 				x.IgnoreErrors = f
 			}
+		case "repOneOnly":
+			{
+				var f bool
+				f = in.Bool()
+				x.RepOneOnly = f
+			}
 		}
 		in.WantComma()
 	}
@@ -6514,6 +6545,7 @@ type StartShardEvacuationRequest_Body struct {
 	Shard_ID     [][]byte `json:"shardID"`
 	IgnoreErrors bool     `json:"ignoreErrors"`
 	Scope        uint32   `json:"scope"`
+	RepOneOnly   bool     `json:"repOneOnly"`
 }
 
 var (
@@ -6533,6 +6565,7 @@ func (x *StartShardEvacuationRequest_Body) StableSize() (size int) {
 	size += proto.RepeatedBytesSize(1, x.Shard_ID)
 	size += proto.BoolSize(2, x.IgnoreErrors)
 	size += proto.UInt32Size(3, x.Scope)
+	size += proto.BoolSize(4, x.RepOneOnly)
 	return size
 }
 
@@ -6558,6 +6591,9 @@ func (x *StartShardEvacuationRequest_Body) EmitProtobuf(mm *easyproto.MessageMar
 	if x.Scope != 0 {
 		mm.AppendUint32(3, x.Scope)
 	}
+	if x.RepOneOnly {
+		mm.AppendBool(4, x.RepOneOnly)
+	}
 }
 
 // UnmarshalProtobuf implements the encoding.ProtoUnmarshaler interface.
@@ -6587,6 +6623,12 @@ func (x *StartShardEvacuationRequest_Body) UnmarshalProtobuf(src []byte) (err er
 				return fmt.Errorf("cannot unmarshal field %s", "Scope")
 			}
 			x.Scope = data
+		case 4: // RepOneOnly
+			data, ok := fc.Bool()
+			if !ok {
+				return fmt.Errorf("cannot unmarshal field %s", "RepOneOnly")
+			}
+			x.RepOneOnly = data
 		}
 	}
 	return nil
@@ -6618,6 +6660,15 @@ func (x *StartShardEvacuationRequest_Body) GetScope() uint32 {
 func (x *StartShardEvacuationRequest_Body) SetScope(v uint32) {
 	x.Scope = v
 }
+func (x *StartShardEvacuationRequest_Body) GetRepOneOnly() bool {
+	if x != nil {
+		return x.RepOneOnly
+	}
+	return false
+}
+func (x *StartShardEvacuationRequest_Body) SetRepOneOnly(v bool) {
+	x.RepOneOnly = v
+}
 
 // MarshalJSON implements the json.Marshaler interface.
 func (x *StartShardEvacuationRequest_Body) MarshalJSON() ([]byte, error) {
@@ -6653,6 +6704,11 @@ func (x *StartShardEvacuationRequest_Body) MarshalEasyJSON(out *jwriter.Writer)
 		out.RawString(prefix)
 		out.Uint32(x.Scope)
 	}
+	{
+		const prefix string = ",\"repOneOnly\":"
+		out.RawString(prefix)
+		out.Bool(x.RepOneOnly)
+	}
 	out.RawByte('}')
 }
 
@@ -6706,6 +6762,12 @@ func (x *StartShardEvacuationRequest_Body) UnmarshalEasyJSON(in *jlexer.Lexer) {
 				f = in.Uint32()
 				x.Scope = f
 			}
+		case "repOneOnly":
+			{
+				var f bool
+				f = in.Bool()
+				x.RepOneOnly = f
+			}
 		}
 		in.WantComma()
 	}