{cmd/storj-sim, satellite/rangedloop}: added rangedloop to storj-sim, removed identity

added in storj-sim rangedloop for each satellite, to verify it works for metrics oveserver,
removed identity from rangedloop peer as we never use it, added logs on service run, added loop
to service instead of endless for loop, interval value to config

Closes: https://github.com/storj/storj/issues/5414

Change-Id: Ibc3b06071b68feda4a35b45da2bbe36e22a02fc8

cmd/satellite/rangedloop.go

@@ -20,12 +20,6 @@ func cmdRangedLoopRun(cmd *cobra.Command, args []string) (err error) {
 
 	runCfg.Debug.Address = *process.DebugAddrFlag
 
-	identity, err := runCfg.Identity.Load()
-	if err != nil {
-		log.Error("Failed to load identity.", zap.Error(err))
-		return errs.New("Failed to load identity: %+v", err)
-	}
-
 	db, err := satellitedb.Open(ctx, log.Named("db"), runCfg.Database, satellitedb.Options{ApplicationName: "satellite-rangedloop"})
 	if err != nil {
 		return errs.New("Error starting master database on satellite rangedloop: %+v", err)
@@ -42,7 +36,7 @@ func cmdRangedLoopRun(cmd *cobra.Command, args []string) (err error) {
 		err = errs.Combine(err, metabaseDB.Close())
 	}()
 
-	peer, err := satellite.NewRangedLoop(log, identity, db, metabaseDB, &runCfg.Config, process.AtomicLevel(cmd))
+	peer, err := satellite.NewRangedLoop(log, db, metabaseDB, &runCfg.Config, process.AtomicLevel(cmd))
 	if err != nil {
 		return err
 	}

cmd/storj-sim/network.go

@@ -54,6 +54,7 @@ const (
 	versioncontrolPeer  = 2
 	storagenodePeer     = 3
 	multinodePeer       = 5
+	rangedloopPeer      = 6
 
 	// Endpoints.
 	publicRPC  = 0
@@ -424,6 +425,20 @@ func newNetwork(flags *Flags) (*Processes, error) {
 		})
 		coreProcess.WaitForExited(migrationProcess)
 
+		rangedLoopProcess := processes.New(Info{
+			Name:       fmt.Sprintf("satellite-rangedloop/%d", i),
+			Executable: "satellite",
+			Directory:  filepath.Join(processes.Directory, "satellite", fmt.Sprint(i)),
+			Address:    "",
+		})
+		rangedLoopProcess.Arguments = withCommon(rangedLoopProcess.Directory, Arguments{
+			"run": {
+				"ranged-loop",
+				"--debug.addr", net.JoinHostPort(host, port(rangedloopPeer, i, debugCoreHTTP)),
+			},
+		})
+		rangedLoopProcess.WaitForExited(migrationProcess)
+
 		adminProcess := processes.New(Info{
 			Name:       fmt.Sprintf("satellite-admin/%d", i),
 			Executable: "satellite",

private/testplanet/satellite.go

@@ -568,7 +568,7 @@ func (planet *Planet) newSatellite(ctx context.Context, prefix string, index int
 		return nil, err
 	}
 
-	rangedLoopPeer, err := planet.newRangedLoop(ctx, index, identity, db, metabaseDB, config)
+	rangedLoopPeer, err := planet.newRangedLoop(ctx, index, db, metabaseDB, config)
 	if err != nil {
 		return nil, err
 	}
@@ -775,13 +775,13 @@ func (planet *Planet) newGarbageCollectionBF(ctx context.Context, index int, ide
 	return satellite.NewGarbageCollectionBF(log, identity, db, metabaseDB, revocationDB, versionInfo, &config, nil)
 }
 
-func (planet *Planet) newRangedLoop(ctx context.Context, index int, identity *identity.FullIdentity, db satellite.DB, metabaseDB *metabase.DB, config satellite.Config) (_ *satellite.RangedLoop, err error) {
+func (planet *Planet) newRangedLoop(ctx context.Context, index int, db satellite.DB, metabaseDB *metabase.DB, config satellite.Config) (_ *satellite.RangedLoop, err error) {
 	defer mon.Task()(&ctx)(&err)
 
 	prefix := "satellite-ranged-loop" + strconv.Itoa(index)
 	log := planet.log.Named(prefix)
 
-	return satellite.NewRangedLoop(log, identity, db, metabaseDB, &config, nil)
+	return satellite.NewRangedLoop(log, db, metabaseDB, &config, nil)
 }
 
 // atLeastOne returns 1 if value < 1, or value otherwise.

satellite/metabase/rangedloop/service.go

@@ -12,6 +12,7 @@ import (
 	"go.uber.org/zap"
 
 	"storj.io/common/errs2"
+	"storj.io/common/sync2"
 	"storj.io/storj/satellite/metabase/segmentloop"
 )
 
@@ -24,6 +25,7 @@ type Config struct {
 	Parallelism        int           `help:"how many chunks of segments to process in parallel" default:"2"`
 	BatchSize          int           `help:"how many items to query in a batch" default:"2500"`
 	AsOfSystemInterval time.Duration `help:"as of system interval" releaseDefault:"-5m" devDefault:"-1us" testDefault:"-1us"`
+	Interval           time.Duration `help:"how often to run the loop" releaseDefault:"2h" devDefault:"10s" testDefault:"10s"`
 }
 
 // Service iterates through all segments and calls the attached observers for every segment
@@ -34,6 +36,8 @@ type Service struct {
 	config    Config
 	provider  RangeSplitter
 	observers []Observer
+
+	Loop *sync2.Cycle
 }
 
 // NewService creates a new instance of the ranged loop service.
@@ -43,6 +47,7 @@ func NewService(log *zap.Logger, config Config, provider RangeSplitter, observer
 		config:    config,
 		provider:  provider,
 		observers: observers,
+		Loop:      sync2.NewCycle(config.Interval),
 	}
 }
 
@@ -68,7 +73,10 @@ type ObserverDuration struct {
 func (service *Service) Run(ctx context.Context) (err error) {
 	defer mon.Task()(&ctx)(&err)
 
-	for {
+	service.log.Info("ranged loop initialized")
+
+	return service.Loop.Run(ctx, func(ctx context.Context) error {
+		service.log.Info("ranged loop started")
 		_, err := service.RunOnce(ctx)
 		if err != nil {
 			service.log.Error("ranged loop failure", zap.Error(err))
@@ -76,13 +84,17 @@ func (service *Service) Run(ctx context.Context) (err error) {
 			if errs2.IsCanceled(err) {
 				return err
 			}
+
 			if ctx.Err() != nil {
 				return errs.Combine(err, ctx.Err())
 			}
 
 			mon.Event("rangedloop_error") //mon:locked
 		}
-	}
+
+		service.log.Info("ranged loop finished")
+		return nil
+	})
 }
 
 // RunOnce goes through one time and sends information to observers.

satellite/rangedloop.go

@@ -14,8 +14,6 @@ import (
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 
-	"storj.io/common/identity"
-	"storj.io/common/storj"
 	"storj.io/private/debug"
 	"storj.io/storj/private/lifecycle"
 	"storj.io/storj/satellite/audit"
@@ -30,9 +28,8 @@ import (
 //
 // architecture: Peer
 type RangedLoop struct {
-	Log      *zap.Logger
-	Identity *identity.FullIdentity
-	DB       DB
+	Log *zap.Logger
+	DB  DB
 
 	Servers  *lifecycle.Group
 	Services *lifecycle.Group
@@ -64,11 +61,10 @@ type RangedLoop struct {
 }
 
 // NewRangedLoop creates a new satellite ranged loop process.
-func NewRangedLoop(log *zap.Logger, full *identity.FullIdentity, db DB, metabaseDB *metabase.DB, config *Config, atomicLogLevel *zap.AtomicLevel) (*RangedLoop, error) {
+func NewRangedLoop(log *zap.Logger, db DB, metabaseDB *metabase.DB, config *Config, atomicLogLevel *zap.AtomicLevel) (*RangedLoop, error) {
 	peer := &RangedLoop{
-		Log:      log,
-		Identity: full, // TODO: figure out if we need Identity here
-		DB:       db,
+		Log: log,
+		DB:  db,
 
 		Servers:  lifecycle.NewGroup(log.Named("servers")),
 		Services: lifecycle.NewGroup(log.Named("services")),
@@ -169,6 +165,3 @@ func (peer *RangedLoop) Close() error {
 		peer.Services.Close(),
 	)
 }
-
-// ID returns the peer ID.
-func (peer *RangedLoop) ID() storj.NodeID { return peer.Identity.ID }

scripts/testdata/satellite-config.yaml.lock

@@ -877,6 +877,9 @@ identity.key-path: /root/.local/share/storj/identity/satellite/identity.key
 # how many items to query in a batch
 # ranged-loop.batch-size: 2500
 
+# how often to run the loop
+# ranged-loop.interval: 2h0m0s
+
 # how many chunks of segments to process in parallel
 # ranged-loop.parallelism: 2